From 865e75fc8b6c079d0b3347c63a730dc6b74c2a79 Mon Sep 17 00:00:00 2001 From: intfoo Date: Tue, 28 Jul 2026 21:13:10 +0800 Subject: [PATCH] =?UTF-8?q?fix(index):=20=E4=BF=AE=E5=A4=8D=20PR=20#46=20?= =?UTF-8?q?=E5=9B=9B=E4=B8=AA=E9=98=BB=E6=96=AD=E9=A1=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 阻断项1: 前端 TypeScript 类型扩展 - MonitorRule.asset_type 加 'index' (api.ts:517) - screenerStrategies 参数加 'index' (api.ts:1412) - klineMinute 响应 asset_type 去重 (api.ts:1296) 阻断项2: 指数监控独立评估 - _evaluate_monitors 股票早期 return 降级为 stock_ready 标志 仅跳过股票轮, ETF/指数轮独立判断数据新鲜度 - 纯指数行情/自选场景下指数规则可正常触发 阻断项3: 核心指数模式不截断分区 - _process_full_market_records 按 index_mode 条件分支: mode=all (完整 CN_Index) → flush 覆盖; mode=core (部分标的) → merge 不截断 - merge_live_enriched_asset 对 index 正确更新 _index_enriched_cache 阻断项4: Free 档额度分批 - _fetch_watchlist_quotes 用 resolve_limit + chunked 按 capability batch 上限分批 - 失败批次跳过不整轮退出, 已有股票实时刷新不受影响 - 复用进程级共享限速器 sleep_between_batches 测试: +7 测试覆盖 4 个阻断项核心场景 --- backend/app/services/quote_service.py | 66 ++++++++----- backend/tests/test_monitor_index.py | 94 ++++++++++++++++++ backend/tests/test_quote_index_merge.py | 78 +++++++++++++++ backend/tests/test_watchlist_batch.py | 123 ++++++++++++++++++++++++ frontend/src/lib/api.ts | 10 +- 5 files changed, 340 insertions(+), 31 deletions(-) create mode 100644 backend/tests/test_quote_index_merge.py create mode 100644 backend/tests/test_watchlist_batch.py diff --git a/backend/app/services/quote_service.py b/backend/app/services/quote_service.py index cf2ae35..5e6f3e2 100644 --- a/backend/app/services/quote_service.py +++ b/backend/app/services/quote_service.py @@ -720,16 +720,21 @@ class QuoteService: self._flush_live_enriched(daily_df, quote_extra, asset_type="stock") if not etf_daily_df.is_empty() and self._repo: self._flush_live_enriched(etf_daily_df, etf_quote_extra, asset_type="etf") - # ---- 指数: 仅有指数监控规则时才 flush 焐热 (无规则零成本) ---- + # ---- 指数: 仅有指数监控规则时才写盘 (无规则零成本) ---- + # mode=all (完整 CN_Index universe) → flush 覆盖; mode=core (部分标的) → merge 不截断分区 engine = getattr(self._app_state, "monitor_engine", None) if self._app_state else None if engine and engine.has_asset_rules("index") and self._repo: index_daily_df = self._build_daily(index_records) if not index_daily_df.is_empty(): + use_flush = preferences.get_realtime_index_mode() == "all" try: - self._repo.flush_live_daily_asset("index", index_daily_df) + if use_flush: + self._repo.flush_live_daily_asset("index", index_daily_df) + else: + self._repo.merge_live_daily_asset("index", index_daily_df) except Exception as e: # noqa: BLE001 logger.warning("指数日K写盘失败: %s", e) - self._flush_live_enriched(index_daily_df, self._build_quote_extra(index_records), asset_type="index") + self._flush_live_enriched(index_daily_df, self._build_quote_extra(index_records), asset_type="index", merge=not use_flush) # ---- 通知 SSE ---- self._broadcast_quote_updated() @@ -738,12 +743,15 @@ class QuoteService: self._evaluate_monitors(daily_df, quote_extra) def _fetch_watchlist_quotes(self) -> None: - """Free 档自选股实时: 只拉取最多 5 个 symbols。""" + """Free 档自选股实时: 按 capability batch 上限分批拉取。""" from app.services import preferences from app.tickflow.client import get_paid_realtime_client + from app.tickflow.capabilities import Cap + from app.tickflow.policy import detect_capabilities + from app.tickflow.rate_limits import chunked, resolve_limit, sleep_between_batches symbols = preferences.get_realtime_watchlist_symbols() - # 指数监控规则标的并入轮询 (独立于股票前5名额) + # 指数监控规则标的并入轮询 (与股票共享 batch 额度) engine = getattr(self._app_state, "monitor_engine", None) if self._app_state else None if engine: for _r in list(engine.rules.values()): @@ -760,13 +768,20 @@ class QuoteService: logger.warning("自选实时拉取失败:未配置付费服务器 API Key") return + # 按 capability batch 上限分批: 股票+指数共享额度, 超过上限会导致整轮失败 + capset = detect_capabilities() + lim = resolve_limit(capset, Cap.QUOTE_BY_SYMBOL, default_batch=5) + batches = chunked(symbols, lim.batch) + t0 = time.perf_counter() now_ts = time.perf_counter() - try: - resp = tf.quotes.get(symbols=symbols) or [] - except Exception as e: # noqa: BLE001 - logger.warning("自选实时拉取失败: %s", e) - return + resp = [] + for i, batch in enumerate(batches): + sleep_between_batches(i, lim.rpm) + try: + resp.extend(tf.quotes.get(symbols=batch) or []) + except Exception as e: # noqa: BLE001 + logger.warning("自选实时批次 %d/%d 拉取失败: %s", i + 1, len(batches), e) if not resp: logger.warning("自选实时行情数据为空") @@ -1050,14 +1065,12 @@ class QuoteService: return # 获取 enriched 数据 (刚算好的) enriched_today, enriched_date = self.get_enriched_today() - if enriched_today.is_empty(): - return - # 快照日期必须是北京当日: 节假日或数据未刷新时 enriched_date 会落后于当日, - # 说明市场未在交易 → 跳过。无需维护 A股交易日历即可挡住节假日与陈旧价告警。 - if enriched_date != cn_today(): - logger.debug("监控评估跳过: enriched 快照日期 %s 非当日 %s (节假日/数据未刷新)", - enriched_date, cn_today()) - return + # 股票快照就绪 = 非空 + 日期为当日。未就绪时仅跳过股票轮, + # ETF/指数轮有各自的空表+日期守卫, 不受影响 (纯指数行情/自选场景可独立评估)。 + stock_ready = (not enriched_today.is_empty()) and (enriched_date == cn_today()) + if not stock_ready: + logger.debug("股票快照未就绪(空=%s, 日期=%s), 跳过股票轮", + enriched_today.is_empty(), enriched_date) all_alerts: list[dict] = [] rule_events: list[dict] = [] @@ -1094,14 +1107,15 @@ class QuoteService: engine.set_name_map(name_map) except Exception as e: # noqa: BLE001 logger.debug("name_map 构建失败 (不影响监控): %s", e) - # 连板梯队封单监控: 有 ladder 规则时, 从 depth_service 注入封单量到 enriched - eval_df = enriched_today - if engine.has_rule_type("ladder"): - eval_df = self._inject_sealed_vol(enriched_today, enriched_date) - eval_df = self._inject_intraday_signals(eval_df, engine, "stock") - rule_events = engine.evaluate(eval_df, asset_type="stock") - if engine.consume_strategy_result_updates(): - self.notify_strategy_results_updated() + # 股票轮: 快照未就绪时跳过 (ladder 封单也依赖股票快照日期, 一并跳过) + if stock_ready: + eval_df = enriched_today + if engine.has_rule_type("ladder"): + eval_df = self._inject_sealed_vol(enriched_today, enriched_date) + eval_df = self._inject_intraday_signals(eval_df, engine, "stock") + 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/tests/test_monitor_index.py b/backend/tests/test_monitor_index.py index a812009..6e19ef4 100644 --- a/backend/tests/test_monitor_index.py +++ b/backend/tests/test_monitor_index.py @@ -106,3 +106,97 @@ def test_reconcile_index_asset_type_keeps_stock_and_mixed(): assert _reconcile_index_asset_type( {"asset_type": "stock", "scope": "all", "symbols": []}, repo, )["asset_type"] == "stock" + + +# ---- 股票快照为空时指数轮仍独立评估 (PR #46 问题 2) ---- + +def test_evaluate_monitors_index_round_survives_empty_stock_snapshot(): + """纯指数行情/自选场景: 股票 enriched 为空时, 指数监控轮仍独立评估。""" + from datetime import date + from unittest.mock import MagicMock, patch + + import polars as pl + + from app.services.quote_service import QuoteService + + svc = QuoteService.__new__(QuoteService) + svc._repo = MagicMock() + + engine = MagicMock() + engine.rule_count = 1 + engine.has_asset_rules.side_effect = lambda at: at == "index" + engine.has_rule_type.return_value = False + engine.evaluate.return_value = [] # 无触发, 简化后续 + + svc._app_state = MagicMock() + svc._app_state.monitor_engine = engine + svc._app_state.repo = svc._repo + + svc._repo.get_instruments.return_value = pl.DataFrame() + svc._repo.get_instruments_asset.return_value = pl.DataFrame() + svc._repo.get_enriched_latest_asset.return_value = ( + pl.DataFrame({"symbol": ["000001.SH"], "close": [3000.0], "rsi_14": [40.0]}), + date(2026, 7, 28), + ) + + with ( + patch.object(QuoteService, "_is_continuous_trading", return_value=True), + patch.object(QuoteService, "get_enriched_today", + return_value=(pl.DataFrame(), None)), # 股票快照为空 + patch.object(QuoteService, "_inject_intraday_signals", + side_effect=lambda df, e, at: df), + patch("app.services.quote_service.cn_today", return_value=date(2026, 7, 28)), + ): + svc._evaluate_monitors(pl.DataFrame(), None) + + # 指数轮执行了 (asset_type="index") + index_calls = [c for c in engine.evaluate.call_args_list + if c[1].get("asset_type") == "index"] + assert len(index_calls) == 1, "股票快照为空时指数轮仍应评估" + # 股票轮被跳过 (stock_ready=False) + stock_calls = [c for c in engine.evaluate.call_args_list + if c[1].get("asset_type") == "stock"] + assert len(stock_calls) == 0, "股票快照为空时股票轮应跳过" + + +def test_evaluate_monitors_stock_round_runs_when_snapshot_ready(): + """股票快照就绪时, 股票轮正常执行 (回归确认未破坏原有行为)。""" + from datetime import date + from unittest.mock import MagicMock, patch + + import polars as pl + + from app.services.quote_service import QuoteService + + svc = QuoteService.__new__(QuoteService) + svc._repo = MagicMock() + + engine = MagicMock() + engine.rule_count = 1 + engine.has_asset_rules.return_value = False + engine.has_rule_type.return_value = False + engine.evaluate.return_value = [] + engine.consume_strategy_result_updates.return_value = False + + svc._app_state = MagicMock() + svc._app_state.monitor_engine = engine + svc._app_state.repo = svc._repo + + svc._repo.get_instruments.return_value = pl.DataFrame() + + stock_df = pl.DataFrame({"symbol": ["600000.SH"], "close": [10.0], "rsi_14": [50.0]}) + + with ( + patch.object(QuoteService, "_is_continuous_trading", return_value=True), + patch.object(QuoteService, "get_enriched_today", + return_value=(stock_df, date(2026, 7, 28))), + patch.object(QuoteService, "_inject_intraday_signals", + side_effect=lambda df, e, at: df), + patch("app.services.quote_service.cn_today", return_value=date(2026, 7, 28)), + ): + svc._evaluate_monitors(pl.DataFrame(), None) + + # 股票轮正常执行 + stock_calls = [c for c in engine.evaluate.call_args_list + if c[1].get("asset_type") == "stock"] + assert len(stock_calls) == 1, "股票快照就绪时股票轮应正常执行" diff --git a/backend/tests/test_quote_index_merge.py b/backend/tests/test_quote_index_merge.py new file mode 100644 index 0000000..56815ea --- /dev/null +++ b/backend/tests/test_quote_index_merge.py @@ -0,0 +1,78 @@ +"""回归测试: 实时指数 merge 不截断盘后管道写入的全量分区 (PR #46 问题 3)。""" +from datetime import date + +import polars as pl + +from app.tickflow.repository import DataStore, KlineRepository + + +def _enriched_row(symbol: str, close: float, dt: date) -> dict: + return { + "symbol": symbol, "date": dt, + "open": close, "high": close, "low": close, "close": close, + "volume": 1000, "amount": 10000.0, + "quote_ts": 1753700400000, + } + + +def test_merge_live_enriched_preserves_full_index_partition(tmp_path): + """盘后管道 flush 写入全量指数后, 实时 merge 部分指数不丢已有数据。""" + repo = KlineRepository(DataStore(tmp_path)) + dt = date(2026, 7, 28) + + # 模拟盘后管道: flush 写入全量 3 只指数 + full_df = pl.DataFrame([ + _enriched_row("000001.SH", 3000.0, dt), + _enriched_row("399001.SZ", 10000.0, dt), + _enriched_row("399006.SZ", 2000.0, dt), + ]) + repo.flush_live_enriched_asset("index", full_df) + + # 模拟实时刷新: 只 merge 核心指数 1 只 (价格更新) + partial_df = pl.DataFrame([ + _enriched_row("000001.SH", 3001.0, dt), + ]) + repo.merge_live_enriched_asset("index", partial_df) + + # 验证: 分区文件仍有 3 只指数, 000001.SH 价格已更新, 其他指数未丢失 + out = tmp_path / "kline_index_enriched" / f"date={dt.isoformat()}" / "part.parquet" + result = pl.read_parquet(out) + assert len(result) == 3, f"merge 后分区应有 3 只指数, 实际 {len(result)}" + + sh = result.filter(pl.col("symbol") == "000001.SH") + assert sh["close"][0] == 3001.0, "merge 应更新 000001.SH 价格" + + sz = result.filter(pl.col("symbol") == "399001.SZ") + assert sz["close"][0] == 10000.0, "399001.SZ 不应被 merge 覆盖" + + cyb = result.filter(pl.col("symbol") == "399006.SZ") + assert cyb["close"][0] == 2000.0, "399006.SZ 不应被 merge 覆盖" + + +def test_merge_live_daily_preserves_full_index_partition(tmp_path): + """日K merge 同样不截断全量分区。""" + repo = KlineRepository(DataStore(tmp_path)) + dt = date(2026, 7, 28) + + # 盘后管道 flush 写入全量 3 只指数日K + full_df = pl.DataFrame([ + {"symbol": "000001.SH", "date": dt, "open": 3000.0, "high": 3010.0, + "low": 2990.0, "close": 3000.0, "volume": 1000, "amount": 10000.0}, + {"symbol": "399001.SZ", "date": dt, "open": 10000.0, "high": 10010.0, + "low": 9990.0, "close": 10000.0, "volume": 2000, "amount": 20000.0}, + {"symbol": "399006.SZ", "date": dt, "open": 2000.0, "high": 2010.0, + "low": 1990.0, "close": 2000.0, "volume": 3000, "amount": 30000.0}, + ]) + repo.flush_live_daily_asset("index", full_df) + + # 实时 merge 部分指数 + partial_df = pl.DataFrame([ + {"symbol": "000001.SH", "date": dt, "open": 3000.0, "high": 3010.0, + "low": 2990.0, "close": 3001.0, "volume": 1000, "amount": 10000.0}, + ]) + repo.merge_live_daily_asset("index", partial_df) + + out = tmp_path / "kline_index_daily" / f"date={dt.isoformat()}" / "part.parquet" + result = pl.read_parquet(out) + assert len(result) == 3, f"merge 后分区应有 3 只指数, 实际 {len(result)}" + assert result.filter(pl.col("symbol") == "000001.SH")["close"][0] == 3001.0 diff --git a/backend/tests/test_watchlist_batch.py b/backend/tests/test_watchlist_batch.py new file mode 100644 index 0000000..251d240 --- /dev/null +++ b/backend/tests/test_watchlist_batch.py @@ -0,0 +1,123 @@ +"""回归测试: Free 档自选实时 symbols 超过 capability batch 上限时分批请求 (PR #46 问题 4)。""" +from contextlib import ExitStack +from unittest.mock import MagicMock, patch + +import polars as pl + +from app.services.quote_service import QuoteService +from app.tickflow.capabilities import Cap, CapabilityLimits, CapabilitySet + + +def _make_svc(engine_rules: dict) -> QuoteService: + """创建最小可用的 QuoteService 实例 (跳过 __init__)。""" + svc = QuoteService.__new__(QuoteService) + svc._app_state = MagicMock() + svc._repo = MagicMock() + svc._lock = MagicMock() + + engine = MagicMock() + engine.rules = engine_rules + svc._app_state.monitor_engine = engine + svc._app_state.repo = svc._repo + + svc._repo.get_index_symbol_set.return_value = {"000001.SH"} + svc._repo.get_etf_symbol_set.return_value = set() + return svc + + +def _run_fetch(svc, tf, watchlist: list[str], capset: CapabilitySet): + """在完整 patch 环境下执行 _fetch_watchlist_quotes。""" + with ExitStack() as stack: + stack.enter_context(patch( + "app.services.preferences.get_realtime_watchlist_symbols", + return_value=watchlist, + )) + stack.enter_context(patch( + "app.tickflow.client.get_paid_realtime_client", return_value=tf, + )) + stack.enter_context(patch( + "app.tickflow.policy.detect_capabilities", return_value=capset, + )) + stack.enter_context(patch("app.tickflow.rate_limits.sleep_between_batches")) + # patch 分批之后的下游处理 + stack.enter_context(patch.object( + QuoteService, "_build_daily", return_value=pl.DataFrame(), + )) + stack.enter_context(patch.object( + QuoteService, "_build_quote_extra", return_value=pl.DataFrame(), + )) + stack.enter_context(patch.object( + QuoteService, "_build_index_quotes", return_value=pl.DataFrame(), + )) + stack.enter_context(patch.object(QuoteService, "_broadcast_quote_updated")) + stack.enter_context(patch.object(QuoteService, "_evaluate_monitors")) + stack.enter_context(patch("app.services.quote_service._persist_last_fetch")) + svc._fetch_watchlist_quotes() + + +def test_watchlist_batch_respects_capability_limit(): + """6 symbols / batch 5 → 分 2 批请求, 不整轮失败。""" + engine_rules = { + "r_idx": {"enabled": True, "asset_type": "index", "scope": "symbols", + "symbols": ["000001.SH"]}, + } + svc = _make_svc(engine_rules) + + tf = MagicMock() + tf.quotes.get.return_value = [ + {"symbol": "600000.SH", "last_price": 10.0, "prev_close": 9.9, "ext": {}}, + ] + capset = CapabilitySet({Cap.QUOTE_BY_SYMBOL: CapabilityLimits(batch=5, rpm=60)}) + + _run_fetch(svc, tf, + ["600000.SH", "600001.SH", "600002.SH", "600003.SH", "600004.SH"], + capset) + + # 5 股票 + 1 指数 = 6 symbols, batch 5 → 2 批 + assert tf.quotes.get.call_count == 2 + first_batch = tf.quotes.get.call_args_list[0][1]["symbols"] + second_batch = tf.quotes.get.call_args_list[1][1]["symbols"] + assert len(first_batch) == 5 + assert len(second_batch) == 1 + assert "000001.SH" in second_batch + + +def test_watchlist_batch_partial_failure_keeps_other_batches(): + """某一批拉取失败不影响其他批次 (已有股票实时刷新不丢失)。""" + engine_rules = { + "r_idx": {"enabled": True, "asset_type": "index", "scope": "symbols", + "symbols": ["000001.SH"]}, + } + svc = _make_svc(engine_rules) + + tf = MagicMock() + # 第一批 (股票) 成功, 第二批 (指数) 失败 + tf.quotes.get.side_effect = [ + [{"symbol": "600000.SH", "last_price": 10.0, "prev_close": 9.9, "ext": {}}], + ConnectionError("timeout"), + ] + capset = CapabilitySet({Cap.QUOTE_BY_SYMBOL: CapabilityLimits(batch=5, rpm=60)}) + + _run_fetch(svc, tf, + ["600000.SH", "600001.SH", "600002.SH", "600003.SH", "600004.SH"], + capset) + + # 两批都被尝试 (第二批失败不阻断) + assert tf.quotes.get.call_count == 2 + + +def test_watchlist_no_index_rules_no_extra_symbols(): + """无指数监控规则时, symbols 不追加指数标的。""" + svc = _make_svc({}) # 无规则 + + tf = MagicMock() + tf.quotes.get.return_value = [ + {"symbol": "600000.SH", "last_price": 10.0, "prev_close": 9.9, "ext": {}}, + ] + capset = CapabilitySet({Cap.QUOTE_BY_SYMBOL: CapabilityLimits(batch=5, rpm=60)}) + + _run_fetch(svc, tf, ["600000.SH", "600001.SH"], capset) + + # 2 symbols / batch 5 → 1 批 + assert tf.quotes.get.call_count == 1 + assert tf.quotes.get.call_args_list[0][1]["symbols"] == ["600000.SH", "600001.SH"] diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 3902ad4..deab1f1 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -514,7 +514,7 @@ export interface MonitorRule { name: string enabled: boolean type: 'strategy' | 'signal' | 'price' | 'market' | 'ladder' - asset_type?: 'stock' | 'etf' + asset_type?: 'stock' | 'etf' | 'index' scope: 'symbols' | 'all' | 'sector' symbols: string[] sector?: string | null @@ -1409,7 +1409,7 @@ export const api = { : '/api/watchlist/enriched', ), - screenerStrategies: async (assetType?: 'stock' | 'etf') => { + screenerStrategies: async (assetType?: 'stock' | 'etf' | 'index') => { const data = await request<{ strategies: StrategyDetail[]; load_errors?: StrategyLoadError[] }>( `/api/strategies?${assetType ? `asset_type=${assetType}&` : ''}timeframe=1d`, ) @@ -1473,7 +1473,7 @@ export const api = { stop_loss_pct?: number max_hold_days?: number matching?: 'close_t' | 'open_t+1' - asset_type?: 'stock' | 'etf' + asset_type?: 'stock' | 'etf' | 'index' }) => request('/api/backtest/run', { method: 'POST', @@ -1493,7 +1493,7 @@ export const api = { weight?: 'equal' | 'factor_weight' fees_pct?: number slippage_bps?: number - asset_type?: 'stock' | 'etf' + asset_type?: 'stock' | 'etf' | 'index' }) => request('/api/backtest/factor/run', { method: 'POST', @@ -1517,7 +1517,7 @@ export const api = { max_positions?: number initial_capital?: number position_sizing?: 'equal' | 'score_weight' - asset_type?: 'stock' | 'etf' + asset_type?: 'stock' | 'etf' | 'index' minute_fill?: boolean }) => request('/api/backtest/strategy/run', {