From f2e77e1bdb563d4f023db8e241230e46b7ea5cc9 Mon Sep 17 00:00:00 2001 From: jackyu66git Date: Wed, 5 Aug 2026 18:10:29 +0800 Subject: [PATCH] =?UTF-8?q?chore:=20=E5=B0=86=20data=5Fprovider=20?= =?UTF-8?q?=E6=8B=86=E5=87=BA=E4=B8=BA=E7=8B=AC=E7=AB=8B=E4=BB=93=E5=BA=93?= =?UTF-8?q?=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 数据服务已迁移至 jack/data_provider，不再随 chan 维护。 Co-authored-by: Cursor --- data_provider/Dockerfile | 20 - data_provider/README.md | 82 -- data_provider/api_docs.html | 576 --------- data_provider/config.json | 31 - data_provider/docker-compose.yml | 15 - data_provider/homepage.html | 425 ------- data_provider/main.py | 2014 ------------------------------ data_provider/onchain_metrics.py | 597 --------- data_provider/requirements.txt | 6 - 9 files changed, 3766 deletions(-) delete mode 100644 data_provider/Dockerfile delete mode 100644 data_provider/README.md delete mode 100644 data_provider/api_docs.html delete mode 100644 data_provider/config.json delete mode 100644 data_provider/docker-compose.yml delete mode 100644 data_provider/homepage.html delete mode 100644 data_provider/main.py delete mode 100644 data_provider/onchain_metrics.py delete mode 100644 data_provider/requirements.txt diff --git a/data_provider/Dockerfile b/data_provider/Dockerfile deleted file mode 100644 index 6f63ba9..0000000 --- a/data_provider/Dockerfile +++ /dev/null @@ -1,20 +0,0 @@ -FROM python:3.11-slim - -ENV PYTHONUNBUFFERED=1 \ - PIP_NO_CACHE_DIR=1 - -WORKDIR /app - -COPY requirements.txt /app/requirements.txt -RUN pip install --no-cache-dir -r requirements.txt - -COPY . /app - -ENV CONFIG_PATH=/app/config.json \ - UVICORN_HOST=0.0.0.0 \ - UVICORN_PORT=80 - -EXPOSE 80 - -CMD ["python", "-m", "main"] - diff --git a/data_provider/README.md b/data_provider/README.md deleted file mode 100644 index f8ce8b2..0000000 --- a/data_provider/README.md +++ /dev/null @@ -1,82 +0,0 @@ -# Chan 数据提供商 (Chan Data Provider) - -从 **Binance 期货** 交易所拉取加密货币 K 线数据，提供 HTTP + WebSocket 数据服务。 - -## 功能 - -- **多交易对**：支持 BTC, ETH, SOL, DOGE 等 9 个交易对 -- **多时间周期**：基础周期 1m/1h/1d/1w，可合成 30+ 种衍生周期（如 5m, 15m, 4h 等） -- **本地缓存**：CSV 持久化到磁盘，重启快速加载 -- **断线恢复**：交易所连接中断时记录断点，自动补拉缺失数据 -- **实时推送**：WebSocket 订阅最新 K 线更新 -- **内存服务**：启动即加载本地数据，不阻塞服务 - -## 启动 - -```bash -# Docker -docker compose up -d - -# 直接运行 -python main.py - -# 或指定配置 -CONFIG_PATH=./config.json python main.py -``` - -服务默认监听 `http://0.0.0.0:9009`。 - -## 配置 - -编辑 `config.json`： - -```json -{ - "exchange": "binance", - "symbols": ["BTC/USDT:USDT", "ETH/USDT:USDT"], - "start_time": "2024-01-01T00:00:00Z", - "timeframes": ["1m", "1h", "1d", "1w"], - "data_dir": "./data" -} -``` - -| 字段 | 说明 | -|------|------| -| `exchange` | 交易所名称（ccxt 支持即可） | -| `symbols` | 交易对列表 | -| `start_time` | 历史数据起始时间 | -| `timeframes` | 基础周期（从交易所直接拉取） | -| `data_dir` | CSV 数据存储目录 | - -## 可用周期 - -### 基础周期（交易所直接拉取） -`1m`, `1h`, `1d`, `1w` - -### 衍生周期（内存中合成） -| 基础周期 | 可合成的衍生周期 | -|----------|----------------| -| 1m | 2m, 3m, 4m, 5m, 10m, 15m, 20m, 25m, 30m, 45m | -| 1h | 2h, 3h, 4h, 5h, 6h, 7h, 8h, 9h, 10h, 11h, 12h, 16h, 20h | -| 1d | 2d, 3d, 4d, 5d, 6d | -| 1w | 2w, 3w | - -## 数据存储 - -数据以 CSV 格式存储，按时间周期分目录： - -``` -./data/ - 1m/ - binance_BTC_USDT_USDT_1m.csv - binance_ETH_USDT_USDT_1m.csv - ... - 1h/ - ... -``` - -每根 K 线包含：`timestamp`, `datetime`, `open`, `high`, `low`, `close`, `volume`。 - ---- - -API 文档请访问 `http://:9009/api/docs`。 diff --git a/data_provider/api_docs.html b/data_provider/api_docs.html deleted file mode 100644 index 681f306..0000000 --- a/data_provider/api_docs.html +++ /dev/null @@ -1,576 +0,0 @@ - - - - - - - - -Chan 数据提供商 - API 文档 - - - -
- -
-

Chan 数据提供商

-

加密货币 K 线 + 衍生品数据 HTTP + WebSocket API

-
v1.0.0  |  binance  |  port 9009
-
- - - - -
-

服务信息

-
-
- GET - / - 服务基本信息 -
-
-

返回服务名称、交易所、交易对列表、可用周期及就绪状态。

-

响应

- 
{
-  "service":        "Data Provider",
-  "exchange":        "binance",
-  "symbols":         ["BTC/USDT:USDT", "ETH/USDT:USDT", ...],
-  "base_timeframes":  ["1m", "1h", "1d", "1w"],
-  "derived_timeframes": ["5m", "15m", "4h", ...],
-  "timeframes":       ["1m", "1h", ..., "5m", "15m", ...],
-  "ready":           true
-}
-
-
-
- - -
-

健康检查

-
-
- GET - /health - 存活检查 -
-
-

返回服务健康状态，与 / 相同结构，适合负载均衡探测器。

-

响应

- 
{
-  "status":  "ok",
-  "exchange": "binance",
-  "symbols":  ["BTC/USDT:USDT", ...],
-  "ready":    true,
-  ...
-}
-
-
-
- - -
-

可用周期

-
-
- GET - /timeframes - 列出所有时间周期 -
-
-

返回基础周期（交易所直接拉取）和衍生周期（合成生成）的完整列表。

-

响应

- 
{
-  "base_timeframes":    ["1m", "1h", "1d", "1w"],
-  "derived_timeframes": ["5m", "15m", "4h", ...],
-  "timeframes":         ["1m", "1h", ..., "5m", "15m", ...]
-}
-
-
-
- - -
-

查询 K 线

-
-
- GET - /api/candles - 获取 OHLCV K 线数据 -
-
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
参数类型必填说明
symbolstring是交易对，如 BTC/USDT:USDT
tfstring否时间周期，默认 1m。支持基础及衍生周期
startint可选开始时间戳（毫秒）
endint可选结束时间戳（毫秒）
limitint可选限制返回的 K 线数量（返回最后 N 根）
- -
若不传 start/end，返回内存中全部数据（可能很多），建议搭配 limit 使用。
- -

请求示例

- 
# 获取 BTC 最近 100 根 5 分钟 K 线
-GET /api/candles?symbol=BTC/USDT:USDT&tf=5m&limit=100
-
-# 指定时间范围
-GET /api/candles?symbol=ETH/USDT:USDT&tf=1h&start=1704067200000&end=1704153600000
-
-# 获取 4 小时周期（衍生周期）
-GET /api/candles?symbol=SOL/USDT:USDT&tf=4h&limit=50
- -

响应

-

返回 OHLCV 对象数组：

- 
[
-  {
-    "timestamp": 1704067200000,
-    "datetime":  "2024-01-01T00:00:00Z",
-    "open":      42850.12,
-    "high":      43100.00,
-    "low":       42780.50,
-    "close":     43050.80,
-    "volume":    125.34
-  },
-  ...
-]
- -

字段说明

- - - - - - - - - -
字段类型说明
timestampintUTC 毫秒时间戳
datetimestringISO 8601 格式（末尾 Z）
openfloat开盘价
highfloat最高价
lowfloat最低价
closefloat收盘价
volumefloat成交量
- -
-
-
- - -
-

查询衍生品数据

-
-
- GET - /api/derivatives - 获取资金费率、持仓量、基差 -
-
- - - - - - - - - -
参数类型必填说明
symbolstring否交易对，默认 BTC/USDT:USDT
- -
数据每 60 秒自动刷新，落盘到 data/derivatives/ 目录。
- -

请求示例

- 
# 获取 BTC 衍生品数据
-GET /api/derivatives?symbol=BTC/USDT:USDT
- -

响应

- 
{
-  "symbol":          "BTC/USDT:USDT",
-  "timestamp":       1719705600000,
-  "datetime":        "2024-06-30T00:00:00Z",
-  "funding_rate":    0.0001,
-  "open_interest":   35120000000.0,
-  "oi_change_pct":   3.52,
-  "basis":           8.5
-}
- -

字段说明

- - - - - - -
字段类型说明
funding_ratefloat当前资金费率（每 8 小时）
open_interestfloat当前持仓量（USD）
oi_change_pctfloat24 小时持仓量变化百分比
basisfloat期货-现货年化基差（%）
-
-
-
- - -
-

WebSocket 实时推送

-
-
- WS - /ws - 实时 K 线订阅 -
-
- -

连接 WebSocket 后，通过 JSON 消息进行订阅管理。服务端在数据更新时主动推送最新 K 线。

- -

客户端 → 服务端

- -
-
订阅 K 线
- 
{
-  "action":    "subscribe",
-  "symbol":    "BTC/USDT:USDT",
-  "timeframe": "1m"
-}
-
- -
-
取消订阅
- 
{
-  "action":    "unsubscribe",
-  "symbol":    "BTC/USDT:USDT",
-  "timeframe": "1m"
-}
-
- -
-
心跳 Ping
- 
{ "action": "ping" }
-
- -

服务端 → 客户端

- -
-
订阅确认
- 
{
-  "type":      "subscribed",
-  "symbol":    "BTC/USDT:USDT",
-  "timeframe": "1m"
-}
-
- -
-
初始快照（订阅后立即推送最近 500 根 K 线）
- 
{
-  "type":      "snapshot",
-  "symbol":    "BTC/USDT:USDT",
-  "timeframe": "1m",
-  "data":      [ ... ]
-}
-
- -
-
K 线更新（增量推送最近 2 根）
- 
{
-  "type":      "kline",
-  "symbol":    "BTC/USDT:USDT",
-  "timeframe": "1m",
-  "data":      [ ... ]
-}
-
- -
-
Pong 响应
- 
{ "type": "pong" }
-
- -
-
错误消息
- 
{ "type": "error", "message": "..." }
-
- -

