jack/Chan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

data_provider: 添加 ccxt.pro WebSocket 实时K线监听；端口 9009→80；web/*.sh 权限修正

Browse Source
This commit is contained in:
jackyu66git
2026-05-19 09:56:37 +08:00
parent f0ea6a6065
commit 5dc0c4cffd
9 changed files with 164 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
data_provider/Dockerfile
+2 -2
View File
@@ -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"]
data_provider/docker-compose.yml
+2 -2
View File
@@ -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"
data_provider/main.py
+160 -73
View File
@@ -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:
web/deploy.sh
Executable → Regular
View File
web/deploy_venv.sh
Executable → Regular
View File
web/status.sh
Executable → Regular
View File
web/status_venv.sh
Executable → Regular
View File
web/stop.sh
Executable → Regular
View File
web/stop_venv.sh
Executable → Regular
View File