From f84cb671ef89c63b3152f1982e189bbf06b7d56c Mon Sep 17 00:00:00 2001 From: shy3130 <415333856@qq.com> Date: Tue, 1 Sep 2026 13:03:54 +0800 Subject: [PATCH] =?UTF-8?q?feat(data):=20=E5=85=A8=E9=87=8F=E5=88=86?= =?UTF-8?q?=E9=92=9F=E5=81=A5=E5=BA=B7=E6=97=B6=E7=AD=96=E7=95=A5=E9=A1=B5?= =?UTF-8?q?=E4=B8=8E=E7=9B=91=E6=8E=A7=E6=B3=A8=E5=85=A5=E5=88=87=E8=AF=BB?= =?UTF-8?q?=E6=9C=AC=E5=9C=B0=E5=88=86=E6=97=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 策略页: healthy 时解除分时列 100 只截断并 prefer_local 读本地分区 - 监控信号注入: 股票 healthy 时直接读本地当日分区, 免每分钟 bucket 一次的全量 API 拉取; ETF 不在服务 universe, 不健康/读失败回落原路径 - minute-refresh status 增加 healthy 读侧新鲜度字段 (前端共享判断) - 测试: 注入本地读 3 用例 (健康读本地/不健康回落/ETF 恒走 API) --- backend/app/services/minute_refresh.py | 3 +- backend/app/services/quote_service.py | 21 +++++- backend/tests/test_custom_provider_indices.py | 75 +++++++++++++++++++ frontend/src/lib/api.ts | 2 + frontend/src/pages/Screener.tsx | 16 +++- 5 files changed, 108 insertions(+), 9 deletions(-) diff --git a/backend/app/services/minute_refresh.py b/backend/app/services/minute_refresh.py index cc809d8..ce3b97f 100644 --- a/backend/app/services/minute_refresh.py +++ b/backend/app/services/minute_refresh.py @@ -298,7 +298,7 @@ class MinuteRefreshService: def is_healthy(self) -> bool: """读侧 freshness 判定: 本地 kline_minute 分区是否正被服务持续写入。 - 条件: 偏好开启 + 轮询线程存活 + 最近一轮距现在 ≤ max(2×间隔, 30s)。 + 条件: 偏好开启 + 轮询线程存活 + 最近一轮距现在 ≤ max(2*间隔, 30s)。 健康时消费方 (自选分时等) 可信任本地分区、跳过批量补拉; 连续失败轮不更新 last_round_at → 自动超时判不健康, 无需单独盯 last_error。 """ @@ -325,6 +325,7 @@ class MinuteRefreshService: return { "enabled": enabled, "running": running, + "healthy": self.is_healthy(), "interval_seconds": preferences.get_minute_refresh_interval(), "capability_ok": self.capability_ok(), "custom_provider_active": self.custom_provider_active(), diff --git a/backend/app/services/quote_service.py b/backend/app/services/quote_service.py index 3a43d13..812cfbb 100644 --- a/backend/app/services/quote_service.py +++ b/backend/app/services/quote_service.py @@ -1241,11 +1241,24 @@ class QuoteService: ) capset = getattr(self._app_state, "capabilities", None) - support = intraday_monitor_support(capset) - if not support["available"] or len(symbols) > int(support["max_symbols"]): - return self._intraday_signal_evaluator.inject(enriched, []) - minute_df = fetch_intraday_monitor_batch(sorted(symbols), capset, now=now) + # 全量分钟健康时股票读本地分区 (服务按间隔持续落盘, 与 API 同一列契约), + # 免去每分钟 bucket 一次的全量 API 拉取; ETF 不在服务 universe 内, + # 本地读空/异常回落原 API 路径 (含能力与上限检查) + minute_df = pl.DataFrame() + if asset_type == "stock": + svc = getattr(self._app_state, "minute_refresh", None) if self._app_state else None + if svc is not None and svc.is_healthy() and self._repo is not None: + try: + minute_df = self._repo.get_minute_batch(sorted(symbols), cn_today()) + except Exception as e: # 本地读异常回落 API + logger.warning("分时信号本地读失败, 回退 API 路径: %s", e) + minute_df = pl.DataFrame() + if minute_df.is_empty(): + support = intraday_monitor_support(capset) + if not support["available"] or len(symbols) > int(support["max_symbols"]): + return self._intraday_signal_evaluator.inject(enriched, []) + minute_df = fetch_intraday_monitor_batch(sorted(symbols), capset, now=now) prev_close: dict[str, float] = {} available_cols = set(enriched.columns) for row in enriched.filter(pl.col("symbol").is_in(sorted(symbols))).iter_rows(named=True): diff --git a/backend/tests/test_custom_provider_indices.py b/backend/tests/test_custom_provider_indices.py index 99fec24..1805539 100644 --- a/backend/tests/test_custom_provider_indices.py +++ b/backend/tests/test_custom_provider_indices.py @@ -11,6 +11,8 @@ from __future__ import annotations from types import SimpleNamespace from typing import ClassVar +import polars as pl + from app.services import quote_service as qs from app.services.index_const import CORE_INDEX_SYMBOLS @@ -114,3 +116,76 @@ def test_custom_provider_index_fetch_error_is_soft(monkeypatch): service, captured = _service_with_provider(monkeypatch, _Boom()) service._fetch_full_market_quotes() assert len(captured) == 1 and captured[0][0]["symbol"] == "600519.SH" + + +# ---- 监控分时注入: 全量分钟健康时股票读本地分区 ---- + + +def _injection_env(monkeypatch, *, healthy, local_df, asset_type="stock", symbols=None): + """构造 _inject_intraday_signals 最小环境, 捕获传入 evaluator 的 minute_df。""" + symbols = symbols or {"600519.SH"} + service = qs.QuoteService() + service._repo = SimpleNamespace( + get_minute_batch=lambda syms, d: local_df, + ) + engine = SimpleNamespace( + intraday_signal_symbols=lambda at: set(symbols) if at == asset_type else set(), + ) + minute_svc = SimpleNamespace(is_healthy=lambda: healthy) + service._app_state = SimpleNamespace(minute_refresh=minute_svc) + + import app.services.quote_service as qsm + api_calls: list[list[str]] = [] + monkeypatch.setattr( + qsm, "_noop", qsm.__dict__.get("_noop", None), raising=False) # 占位无操作 + from app.services.kline_sync import intraday_monitor_support + monkeypatch.setattr( + "app.services.quote_service.logger", qsm.logger, raising=False) + # 打桩 API 拉取路径 (健康时不应被调) + import app.services.kline_sync as ks + monkeypatch.setattr( + ks, "fetch_intraday_monitor_batch", + lambda symbols, capset, *, now=None: (api_calls.append(list(symbols)), local_df)[1]) + monkeypatch.setattr( + ks, "intraday_monitor_support", + lambda capset: {"available": True, "max_symbols": 200, "source": "minute_batch"}) + + evaluator = SimpleNamespace( + evaluate=lambda minute_df, **kw: (captured.append(minute_df), [])[1], + inject=lambda enriched, signals: enriched, + ) + captured: list = [] + service._intraday_signal_evaluator = evaluator + service._intraday_signal_bucket = {} + return service, engine, api_calls, captured + + +def test_intraday_signals_read_local_when_healthy(monkeypatch): + """健康时股票读本地分区, 不触发 API 拉取。""" + local = pl.DataFrame({ + "symbol": ["600519.SH"], "datetime": ["2026-01-15 09:31:00"], + "open": [100.0], "high": [101.0], "low": [99.0], "close": [100.5], + "volume": [1000.0], "amount": [100500.0], + }) + service, engine, api_calls, captured = _injection_env(monkeypatch, healthy=True, local_df=local) + enriched = pl.DataFrame({"symbol": ["600519.SH"], "close": [100.5]}) + service._inject_intraday_signals(enriched, engine, asset_type="stock") + assert api_calls == [] # 未走 API + assert captured and captured[0].height == 1 # evaluator 拿到本地数据 + + +def test_intraday_signals_fall_back_to_api_when_unhealthy(monkeypatch): + """不健康 (服务关/挂) 时回落原 API 拉取路径。""" + service, engine, api_calls, captured = _injection_env(monkeypatch, healthy=False, local_df=pl.DataFrame()) + enriched = pl.DataFrame({"symbol": ["600519.SH"], "close": [100.5]}) + service._inject_intraday_signals(enriched, engine, asset_type="stock") + assert api_calls == [["600519.SH"]] # 走了 API + + +def test_intraday_signals_etf_never_reads_local(monkeypatch): + """ETF 不在全量分钟 universe: 即使健康也走 API 路径。""" + service, engine, api_calls, captured = _injection_env( + monkeypatch, healthy=True, local_df=pl.DataFrame(), asset_type="etf", symbols={"510300.SH"}) + enriched = pl.DataFrame({"symbol": ["510300.SH"], "close": [4.0]}) + service._inject_intraday_signals(enriched, engine, asset_type="etf") + assert api_calls == [["510300.SH"]] diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 03dd8b8..f7f703e 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -1803,6 +1803,8 @@ export const api = { available: boolean enabled?: boolean running?: boolean + /** 读侧 freshness: 本地分区正被服务持续写入 (前端据此切本地读/解除截断) */ + healthy?: boolean interval_seconds?: number capability_ok?: boolean custom_provider_active?: boolean diff --git a/frontend/src/pages/Screener.tsx b/frontend/src/pages/Screener.tsx index 4fa37a1..c7f09ee 100644 --- a/frontend/src/pages/Screener.tsx +++ b/frontend/src/pages/Screener.tsx @@ -438,6 +438,14 @@ export function Screener() { ) // 分时图依赖分钟K批量数据 (kline.minute.batch), 无数据时开了列也不拉 const caps = useCapabilities() + // 全量分钟服务健康 (freshness 契约): 健康时本地分区按配置间隔持续落盘, + // 分时读本地不受批量上限约束 → 不截断 + prefer_local; 与监控设置页共享缓存 + const refreshStatus = useQuery({ + queryKey: ['minute-refresh-status'], + queryFn: api.minuteRefreshStatus, + refetchInterval: 15000, + }) + const fullMinuteHealthy = !!refreshStatus.data?.healthy const hasMinuteBatch = !!caps.data?.capabilities?.['kline.minute.batch'] const intradayVisible = !!intradayColumn && hasMinuteBatch && intradayChartVisible @@ -456,11 +464,11 @@ export function Screener() { () => displayRows.map((r: any) => r.symbol), [displayRows], ) - const intradayTruncated = intradayVisible && allIntradaySymbols.length > minuteBatchCap - // 截断到 batch 上限, 一次请求 = 一次数据源调用 + // 拉模型 (走批量接口) 才截断到 batch 上限; 全量分钟健康时读本地分区无上限 + const intradayTruncated = intradayVisible && !fullMinuteHealthy && allIntradaySymbols.length > minuteBatchCap const intradaySymbols = useMemo( () => intradayTruncated ? allIntradaySymbols.slice(0, minuteBatchCap) : allIntradaySymbols, - [allIntradaySymbols, intradayTruncated, minuteBatchCap], + [allIntradaySymbols, intradayTruncated, minuteBatchCap, fullMinuteHealthy], ) const intradayRequestSymbols = useMemo( () => [...new Set(intradaySymbols)].sort(), @@ -471,7 +479,7 @@ export function Screener() { const minuteBatch = useQuery({ queryKey: QK.minuteBatch(intradaySymbolsKey), // 增量轮询: 读缓存以最后一根为 since 只拉新增, 本地合并为完整序列 - queryFn: () => fetchMinuteBatchIncremental(qc, QK.minuteBatch(intradaySymbolsKey), intradayRequestSymbols), + queryFn: () => fetchMinuteBatchIncremental(qc, QK.minuteBatch(intradaySymbolsKey), intradayRequestSymbols, fullMinuteHealthy), enabled: intradayVisible && intradayRequestSymbols.length > 0, staleTime: 10_000, placeholderData: previousData => previousData,