JavaScript 示例

- 
// 连接
-const ws = new WebSocket("ws://localhost:9009/ws");
-
-ws.onopen = () => {
-  // 订阅 BTC 1m K 线
-  ws.send(JSON.stringify({
-    action: "subscribe",
-    symbol: "BTC/USDT:USDT",
-    timeframe: "1m"
-  }));
-};
-
-ws.onmessage = (event) => {
-  const msg = JSON.parse(event.data);
-  if (msg.type === "kline") {
-    console.log(msg.data); // 最新 K 线数组
-  }
-};
- -
-
-
- - -
-

时间周期参考

-

以下是完整的周期对照表：

- - - - - - - -
基础周期合成衍生周期
1m2m, 3m, 4m, 5m, 10m, 15m, 20m, 25m, 30m, 45m
1h2h, 3h, 4h, 5h, 6h, 7h, 8h, 9h, 10h, 11h, 12h, 16h, 20h
1d2d, 3d, 4d, 5d, 6d
1w2w, 3w
- -

衍生周期由对应基础周期的 K 线通过 OHLCV 聚合合成，查询方式与基础周期完全一致。

-
- -
- Chan Data Provider — Built with FastAPI + ccxt + pandas -
- -
- - - - diff --git a/data_provider/config.json b/data_provider/config.json deleted file mode 100644 index 90a7ff3..0000000 --- a/data_provider/config.json +++ /dev/null @@ -1,31 +0,0 @@ -{ - "exchange": "binance", - "symbols": [ - "BTC/USDT:USDT", - "ETH/USDT:USDT", - "SOL/USDT:USDT", - "XAU/USDT:USDT", - "XAG/USDT:USDT", - "SAGA/USDT:USDT", - "CL/USDT:USDT", - "ZEC/USDT:USDT", - "XRP/USDT:USDT", - "DOGE/USDT:USDT", - "BNB/USDT:USDT", - "SUI/USDT:USDT", - "BILL/USDT:USDT", - "BZ/USDT:USDT", - "LAB/USDT:USDT", - "TON/USDT:USDT", - "CRCL/USDT:USDT", - "SNDK/USDT:USDT", - "1000PEPE/USDT:USDT", - "CHIP/USDT:USDT" - ], - "start_time": "2024-01-01T00:00:00Z", - "start_time_per_tf": { - "1m": "2026-01-01T00:00:00Z" - }, - "timeframes": ["1m", "1h", "1d", "1w"], - "data_dir": "./data" -} diff --git a/data_provider/docker-compose.yml b/data_provider/docker-compose.yml deleted file mode 100644 index 72bfe85..0000000 --- a/data_provider/docker-compose.yml +++ /dev/null @@ -1,15 +0,0 @@ -services: - data_provider: - build: . - container_name: data-provider - restart: unless-stopped - environment: - CONFIG_PATH: /app/config.json - UVICORN_HOST: 0.0.0.0 - UVICORN_PORT: "80" - volumes: - - ./config.json:/app/config.json:ro - - ./data:/app/data - ports: - - "80:80" - diff --git a/data_provider/homepage.html b/data_provider/homepage.html deleted file mode 100644 index 70a8628..0000000 --- a/data_provider/homepage.html +++ /dev/null @@ -1,425 +0,0 @@ - - - - - - - - -Chan 数据提供商 - - - -
- -
-
- -
-

Chan 数据提供商

-
加密货币 K 线数据服务
-
-
-
—
-
- - -
-
服务状态
加载中...
-
交易所
—
-
交易对
—
-
基础周期
—
-
衍生周期
—
-
数据就绪
—
-
- - -
- 📖 API 文档 - 📋 Swagger UI - 📄 ReDoc - 📊 BTC 1m 示例 - 📊 ETH 4h 示例 -
- - -
-
-

📈 交易对

- 0 -
-
-
加载中...
-
-
- - -
-
-

⏱ 时间周期

- 0 -
-
-
基础周期（交易所直拉）
-
-
衍生周期（内存合成）
-
-
-
- - -
-
-

⚡ 快速查询

