From 5dc0c4cffd511ede15bfc0f719d2c3e5b5f7365f Mon Sep 17 00:00:00 2001 From: jackyu66git Date: Tue, 19 May 2026 09:56:37 +0800 Subject: [PATCH] =?UTF-8?q?data=5Fprovider:=20=E6=B7=BB=E5=8A=A0=20ccxt.pr?= =?UTF-8?q?o=20WebSocket=20=E5=AE=9E=E6=97=B6K=E7=BA=BF=E7=9B=91=E5=90=AC?= =?UTF-8?q?=EF=BC=9B=E7=AB=AF=E5=8F=A3=209009=E2=86=9280=EF=BC=9Bweb/*.sh?= =?UTF-8?q?=20=E6=9D=83=E9=99=90=E4=BF=AE=E6=AD=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- data_provider/Dockerfile | 4 +- data_provider/docker-compose.yml | 4 +- data_provider/main.py | 233 +++++++++++++++++++++---------- web/deploy.sh | 0 web/deploy_venv.sh | 0 web/status.sh | 0 web/status_venv.sh | 0 web/stop.sh | 0 web/stop_venv.sh | 0 9 files changed, 164 insertions(+), 77 deletions(-) mode change 100755 => 100644 web/deploy.sh mode change 100755 => 100644 web/deploy_venv.sh mode change 100755 => 100644 web/status.sh mode change 100755 => 100644 web/status_venv.sh mode change 100755 => 100644 web/stop.sh mode change 100755 => 100644 web/stop_venv.sh diff --git a/data_provider/Dockerfile b/data_provider/Dockerfile index d07a269..6f63ba9 100644 --- a/data_provider/Dockerfile +++ b/data_provider/Dockerfile @@ -12,9 +12,9 @@ COPY . /app ENV CONFIG_PATH=/app/config.json \ UVICORN_HOST=0.0.0.0 \ - UVICORN_PORT=9009 + UVICORN_PORT=80 -EXPOSE 9009 +EXPOSE 80 CMD ["python", "-m", "main"] diff --git a/data_provider/docker-compose.yml b/data_provider/docker-compose.yml index 4906e8f..72bfe85 100644 --- a/data_provider/docker-compose.yml +++ b/data_provider/docker-compose.yml @@ -6,10 +6,10 @@ services: environment: CONFIG_PATH: /app/config.json UVICORN_HOST: 0.0.0.0 - UVICORN_PORT: "9009" + UVICORN_PORT: "80" volumes: - ./config.json:/app/config.json:ro - ./data:/app/data ports: - - "9009:9009" + - "80:80" diff --git a/data_provider/main.py b/data_provider/main.py index 37d4235..91f5652 100644 --- a/data_provider/main.py +++ b/data_provider/main.py @@ -16,6 +16,7 @@ from pathlib import Path from typing import Dict, Iterable, List, Optional import ccxt # type: ignore +import ccxt.pro as ccxt_pro # type: ignore import pandas as pd # type: ignore from fastapi import FastAPI, HTTPException, Query, WebSocket, WebSocketDisconnect from fastapi.middleware.cors import CORSMiddleware @@ -247,9 +248,10 @@ class DataProvider: 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._watch_tasks: List = [] # asyncio Task — 每个基础周期一个 watch + self._watch_loop_task = None # master watch 协程 self._cold_start_done = False # 记录断线后需要从哪个 since 重新拉取（symbol -> timeframe -> since_ms） self._resume_since: Dict[str, Dict[str, int]] = {} @@ -309,7 +311,8 @@ class DataProvider: return unique def _init_exchange(self): - """实例化 ccxt 交易所，币安期货默认 defaultType=future，并 load_markets。""" + """实例化两个交易所实例：REST 用 ccxt，WebSocket 用 ccxt.pro。""" + # REST 实例：供冷启动回填、历史补数、定时落盘等同步代码使用 if not hasattr(ccxt, self.exchange_name): raise ValueError(f"不支持的交易所: {self.exchange_name}") exchange_class = getattr(ccxt, self.exchange_name) @@ -317,9 +320,141 @@ class DataProvider: if exchange.id == "binance": exchange.options.setdefault("defaultType", "future") exchange.load_markets() - logger.info("已初始化交易所 %s", exchange.id) + logger.info("已初始化交易所 %s (ccxt REST)", exchange.id) + + # WebSocket 实例：供实时 K 线监听使用 + if not hasattr(ccxt_pro, self.exchange_name): + raise ValueError(f"不支持的交易所: {self.exchange_name} (ccxt.pro)") + ws_exchange_class = getattr(ccxt_pro, self.exchange_name) + self._ws_exchange = ws_exchange_class({"newUpdates": True}) + if self._ws_exchange.id == "binance": + self._ws_exchange.options.setdefault("defaultType", "future") + logger.info("已初始化交易所 %s (ccxt.pro, WebSocket 已启用)", self._ws_exchange.id) + return exchange + def _record_resume_for_timeframe(self, tf: str) -> None: + """WebSocket 断线时为该周期所有 symbol 记录恢复点，避免 gap。""" + for symbol in self.symbols: + with self._lock: + current = self.data.get(symbol, {}).get(tf, []) + if current: + last_ts = int(current[-1]["timestamp"]) + else: + last_ts = self._get_start_time(tf) + tf_ms = TIMEFRAME_TO_MS[tf] + since_ms = max(self._get_start_time(tf), last_ts - tf_ms) + existing = self._get_resume_since(symbol, tf) + if existing is None: + self._set_resume_since(symbol, tf, since_ms) + + async def _backfill_resume_data(self) -> None: + """启动 WebSocket watch 前的历史补数，通过 REST 补齐 resume_since gap。""" + loop = asyncio.get_running_loop() + for symbol in self.symbols: + for tf in self.timeframes: + if self._stop_event.is_set(): + return + resume_since = self._get_resume_since(symbol, tf) + if resume_since is None: + continue + logger.info("WebSocket 启动前补数: %s %s since=%s", symbol, tf, + to_utc_iso(resume_since)) + try: + history = await loop.run_in_executor( + None, self._fetch_history, symbol, tf, resume_since, + ) + with self._lock: + current = self.data.setdefault(symbol, {}).get(tf, []) + merged = self._merge_candles(tf, current, history) + self.data[symbol][tf] = merged + self._notify_update(symbol, tf) + self._clear_resume_since(symbol, tf) + self._trim(symbol, tf) + except Exception as exc: + logger.error("启动前补数失败 %s %s: %s", symbol, tf, exc) + + async def _watch_single_tf(self, tf: str) -> None: + """单个基础周期的 WebSocket 监听协程。 + + 循环调用 watch_ohlcv_for_symbols，新 K 线到达后从 exchange.ohlcvs 缓存 + 读取最新数据合并到内存并广播。 + """ + pairs = [[symbol, tf] for symbol in self.symbols] + backoff = 2.0 + + while not self._stop_event.is_set(): + try: + await self._ws_exchange.watch_ohlcv_for_symbols(pairs) + + # watch_ohlcv_for_symbols 更新了 exchange.ohlcvs 缓存， + # 从缓存中读取每个 symbol 的最新 candle 合并到本地 + for symbol in self.symbols: + ohlcv_cache = self._ws_exchange.ohlcvs.get(symbol, {}).get(tf) + if ohlcv_cache is None: + continue + cached = list(ohlcv_cache) + if not cached: + continue + + with self._lock: + current = self.data.setdefault(symbol, {}).get(tf, []) + if current and current[-1]["timestamp"] >= cached[-1][0]: + continue # 没有新数据 + merged = self._merge_candles(tf, current, cached) + self.data[symbol][tf] = merged + + self._notify_update(symbol, tf) + self._trim(symbol, tf) + + backoff = 2.0 # 成功后重置 + + except ccxt.NetworkError as exc: + logger.warning("WebSocket 网络错误 (tf=%s): %s，%.1fs 后重连", tf, exc, backoff) + self._record_resume_for_timeframe(tf) + await asyncio.sleep(backoff) + backoff = min(backoff * 1.5, 60.0) + + except ccxt.BaseError as exc: + logger.error("WebSocket 交易所错误 (tf=%s): %s，%.1fs 后重连", tf, exc, backoff) + self._record_resume_for_timeframe(tf) + await asyncio.sleep(backoff) + backoff = min(backoff * 1.5, 60.0) + + except asyncio.CancelledError: + logger.info("WebSocket 监听 (tf=%s) 被取消", tf) + break + + except Exception: + logger.exception("WebSocket 监听 (tf=%s) 未预期错误", tf) + self._record_resume_for_timeframe(tf) + await asyncio.sleep(backoff) + backoff = min(backoff * 1.5, 60.0) + + logger.info("WebSocket 监听 (tf=%s) 已退出", tf) + + async def _watch_loop(self) -> None: + """主 WebSocket 监听协程：先启动 watch 任务，再并发补历史 gap。""" + for tf in self.timeframes: + task = asyncio.create_task(self._watch_single_tf(tf), name=f"watch-{tf}") + self._watch_tasks.append(task) + + logger.info("WebSocket 监听已启动: %d 个时间周期, %d 个任务", + len(self.timeframes), len(self._watch_tasks)) + + # 补历史数据与 watch 任务并发进行，不阻塞实时数据接收 + try: + await self._backfill_resume_data() + except Exception: + logger.exception("启动前补数失败") + + try: + await asyncio.gather(*self._watch_tasks, return_exceptions=True) + except asyncio.CancelledError: + pass + finally: + logger.info("所有 WebSocket 监听任务已结束") + def _data_file_path(self, symbol: str, timeframe: str) -> Path: """单交易对单周期的 CSV 路径：data_dir/tf/exchange_symbol_tf.csv。""" symbol_safe = symbol.replace("/", "_").replace(":", "_") @@ -665,86 +800,43 @@ class DataProvider: logger.error("数据更新回调异常: %s", exc) def start_background_workers(self) -> None: - """启动后台线程：冷启动回填、增量刷新、周期性落盘。""" + """启动后台线程：冷启动回填 + 周期性落盘。WebSocket 监听由 lifespan 异步启动。""" self._stop_event.clear() self._backfill_thread = threading.Thread(target=self._cold_start_backfill, name="backfill-loop", daemon=True) - self._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("后台线程已启动（回填 + 增量刷新 + 落盘）") + logger.info("后台线程已启动（回填 + 落盘）") + + def start_watch_tasks(self) -> None: + """在当前 asyncio event loop 上启动 WebSocket 监听。必须在 lifespan 内调用。""" + loop = asyncio.get_running_loop() + self._watch_loop_task = loop.create_task(self._watch_loop(), name="watch-master") + logger.info("WebSocket 监听主任务已创建") def stop(self) -> None: - """停止后台线程（应用关闭时 lifespan finally 调用）。""" + """停止后台线程和异步 WebSocket 监听任务。""" self._stop_event.set() + + # 取消 WebSocket watch 任务（它们检查 stop_event 后会退出） + for task in getattr(self, '_watch_tasks', []): + if not task.done(): + task.cancel() + if hasattr(self, '_watch_loop_task') and self._watch_loop_task is not None: + if not self._watch_loop_task.done(): + self._watch_loop_task.cancel() + + # Join 所有后台线程 for thread, name in [ (self._backfill_thread, "backfill"), - (self._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 + logger.info("数据提供商已停止") def _persist_loop(self) -> None: """每隔 PERSIST_INTERVAL 秒把内存快照写 CSV 并保存恢复点。""" @@ -922,15 +1014,10 @@ def create_app(provider: DataProvider) -> 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() + # 启动 WebSocket 实时监听（asyncio 后台任务） + provider.start_watch_tasks() try: yield finally: diff --git a/web/deploy.sh b/web/deploy.sh old mode 100755 new mode 100644 diff --git a/web/deploy_venv.sh b/web/deploy_venv.sh old mode 100755 new mode 100644 diff --git a/web/status.sh b/web/status.sh old mode 100755 new mode 100644 diff --git a/web/status_venv.sh b/web/status_venv.sh old mode 100755 new mode 100644 diff --git a/web/stop.sh b/web/stop.sh old mode 100755 new mode 100644 diff --git a/web/stop_venv.sh b/web/stop_venv.sh old mode 100755 new mode 100644