From 2e8be527b61e058687787696a8168631e7073c0f Mon Sep 17 00:00:00 2001 From: 0112020179 Date: Thu, 3 Sep 2026 14:42:32 +0800 Subject: [PATCH] fix(index): preserve realtime cache on provider failure --- CONTRIBUTING.md | 2 +- backend/app/plugins/fuyao/provider.py | 7 +- backend/app/services/quote_service.py | 31 ++++++- backend/tests/test_custom_provider_indices.py | 92 +++++++++++++++++-- backend/tests/test_fuyao_provider.py | 4 +- docs/plugin-development.md | 7 +- 6 files changed, 122 insertions(+), 21 deletions(-) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 724f6ab..1ccc4d3 100644 --- a/CONTRIBUTING.md +++ b/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. 领域专项要求 diff --git a/backend/app/plugins/fuyao/provider.py b/backend/app/plugins/fuyao/provider.py index 3e8d9a4..a457904 100644 --- a/backend/app/plugins/fuyao/provider.py +++ b/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 = [] diff --git a/backend/app/services/quote_service.py b/backend/app/services/quote_service.py index da21014..16a05db 100644 --- a/backend/app/services/quote_service.py +++ b/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) diff --git a/backend/tests/test_custom_provider_indices.py b/backend/tests/test_custom_provider_indices.py index 1805539..de4603f 100644 --- a/backend/tests/test_custom_provider_indices.py +++ b/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() # ---- 监控分时注入: 全量分钟健康时股票读本地分区 ---- diff --git a/backend/tests/test_fuyao_provider.py b/backend/tests/test_fuyao_provider.py index 4c32f7f..aaddf77 100644 --- a/backend/tests/test_fuyao_provider.py +++ b/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(): diff --git a/docs/plugin-development.md b/docs/plugin-development.md index 1d7d422..6178fa7 100644 --- a/docs/plugin-development.md +++ b/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 |