From cf53953c8d71d8bd56b5bfc3206e8f791d6aed1b Mon Sep 17 00:00:00 2001 From: wshy Date: Wed, 8 Jul 2026 18:18:55 +0800 Subject: [PATCH] fix(realtime): ensure final quote sync and badge init (#79) --- backend/app/services/quote_service.py | 92 ++++++++++++++++++++++----- backend/uv.lock | 2 +- frontend/src/lib/api.ts | 4 ++ frontend/src/lib/monitorBadge.ts | 9 ++- frontend/src/lib/useSharedQueries.ts | 10 ++- 5 files changed, 90 insertions(+), 27 deletions(-) diff --git a/backend/app/services/quote_service.py b/backend/app/services/quote_service.py index 4f002a3..22252fd 100644 --- a/backend/app/services/quote_service.py +++ b/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: diff --git a/backend/uv.lock b/backend/uv.lock index 80a5e17..85a1f0f 100644 --- a/backend/uv.lock +++ b/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" }, diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 8e8183b..638d56e 100644 --- a/frontend/src/lib/api.ts +++ b/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: () => diff --git a/frontend/src/lib/monitorBadge.ts b/frontend/src/lib/monitorBadge.ts index be9ca52..65afc03 100644 --- a/frontend/src/lib/monitorBadge.ts +++ b/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) } diff --git a/frontend/src/lib/useSharedQueries.ts b/frontend/src/lib/useSharedQueries.ts index ee1dfba..c6dd175 100644 --- a/frontend/src/lib/useSharedQueries.ts +++ b/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, }) }