diff --git a/backend/app/api/intraday.py b/backend/app/api/intraday.py index 988d476..6271b8b 100644 --- a/backend/app/api/intraday.py +++ b/backend/app/api/intraday.py @@ -1,9 +1,10 @@ """行情状态 / SSE 推送 API。 盘中选股相关端点已迁移至策略页面，此处仅保留全局行情基础设施。 -SSE 推送两种事件 (使用标准 SSE event 字段): +SSE 推送三种事件 (使用标准 SSE event 字段): - quotes_updated: 行情数据刷新，前端 invalidate 对应 query - strategy_alert: 策略监控/告警触发，前端弹通知 + - depth_updated: 五档盘口修正完成，前端刷新连板梯队/看板封单数据 """ from __future__ import annotations @@ -113,7 +114,7 @@ def index_quotes( @router.get("/stream") async def quote_stream(request: Request): - """SSE 端点: 行情更新 + 告警推送。 + """SSE 端点: 行情更新 + 告警推送 + 五档修正。 使用 sse-starlette EventSourceResponse: - 标准 SSE event 字段，前端按 event name 监听 @@ -124,16 +125,21 @@ async def quote_stream(request: Request): async def event_generator(): while True: - # 同时等待行情更新和告警 - update_task = asyncio.ensure_future( - asyncio.to_thread(qs.wait_for_update, timeout=5.0) if qs else asyncio.sleep(5) - ) - alert_task = asyncio.ensure_future( - asyncio.to_thread(qs.wait_for_alert, timeout=5.0) if qs else asyncio.sleep(5) - ) + # 同时等待三类信号: 行情更新 / 告警 / 五档修正 + tasks: dict[str, asyncio.Future] = { + "quote": asyncio.ensure_future( + asyncio.to_thread(qs.wait_for_update, timeout=5.0) if qs else asyncio.sleep(5) + ), + "alert": asyncio.ensure_future( + asyncio.to_thread(qs.wait_for_alert, timeout=5.0) if qs else asyncio.sleep(5) + ), + "depth": asyncio.ensure_future( + asyncio.to_thread(qs.wait_for_depth_update, timeout=5.0) if qs else asyncio.sleep(5) + ), + } done, pending = await asyncio.wait( - [update_task, alert_task], + list(tasks.values()), timeout=30.0, return_when=asyncio.FIRST_COMPLETED, ) @@ -154,22 +160,34 @@ async def quote_stream(request: Request): }, ensure_ascii=False), } - # 推送行情更新 - update_result = None - for t in done: + # 推送行情更新 (行情信号触发) + if tasks["quote"] in done: try: - update_result = t.result() + update_result = tasks["quote"].result() except Exception: # noqa: BLE001 - pass + update_result = False + if update_result: + yield { + "event": "quotes_updated", + "data": json.dumps({ + "ts": int(time.time() * 1000), + "symbol_count": qs._symbol_count if qs else 0, + }), + } - if update_result: - yield { - "event": "quotes_updated", - "data": json.dumps({ - "ts": int(time.time() * 1000), - "symbol_count": qs._symbol_count if qs else 0, - }), - } + # 推送五档修正完成 (depth 信号触发) — 前端刷新连板梯队封单数据 + if tasks["depth"] in done: + try: + depth_result = tasks["depth"].result() + except Exception: # noqa: BLE001 + depth_result = False + if depth_result: + yield { + "event": "depth_updated", + "data": json.dumps({ + "ts": int(time.time() * 1000), + }), + } return EventSourceResponse(event_generator()) diff --git a/backend/app/services/depth_service.py b/backend/app/services/depth_service.py index 9567e2d..4978a92 100644 --- a/backend/app/services/depth_service.py +++ b/backend/app/services/depth_service.py @@ -255,6 +255,9 @@ class DepthService: len(new_cache), len(syms_up), len(syms_down), enriched_date, " → 落盘" if persist else "") + # 缓存已更新: 通知 SSE 推 depth_updated, 触发连板梯队刷新封单数据。 + self._notify_depth_updated(len(new_cache)) + if persist and enriched_date: self._persist(enriched_date) @@ -533,6 +536,18 @@ class DepthService: except Exception as e: # noqa: BLE001 logger.debug("depth 接管通知推送失败: %s", e) + def _notify_depth_updated(self, count: int) -> None: + """修正完成通知: set quote_service._depth_update_event, SSE 推 depth_updated 刷新连板梯队。""" + if not self._app_state: + return + qs = getattr(self._app_state, "quote_service", None) + if not qs: + return + try: + qs.notify_depth_updated() + except Exception as e: # noqa: BLE001 + logger.debug("depth 更新通知推送失败: %s", e) + # ================================================================ # 工具 # ================================================================ diff --git a/backend/app/services/quote_service.py b/backend/app/services/quote_service.py index f4d4918..2fd3b7a 100644 --- a/backend/app/services/quote_service.py +++ b/backend/app/services/quote_service.py @@ -56,6 +56,7 @@ class QuoteService: self._repo = None # 延迟注入, 避免循环导入 self._update_event = threading.Event() # SSE 通知: 行情更新后 set self._alert_event = threading.Event() # SSE 通知: 有告警时 set + self._depth_update_event = threading.Event() # SSE 通知: depth 五档修正后 set (刷新连板梯队) self._pending_alerts: list[dict] = [] # 待推送的告警 self._max_pending_alerts: int = 1000 # 背压上限: 超出丢弃最旧 self._strategy_monitor = None # 延迟注入 @@ -152,6 +153,18 @@ class QuoteService: self._alert_event.clear() return self._alert_event.wait(timeout=timeout) + def notify_depth_updated(self) -> None: + """五档盘口修正完成后调用: 通知 SSE 推送 depth_updated, 触发连板梯队刷新。 + + 与行情/告警通道独立 — 只刷新连板梯队, 不连带刷新 watchlist 等。 + """ + self._depth_update_event.set() + + def wait_for_depth_update(self, timeout: float = 30.0) -> bool: + """阻塞等待 depth 修正 (供 SSE 线程使用)。""" + self._depth_update_event.clear() + return self._depth_update_event.wait(timeout=timeout) + def pop_alerts(self) -> list[dict]: """取走所有待推送的告警 (线程安全)。""" with self._lock: diff --git a/frontend/src/lib/useQuoteStream.ts b/frontend/src/lib/useQuoteStream.ts index 8b3718b..ae15e15 100644 --- a/frontend/src/lib/useQuoteStream.ts +++ b/frontend/src/lib/useQuoteStream.ts @@ -88,6 +88,13 @@ export function useQuoteStream( } }) + es.addEventListener('depth_updated', () => { + // 五档修正完成: 刷新连板梯队 + 看板封单数据。 + // 不受实时行情开关限制 — 修正轮询独立于行情轮询, 用户开了修正就想看实时封单。 + qc.invalidateQueries({ queryKey: ['limit-ladder'] }) + qc.invalidateQueries({ queryKey: ['overview-market'] }) + }) + es.addEventListener('strategy_alert', (e: MessageEvent) => { try { const data = JSON.parse(e.data) diff --git a/frontend/src/pages/LimitUpLadder.tsx b/frontend/src/pages/LimitUpLadder.tsx index acefb5a..5da55fd 100644 --- a/frontend/src/pages/LimitUpLadder.tsx +++ b/frontend/src/pages/LimitUpLadder.tsx @@ -762,7 +762,21 @@ function TierGroup({ tier, defaultOpen, extFields, filterKeys, bf, onStockClick, if (s === 'broken' || s === 'recovery') return 1 return 2 } - return ord(a.status ?? '') - ord(b.status ?? '') + const oa = ord(a.status ?? '') + const ob = ord(b.status ?? '') + if (oa !== ob) return oa - ob + // 同状态(主状态=涨停/跌停)内: 按封单从高到低排, 无封单排末尾。 + // 封单额 = sealed_vol(手) × 100 × close, 与展示口径一致。 + if (oa === 0) { + const sealVal = (s: typeof a) => { + if (s.sealed_vol == null) return -1 + return sealMode === 'amount' && s.close + ? s.sealed_vol * 100 * s.close + : s.sealed_vol + } + return sealVal(b) - sealVal(a) + } + return 0 }).map(s => (