-
-
-
-
-
交易对
- -
-
-
周期
- -
-
-
数量
- -
- -
- -
-
- -
- -
- Chan Data Provider  ·  Built with FastAPI + ccxt + pandas  ·  - 更新于 — -
- - - - diff --git a/data_provider/main.py b/data_provider/main.py deleted file mode 100644 index fe43a1b..0000000 --- a/data_provider/main.py +++ /dev/null @@ -1,2014 +0,0 @@ -""" -Chan 数据服务：用 ccxt 从交易所拉取 K 线，内存缓存 + CSV 落盘； -后台线程定期增量刷新，断线时记录 resume_since 以免漏 K； -配置中的基础周期（如 1m/1h）可合成 DERIVED_TIMEFRAME_PLAN 中的衍生周期。 -""" -import asyncio -import csv -import json -import logging -import os -import threading -import time -from contextlib import asynccontextmanager -from datetime import datetime, timezone -from pathlib import Path -from typing import Dict, Iterable, List, Optional - -import ccxt # type: ignore -import ccxt.pro as ccxt_pro # type: ignore -import pandas as pd # type: ignore -from fastapi import FastAPI, HTTPException, Query, WebSocket, WebSocketDisconnect -from fastapi.middleware.cors import CORSMiddleware -from fastapi.responses import HTMLResponse -import uvicorn -from technical.util import resample_to_interval - -from onchain_metrics import ( - OnchainMetricsManager, - api_keys_from_env, - create_onchain_router, -) - -# docker compose logs --tail=200 -# docker compose down && docker compose build --no-cache && docker compose up -d - -# 基础周期枚举顺序（用于衍生周期展示顺序）；仅允许集合内周期作为交易所直接拉取的 tf -TIMEFRAME_ORDER = ["1m", "1h", "1d", "1w"] -ALLOWED_TIMEFRAMES = set(TIMEFRAME_ORDER) -# 各基础周期一根 K 线的毫秒长度（用于历史分页与断线回退） -TIMEFRAME_TO_MS: Dict[str, int] = { - "1m": 60_000, - "1h": 3_600_000, - "1d": 86_400_000, - "1w": 604_800_000, -} -# 内存最大缓存的 K 线数量（超出部分从 CSV 按需读取，不限制则不限量） -MAX_CANDLES_IN_MEMORY: Dict[str, int] = { - "1m": 129_600, # 3 个月（3 * 30 * 24 * 60） -} -# 每个基础周期可派生出的合成周期列表（由该基础周期 K 线 resample 得到） -DERIVED_TIMEFRAME_PLAN: Dict[str, List[str]] = { - "1m": ["2m", "3m", "4m", "5m", "10m", "15m", "20m", "25m", "30m", "45m"], - "1h": ["2h", "3h", "4h", "5h", "6h", "7h", "8h", "9h", "10", "11h", "12h", "16h", "20h"], - "1d": ["2d", "3d", "4d", "5d", "6d"], - "1w": ["2w", "3w"], -} -CSV_FIELDNAMES = ["timestamp", "datetime", "open", "high", "low", "close", "volume"] -DEFAULT_LIMIT = 500 -RECENT_CANDLE_LIMIT = 10 -RECENT_FETCH_INTERVAL = 5 # 后台刷新循环休眠秒数 -PERSIST_INTERVAL = 600 # 全量落盘周期（秒） -WS_UPDATE_CANDLE_COUNT = 2 # WebSocket 增量推送最近 K 线根数 - - -logger = logging.getLogger("data_provider") -logging.basicConfig( - level=logging.INFO, - format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", -) - - -def to_utc_iso(timestamp_ms: int) -> str: - """将毫秒时间戳格式化为 UTC ISO 字符串（末尾 Z）。""" - dt = datetime.fromtimestamp(timestamp_ms / 1000, tz=timezone.utc) - return dt.isoformat().replace("+00:00", "Z") - - -def parse_timestamp(value: Optional[object]) -> Optional[int]: - """解析查询参数中的时间为 UTC 毫秒时间戳；支持数字或 ISO 字符串。""" - if value is None: - return None - if isinstance(value, (int, float)): - return int(value) - if isinstance(value, str): - text = value.strip() - if not text: - return None - if text.isdigit(): - return int(text) - if text.endswith("Z"): - text = text[:-1] + "+00:00" - try: - dt = datetime.fromisoformat(text) - except ValueError as exc: # pragma: no cover - informative logging - raise ValueError(f"无法解析时间字符串: {value}") from exc - if dt.tzinfo is None: - dt = dt.replace(tzinfo=timezone.utc) - else: - dt = dt.astimezone(timezone.utc) - return int(dt.timestamp() * 1000) - raise ValueError(f"不支持的时间格式: {value}") - - -def candle_to_dict(candle: Iterable[float]) -> Dict[str, float]: - """ccxt OHLCV 单根 [ts, o, h, l, c, v] 转为内部字典结构。""" - ts = int(candle[0]) - return { - "timestamp": ts, - "datetime": to_utc_iso(ts), - "open": float(candle[1]), - "high": float(candle[2]), - "low": float(candle[3]), - "close": float(candle[4]), - "volume": float(candle[5]), - } - - -def timeframe_to_minutes(tf: str) -> Optional[int]: - """将如 15m、2h 转为「分钟数」，供 resample 与衍生周期计算。""" - if not tf: - return None - unit = tf[-1] - try: - value = int(tf[:-1]) - except ValueError: - return None - multiplier = { - "m": 1, - "h": 60, - "d": 1_440, - "w": 10_080, - }.get(unit) - if multiplier is None: - return None - return value * multiplier - - -class WebSocketManager: - """管理 WebSocket 连接及订阅，线程安全地广播 K 线更新。""" - - def __init__(self) -> None: - self._subscriptions: Dict[tuple, set] = {} - self._async_lock = asyncio.Lock() - self._loop: Optional[asyncio.AbstractEventLoop] = None - - def set_loop(self, loop: asyncio.AbstractEventLoop) -> None: - self._loop = loop - - async def connect(self, ws: WebSocket) -> None: - await ws.accept() - logger.info("WebSocket 客户端已连接") - - async def disconnect(self, ws: WebSocket) -> None: - async with self._async_lock: - for key in list(self._subscriptions): - self._subscriptions[key].discard(ws) - if not self._subscriptions[key]: - del self._subscriptions[key] - logger.info("WebSocket 客户端已断开") - - async def subscribe(self, ws: WebSocket, symbol: str, timeframe: str) -> None: - key = (symbol, timeframe) - async with self._async_lock: - self._subscriptions.setdefault(key, set()).add(ws) - logger.info("WebSocket 订阅: %s %s", symbol, timeframe) - - async def unsubscribe(self, ws: WebSocket, symbol: str, timeframe: str) -> None: - key = (symbol, timeframe) - async with self._async_lock: - if key in self._subscriptions: - self._subscriptions[key].discard(ws) - if not self._subscriptions[key]: - del self._subscriptions[key] - - def has_subscribers(self, symbol: str, timeframe: str) -> bool: - """非异步快速检查（供同步线程调用）。""" - return bool(self._subscriptions.get((symbol, timeframe))) - - async def broadcast( - self, symbol: str, timeframe: str, candles: List[Dict], msg_type: str = "kline", - ) -> None: - key = (symbol, timeframe) - async with self._async_lock: - subscribers = list(self._subscriptions.get(key, set())) - if not subscribers: - return - message = json.dumps( - {"type": msg_type, "symbol": symbol, "timeframe": timeframe, "data": candles}, - ensure_ascii=False, - ) - dead: list = [] - for ws in subscribers: - try: - await ws.send_text(message) - except Exception: - dead.append(ws) - if dead: - async with self._async_lock: - for ws in dead: - self._subscriptions.get(key, set()).discard(ws) - - def broadcast_from_thread( - self, symbol: str, timeframe: str, candles: List[Dict], msg_type: str = "kline", - ) -> None: - """供同步后台线程调用，将广播提交到 asyncio 事件循环。""" - if self._loop is None or self._loop.is_closed(): - return - asyncio.run_coroutine_threadsafe( - self.broadcast(symbol, timeframe, candles, msg_type), - self._loop, - ) - - -class DataProvider: - """封装交易所连接、本地 CSV、内存缓存、断线恢复与衍生周期聚合。""" - - def __init__(self, config_path: Path) -> None: - self.config_path = config_path - self.config = self._load_config() - self.exchange_name: str = self.config["exchange"] - self.symbols: List[str] = self._load_symbols(self.config) - self.timeframes: List[str] = self._validate_timeframes(self.config.get("timeframes")) - self.data_dir = Path(self.config.get("data_dir", "./data")).expanduser() - start = parse_timestamp(self.config.get("start_time")) - if start is None: - raise ValueError("配置文件必须包含 start_time 字段") - self.start_time_ms: int = start - # 按周期独立 start_time（从 start_time_per_tf 字段读取），未配置则回退到全局 start_time - self.start_time_map: Dict[str, int] = {} - start_per_tf = self.config.get("start_time_per_tf", {}) - if isinstance(start_per_tf, dict): - for tf, ts_str in start_per_tf.items(): - tf_start = parse_timestamp(ts_str) - if tf_start is not None: - self.start_time_map[tf] = tf_start - self.exchange = self._init_exchange() - self.data: Dict[str, Dict[str, List[Dict[str, float]]]] = { - symbol: {tf: [] for tf in self.timeframes} for symbol in self.symbols - } - # 衍生周期 -> 用于合成的交易所基础周期（每个衍生只对应一个 base） - self.derived_map: Dict[str, str] = {} - for base_tf in self.timeframes: - for derived_tf in DERIVED_TIMEFRAME_PLAN.get(base_tf, []): - self.derived_map.setdefault(derived_tf, base_tf) - # 衍生周期展示顺序：按 TIMEFRAME_ORDER 中的基础周期依次展开 - derived_order: List[str] = [] - for base_tf in TIMEFRAME_ORDER: - if base_tf not in self.timeframes: - continue - for derived_tf in DERIVED_TIMEFRAME_PLAN.get(base_tf, []): - if derived_tf in self.derived_map and derived_tf not in derived_order: - derived_order.append(derived_tf) - self.available_timeframes: List[str] = list(self.timeframes) + derived_order - self._lock = threading.RLock() - self._ready = threading.Event() - self._stop_event = threading.Event() - self._persist_thread: Optional[threading.Thread] = None - self._backfill_thread: Optional[threading.Thread] = None - self._watch_tasks: List = [] # asyncio Task — 每个基础周期一个 watch - self._watch_loop_task = None # master watch 协程 - self._cold_start_done = False - # 记录断线后需要从哪个 since 重新拉取（symbol -> timeframe -> since_ms） - self._resume_since: Dict[str, Dict[str, int]] = {} - # 恢复点持久化文件 - self._resume_file: Path = self.data_dir / "resume_since.json" - # 尝试加载历史恢复点 - self._load_resume_since() - self._update_callbacks: List = [] - # 衍生品数据: 与 K 线一样的模式 — symbol -> List[Dict], 按 timestamp 去重 - self._derivatives: Dict[str, List[Dict]] = {symbol: [] for symbol in self.symbols} - self._derivatives_dir: Path = self.data_dir / "derivatives" - - def _load_config(self) -> Dict[str, object]: - """读取 JSON 配置文件。""" - if not self.config_path.exists(): - raise FileNotFoundError(f"未找到配置文件: {self.config_path}") - with self.config_path.open("r", encoding="utf-8") as fp: - return json.load(fp) - - def _load_symbols(self, config: Dict[str, object]) -> List[str]: - """从 symbols 列表、逗号分隔字符串或单字段 symbol 解析交易对，去重保序。""" - raw_symbols: List[str] = [] - symbols_value = config.get("symbols") - if isinstance(symbols_value, list): - raw_symbols = [str(item).strip() for item in symbols_value if isinstance(item, str) and item.strip()] - elif isinstance(symbols_value, str) and symbols_value.strip(): - raw_symbols = [item.strip() for item in symbols_value.split(",") if item.strip()] - symbol_single = config.get("symbol") - if not raw_symbols and isinstance(symbol_single, str) and symbol_single.strip(): - raw_symbols = [symbol_single.strip()] - if not raw_symbols: - raise ValueError("配置文件必须提供 symbols（列表或逗号分隔字符串）或 symbol 字段") - unique: List[str] = [] - for item in raw_symbols: - if item not in unique: - unique.append(item) - return unique - - def _get_start_time(self, timeframe: str) -> int: - """获取指定周期的 start_time（毫秒）。优先使用 start_time_per_tf，否则回退全局。""" - return self.start_time_map.get(timeframe, self.start_time_ms) - - def _validate_timeframes(self, configured: Optional[Iterable[str]]) -> List[str]: - """校验周期在允许集合内；未配置则默认 TIMEFRAME_ORDER 全部；顺序优先按 TIMEFRAME_ORDER。""" - if not configured: - return list(TIMEFRAME_ORDER) - invalid = [tf for tf in configured if tf not in ALLOWED_TIMEFRAMES] - if invalid: - raise ValueError(f"不支持的时间周期: {invalid}. 允许值: {sorted(ALLOWED_TIMEFRAMES)}") - unique = [] - seen = set() - for tf in TIMEFRAME_ORDER: - if tf in configured and tf not in seen: - unique.append(tf) - seen.add(tf) - for tf in configured: - if tf not in seen: - unique.append(tf) - seen.add(tf) - return unique - - def _init_exchange(self): - """实例化两个交易所实例：REST 用 ccxt，WebSocket 用 ccxt.pro。""" - # REST 实例：供冷启动回填、历史补数、定时落盘等同步代码使用 - if not hasattr(ccxt, self.exchange_name): - raise ValueError(f"不支持的交易所: {self.exchange_name}") - exchange_class = getattr(ccxt, self.exchange_name) - exchange = exchange_class({"enableRateLimit": True}) - if exchange.id == "binance": - exchange.options.setdefault("defaultType", "future") - exchange.load_markets() - logger.info("已初始化交易所 %s (ccxt REST)", exchange.id) - - # WebSocket 实例：供实时 K 线监听使用 - if not hasattr(ccxt_pro, self.exchange_name): - raise ValueError(f"不支持的交易所: {self.exchange_name} (ccxt.pro)") - ws_exchange_class = getattr(ccxt_pro, self.exchange_name) - self._ws_exchange = ws_exchange_class({"newUpdates": True}) - if self._ws_exchange.id == "binance": - self._ws_exchange.options.setdefault("defaultType", "future") - logger.info("已初始化交易所 %s (ccxt.pro, WebSocket 已启用)", self._ws_exchange.id) - - return exchange - - def _record_resume_for_timeframe(self, tf: str) -> None: - """WebSocket 断线时为该周期所有 symbol 记录恢复点，避免 gap。""" - for symbol in self.symbols: - with self._lock: - current = self.data.get(symbol, {}).get(tf, []) - if current: - last_ts = int(current[-1]["timestamp"]) - else: - last_ts = self._get_start_time(tf) - tf_ms = TIMEFRAME_TO_MS[tf] - since_ms = max(self._get_start_time(tf), last_ts - tf_ms) - existing = self._get_resume_since(symbol, tf) - if existing is None: - self._set_resume_since(symbol, tf, since_ms) - - async def _backfill_resume_data(self) -> None: - """启动 WebSocket watch 前的历史补数，通过 REST 补齐 resume_since gap。""" - loop = asyncio.get_running_loop() - for symbol in self.symbols: - for tf in self.timeframes: - if self._stop_event.is_set(): - return - resume_since = self._get_resume_since(symbol, tf) - if resume_since is None: - continue - logger.info("WebSocket 启动前补数: %s %s since=%s", symbol, tf, - to_utc_iso(resume_since)) - try: - history = await loop.run_in_executor( - None, self._fetch_history, symbol, tf, resume_since, - ) - with self._lock: - current = self.data.setdefault(symbol, {}).get(tf, []) - merged = self._merge_candles(tf, current, history) - self.data[symbol][tf] = merged - self._notify_update(symbol, tf) - self._clear_resume_since(symbol, tf) - self._trim(symbol, tf) - except Exception as exc: - logger.error("启动前补数失败 %s %s: %s", symbol, tf, exc) - - async def _watch_single_tf(self, tf: str) -> None: - """单个基础周期的 WebSocket 监听协程。 - - 循环调用 watch_ohlcv_for_symbols，新 K 线到达后从 exchange.ohlcvs 缓存 - 读取最新数据合并到内存并广播。 - """ - pairs = [[symbol, tf] for symbol in self.symbols] - backoff = 2.0 - - while not self._stop_event.is_set(): - try: - await self._ws_exchange.watch_ohlcv_for_symbols(pairs) - - # watch_ohlcv_for_symbols 更新了 exchange.ohlcvs 缓存， - # 从缓存中读取每个 symbol 的最新 candle 合并到本地 - for symbol in self.symbols: - ohlcv_cache = self._ws_exchange.ohlcvs.get(symbol, {}).get(tf) - if ohlcv_cache is None: - continue - cached = list(ohlcv_cache) - if not cached: - continue - - with self._lock: - current = self.data.setdefault(symbol, {}).get(tf, []) - if current and current[-1]["timestamp"] >= cached[-1][0]: - continue # 没有新数据 - merged = self._merge_candles(tf, current, cached) - self.data[symbol][tf] = merged - - self._notify_update(symbol, tf) - self._trim(symbol, tf) - - backoff = 2.0 # 成功后重置 - - except ccxt.NetworkError as exc: - logger.warning("WebSocket 网络错误 (tf=%s): %s，%.1fs 后重连", tf, exc, backoff) - self._record_resume_for_timeframe(tf) - await asyncio.sleep(backoff) - backoff = min(backoff * 1.5, 60.0) - - except ccxt.BaseError as exc: - logger.error("WebSocket 交易所错误 (tf=%s): %s，%.1fs 后重连", tf, exc, backoff) - self._record_resume_for_timeframe(tf) - await asyncio.sleep(backoff) - backoff = min(backoff * 1.5, 60.0) - - except asyncio.CancelledError: - logger.info("WebSocket 监听 (tf=%s) 被取消", tf) - break - - except Exception: - logger.exception("WebSocket 监听 (tf=%s) 未预期错误", tf) - self._record_resume_for_timeframe(tf) - await asyncio.sleep(backoff) - backoff = min(backoff * 1.5, 60.0) - - logger.info("WebSocket 监听 (tf=%s) 已退出", tf) - - async def _watch_loop(self) -> None: - """主 WebSocket 监听协程：先启动 watch 任务，再并发补历史 gap。""" - for tf in self.timeframes: - task = asyncio.create_task(self._watch_single_tf(tf), name=f"watch-{tf}") - self._watch_tasks.append(task) - - logger.info("WebSocket 监听已启动: %d 个时间周期, %d 个任务", - len(self.timeframes), len(self._watch_tasks)) - - # 补历史数据与 watch 任务并发进行，不阻塞实时数据接收 - try: - await self._backfill_resume_data() - except Exception: - logger.exception("启动前补数失败") - - try: - await asyncio.gather(*self._watch_tasks, return_exceptions=True) - except asyncio.CancelledError: - pass - finally: - logger.info("所有 WebSocket 监听任务已结束") - - def _data_file_path(self, symbol: str, timeframe: str) -> Path: - """单交易对单周期的 CSV 路径：data_dir/tf/exchange_symbol_tf.csv。""" - symbol_safe = symbol.replace("/", "_").replace(":", "_") - return self.data_dir / timeframe / f"{self.exchange.id}_{symbol_safe}_{timeframe}.csv" - - def _derivatives_file_path(self, symbol: str) -> Path: - """衍生品 CSV 路径：data_dir/derivatives/exchange_symbol.csv。""" - symbol_safe = symbol.replace("/", "_").replace(":", "_") - return self._derivatives_dir / f"{self.exchange.id}_{symbol_safe}.csv" - - def _load_derivatives_local(self, symbol: str) -> List[Dict]: - """加载本地衍生品历史数据。""" - path = self._derivatives_file_path(symbol) - if not path.exists(): - return [] - loaded = [] - with path.open("r", encoding="utf-8", newline="") as fp: - for row in csv.DictReader(fp): - try: - loaded.append({ - "timestamp": int(row["timestamp"]), - "datetime": row.get("datetime", ""), - "funding_rate": float(row.get("funding_rate", 0)), - "open_interest": float(row.get("open_interest", 0)), - "oi_change_pct": float(row.get("oi_change_pct", 0)) if row.get("oi_change_pct") else None, - "basis": float(row.get("basis", 0)) if row.get("basis") else None, - }) - except (KeyError, ValueError): - continue - loaded.sort(key=lambda item: item["timestamp"]) - return loaded - - def _fetch_derivatives_sync(self, symbol: str) -> Optional[List[Dict]]: - """用 REST 获取单个币对的费率、OI 和基差（同步，供后台线程调用）。 - 返回单条记录列表，与 K 线 fetcher 返回格式一致。""" - try: - funding = self.exchange.fetch_funding_rate(symbol) - oi = self.exchange.fetch_open_interest(symbol) - ticker = self.exchange.fetch_ticker(symbol) - except Exception as e: - logger.warning("衍生品获取失败 %s: %s", symbol, e) - return None - - now_ms = int(time.time() * 1000) - record = { - "timestamp": now_ms, - "datetime": to_utc_iso(now_ms), - "funding_rate": float(funding.get("fundingRate", 0)) if funding else 0, - "open_interest": float(oi.get("openInterestAmount", 0)) if oi else 0, - "oi_change_pct": None, - "basis": None, - } - - # OI 变化 (与上一条对比) - with self._lock: - history = self._derivatives.get(symbol, []) - if history: - prev = history[-1] - prev_oi = prev.get("open_interest", 0) - if prev_oi > 0 and record["open_interest"] > 0: - record["oi_change_pct"] = round( - (record["open_interest"] - prev_oi) / prev_oi * 100, 2 - ) - - # 基差 (期货-现货)/现货 - if ticker: - spot_symbol = symbol.split(":")[0] - try: - spot_ticker = self.exchange.fetch_ticker(spot_symbol) - future_price = float(ticker.get("last", 0)) - spot_price = float(spot_ticker.get("last", 0)) - if spot_price > 0 and future_price > 0: - record["basis"] = round( - (future_price - spot_price) / spot_price * 100, 2 - ) - except Exception: - pass - - return [record] - - def _merge_derivatives( - self, - base: List[Dict], - new_records: List[Dict], - ) -> List[Dict]: - """按 timestamp 去重合并衍生品记录，新数据覆盖同时间戳旧数据。与 _merge_candles 模式一致。""" - merged = {entry["timestamp"]: entry for entry in base} - for record in new_records: - merged[record["timestamp"]] = record - return list(sorted(merged.values(), key=lambda item: item["timestamp"])) - - def _write_derivatives_to_disk(self, symbol: str, records: List[Dict]) -> None: - """将衍生品历史数据写入 CSV。与 _write_to_disk 模式一致：先写 tmp 再 replace。""" - if not records: - return - path = self._derivatives_file_path(symbol) - path.parent.mkdir(parents=True, exist_ok=True) - fieldnames = ["timestamp", "datetime", "funding_rate", "open_interest", - "oi_change_pct", "basis"] - tmp_path = path.with_suffix(".tmp") - try: - with tmp_path.open("w", encoding="utf-8", newline="") as fp: - writer = csv.DictWriter(fp, fieldnames=fieldnames, extrasaction="ignore") - writer.writeheader() - for r in records: - row = {} - for k in fieldnames: - val = r.get(k, "") - row[k] = "" if val is None else val - writer.writerow(row) - tmp_path.replace(path) - except Exception as e: - logger.error("衍生品落盘失败 %s: %s", symbol, e) - if tmp_path.exists(): - tmp_path.unlink() - - def _load_local(self, symbol: str, timeframe: str) -> List[Dict[str, float]]: - """启动时从磁盘加载已有 K 线，损坏行跳过，按时间排序。""" - path = self._data_file_path(symbol, timeframe) - if not path.exists(): - return [] - loaded: List[Dict[str, float]] = [] - with path.open("r", encoding="utf-8", newline="") as fp: - reader = csv.DictReader(fp) - for row in reader: - try: - loaded.append( - { - "timestamp": int(row["timestamp"]), - "datetime": row.get("datetime") or to_utc_iso(int(row["timestamp"])), - "open": float(row["open"]), - "high": float(row["high"]), - "low": float(row["low"]), - "close": float(row["close"]), - "volume": float(row["volume"]), - } - ) - except (KeyError, ValueError): - logger.warning("忽略损坏的行: %s", row) - loaded.sort(key=lambda item: item["timestamp"]) - logger.info("交易对 %s 时间周期 %s 加载本地K线数量: %s", symbol, timeframe, len(loaded)) - return loaded - - def _merge_candles( - self, - timeframe: str, - base: List[Dict[str, float]], - new_candles: Iterable[Iterable[float]], - ) -> List[Dict[str, float]]: - """按 timestamp 去重合并，新数据覆盖同时间戳旧数据。""" - merged = {entry["timestamp"]: entry for entry in base} - for candle in new_candles: - entry = candle_to_dict(candle) - merged[entry["timestamp"]] = entry - ordered = list(sorted(merged.values(), key=lambda item: item["timestamp"])) - logger.debug("时间周期 %s 合并后K线数量: %s", timeframe, len(ordered)) - return ordered - - def _trim(self, symbol: str, timeframe: str) -> None: - """将内存缓存裁剪到 MAX_CANDLES_IN_MEMORY 上限（全量数据已在 CSV 中）。""" - max_count = MAX_CANDLES_IN_MEMORY.get(timeframe) - if max_count is None: - return - sym_data = self.data.get(symbol) - if sym_data is None: - return - candles = sym_data.get(timeframe) - if candles is None or len(candles) <= max_count: - return - sym_data[timeframe] = candles[-max_count:] - - def _write_to_disk(self, symbol: str, timeframe: str, data: List[Dict[str, float]]) -> None: - """先写临时文件再 replace，避免写入中断导致 CSV 损坏。""" - path = self._data_file_path(symbol, timeframe) - path.parent.mkdir(parents=True, exist_ok=True) - tmp_path = path.with_suffix(path.suffix + ".tmp") - try: - with tmp_path.open("w", encoding="utf-8", newline="") as fp: - writer = csv.DictWriter(fp, fieldnames=CSV_FIELDNAMES) - writer.writeheader() - writer.writerows(data) - os.replace(tmp_path, path) - finally: - if tmp_path.exists(): - try: - tmp_path.unlink() - except OSError: - pass - logger.info("交易对 %s 时间周期 %s 已写入磁盘 (%s 根K线)", symbol, timeframe, len(data)) - - def _fetch_history(self, symbol: str, timeframe: str, since_ms: int) -> List[List[float]]: - """从 since_ms 分页拉取直到接近当前时间。 - - 当 since_ms 早于交易所数据保留期时，先从最新数据往回探测有效起点， - 再向后分页补齐。""" - results: List[List[float]] = [] - limit = 1500 - now_ms = self.exchange.milliseconds() - tf_ms = TIMEFRAME_TO_MS[timeframe] - - # 先尝试直接拉取 - first_batch = self._fetch_ohlcv_retry(symbol, timeframe, since=since_ms, limit=limit) - - # 若返回空，或首根时间远晚于 since（交易所忽略了过老的 since），启用反向探测 - gap_ms = tf_ms * limit * 2 - first_skipped = ( - not first_batch - or (first_batch[0][0] - since_ms > gap_ms and first_batch[-1][0] >= now_ms - tf_ms * 10) - ) - if first_skipped: - logger.info("since 过老被交易所跳过，启用反向探测 %s %s (since=%s)", symbol, timeframe, to_utc_iso(since_ms)) - # 反向探测：从当前时间往前翻页，找到实际最早可用的 K 线 - all_collected: List[List[float]] = [] - search_since = max(0, now_ms - tf_ms * limit) - search_rounds = 0 - max_search = 200 - earliest_ts = None - while search_rounds < max_search: - search_rounds += 1 - batch = self._fetch_ohlcv_retry(symbol, timeframe, since=search_since, limit=limit) - if not batch: - break - all_collected = batch + all_collected - first_ts = batch[0][0] - if first_ts == search_since or first_ts <= since_ms: - earliest_ts = first_ts - break - search_since = max(0, first_ts - tf_ms * limit) - time.sleep(self.exchange.rateLimit / 1000 if self.exchange.rateLimit else 0.2) - if earliest_ts is None and all_collected: - earliest_ts = all_collected[0][0] - if earliest_ts is not None: - logger.info("反向探测找到最早 %s: %s (%s)", symbol, to_utc_iso(earliest_ts), timeframe) - # 用收集到的数据作为起点，继续向前分页到 now - results = all_collected - if results: - last_ts = results[-1][0] - if last_ts < now_ms - tf_ms: - extra = self._fetch_ohlcv_paginate(symbol, timeframe, last_ts + tf_ms, now_ms, limit) - results.extend(extra) - logger.info("交易对 %s 时间周期 %s 拉取历史K线数量: %s", symbol, timeframe, len(results)) - return results - - # 直接拉取成功，正常向后分页 - results = list(first_batch) - last_ts = results[-1][0] - if last_ts < now_ms - tf_ms: - extra = self._fetch_ohlcv_paginate(symbol, timeframe, last_ts + tf_ms, now_ms, limit) - results.extend(extra) - logger.info("交易对 %s 时间周期 %s 拉取历史K线数量: %s", symbol, timeframe, len(results)) - return results - - def _fetch_ohlcv_retry(self, symbol: str, timeframe: str, since: int, limit: int) -> List[List[float]]: - """带重试的单次 fetch_ohlcv，限频/网络错误时等待重试。""" - max_attempts = 5 - for attempt in range(max_attempts): - try: - return self.exchange.fetch_ohlcv(symbol, timeframe=timeframe, since=since, limit=limit) - except ccxt.RateLimitExceeded as exc: - wait_s = (self.exchange.rateLimit / 1000 if self.exchange.rateLimit else 1) * (attempt + 1) - logger.warning("触发频率限制，等待 %.1fs: %s", wait_s, exc) - time.sleep(wait_s) - except ccxt.BaseError as exc: - logger.error("拉取K线失败 (%s, %s, since=%s): %s (attempt %s/%s)", - timeframe, symbol, since, exc, attempt + 1, max_attempts) - time.sleep(3 * (attempt + 1)) - logger.error("拉取K线失败，已达最大重试次数: %s %s", symbol, timeframe) - return [] - - def _fetch_ohlcv_paginate(self, symbol: str, timeframe: str, start_ms: int, end_ms: int, limit: int) -> List[List[float]]: - """从 start_ms 向后分页拉取到 end_ms。""" - results: List[List[float]] = [] - tf_ms = TIMEFRAME_TO_MS[timeframe] - fetch_since = start_ms - max_rounds = 5000 - rounds = 0 - while fetch_since < end_ms and rounds < max_rounds: - rounds += 1 - candles = self._fetch_ohlcv_retry(symbol, timeframe, since=fetch_since, limit=limit) - if not candles: - break - results.extend(candles) - last_ts = candles[-1][0] - fetch_since = last_ts + tf_ms - if last_ts >= end_ms - tf_ms: - break - time.sleep(self.exchange.rateLimit / 1000 if self.exchange.rateLimit else 0.2) - return results - - def initialize(self) -> None: - """快速启动：仅加载本地磁盘已有数据到内存，立即设 _ready 让服务可用。后台再补拉交易所数据。""" - logger.info("开始初始化数据提供商（仅加载本地数据）") - for symbol in self.symbols: - for timeframe in self.timeframes: - existing = self._load_local(symbol, timeframe) - with self._lock: - self.data.setdefault(symbol, {})[timeframe] = existing - self._trim(symbol, timeframe) - logger.info("交易对 %s 时间周期 %s 已加载本地K线: %s 根", symbol, timeframe, len(existing)) - self._ready.set() - logger.info("数据初始化完成，服务已就绪（后台将补拉交易所数据）") - - def _cold_start_backfill(self) -> None: - """后台一次性回填：从交易所拉取每个 symbol/timeframe 的缺失历史，合并后写盘。""" - logger.info("开始后台回填交易所数据") - for symbol in self.symbols: - for timeframe in self.timeframes: - if self._stop_event.is_set(): - return - with self._lock: - existing = list(self.data.get(symbol, {}).get(timeframe, [])) - tf_ms = TIMEFRAME_TO_MS[timeframe] - last_ts = existing[-1]["timestamp"] if existing else None - - # 计算 fetch_since：优先从数据末尾补，但若数据量极少 - #（刷新线程抢先写入的少量近期数据），则回退到 start_time 做全量回填 - if last_ts is not None: - if len(existing) > RECENT_CANDLE_LIMIT * 3: - # 已有足够历史数据，仅增量补拉 - fetch_since = (existing[-2]["timestamp"] if len(existing) >= 2 - else max(0, existing[-1]["timestamp"] - tf_ms)) - else: - # 只有刷新线程抢先写入的少量数据，从 start_time 全量回填 - fetch_since = self._get_start_time(timeframe) - logger.info("数据量不足 (%s 根)，从 start_time 回填 %s %s", - len(existing), symbol, timeframe) - else: - fetch_since = self._get_start_time(timeframe) - try: - history = self._fetch_history(symbol, timeframe, fetch_since) - # 再次读取最新数据后合并，避免覆盖并发刷新线程写入的更新 - with self._lock: - current = list(self.data.get(symbol, {}).get(timeframe, [])) - merged = self._merge_candles(timeframe, current, history) - self.data[symbol][timeframe] = merged - self._write_to_disk(symbol, timeframe, merged) - self._trim(symbol, timeframe) - self._notify_update(symbol, timeframe) - logger.info("回填完成 %s %s: +%s 根新K线", symbol, timeframe, len(history)) - except Exception as exc: - logger.error("回填失败 %s %s: %s", symbol, timeframe, exc) - self._set_resume_since(symbol, timeframe, fetch_since) - logger.info("后台回填全部完成") - self._cold_start_done = True - - def resample_df(self, df: pd.DataFrame, interval: int) -> pd.DataFrame: - """将基础周期 DataFrame 聚合为 interval 分钟周期（freqtrade technical.util）。""" - return resample_to_interval(df, interval) - - def _save_resume_since(self) -> None: - """将断线恢复点持久化到 resume_since.json（原子替换）。""" - path = self._resume_file - path.parent.mkdir(parents=True, exist_ok=True) - tmp_path = path.with_suffix(path.suffix + ".tmp") - with self._lock: - snapshot = { - symbol: {tf: int(since) for tf, since in tf_map.items()} - for symbol, tf_map in self._resume_since.items() - } - try: - with tmp_path.open("w", encoding="utf-8") as fp: - json.dump(snapshot, fp, ensure_ascii=False, separators=(",", ":")) - os.replace(tmp_path, path) - finally: - if tmp_path.exists(): - try: - tmp_path.unlink() - except OSError: - pass - logger.debug("恢复点已保存到磁盘: %s", path) - - def _load_resume_since(self) -> None: - """启动时加载恢复点；与内存合并时取更早的 since，避免漏拉。""" - path = self._resume_file - if not path.exists(): - return - try: - with path.open("r", encoding="utf-8") as fp: - raw = json.load(fp) - except Exception as exc: - logger.warning("恢复点文件读取失败，忽略: %s (%s)", path, exc) - return - if not isinstance(raw, dict): - logger.warning("恢复点文件格式错误，忽略: %s", path) - return - loaded: Dict[str, Dict[str, int]] = {} - for symbol, tf_map in raw.items(): - if not isinstance(tf_map, dict): - continue - per_symbol: Dict[str, int] = {} - for timeframe, since in tf_map.items(): - try: - per_symbol[str(timeframe)] = int(since) - except Exception: - continue - if per_symbol: - loaded[str(symbol)] = per_symbol - if not loaded: - return - with self._lock: - # 合并为更早的 since，避免遗漏 - for symbol, tf_map in loaded.items(): - cur = self._resume_since.setdefault(symbol, {}) - for timeframe, since in tf_map.items(): - prev = cur.get(timeframe) - if prev is None or since < prev: - cur[timeframe] = since - logger.info("已加载恢复点: %s", path) - - def _get_resume_since(self, symbol: str, timeframe: str) -> Optional[int]: - """若曾断线，返回应从哪一毫秒起补拉该 symbol/tf。""" - with self._lock: - return self._resume_since.get(symbol, {}).get(timeframe) - - def _set_resume_since(self, symbol: str, timeframe: str, since_ms: int) -> None: - """断线时写入恢复点（取更早的 since 以免漏数据），并持久化到磁盘。""" - with self._lock: - per_symbol = self._resume_since.setdefault(symbol, {}) - prev = per_symbol.get(timeframe) - # 取更早的 since，避免跳过数据 - if prev is None or since_ms < prev: - per_symbol[timeframe] = since_ms - logger.warning( - "记录断线恢复点: %s %s since=%s (%s)", - symbol, - timeframe, - since_ms, - to_utc_iso(since_ms), - ) - # 同步写盘 - self._save_resume_since() - - def _clear_resume_since(self, symbol: str, timeframe: str) -> None: - """补数成功后清除该 symbol/tf 的恢复点。""" - with self._lock: - if symbol in self._resume_since and timeframe in self._resume_since[symbol]: - del self._resume_since[symbol][timeframe] - if not self._resume_since[symbol]: - del self._resume_since[symbol] - logger.info("清除断线恢复点: %s %s", symbol, timeframe) - # 同步写盘 - self._save_resume_since() - - def on_update(self, callback) -> None: - """注册数据更新回调（签名: callback(symbol, timeframe)）。""" - self._update_callbacks.append(callback) - - def _notify_update(self, symbol: str, timeframe: str) -> None: - """通知所有回调：某 symbol/timeframe 数据已更新。""" - for cb in self._update_callbacks: - try: - cb(symbol, timeframe) - except Exception as exc: - logger.error("数据更新回调异常: %s", exc) - - def _derivatives_refresh_loop(self) -> None: - """后台线程：每 60 秒拉取一次衍生品数据。与 K 线 watch 模式一致：拉取 → 合并 → 写入内存。""" - # 首次启动先加载本地历史 - for symbol in self.symbols: - history = self._load_derivatives_local(symbol) - if history: - with self._lock: - self._derivatives[symbol] = history - logger.info("衍生品 %s 加载本地记录: %d 条", symbol, len(history)) - - DERIVATIVES_INTERVAL = 60 - while not self._stop_event.wait(DERIVATIVES_INTERVAL): - for symbol in self.symbols: - try: - new_records = self._fetch_derivatives_sync(symbol) - if new_records: - with self._lock: - base = self._derivatives.get(symbol, []) - self._derivatives[symbol] = self._merge_derivatives(base, new_records) - except Exception as e: - logger.debug("衍生品刷新失败 %s: %s", symbol, e) - - def start_background_workers(self) -> None: - """启动后台线程：冷启动回填 + 周期性落盘 + 衍生品刷新。WebSocket 监听由 lifespan 异步启动。""" - self._stop_event.clear() - self._backfill_thread = threading.Thread(target=self._cold_start_backfill, name="backfill-loop", daemon=True) - self._persist_thread = threading.Thread(target=self._persist_loop, name="persist-loop", daemon=True) - self._derivatives_thread = threading.Thread(target=self._derivatives_refresh_loop, name="derivatives-loop", daemon=True) - self._backfill_thread.start() - self._persist_thread.start() - self._derivatives_thread.start() - logger.info("后台线程已启动（回填 + 落盘 + 衍生品）") - - def start_watch_tasks(self) -> None: - """在当前 asyncio event loop 上启动 WebSocket 监听。必须在 lifespan 内调用。""" - loop = asyncio.get_running_loop() - self._watch_loop_task = loop.create_task(self._watch_loop(), name="watch-master") - logger.info("WebSocket 监听主任务已创建") - - def stop(self) -> None: - """停止后台线程和异步 WebSocket 监听任务。""" - self._stop_event.set() - - # 取消 WebSocket watch 任务（它们检查 stop_event 后会退出） - for task in getattr(self, '_watch_tasks', []): - if not task.done(): - task.cancel() - if hasattr(self, '_watch_loop_task') and self._watch_loop_task is not None: - if not self._watch_loop_task.done(): - self._watch_loop_task.cancel() - - # Join 所有后台线程 - for thread, name in [ - (self._backfill_thread, "backfill"), - (self._persist_thread, "persist"), - ]: - if thread: - thread.join(timeout=5) - if thread.is_alive(): - logger.warning("后台线程 %s 未能在 5s 内结束", name) - - logger.info("数据提供商已停止") - - def _persist_loop(self) -> None: - """每隔 PERSIST_INTERVAL 秒把内存快照写 CSV 并保存恢复点。""" - while not self._stop_event.wait(PERSIST_INTERVAL): - self._persist_all() - - def _persist_all(self) -> None: - """在锁内复制 data 后落盘，避免长时间持锁。有限制的时间周期跳过不写（CSV 已有全量历史）。""" - if not self._ready.is_set(): - return - with self._lock: - snapshot = { - symbol: {tf: list(data) for tf, data in tf_map.items()} - for symbol, tf_map in self.data.items() - } - for symbol, tf_map in snapshot.items(): - for timeframe, data in tf_map.items(): - if timeframe in MAX_CANDLES_IN_MEMORY: - continue # 内存只有尾部 N 根，不覆盖 CSV 全量 - self._write_to_disk(symbol, timeframe, data) - # 周期性落盘衍生品数据（与 K 线一致：锁内复制 → 完整覆写 CSV） - with self._lock: - deriv_snapshot = { - symbol: list(records) for symbol, records in self._derivatives.items() - } - for symbol, records in deriv_snapshot.items(): - if records: - self._write_derivatives_to_disk(symbol, records) - - # 周期性也保存一次恢复点，保证一致性 - self._save_resume_since() - - def is_ready(self) -> bool: - return self._ready.is_set() - - def wait_ready(self, timeout: Optional[float] = None) -> bool: - return self._ready.wait(timeout) - - def get_available_timeframes(self) -> List[str]: - return list(self.available_timeframes) - - def get_derived_timeframes(self) -> List[str]: - return list(self.derived_map.keys()) - - def _get_base_klines( - self, - symbol: str, - timeframe: str, - start_ms: Optional[int], - end_ms: Optional[int], - limit: Optional[int], - ) -> List[Dict[str, float]]: - """从内存读取已缓存的基础周期 K 线并按时间/limit 裁剪；内存不足时从 CSV 补充。""" - with self._lock: - candles = list(self.data.get(symbol, {}).get(timeframe, [])) - # 有限制的时间周期，若内存不够老则从 CSV 补充历史 - if timeframe in MAX_CANDLES_IN_MEMORY and candles and start_ms is not None: - if start_ms < candles[0]["timestamp"]: - csv_candles = self._load_local(symbol, timeframe) - if csv_candles: - merged = {c["timestamp"]: c for c in csv_candles} - for c in candles: - merged[c["timestamp"]] = c - candles = sorted(merged.values(), key=lambda x: x["timestamp"]) - if start_ms is not None: - candles = [row for row in candles if row["timestamp"] >= start_ms] - if end_ms is not None: - candles = [row for row in candles if row["timestamp"] <= end_ms] - if limit: - candles = candles[-limit:] - return candles - - def get_klines( - self, - symbol: str, - timeframe: str, - start_time: Optional[object] = None, - end_time: Optional[object] = None, - limit: Optional[int] = None, - ) -> List[Dict[str, float]]: - """对外查询：基础周期直接返回；衍生周期从 derived_map 取 base，resample 后对齐时间戳再裁剪。""" - if symbol not in self.symbols: - raise HTTPException(status_code=404, detail=f"symbol {symbol} 不可用") - self.wait_ready() - start_ms = parse_timestamp(start_time) - end_ms = parse_timestamp(end_time) - if timeframe in self.timeframes: - return self._get_base_klines(symbol, timeframe, start_ms, end_ms, limit) - base_tf = self.derived_map.get(timeframe) - if not base_tf: - raise HTTPException(status_code=404, detail=f"{symbol} 时间周期 {timeframe} 不可用") - target_minutes = timeframe_to_minutes(timeframe) - if target_minutes is None: - raise HTTPException(status_code=400, detail=f"不支持的时间周期: {timeframe}") - target_ms = target_minutes * 60_000 - # 起点前移一根目标周期长度，保证首根合成 K 边界完整 - adjusted_start = None if start_ms is None else max(0, start_ms - target_ms) - base_candles = self._get_base_klines(symbol, base_tf, adjusted_start, end_ms, None) - if not base_candles: - return [] - df = pd.DataFrame(base_candles) - if df.empty: - return [] - df = df.drop_duplicates(subset=["timestamp"], keep="last").sort_values("timestamp") - df["date"] = pd.to_datetime(df["timestamp"], unit="ms", utc=True) - # resample_to_interval 按「分钟」目标周期聚合 OHLCV - resampled = self.resample_df(df, target_minutes) - if resampled is None or resampled.empty: - return [] - # 统一得到毫秒 timestamp 列（resample 可能返回 date 或 DatetimeIndex） - if "timestamp" in resampled.columns: - resampled_df = resampled.copy() - else: - resampled_df = resampled.copy() - if "date" in resampled_df.columns: - dates = pd.to_datetime(resampled_df["date"], utc=True, errors="coerce") - resampled_df["timestamp"] = (dates.astype("int64", copy=False) // 1_000_000).astype("int64") - elif isinstance(resampled_df.index, pd.DatetimeIndex): - idx = resampled_df.index - if idx.tz is None: - idx = idx.tz_localize("UTC") - else: - idx = idx.tz_convert("UTC") - resampled_df["timestamp"] = (idx.astype("int64", copy=False) // 1_000_000).astype("int64") - else: - raise HTTPException(status_code=500, detail=f"聚合结果缺少 timestamp 列 ({timeframe})") - resampled_df = resampled_df.dropna(subset=["timestamp"]).sort_values("timestamp") - if start_ms is not None: - resampled_df = resampled_df[resampled_df["timestamp"] >= start_ms] - if end_ms is not None: - resampled_df = resampled_df[resampled_df["timestamp"] <= end_ms] - if resampled_df.empty: - return [] - resampled_df["datetime"] = resampled_df["timestamp"].apply(to_utc_iso) - for column in ["open", "high", "low", "close", "volume"]: - if column not in resampled_df.columns: - resampled_df[column] = 0.0 - resampled_df = resampled_df[["timestamp", "datetime", "open", "high", "low", "close", "volume"]] - result = resampled_df.to_dict("records") - if limit: - result = result[-limit:] - logger.debug( - "衍生周期返回", - extra={ - "symbol": symbol, - "timeframe": timeframe, - "base_timeframe": base_tf, - "count": len(result), - }, - ) - return result - - -def create_app(provider: DataProvider) -> FastAPI: - """构造 FastAPI 应用：lifespan 内同步 initialize 并启动后台拉数；WebSocket 实时推送。""" - ws_manager = WebSocketManager() - - # 初始化链上指标模块 - onchain_manager = OnchainMetricsManager( - data_dir=provider.data_dir, - api_keys=api_keys_from_env(), - ) - onchain_router = create_onchain_router(onchain_manager) - - def _on_data_update(symbol: str, base_tf: str) -> None: - """后台刷新线程回调：广播基础及衍生周期更新给 WebSocket 订阅者。""" - with provider._lock: - base_data = list(provider.data.get(symbol, {}).get(base_tf, [])) - recent = base_data[-WS_UPDATE_CANDLE_COUNT:] if base_data else [] - if recent: - ws_manager.broadcast_from_thread(symbol, base_tf, recent) - for derived_tf, src_base in provider.derived_map.items(): - if src_base != base_tf or not ws_manager.has_subscribers(symbol, derived_tf): - continue - try: - target_min = timeframe_to_minutes(derived_tf) - if target_min is None: - continue - now_ms = int(time.time() * 1000) - window_ms = target_min * 60_000 * (WS_UPDATE_CANDLE_COUNT + 2) - derived = provider.get_klines( - symbol, derived_tf, start_time=now_ms - window_ms, limit=WS_UPDATE_CANDLE_COUNT, - ) - if derived: - ws_manager.broadcast_from_thread(symbol, derived_tf, derived) - except Exception as exc: - logger.debug("衍生周期广播失败 %s %s: %s", symbol, derived_tf, exc) - - @asynccontextmanager - async def lifespan(app: FastAPI): - loop = asyncio.get_running_loop() - ws_manager.set_loop(loop) - provider.on_update(_on_data_update) - await loop.run_in_executor(None, provider.initialize) - provider.start_background_workers() - # 启动 WebSocket 实时监听（asyncio 后台任务） - provider.start_watch_tasks() - # 启动链上指标后台刷新 - onchain_manager.start() - try: - yield - finally: - onchain_manager.stop() - provider.stop() - - app = FastAPI(title="Chan 数据提供商", version="1.0.0", lifespan=lifespan) - - app.add_middleware( - CORSMiddleware, - allow_origins=["*"], - allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], - ) - - # 注册链上指标 API - app.include_router(onchain_router) - - @app.get("/health") - async def health() -> Dict[str, object]: - """存活检查：交易所、交易对、基础/衍生周期、是否已完成冷启动。""" - return { - "status": "ok", - "exchange": provider.exchange_name, - "symbols": provider.symbols, - "base_timeframes": provider.timeframes, - "derived_timeframes": provider.get_derived_timeframes(), - "timeframes": provider.get_available_timeframes(), - "ready": provider.is_ready(), - } - - @app.get("/timeframes") - async def list_timeframes() -> Dict[str, List[str]]: - """返回配置的基础周期与可合成的衍生周期列表。""" - provider.wait_ready() - return { - "base_timeframes": provider.timeframes, - "derived_timeframes": provider.get_derived_timeframes(), - "timeframes": provider.get_available_timeframes(), - } - - @app.get("/api/candles") - async def api_candles( - symbol: str = Query(..., description="如 BTC/USDT"), - tf: str = Query("1m", description="时间周期"), - start: Optional[int] = Query(None, description="开始时间戳(ms)"), - end: Optional[int] = Query(None, description="结束时间戳(ms)"), - limit: Optional[int] = Query(None, description="可选，限制返回数量"), - ): - """按交易对与时间周期返回 OHLCV；tf 支持配置的基础周期及衍生合成周期。""" - data = provider.get_klines(symbol=symbol, timeframe=tf, start_time=start, end_time=end, limit=limit) - return data - - @app.get("/api/derivatives") - async def api_derivatives( - symbol: str = Query("BTC/USDT:USDT", description="如 BTC/USDT:USDT"), - ): - """返回指定币对的衍生品数据快照（资金费率、OI、基差）。""" - with provider._lock: - records = list(provider._derivatives.get(symbol, [])) - if not records: - raise HTTPException(status_code=404, detail=f"衍生品数据不可用: {symbol}") - record = records[-1] # 最新一条 - return { - "symbol": symbol, - "timestamp": record.get("timestamp"), - "datetime": record.get("datetime"), - "funding_rate": record.get("funding_rate"), - "open_interest": record.get("open_interest"), - "oi_change_pct": record.get("oi_change_pct"), - "basis": record.get("basis"), - } - - homepage_path = Path(__file__).resolve().parent / "homepage.html" - docs_path = Path(__file__).resolve().parent / "api_docs.html" - - @app.get("/", response_class=HTMLResponse) - async def root(): - """服务主页。""" - if homepage_path.exists(): - return HTMLResponse(content=homepage_path.read_text(encoding="utf-8")) - return HTMLResponse(content="

