From 3c6ed99922e4e8f26e9bbc9832f8323a38f21c12 Mon Sep 17 00:00:00 2001 From: shy3130 <415333856@qq.com> Date: Mon, 31 Aug 2026 16:50:21 +0800 Subject: [PATCH] =?UTF-8?q?feat(index):=20=E6=8C=87=E6=95=B0=E6=94=B6?= =?UTF-8?q?=E6=95=9B=E4=B8=BA=E5=9B=BA=E5=AE=9A=E6=A0=B8=E5=BF=83=E5=9B=9B?= =?UTF-8?q?=E5=8F=AA=20+=20fuyao=20=E6=8C=87=E6=95=B0=E5=BF=AB=E7=85=A7?= =?UTF-8?q?=E6=8E=A5=E5=85=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新建 app/services/index_const.py 单一权威: 展示层固定四只核心指数 (上证/深证成指/创业板/科创综指), quote_service/overview/ market_overview_builder/sector_monitor 四处重复定义全部收敛 - fuyao 插件接入指数快照端点 (/api/a-share-index/prices/snapshot): client 增加 index_snapshot (批量上限 627 码), provider 实现可选协议 get_realtime_indices — .BJ 后缀先行过滤 (未知代码整批 1002 连坐), volume 不做股转手; 修复 fuyao 路由下指数冻结在日K兜底的 bug - quote_service 自定义源分支鸭子类型调用 get_realtime_indices 补拉指数, 请求清单 = 核心四只 + 启用指数监控规则标的; 未实现的源走日K兜底 - TickFlow 分支指数固定按码显式拉取, 移除 CN_Index 全量 universe 与 mode core/all 分支; 指数落盘固定 merge 不截断 - 指数偏好全套下线: realtime_index_symbols / sidebar_index_symbols / indices_nav_pinned / realtime_pull_index / realtime_index_mode - /api/index/list 与 /search (全指数浏览搜索) 删除; daily/minute 保留 供指数详情页, sync_instruments/sync_daily 保留供数据页 - sector_monitor 指数标的恒可监控, catalog 签名不再依赖指数偏好 - 新增 7 个用例: 指数快照映射/.BJ 过滤/软失败 + 自定义源指数补充链路 (含监控规则并入/无协议源静默/指数失败软降级) --- backend/app/api/indices.py | 43 +------ backend/app/api/overview.py | 9 +- backend/app/api/settings.py | 19 --- backend/app/plugins/fuyao/client.py | 22 ++++ backend/app/plugins/fuyao/provider.py | 29 ++++- backend/app/services/index_const.py | 18 +++ .../app/services/market_overview_builder.py | 11 +- backend/app/services/preferences.py | 46 +------ backend/app/services/quote_service.py | 60 +++++---- backend/app/services/sector_monitor.py | 18 +-- backend/tests/test_custom_provider_indices.py | 116 ++++++++++++++++++ backend/tests/test_fuyao_provider.py | 60 +++++++++ backend/tests/test_minute_routing.py | 2 +- backend/tests/test_sector_monitor.py | 25 ++-- 14 files changed, 299 insertions(+), 179 deletions(-) create mode 100644 backend/app/services/index_const.py create mode 100644 backend/tests/test_custom_provider_indices.py diff --git a/backend/app/api/indices.py b/backend/app/api/indices.py index febd6bb..6d76b8a 100644 --- a/backend/app/api/indices.py +++ b/backend/app/api/indices.py @@ -1,4 +1,4 @@ -"""指数 API。""" +"""指数 API (核心四只固定清单, 浏览/搜索全量指数已下线; 仅保留详情读数与同步)。""" from __future__ import annotations import logging @@ -27,47 +27,6 @@ def _index_info(repo, symbol: str) -> dict: return hit.to_dicts()[0] -@router.get("/list") -def list_indices(request: Request): - """返回已缓存的 CN_Index 指数列表。""" - repo = request.app.state.repo - df = repo.get_index_instruments() - if df.is_empty(): - return {"results": [], "count": 0} - cols = [c for c in ["symbol", "name", "code", "asset_type"] if c in df.columns] - rows = df.select(cols).sort("symbol").to_dicts() - return {"results": rows, "count": len(rows)} - - -@router.get("/search") -def search_indices( - request: Request, - q: str = Query("", min_length=0, max_length=50, description="搜索关键词"), - limit: int = Query(20, ge=1, le=100), -): - """模糊搜索指数。""" - repo = request.app.state.repo - df = repo.get_index_instruments() - if df.is_empty(): - return {"results": []} - if not q.strip(): - rows = df.head(limit).to_dicts() - return {"results": rows} - - keyword = q.strip().upper() - masks = [] - if "code" in df.columns: - masks.append(pl.col("code").cast(pl.Utf8).str.contains(keyword, literal=True)) - masks.append(pl.col("symbol").cast(pl.Utf8).str.to_uppercase().str.contains(keyword, literal=True)) - if "name" in df.columns: - masks.append(pl.col("name").cast(pl.Utf8).str.contains(q.strip(), literal=True)) - - mask = masks[0] - for m in masks[1:]: - mask = mask | m - rows = df.filter(mask).head(limit).to_dicts() - return {"results": rows} - @router.get("/daily") def get_index_daily( diff --git a/backend/app/api/overview.py b/backend/app/api/overview.py index 4cce44d..f58fd99 100644 --- a/backend/app/api/overview.py +++ b/backend/app/api/overview.py @@ -12,6 +12,7 @@ import polars as pl from fastapi import APIRouter, Request from app.services.ext_data import ExtConfig, ExtConfigStore +from app.services.index_const import CORE_INDEX_NAMES, CORE_INDEX_SYMBOLS from app.services.screener import ScreenerService router = APIRouter(prefix="/api/overview", tags=["overview"]) @@ -37,14 +38,6 @@ def invalidate_overview_cache() -> None: _cache_ts = 0.0 -CORE_INDEX_NAMES = { - "000001.SH": "上证指数", - "399001.SZ": "深证成指", - "399006.SZ": "创业板指", - "000680.SH": "科创综指", -} -CORE_INDEX_SYMBOLS = tuple(CORE_INDEX_NAMES.keys()) - _DIMENSION_SEP = re.compile(r"[、,，;；|/\s]+") diff --git a/backend/app/api/settings.py b/backend/app/api/settings.py index 8b305bb..9be844c 100644 --- a/backend/app/api/settings.py +++ b/backend/app/api/settings.py @@ -489,7 +489,6 @@ def get_preferences() -> dict: return { "realtime_quotes_enabled": preferences.get_realtime_quotes_enabled(), "realtime_allowed": _realtime_allowed(), - "indices_nav_pinned": preferences.get_indices_nav_pinned(), "watchlist_groups_in_nav": preferences.get_watchlist_groups_in_nav(), "minute_sync_enabled": preferences.get_minute_sync_enabled(), "minute_sync_days": preferences.get_minute_sync_days(), @@ -531,7 +530,6 @@ def get_preferences() -> dict: "wecom_bot_enabled": preferences.get_wecom_bot_enabled(), "webhook_enabled_default": preferences.get_webhook_enabled_default(), "webhook_default_channels": preferences.get_webhook_default_channels(), - "sidebar_index_symbols": preferences.get_sidebar_index_symbols(), "minute_intraday_refresh": preferences.get_minute_intraday_refresh(), "minute_intraday_refresh_interval": preferences.get_minute_intraday_refresh_interval(), "monitor_ext_fields": preferences.get_monitor_ext_fields(), @@ -901,9 +899,6 @@ class RealtimeQuotesPrefs(BaseModel): class RealtimeQuoteScopePrefs(BaseModel): realtime_pull_stock: bool | None = None realtime_pull_etf: bool | None = None - realtime_pull_index: bool | None = None - realtime_index_mode: str | None = None - realtime_index_symbols: list[str] | None = None @router.put("/preferences/realtime-quotes") @@ -985,19 +980,6 @@ def update_realtime_quote_scope(req: RealtimeQuoteScopePrefs) -> dict: return preferences.set_realtime_quote_scope(cfg) -class IndicesNavPinnedPrefs(BaseModel): - indices_nav_pinned: bool - - -@router.put("/preferences/indices-nav-pinned") -def update_indices_nav_pinned(req: IndicesNavPinnedPrefs) -> dict: - """保存侧栏指数报价卡片固定显示开关。 - ON=常驻显示；OFF=跟随实时行情开关（仅实时开时显示）。""" - from app.services import preferences - preferences.save({"indices_nav_pinned": req.indices_nav_pinned}) - return {"indices_nav_pinned": req.indices_nav_pinned} - - class WatchlistGroupsInNavPrefs(BaseModel): watchlist_groups_in_nav: bool @@ -1014,7 +996,6 @@ class RealtimeMonitorConfigIn(BaseModel): sse_refresh_pages: dict[str, bool] | None = None strategy_monitor_enabled: bool | None = None strategy_monitor_ids: list[str] | None = None - sidebar_index_symbols: list[str] | None = None screener_auto_run: bool | None = None minute_intraday_refresh: bool | None = None minute_intraday_refresh_interval: int | None = None diff --git a/backend/app/plugins/fuyao/client.py b/backend/app/plugins/fuyao/client.py index b090871..4a93396 100644 --- a/backend/app/plugins/fuyao/client.py +++ b/backend/app/plugins/fuyao/client.py @@ -115,6 +115,28 @@ class FuyaoClient: raise FuyaoError("全市场快照为空") return out, server_ts + # ---- 指数快照 ---- + def index_snapshot(self, thscodes: list[str]) -> tuple[list[dict], int]: + """拉取指数行情快照 (/api/a-share-index/prices/snapshot)。返回 (rows, 服务端时间戳ms)。 + + 与 A 股快照不同: 必须显式传 thscodes (逗号分隔), 无全量枚举; + 单次批量上限实测 627 个代码 (~6.3KB 参数, 超出 HTTP 400); + 混入未知代码整批失败 (code=1002 连坐), 调用方需自行过滤。 + 覆盖范围: 沪深交易所指数 + 同花顺板块指数, 无北交所 (官方文档明确)。 + """ + if not thscodes: + return [], 0 + joined = ",".join(thscodes[:627]) + data = self._get("/api/a-share-index/prices/snapshot", {"thscodes": joined}) + try: + server_ts = int(data.get("timestamp") or 0) + except (TypeError, ValueError): + server_ts = 0 + rows = data.get("item") + if not isinstance(rows, list): + rows = [] + return rows, server_ts + # ---- 历史日K ---- def historical_kline( self, thscode: str, start_ms: int, end_ms: int, adjust: str = "none" diff --git a/backend/app/plugins/fuyao/provider.py b/backend/app/plugins/fuyao/provider.py index 76ce1d9..3e8d9a4 100644 --- a/backend/app/plugins/fuyao/provider.py +++ b/backend/app/plugins/fuyao/provider.py @@ -250,11 +250,12 @@ def _dump_date_range(path: Path) -> tuple[date | None, date | None]: return row["dmin"][0], row["dmax"][0] -def _map_snapshot_row(row: dict, fetched_ms: int) -> dict | None: +def _map_snapshot_row(row: dict, fetched_ms: int, *, volume_to_hand: bool = True) -> dict | None: """扶摇快照行 → 内部 realtime record。字段缺失时按依赖推导, 不伪造数据。 实测字段(2026-08): high_price / low_price / prev_price; 官方文档示例: highest_price / lowest_price / prev_close_price。两者都取。 + volume_to_hand: A 股快照 volume 为股 → 手; 指数快照无此口径, 直接透传。 """ symbol = row.get("thscode") if not symbol: @@ -283,7 +284,7 @@ def _map_snapshot_row(row: dict, fetched_ms: int) -> dict | None: "open": _to_float(row.get("open_price")), "high": _to_float(_first(row, "high_price", "highest_price")), "low": _to_float(_first(row, "low_price", "lowest_price")), - "volume": math.floor(volume / 100.0) if volume is not None else None, # 股 → 手 + "volume": math.floor(volume / 100.0) if (volume is not None and volume_to_hand) else volume, "amount": _to_float(row.get("turnover")), "change_pct": change_pct, "change_amount": change_amount, @@ -400,6 +401,30 @@ class FuyaoProvider: logger.info("扶摇实时行情拉取完成: %d 条(丢弃 %d 行)", len(records), dropped) return records + def get_realtime_indices(self, symbols: list[str]) -> list[dict]: + """指数实时快照 → 内部 realtime record (可选插件协议, quote_service 鸭子类型调用)。 + + A 股快照不含指数, 指数在扶摇是独立端点; 覆盖沪深交易所指数 + 同花顺板块, + 无北交所 (未知代码会整批 1002 连坐, .BJ 直接跳过)。失败软返回空列表。 + """ + wanted = [s for s in symbols if s and not s.upper().endswith(".BJ")] + if not wanted: + return [] + try: + rows, server_ts = self._get_client().index_snapshot(wanted) + except FuyaoError as e: + logger.warning("扶摇指数行情拉取失败: %s", e) + return [] + + fetched_ms = server_ts or int(time.time() * 1000) + records = [] + for row in rows: + rec = _map_snapshot_row(row, fetched_ms, volume_to_hand=False) + if rec is not None: + records.append(rec) + logger.info("扶摇指数行情拉取完成: %d 条(请求 %d 只)", len(records), len(wanted)) + return records + # ---- daily ---- def get_daily( self, diff --git a/backend/app/services/index_const.py b/backend/app/services/index_const.py new file mode 100644 index 0000000..d19f9ee --- /dev/null +++ b/backend/app/services/index_const.py @@ -0,0 +1,18 @@ +"""实时指数核心清单 — 产品级固定契约 (单一权威)。 + +指数展示层 (侧栏指数条 / 市场总览) 固定四只核心指数, 不开放配置: +- 数据源边界: TickFlow 与 fuyao 指数快照双源均完整覆盖, 无降级分歧; +- 后端消费方 (quote_service / overview / sector_monitor) 与前端 Layout + 统一引用此处, 不得各自维护副本。 + +监控规则的指数标的不受此限 — quote_service 会把启用规则的指数并入显式拉取。 +""" + +CORE_INDEX_NAMES: dict[str, str] = { + "000001.SH": "上证指数", + "399001.SZ": "深证成指", + "399006.SZ": "创业板指", + "000680.SH": "科创综指", +} + +CORE_INDEX_SYMBOLS: tuple[str, ...] = tuple(CORE_INDEX_NAMES.keys()) diff --git a/backend/app/services/market_overview_builder.py b/backend/app/services/market_overview_builder.py index a6c549b..f879fb4 100644 --- a/backend/app/services/market_overview_builder.py +++ b/backend/app/services/market_overview_builder.py @@ -19,20 +19,13 @@ from typing import Any import polars as pl from app.services.ext_data import ExtConfig, ExtConfigStore +from app.services.index_const import CORE_INDEX_NAMES, CORE_INDEX_SYMBOLS from app.services.screener import ScreenerService # ================================================================ -# 常量(与 overview.py 保持同步;复盘复盘仅 A 股核心指数) +# 常量(核心指数清单单一权威: app.services.index_const) # ================================================================ -CORE_INDEX_NAMES = { - "000001.SH": "上证指数", - "399001.SZ": "深证成指", - "399006.SZ": "创业板指", - "000680.SH": "科创综指", -} -CORE_INDEX_SYMBOLS = tuple(CORE_INDEX_NAMES.keys()) - _DIMENSION_SEP = re.compile(r"[、,，;；|/\s]+") diff --git a/backend/app/services/preferences.py b/backend/app/services/preferences.py index be062e5..f5daaa6 100644 --- a/backend/app/services/preferences.py +++ b/backend/app/services/preferences.py @@ -69,12 +69,6 @@ def get_realtime_quotes_enabled() -> bool: return load().get("realtime_quotes_enabled", False) -def get_indices_nav_pinned() -> bool: - """侧栏指数报价卡片是否固定显示。默认 True（常驻）。 - 关闭后，卡片跟随实时行情开关（仅实时开时显示）。""" - return load().get("indices_nav_pinned", True) - - def get_watchlist_groups_in_nav() -> bool: """自选分组是否显示在侧边栏（可展开二级子菜单）。默认 False。""" return load().get("watchlist_groups_in_nav", False) @@ -672,10 +666,9 @@ SSE_REFRESH_PAGES_DEFAULT = { "limit-ladder": False, } -SIDEBAR_INDEX_SYMBOLS_DEFAULT = ["000001.SH", "399001.SZ", "399006.SZ", "000680.SH"] - # ===== 盘中实时行情范围 (独立于盘后管道范围) ===== +# 指数不在其中: 展示层固定核心四只 (app.services.index_const), 不开放配置。 def get_realtime_pull_stock() -> bool: @@ -687,32 +680,11 @@ def get_realtime_pull_etf() -> bool: return load().get("realtime_pull_etf", False) -def get_realtime_pull_index() -> bool: - return load().get("realtime_pull_index", True) - - -def get_realtime_index_mode() -> str: - mode = str(load().get("realtime_index_mode", "core") or "core").lower() - return mode if mode in {"core", "all"} else "core" - - -def get_realtime_index_symbols() -> list[str]: - stored = load().get("realtime_index_symbols", SIDEBAR_INDEX_SYMBOLS_DEFAULT) - if isinstance(stored, str): - import re - stored = [s.strip() for s in re.split(r"[,\s]+", stored) if s.strip()] - return [str(s) for s in stored if str(s).strip()] - - def set_realtime_quote_scope(cfg: dict) -> dict: updates = {} - for key in ("realtime_pull_stock", "realtime_pull_etf", "realtime_pull_index"): + for key in ("realtime_pull_stock", "realtime_pull_etf"): if key in cfg and cfg[key] is not None: updates[key] = bool(cfg[key]) - if "realtime_index_mode" in cfg and cfg["realtime_index_mode"] in {"core", "all"}: - updates["realtime_index_mode"] = cfg["realtime_index_mode"] - if "realtime_index_symbols" in cfg and cfg["realtime_index_symbols"] is not None: - updates["realtime_index_symbols"] = cfg["realtime_index_symbols"] if updates: save(updates) return get_realtime_quote_scope() @@ -722,9 +694,6 @@ def get_realtime_quote_scope() -> dict: return { "realtime_pull_stock": get_realtime_pull_stock(), "realtime_pull_etf": get_realtime_pull_etf(), - "realtime_pull_index": get_realtime_pull_index(), - "realtime_index_mode": get_realtime_index_mode(), - "realtime_index_symbols": get_realtime_index_symbols(), } @@ -743,13 +712,6 @@ def set_sse_refresh_pages(pages: dict[str, bool]) -> dict[str, bool]: return get_sse_refresh_pages() -def get_sidebar_index_symbols() -> list[str]: - """返回左侧菜单显示的指数代码。""" - stored = load().get("sidebar_index_symbols", SIDEBAR_INDEX_SYMBOLS_DEFAULT) - allowed = set(SIDEBAR_INDEX_SYMBOLS_DEFAULT) - return [s for s in stored if s in allowed] - - def get_strategy_monitor_enabled() -> bool: """策略告警评估总开关。""" return load().get("strategy_monitor_enabled", False) @@ -911,9 +873,6 @@ def set_realtime_monitor_config(cfg: dict) -> dict: updates["strategy_monitor_enabled"] = cfg["strategy_monitor_enabled"] if "strategy_monitor_ids" in cfg: updates["strategy_monitor_ids"] = cfg["strategy_monitor_ids"] - if "sidebar_index_symbols" in cfg: - allowed = set(SIDEBAR_INDEX_SYMBOLS_DEFAULT) - updates["sidebar_index_symbols"] = [s for s in cfg["sidebar_index_symbols"] if s in allowed] if "screener_auto_run" in cfg: updates["screener_auto_run"] = bool(cfg["screener_auto_run"]) if "minute_intraday_refresh" in cfg: @@ -947,7 +906,6 @@ def get_realtime_monitor_config() -> dict: "sse_refresh_pages": get_sse_refresh_pages(), "strategy_monitor_enabled": get_strategy_monitor_enabled(), "strategy_monitor_ids": get_strategy_monitor_ids(), - "sidebar_index_symbols": get_sidebar_index_symbols(), "screener_auto_run": get_screener_auto_run(), "minute_intraday_refresh": get_minute_intraday_refresh(), "minute_intraday_refresh_interval": get_minute_intraday_refresh_interval(), diff --git a/backend/app/services/quote_service.py b/backend/app/services/quote_service.py index fade3c4..3a43d13 100644 --- a/backend/app/services/quote_service.py +++ b/backend/app/services/quote_service.py @@ -3,7 +3,8 @@ 集中管理全市场行情拉取 + enriched 缓存，供盘中选股、自选股等所有模块复用。 架构: - - 后台线程轮询 TickFlow get_by_universes(["CN_Equity_A", "CN_Index"]) + - 后台线程轮询 TickFlow get_by_universes(["CN_Equity_A", "CN_ETF"]) + 核心指数按码拉取 + (自定义源走 provider.get_realtime() + 可选 get_realtime_indices() 指数补充) - 拉取行情 → 写 kline_daily (不复权) + 增量计算 enriched → 写盘 + 更新缓存 - _enriched_cache 是唯一的盘中数据源 (OHLCV + 全套技术指标) - _live_agg_cache 是递推状态 (只加载一次, 盘中不变) @@ -34,6 +35,7 @@ import polars as pl from app.market_time import cn_now, cn_today from app.parquet import scan_daily_parquet +from app.services.index_const import CORE_INDEX_SYMBOLS from app.strategy.intraday_signals import IntradaySignalEvaluator logger = logging.getLogger(__name__) @@ -165,8 +167,6 @@ def _monitor_name_map(repo) -> dict[str, str]: class QuoteService: """全局实时行情服务 — 单例。""" - CORE_INDEX_SYMBOLS = ("000001.SH", "399001.SZ", "399006.SZ", "000680.SH") - # 档位 → 最小轮询间隔 (秒) — TickFlow 档位限速保护, 仅实时源为 tickflow 时适用 TIER_MIN_INTERVAL = { "expert": 1.0, @@ -599,7 +599,18 @@ class QuoteService: try: t0 = time.perf_counter() now_ts = time.perf_counter() - records = custom_sources.get_provider(provider_name).get_realtime() + provider = custom_sources.get_provider(provider_name) + records = provider.get_realtime() + # 指数补充: A 股快照通常不含指数。插件可选实现 + # get_realtime_indices(symbols) 用独立端点补拉 (如 fuyao 指数快照); + # 未实现的源指数缓存为空, 由日K兜底接管。 + 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 []) + except Exception as e: # noqa: BLE001 + logger.warning("自定义源指数行情拉取失败: %s", e) except Exception as e: # noqa: BLE001 logger.warning("自定义实时行情拉取失败: %s", e) return @@ -619,15 +630,10 @@ class QuoteService: try: from app.services import preferences all_index_symbols = set(self._repo.get_index_symbol_set()) if self._repo else set() - core_index_symbols = set(preferences.get_realtime_index_symbols() or self.CORE_INDEX_SYMBOLS) + core_index_symbols = set(CORE_INDEX_SYMBOLS) all_index_symbols.update(core_index_symbols) - # 指数监控规则标的并入轮询 (mode=core 时 quotes.get 显式拉取覆盖; mode=all 被 CN_Index 全覆盖) - monitor_index_symbols: set[str] = set() - engine = getattr(self._app_state, "monitor_engine", None) if self._app_state else None - if engine: - for _r in list(engine.rules.values()): - if _r.get("enabled", True) and _r.get("asset_type") == "index" and _r.get("scope") == "symbols": - monitor_index_symbols.update(s for s in _r.get("symbols", []) if s) + # 指数监控规则标的并入显式拉取 (quotes.get 按码覆盖) + monitor_index_symbols = self._collect_monitor_index_symbols() all_index_symbols.update(monitor_index_symbols) all_etf_symbols = set() if self._repo: @@ -640,8 +646,6 @@ class QuoteService: universes.append("CN_Equity_A") if preferences.get_realtime_pull_etf() and all_etf_symbols: universes.append("CN_ETF") - if preferences.get_realtime_pull_index() and preferences.get_realtime_index_mode() == "all": - universes.append("CN_Index") resp = [] if universes: @@ -649,9 +653,10 @@ class QuoteService: logger.info("拉取全市场行情 (universes=%s, SDK超时=30s×重试3)", universes) resp.extend(tf.quotes.get_by_universes(universes=universes) or []) logger.info("全市场行情拉取完成: %d 条 (%.2fs)", len(resp), time.perf_counter() - _u0) - if preferences.get_realtime_pull_index() and preferences.get_realtime_index_mode() == "core": + # 指数: 固定核心四只 + 监控规则标的, 按码显式拉取 + _core_syms = sorted(core_index_symbols | monitor_index_symbols) + if _core_syms: _i0 = time.perf_counter() - _core_syms = sorted(core_index_symbols | monitor_index_symbols) resp.extend(tf.quotes.get(symbols=_core_syms) or []) logger.info("核心指数行情拉取完成: %d 只 (%.2fs)", len(_core_syms), time.perf_counter() - _i0) except Exception as e: # noqa: BLE001 @@ -700,7 +705,7 @@ class QuoteService: """把全市场 records 写盘并增量计算 enriched。""" from app.services import preferences all_index_symbols = set(self._repo.get_index_symbol_set()) if self._repo else set() - core_index_symbols = set(preferences.get_realtime_index_symbols() or self.CORE_INDEX_SYMBOLS) + core_index_symbols = set(CORE_INDEX_SYMBOLS) all_index_symbols.update(core_index_symbols) all_etf_symbols = set() if self._repo: @@ -763,20 +768,16 @@ class QuoteService: if not etf_daily_df.is_empty() and self._repo: self._flush_live_enriched(etf_daily_df, etf_quote_extra, asset_type="etf") # ---- 指数: 仅有指数监控规则时才写盘 (无规则零成本) ---- - # mode=all (完整 CN_Index universe) → flush 覆盖; mode=core (部分标的) → merge 不截断分区 + # 指数为按码显式拉取 (部分标的) → 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: - if use_flush: - self._repo.flush_live_daily_asset("index", index_daily_df) - else: - self._repo.merge_live_daily_asset("index", index_daily_df) + 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", merge=not use_flush) + self._flush_live_enriched(index_daily_df, self._build_quote_extra(index_records), asset_type="index", merge=True) # ---- 通知 SSE ---- self._broadcast_quote_updated() @@ -788,6 +789,17 @@ class QuoteService: # 工具 # ================================================================ + def _collect_monitor_index_symbols(self) -> set[str]: + """启用中的指数监控规则标的 (asset_type=index & scope=symbols)。""" + engine = getattr(self._app_state, "monitor_engine", None) if self._app_state else None + if not engine: + return set() + out: set[str] = set() + for _r in list(engine.rules.values()): + if _r.get("enabled", True) and _r.get("asset_type") == "index" and _r.get("scope") == "symbols": + out.update(s for s in _r.get("symbols", []) if s) + return out + @staticmethod @staticmethod def _build_daily(records: list[dict]) -> pl.DataFrame: diff --git a/backend/app/services/sector_monitor.py b/backend/app/services/sector_monitor.py index 31e44e9..a5180e8 100644 --- a/backend/app/services/sector_monitor.py +++ b/backend/app/services/sector_monitor.py @@ -13,13 +13,7 @@ import polars as pl from app.services import preferences from app.services.ext_data import ExtConfig, ExtConfigStore - -CORE_INDICES = { - "000001.SH": "上证指数", - "399001.SZ": "深证成指", - "399006.SZ": "创业板指", - "000680.SH": "科创综指", -} +from app.services.index_const import CORE_INDEX_NAMES as CORE_INDICES SECTOR_KINDS = {"index", "concept", "industry"} _VALUE_SEP = re.compile(r"[\u3001,\uff0c;\uff1b|]+") _NULL_VALUES = {"nan", "none", "null", "", "n/a", "-"} @@ -147,16 +141,14 @@ class SectorMonitorService: except Exception: pass - realtime_index_enabled = preferences.get_realtime_pull_index() - realtime_indices = set(preferences.get_realtime_index_symbols() or CORE_INDICES) - all_indices_enabled = preferences.get_realtime_index_mode() == "all" + # 指数标的恒可实时监控: quote_service 对核心四只 + 启用规则标的显式拉取 for symbol, name in sorted(index_names.items()): target = { "key": f"index:{symbol}", "kind": "index", "name": name, "symbol": symbol, - "available": realtime_index_enabled and (all_indices_enabled or symbol in realtime_indices), + "available": True, "member_count": 1, } catalog["index"].append(target) @@ -223,10 +215,6 @@ class SectorMonitorService: for path in sorted(paths) if path.is_file() ] - index_mode = preferences.get_realtime_index_mode() - index_enabled = preferences.get_realtime_pull_index() - index_symbols = sorted(preferences.get_realtime_index_symbols() or CORE_INDICES) - signature.append((f"realtime_indices:{index_enabled}:{index_mode}:{','.join(index_symbols)}", 0, 0)) return tuple(signature) def _read_ext_dataframe(self, config: ExtConfig) -> pl.DataFrame: diff --git a/backend/tests/test_custom_provider_indices.py b/backend/tests/test_custom_provider_indices.py new file mode 100644 index 0000000..99fec24 --- /dev/null +++ b/backend/tests/test_custom_provider_indices.py @@ -0,0 +1,116 @@ +"""自定义源实时行情的指数补充链路测试。 + +契约 (CONTRIBUTING §4 能力路由矩阵): 实时源路由到自定义 provider 时, +quote_service 在 get_realtime() 之外鸭子类型调用可选方法 +get_realtime_indices(symbols) 补拉指数 — A 股快照普遍不含指数 +(fuyao 实测无指数, 指数在其独立端点)。 +""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import ClassVar + +from app.services import quote_service as qs +from app.services.index_const import CORE_INDEX_SYMBOLS + + +class _FakeProvider: + """带指数能力的假自定义源: 记录请求的 symbols, 返回预置 records。""" + + def __init__(self, stocks: list[dict], indices: list[dict]): + self._stocks = stocks + self._indices = indices + self.index_calls: list[list[str]] = [] + + def get_realtime(self) -> list[dict]: + return list(self._stocks) + + def get_realtime_indices(self, symbols: list[str]) -> list[dict]: + self.index_calls.append(list(symbols)) + wanted = set(symbols) + return [r for r in self._indices if r["symbol"] in wanted] + + +class _ProviderNoIndices: + """未实现可选协议的源: 指数不得报错, 只是不补充。""" + + def get_realtime(self) -> list[dict]: + return [{"symbol": "600519.SH", "last_price": 1480.0}] + + +def _stock_rec(symbol: str = "600519.SH") -> dict: + return {"symbol": symbol, "last_price": 1480.0, "prev_close": 1455.0, "volume": 12345} + + +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]]]: + """构造最小 QuoteService: 自定义源路由 + 捕获 _process_full_market_records 入参。""" + from app.services import preferences as prefs_mod + + service = qs.QuoteService() + captured: list[list[dict]] = [] + monkeypatch.setattr(prefs_mod, "get_realtime_data_provider", lambda: "fuyao") + import app.data_providers.custom as custom_mod + + monkeypatch.setattr(custom_mod, "provider_has_dataset", lambda name, dataset: dataset == "realtime") + monkeypatch.setattr(custom_mod, "get_provider", lambda name: provider) + monkeypatch.setattr( + service, "_process_full_market_records", + lambda records, *, t0, now_ts: captured.append(records), + ) + return service, captured + + +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._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 provider.index_calls == [sorted(CORE_INDEX_SYMBOLS)] + + +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) + + class _Engine: + rules: ClassVar[dict] = { + "r1": {"enabled": True, "asset_type": "index", "scope": "symbols", "symbols": ["000300.SH"]}, + "r2": {"enabled": False, "asset_type": "index", "scope": "symbols", "symbols": ["000016.SH"]}, + "r3": {"enabled": True, "asset_type": "stock", "scope": "symbols", "symbols": ["600519.SH"]}, + } + + service._app_state = SimpleNamespace(monitor_engine=_Engine()) + service._fetch_full_market_quotes() + + assert provider.index_calls == [sorted(set(CORE_INDEX_SYMBOLS) | {"000300.SH"})] + + +def test_custom_provider_without_indices_protocol_is_silent(monkeypatch): + """未实现 get_realtime_indices 的源: 个股 records 照常, 指数不补充不报错。""" + service, captured = _service_with_provider(monkeypatch, _ProviderNoIndices()) + service._fetch_full_market_quotes() + assert captured == [[{"symbol": "600519.SH", "last_price": 1480.0}]] + + +def test_custom_provider_index_fetch_error_is_soft(monkeypatch): + """指数补充失败软降级: 警告不抛出, 个股 records 仍然进入处理链。""" + class _Boom: + def get_realtime(self) -> list[dict]: + return [_stock_rec()] + + def get_realtime_indices(self, symbols: list[str]) -> list[dict]: + raise RuntimeError("index endpoint down") + + service, captured = _service_with_provider(monkeypatch, _Boom()) + service._fetch_full_market_quotes() + assert len(captured) == 1 and captured[0][0]["symbol"] == "600519.SH" diff --git a/backend/tests/test_fuyao_provider.py b/backend/tests/test_fuyao_provider.py index c5d3693..4c32f7f 100644 --- a/backend/tests/test_fuyao_provider.py +++ b/backend/tests/test_fuyao_provider.py @@ -266,6 +266,66 @@ def test_realtime_error_returns_empty_list(monkeypatch): assert provider.get_realtime() == [] +# ---- 指数快照 (可选插件协议 get_realtime_indices) ---- + + +class _FakeIndexClient: + def __init__(self, rows=None, server_ts=0, error=None): + self.rows = rows or [] + self.server_ts = server_ts + self.error = error + self.calls: list[list[str]] = [] + + def index_snapshot(self, thscodes): + self.calls.append(list(thscodes)) + if self.error: + raise self.error + return list(self.rows), self.server_ts + + def close(self): + pass + + +def _index_provider_with(monkeypatch, **kwargs): + fake = _FakeIndexClient(**kwargs) + monkeypatch.setattr(fp, "fuyao_client", type("M", (), {"FuyaoClient": lambda **kw: fake})) + monkeypatch.setattr(fp, "get_api_key", lambda: "test-key") + return FuyaoProvider(), fake + + +def test_realtime_indices_maps_and_keeps_volume_unit(monkeypatch): + """指数快照 → realtime record; volume 无股→手口径, 直接透传。""" + provider, fake = _index_provider_with( + monkeypatch, + rows=[_row("000001.SH", volume=576656606, price_change_ratio_pct=0.86)], + server_ts=1787542612000, + ) + records = provider.get_realtime_indices(["000001.SH", "399001.SZ"]) + assert fake.calls == [["000001.SH", "399001.SZ"]] + assert len(records) == 1 + r = records[0] + assert r["symbol"] == "000001.SH" + assert r["change_pct"] == pytest.approx(0.0086) + assert r["timestamp"] == 1787542612000 + assert r["volume"] == 576656606 # 不做 /100 + + +def test_realtime_indices_skips_bj_symbols(monkeypatch): + """北交所指数扶摇不支持, 未知代码会整批 1002 连坐 → .BJ 直接跳过不进请求。""" + provider, fake = _index_provider_with(monkeypatch, rows=[]) + assert provider.get_realtime_indices(["899050.BJ", "000001.SH"]) == [] + assert fake.calls == [["000001.SH"]] + assert provider.get_realtime_indices(["899050.BJ"]) == [] + assert fake.calls == [["000001.SH"]] # 全 .BJ 时根本不发请求 + + +def test_realtime_indices_error_returns_empty(monkeypatch): + provider, _ = _index_provider_with( + monkeypatch, error=fc.FuyaoError("扶摇接口错误 code=1002: Unknown thscode") + ) + assert provider.get_realtime_indices(["000001.SH"]) == [] + + def test_client_requires_api_key(): with pytest.raises(fc.FuyaoError): fc.FuyaoClient(api_key="") diff --git a/backend/tests/test_minute_routing.py b/backend/tests/test_minute_routing.py index 1e30e27..91516b3 100644 --- a/backend/tests/test_minute_routing.py +++ b/backend/tests/test_minute_routing.py @@ -275,7 +275,7 @@ def test_get_minute_batch_splits_stock_and_etf(monkeypatch): asset_type='stock'/'etf' 调用 sync_minute_batch, 结果 concat 返回。 覆盖 kline.py get_minute_batch 的双调用拼接逻辑 (本次提交改动量最大的部分)。 - 契约: 本端点只接受 stock/ETF (指数走 /api/index/minute), 故两分支覆盖全部 incomplete。 + 契约: 本端点只接受 stock/ETF, 故两分支覆盖全部 incomplete。 """ from app.api import kline as kline_api diff --git a/backend/tests/test_sector_monitor.py b/backend/tests/test_sector_monitor.py index b6af59e..8a6d310 100644 --- a/backend/tests/test_sector_monitor.py +++ b/backend/tests/test_sector_monitor.py @@ -110,28 +110,23 @@ def test_index_targets_are_evaluated_independently(tmp_path): assert events[0]["change_pct"] == 0.012 -def test_index_availability_updates_when_realtime_pool_changes(tmp_path, monkeypatch): - selected = ["000001.SH"] - monkeypatch.setattr(sector_monitor.preferences, "get_realtime_pull_index", lambda: True) - monkeypatch.setattr(sector_monitor.preferences, "get_realtime_index_mode", lambda: "core") - monkeypatch.setattr(sector_monitor.preferences, "get_realtime_index_symbols", lambda: selected) +def test_index_targets_always_available(tmp_path): + """指数标的恒可实时监控: quote_service 对核心四只 + 启用规则标的显式拉取, + 不再有按偏好池翻转 available 的行为。""" service = SectorMonitorService(_Repo(tmp_path)) + targets = {target["symbol"]: target for target in service.list_targets()["index"]} + assert targets["000001.SH"]["available"] is True + assert targets["399006.SZ"]["available"] is True - first = {target["symbol"]: target for target in service.list_targets()["index"]} - assert first["000001.SH"]["available"] is True - assert first["399006.SZ"]["available"] is False + first = targets["000001.SH"] initial_quote = pl.DataFrame({"symbol": ["000001.SH"], "change_pct": [0.2]}) - service.build_snapshots(pl.DataFrame(), initial_quote, [first["000001.SH"]], {5}, now=1000.0) + service.build_snapshots(pl.DataFrame(), initial_quote, [first], {5}, now=1000.0) - selected[:] = ["399006.SZ"] - second = {target["symbol"]: target for target in service.list_targets()["index"]} - assert second["000001.SH"]["available"] is False - assert second["399006.SZ"]["available"] is True changed_quote = pl.DataFrame({"symbol": ["000001.SH"], "change_pct": [1.3]}) snapshot = service.build_snapshots( - pl.DataFrame(), changed_quote, [second["000001.SH"]], {5}, now=1300.0, + pl.DataFrame(), changed_quote, [first], {5}, now=1300.0, ) - assert snapshot["index:000001.SH"]["window_changes"][5] is None + assert snapshot["index:000001.SH"]["window_changes"][5] is not None def test_concept_snapshot_uses_member_average_and_full_window(tmp_path):