fix(realtime): ensure final quote sync and badge init (#79)

backend/app/services/quote_service.py

@@ -152,6 +152,9 @@ class QuoteService:
         self._index_symbol_count: int = 0
         self._etf_symbol_count: int = 0
         self._index_quotes_cache: pl.DataFrame | None = None
+        # 午休/收盘最终同步状态: 到边界后必须成功拉取一版行情, 再进入休盘态。
+        self._final_sync_done: set[tuple[date, str]] = set()
+        self._final_sync_failed: dict[tuple[date, str], str] = {}
 
     # ================================================================
     # 生命周期
@@ -381,6 +384,10 @@ class QuoteService:
         from app.services import preferences
         age = (time.perf_counter() - self._fetch_time) * 1000 if self._fetch_time else -1
         mode = self.realtime_mode()
+        phase = self._market_phase()
+        final_key = self._final_sync_key(phase)
+        final_done = bool(final_key and final_key in self._final_sync_done)
+        final_failed = self._final_sync_failed.get(final_key) if final_key else None
         return {
             "enabled": self._enabled,
             "running": self._running,
@@ -392,7 +399,12 @@ class QuoteService:
             "index_symbol_count": self._index_symbol_count,
             "etf_symbol_count": self._etf_symbol_count,
             "quote_age_ms": round(age, 0) if age >= 0 else None,
-            "is_trading_hours": self._is_trading_hours(),
+            # 交易时段 = 连续竞价; polling_window 另行返回,避免午休/收盘缓冲误显示为交易中。
+            "is_trading_hours": self._is_continuous_trading(),
+            "is_polling_window": self._should_poll_for_phase(phase),
+            "market_phase": phase,
+            "final_sync_done": final_done,
+            "final_sync_failed": final_failed,
             "last_fetch_ms": round(self._fetched_at, 0) if self._fetched_at else None,
         }
 
@@ -408,10 +420,21 @@ class QuoteService:
     def _poll_loop(self) -> None:
         while self._running and self._enabled:
             try:
-                if self._is_trading_hours():
-                    self._fetch_quotes()
+                phase = self._market_phase()
+                if self._should_fetch_for_phase(phase):
+                    is_final = phase in {"morning_final", "close_final"}
+                    ok = self._fetch_quotes(final=is_final)
+                    if is_final:
+                        key = self._final_sync_key(phase)
+                        if key and ok:
+                            self._final_sync_done.add(key)
+                            self._final_sync_failed.pop(key, None)
+                            logger.info("%s 最终行情同步完成, 进入休盘态", "午休" if phase == "morning_final" else "收盘")
+                        elif key:
+                            self._final_sync_failed[key] = "fetch_failed"
+                            logger.warning("%s 最终行情同步失败, 将继续重试", "午休" if phase == "morning_final" else "收盘")
                 else:
-                    logger.debug("非交易时段, 跳过行情轮询")
+                    logger.debug("非轮询阶段(%s), 跳过行情轮询", phase)
             except Exception as e:  # noqa: BLE001
                 logger.warning("行情轮询异常: %s", e)
 
@@ -420,13 +443,17 @@ class QuoteService:
                 time.sleep(0.5)
                 waited += 0.5
 
-    def _fetch_quotes(self) -> None:
-        """按当前档位拉取行情。加锁串行化 (后台轮询 vs 手动 refresh)。"""
+    def _fetch_quotes(self, *, final: bool = False) -> bool:
+        """按当前档位拉取行情。加锁串行化 (后台轮询 vs 手动 refresh)。返回本轮是否成功更新。"""
         with self._fetch_lock:
+            before = self._fetched_at
+            if final:
+                logger.info("最终行情同步开始")
             if self.realtime_mode() == "watchlist":
                 self._fetch_watchlist_quotes()
-                return
-            self._fetch_full_market_quotes()
+            else:
+                self._fetch_full_market_quotes()
+            return self._fetched_at > before
 
     def _fetch_full_market_quotes(self) -> None:
         """拉取全市场行情 → 写 daily + 计算 enriched + 更新缓存。"""
@@ -763,15 +790,50 @@ class QuoteService:
         return df
 
     @staticmethod
-    def _is_trading_hours() -> bool:
-        # 显式北京时间: 容器/服务器本地时区可能是 UTC, 用 naive now() 会整体错开轮询窗口
-        # 注: 这是**轮询**窗口(含 9:15 集合竞价与 15:05 收盘缓冲, 用于盘前预热/收盘捕捉),
-        # 比连续竞价宽。监控告警用更严格的 _is_continuous_trading。
+    def _market_phase() -> str:
+        """A股行情轮询阶段(北京时间)。
+
+        final 阶段用于午休/收盘定版: 需要至少成功拉取一版边界后的行情, 才算进入休盘。
+        """
         now = cn_now()
+        if now.weekday() >= 5:
+            return "closed"
         t = now.time()
-        morning = dt_time(9, 15) <= t <= dt_time(11, 35)
-        afternoon = dt_time(12, 55) <= t <= dt_time(15, 5)
-        return now.weekday() < 5 and (morning or afternoon)
+        if dt_time(9, 15) <= t < dt_time(9, 30):
+            return "preopen"
+        if dt_time(9, 30) <= t < dt_time(11, 30):
+            return "morning"
+        if dt_time(11, 30) <= t < dt_time(12, 55):
+            return "morning_final"
+        if dt_time(12, 55) <= t < dt_time(13, 0):
+            return "pre_afternoon"
+        if dt_time(13, 0) <= t < dt_time(15, 0):
+            return "afternoon"
+        if t >= dt_time(15, 0):
+            return "close_final"
+        return "closed"
+
+    @staticmethod
+    def _final_sync_key(phase: str) -> tuple[date, str] | None:
+        if phase == "morning_final":
+            return (cn_today(), "morning")
+        if phase == "close_final":
+            return (cn_today(), "close")
+        return None
+
+    def _should_poll_for_phase(self, phase: str) -> bool:
+        """是否处于会主动拉行情的阶段。final 阶段成功后即停止。"""
+        if phase in {"preopen", "morning", "pre_afternoon", "afternoon"}:
+            return True
+        key = self._final_sync_key(phase)
+        return bool(key and key not in self._final_sync_done)
+
+    def _should_fetch_for_phase(self, phase: str) -> bool:
+        return self._should_poll_for_phase(phase)
+
+    def _is_trading_hours(self) -> bool:
+        """行情轮询窗口(兼容旧调用): 包含盘前预热和未完成的午休/收盘定版。"""
+        return self._should_poll_for_phase(self._market_phase())
 
     @staticmethod
     def _is_continuous_trading() -> bool:

backend/uv.lock

@@ -2491,7 +2491,7 @@ all = [
 
 [[package]]
 name = "tickflow-stock-panel-backend"
-version = "0.1.81"
+version = "0.1.82"
 source = { editable = "." }
 dependencies = [
     { name = "apscheduler" },

frontend/src/lib/api.ts

@@ -945,6 +945,10 @@ export const api = {
       etf_symbol_count?: number
       quote_age_ms: number | null
       is_trading_hours: boolean
+      is_polling_window?: boolean
+      market_phase?: string
+      final_sync_done?: boolean
+      final_sync_failed?: string | null
       last_fetch_ms: number | null
     }>('/api/intraday/status'),
   quoteInterval: () =>

frontend/src/lib/monitorBadge.ts

@@ -48,17 +48,16 @@ function subscribe(fn: () => void) {
 }
 
 function getSnapshot() {
-  return Math.max(0, currentTotal - lastSeenTotal)
+  return Math.max(0, currentTotal - Math.max(0, lastSeenTotal))
 }
 
 /** 轮询更新最新总数 (Layout 层调用)。 */
 export function setCurrentTotal(total: number): void {
-  // total=0 视为未初始化, 不更新 (避免渲染期 data=undefined 传 0 重置 lastSeen)
-  if (total <= 0) return
+  if (total < 0) return
 
-  // 首次初始化: lastSeen <= 0 (从未设置 -1, 或历史为 0) → 把已读基线设为当前总数
+  // 首次初始化: lastSeen < 0 (从未设置) → 把已读基线设为当前总数
   // 否则 lastSeen=0 + total=1 会被误算成"1条未读" (首次进入就显示徽标的 bug)
-  if (lastSeenTotal <= 0) {
+  if (lastSeenTotal < 0) {
     lastSeenTotal = total
     writeSeen(total)
   }

frontend/src/lib/useSharedQueries.ts

@@ -36,9 +36,9 @@ export function usePreferences() {
 
 /** 行情状态 — SSE quotes_updated 自动刷新。
 
- * poll=true 时启用条件轮询兜底: 仅在非交易时段每 60s 轮询一次,
- * 用于在交易时段边界 (11:30午休 / 12:55开盘 / 15:05收盘) 同步 is_trading_hours。
- * 交易时段不轮询 (SSE 已驱动刷新), 非交易时段无 SSE 推送, 需要兜底。
+ * poll=true 时启用 60s 状态轮询兜底, 用于在交易时段边界
+ * (11:30午休 / 13:00开盘 / 15:00收盘) 同步 quote status。
+ * SSE 会在行情更新时即时刷新, 轮询负责没有 SSE 的休盘边界。
  * 只应在全局唯一挂载处 (Layout) 传 poll=true, 避免多页面重复轮询;
  * 其他调用方共享同一 queryKey 缓存, 无需自行轮询。
  */
@@ -47,9 +47,7 @@ export function useQuoteStatus(opts?: { enabled?: boolean; poll?: boolean }) {
     queryKey: QK.quoteStatus,
     queryFn: api.quoteStatus,
     enabled: opts?.enabled ?? true,
-    refetchInterval: opts?.poll
-      ? (query) => (query.state.data?.is_trading_hours ? false : 60_000)
-      : false,
+    refetchInterval: opts?.poll ? 60_000 : false,
   })
 }