主页页面未找到

", status_code=404) - - @app.get("/api-manual", response_class=HTMLResponse) - async def api_manual(): - """数据服务 API 使用手册页面。""" - derived_plan = DERIVED_TIMEFRAME_PLAN - timeframe_ms = TIMEFRAME_TO_MS - memory_limits = MAX_CANDLES_IN_MEMORY - base_tfs = provider.timeframes - derived_tfs = provider.get_derived_timeframes() - all_tfs = provider.get_available_timeframes() - symbols = provider.symbols - - # Build derived timeframe table rows - derived_rows = "" - for base, derived_list in derived_plan.items(): - present = "✓" if base in base_tfs else "✗" - derived_rows += f"{base}{present}{', '.join(f'{d}' for d in derived_list)}" - - # Build memory limits table rows - memory_rows = "" - for tf in base_tfs: - limit = memory_limits.get(tf) - if limit is None: - span = "unlimited" - elif tf == "1m": - span = f"{limit:,} candles (~3 months)" - elif tf == "1h": - span = f"{limit:,} candles (~1 year)" - else: - span = f"{limit:,} candles" - memory_rows += f"{tf}{span}" - - # Pre-build config JSON example to avoid f-string escaping issues - import json as _json - _config_obj = { - "exchange": "binance", - "symbols": ["BTC/USDT:USDT", "ETH/USDT:USDT"], - "start_time": "2024-01-01T00:00:00Z", - "start_time_per_tf": {"1m": "2026-01-01T00:00:00Z"}, - "timeframes": ["1m", "1h", "1d", "1w"], - "data_dir": "./data", - } - config_example = _json.dumps(_config_obj, indent=2, ensure_ascii=False) - - return f""" - - - - - - - - Data Provider — API Manual - - - - - - -
-
-

