""" 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 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 线 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 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 # 记录断线后需要从哪个 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 _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 _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 分页拉取直到接近当前时间；遇限频则 sleep 重试。""" results: List[List[float]] = [] limit = 1500 now_ms = self.exchange.milliseconds() tf_ms = TIMEFRAME_TO_MS[timeframe] fetch_since = since_ms max_rounds = 5000 rounds = 0 while fetch_since < now_ms and rounds < max_rounds: rounds += 1 try: candles = self.exchange.fetch_ohlcv( symbol, timeframe=timeframe, since=fetch_since, limit=limit, ) except ccxt.RateLimitExceeded as exc: logger.warning("触发频率限制，等待: %s", exc) time.sleep(self.exchange.rateLimit / 1000 if self.exchange.rateLimit else 1) continue except ccxt.BaseError as exc: logger.error("拉取历史K线失败 (%s, %s): %s", timeframe, fetch_since, exc) time.sleep(5) continue if not candles: break results.extend(candles) last_ts = candles[-1][0] fetch_since = last_ts + tf_ms if last_ts >= now_ms - tf_ms: break time.sleep(self.exchange.rateLimit / 1000 if self.exchange.rateLimit else 0.2) logger.info("交易对 %s 时间周期 %s 拉取历史K线数量: %s", symbol, timeframe, len(results)) return results def initialize(self) -> None: """阻塞式启动：加载本地、从倒数第二根或配置起点补历史、写盘并 set _ready。""" logger.info("开始初始化数据提供商") for symbol in self.symbols: for timeframe in self.timeframes: existing = self._load_local(symbol, timeframe) tf_ms = TIMEFRAME_TO_MS[timeframe] last_ts = existing[-1]["timestamp"] if existing else None if last_ts is not None: if len(existing) >= 2: # 从倒数第二根起拉，避免最后一根未收盘重复/缺口 fetch_since = existing[-2]["timestamp"] else: fetch_since = max(0, last_ts - tf_ms) else: fetch_since = self.start_time_ms logger.debug( "初始化拉取参数", extra={ "symbol": symbol, "timeframe": timeframe, "existing_last": last_ts, "fetch_since": fetch_since, "tf_ms": tf_ms, }, ) history = self._fetch_history(symbol, timeframe, fetch_since) merged = self._merge_candles(timeframe, existing, history) with self._lock: self.data.setdefault(symbol, {})[timeframe] = merged self._write_to_disk(symbol, timeframe, merged) self._ready.set() logger.info("数据初始化完成") 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: """启动增量刷新线程与周期性落盘线程。""" if self._fetch_thread and self._fetch_thread.is_alive(): return self._stop_event.clear() 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._fetch_thread.start() self._persist_thread.start() logger.info("后台线程已启动") def stop(self) -> None: """停止后台线程（应用关闭时 lifespan finally 调用）。""" self._stop_event.set() if self._fetch_thread: self._fetch_thread.join(timeout=5) if self._persist_thread: self._persist_thread.join(timeout=5) 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) 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) 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.start_time_ms tf_ms = TIMEFRAME_TO_MS[timeframe] # 回退一个周期，确保包含可能未完全收盘的K线，去重由 _merge_candles 处理 since_ms = max(self.start_time_ms, 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 后落盘，避免长时间持锁。""" 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(): 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 裁剪。""" with self._lock: candles = list(self.data.get(symbol, {}).get(timeframe, [])) 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() 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 @app.get("/") async def root() -> Dict[str, object]: """根路径：服务名、交易所、交易对与可用周期（含 ready 标志）。""" return { "service": "Data Provider", "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.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) 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", "9009")) uvicorn.run(app, host=host, port=port, log_level=os.getenv("UVICORN_LOG_LEVEL", "info")) if __name__ == "__main__": main()