fix(index): preserve realtime cache on provider failure

CONTRIBUTING.md

@@ -140,7 +140,7 @@
 - 每个能力独立路由，禁止跟随/派生特殊值（`same_as_daily` 已下线）；存量非法偏好值由 preferences getter 回退默认自愈，不做迁移。
 - 边界注记：分时监控由分钟能力兜底（`intraday_monitor_support`），不单设分时能力；`full_minute`（全量分钟）数据集已开放插件/自定义源声明；`depth5` 已进矩阵但插件数据集白名单暂未开放，当前仅 TickFlow 提供。
 - 实时指数为产品级固定契约，不走路由矩阵：展示层（侧栏指数条、市场总览）固定核心四只（`backend/app/services/index_const.py` 单一权威：上证/深成/创业板/科创综指），后端各消费方与前端 Layout 引用同一份定义不建副本；指数页保留但标的固定为核心四只（无全指数搜索/浏览，`/api/index/list`、`/api/index/search` 已下线）；侧栏指数多选配置已下线，相关偏好（`realtime_index_symbols`/`sidebar_index_symbols`/`indices_nav_pinned`/`realtime_pull_index`/`realtime_index_mode`）已删除。监控规则的指数标的不受限——quote_service 把核心四只 + 启用规则的指数并入显式拉取。
-- 自定义源指数补充协议：A 股快照普遍不含指数（fuyao 实测无指数，指数在其独立端点）。provider 可实现可选方法 `get_realtime_indices(symbols) -> list[dict]`（record 结构与 realtime 一致），quote_service 在自定义源分支鸭子类型调用补拉；未实现的源指数缓存为空，由本地日K兜底接管。fuyao 指数快照有连坐语义——请求混入未知代码整批失败，插件侧必须先行过滤不支持的后缀（如 `.BJ`）。
+- 自定义源指数补充协议：A 股快照普遍不含指数（fuyao 实测无指数，指数在其独立端点）。provider 可实现可选方法 `get_realtime_indices(symbols) -> list[dict] | None`（record 结构与 realtime 一致），quote_service 在自定义源分支鸭子类型调用补拉；`None` 表示请求失败，保留上轮有效指数缓存，`[]` 表示成功但无数据；未实现的源指数缓存为空，由本地日K兜底接管。fuyao 指数快照有连坐语义——请求混入未知代码整批失败，插件侧必须先行过滤不支持的后缀（如 `.BJ`）。
 
 ## 5. 领域专项要求
 

backend/app/plugins/fuyao/provider.py

@@ -401,11 +401,12 @@ class FuyaoProvider:
         logger.info("扶摇实时行情拉取完成: %d 条(丢弃 %d 行)", len(records), dropped)
         return records
 
-    def get_realtime_indices(self, symbols: list[str]) -> list[dict]:
+    def get_realtime_indices(self, symbols: list[str]) -> list[dict] | None:
         """指数实时快照 → 内部 realtime record (可选插件协议, quote_service 鸭子类型调用)。
 
         A 股快照不含指数, 指数在扶摇是独立端点; 覆盖沪深交易所指数 + 同花顺板块,
-        无北交所 (未知代码会整批 1002 连坐, .BJ 直接跳过)。失败软返回空列表。
+        无北交所 (未知代码会整批 1002 连坐, .BJ 直接跳过)。失败返回 None,
+        让上层与“成功但无数据”的空列表区分, 保留上轮有效指数缓存。
         """
         wanted = [s for s in symbols if s and not s.upper().endswith(".BJ")]
         if not wanted:
@@ -414,7 +415,7 @@ class FuyaoProvider:
             rows, server_ts = self._get_client().index_snapshot(wanted)
         except FuyaoError as e:
             logger.warning("扶摇指数行情拉取失败: %s", e)
-            return []
+            return None
 
         fetched_ms = server_ts or int(time.time() * 1000)
         records = []

backend/app/services/quote_service.py

@@ -624,17 +624,28 @@ class QuoteService:
                     # 指数补充: A 股快照通常不含指数。插件可选实现
                     # get_realtime_indices(symbols) 用独立端点补拉 (如 fuyao 指数快照);
                     # 未实现的源指数缓存为空, 由日K兜底接管。