Data Provider v1.0.0

-

Cryptocurrency OHLCV data microservice — REST + WebSocket API for candlestick (K-line) data

-

Exchange: {provider.exchange_name} · Symbols: {len(symbols)} · Timeframes: {len(all_tfs)} · Status: {'Ready' if provider.is_ready() else 'Initializing…'}

-
-
- -
- - - - - -
-
-
What is Data Provider?
-
-

Data Provider is a standalone microservice that fetches, caches, and serves cryptocurrency OHLCV (Open/High/Low/Close/Volume) candlestick data from the Binance Futures exchange via CCXT.

-

It decouples data acquisition from analysis — consumers request historical or real-time market data through a simple REST or WebSocket API instead of talking to exchange APIs directly.

-
-
Key Features
-
-
-
    -
  • Fetches base OHLCV at 1m, 1h, 1d, 1w from Binance
    • -
  • Synthesizes {len(derived_tfs)} derived timeframes by resampling
    • -
  • Dual storage: in-memory cache + CSV persistence
    • -
  • Background refresh every {RECENT_FETCH_INTERVAL}s
    • -
-
-
-
    -
  • REST endpoint at /api/candles
    • -
  • WebSocket real-time push at /ws
    • -
  • Automatic disconnect recovery
    • -
  • Fully Dockerized, single JSON config file
    • -
