1619 lines
77 KiB
Python
1619 lines
77 KiB
Python
"""
|
||
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 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
|
||
|
||
# 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._fetch_thread: Optional[threading.Thread] = None
|
||
self._persist_thread: Optional[threading.Thread] = None
|
||
self._backfill_thread: Optional[threading.Thread] = None
|
||
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 = []
|
||
|
||
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):
|
||
"""实例化 ccxt 交易所，币安期货默认 defaultType=future，并 load_markets。"""
|
||
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", exchange.id)
|
||
return exchange
|
||
|
||
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 _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 start_background_workers(self) -> None:
|
||
"""启动后台线程：冷启动回填、增量刷新、周期性落盘。"""
|
||
self._stop_event.clear()
|
||
self._backfill_thread = threading.Thread(target=self._cold_start_backfill, name="backfill-loop", daemon=True)
|
||
self._fetch_thread = threading.Thread(target=self._refresh_loop, name="refresh-loop", daemon=True)
|
||
self._persist_thread = threading.Thread(target=self._persist_loop, name="persist-loop", daemon=True)
|
||
self._backfill_thread.start()
|
||
self._fetch_thread.start()
|
||
self._persist_thread.start()
|
||
logger.info("后台线程已启动（回填 + 增量刷新 + 落盘）")
|
||
|
||
def stop(self) -> None:
|
||
"""停止后台线程（应用关闭时 lifespan finally 调用）。"""
|
||
self._stop_event.set()
|
||
for thread, name in [
|
||
(self._backfill_thread, "backfill"),
|
||
(self._fetch_thread, "fetch"),
|
||
(self._persist_thread, "persist"),
|
||
]:
|
||
if thread:
|
||
thread.join(timeout=5)
|
||
if thread.is_alive():
|
||
logger.warning("后台线程 %s 未能在 5s 内结束", name)
|
||
logger.info("数据提供商已停止")
|
||
|
||
def _refresh_loop(self) -> None:
|
||
"""轮询各 symbol/tf：有恢复点则先补历史，否则 fetch 最近 RECENT_CANDLE_LIMIT 根。"""
|
||
while not self._stop_event.is_set():
|
||
for symbol in self.symbols:
|
||
for timeframe in self.timeframes:
|
||
try:
|
||
# 若存在断线恢复点，则优先从该 since 补齐历史数据
|
||
resume_since = self._get_resume_since(symbol, timeframe)
|
||
if resume_since is not None:
|
||
logger.info(
|
||
"开始断线后补数: %s %s since=%s (%s)",
|
||
symbol,
|
||
timeframe,
|
||
resume_since,
|
||
to_utc_iso(resume_since),
|
||
)
|
||
history = self._fetch_history(symbol, timeframe, resume_since)
|
||
with self._lock:
|
||
current = self.data.setdefault(symbol, {}).get(timeframe, [])
|
||
merged = self._merge_candles(timeframe, current, history)
|
||
self.data[symbol][timeframe] = merged
|
||
self._notify_update(symbol, timeframe)
|
||
self._clear_resume_since(symbol, timeframe)
|
||
self._trim(symbol, timeframe)
|
||
else:
|
||
# 正常增量获取最近若干根K线
|
||
candles = self.exchange.fetch_ohlcv(
|
||
symbol,
|
||
timeframe=timeframe,
|
||
limit=RECENT_CANDLE_LIMIT,
|
||
)
|
||
if not candles:
|
||
continue
|
||
with self._lock:
|
||
current = self.data.setdefault(symbol, {}).get(timeframe, [])
|
||
merged = self._merge_candles(timeframe, current, candles)
|
||
self.data[symbol][timeframe] = merged
|
||
self._notify_update(symbol, timeframe)
|
||
self._trim(symbol, timeframe)
|
||
except ccxt.BaseError as exc:
|
||
logger.error("更新最新K线失败 (%s %s): %s", symbol, timeframe, exc)
|
||
# 记录应当从何时恢复拉取，避免重连后从当前时间开始导致丢K
|
||
with self._lock:
|
||
current = self.data.get(symbol, {}).get(timeframe, [])
|
||
if current:
|
||
last_ts = int(current[-1]["timestamp"])
|
||
else:
|
||
last_ts = self._get_start_time(timeframe)
|
||
tf_ms = TIMEFRAME_TO_MS[timeframe]
|
||
# 回退一个周期，确保包含可能未完全收盘的K线，去重由 _merge_candles 处理
|
||
since_ms = max(self._get_start_time(timeframe), last_ts - tf_ms)
|
||
self._set_resume_since(symbol, timeframe, since_ms)
|
||
time.sleep(2)
|
||
continue
|
||
if self._stop_event.wait(RECENT_FETCH_INTERVAL):
|
||
break
|
||
|
||
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)
|
||
# 周期性也保存一次恢复点，保证一致性
|
||
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()
|
||
|
||
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()
|
||
# 后台拉取历史数据补齐（不阻塞 HTTP/WS 服务）
|
||
threading.Thread(
|
||
target=provider.run_initial_history_fetch,
|
||
name="initial-history-fetch",
|
||
daemon=True,
|
||
).start()
|
||
try:
|
||
yield
|
||
finally:
|
||
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=["*"],
|
||
)
|
||
|
||
@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
|
||
|
||
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="<h1>主页页面未找到</h1>", 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"<tr><td><code>{base}</code></td><td>{present}</td><td>{', '.join(f'<code>{d}</code>' for d in derived_list)}</td></tr>"
|
||
|
||
# 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"<tr><td><code>{tf}</code></td><td>{span}</td></tr>"
|
||
|
||
# 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"""<!DOCTYPE html>
|
||
<html lang="zh-CN">
|
||
<head>
|
||
<meta charset="utf-8">
|
||
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
||
<title>Data Provider — API Manual</title>
|
||
<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.1.3/dist/css/bootstrap.min.css" rel="stylesheet">
|
||
<link href="https://cdn.jsdelivr.net/npm/bootstrap-icons@1.8.1/font/bootstrap-icons.css" rel="stylesheet">
|
||
<style>
|
||
body {{
|
||
font-family: "Helvetica Neue", Arial, "PingFang SC", "Microsoft YaHei", sans-serif;
|
||
background: #f8f9fa; color: #333; padding: 0;
|
||
}}
|
||
.hero {{
|
||
background: linear-gradient(135deg, #1a1a2e 0%, #16213e 50%, #0f3460 100%);
|
||
color: #fff; padding: 48px 0 40px; margin-bottom: 32px;
|
||
}}
|
||
.hero h1 {{ font-size: 2rem; font-weight: 700; margin: 0; }}
|
||
.hero .badge {{ font-size: 0.8rem; vertical-align: middle; margin-left: 10px; }}
|
||
.hero p {{ opacity: 0.85; margin: 8px 0 0; font-size: 0.95rem; }}
|
||
.nav-pills .nav-link {{ color: #495057; border-radius: 6px; padding: 6px 16px; font-size: 0.9rem; }}
|
||
.nav-pills .nav-link.active {{ background: #0d6efd; }}
|
||
.section {{ display: none; }}
|
||
.section.active {{ display: block; }}
|
||
.card {{ border: 1px solid #e9ecef; box-shadow: 0 1px 3px rgba(0,0,0,0.04); margin-bottom: 20px; }}
|
||
.card-header {{ background: #fff; border-bottom: 1px solid #e9ecef; font-weight: 600; font-size: 0.95rem; padding: 12px 18px; }}
|
||
pre {{ background: #1e1e2e; color: #cdd6f4; border-radius: 8px; padding: 16px; font-size: 0.85rem; line-height: 1.6; overflow-x: auto; }}
|
||
pre code {{ color: inherit; }}
|
||
.method-badge {{ display: inline-block; padding: 2px 8px; border-radius: 4px; font-size: 0.75rem; font-weight: 700; margin-right: 8px; width: 52px; text-align: center; }}
|
||
.method-GET {{ background: #d1fae5; color: #065f46; }}
|
||
.method-WS {{ background: #ede9fe; color: #5b21b6; }}
|
||
.endpoint-path {{ font-family: monospace; font-size: 0.9rem; }}
|
||
.table {{ font-size: 0.875rem; }}
|
||
.table code {{ background: #f1f3f5; padding: 1px 5px; border-radius: 3px; font-size: 0.82rem; }}
|
||
.status-ready {{ color: #198754; }}
|
||
.status-notready {{ color: #dc3545; }}
|
||
.param-required {{ color: #dc3545; font-weight: 600; }}
|
||
.toc-item {{ cursor: pointer; }}
|
||
footer {{ border-top: 1px solid #dee2e6; margin-top: 48px; padding: 20px 0; color: #868e96; font-size: 0.85rem; text-align: center; }}
|
||
.endpoint-card {{ border-left: 3px solid #0d6efd; }}
|
||
.endpoint-card.ws {{ border-left-color: #7c3aed; }}
|
||
</style>
|
||
</head>
|
||
<body>
|
||
|
||
<div class="hero">
|
||
<div class="container">
|
||
<h1>Data Provider <span class="badge bg-success">v1.0.0</span></h1>
|
||
<p>Cryptocurrency OHLCV data microservice — REST + WebSocket API for candlestick (K-line) data</p>
|
||
<p style="font-size:0.82rem; opacity:0.7;">Exchange: {provider.exchange_name} · Symbols: {len(symbols)} · Timeframes: {len(all_tfs)} · Status: <span class="{'status-ready' if provider.is_ready() else 'status-notready'}">{'Ready' if provider.is_ready() else 'Initializing…'}</span></p>
|
||
</div>
|
||
</div>
|
||
|
||
<div class="container" style="max-width: 960px;">
|
||
|
||
<!-- Tabs -->
|
||
<ul class="nav nav-pills mb-4" id="manualTabs" role="tablist">
|
||
<li class="nav-item"><button class="nav-link active" data-section="overview">Overview</button></li>
|
||
<li class="nav-item"><button class="nav-link" data-section="rest">REST API</button></li>
|
||
<li class="nav-item"><button class="nav-link" data-section="ws">WebSocket</button></li>
|
||
<li class="nav-item"><button class="nav-link" data-section="timeframes">Timeframes</button></li>
|
||
<li class="nav-item"><button class="nav-link" data-section="config">Configuration</button></li>
|
||
<li class="nav-item"><button class="nav-link" data-section="examples">Code Examples</button></li>
|
||
</ul>
|
||
|
||
<!-- ==================== OVERVIEW ==================== -->
|
||
<div class="section active" id="section-overview">
|
||
<div class="card">
|
||
<div class="card-header">What is Data Provider?</div>
|
||
<div class="card-body">
|
||
<p><strong>Data Provider</strong> 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.</p>
|
||
<p>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.</p>
|
||
<hr>
|
||
<h6>Key Features</h6>
|
||
<div class="row">
|
||
<div class="col-md-6">
|
||
<ul class="small">
|
||
<li>Fetches base OHLCV at <b>1m, 1h, 1d, 1w</b> from Binance</li>
|
||
<li>Synthesizes <b>{len(derived_tfs)} derived timeframes</b> by resampling</li>
|
||
<li>Dual storage: in-memory cache + CSV persistence</li>
|
||
<li>Background refresh every <b>{RECENT_FETCH_INTERVAL}s</b></li>
|
||
</ul>
|
||
</div>
|
||
<div class="col-md-6">
|
||
<ul class="small">
|
||
<li>REST endpoint at <code>/api/candles</code></li>
|
||
<li>WebSocket real-time push at <code>/ws</code></li>
|
||
<li>Automatic disconnect recovery</li>
|
||
<li>Fully Dockerized, single JSON config file</li>
|
||
</ul>
|
||
</div>
|
||
</div>
|
||
</div>
|
||
</div>
|
||
|
||
<div class="card">
|
||
<div class="card-header">Architecture</div>
|
||
<div class="card-body">
|
||
<pre style="background:#f8f9fa; color:#333; text-align:center; font-size:0.8rem; line-height:1.4;">
|
||
┌───────────────────────┐
|
||
│ 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…) │
|
||
└────────────────────────┘</pre>
|
||
</div>
|
||
</div>
|
||
|
||
<div class="card">
|
||
<div class="card-header">Endpoints at a Glance</div>
|
||
<div class="card-body p-0">
|
||
<table class="table table-hover mb-0">
|
||
<thead class="table-light"><tr><th>Method</th><th>Path</th><th>Description</th></tr></thead>
|
||
<tbody>
|
||
<tr><td><span class="method-badge method-GET">GET</span></td><td class="endpoint-path">/</td><td>Service info & health summary</td></tr>
|
||
<tr><td><span class="method-badge method-GET">GET</span></td><td class="endpoint-path">/health</td><td>Structured health check (JSON)</td></tr>
|
||
<tr><td><span class="method-badge method-GET">GET</span></td><td class="endpoint-path">/timeframes</td><td>List available base & derived timeframes</td></tr>
|
||
<tr><td><span class="method-badge method-GET">GET</span></td><td class="endpoint-path">/api/candles</td><td>Fetch OHLCV candles</td></tr>
|
||
<tr><td><span class="method-badge method-WS">WS</span></td><td class="endpoint-path">/ws</td><td>Real-time K-line streaming</td></tr>
|
||
<tr><td><span class="method-badge method-GET">GET</span></td><td class="endpoint-path">/api-manual</td><td>This page</td></tr>
|
||
</tbody>
|
||
</table>
|
||
</div>
|
||
</div>
|
||
|
||
<div class="card">
|
||
<div class="card-header">Data Model</div>
|
||
<div class="card-body">
|
||
<p class="small">Every candle object (REST response, WebSocket payload) has this structure:</p>
|
||
<table class="table table-bordered table-sm">
|
||
<thead class="table-light"><tr><th>Field</th><th>Type</th><th>Example</th></tr></thead>
|
||
<tbody>
|
||
<tr><td><code>timestamp</code></td><td>int</td><td><code>1704067200000</code></td></tr>
|
||
<tr><td><code>datetime</code></td><td>string</td><td><code>"2024-01-01T00:00:00Z"</code></td></tr>
|
||
<tr><td><code>open</code></td><td>float</td><td><code>42314.0</code></td></tr>
|
||
<tr><td><code>high</code></td><td>float</td><td><code>44266.0</code></td></tr>
|
||
<tr><td><code>low</code></td><td>float</td><td><code>42207.9</code></td></tr>
|
||
<tr><td><code>close</code></td><td>float</td><td><code>44230.2</code></td></tr>
|
||
<tr><td><code>volume</code></td><td>float</td><td><code>206424.144</code></td></tr>
|
||
</tbody>
|
||
</table>
|
||
</div>
|
||
</div>
|
||
</div>
|
||
|
||
<!-- ==================== REST API ==================== -->
|
||
<div class="section" id="section-rest">
|
||
|
||
<h5 class="mb-3"><span class="method-badge method-GET">GET</span> /health</h5>
|
||
<div class="card endpoint-card">
|
||
<div class="card-body">
|
||
<p class="small">Health check — use <code>ready</code> to determine if the service finished cold-start initialization.</p>
|
||
<p><strong>Response</strong> <span class="text-muted small">200 OK</span></p>
|
||
<pre><code>{{
|
||
"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
|
||
}}</code></pre>
|
||
<button class="btn btn-sm btn-outline-secondary" onclick="fetch('/health').then(r=>r.json()).then(d=>alert(JSON.stringify(d,null,2)))">Try it</button>
|
||
</div>
|
||
</div>
|
||
|
||
<h5 class="mb-3 mt-4"><span class="method-badge method-GET">GET</span> /api/candles</h5>
|
||
<div class="card endpoint-card">
|
||
<div class="card-body">
|
||
<p class="small">The primary data endpoint. Returns OHLCV candlestick data for a given symbol and timeframe.</p>
|
||
<p><strong>Query Parameters</strong></p>
|
||
<table class="table table-bordered table-sm">
|
||
<thead class="table-light"><tr><th>Param</th><th>Type</th><th>Required</th><th>Default</th><th>Description</th></tr></thead>
|
||
<tbody>
|
||
<tr><td><code>symbol</code></td><td>string</td><td><span class="param-required">Yes</span></td><td>—</td><td>Trading pair, e.g. <code>BTC/USDT:USDT</code></td></tr>
|
||
<tr><td><code>tf</code></td><td>string</td><td>No</td><td><code>1m</code></td><td>Timeframe — any base or derived</td></tr>
|
||
<tr><td><code>start</code></td><td>int</td><td>No</td><td>—</td><td>Start time, UTC milliseconds</td></tr>
|
||
<tr><td><code>end</code></td><td>int</td><td>No</td><td>—</td><td>End time, UTC milliseconds</td></tr>
|
||
<tr><td><code>limit</code></td><td>int</td><td>No</td><td>—</td><td>Max candles to return</td></tr>
|
||
</tbody>
|
||
</table>
|
||
<p><strong>Response</strong> <span class="text-muted small">200 OK — array of candle objects</span></p>
|
||
<pre><code>[
|
||
{{
|
||
"timestamp": 1704067200000,
|
||
"datetime": "2024-01-01T00:00:00Z",
|
||
"open": 42314.0,
|
||
"high": 44266.0,
|
||
"low": 42207.9,
|
||
"close": 44230.2,
|
||
"volume": 206424.144
|
||
}}
|
||
]</code></pre>
|
||
<p class="small text-muted">At least one of <code>start</code>/<code>end</code> or <code>limit</code> should be provided. If none given, all in-memory candles are returned (subject to memory limits).</p>
|
||
</div>
|
||
</div>
|
||
|
||
<h5 class="mb-3 mt-4"><span class="method-badge method-GET">GET</span> /timeframes</h5>
|
||
<div class="card endpoint-card">
|
||
<div class="card-body">
|
||
<p class="small">Lists all available timeframes: base (fetched from exchange) and derived (synthesized by resampling).</p>
|
||
<pre><code>{{
|
||
"base_timeframes": ["1m","1h","1d","1w"],
|
||
"derived_timeframes": ["2m","3m","4m","5m",...],
|
||
"timeframes": ["1m","2m","3m",...]
|
||
}}</code></pre>
|
||
</div>
|
||
</div>
|
||
|
||
<h5 class="mb-3 mt-4"><span class="method-badge method-GET">GET</span> /</h5>
|
||
<div class="card endpoint-card">
|
||
<div class="card-body">
|
||
<p class="small">Root endpoint — returns service name, exchange, symbols, timeframes, and ready status. Same data as <code>/health</code>.</p>
|
||
</div>
|
||
</div>
|
||
</div>
|
||
|
||
<!-- ==================== WEBSOCKET ==================== -->
|
||
<div class="section" id="section-ws">
|
||
|
||
<div class="card endpoint-card ws">
|
||
<div class="card-header"><span class="method-badge method-WS">WS</span> /ws — Real-time K-line Streaming</div>
|
||
<div class="card-body">
|
||
<p class="small">Connect to receive live incremental candle updates. Subscribe to specific symbol/timeframe pairs; receive a full snapshot followed by real-time push.</p>
|
||
|
||
<h6 class="mt-3">Connection URL</h6>
|
||
<pre><code>ws://localhost:80/ws</code></pre>
|
||
<p class="small text-muted">Replace <code>localhost:80</code> with the server address if connecting remotely.</p>
|
||
|
||
<h6 class="mt-3">Client → Server Messages</h6>
|
||
<table class="table table-bordered table-sm">
|
||
<thead class="table-light"><tr><th>Action</th><th>Payload</th><th>Description</th></tr></thead>
|
||
<tbody>
|
||
<tr>
|
||
<td><code>subscribe</code></td>
|
||
<td><pre style="margin:0;padding:4px 8px;font-size:0.78rem;"><code>{{"action":"subscribe","symbol":"BTC/USDT:USDT","timeframe":"1m"}}</code></pre></td>
|
||
<td>Subscribe to a symbol/timeframe. Server replies with <code>subscribed</code>, then a <code>snapshot</code> of recent candles.</td>
|
||
</tr>
|
||
<tr>
|
||
<td><code>unsubscribe</code></td>
|
||
<td><pre style="margin:0;padding:4px 8px;font-size:0.78rem;"><code>{{"action":"unsubscribe","symbol":"BTC/USDT:USDT","timeframe":"1m"}}</code></pre></td>
|
||
<td>Stop receiving updates for this pair.</td>
|
||
</tr>
|
||
<tr>
|
||
<td><code>ping</code></td>
|
||
<td><pre style="margin:0;padding:4px 8px;font-size:0.78rem;"><code>{{"action":"ping"}}</code></pre></td>
|
||
<td>Keepalive ping. Server replies with <code>pong</code>.</td>
|
||
</tr>
|
||
</tbody>
|
||
</table>
|
||
|
||
<h6 class="mt-3">Server → Client Messages</h6>
|
||
<table class="table table-bordered table-sm">
|
||
<thead class="table-light"><tr><th>Type</th><th>Description</th></tr></thead>
|
||
<tbody>
|
||
<tr><td><code>subscribed</code></td><td>Confirms subscription: <code>{{"type":"subscribed","symbol":"...","timeframe":"..."}}</code></td></tr>
|
||
<tr><td><code>snapshot</code></td><td>Full recent candles on subscribe: <code>{{"type":"snapshot","symbol":"...","timeframe":"...","data":[...]}}</code></td></tr>
|
||
<tr><td><code>kline</code></td><td>Incremental candle updates (last 2 candles): <code>{{"type":"kline","symbol":"...","timeframe":"...","data":[...]}}</code></td></tr>
|
||
<tr><td><code>pong</code></td><td>Ping response.</td></tr>
|
||
<tr><td><code>unsubscribed</code></td><td>Unsubscribe confirmation.</td></tr>
|
||
<tr><td><code>error</code></td><td>Error message: <code>{{"type":"error","message":"..."}}</code></td></tr>
|
||
</tbody>
|
||
</table>
|
||
|
||
<h6 class="mt-3">Python Client Example</h6>
|
||
<pre><code>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())</code></pre>
|
||
<p class="small text-muted">Requires: <code>pip install websockets</code>. Subscribe to multiple symbol/timeframe pairs by sending additional subscribe messages.</p>
|
||
</div>
|
||
</div>
|
||
</div>
|
||
|
||
<!-- ==================== TIMEFRAMES ==================== -->
|
||
<div class="section" id="section-timeframes">
|
||
|
||
<div class="card">
|
||
<div class="card-header">Base & Derived Timeframes</div>
|
||
<div class="card-body">
|
||
<p class="small"><strong>Base</strong> timeframes are fetched directly from the exchange. <strong>Derived</strong> timeframes are synthesized on-the-fly by aggregating base candles — no extra API calls needed.</p>
|
||
</div>
|
||
<div class="card-body p-0">
|
||
<table class="table table-hover mb-0">
|
||
<thead class="table-light"><tr><th>Base Timeframe</th><th>Configured</th><th>Derived Timeframes Synthesized</th></tr></thead>
|
||
<tbody>{derived_rows}</tbody>
|
||
</table>
|
||
</div>
|
||
</div>
|
||
|
||
<div class="card">
|
||
<div class="card-header">Memory Limits</div>
|
||
<div class="card-body">
|
||
<p class="small">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.</p>
|
||
</div>
|
||
<div class="card-body p-0">
|
||
<table class="table table-hover mb-0">
|
||
<thead class="table-light"><tr><th>Timeframe</th><th>Max Candles in Memory</th></tr></thead>
|
||
<tbody>{memory_rows}</tbody>
|
||
</table>
|
||
</div>
|
||
</div>
|
||
|
||
<div class="card">
|
||
<div class="card-header">CSV Storage</div>
|
||
<div class="card-body">
|
||
<p class="small">Data is persisted under <code>data/<timeframe>/</code> as one CSV file per symbol (e.g. <code>data/1m/BTC_USDT_USDT.csv</code>). Schema: <code>timestamp,datetime,open,high,low,close,volume</code>.</p>
|
||
</div>
|
||
</div>
|
||
</div>
|
||
|
||
<!-- ==================== CONFIGURATION ==================== -->
|
||
<div class="section" id="section-config">
|
||
|
||
<div class="card">
|
||
<div class="card-header">config.json</div>
|
||
<div class="card-body">
|
||
<pre><code>{config_example}</code></pre>
|
||
</div>
|
||
<div class="card-body p-0">
|
||
<table class="table table-hover mb-0">
|
||
<thead class="table-light"><tr><th>Field</th><th>Type</th><th>Description</th></tr></thead>
|
||
<tbody>
|
||
<tr><td><code>exchange</code></td><td>string</td><td>CCXT exchange ID. Tested with <code>binance</code> (Binance Futures).</td></tr>
|
||
<tr><td><code>symbols</code></td><td>string[]</td><td>Trading pairs in Binance Futures format: <code>BTC/USDT:USDT</code>. Also accepts comma-separated string.</td></tr>
|
||
<tr><td><code>start_time</code></td><td>string</td><td>Global earliest backfill date, ISO 8601 UTC. Used as fallback for timeframes not listed in <code>start_time_per_tf</code>.</td></tr>
|
||
<tr><td><code>start_time_per_tf</code></td><td>object</td><td><em>Optional.</em> Per-timeframe override. Keys are timeframe strings (e.g. <code>"1m"</code>), values are ISO 8601 UTC dates. Useful for limiting high-frequency data volume (e.g. 1m from recent date, longer timeframes from earlier).</td></tr>
|
||
<tr><td><code>timeframes</code></td><td>string[]</td><td>Base timeframes to fetch. Supported: <code>1m</code>, <code>1h</code>, <code>1d</code>, <code>1w</code>.</td></tr>
|
||
<tr><td><code>data_dir</code></td><td>string</td><td>Directory for CSV storage (created if missing).</td></tr>
|
||
</tbody>
|
||
</table>
|
||
</div>
|
||
</div>
|
||
|
||
<div class="card">
|
||
<div class="card-header">Environment Variables</div>
|
||
<div class="card-body p-0">
|
||
<table class="table table-hover mb-0">
|
||
<thead class="table-light"><tr><th>Variable</th><th>Default</th><th>Description</th></tr></thead>
|
||
<tbody>
|
||
<tr><td><code>CONFIG_PATH</code></td><td><code>config.json</code></td><td>Path to the JSON config file.</td></tr>
|
||
<tr><td><code>UVICORN_HOST</code></td><td><code>0.0.0.0</code></td><td>Host to bind.</td></tr>
|
||
<tr><td><code>UVICORN_PORT</code></td><td><code>80</code></td><td>Port to bind.</td></tr>
|
||
<tr><td><code>UVICORN_LOG_LEVEL</code></td><td><code>info</code></td><td>Log level: <code>debug</code>, <code>info</code>, <code>warning</code>, <code>error</code>.</td></tr>
|
||
</tbody>
|
||
</table>
|
||
</div>
|
||
</div>
|
||
</div>
|
||
|
||
<!-- ==================== EXAMPLES ==================== -->
|
||
<div class="section" id="section-examples">
|
||
|
||
<div class="card">
|
||
<div class="card-header">cURL — Fetch Last 500 5-Minute Candles</div>
|
||
<div class="card-body">
|
||
<pre><code>curl "http://localhost:80/api/candles?symbol=BTC/USDT:USDT&tf=5m&limit=500"</code></pre>
|
||
</div>
|
||
</div>
|
||
|
||
<div class="card">
|
||
<div class="card-header">cURL — Fetch Date Range</div>
|
||
<div class="card-body">
|
||
<pre><code># 4-hour candles for ETH, Jan–Mar 2024
|
||
curl "http://localhost:80/api/candles?symbol=ETH/USDT:USDT&tf=4h&start=1704067200000&end=1711929600000"</code></pre>
|
||
</div>
|
||
</div>
|
||
|
||
<div class="card">
|
||
<div class="card-header">Python — requests</div>
|
||
<div class="card-body">
|
||
<pre><code>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"])</code></pre>
|
||
</div>
|
||
</div>
|
||
|
||
<div class="card">
|
||
<div class="card-header">JavaScript — WebSocket</div>
|
||
<div class="card-body">
|
||
<pre><code>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);</code></pre>
|
||
</div>
|
||
</div>
|
||
|
||
<div class="card">
|
||
<div class="card-header">JavaScript — fetch (browser)</div>
|
||
<div class="card-body">
|
||
<pre><code>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`);</code></pre>
|
||
</div>
|
||
</div>
|
||
</div>
|
||
|
||
<footer>
|
||
Data Provider v1.0.0 · Chan Project · Served from {provider.exchange_name}
|
||
</footer>
|
||
</div>
|
||
|
||
<script>
|
||
// Simple tab switcher
|
||
document.querySelectorAll('#manualTabs .nav-link').forEach(btn => {{
|
||
btn.addEventListener('click', () => {{
|
||
document.querySelectorAll('#manualTabs .nav-link').forEach(b => b.classList.remove('active'));
|
||
document.querySelectorAll('.section').forEach(s => s.classList.remove('active'));
|
||
btn.classList.add('active');
|
||
document.getElementById('section-' + btn.dataset.section).classList.add('active');
|
||
}});
|
||
}});
|
||
</script>
|
||
</body>
|
||
</html>"""
|
||
|
||
@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="<h1>API 文档页面未找到</h1>", 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()
|
||
|