+                    replace_index_cache = True
                     fetch_indices = getattr(provider, "get_realtime_indices", None)
                     if callable(fetch_indices):
                         wanted = sorted(set(CORE_INDEX_SYMBOLS) | self._collect_monitor_index_symbols())
                         try:
-                            records = records + (fetch_indices(wanted) or [])
+                            fetched_indices = fetch_indices(wanted)
+                            if fetched_indices is None:
+                                replace_index_cache = False
+                            else:
+                                records = records + fetched_indices
                         except Exception as e:  # noqa: BLE001
                             logger.warning("自定义源指数行情拉取失败: %s", e)
+                            replace_index_cache = False
                 except Exception as e:  # noqa: BLE001
                     logger.warning("自定义实时行情拉取失败: %s", e)
                     return
-                self._process_full_market_records(records, t0=t0, now_ts=now_ts)
+                self._process_full_market_records(
+                    records,
+                    t0=t0,
+                    now_ts=now_ts,
+                    replace_index_cache=replace_index_cache,
+                )
                 return
             # 自定义源未配置 realtime → 回退 TickFlow
 
@@ -721,7 +732,14 @@ class QuoteService:
 
         self._process_full_market_records(records, t0=t0, now_ts=now_ts)
 
-    def _process_full_market_records(self, records: list[dict], *, t0: float, now_ts: float) -> None:
+    def _process_full_market_records(
+        self,
+        records: list[dict],
+        *,
+        t0: float,
+        now_ts: float,
+        replace_index_cache: bool = True,
+    ) -> None:
         """把全市场 records 写盘并增量计算 enriched。"""
         from app.services import preferences
         all_index_symbols = set(self._repo.get_index_symbol_set()) if self._repo else set()
@@ -753,9 +771,12 @@ class QuoteService:
             self._fetch_ms = fetch_ms
             self._fetched_at = fetched_at
             self._symbol_count = len(stock_records)
-            self._index_symbol_count = len(index_records)
             self._etf_symbol_count = len(etf_records)
-            self._index_quotes_cache = self._build_index_quotes(index_records)
+            if replace_index_cache:
+                self._index_symbol_count = len(index_records)
+                self._index_quotes_cache = self._build_index_quotes(index_records)
+            else:
+                logger.info("指数本轮获取失败,沿用上轮缓存: %d 只", self._index_symbol_count)
 
         _persist_last_fetch(fetched_at)
         logger.info("行情刷新: %d 只股票, %d 只ETF, %d 只指数, 耗时 %.0fms", len(stock_records), len(etf_records), len(index_records), fetch_ms)

backend/tests/test_custom_provider_indices.py

@@ -8,6 +8,7 @@ get_realtime_indices(symbols) 补拉指数 — A 股快照普遍不含指数
 
 from __future__ import annotations
 
+import time
 from types import SimpleNamespace
 from typing import ClassVar
 
@@ -49,12 +50,15 @@ def _index_rec(symbol: str) -> dict:
     return {"symbol": symbol, "last_price": 3986.3, "prev_close": 3952.2, "change_pct": 0.0086}
 
 
-def _service_with_provider(monkeypatch, provider) -> tuple[qs.QuoteService, list[list[dict]]]:
+def _service_with_provider(
+    monkeypatch, provider,
+) -> tuple[qs.QuoteService, list[list[dict]], list[bool]]:
     """构造最小 QuoteService: 自定义源路由 + 捕获 _process_full_market_records 入参。"""
     from app.services import preferences as prefs_mod
 
     service = qs.QuoteService()
     captured: list[list[dict]] = []
+    index_cache_replacements: list[bool] = []
     monkeypatch.setattr(prefs_mod, "get_realtime_data_provider", lambda: "fuyao")
     import app.data_providers.custom as custom_mod
 