-
-
-
-
- -
-
Architecture
-
- 
-┌───────────────────────┐
-│   Binance Futures     │
-│   (ccxt driver)       │
-└──────────┬────────────┘
-           │ fetch_ohlcv()
-┌──────────┴────────────┐
-│    DataProvider        │
-│                        │
-│  · In-memory cache     │
-│  · CSV persistence     │
-│  · Background refresh  │
-│  · Derived timeframe   │
-│    resample engine     │
-└──────┬────────┬────────┘
-       │        │
-  REST │        │ WebSocket
-       │        │
-┌──────┴────────┴────────┐
-│  Consumers (Flask app, │
-│  browser, scripts…)    │
-└────────────────────────┘
-
-
- -
-
Endpoints at a Glance
-
- - - - - - - - - - - - - -
MethodPathDescription
GET/Service info & health summary
GET/healthStructured health check (JSON)
GET/timeframesList available base & derived timeframes
GET/api/candlesFetch OHLCV candles
GET/api/derivativesFunding rate, OI, basis
GET/api/onchain/metricsOn-chain metric time series
GET/api/onchain/latestLatest on-chain snapshot
WS/wsReal-time K-line streaming
GET/api-manualThis page
-
-
- -
-
Data Model
-
-

Every candle object (REST response, WebSocket payload) has this structure:

