fix(realtime): 策略页实时行情闪烁 + 策略结果刷新机制

策略页开启实时行情时, 被监控的策略列表反复闪烁 (变 0 → 全部失效 → 又出现),
非监控策略不受影响。

根因: 每个行情周期后端先广播 quotes_updated 再重算策略, 重算时先清空内存结果
再逐个回填 (非原子窗口)。前端 quotes_updated 与 strategy_results_updated 两个
SSE 事件都刷新 screener-cached, 第一次撞上清空窗口拿到空结果, 第二次拿到重算
结果, 每周期重复 → 闪烁。

修复:
- 前端: 从 SSE_INVALIDATE_PREFIXES 移除 screener, quotes_updated 不再刷新策略页,
  每周期只剩 strategy_results_updated (重算完成后才发) 触发一次刷新。
- 后端: monitor.evaluate 改为临时容器收集结果, 算完后整体替换 _latest_strategy_results,
  /cached 并发读取永远拿到完整结果, 不会读到空中间态。

配套:
- 新增 strategy_results_updated SSE 事件 + subscriber 合并通知机制
- 策略卡片 loading 时不显示旧命中数, 避免刷新时数字跳动

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 {

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 实时

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
 

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

frontend/src/components/screener/StrategyCard.tsx

@@ -129,7 +129,7 @@ export function StrategyCard({
             {description && (
               <span className="text-[10px] text-muted leading-tight mt-0.5 line-clamp-1">{description}</span>
             )}
-            {count != null && (
+            {count != null && !loading && (
               <div className="mt-1.5 flex items-center gap-2">
                 <div className="flex items-center gap-1">
                   <span className={`text-sm font-mono font-bold tabular-nums ${countCls}`}>{count}</span>
@@ -164,7 +164,7 @@ export function StrategyCard({
             <div className="flex items-center gap-1.5 min-w-0">
               <span className={`text-[9px] px-1 py-px rounded border font-medium leading-tight shrink-0 ${badgeCls}`}>{srcLabel}</span>
               <span className="text-xs font-medium truncate text-foreground">{name}</span>
-              {count != null && (
+              {count != null && !loading && (
                 <span className={`text-xs font-mono font-bold tabular-nums shrink-0 ${countCls}`}>{count}</span>
               )}
               {loading && <span className="w-5 h-3 rounded bg-elevated animate-pulse shrink-0" />}
@@ -197,7 +197,7 @@ export function StrategyCard({
             className="flex items-center gap-1 cursor-pointer disabled:opacity-50 disabled:cursor-wait">
             <span className="text-[8px] px-0.5 rounded bg-secondary/10 text-muted border border-border font-medium leading-tight">{srcLabel}</span>
             <span className="text-[10px] font-medium whitespace-nowrap text-foreground">{name}</span>
-            {count != null && (
+            {count != null && !loading && (
               <span className={`text-xs font-mono font-bold tabular-nums ${countCls}`}>{count}</span>
             )}
             {hasExpired && (

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

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', () => {
         // 五档修正完成: 刷新连板梯队 + 看板封单数据。
         // 不受实时行情开关限制 — 修正轮询独立于行情轮询, 用户开了修正就想看实时封单。

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}