@@ -62,19 +66,23 @@ def _service_with_provider(monkeypatch, provider) -> tuple[qs.QuoteService, list
     monkeypatch.setattr(custom_mod, "get_provider", lambda name: provider)
     monkeypatch.setattr(
         service, "_process_full_market_records",
-        lambda records, *, t0, now_ts: captured.append(records),
+        lambda records, *, t0, now_ts, replace_index_cache=True: (
+            captured.append(records),
+            index_cache_replacements.append(replace_index_cache),
+        ),
     )
-    return service, captured
+    return service, captured, index_cache_replacements
 
 
 def test_custom_provider_fetch_appends_index_records(monkeypatch):
     provider = _FakeProvider([_stock_rec()], [_index_rec("000001.SH"), _index_rec("399001.SZ")])
-    service, captured = _service_with_provider(monkeypatch, provider)
+    service, captured, replacements = _service_with_provider(monkeypatch, provider)
     service._fetch_full_market_quotes()
 
     assert len(captured) == 1
     symbols = [r["symbol"] for r in captured[0]]
     assert "600519.SH" in symbols and "000001.SH" in symbols and "399001.SZ" in symbols
+    assert replacements == [True]
     # 请求清单 = 核心四只 (无指数监控规则时)
     assert provider.index_calls == [sorted(CORE_INDEX_SYMBOLS)]
 
@@ -82,7 +90,7 @@ def test_custom_provider_fetch_appends_index_records(monkeypatch):
 def test_custom_provider_monitor_indices_join_fetch(monkeypatch):
     """指数监控规则标的并入请求清单 (quote_service._collect_monitor_index_symbols)。"""
     provider = _FakeProvider([_stock_rec()], [_index_rec("000300.SH")])
-    service, _captured = _service_with_provider(monkeypatch, provider)
+    service, _captured, _replacements = _service_with_provider(monkeypatch, provider)
 
     class _Engine:
         rules: ClassVar[dict] = {
@@ -99,9 +107,10 @@ def test_custom_provider_monitor_indices_join_fetch(monkeypatch):
 
 def test_custom_provider_without_indices_protocol_is_silent(monkeypatch):
     """未实现 get_realtime_indices 的源: 个股 records 照常, 指数不补充不报错。"""
-    service, captured = _service_with_provider(monkeypatch, _ProviderNoIndices())
+    service, captured, replacements = _service_with_provider(monkeypatch, _ProviderNoIndices())
     service._fetch_full_market_quotes()
     assert captured == [[{"symbol": "600519.SH", "last_price": 1480.0}]]
+    assert replacements == [True]
 
 
 def test_custom_provider_index_fetch_error_is_soft(monkeypatch):
@@ -113,9 +122,78 @@ def test_custom_provider_index_fetch_error_is_soft(monkeypatch):
         def get_realtime_indices(self, symbols: list[str]) -> list[dict]:
             raise RuntimeError("index endpoint down")
 
-    service, captured = _service_with_provider(monkeypatch, _Boom())
+    service, captured, replacements = _service_with_provider(monkeypatch, _Boom())
     service._fetch_full_market_quotes()
     assert len(captured) == 1 and captured[0][0]["symbol"] == "600519.SH"
+    assert replacements == [False]
+
+
+def test_custom_provider_index_fetch_failure_preserves_cache(monkeypatch):
+    """None 表示指数请求失败: 股票继续更新, 但不得替换上一轮指数缓存。"""
+    class _Unavailable:
+        def get_realtime(self) -> list[dict]:
+            return [_stock_rec()]
+
+        def get_realtime_indices(self, symbols: list[str]) -> None:
+            return None
+
+    service, captured, replacements = _service_with_provider(monkeypatch, _Unavailable())
+    service._fetch_full_market_quotes()
+
+    assert captured == [[_stock_rec()]]
+    assert replacements == [False]
+
+
+def test_custom_provider_successful_empty_index_fetch_replaces_cache(monkeypatch):
+    """空 list 是成功响应: 与失败 None 区分, 仍按现有语义替换缓存。"""
+    service, captured, replacements = _service_with_provider(
+        monkeypatch, _FakeProvider([_stock_rec()], []),
+    )
+    service._fetch_full_market_quotes()
+
+    assert captured == [[_stock_rec()]]
+    assert replacements == [True]
+
+
+def _disable_record_processing_side_effects(monkeypatch, service: qs.QuoteService) -> None:
+    monkeypatch.setattr(qs, "_persist_last_fetch", lambda fetched_at: None)
+    monkeypatch.setattr(service, "_update_volume_delta", lambda records, fetched_at: None)
+    monkeypatch.setattr(service, "_broadcast_quote_updated", lambda: None)
+    monkeypatch.setattr(service, "_evaluate_monitors", lambda daily, extra: None)
+
+
+def test_failed_index_refresh_keeps_last_known_good_cache(monkeypatch):
+    service = qs.QuoteService()
+    _disable_record_processing_side_effects(monkeypatch, service)
+    cached = service._build_index_quotes([_index_rec("000001.SH")])
+    service._index_quotes_cache = cached
+    service._index_symbol_count = cached.height
+
+    service._process_full_market_records(
+        [_stock_rec()],
+        t0=time.perf_counter(),
+        now_ts=time.perf_counter(),
+        replace_index_cache=False,
+    )
+
+    assert service._index_symbol_count == 1
+    assert service.get_index_quotes().to_dicts() == cached.to_dicts()
+
+
+def test_successful_empty_index_refresh_clears_cache(monkeypatch):
+    service = qs.QuoteService()
+    _disable_record_processing_side_effects(monkeypatch, service)
+    service._index_quotes_cache = service._build_index_quotes([_index_rec("000001.SH")])
+    service._index_symbol_count = 1
+
+    service._process_full_market_records(
+        [_stock_rec()],
+        t0=time.perf_counter(),
+        now_ts=time.perf_counter(),
+    )
+
+    assert service._index_symbol_count == 0
+    assert service.get_index_quotes().is_empty()
 
 
 # ---- 监控分时注入: 全量分钟健康时股票读本地分区 ----

backend/tests/test_fuyao_provider.py

@@ -319,11 +319,11 @@ def test_realtime_indices_skips_bj_symbols(monkeypatch):
     assert fake.calls == [["000001.SH"]]  # 全 .BJ 时根本不发请求
 
 
-def test_realtime_indices_error_returns_empty(monkeypatch):
+def test_realtime_indices_error_returns_none(monkeypatch):
     provider, _ = _index_provider_with(
         monkeypatch, error=fc.FuyaoError("扶摇接口错误 code=1002: Unknown thscode")
     )
-    assert provider.get_realtime_indices(["000001.SH"]) == []
+    assert provider.get_realtime_indices(["000001.SH"]) is None
 
 
 def test_client_requires_api_key():

docs/plugin-development.md

@@ -160,10 +160,11 @@ class MyProvider:
     def get_realtime(self) -> list[dict]:
         """全市场实时快照 → list[dict]。失败软返回 [], 不抛异常(不阻断轮询线程)。"""
 
-    def get_realtime_indices(self, symbols: list[str]) -> list[dict]:
+    def get_realtime_indices(self, symbols: list[str]) -> list[dict] | None:
         """(可选)指数实时快照 → list[dict], 行字段与 get_realtime 一致。
         A 股快照普遍不含指数(fuyao 的指数在独立端点); 声明 realtime 的源
-        强烈建议实现本方法, 否则指数行情冻结在本地日K兜底。失败软返回 []。"""
+        强烈建议实现本方法, 否则指数行情冻结在本地日K兜底。失败返回 None,
+        成功但无数据返回 []。"""
 
     def get_financials(self, table, symbols, latest_only=False) -> pl.DataFrame:
         """财务数据(声明 financial 数据集时实现, table 见 financial_sync 调用)。"""
@@ -217,7 +218,7 @@ class MyProvider:
 | 方法 | 失败行为 |
 | --- | --- |
 | `get_realtime` | **软失败**: 返回 `[]` + warning 日志, 保证轮询线程不中断 |
-| `get_realtime_indices` | **软失败**: 返回 `[]` + warning 日志; 指数缓存为空走日K兜底 |
+| `get_realtime_indices` | **软失败**: 返回 `None` + warning 日志, 保留上轮有效缓存; 成功无数据返回 `[]` |
 | `get_minute` | 抛异常时调用方自动回退 TickFlow 重试 |
 | `get_daily` / `get_adj_factors` / `get_financials` | 异常由上层同步流程捕获记录; 无数据返回空 DataFrame |