jackyu66git
ffe7074fef
- onchain_metrics.py: 独立模块，CoinMetrics/CoinGecko/Farside 免费数据源 - main.py: 集成后台线程 + REST API (/api/onchain/metrics, /latest, /available) - requirements.txt: 添加 requests 依赖 - 5分钟自动刷新，CSV 落盘到 data/onchain/
598 lines
25 KiB
Python
598 lines
25 KiB
Python
"""
|
||
链上指标数据模块：拉取 BTC 交易所净流量、稳定币供应、ETF 净流入、MVRV Z-Score、SOPR。
|
||
本地 CSV 落盘 + 内存缓存，参照 derivatives 模式。
|
||
|
||
免费数据源：
|
||
- CoinMetrics Community API: 交易所流入/流出(USD)、MVRV Ratio
|
||
- CoinGecko: 稳定币市值
|
||
需要 API key 的指标（可配置）：
|
||
- ETF 净流入: Coinglass / Farside / Glassnode
|
||
- SOPR: Glassnode / CoinMetrics Pro
|
||
"""
|
||
import csv
|
||
import json
|
||
import logging
|
||
import os
|
||
import re
|
||
import threading
|
||
import time
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
import requests
|
||
from fastapi import APIRouter, HTTPException, Query
|
||
|
||
logger = logging.getLogger("onchain_metrics")
|
||
|
||
# ── 常量 ─────────────────────────────────────────────────────────────
|
||
METRIC_NAMES = [
|
||
"btc_netflow",
|
||
"stablecoin_supply",
|
||
"etf_flow",
|
||
"mvrv_zscore",
|
||
"sopr",
|
||
]
|
||
REFRESH_INTERVAL = 300 # 后台刷新间隔（秒），免费 API 限频较严
|
||
COINMETRICS_BASE = "https://community-api.coinmetrics.io/v4"
|
||
COINGECKO_BASE = "https://api.coingecko.com/api/v3"
|
||
REQUEST_TIMEOUT = 30
|
||
HTTP_HEADERS = {"User-Agent": "Mozilla/5.0 (compatible; chan-data-provider/1.0)"}
|
||
|
||
|
||
def to_utc_iso(ts_ms: int) -> str:
|
||
dt = datetime.fromtimestamp(ts_ms / 1000, tz=timezone.utc)
|
||
return dt.isoformat().replace("+00:00", "Z")
|
||
|
||
|
||
# ── OnchainMetricsManager ────────────────────────────────────────────
|
||
class OnchainMetricsManager:
|
||
"""管理链上指标的拉取、缓存和持久化。每个指标独立存储。"""
|
||
|
||
def __init__(self, data_dir: Path, api_keys: Optional[Dict[str, str]] = None) -> None:
|
||
self.data_dir = data_dir / "onchain"
|
||
self.data_dir.mkdir(parents=True, exist_ok=True)
|
||
self.api_keys = api_keys or {}
|
||
|
||
# 内存缓存: metric_name -> List[Dict]
|
||
self._metrics: Dict[str, List[Dict]] = {name: [] for name in METRIC_NAMES}
|
||
self._lock = threading.RLock()
|
||
self._stop_event = threading.Event()
|
||
self._thread: Optional[threading.Thread] = None
|
||
|
||
# 加载本地历史
|
||
for name in METRIC_NAMES:
|
||
local = self._load_local(name)
|
||
if local:
|
||
self._metrics[name] = local
|
||
logger.info("链上指标 %s 加载本地记录: %d 条", name, len(local))
|
||
|
||
# ── CSV 路径 ─────────────────────────────────────────────────────
|
||
def _path(self, metric: str) -> Path:
|
||
return self.data_dir / f"{metric}.csv"
|
||
|
||
# ── CSV 读写 ─────────────────────────────────────────────────────
|
||
def _load_local(self, metric: str) -> List[Dict]:
|
||
path = self._path(metric)
|
||
if not path.exists():
|
||
return []
|
||
records = []
|
||
with path.open("r", encoding="utf-8", newline="") as fp:
|
||
for row in csv.DictReader(fp):
|
||
try:
|
||
records.append({
|
||
"timestamp": int(row["timestamp"]),
|
||
"datetime": row.get("datetime", ""),
|
||
"value": float(row.get("value", 0)),
|
||
"sub_value": float(row.get("sub_value", 0)) if row.get("sub_value") else None,
|
||
"extra": json.loads(row.get("extra", "{}")) if row.get("extra") else {},
|
||
})
|
||
except (KeyError, ValueError):
|
||
continue
|
||
records.sort(key=lambda r: r["timestamp"])
|
||
return records
|
||
|
||
def _write_local(self, metric: str, records: List[Dict]) -> None:
|
||
path = self._path(metric)
|
||
fieldnames = ["timestamp", "datetime", "value", "sub_value", "extra"]
|
||
tmp = path.with_suffix(".tmp")
|
||
try:
|
||
with tmp.open("w", encoding="utf-8", newline="") as fp:
|
||
w = csv.DictWriter(fp, fieldnames=fieldnames, extrasaction="ignore")
|
||
w.writeheader()
|
||
for r in records:
|
||
row = {
|
||
"timestamp": r.get("timestamp", 0),
|
||
"datetime": r.get("datetime", ""),
|
||
"value": r.get("value", 0),
|
||
"sub_value": "" if r.get("sub_value") is None else r["sub_value"],
|
||
"extra": json.dumps(r.get("extra", {}), ensure_ascii=False),
|
||
}
|
||
w.writerow(row)
|
||
tmp.replace(path)
|
||
logger.info("链上指标 %s 已写入磁盘: %d 条", metric, len(records))
|
||
except Exception as e:
|
||
logger.error("链上指标 %s 落盘失败: %s", metric, e)
|
||
if tmp.exists():
|
||
tmp.unlink()
|
||
|
||
def _merge(self, base: List[Dict], new: List[Dict]) -> List[Dict]:
|
||
"""按 timestamp 去重合并，新覆盖旧。"""
|
||
merged = {r["timestamp"]: r for r in base}
|
||
for r in new:
|
||
merged[r["timestamp"]] = r
|
||
return sorted(merged.values(), key=lambda r: r["timestamp"])
|
||
|
||
# ── 数据拉取 ─────────────────────────────────────────────────────
|
||
def _fetch_cm_metrics(
|
||
self, metrics: str, days: int = 365
|
||
) -> Optional[List[Dict]]:
|
||
"""从 CoinMetrics Community API 拉取指标。返回 [{time, metric: value}, ...]"""
|
||
page_size = min(days, 10000)
|
||
url = (
|
||
f"{COINMETRICS_BASE}/timeseries/asset-metrics"
|
||
f"?assets=btc&metrics={metrics}&frequency=1d&page_size={page_size}"
|
||
)
|
||
try:
|
||
r = requests.get(url, timeout=REQUEST_TIMEOUT, headers=HTTP_HEADERS)
|
||
if r.status_code != 200:
|
||
logger.warning("CoinMetrics %s 返回 %d: %s", metrics, r.status_code, r.text[:200])
|
||
return None
|
||
data = r.json().get("data", [])
|
||
if not data:
|
||
return None
|
||
# 翻页取更多历史
|
||
all_data = list(data)
|
||
next_url = r.json().get("next_page_url")
|
||
pages = 0
|
||
while next_url and pages < 10:
|
||
pages += 1
|
||
time.sleep(0.5)
|
||
r2 = requests.get(next_url, timeout=REQUEST_TIMEOUT, headers=HTTP_HEADERS)
|
||
if r2.status_code != 200:
|
||
break
|
||
batch = r2.json()
|
||
all_data.extend(batch.get("data", []))
|
||
next_url = batch.get("next_page_url")
|
||
if not next_url or next_url == r.json().get("next_page_url"):
|
||
break
|
||
return all_data
|
||
except Exception as e:
|
||
logger.error("CoinMetrics %s 拉取失败: %s", metrics, e)
|
||
return None
|
||
|
||
def _fetch_btc_netflow(self) -> List[Dict]:
|
||
"""BTC 交易所净流量 (USD) = 流入 - 流出。"""
|
||
data = self._fetch_cm_metrics("FlowInExUSD,FlowOutExUSD", days=365)
|
||
if not data:
|
||
return []
|
||
|
||
records = []
|
||
for d in data:
|
||
ts_str = d.get("time", "")
|
||
inflow = float(d.get("FlowInExUSD") or 0)
|
||
outflow = float(d.get("FlowOutExUSD") or 0)
|
||
netflow = inflow - outflow
|
||
try:
|
||
ts = int(datetime.fromisoformat(ts_str.replace("Z", "+00:00")).timestamp() * 1000)
|
||
except Exception:
|
||
continue
|
||
records.append({
|
||
"timestamp": ts,
|
||
"datetime": to_utc_iso(ts),
|
||
"value": round(netflow, 2),
|
||
"sub_value": round(inflow, 2),
|
||
"extra": {"inflow_usd": round(inflow, 2), "outflow_usd": round(outflow, 2)},
|
||
})
|
||
return sorted(records, key=lambda r: r["timestamp"])
|
||
|
||
def _fetch_stablecoin_supply(self) -> List[Dict]:
|
||
"""稳定币总供应（USDT + USDC + DAI + FDUSD + TUSD 市值之和）。"""
|
||
try:
|
||
url = (
|
||
f"{COINGECKO_BASE}/coins/markets"
|
||
f"?vs_currency=usd&category=stablecoins&order=market_cap_desc"
|
||
f"&per_page=5&page=1&sparkline=false"
|
||
)
|
||
r = requests.get(url, timeout=REQUEST_TIMEOUT, headers=HTTP_HEADERS)
|
||
if r.status_code != 200:
|
||
logger.warning("CoinGecko stablecoins 返回 %d", r.status_code)
|
||
return []
|
||
coins = r.json()
|
||
if not isinstance(coins, list):
|
||
return []
|
||
|
||
total_mcap = sum(c.get("market_cap", 0) or 0 for c in coins)
|
||
now_ms = int(time.time() * 1000)
|
||
breakdown = {
|
||
c.get("symbol", "?").upper(): c.get("market_cap", 0) or 0
|
||
for c in coins[:5]
|
||
}
|
||
return [{
|
||
"timestamp": now_ms,
|
||
"datetime": to_utc_iso(now_ms),
|
||
"value": round(total_mcap, 2),
|
||
"sub_value": round(total_mcap / 1e9, 2), # Billions
|
||
"extra": breakdown,
|
||
}]
|
||
except Exception as e:
|
||
logger.error("稳定币供应拉取失败: %s", e)
|
||
return []
|
||
|
||
def _fetch_etf_flow(self) -> List[Dict]:
|
||
"""BTC ETF 净流入/流出。优先从 Farside 免费爬取，Coinglass 为备用。
|
||
|
||
数据源优先级:
|
||
1. Farside (免费, HTML 爬取)
|
||
2. Coinglass (需 coinglass_key)
|
||
3. Glassnode (需 glassnode_key)
|
||
"""
|
||
# ── 优先：Farside 免费爬取 ──
|
||
try:
|
||
records = self._fetch_etf_farside()
|
||
if records:
|
||
logger.info("ETF 数据从 Farside 拉取: %d 条", len(records))
|
||
return records
|
||
except Exception as e:
|
||
logger.warning("Farside ETF 爬取失败: %s", e)
|
||
|
||
# ── 备用：Coinglass ──
|
||
cg_key = self.api_keys.get("coinglass_key")
|
||
if cg_key:
|
||
try:
|
||
url = "https://open-api-v3.coinglass.com/api/bitcoin/etf/net-inflow?interval=30"
|
||
r = requests.get(url, timeout=REQUEST_TIMEOUT, headers={
|
||
**HTTP_HEADERS, "coinglassSecret": cg_key,
|
||
})
|
||
if r.status_code == 200:
|
||
data = r.json()
|
||
records = []
|
||
if isinstance(data, dict) and "data" in data:
|
||
for item in data["data"]:
|
||
ts = int(item.get("date", 0)) * 1000 if item.get("date") else 0
|
||
if ts:
|
||
records.append({
|
||
"timestamp": ts,
|
||
"datetime": to_utc_iso(ts),
|
||
"value": float(item.get("netInflow", 0)),
|
||
"sub_value": None,
|
||
"extra": item,
|
||
})
|
||
return records
|
||
logger.warning("Coinglass ETF 返回 %d", r.status_code)
|
||
except Exception as e:
|
||
logger.error("Coinglass ETF 拉取失败: %s", e)
|
||
|
||
logger.warning("ETF 数据源均不可用（Farside/Coinglass/Glassnode）")
|
||
return []
|
||
|
||
def _fetch_etf_farside(self) -> List[Dict]:
|
||
"""从 Farside 网站爬取 BTC ETF 每日净流量。
|
||
|
||
https://farside.co.uk/btc/ 页面包含一个 HTML 表格，
|
||
每行包含日期和各 ETF 的当日流量（百万美元），最后一列为总计。
|
||
"""
|
||
url = "https://farside.co.uk/btc/"
|
||
r = requests.get(url, timeout=REQUEST_TIMEOUT, headers={
|
||
**HTTP_HEADERS,
|
||
"Accept": "text/html,application/xhtml+xml,*/*",
|
||
})
|
||
if r.status_code != 200:
|
||
logger.warning("Farside 返回 %d", r.status_code)
|
||
return []
|
||
|
||
html = r.text
|
||
|
||
# 查找表格
|
||
table_match = re.search(r'<table[^>]*>(.*?)</table>', html, re.DOTALL)
|
||
if not table_match:
|
||
logger.warning("Farside 页面未找到表格")
|
||
return []
|
||
|
||
rows_html = re.findall(r'<tr[^>]*>(.*?)</tr>', table_match.group(1), re.DOTALL)
|
||
month_map = {
|
||
"jan": 1, "feb": 2, "mar": 3, "apr": 4, "may": 5, "jun": 6,
|
||
"jul": 7, "aug": 8, "sep": 9, "oct": 10, "nov": 11, "dec": 12,
|
||
}
|
||
|
||
records = []
|
||
for row_html in rows_html:
|
||
cells = re.findall(r'<t[dh][^>]*>(.*?)</t[dh]>', row_html, re.DOTALL)
|
||
# 清理 HTML 标签和空白
|
||
clean = []
|
||
for c in cells:
|
||
t = re.sub(r'<[^>]+>', '', c).strip()
|
||
t = t.replace('\xa0', ' ').replace(' ', ' ').strip()
|
||
clean.append(t)
|
||
|
||
if not clean:
|
||
continue
|
||
|
||
# 第一列应为日期格式 "15 Jun 2026"
|
||
date_str = clean[0]
|
||
parts = date_str.split()
|
||
if len(parts) != 3:
|
||
continue
|
||
day_str, mon_str, year_str = parts
|
||
mon = month_map.get(mon_str.lower()[:3])
|
||
if mon is None:
|
||
continue
|
||
try:
|
||
day = int(day_str)
|
||
year = int(year_str)
|
||
except ValueError:
|
||
continue
|
||
|
||
# 最后一列为总计（可能带括号表示负值）
|
||
total_str = clean[-1] if len(clean) > 1 else ""
|
||
if not total_str or total_str in ("", "Total", "-"):
|
||
continue
|
||
# 解析 "(123.4)" → -123.4, "123.4" → 123.4
|
||
total_str = total_str.replace(",", "")
|
||
is_negative = total_str.startswith("(") and total_str.endswith(")")
|
||
if is_negative:
|
||
total_str = total_str[1:-1]
|
||
try:
|
||
total_m = float(total_str)
|
||
except ValueError:
|
||
continue
|
||
if is_negative:
|
||
total_m = -total_m
|
||
|
||
# 构建时间戳（UTC 午夜）
|
||
from datetime import datetime, timezone as tz
|
||
dt = datetime(year, mon, day, tzinfo=tz.utc)
|
||
ts = int(dt.timestamp() * 1000)
|
||
|
||
# 分解各 ETF 明细
|
||
etf_breakdown = {}
|
||
if len(clean) > 2:
|
||
etf_names = ["IBIT", "FBTC", "BITB", "ARKB", "BTCO", "EZBC",
|
||
"BRRR", "HODL", "BTCW", "MSBT", "GBTC", "BTC"]
|
||
for i, name in enumerate(etf_names):
|
||
idx = i + 1
|
||
if idx < len(clean) - 1:
|
||
val_str = clean[idx].replace(",", "").replace("(", "").replace(")", "")
|
||
neg = clean[idx].startswith("(")
|
||
try:
|
||
val = float(val_str)
|
||
etf_breakdown[name] = -val if neg else val
|
||
except ValueError:
|
||
pass
|
||
|
||
records.append({
|
||
"timestamp": ts,
|
||
"datetime": dt.isoformat().replace("+00:00", "Z"),
|
||
"value": round(total_m, 2), # 净流入 (百万 USD)
|
||
"sub_value": round(total_m * 1_000_000, 2), # 净流入 (USD)
|
||
"extra": {"total_million_usd": round(total_m, 2), "breakdown": etf_breakdown},
|
||
})
|
||
|
||
return sorted(records, key=lambda r: r["timestamp"])
|
||
|
||
def _fetch_mvrv_zscore(self) -> List[Dict]:
|
||
"""MVRV Z-Score = (当前 MVRV - 滚动均值) / 滚动标准差。
|
||
|
||
从 CoinMetrics 拉取 CapMVRVCur（免费），计算 365 天滚动 Z-Score。
|
||
"""
|
||
data = self._fetch_cm_metrics("CapMVRVCur", days=400)
|
||
if not data:
|
||
return []
|
||
|
||
# 解析时间序列
|
||
mvrv_series = []
|
||
for d in data:
|
||
ts_str = d.get("time", "")
|
||
mvrv_val = d.get("CapMVRVCur")
|
||
if mvrv_val is None:
|
||
continue
|
||
try:
|
||
ts = int(datetime.fromisoformat(ts_str.replace("Z", "+00:00")).timestamp() * 1000)
|
||
except Exception:
|
||
continue
|
||
mvrv_series.append((ts, float(mvrv_val)))
|
||
mvrv_series.sort(key=lambda x: x[0])
|
||
|
||
# 计算滚动 Z-Score (365 天窗口，即约 365 个数据点)
|
||
window = 365
|
||
records = []
|
||
values = []
|
||
timestamps = []
|
||
for ts, val in mvrv_series:
|
||
values.append(val)
|
||
timestamps.append(ts)
|
||
if len(values) < window:
|
||
continue
|
||
window_vals = values[-window:]
|
||
mean = sum(window_vals) / window
|
||
variance = sum((v - mean) ** 2 for v in window_vals) / window
|
||
stddev = variance ** 0.5
|
||
zscore = (val - mean) / stddev if stddev > 0 else 0.0
|
||
records.append({
|
||
"timestamp": ts,
|
||
"datetime": to_utc_iso(ts),
|
||
"value": round(zscore, 4),
|
||
"sub_value": round(val, 4),
|
||
"extra": {"mvrv_ratio": round(val, 4), "rolling_mean": round(mean, 4), "rolling_stddev": round(stddev, 4)},
|
||
})
|
||
return records
|
||
|
||
def _fetch_sopr(self) -> List[Dict]:
|
||
"""SOPR (Spent Output Profit Ratio)。需要 API key。
|
||
|
||
支持:
|
||
- glassnode_key: Glassnode v1/metrics/indicators/sopr
|
||
"""
|
||
gn_key = self.api_keys.get("glassnode_key")
|
||
if gn_key:
|
||
try:
|
||
url = (
|
||
f"https://api.glassnode.com/v1/metrics/indicators/sopr"
|
||
f"?a=btc&f=json&api_key={gn_key}"
|
||
)
|
||
r = requests.get(url, timeout=REQUEST_TIMEOUT, headers=HTTP_HEADERS)
|
||
if r.status_code == 200:
|
||
data = r.json()
|
||
records = []
|
||
for item in data if isinstance(data, list) else []:
|
||
ts = int(item.get("t", 0)) * 1000
|
||
if ts:
|
||
records.append({
|
||
"timestamp": ts,
|
||
"datetime": to_utc_iso(ts),
|
||
"value": float(item.get("v", 0)),
|
||
"sub_value": None,
|
||
"extra": {"raw": item},
|
||
})
|
||
return records
|
||
logger.warning("Glassnode SOPR 返回 %d", r.status_code)
|
||
except Exception as e:
|
||
logger.error("Glassnode SOPR 拉取失败: %s", e)
|
||
|
||
logger.warning(
|
||
"SOPR 数据未配置 API key。请设置环境变量 GLASSNODE_API_KEY"
|
||
)
|
||
return []
|
||
|
||
# ── 刷新入口 ─────────────────────────────────────────────────────
|
||
def refresh_all(self) -> None:
|
||
"""拉取全部 5 个指标，合并到内存并写盘。"""
|
||
fetchers = {
|
||
"btc_netflow": self._fetch_btc_netflow,
|
||
"stablecoin_supply": self._fetch_stablecoin_supply,
|
||
"etf_flow": self._fetch_etf_flow,
|
||
"mvrv_zscore": self._fetch_mvrv_zscore,
|
||
"sopr": self._fetch_sopr,
|
||
}
|
||
|
||
for name, fetcher in fetchers.items():
|
||
try:
|
||
new_records = fetcher()
|
||
if not new_records:
|
||
continue
|
||
with self._lock:
|
||
base = list(self._metrics.get(name, []))
|
||
merged = self._merge(base, new_records)
|
||
self._metrics[name] = merged
|
||
self._write_local(name, merged)
|
||
logger.info("链上指标 %s 刷新: +%d 条, 总计 %d 条",
|
||
name, len(new_records), len(merged))
|
||
except Exception:
|
||
logger.exception("链上指标 %s 刷新异常", name)
|
||
|
||
# ── 后台线程 ─────────────────────────────────────────────────────
|
||
def _refresh_loop(self) -> None:
|
||
"""后台线程：启动立即拉取一次，之后每 REFRESH_INTERVAL 秒刷新。"""
|
||
logger.info("链上指标后台线程启动，间隔 %ds", REFRESH_INTERVAL)
|
||
# 启动立即拉取
|
||
self.refresh_all()
|
||
while not self._stop_event.wait(REFRESH_INTERVAL):
|
||
self.refresh_all()
|
||
logger.info("链上指标后台线程已退出")
|
||
|
||
def start(self) -> None:
|
||
self._stop_event.clear()
|
||
self._thread = threading.Thread(
|
||
target=self._refresh_loop, name="onchain-loop", daemon=True,
|
||
)
|
||
self._thread.start()
|
||
logger.info("链上指标模块已启动")
|
||
|
||
def stop(self) -> None:
|
||
self._stop_event.set()
|
||
if self._thread:
|
||
self._thread.join(timeout=5)
|
||
|
||
# ── 查询接口 ─────────────────────────────────────────────────────
|
||
def get_metric(
|
||
self,
|
||
metric: str,
|
||
limit: int = 100,
|
||
start_ms: Optional[int] = None,
|
||
end_ms: Optional[int] = None,
|
||
) -> List[Dict]:
|
||
if metric not in METRIC_NAMES:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail=f"不支持的指标: {metric}。可选: {METRIC_NAMES}",
|
||
)
|
||
with self._lock:
|
||
records = list(self._metrics.get(metric, []))
|
||
if start_ms is not None:
|
||
records = [r for r in records if r["timestamp"] >= start_ms]
|
||
if end_ms is not None:
|
||
records = [r for r in records if r["timestamp"] <= end_ms]
|
||
return records[-limit:] if limit else records
|
||
|
||
def get_latest(self) -> Dict[str, Any]:
|
||
"""获取所有指标的最新值。"""
|
||
result = {"timestamp": int(time.time() * 1000)}
|
||
for name in METRIC_NAMES:
|
||
with self._lock:
|
||
records = self._metrics.get(name, [])
|
||
if records:
|
||
latest = records[-1]
|
||
result[name] = {
|
||
"timestamp": latest["timestamp"],
|
||
"datetime": latest["datetime"],
|
||
"value": latest["value"],
|
||
"sub_value": latest.get("sub_value"),
|
||
"extra": latest.get("extra", {}),
|
||
}
|
||
else:
|
||
result[name] = None
|
||
return result
|
||
|
||
|
||
# ── FastAPI Router ────────────────────────────────────────────────────
|
||
def create_onchain_router(manager: OnchainMetricsManager) -> APIRouter:
|
||
router = APIRouter(prefix="/api/onchain", tags=["onchain"])
|
||
|
||
@router.get("/metrics")
|
||
async def get_metrics(
|
||
metric: str = Query(..., description=f"指标名称: {', '.join(METRIC_NAMES)}"),
|
||
limit: int = Query(100, ge=1, le=5000, description="返回条数上限"),
|
||
start: Optional[int] = Query(None, description="起始时间戳(ms)"),
|
||
end: Optional[int] = Query(None, description="结束时间戳(ms)"),
|
||
):
|
||
"""获取单个链上指标的时间序列。"""
|
||
data = manager.get_metric(metric, limit=limit, start_ms=start, end_ms=end)
|
||
return {"metric": metric, "count": len(data), "data": data}
|
||
|
||
@router.get("/latest")
|
||
async def get_latest():
|
||
"""获取全部 5 个指标的最新快照。"""
|
||
return manager.get_latest()
|
||
|
||
@router.get("/available")
|
||
async def get_available():
|
||
"""列出可用指标及当前数据量。"""
|
||
info = {}
|
||
for name in METRIC_NAMES:
|
||
records = manager.get_metric(name, limit=0)
|
||
info[name] = {
|
||
"count": len(records),
|
||
"has_data": len(records) > 0,
|
||
"first_ts": records[0]["datetime"] if records else None,
|
||
"last_ts": records[-1]["datetime"] if records else None,
|
||
}
|
||
return info
|
||
|
||
return router
|
||
|
||
|
||
# ── 环境变量辅助 ─────────────────────────────────────────────────────
|
||
def api_keys_from_env() -> Dict[str, str]:
|
||
"""从环境变量读取 API keys。"""
|
||
keys = {}
|
||
for env_var, key_name in [
|
||
("COINGLASS_API_KEY", "coinglass_key"),
|
||
("GLASSNODE_API_KEY", "glassnode_key"),
|
||
("FARSIDE_API_KEY", "farside_key"),
|
||
("COINMETRICS_API_KEY", "coinmetrics_key"),
|
||
]:
|
||
val = os.getenv(env_var, "").strip()
|
||
if val and "***" not in val: # 忽略占位符
|
||
keys[key_name] = val
|
||
return keys
|