feat(index): 指数收敛为固定核心四只 + fuyao 指数快照接入

- 新建 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 过滤/软失败 + 自定义源指数补充链路
  (含监控规则并入/无协议源静默/指数失败软降级)
This commit is contained in:
shy3130
2026-08-31 16:50:21 +08:00
parent e287a55e82
commit 3c6ed99922
14 changed files with 299 additions and 179 deletions
+1 -42
View File
@@ -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(
+1 -8
View File
@@ -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]+")
-19
View File
@@ -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
+22
View File
@@ -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"
+27 -2
View File
@@ -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,
+18
View File
@@ -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())
@@ -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]+")
+2 -44
View File
@@ -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(),
+36 -24
View File
@@ -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:
+3 -15
View File
@@ -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", "<na>", "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:
@@ -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"
+60
View File
@@ -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="")
+1 -1
View File
@@ -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
+10 -15
View File
@@ -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):