web端进行优化，减少内存开销，data provider提供websocket服务

data_provider/main.py

@@ -17,7 +17,7 @@ from typing import Dict, Iterable, List, Optional
 
 import ccxt  # type: ignore
 import pandas as pd  # type: ignore
-from fastapi import FastAPI, HTTPException, Query
+from fastapi import FastAPI, HTTPException, Query, WebSocket, WebSocketDisconnect
 from fastapi.middleware.cors import CORSMiddleware
 import uvicorn
 from technical.util import resample_to_interval
@@ -47,6 +47,7 @@ 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")
@@ -122,6 +123,82 @@ def timeframe_to_minutes(tf: str) -> Optional[int]:
     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、内存缓存、断线恢复与衍生周期聚合。"""
 
@@ -165,6 +242,7 @@ class DataProvider:
         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 配置文件。"""
@@ -458,6 +536,18 @@ class DataProvider:
         # 同步写盘
         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():
@@ -499,6 +589,7 @@ class DataProvider:
                                 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线
@@ -513,6 +604,7 @@ class DataProvider:
                                 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
@@ -664,11 +756,38 @@ class DataProvider:
 
 
 def create_app(provider: DataProvider) -> FastAPI:
-    """构造 FastAPI 应用：lifespan 内同步 initialize 并启动后台拉数。"""
+    """构造 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:
@@ -734,6 +853,74 @@ def create_app(provider: DataProvider) -> FastAPI:
             "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