
本文详解如何通过逆向分析 coincodex 的前端 api,稳定获取加密货币全市场总市值等高频时序数据,并封装为可复用的 python 脚本,支持批量采集、时间对齐与 pandas 标准化处理。
CoinCodex(https://www.php.cn/link/f5fee8f7da74f4887f5bcae2bafb6dd6)以交互式图表形式展示加密货币市场总市值(SUM_ALL_COINS)、比特币(BTC)、以太坊(ETH)等关键资产的历史走势。这些图表由前端 JavaScript 动态加载,其真实数据源并非 HTML 页面本身,而是隐藏在后台的 RESTful API 接口。直接请求网页 HTML 无法获取原始时间序列,而盲目构造参数又易因签名、时间戳或反爬机制失败。所幸该站 API 设计相对开放——无需认证密钥,仅需正确传递查询参数即可返回结构化 JSON 数据。
以下是一个完整、健壮的采集示例,适用于 SUM_ALL_COINS(全市场总市值),并可轻松扩展至其他资产:
import pandas as pd
import requests
from typing import Optional, Dict, List
def fetch_chart_data(
asset: str = "SUM_ALL_COINS",
include: str = "market_cap",
samples: str = "md", # 'sm' (small), 'md' (medium), 'lg' (large)
charts: str = "ALL",
timeout: int = 10
) -> pd.DataFrame:
"""
从 CoinCodex API 抓取指定资产的图表数据
Parameters:
-----------
asset : str
资产标识符,如 "SUM_ALL_COINS", "BTC", "ETH"
include : str
数据类型,如 "market_cap", "price", "volume"
samples : str
数据采样粒度:'sm'(约 300 点)、'md'(约 600 点)、'lg'(约 1200 点)
charts : str
图表类型,"ALL" 表示主趋势线(推荐)
timeout : int
HTTP 请求超时(秒)
Returns:
--------
pd.DataFrame with columns ['Date', 'Value', 'Cap'] (for market_cap)
"""
api_url = "https://coincodex.com/api/v1/assets/get_charts"
# 注意:t 参数为 Unix 时间戳(秒),但实际作用是缓存键/防重放,非必需
# 实测留空或使用当前时间戳均可成功;为稳定性建议固定或省略
params = {
"charts": charts,
"samples": samples,
"assets": asset,
"include": include,
}
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}
try:
response = requests.get(api_url, params=params, headers=headers, timeout=timeout)
response.raise_for_status()
data = response.json()
# 解析响应:data[asset]["ALL"] 是核心时间序列列表
if asset not in data:
raise KeyError(f"Asset '{asset}' not found in API response")
if "ALL" not in data[asset]:
raise KeyError(f"'ALL' chart not available for asset '{asset}'")
raw_series = data[asset]["ALL"]
# 构建 DataFrame(注意字段顺序可能随 API 微调变化,此处按典型结构)
df = pd.DataFrame(raw_series, columns=["Date", "Value", "Cap"])
df["Date"] = pd.to_datetime(df["Date"], unit="s", utc=True).dt.tz_localize(None)
return df.sort_values("Date").reset_index(drop=True)
except requests.exceptions.RequestException as e:
raise ConnectionError(f"API request failed: {e}")
except (KeyError, ValueError, TypeError) as e:
raise RuntimeError(f"Failed to parse API response: {e}")
# ✅ 使用示例:获取全市场总市值(中等粒度)
df_market = fetch_chart_data(
asset="SUM_ALL_COINS",
include="market_cap",
samples="md"
)
print("✅ 全市场总市值数据(最后5条):")
print(df_market.tail())
# ✅ 批量采集示例:BTC + ETH 市值对比
assets = ["BTC", "ETH"]
dfs = {}
for a in assets:
dfs[a] = fetch_chart_data(asset=a, include="market_cap", samples="sm")
# 合并并按日期对齐(可选)
aligned = pd.concat([
dfs["BTC"][["Date", "Value"]].rename(columns={"Value": "BTC_MCap"}),
dfs["ETH"][["Date", "Value"]].rename(columns={"Value": "ETH_MCap"})
], on="Date", how="outer").sort_values("Date").reset_index(drop=True)关键注意事项与最佳实践:
- ? 参数 t 并非必需:原问题中尝试的 t=5693725 实为过期或无效时间戳(远小于 Unix 纪元)。实测该参数可完全省略,或设为当前时间戳 int(time.time()) 以提升缓存命中率,但不影响数据可用性。
- ? 响应结构依赖资产类型:data["BTC"]["ALL"] 是标准路径,但部分小币种可能仅返回 data["XRP"] 下单层结构。建议始终校验 asset in data 和 data[asset].get("ALL")。
- ⏱️ 采样粒度说明:samples=sm/md/lg 控制返回点数(非时间跨度),md 在精度与体积间较平衡;若需更长历史,优先增大 samples 而非调整 t。
- ? 反爬友好策略:添加 User-Agent 头已足够;避免高频请求(建议 ≥5 秒间隔),生产环境可加入 requests.Session() 复用连接及简单重试逻辑。
- ? 数据字段含义:对于 include="market_cap",三列通常为 [timestamp_s, market_cap_usd, volume_usd];若 include="price",则为 [timestamp_s, price_usd, market_cap_usd] —— 请以实际响应为准并做字段验证。
通过上述方法,你不仅能自动化获取 CoinCodex 的高质量时序特征,还可将 fetch_chart_data() 函数封装进数据管道,配合定时任务(如 APScheduler 或 cron)实现每日更新,为量化模型训练持续注入可靠的历史数据源。










