fix(index): 修复 PR #46 四个阻断项

阻断项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

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

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, "股票快照就绪时股票轮应正常执行"

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

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"]