- - - - - - - - - - - -
FieldTypeExample
timestampint1704067200000
datetimestring"2024-01-01T00:00:00Z"
openfloat42314.0
highfloat44266.0
lowfloat42207.9
closefloat44230.2
volumefloat206424.144
-
-
-
- - -
- -
GET /health
-
-
-

Health check — use ready to determine if the service finished cold-start initialization.

-

Response 200 OK

- 
{{
-  "status": "ok",
-  "exchange": "binance",
-  "symbols": ["BTC/USDT:USDT", "ETH/USDT:USDT"],
-  "base_timeframes": ["1m", "1h", "1d", "1w"],
-  "derived_timeframes": ["2m","3m","4m","5m",...],
-  "timeframes": ["1m","2m","3m",...],
-  "ready": true
-}}
- -
-
- -
GET /api/candles
-
-
-

The primary data endpoint. Returns OHLCV candlestick data for a given symbol and timeframe.

-

Query Parameters

- - - - - - - - - -
ParamTypeRequiredDefaultDescription
symbolstringYes—Trading pair, e.g. BTC/USDT:USDT
tfstringNo1mTimeframe — any base or derived
startintNo—Start time, UTC milliseconds
endintNo—End time, UTC milliseconds
limitintNo—Max candles to return
-

Response 200 OK — array of candle objects

- 
[
-  {{
-    "timestamp": 1704067200000,
-    "datetime": "2024-01-01T00:00:00Z",
-    "open": 42314.0,
-    "high": 44266.0,
-    "low": 42207.9,
-    "close": 44230.2,
-    "volume": 206424.144
-  }}
-]
-

At least one of start/end or limit should be provided. If none given, all in-memory candles are returned (subject to memory limits).

-
-
- -
GET /timeframes
-
-
-

Lists all available timeframes: base (fetched from exchange) and derived (synthesized by resampling).

- 
{{
-  "base_timeframes": ["1m","1h","1d","1w"],
-  "derived_timeframes": ["2m","3m","4m","5m",...],
-  "timeframes": ["1m","2m","3m",...]
-}}
-
-
- -
GET /
-
-
-

Root endpoint — returns service name, exchange, symbols, timeframes, and ready status. Same data as /health.

-
-
-
- - -
- -
GET /api/onchain/latest
-
-
-

获取全部链上指标最新快照。

-

Response 200 OK

- 
{{
-  "timestamp": 1704067200000,
-  "btc_netflow": {{
-    "timestamp": 1704067200000,
-    "datetime": "2024-01-01T00:00:00Z",
-    "value": -33361122.64,
-    "sub_value": 1366105343.72,
-    "extra": {{ "inflow_usd": 1366105343.72, "outflow_usd": 1399466466.36 }}
-  }},
-  "stablecoin_supply": {{
-    "timestamp": 1704067200000,
-    "datetime": "2024-01-01T00:00:00Z",
-    "value": 276932114785,
-    "sub_value": 276.93,
-    "extra": {{ "USDT": 184419658464, "USDC": 73258860594, "DAI": 4625356465 }}
-  }},
-  "etf_flow": {{
-    "timestamp": 1704067200000,
-    "datetime": "2024-01-01T00:00:00Z",
-    "value": -296.0,
-    "sub_value": -296000000.0,
-    "extra": {{ "total_million_usd": -296.0, "breakdown": {{ "IBIT": -219.4, "GBTC": -62.8 }} }}
-  }},
-  "mvrv_zscore": {{
-    "timestamp": 1704067200000,
-    "datetime": "2024-01-01T00:00:00Z",
-    "value": -1.3895,
-    "sub_value": 1.1321,
-    "extra": {{ "mvrv_ratio": 1.1321, "rolling_mean": 1.6787, "rolling_stddev": 0.3934 }}
-  }},
-  "sopr": null
-}}
-

4个免费指标 (netflow/stablecoin/etf/mvrv) 每5分钟自动刷新。SOPR 需 Glassnode API key。

- -
-
- -
GET /api/onchain/metrics
-
-
-

获取单个链上指标的时间序列。

-

Query Parameters

- - - - - - - - -
ParamTypeRequiredDefaultDescription
metricstringYes—指标: btc_netflow, stablecoin_supply, etf_flow, mvrv_zscore, sopr
limitintNo100返回条数上限 (1-5000)
startintNo—起始时间戳(ms)
endintNo—结束时间戳(ms)
-

cURL Example

- 
curl "http://localhost:80/api/onchain/metrics?metric=btc_netflow&limit=10"
-curl "http://localhost:80/api/onchain/metrics?metric=mvrv_zscore&start=1704067200000&end=1711929600000"
-
-
- -
GET /api/onchain/available
-
-
-

列出所有指标及其数据量、时间范围。

- 
{{
-  "btc_netflow": {{"count": 4015, "has_data": true, "first_ts": "2024-01-01T00:00:00Z", "last_ts": "2026-07-01T00:00:00Z"}},
-  "stablecoin_supply": {{"count": 2, "has_data": true, ...}},
-  "etf_flow": {{"count": 12, "has_data": true, ...}},
-  "mvrv_zscore": {{"count": 4036, "has_data": true, ...}},
-  "sopr": {{"count": 0, "has_data": false, ...}}
-}}
-
-
- -
指标说明
-
-
- - - - - - - - - -
Metric来源value 含义sub_value费用
btc_netflowCoinMetrics净流量 (USD)流入 (USD)免费
stablecoin_supplyCoinGecko总市值 (USD)总市值 (B)免费
etf_flowFarside净流入 (M USD)净流入 (USD)免费
mvrv_zscoreCoinMetricsZ-ScoreMVRV Ratio免费
soprGlassnodeSOPR 值—需 key
-
-
- -
CSV 存储
-
-
-

数据落盘在 data/onchain/ 下，每个指标一个 CSV 文件：

-
    -
  • btc_netflow.csv — 交易所净流量 (timestamp, datetime, value, sub_value, extra)
    • -
  • stablecoin_supply.csv — 稳定币供应
    • -
  • etf_flow.csv — ETF 净流入
    • -
  • mvrv_zscore.csv — MVRV Z-Score
    • -
  • sopr.csv — SOPR
    • -
-

每5分钟自动刷新并写盘。extra 字段存 JSON (明细/分解数据)。

