diff --git a/backend/app/api/intraday.py b/backend/app/api/intraday.py index 51dcbff..593dfb6 100644 --- a/backend/app/api/intraday.py +++ b/backend/app/api/intraday.py @@ -1,8 +1,9 @@ """行情状态 / SSE 推送 API。 盘中选股相关端点已迁移至策略页面，此处仅保留全局行情基础设施。 -SSE 推送三种事件 (使用标准 SSE event 字段): +SSE 推送四种事件 (使用标准 SSE event 字段): - quotes_updated: 行情数据刷新，前端 invalidate 对应 query + - strategy_results_updated: 策略监控已写入最新结果，前端刷新策略个股列表 - strategy_alert: 策略监控/告警触发，前端弹通知 - depth_updated: 五档盘口修正完成，前端刷新连板梯队/看板封单数据 """ @@ -170,6 +171,13 @@ async def quote_stream(request: Request): }), } + # 策略监控完成后, 结果已写入内存缓存; 独立通知只刷新策略个股列表。 + if data["strategy_results_updated"]: + yield { + "event": "strategy_results_updated", + "data": json.dumps({"ts": int(time.time() * 1000)}), + } + # 五档修正完成 — 前端刷新连板梯队封单数据 if data["depth_updated"]: yield { diff --git a/backend/app/services/quote_service.py b/backend/app/services/quote_service.py index 0e80c87..09e4054 100644 --- a/backend/app/services/quote_service.py +++ b/backend/app/services/quote_service.py @@ -59,6 +59,7 @@ class QuoteSubscriber: self._max_alerts = max_alerts self._max_reviews = max_reviews self._quote_updated = False + self._strategy_results_updated = False self._depth_updated = False self._alerts: list[dict] = [] self._reviews: list[str] = [] @@ -73,11 +74,13 @@ class QuoteSubscriber: with self._lock: out = { "quote_updated": self._quote_updated, + "strategy_results_updated": self._strategy_results_updated, "depth_updated": self._depth_updated, "alerts": self._alerts, "reviews": self._reviews, } self._quote_updated = False + self._strategy_results_updated = False self._depth_updated = False self._alerts = [] self._reviews = [] @@ -102,7 +105,12 @@ class QuoteSubscriber: def clear_alerts(self) -> None: with self._lock: self._alerts = [] - if not self._quote_updated and not self._depth_updated and not self._reviews: + if ( + not self._quote_updated + and not self._strategy_results_updated + and not self._depth_updated + and not self._reviews + ): self._event.clear() def notify_quote(self) -> None: @@ -110,6 +118,11 @@ class QuoteSubscriber: self._quote_updated = True self._event.set() + def notify_strategy_results(self) -> None: + with self._lock: + self._strategy_results_updated = True + self._event.set() + def notify_depth(self) -> None: with self._lock: self._depth_updated = True @@ -321,6 +334,11 @@ class QuoteService: for sub in self._snapshot_subscribers(): sub.notify_quote() + def notify_strategy_results_updated(self) -> None: + """策略监控完成实时结果更新后调用，仅刷新策略页结果缓存。""" + for sub in self._snapshot_subscribers(): + sub.notify_strategy_results() + def notify_depth_updated(self) -> None: """五档盘口修正完成后调用: 通知 SSE 推送 depth_updated, 触发连板梯队刷新。 @@ -972,6 +990,8 @@ class QuoteService: if engine.has_rule_type("ladder"): eval_df = self._inject_sealed_vol(enriched_today, enriched_date) rule_events = engine.evaluate(eval_df, asset_type="stock") + if engine.consume_strategy_result_updates(): + self.notify_strategy_results_updated() # ETF 规则轮: 股票快照不含 ETF, 用 ETF enriched 快照单独评估。 # 独立 try —— ETF 轮任何异常都不得丢弃本轮已算出的股票告警。 # refresh=False —— 不在轮询线程上触发 ETF 冷缓存的同步重算 (缓存由 ETF 实时 diff --git a/backend/app/strategy/monitor.py b/backend/app/strategy/monitor.py index c407410..bdb8902 100644 --- a/backend/app/strategy/monitor.py +++ b/backend/app/strategy/monitor.py @@ -335,8 +335,15 @@ class MonitorRuleEngine: # ETF 版历史窗口加载器 (asset_type=etf 的规则用)。为 None 时 ETF filter_history 策略跳过。 self._history_loader_etf: Callable[[_dt.date, int], "pl.DataFrame"] | None = None # 本轮 evaluate() 产出的策略选股结果: strategy_id → {rows, total, as_of} - # 供策略页实时回显复用 (/api/screener/cached 端点直接读取此内存结果), 避免重跑 + # 供策略页实时回显复用 (/api/screener/cached 端点直接读取, 避免重跑)。 + # 注意: 始终是「完整」的 dict —— evaluate 重算时先写到 _building_strategy_results, + # 算完后整体替换此属性, 保证 /cached 并发读取永远拿到完整结果, 不会读到空中间态。 self._latest_strategy_results: dict[str, dict] = {} + # 本轮重算的临时容器 (_match_strategy 写入它); reset 轮开始时初始化为空 dict, + # evaluate 结束后一次性替换 _latest_strategy_results。 + self._building_strategy_results: dict[str, dict] = {} + # 本轮成功写入股票策略实时结果的策略 ID, 供 QuoteService 在计算完成后精确通知策略页。 + self._latest_strategy_result_ids: set[str] = set() def set_strategy_engine(self, engine) -> None: """注入 StrategyEngine, type=strategy 规则据此跑选股。""" @@ -422,6 +429,12 @@ class MonitorRuleEngine: """ return self._latest_strategy_results + def consume_strategy_result_updates(self) -> bool: + """返回并清除本轮成功写入的股票策略实时结果标记。""" + updated = bool(self._latest_strategy_result_ids) + self._latest_strategy_result_ids.clear() + return updated + def has_rule_type(self, rtype: str) -> bool: """是否存在指定类型的 (已启用) 规则。供 quote_service 判断是否需要注入特殊数据。""" if not self._rules: @@ -462,9 +475,13 @@ class MonitorRuleEngine: now = time.time() events: list[dict] = [] - # 每轮重置: 只保留本次 evaluate 产出的策略结果 + # 原子化: reset 轮 (股票轮) 时先把本轮结果写到临时容器, 算完后一次性替换 + # _latest_strategy_results。这样 /cached 并发读取永远拿到完整结果, + # 不会在「清空 → 逐个回填」窗口里读到空中间态 (曾导致策略页闪烁)。 + # 非 reset 轮 (ETF 轮) 继续往同一临时容器追加 (_match_strategy 仅写 stock, 实际不追加)。 if reset_strategy_results: - self._latest_strategy_results = {} + self._building_strategy_results = {} + self._latest_strategy_result_ids.clear() # list() 快照: 本方法跑在行情轮询线程, API 线程同时 add/remove 规则 # 会触发 "dictionary changed size during iteration", 整轮告警丢失 @@ -476,6 +493,10 @@ class MonitorRuleEngine: except Exception as e: logger.warning("规则评估失败 %s: %s", rule_id, e) + # 一次性提交本轮结果 (原子替换): /cached 读方要么拿到上一轮完整结果, + # 要么拿到本轮完整结果, 不会读到空中间态。 + self._latest_strategy_results = self._building_strategy_results + return events def _evaluate_rule(self, df: pl.DataFrame, rule: dict, now: float) -> list[dict]: @@ -659,10 +680,12 @@ class MonitorRuleEngine: # 记录本轮完整选股结果 (供策略页实时回显: /cached 端点直接读取, 不落盘)。 # 与下面的 diff 事件无关 — 无论是否产生 new_entry/dropped, 结果都该可用于回显。 # 策略结果缓存仅用于股票策略页 /cached 回显; ETF 策略页走实时单跑, 不写入。 + # 写到 evaluate 提供的临时容器 (_building_strategy_results), 算完后整体替换, + # 避免并发读到半填充状态。 if at == "stock": try: import math - self._latest_strategy_results[sid] = { + self._building_strategy_results[sid] = { "total": result.total, "as_of": str(cn_today()), "rows": [ @@ -671,6 +694,7 @@ class MonitorRuleEngine: for row in result.rows ], } + self._latest_strategy_result_ids.add(sid) except Exception: # noqa: BLE001 pass diff --git a/backend/tests/test_strategy_realtime_refresh.py b/backend/tests/test_strategy_realtime_refresh.py new file mode 100644 index 0000000..6da0a78 --- /dev/null +++ b/backend/tests/test_strategy_realtime_refresh.py @@ -0,0 +1,149 @@ +"""策略页实时结果刷新 SSE 回归测试。""" +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import patch + +import polars as pl + +from app.services import quote_service +from app.services.quote_service import QuoteService, QuoteSubscriber +from app.strategy.monitor import MonitorRuleEngine + + +def _strategy_rule(scope: str = "all") -> dict: + return { + "id": "strategy_rule", + "name": "策略监控", + "type": "strategy", + "asset_type": "stock", + "strategy_id": "strategy_1", + "scope": scope, + "symbols": ["600000.SH"], + "cooldown_seconds": 0, + } + + +def _quote_df() -> pl.DataFrame: + return pl.DataFrame({ + "symbol": ["600000.SH"], + "close": [10.0], + "change_pct": [0.01], + }) + + +def test_strategy_result_subscriber_notification_is_coalesced(): + sub = QuoteSubscriber() + + sub.notify_strategy_results() + sub.notify_strategy_results() + + assert sub.wait(timeout=0.01) is True + data = sub.pop() + assert data["strategy_results_updated"] is True + assert data["quote_updated"] is False + assert data["depth_updated"] is False + assert sub.wait(timeout=0.01) is False + + +def test_strategy_result_notification_fans_out_to_all_subscribers(): + service = QuoteService() + first = service.subscribe() + second = service.subscribe() + + service.notify_strategy_results_updated() + + assert first.pop()["strategy_results_updated"] is True + assert second.pop()["strategy_results_updated"] is True + + +class _EmptyResultStrategyEngine: + def get(self, strategy_id: str): + assert strategy_id == "strategy_1" + return SimpleNamespace(filter_history_fn=None) + + def run(self, strategy_id: str, **kwargs): + assert strategy_id == "strategy_1" + assert kwargs["precomputed"].height == 1 + return SimpleNamespace(total=0, rows=[]) + + +class _FailingStrategyEngine(_EmptyResultStrategyEngine): + def run(self, strategy_id: str, **kwargs): + raise RuntimeError("strategy failed") + + +def test_successful_zero_match_strategy_marks_result_refresh(): + engine = MonitorRuleEngine() + engine.set_strategy_engine(_EmptyResultStrategyEngine()) + engine.set_rules([_strategy_rule()]) + + assert engine.evaluate(_quote_df()) == [] + assert engine.latest_strategy_results()["strategy_1"]["total"] == 0 + assert engine.consume_strategy_result_updates() is True + assert engine.consume_strategy_result_updates() is False + + +def test_failed_or_skipped_strategy_does_not_mark_result_refresh(): + failed = MonitorRuleEngine() + failed.set_strategy_engine(_FailingStrategyEngine()) + failed.set_rules([_strategy_rule()]) + + assert failed.evaluate(_quote_df()) == [] + assert failed.latest_strategy_results() == {} + assert failed.consume_strategy_result_updates() is False + + skipped = MonitorRuleEngine() + skipped.set_strategy_engine(_EmptyResultStrategyEngine()) + skipped.set_rules([_strategy_rule(scope="symbols")]) + + assert skipped.evaluate(pl.DataFrame({"symbol": ["000001.SZ"]})) == [] + assert skipped.latest_strategy_results() == {} + assert skipped.consume_strategy_result_updates() is False + + +class _MonitorWithUpdate: + rule_count = 1 + + def __init__(self, updated: bool): + self.updated = updated + + def set_name_map(self, name_map): + pass + + def has_rule_type(self, rtype: str) -> bool: + return False + + def has_asset_rules(self, asset_type: str) -> bool: + return False + + def evaluate(self, df, asset_type: str): + assert asset_type == "stock" + return [] + + def consume_strategy_result_updates(self) -> bool: + return self.updated + + +def test_quote_service_notifies_only_after_strategy_result_update(): + service = QuoteService() + subscriber = service.subscribe() + service.set_app_state(SimpleNamespace(monitor_engine=_MonitorWithUpdate(updated=True))) + service.get_enriched_today = lambda: (_quote_df(), quote_service.cn_today()) + + with patch.object(QuoteService, "_is_continuous_trading", return_value=True): + service._evaluate_monitors(pl.DataFrame(), None) + + assert subscriber.pop()["strategy_results_updated"] is True + + +def test_quote_service_skips_notification_without_strategy_result_update(): + service = QuoteService() + subscriber = service.subscribe() + service.set_app_state(SimpleNamespace(monitor_engine=_MonitorWithUpdate(updated=False))) + service.get_enriched_today = lambda: (_quote_df(), quote_service.cn_today()) + + with patch.object(QuoteService, "_is_continuous_trading", return_value=True): + service._evaluate_monitors(pl.DataFrame(), None) + + assert subscriber.pop()["strategy_results_updated"] is False diff --git a/frontend/src/components/screener/StrategyCard.tsx b/frontend/src/components/screener/StrategyCard.tsx index 6f88000..9dc8547 100644 --- a/frontend/src/components/screener/StrategyCard.tsx +++ b/frontend/src/components/screener/StrategyCard.tsx @@ -129,7 +129,7 @@ export function StrategyCard({ {description && ( {description} )} - {count != null && ( + {count != null && !loading && (
{count} @@ -164,7 +164,7 @@ export function StrategyCard({
{srcLabel} {name} - {count != null && ( + {count != null && !loading && ( {count} )} {loading && } @@ -197,7 +197,7 @@ export function StrategyCard({ className="flex items-center gap-1 cursor-pointer disabled:opacity-50 disabled:cursor-wait"> {srcLabel} {name} - {count != null && ( + {count != null && !loading && ( {count} )} {hasExpired && ( diff --git a/frontend/src/lib/queryKeys.ts b/frontend/src/lib/queryKeys.ts index d2dadba..8f82dfb 100644 --- a/frontend/src/lib/queryKeys.ts +++ b/frontend/src/lib/queryKeys.ts @@ -85,6 +85,11 @@ export const QK = { // ===== SSE 应该 invalidate 的 key 前缀列表 ===== // 新增需要 SSE 推送的查询，只需在此加一行 +// +// 注意: 策略页 (screener-cached) 不在此列表 —— 行情刷新时策略结果不变 +// (非监控策略读盘后静态缓存, 监控策略由独立的 strategy_results_updated 事件在 +// 重算完成后刷新)。若加入 'screener', 会导致每个行情 tick 双重刷新策略页, +// 且在 monitor "重算" 窗口内读到空结果, 造成策略列表闪烁 (变 0 → 空失效 → 又出现)。 export const SSE_INVALIDATE_PREFIXES = [ 'watchlist', @@ -92,5 +97,4 @@ export const SSE_INVALIDATE_PREFIXES = [ 'index-quotes', 'overview-market', 'limit-ladder', - 'screener', ] as const diff --git a/frontend/src/lib/useQuoteStream.ts b/frontend/src/lib/useQuoteStream.ts index 95acce7..834d20f 100644 --- a/frontend/src/lib/useQuoteStream.ts +++ b/frontend/src/lib/useQuoteStream.ts @@ -140,6 +140,11 @@ export function useQuoteStream( } }) + es.addEventListener('strategy_results_updated', () => { + // 策略监控完成后只刷新策略结果缓存，不扩散到其他行情页面。 + qc.invalidateQueries({ queryKey: ['screener-cached'] }) + }) + es.addEventListener('depth_updated', () => { // 五档修正完成: 刷新连板梯队 + 看板封单数据。 // 不受实时行情开关限制 — 修正轮询独立于行情轮询, 用户开了修正就想看实时封单。 diff --git a/frontend/src/pages/Screener.tsx b/frontend/src/pages/Screener.tsx index dfff8ef..de6f7f7 100644 --- a/frontend/src/pages/Screener.tsx +++ b/frontend/src/pages/Screener.tsx @@ -643,7 +643,7 @@ export function Screener() { active={activeStrategy === s.id} count={hitCounts[id]} expiredCount={expiredCounts[id]} - loading={runAll.isPending && hitCounts[id] == null} + loading={runAll.isPending} cardSize={cardSize} onRun={() => handleRun(s)} disabled={run.isPending && activeStrategy === s.id}