-
-
-
- - -
- -
-
WS /ws — Real-time K-line Streaming
-
-

Connect to receive live incremental candle updates. Subscribe to specific symbol/timeframe pairs; receive a full snapshot followed by real-time push.

- -
Connection URL
- 
ws://localhost:80/ws
-

Replace localhost:80 with the server address if connecting remotely.

- -
Client → Server Messages
- - - - - - - - - - - - - - - - - - - -
ActionPayloadDescription
subscribe
{{"action":"subscribe","symbol":"BTC/USDT:USDT","timeframe":"1m"}}
Subscribe to a symbol/timeframe. Server replies with subscribed, then a snapshot of recent candles.
unsubscribe
{{"action":"unsubscribe","symbol":"BTC/USDT:USDT","timeframe":"1m"}}
Stop receiving updates for this pair.
ping
{{"action":"ping"}}
Keepalive ping. Server replies with pong.
- -
Server → Client Messages
- - - - - - - - - - -
TypeDescription
subscribedConfirms subscription: {{"type":"subscribed","symbol":"...","timeframe":"..."}}
snapshotFull recent candles on subscribe: {{"type":"snapshot","symbol":"...","timeframe":"...","data":[...]}}
klineIncremental candle updates (last 2 candles): {{"type":"kline","symbol":"...","timeframe":"...","data":[...]}}
pongPing response.
unsubscribedUnsubscribe confirmation.
errorError message: {{"type":"error","message":"..."}}
- -
Python Client Example
- 
import asyncio, json
-import websockets
-
-async def listen():
-    async with websockets.connect("ws://localhost:80/ws") as ws:
-        # Subscribe to BTC 1m candles
-        await ws.send(json.dumps({{
-            "action": "subscribe",
-            "symbol": "BTC/USDT:USDT",
-            "timeframe": "1m"
-        }}))
-        while True:
-            msg = json.loads(await ws.recv())
-            print(f"[{{msg['type']}}] {{msg.get('symbol','')}} {{msg.get('timeframe','')}}")
-            if msg['type'] == 'kline':
-                for candle in msg.get('data', []):
-                    print(f"  o={{candle['open']}} h={{candle['high']}} "
-                          f"l={{candle['low']}} c={{candle['close']}}")
-
-asyncio.run(listen())
-

Requires: pip install websockets. Subscribe to multiple symbol/timeframe pairs by sending additional subscribe messages.

-
-
-
- - -
- -
-
Base & Derived Timeframes
-
-

Base timeframes are fetched directly from the exchange. Derived timeframes are synthesized on-the-fly by aggregating base candles — no extra API calls needed.

-
-
- - - {derived_rows} -
Base TimeframeConfiguredDerived Timeframes Synthesized
-
-
- -
-
Memory Limits
-
-

To bound memory usage, only a sliding window is held in RAM for high-frequency timeframes. Data beyond the window is served from CSV on disk lazily.

-
-
- - - {memory_rows} -
TimeframeMax Candles in Memory
-
-
- -
-
CSV Storage
-
-

Data is persisted under data/<timeframe>/ as one CSV file per symbol (e.g. data/1m/BTC_USDT_USDT.csv). Schema: timestamp,datetime,open,high,low,close,volume.

-
-
-
- - -
- -
-
config.json
-
- 
{config_example}
-
-
- - - - - - - - - - -
FieldTypeDescription
exchangestringCCXT exchange ID. Tested with binance (Binance Futures).
symbolsstring[]Trading pairs in Binance Futures format: BTC/USDT:USDT. Also accepts comma-separated string.
start_timestringGlobal earliest backfill date, ISO 8601 UTC. Used as fallback for timeframes not listed in start_time_per_tf.
start_time_per_tfobjectOptional. Per-timeframe override. Keys are timeframe strings (e.g. "1m"), values are ISO 8601 UTC dates. Useful for limiting high-frequency data volume (e.g. 1m from recent date, longer timeframes from earlier).
timeframesstring[]Base timeframes to fetch. Supported: 1m, 1h, 1d, 1w.
data_dirstringDirectory for CSV storage (created if missing).
-
-
- -
-
Environment Variables
-
- - - - - - - - -
VariableDefaultDescription
CONFIG_PATHconfig.jsonPath to the JSON config file.
UVICORN_HOST0.0.0.0Host to bind.
UVICORN_PORT80Port to bind.
UVICORN_LOG_LEVELinfoLog level: debug, info, warning, error.
-
-
-
- - -
- -
-
cURL — Fetch Last 500 5-Minute Candles
-
- 
curl "http://localhost:80/api/candles?symbol=BTC/USDT:USDT&tf=5m&limit=500"
-
-
- -
-
cURL — Fetch Date Range
-
- 
# 4-hour candles for ETH, Jan–Mar 2024
-curl "http://localhost:80/api/candles?symbol=ETH/USDT:USDT&tf=4h&start=1704067200000&end=1711929600000"
-
-
- -
-
Python — requests
-
- 
import requests
-
-# Check service is ready
-resp = requests.get("http://localhost:80/health")
-meta = resp.json()
-print(f"Ready: {{meta['ready']}}")
-
-# Fetch candles
-params = {{
-    "symbol": "BTC/USDT:USDT",
-    "tf": "1h",
-    "limit": 200,
-}}
-resp = requests.get("http://localhost:80/api/candles", params=params)
-candles = resp.json()
-for c in candles[:3]:
-    print(c["datetime"], c["open"], c["close"])
-
-
- -
-
JavaScript — WebSocket
-
- 
const ws = new WebSocket("ws://localhost:80/ws");
-
-ws.onopen = () => {{
-  ws.send(JSON.stringify({{
-    action: "subscribe",
-    symbol: "BTC/USDT:USDT",
-    timeframe: "1m"
-  }}));
-}};
-
-ws.onmessage = (event) => {{
-  const msg = JSON.parse(event.data);
-  if (msg.type === "snapshot") {{
-    console.log("Initial data:", msg.data.length, "candles");
-  }} else if (msg.type === "kline") {{
-    console.log("Update:", msg.data);
-  }}
-}};
-
-// Keepalive every 30s
-setInterval(() => ws.send(JSON.stringify({{action: "ping"}})), 30000);
-
-
- -
-
JavaScript — fetch (browser)
-
- 
const params = new URLSearchParams({{
-  symbol: "BTC/USDT:USDT",
-  tf: "5m",
-  limit: "500"
-}});
-const resp = await fetch(`/api/candles?${{params}}`);
-const candles = await resp.json();
-console.log(`Got ${{candles.length}} candles`);
-
-
-
- -
- Data Provider v1.0.0 · Chan Project · Served from {provider.exchange_name} -
-
- - - -""" - - @app.websocket("/ws") - async def websocket_endpoint(ws: WebSocket): - """WebSocket 实时 K 线推送。 - - 客户端发送 JSON: - {"action": "subscribe", "symbol": "BTC/USDT:USDT", "timeframe": "1m"} - {"action": "unsubscribe", "symbol": "BTC/USDT:USDT", "timeframe": "1m"} - {"action": "ping"} - 服务端推送: - {"type": "subscribed", "symbol": "...", "timeframe": "..."} - {"type": "snapshot", "symbol": "...", "timeframe": "...", "data": [...]} - {"type": "kline", "symbol": "...", "timeframe": "...", "data": [...]} - {"type": "pong"} - {"type": "error", "message": "..."} - """ - await ws_manager.connect(ws) - try: - while True: - raw = await ws.receive_text() - try: - msg = json.loads(raw) - except json.JSONDecodeError: - await ws.send_text(json.dumps({"type": "error", "message": "invalid JSON"})) - continue - - action = msg.get("action", "") - symbol = str(msg.get("symbol", "")).strip() - timeframe = str(msg.get("timeframe", "")).strip() - - if action == "ping": - await ws.send_text(json.dumps({"type": "pong"})) - - elif action == "subscribe": - if not symbol or not timeframe: - await ws.send_text(json.dumps( - {"type": "error", "message": "需要 symbol 和 timeframe 字段"} - )) - continue - await ws_manager.subscribe(ws, symbol, timeframe) - await ws.send_text(json.dumps( - {"type": "subscribed", "symbol": symbol, "timeframe": timeframe}, - ensure_ascii=False, - )) - try: - snapshot = provider.get_klines(symbol, timeframe, limit=DEFAULT_LIMIT) - if snapshot: - await ws.send_text(json.dumps( - {"type": "snapshot", "symbol": symbol, "timeframe": timeframe, "data": snapshot}, - ensure_ascii=False, - )) - except Exception as exc: - await ws.send_text(json.dumps({"type": "error", "message": str(exc)})) - - elif action == "unsubscribe": - await ws_manager.unsubscribe(ws, symbol, timeframe) - await ws.send_text(json.dumps( - {"type": "unsubscribed", "symbol": symbol, "timeframe": timeframe}, - ensure_ascii=False, - )) - - else: - await ws.send_text(json.dumps({"type": "error", "message": f"未知 action: {action}"})) - - except WebSocketDisconnect: - pass - finally: - await ws_manager.disconnect(ws) - - @app.get("/api/docs", response_class=HTMLResponse, include_in_schema=False) - async def api_docs(): - """返回自定义 API 文档页面。""" - if docs_path.exists(): - return HTMLResponse(content=docs_path.read_text(encoding="utf-8")) - return HTMLResponse(content="

API 文档页面未找到

", status_code=404) - - return app - - -def build_app() -> FastAPI: - """默认入口：从环境变量 CONFIG_PATH（或 config.json）加载配置并创建 FastAPI app。""" - config_path = Path(os.getenv("CONFIG_PATH", "config.json")) - provider = DataProvider(config_path) - return create_app(provider) - - -app = build_app() - - -def main() -> None: - """直接运行本模块时启动 uvicorn（监听 UVICORN_HOST / UVICORN_PORT）。""" - host = os.getenv("UVICORN_HOST", "0.0.0.0") - port = int(os.getenv("UVICORN_PORT", "80")) - - uvicorn.run(app, host=host, port=port, log_level=os.getenv("UVICORN_LOG_LEVEL", "info")) - - -if __name__ == "__main__": - main() - diff --git a/data_provider/onchain_metrics.py b/data_provider/onchain_metrics.py deleted file mode 100644 index 8bcf95a..0000000 --- a/data_provider/onchain_metrics.py +++ /dev/null @@ -1,597 +0,0 @@ -""" -链上指标数据模块：拉取 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']*>(.*?)', html, re.DOTALL) - if not table_match: - logger.warning("Farside 页面未找到表格") - return [] - - rows_html = re.findall(r']*>(.*?)', 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']*>(.*?)', 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 diff --git a/data_provider/requirements.txt b/data_provider/requirements.txt deleted file mode 100644 index 0545ac1..0000000 --- a/data_provider/requirements.txt +++ /dev/null @@ -1,6 +0,0 @@ -ccxt>=4.0.0,<5.0.0 -fastapi>=0.110.0,<1.0.0 -uvicorn[standard]>=0.23.0,<1.0.0 -pandas>=2.0.0,<3.0.0 -technical==1.5.0 -requests>=2.28.0