diff --git a/backend/app/api/kline.py b/backend/app/api/kline.py
index 2c0a545..d1a1627 100644
--- a/backend/app/api/kline.py
+++ b/backend/app/api/kline.py
@@ -95,7 +95,7 @@ def search_instruments(
@router.post("/instruments/names")
def instruments_names(request: Request, symbols: list[str]):
- """批量查标的名称 (股票 + ETF)。传入 symbol 列表, 返回 {symbol: name}。"""
+ """批量查标的名称 (股票 + ETF + 指数)。传入 symbol 列表, 返回 {symbol: name}。"""
if not symbols:
return {"names": {}}
repo = request.app.state.repo
@@ -430,10 +430,37 @@ def get_daily_batch(request: Request, body: dict):
start = end - timedelta(days=days * 2) # 多取一些确保交易日够
cols = ["symbol", "date", "open", "high", "low", "close", "volume"]
- df = repo.get_daily_batch(symbols, start, end, columns=cols)
- if df.is_empty():
+ # 按资产类型分组: stock 走批量缓存; etf/index 逐只查独立存储 (数量少, 成本可忽略)
+ stock_symbols: list[str] = []
+ etf_symbols: list[str] = []
+ index_symbols: list[str] = []
+ for s in symbols:
+ t = repo.resolve_asset_type(s)
+ if t == "etf":
+ etf_symbols.append(s)
+ elif t == "index":
+ index_symbols.append(s)
+ else:
+ stock_symbols.append(s)
+
+ frames: list[pl.DataFrame] = []
+ if stock_symbols:
+ df_stock = repo.get_daily_batch(stock_symbols, start, end, columns=cols)
+ if not df_stock.is_empty():
+ frames.append(df_stock)
+ for sym in etf_symbols:
+ sub = repo.get_etf_daily(sym, start, end, columns=cols)
+ if not sub.is_empty():
+ frames.append(sub)
+ for sym in index_symbols:
+ sub = repo.get_index_daily(sym, start, end, columns=cols)
+ if not sub.is_empty():
+ frames.append(sub)
+
+ if not frames:
return {"data": {}}
+ df = pl.concat(frames, how="diagonal_relaxed")
# 按 symbol 分组, 每只取最近 N 条
result: dict[str, list[dict]] = {}
@@ -625,6 +652,7 @@ def get_minute(
return {
"symbol": symbol, "name": stock_name, "stock_info": stock_info,
"date": str(trade_date), "rows": df.to_dicts(), "source": "live",
+ "asset_type": asset_type,
"price_limit": price_limit,
}
@@ -656,6 +684,7 @@ def get_minute(
return {
"symbol": symbol, "name": stock_name, "stock_info": stock_info,
"date": str(trade_date), "rows": df.to_dicts(), "source": "local",
+ "asset_type": asset_type,
"price_limit": price_limit,
}
@@ -665,6 +694,7 @@ def get_minute(
"symbol": symbol, "name": stock_name, "stock_info": stock_info,
"date": str(trade_date), "rows": live_df.to_dicts(),
"source": "live" if not live_df.is_empty() else "none",
+ "asset_type": asset_type,
"price_limit": price_limit,
}
@@ -760,6 +790,9 @@ async def sync_minute(request: Request):
universe = sorted(set(universe) | set(inst["symbol"].to_list()))
except Exception: # noqa: BLE001
pass
+ # 剔除指数 symbol: 指数分钟K无本地存储, 落库会污染 kline_minute
+ index_set = repo.get_index_symbol_set()
+ universe = [s for s in universe if s not in index_set]
progress("sync_minute", 10, f"标的池 {len(universe)} 只")
days = override_days if override_days else get_minute_sync_days()
@@ -814,6 +847,11 @@ async def sync_minute_single(request: Request, body: dict):
repo = request.app.state.repo
capset = request.app.state.capabilities
+ # 指数分钟K无本地存储, 落库会污染股票分钟表 kline_minute;
+ # 指数分钟数据走 /api/index/minute 实时读取, 此端点显式拒绝。
+ if repo.resolve_asset_type(symbol) == "index":
+ raise HTTPException(status_code=400, detail="指数分钟K不支持落库同步 (指数分钟数据走 /api/index/minute 实时读取)")
+
if not _minute_allowed(capset):
raise HTTPException(status_code=403, detail="需要 Pro+ 权限")
diff --git a/backend/app/api/monitor_rules.py b/backend/app/api/monitor_rules.py
index 9cd0c30..34207f5 100644
--- a/backend/app/api/monitor_rules.py
+++ b/backend/app/api/monitor_rules.py
@@ -20,11 +20,35 @@ def _data_dir(request: Request) -> Path:
return request.app.state.repo.store.data_dir
+def _reconcile_index_asset_type(rule: dict, repo) -> dict:
+ """纠正误存为 stock 的指数规则 (asset_type → index)。
+
+ 个股弹窗加监控 / 点位提醒等入口未传 asset_type, 指数 symbol 的规则被存成
+ stock, 导致监控中心显示「个股」、引擎在股票轮评估 (指数 symbol 永不命中)。
+ 仅当规则全部 symbols 都 resolve 为指数时纠正 (股票+指数混合池不动)。
+ """
+ if rule.get("asset_type", "stock") != "stock" or rule.get("scope") != "symbols":
+ return rule
+ symbols = [s for s in rule.get("symbols", []) if s]
+ if not symbols:
+ return rule
+ try:
+ if all(repo.resolve_asset_type(s) == "index" for s in symbols):
+ rule["asset_type"] = "index"
+ except Exception: # noqa: BLE001
+ pass
+ return rule
+
+
def _sync_engine(request: Request) -> None:
"""保存/删除后,把最新规则集 reload 到引擎内存态。"""
engine = getattr(request.app.state, "monitor_engine", None)
if engine is not None:
- rules = monitor_rules.load_all(_data_dir(request))
+ repo = request.app.state.repo
+ rules = [
+ _reconcile_index_asset_type(r, repo)
+ for r in monitor_rules.load_all(_data_dir(request))
+ ]
engine.set_rules(rules)
@@ -101,13 +125,13 @@ def get_options(request: Request):
"custom_signals": custom_sigs,
"operators": [">", ">=", "<", "<=", "==", "!="],
"types": [
- {"key": "signal", "label": "个股信号"},
+ {"key": "signal", "label": "信号"},
{"key": "price", "label": "价格/涨跌"},
{"key": "market", "label": "市场异动"},
{"key": "strategy", "label": "策略监控"},
],
"scopes": [
- {"key": "symbols", "label": "指定股票"},
+ {"key": "symbols", "label": "指定标的"},
{"key": "all", "label": "全市场"},
{"key": "sector", "label": "板块"},
],
@@ -134,7 +158,11 @@ def get_options(request: Request):
# ── 列表 ───────────────────────────────────────────────
@router.get("")
def list_rules(request: Request):
- rules = monitor_rules.load_all(_data_dir(request))
+ repo = request.app.state.repo
+ rules = [
+ _reconcile_index_asset_type(r, repo)
+ for r in monitor_rules.load_all(_data_dir(request))
+ ]
from app.services.kline_sync import intraday_monitor_support
support = intraday_monitor_support(getattr(request.app.state, "capabilities", None))
@@ -167,6 +195,7 @@ def list_rules(request: Request):
@router.post("")
def save_rule(req: RuleModel, request: Request):
rule = monitor_rules.normalize(req.model_dump())
+ rule = _reconcile_index_asset_type(rule, request.app.state.repo)
# 连板梯队封单监控 (type=ladder) 依赖五档盘口数据, 需 Pro+ (DEPTH5_BATCH 能力)。
# 无能力时拒绝创建, 避免规则存了却永远无法触发。
if rule.get("type") == "ladder":
diff --git a/backend/app/api/watchlist.py b/backend/app/api/watchlist.py
index 13db6d8..ef8e6f6 100644
--- a/backend/app/api/watchlist.py
+++ b/backend/app/api/watchlist.py
@@ -187,8 +187,10 @@ def watchlist_enriched(
# 按资产拆分自选 symbol; ETF enriched 是独立缓存, 仅自选真的含 ETF 才去加载
# (避免无 ETF 用户在缓存冷启动时触发 ETF 全量懒加载)
etf_set = repo.get_etf_symbol_set()
- stock_symbols = [s for s in symbols if s not in etf_set]
+ index_set = repo.get_index_symbol_set()
etf_symbols = [s for s in symbols if s in etf_set]
+ index_symbols = [s for s in symbols if s not in etf_set and s in index_set]
+ stock_symbols = [s for s in symbols if s not in etf_set and s not in index_set]
df_e, cache_date = repo.get_enriched_latest()
@@ -217,8 +219,19 @@ def watchlist_enriched(
df_etf = etf_watchlist_df
df = df_etf if df.is_empty() else pl.concat([df, df_etf], how="diagonal_relaxed")
- # as_of 取两类缓存中较旧者, 避免把旧的 ETF 行标成股票缓存日期
- dates = [d for d in (cache_date if stock_symbols else None, etf_date) if d is not None]
+ # 指数行合并 (镜像 ETF 分支); 缺失列 (换手率/涨跌停信号等) 为 null
+ index_date = None
+ if index_symbols:
+ df_idx_all, index_date = repo.get_enriched_latest_asset("index")
+ idx_watchlist_df = pl.DataFrame({"symbol": index_symbols})
+ if not df_idx_all.is_empty():
+ df_idx = idx_watchlist_df.join(df_idx_all, on="symbol", how="left")
+ else:
+ df_idx = idx_watchlist_df
+ df = df_idx if df.is_empty() else pl.concat([df, df_idx], how="diagonal_relaxed")
+
+ # as_of 取三类缓存中较旧者
+ dates = [d for d in (cache_date if stock_symbols else None, etf_date, index_date) if d is not None]
as_of = min(dates) if dates else None
if df.is_empty():
return {"rows": [], "as_of": str(as_of) if as_of else None, "elapsed_ms": 0}
@@ -232,8 +245,14 @@ def watchlist_enriched(
pl.col("symbol").replace_strict(name_map, default=None, return_dtype=pl.Utf8).alias("name")
)
+ # 标注资产类型: 前端据此渲染徽标/豁免板块筛选/分时列降级
+ asset_map = {**{s: "etf" for s in etf_symbols}, **{s: "index" for s in index_symbols}}
+ df = df.with_columns(
+ pl.col("symbol").replace_strict(asset_map, default="stock", return_dtype=pl.Utf8).alias("asset_type")
+ )
+
# 选择内置需要的列
- keep = [c for c in _WATCHLIST_COLS + ["name", "float_shares"] if c in df.columns]
+ keep = [c for c in _WATCHLIST_COLS + ["name", "float_shares", "asset_type"] if c in df.columns]
df = df.select(keep)
# 动态 JOIN 扩展数据表
diff --git a/backend/app/jobs/daily_pipeline.py b/backend/app/jobs/daily_pipeline.py
index 861997e..f5edef1 100644
--- a/backend/app/jobs/daily_pipeline.py
+++ b/backend/app/jobs/daily_pipeline.py
@@ -54,11 +54,14 @@ def _invalidate(table: str | None = None) -> None:
invalidate_data_cache(table)
-def _resolve_universe(capset: CapabilitySet) -> list[str]:
+def _resolve_universe(capset: CapabilitySet, repo=None) -> list[str]:
"""解析标的池 — 以 CN_Equity_A (沪深京A股 ~5522只) 为主。
有 batch 能力 → 直接拉 CN_Equity_A universe
其他用户 → 用 instruments parquet + watchlist 兜底
+
+ repo 传入时过滤自选兜底里的指数 symbol (指数日K走独立 kline_index_* 存储,
+ 进股票池会污染 kline_daily/kline_minute)。ETF 刻意保留 (既有行为)。
"""
if capset.has(Cap.KLINE_DAILY_BATCH):
try:
@@ -79,6 +82,10 @@ def _resolve_universe(capset: CapabilitySet) -> list[str]:
base.update(inst["symbol"].to_list())
except Exception as e: # noqa: BLE001
logger.warning("instruments supplement failed: %s", e)
+ # 过滤自选兜底里的指数 symbol (指数日K走独立 kline_index_* 存储,
+ # 进股票池会污染 kline_daily/kline_minute)。ETF 刻意保留 (既有行为)。
+ if repo is not None:
+ base -= set(repo.get_index_symbol_set())
return sorted(base)
@@ -127,7 +134,7 @@ def run_now(
_invalidate("instruments")
emit("resolve_universe", 9, "解析标的池…")
- universe = _resolve_universe(capset)
+ universe = _resolve_universe(capset, repo)
emit("resolve_universe", 10, f"标的池规模:{len(universe)} 只")
# Step 1: 日 K 同步
@@ -488,7 +495,7 @@ def run_now(
minute_start = today - _td(days=minute_days)
emit("sync_minute", 90, f"获取分钟K [{minute_start} ~ {today}]…")
logger.info("sync_minute: [%s ~ %s] start", minute_start, today)
- minute_symbols = _resolve_minute_symbols(capset)
+ minute_symbols = _resolve_minute_symbols(capset, repo)
def _minute_chunk_progress(cur: int, tot: int, seg_label: str = "") -> None:
emit("sync_minute", 90 + int(3 * cur / tot),
f"分钟K 批次 {cur}/{tot}" + (f" [{seg_label}]" if seg_label else ""),
@@ -575,9 +582,9 @@ def _refresh_single_view(repo: KlineRepository, name: str) -> None:
logger.warning("refresh view %s failed: %s", name, e)
-def _resolve_minute_symbols(capset: CapabilitySet) -> list[str]:
+def _resolve_minute_symbols(capset: CapabilitySet, repo=None) -> list[str]:
"""分钟 K 同步标的 — 与日K共用同一标的池。"""
- return _resolve_universe(capset)
+ return _resolve_universe(capset, repo)
def _refresh_instruments_view(repo: KlineRepository) -> None:
diff --git a/backend/app/services/quote_service.py b/backend/app/services/quote_service.py
index eaf29aa..2fa064c 100644
--- a/backend/app/services/quote_service.py
+++ b/backend/app/services/quote_service.py
@@ -582,6 +582,14 @@ class QuoteService:
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)
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)
+ all_index_symbols.update(monitor_index_symbols)
all_etf_symbols = set()
if self._repo:
etf_inst = self._repo.get_etf_instruments()
@@ -604,7 +612,7 @@ class QuoteService:
logger.info("全市场行情拉取完成: %d 条 (%.2fs)", len(resp), time.perf_counter() - _u0)
if preferences.get_realtime_pull_index() and preferences.get_realtime_index_mode() == "core":
_i0 = time.perf_counter()
- _core_syms = sorted(core_index_symbols)
+ _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
@@ -712,6 +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")
+ # ---- 指数: 仅有指数监控规则时才写盘 (无规则零成本) ----
+ # 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:
+ 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", merge=not use_flush)
# ---- 通知 SSE ----
self._broadcast_quote_updated()
@@ -720,11 +743,22 @@ 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()
+ # 指数监控规则标的并入轮询 (与股票共享 batch 额度)
+ 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":
+ for _s in _r.get("symbols", []):
+ if _s and _s not in symbols:
+ symbols.append(_s)
if not symbols:
logger.info("自选实时未配置标的, 跳过行情拉取")
return
@@ -734,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("自选实时行情数据为空")
@@ -776,22 +817,27 @@ class QuoteService:
"session": q.get("session"),
})
+ index_set = self._repo.get_index_symbol_set() if self._repo else set()
+ etf_set = self._repo.get_etf_symbol_set() if self._repo else set()
+ index_records, etf_records, stock_records = self._split_records_by_asset(records, index_set, etf_set)
+
fetch_ms = (time.perf_counter() - t0) * 1000
fetched_at = time.time() * 1000
with self._lock:
self._fetch_time = now_ts
self._fetch_ms = fetch_ms
self._fetched_at = fetched_at
- self._symbol_count = len(records)
- self._index_symbol_count = 0
- self._etf_symbol_count = 0
- self._index_quotes_cache = None
+ 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 index_records else None
_persist_last_fetch(fetched_at)
- logger.info("自选实时刷新: %d 只股票, 耗时 %.0fms", len(records), fetch_ms)
+ logger.info("自选实时刷新: %d 只股票, %d 只ETF, %d 只指数, 耗时 %.0fms",
+ len(stock_records), len(etf_records), len(index_records), fetch_ms)
- daily_df = self._build_daily(records)
- quote_extra = self._build_quote_extra(records)
+ daily_df = self._build_daily(stock_records)
+ quote_extra = self._build_quote_extra(stock_records)
if not daily_df.is_empty() and self._repo:
try:
self._repo.merge_live_daily_asset("stock", daily_df)
@@ -799,6 +845,22 @@ class QuoteService:
logger.warning("自选实时日K写盘失败: %s", e)
self._flush_live_enriched(daily_df, quote_extra, asset_type="stock", merge=True)
+ # ETF/指数进自选前5时按各自资产落盘, 不污染股票表
+ etf_daily_df = self._build_daily(etf_records)
+ if not etf_daily_df.is_empty() and self._repo:
+ try:
+ self._repo.merge_live_daily_asset("etf", etf_daily_df)
+ except Exception as e: # noqa: BLE001
+ logger.warning("自选实时 ETF 日K写盘失败: %s", e)
+ self._flush_live_enriched(etf_daily_df, self._build_quote_extra(etf_records), asset_type="etf", merge=True)
+ index_daily_df = self._build_daily(index_records)
+ if not index_daily_df.is_empty() and self._repo:
+ try:
+ 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=True)
+
self._broadcast_quote_updated()
self._evaluate_monitors(daily_df, quote_extra)
@@ -806,6 +868,24 @@ class QuoteService:
# 工具
# ================================================================
+ @staticmethod
+ def _split_records_by_asset(
+ records: list[dict], index_set: set[str], etf_set: set[str],
+ ) -> tuple[list[dict], list[dict], list[dict]]:
+ """把行情 records 按资产拆成 (index, etf, stock)。判定顺序与 resolve_asset_type 一致: 先 ETF 后指数。"""
+ index_records: list[dict] = []
+ etf_records: list[dict] = []
+ stock_records: list[dict] = []
+ for r in records:
+ sym = r.get("symbol")
+ if sym in etf_set:
+ etf_records.append(r)
+ elif sym in index_set:
+ index_records.append(r)
+ else:
+ stock_records.append(r)
+ return index_records, etf_records, stock_records
+
@staticmethod
def _build_daily(records: list[dict]) -> pl.DataFrame:
"""将 API records 转为日K格式 DataFrame (OHLCV + quote_ts, 写 kline_daily 用)。"""
@@ -985,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] = []
@@ -1018,18 +1096,26 @@ class QuoteService:
for row in etf_inst.select(["symbol", "name"]).iter_rows(named=True):
if row.get("name"):
name_map.setdefault(row["symbol"], row["name"])
+ # 仅当存在指数规则时补指数维表 (setdefault 不覆盖股票/ETF)
+ if engine.has_asset_rules("index"):
+ idx_inst = self._app_state.repo.get_instruments_asset("index")
+ if not idx_inst.is_empty() and "symbol" in idx_inst.columns and "name" in idx_inst.columns:
+ for row in idx_inst.select(["symbol", "name"]).iter_rows(named=True):
+ if row.get("name"):
+ name_map.setdefault(row["symbol"], row["name"])
if name_map:
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 实时
@@ -1044,6 +1130,19 @@ class QuoteService:
)
except Exception as e: # noqa: BLE001
logger.warning("ETF 监控评估失败 (不影响股票告警): %s", e)
+ # 指数规则轮: 复刻 ETF 轮。快照由指数实时 flush 焐热;
+ # refresh=False 冷缓存不同步重算; 显式日期守卫防陈旧 parquet 误告警
+ # (ETF 轮靠空表隐式跳过, 指数轮更显式, 行为等价)。
+ if engine.has_asset_rules("index") and self._repo is not None:
+ try:
+ index_enriched, index_date = self._repo.get_enriched_latest_asset("index", refresh=False)
+ if not index_enriched.is_empty() and index_date == cn_today():
+ index_enriched = self._inject_intraday_signals(index_enriched, engine, "index")
+ rule_events = rule_events + engine.evaluate(
+ index_enriched, asset_type="index", reset_strategy_results=False,
+ )
+ except Exception as e: # noqa: BLE001
+ logger.warning("指数监控评估失败 (不影响股票/ETF 告警): %s", e)
if rule_events:
# 落盘到 alerts.jsonl
try:
@@ -1380,7 +1479,7 @@ class QuoteService:
"ok" if not live_agg.is_empty() else "空", prev_date)
cutoff = today - timedelta(days=90)
- table = "kline_etf_daily" if asset_type == "etf" else "kline_daily"
+ table = {"etf": "kline_etf_daily", "index": "kline_index_daily"}.get(asset_type, "kline_daily")
daily_glob = str(self._repo.store.data_dir / table / "**" / "*.parquet")
ohlcv_cols = ["symbol", "date", "open", "high", "low", "close", "volume", "amount", "quote_ts"]
hist_df = (
@@ -1398,10 +1497,10 @@ class QuoteService:
full_df = pl.concat([hist_df, daily_ohlcv], how="diagonal_relaxed")
full_df = full_df.sort(["symbol", "date"])
- factor_dir = "adj_factor_etf" if asset_type == "etf" else "adj_factor"
- factor_path = self._repo.store.data_dir / factor_dir / "all.parquet"
+ factor_dir = {"stock": "adj_factor", "etf": "adj_factor_etf"}.get(asset_type)
+ factor_path = self._repo.store.data_dir / factor_dir / "all.parquet" if factor_dir else None
factors = pl.DataFrame()
- if factor_path.exists():
+ if factor_path and factor_path.exists():
try:
factors = pl.read_parquet(factor_path)
except Exception:
diff --git a/backend/app/services/stock_analyzer.py b/backend/app/services/stock_analyzer.py
index 298739f..0c2af8e 100644
--- a/backend/app/services/stock_analyzer.py
+++ b/backend/app/services/stock_analyzer.py
@@ -198,8 +198,12 @@ def _build_user_prompt(
close: float | None,
symbol: str,
focus: str,
+ asset_type: str = "stock",
) -> str:
- """构建用户消息:标的 + 价位摘要 + 技术指标 JSON + 财务摘要 + 关注点。"""
+ """构建用户消息:标的 + 价位摘要 + 技术指标 JSON + 财务摘要 + 关注点。
+
+ asset_type 用于区分无财务数据时的文案:指数无财务是常态,不走 Free 文案。
+ """
parts: list[str] = [
f"标的标准代码: {symbol}",
f"关键价位概览: {summarize_levels(levels, close)}",
@@ -220,6 +224,13 @@ def _build_user_prompt(
json.dumps(fins, ensure_ascii=False),
"```",
])
+ elif asset_type == "index":
+ parts.extend([
+ "",
+ "(该标的为指数: 无财务、股本与涨跌停数据。请按系统提示词第 4 节的说明,"
+ "在基本面/财务面维度给出\"接入中\"的友好提示,不要编造数据;"
+ "消息面维度基于价量异动推断即可。)",
+ ])
else:
parts.extend([
"",
@@ -302,7 +313,8 @@ async def analyze_stock_stream(
from app.services.ai_provider import stream_ai_text
kline_tail = _clean_rows(df, _KLINE_KEEP_COLS)
- user_prompt = _build_user_prompt(kline_tail, fins, levels, close, symbol, focus)
+ user_prompt = _build_user_prompt(kline_tail, fins, levels, close, symbol, focus,
+ asset_type=repo.resolve_asset_type(symbol))
async for delta in stream_ai_text(
[
{"role": "system", "content": _SYSTEM_PROMPT},
diff --git a/backend/app/strategy/monitor_rules.py b/backend/app/strategy/monitor_rules.py
index dd2d2a6..ada8cfd 100644
--- a/backend/app/strategy/monitor_rules.py
+++ b/backend/app/strategy/monitor_rules.py
@@ -107,6 +107,16 @@ def validate(rule: dict) -> None:
if rule.get("type") not in RULE_TYPES:
raise ValueError(f"type 必须是 {RULE_TYPES} 之一")
+ # 指数规则: 仅 signal/price + symbols 作用域 + 不含分时信号
+ # (指数无涨跌停/策略/封单语义; 无本地分钟K, 分时信号会静默不触发)
+ if rule.get("asset_type") == "index":
+ if rule.get("type") not in ("signal", "price"):
+ raise ValueError("指数监控仅支持 signal/price 类型 (无涨跌停/策略/封单语义)")
+ if rule.get("scope") != "symbols":
+ raise ValueError("指数监控仅支持指定标的 (scope=symbols)")
+ if uses_intraday_signals(rule):
+ raise ValueError("指数无本地分钟K数据, 不支持分时信号条件")
+
# 策略类型: 需要 strategy_id + direction,conditions 可空
if rule.get("type") == "strategy":
if not rule.get("strategy_id"):
@@ -163,7 +173,7 @@ def validate(rule: dict) -> None:
if not isinstance(syms, list) or len(syms) == 0:
raise ValueError("scope=symbols 时 symbols 不能为空")
if uses_intraday_signals(rule) and rule.get("scope") != "symbols":
- raise ValueError("分时穿越信号仅支持指定股票")
+ raise ValueError("分时穿越信号仅支持指定标的")
# sector 作用域的板块 JOIN 尚未实现: _apply_scope 目前会退化为「全市场」,
# 一条本意针对某板块的规则会对全市场每只命中都触发(告警风暴)。在板块 JOIN
# 落地前, 拒绝创建 sector 规则(fail-closed), 避免用户建出会刷屏的规则。
diff --git a/backend/app/tickflow/repository.py b/backend/app/tickflow/repository.py
index a8bf231..c2ca060 100644
--- a/backend/app/tickflow/repository.py
+++ b/backend/app/tickflow/repository.py
@@ -327,6 +327,8 @@ class KlineRepository:
# symbol 集合 memo (随对应 instruments 缓存失效): 供每请求资产分流用
self._index_symbol_set_cache: set[str] | None = None
self._etf_symbol_set_cache: set[str] | None = None
+ self._index_enriched_cache: pl.DataFrame | None = None
+ self._index_enriched_cache_date: date | None = None
# ---- enriched 后台预热 ----
# 启动时 compute_indicators (107万行, 低配机 50s+) 移出 lifespan 关键路径,
@@ -394,6 +396,9 @@ class KlineRepository:
# 避免自选无 ETF 的用户在管道后白付全量重算成本
self._etf_enriched_cache = None
self._etf_enriched_cache_date = None
+ # 指数 enriched 同样只失效不重建 (懒加载)
+ self._index_enriched_cache = None
+ self._index_enriched_cache_date = None
if background:
logger.info("cache refresh: enriched 推后台线程预热")
@@ -484,6 +489,8 @@ class KlineRepository:
self._etf_instruments_cache = None
self._index_symbol_set_cache = None
self._etf_symbol_set_cache = None
+ self._index_enriched_cache = None
+ self._index_enriched_cache_date = None
def _refresh_enriched(self) -> None:
"""从 parquet 加载 enriched 最新日到内存 + 构建聚合表。
@@ -986,6 +993,50 @@ class KlineRepository:
except Exception as e: # noqa: BLE001
logger.debug("ETF enriched 缓存刷新跳过: %s", e)
+ def _refresh_index_enriched(self) -> None:
+ """从指数 enriched parquet 加载最新日到内存缓存 (300天重算通用指标)。
+
+ 磁盘窄表无指标列, 必须 scan 近 300 天重算, 否则监控信号规则无列可评估。
+ 指数无复权需求, 不读 raw_close/raw_high/raw_low。
+ """
+ try:
+ enriched_dir = self.store.data_dir / "kline_index_enriched"
+ dates = sorted(
+ p.name[5:] for p in enriched_dir.glob("date=*")
+ if p.is_dir() and p.name.startswith("date=")
+ ) if enriched_dir.exists() else []
+ if not dates:
+ self._index_enriched_cache = None
+ self._index_enriched_cache_date = None
+ return
+ latest = date.fromisoformat(dates[-1])
+ target_parquet = enriched_dir / f"date={dates[-1]}" / "part.parquet"
+ df_latest = pl.read_parquet(target_parquet)
+ if df_latest.is_empty():
+ return
+
+ from datetime import timedelta
+ start_full = latest - timedelta(days=300)
+ read_cols = [c for c in ["symbol", "date", "open", "high", "low", "close",
+ "volume", "amount"]
+ if c in df_latest.columns]
+ df_hist = (
+ scan_enriched_parquet(self._index_enriched_glob,
+ cast_options=pl.ScanCastOptions(integer_cast="allow-float"))
+ .filter(pl.col("date") >= start_full)
+ .select(read_cols)
+ .sort(["symbol", "date"])
+ .collect()
+ )
+ if df_hist.is_empty():
+ self._index_enriched_cache = df_latest.sort(["symbol"])
+ else:
+ df_full = self._compute_index_enriched_range(df_hist)
+ self._index_enriched_cache = df_full.filter(pl.col("date") == latest).sort(["symbol"])
+ self._index_enriched_cache_date = latest
+ except Exception as e: # noqa: BLE001
+ logger.debug("指数 enriched 缓存刷新跳过: %s", e)
+
def _refresh_instruments(self) -> None:
"""加载 instruments 到内存。"""
try:
@@ -1059,6 +1110,12 @@ class KlineRepository:
if self._etf_enriched_cache is None:
return pl.DataFrame(), self._etf_enriched_cache_date
return self._etf_enriched_cache, self._etf_enriched_cache_date
+ if asset_type == "index":
+ if self._index_enriched_cache is None and refresh:
+ self._refresh_index_enriched()
+ if self._index_enriched_cache is None:
+ return pl.DataFrame(), self._index_enriched_cache_date
+ return self._index_enriched_cache, self._index_enriched_cache_date
return pl.DataFrame(), None
def get_enriched_history(self, target_date: date, lookback_days: int) -> pl.DataFrame | None:
@@ -1237,13 +1294,13 @@ class KlineRepository:
return "stock"
def get_name_map(self, symbols: list[str] | None = None) -> dict[str, str]:
- """返回 {symbol: name} 映射, 合并股票 + ETF instruments (股票优先去重)。
+ """返回 {symbol: name} 映射, 合并股票 + ETF + 指数 instruments (股票优先去重)。
自选列表/名称批查等场景的统一名称解析入口, 避免各调用方自行合并两份缓存。
symbols 非 None 时只返回命中的条目。
"""
name_map: dict[str, str] = {}
- for df in (self.get_instruments(), self.get_etf_instruments()):
+ for df in (self.get_instruments(), self.get_etf_instruments(), self.get_instruments_asset("index")):
if df.is_empty() or "symbol" not in df.columns or "name" not in df.columns:
continue
if symbols is not None:
@@ -2031,7 +2088,7 @@ class KlineRepository:
existing_cache = self._etf_enriched_cache if self._etf_enriched_cache_date == dt else pl.DataFrame()
elif asset_type == "index":
table = "kline_index_enriched"
- existing_cache = pl.DataFrame()
+ existing_cache = self._index_enriched_cache if self._index_enriched_cache_date == dt else pl.DataFrame()
else:
return
@@ -2048,6 +2105,9 @@ class KlineRepository:
elif asset_type == "etf":
self._etf_enriched_cache = merged_cache
self._etf_enriched_cache_date = dt
+ elif asset_type == "index":
+ self._index_enriched_cache = merged_cache
+ self._index_enriched_cache_date = dt
from app.indicators.pipeline import ENRICHED_STORAGE_COLS
storage_cols = [c for c in ENRICHED_STORAGE_COLS if c in df.columns]
@@ -2113,6 +2173,8 @@ class KlineRepository:
self._etf_enriched_cache_date = dt
table = "kline_etf_enriched"
elif asset_type == "index":
+ self._index_enriched_cache = cache_df
+ self._index_enriched_cache_date = dt
table = "kline_index_enriched"
else:
return
diff --git a/backend/tests/test_daily_batch_asset.py b/backend/tests/test_daily_batch_asset.py
new file mode 100644
index 0000000..bba6e4f
--- /dev/null
+++ b/backend/tests/test_daily_batch_asset.py
@@ -0,0 +1,42 @@
+"""daily-batch 混合资产分组测试。"""
+import datetime as _dt
+
+import polars as pl
+import pytest
+
+from app.tickflow.repository import DataStore, KlineRepository
+
+
+@pytest.fixture()
+def repo(tmp_path):
+ return KlineRepository(DataStore(tmp_path))
+
+
+def test_daily_batch_groups_index_symbols(repo, monkeypatch):
+ from app.api import kline as kline_api
+
+ calls = {"stock_batch": [], "index": []}
+
+ def fake_stock_batch(symbols, start, end, columns=None):
+ calls["stock_batch"].append(list(symbols))
+ return pl.DataFrame()
+
+ def fake_index_daily(symbol, start, end, columns=None):
+ calls["index"].append(symbol)
+ return pl.DataFrame({
+ "symbol": [symbol], "date": [_dt.date(2026, 7, 24)],
+ "open": [1.0], "high": [1.0], "low": [1.0], "close": [1.0], "volume": [1],
+ })
+
+ monkeypatch.setattr(repo, "get_daily_batch", fake_stock_batch)
+ monkeypatch.setattr(repo, "get_index_daily", fake_index_daily)
+ monkeypatch.setattr(repo, "get_index_symbol_set", lambda: {"000001.SH"})
+ monkeypatch.setattr(repo, "get_etf_symbol_set", lambda: set())
+
+ state = type("S", (), {"repo": repo})()
+ req = type("R", (), {"app": type("A", (), {"state": state})()})()
+
+ out = kline_api.get_daily_batch(req, {"symbols": ["600000.SH", "000001.SH"], "days": 12})
+ assert calls["stock_batch"] == [["600000.SH"]]
+ assert calls["index"] == ["000001.SH"]
+ assert "000001.SH" in out["data"]
diff --git a/backend/tests/test_intraday_monitor_signals.py b/backend/tests/test_intraday_monitor_signals.py
index 5473227..efd9a11 100644
--- a/backend/tests/test_intraday_monitor_signals.py
+++ b/backend/tests/test_intraday_monitor_signals.py
@@ -144,7 +144,7 @@ def test_intraday_rule_pool_is_derived_from_enabled_rules():
def test_intraday_rule_rejects_non_symbol_scope():
- with pytest.raises(ValueError, match="仅支持指定股票"):
+ with pytest.raises(ValueError, match="仅支持指定标的"):
monitor_rules.validate(_intraday_rule("all"))
diff --git a/backend/tests/test_minute_routing.py b/backend/tests/test_minute_routing.py
index 0aa1274..6792d35 100644
--- a/backend/tests/test_minute_routing.py
+++ b/backend/tests/test_minute_routing.py
@@ -680,3 +680,25 @@ def test_intraday_monitor_support_resolver_exception_falls_back(monkeypatch):
assert support["available"] is True
assert support["source"] == "minute_batch"
+
+
+# ---------- 测试 20: sync_minute_single 拒绝指数 symbol (防污染 kline_minute) ----------
+
+def test_sync_minute_single_rejects_index_symbol():
+ """指数分钟K无本地存储, 落库会污染股票分钟表; 端点应显式 400 而非 500。"""
+ import asyncio
+
+ import pytest
+ from fastapi import HTTPException
+
+ from app.api import kline as kline_api
+
+ mock_repo = MagicMock()
+ mock_repo.resolve_asset_type.return_value = "index"
+ mock_request = MagicMock()
+ mock_request.app.state.repo = mock_repo
+
+ with pytest.raises(HTTPException) as exc_info:
+ asyncio.run(kline_api.sync_minute_single(mock_request, {"symbol": "000001.SH"}))
+ assert exc_info.value.status_code == 400
+ assert "指数" in str(exc_info.value.detail)
diff --git a/backend/tests/test_monitor_index.py b/backend/tests/test_monitor_index.py
new file mode 100644
index 0000000..6e19ef4
--- /dev/null
+++ b/backend/tests/test_monitor_index.py
@@ -0,0 +1,202 @@
+"""指数监控规则校验测试。"""
+import pytest
+
+from app.strategy import monitor_rules
+
+
+def _index_rule(rid="r_idx", **over):
+ rule = {
+ "id": rid, "name": rid, "type": "signal", "asset_type": "index",
+ "scope": "symbols", "symbols": ["000001.SH"], "logic": "and",
+ "conditions": [{"field": "rsi_14", "op": "<", "value": 30}],
+ "cooldown_seconds": 0, "enabled": True,
+ }
+ rule.update(over)
+ return rule
+
+
+def test_index_signal_price_allowed():
+ monitor_rules.validate(_index_rule())
+ monitor_rules.validate(_index_rule(type="price"))
+
+
+def test_index_strategy_rejected():
+ with pytest.raises(ValueError, match="指数"):
+ monitor_rules.validate(_index_rule(type="strategy", strategy_id="s1"))
+
+
+def test_index_market_rejected():
+ with pytest.raises(ValueError, match="指数"):
+ monitor_rules.validate(_index_rule(type="market"))
+
+
+def test_index_scope_all_rejected():
+ with pytest.raises(ValueError, match="指数"):
+ monitor_rules.validate(_index_rule(scope="all", symbols=[]))
+
+
+def test_index_intraday_signal_rejected():
+ with pytest.raises(ValueError, match="分时"):
+ monitor_rules.validate(_index_rule(
+ conditions=[{"field": "signal_intraday_avg_cross_up", "op": "truth"}],
+ ))
+
+
+# ---- Task 7: B5 监控指数评估轮 ----
+
+def _signal_rule(rid, asset_type, sym):
+ return {
+ "id": rid, "name": rid, "type": "signal", "asset_type": asset_type,
+ "scope": "symbols", "symbols": [sym], "logic": "and",
+ "conditions": [{"field": "rsi_14", "op": "<", "value": 100}],
+ "cooldown_seconds": 0, "enabled": True,
+ }
+
+
+def test_evaluate_index_round_triggers_and_isolates():
+ """指数轮只评估指数规则, 且不触碰策略结果缓存。"""
+ import polars as pl
+ from app.strategy.monitor import MonitorRuleEngine
+
+ eng = MonitorRuleEngine()
+ eng.set_rules([_signal_rule("r_idx", "index", "000001.SH"),
+ _signal_rule("r_stock", "stock", "000001.SH")])
+ eng.set_name_map({"000001.SH": "上证指数"})
+ df = pl.DataFrame({"symbol": ["000001.SH"], "close": [3000.0],
+ "change_pct": [0.01], "rsi_14": [40.0]})
+
+ events = eng.evaluate(df, asset_type="index", reset_strategy_results=False)
+ assert any(e["rule_id"] == "r_idx" for e in events)
+ assert all(e["rule_id"] != "r_stock" for e in events)
+ assert events[0]["name"] == "上证指数"
+ assert eng.latest_strategy_results() == {} # 策略结果缓存未被触碰
+
+
+# ---- 资产类型纠正: 误存为 stock 的指数规则 ----
+
+class _FakeRepo:
+ def resolve_asset_type(self, symbol):
+ return {"000001.SH": "index"}.get(symbol, "stock")
+
+
+def test_reconcile_index_asset_type_corrects_index_only_rule():
+ from app.api.monitor_rules import _reconcile_index_asset_type
+
+ rule = {"asset_type": "stock", "scope": "symbols", "symbols": ["000001.SH"]}
+ assert _reconcile_index_asset_type(rule, _FakeRepo())["asset_type"] == "index"
+
+
+def test_reconcile_index_asset_type_keeps_stock_and_mixed():
+ from app.api.monitor_rules import _reconcile_index_asset_type
+
+ repo = _FakeRepo()
+ # 纯股票 → 不动
+ assert _reconcile_index_asset_type(
+ {"asset_type": "stock", "scope": "symbols", "symbols": ["600000.SH"]}, repo,
+ )["asset_type"] == "stock"
+ # 股票+指数混合 → 不动 (asset_type 语义覆盖整条规则)
+ assert _reconcile_index_asset_type(
+ {"asset_type": "stock", "scope": "symbols", "symbols": ["000001.SH", "600000.SH"]}, repo,
+ )["asset_type"] == "stock"
+ # 已是 index → 不动
+ assert _reconcile_index_asset_type(
+ {"asset_type": "index", "scope": "symbols", "symbols": ["000001.SH"]}, repo,
+ )["asset_type"] == "index"
+ # 非 symbols 范围 → 不动
+ 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, "股票快照就绪时股票轮应正常执行"
diff --git a/backend/tests/test_quote_index_merge.py b/backend/tests/test_quote_index_merge.py
new file mode 100644
index 0000000..56815ea
--- /dev/null
+++ b/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
diff --git a/backend/tests/test_repository_index.py b/backend/tests/test_repository_index.py
new file mode 100644
index 0000000..28f12ef
--- /dev/null
+++ b/backend/tests/test_repository_index.py
@@ -0,0 +1,109 @@
+"""指数资产路由 — repository 层测试。"""
+import polars as pl
+import pytest
+
+from app.tickflow.repository import DataStore, KlineRepository
+
+
+@pytest.fixture()
+def repo(tmp_path):
+ return KlineRepository(DataStore(tmp_path))
+
+
+def _write_index_instruments(repo, rows):
+ pl.DataFrame(rows).write_parquet(
+ repo.store.data_dir / "instruments_index" / "part.parquet"
+ )
+ repo._refresh_index_instruments()
+
+
+def test_name_map_includes_index(repo):
+ _write_index_instruments(repo, {
+ "symbol": ["000001.SH"], "name": ["上证指数"],
+ "code": ["000001"], "asset_type": ["index"],
+ })
+ names = repo.get_name_map(["000001.SH", "600000.SH"])
+ assert names.get("000001.SH") == "上证指数"
+ assert "600000.SH" not in names # 未收录不造名
+
+
+def test_name_map_stock_beats_index(repo):
+ """同名 symbol 同时出现在股票/指数维表时, 股票名称优先。"""
+ _write_index_instruments(repo, {
+ "symbol": ["600000.SH"], "name": ["某指数"],
+ "code": ["600000"], "asset_type": ["index"],
+ })
+ pl.DataFrame({
+ "symbol": ["600000.SH"], "name": ["浦发银行"], "code": ["600000"],
+ "exchange": ["SH"], "region": ["CN"], "type": ["stock"],
+ "listing_date": [None], "total_shares": [None], "float_shares": [None],
+ "tick_size": [None], "limit_up": [None], "limit_down": [None],
+ "as_of": ["2026-07-25"],
+ }).write_parquet(repo.store.data_dir / "instruments" / "instruments.parquet")
+ repo._refresh_instruments()
+ assert repo.get_name_map(["600000.SH"]).get("600000.SH") == "浦发银行"
+
+
+import datetime as _dt
+
+
+def _write_index_enriched(repo, dates_rows):
+ for ds, rows in dates_rows.items():
+ d = repo.store.data_dir / "kline_index_enriched" / f"date={ds}"
+ d.mkdir(parents=True, exist_ok=True)
+ pl.DataFrame(rows).write_parquet(d / "part.parquet")
+
+
+def _index_rows(ds, close=3000.0):
+ return [{
+ "symbol": "000001.SH", "date": _dt.date.fromisoformat(ds),
+ "open": close - 10, "high": close + 20, "low": close - 20, "close": close,
+ "volume": 1_000_000, "amount": 1e9,
+ }]
+
+
+def test_get_enriched_latest_asset_index(repo):
+ _write_index_enriched(repo, {
+ "2026-07-23": _index_rows("2026-07-23", 2990.0),
+ "2026-07-24": _index_rows("2026-07-24", 3000.0),
+ })
+ df, dt = repo.get_enriched_latest_asset("index")
+ assert str(dt) == "2026-07-24"
+ assert df["symbol"].to_list() == ["000001.SH"]
+ assert "ma5" in df.columns or "rsi_14" in df.columns # 重算产出指标列
+
+
+def test_get_enriched_latest_asset_index_cold_no_refresh(repo):
+ df, dt = repo.get_enriched_latest_asset("index", refresh=False)
+ assert df.is_empty() and dt is None
+
+
+def test_flush_live_enriched_asset_index_updates_cache(repo):
+ df = pl.DataFrame([{
+ "symbol": "000001.SH", "date": _dt.date(2026, 7, 25),
+ "open": 3000.0, "high": 3010.0, "low": 2990.0, "close": 3005.0,
+ "volume": 1_000_000, "amount": 1e9, "ma5": 3001.0, "rsi_14": 55.0,
+ }])
+ repo.flush_live_enriched_asset("index", df)
+ cached, dt = repo.get_enriched_latest_asset("index", refresh=False)
+ assert str(dt) == "2026-07-25"
+ assert cached["close"].to_list() == [3005.0]
+ assert (repo.store.data_dir / "kline_index_enriched" / "date=2026-07-25" / "part.parquet").exists()
+
+
+def _merge_row(symbol, close):
+ return {
+ "symbol": symbol, "date": _dt.date(2026, 7, 25),
+ "open": close - 5, "high": close + 5, "low": close - 6, "close": close,
+ "volume": 1_000, "amount": 1e6,
+ }
+
+
+def test_merge_live_enriched_asset_index_merges_cache(repo):
+ """merge 路径: 两次合并缓存取并集 (不 NameError, 不丢已有缓存)。"""
+ repo.merge_live_enriched_asset("index", pl.DataFrame([_merge_row("000001.SH", 3000.0)]))
+ repo.merge_live_enriched_asset("index", pl.DataFrame([_merge_row("000300.SH", 4000.0)]))
+ cached, dt = repo.get_enriched_latest_asset("index", refresh=False)
+ assert str(dt) == "2026-07-25"
+ assert set(cached["symbol"].to_list()) == {"000001.SH", "000300.SH"}
+
diff --git a/backend/tests/test_resolve_universe_index.py b/backend/tests/test_resolve_universe_index.py
new file mode 100644
index 0000000..ef27d0e
--- /dev/null
+++ b/backend/tests/test_resolve_universe_index.py
@@ -0,0 +1,29 @@
+"""_resolve_universe 指数过滤测试。"""
+import pytest
+
+from app.jobs import daily_pipeline
+from app.tickflow.repository import DataStore, KlineRepository
+
+
+@pytest.fixture()
+def repo(tmp_path):
+ return KlineRepository(DataStore(tmp_path))
+
+
+def test_resolve_universe_excludes_index_symbols(repo, monkeypatch, tmp_path):
+ """自选里的指数不进入股票日K/分钟K同步池。"""
+ class _Capset:
+ def has(self, cap):
+ return False
+
+ monkeypatch.setattr(
+ daily_pipeline, "get_pool",
+ lambda name, refresh=False: ["600000.SH", "000001.SH"] if name == "watchlist" else [],
+ )
+ monkeypatch.setattr(daily_pipeline, "DEMO_SYMBOLS", [])
+ monkeypatch.setattr(daily_pipeline.settings, "data_dir", tmp_path)
+ monkeypatch.setattr(repo, "get_index_symbol_set", lambda: {"000001.SH"})
+
+ universe = daily_pipeline._resolve_universe(_Capset(), repo)
+ assert "600000.SH" in universe
+ assert "000001.SH" not in universe
diff --git a/backend/tests/test_stock_analyzer_index.py b/backend/tests/test_stock_analyzer_index.py
new file mode 100644
index 0000000..f079e57
--- /dev/null
+++ b/backend/tests/test_stock_analyzer_index.py
@@ -0,0 +1,13 @@
+"""AI 分析 prompt 指数文案测试。"""
+from app.services.stock_analyzer import _build_user_prompt
+
+
+def test_user_prompt_index_no_financials():
+ prompt = _build_user_prompt(
+ kline_tail=[{"date": "2026-07-24", "close": 3000.0}],
+ fins={"metrics": [], "income": []},
+ levels={}, close=3000.0, symbol="000001.SH", focus="",
+ asset_type="index",
+ )
+ assert "指数" in prompt
+ assert "Free 模式" not in prompt # 指数无财务是常态, 不走 Free 文案
diff --git a/backend/tests/test_watchlist_batch.py b/backend/tests/test_watchlist_batch.py
new file mode 100644
index 0000000..251d240
--- /dev/null
+++ b/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"]
diff --git a/backend/tests/test_watchlist_enriched_join.py b/backend/tests/test_watchlist_enriched_join.py
index ff9925f..d4a4e4d 100644
--- a/backend/tests/test_watchlist_enriched_join.py
+++ b/backend/tests/test_watchlist_enriched_join.py
@@ -20,14 +20,18 @@ class _FakeRepo:
"""最小化 repo mock: 只实现 watchlist_enriched 调用到的方法."""
def __init__(self, enriched_df, enriched_date, etf_df=None, etf_date=None,
- instruments_df=None, name_map=None, etf_set=None):
+ instruments_df=None, name_map=None, etf_set=None,
+ index_df=None, index_date=None, index_set=None):
self._enriched = enriched_df
self._enriched_date = enriched_date
self._etf = etf_df
self._etf_date = etf_date
+ self._index = index_df
+ self._index_date = index_date
self._instruments = instruments_df or pl.DataFrame()
self._name_map = name_map or {}
self._etf_set = etf_set or set()
+ self._index_set = index_set or set()
def get_enriched_latest(self):
return self._enriched, self._enriched_date
@@ -36,11 +40,17 @@ class _FakeRepo:
if asset == "etf":
etf = self._etf if self._etf is not None else pl.DataFrame()
return etf, self._etf_date
+ if asset == "index":
+ idx = self._index if self._index is not None else pl.DataFrame()
+ return idx, self._index_date
return pl.DataFrame(), None
def get_etf_symbol_set(self):
return self._etf_set
+ def get_index_symbol_set(self):
+ return self._index_set
+
def get_instruments(self):
return self._instruments
@@ -212,3 +222,45 @@ def test_mixed_watchlist_keeps_pending_etf_rows(monkeypatch):
assert all(next(r for r in res["rows"] if r["symbol"] == symbol).get("close") is None
for symbol in ("510300", "510500"))
assert res["as_of"] == "2026-07-08"
+
+
+def test_watchlist_enriched_index_branch(monkeypatch):
+ """自选含指数: 行走 index enriched, asset_type=index, 名称回填, 股票/ETF 行不受影响。
+
+ 断言 (计划 Task 4 Step 1):
+ - 指数行存在, close == 3000.0, asset_type == "index", name == "上证指数"
+ - 指数行 turnover_rate 为 None (列不存在或 null, 不报错)
+ - 股票行 asset_type == "stock"; ETF 行 == "etf"
+ - as_of == min(股票日期, etf日期, index日期)
+ """
+ monkeypatch.setattr(wl_api.watchlist, "list_symbols",
+ lambda: [{"symbol": "600000.SH"}, {"symbol": "510300.SH"},
+ {"symbol": "000001.SH"}])
+ repo = _FakeRepo(
+ enriched_df=_enriched_df([("600000.SH", 10.0, 0.3, 2e9)]),
+ enriched_date="2026-07-23",
+ etf_df=_enriched_df([("510300.SH", 4.0, 0.5, 1e8)]),
+ etf_date="2026-07-24",
+ etf_set={"510300.SH"},
+ index_df=pl.DataFrame([{"symbol": "000001.SH", "close": 3000.0, "change_pct": 0.01,
+ "amount": 1e9, "ma5": 2990.0}]),
+ index_date="2026-07-24",
+ index_set={"000001.SH"},
+ name_map={"600000.SH": "浦发银行", "510300.SH": "沪深300ETF", "000001.SH": "上证指数"},
+ )
+
+ res = wl_api.watchlist_enriched(_make_request(repo), ext_columns=None)
+
+ rows = {r["symbol"]: r for r in res["rows"]}
+ # 指数行
+ idx = rows["000001.SH"]
+ assert idx["close"] == 3000.0
+ assert idx["asset_type"] == "index"
+ assert idx["name"] == "上证指数"
+ # 指数无换手率: 列缺失或 null, 不报错
+ assert idx.get("turnover_rate") is None
+ # 股票行 / ETF 行 asset_type
+ assert rows["600000.SH"]["asset_type"] == "stock"
+ assert rows["510300.SH"]["asset_type"] == "etf"
+ # as_of == min(三类缓存日期)
+ assert res["as_of"] == "2026-07-23"
diff --git a/backend/tests/test_watchlist_realtime_split.py b/backend/tests/test_watchlist_realtime_split.py
new file mode 100644
index 0000000..7529a96
--- /dev/null
+++ b/backend/tests/test_watchlist_realtime_split.py
@@ -0,0 +1,19 @@
+"""Free 档自选实时资产分流测试。"""
+from app.services.quote_service import QuoteService
+
+
+def test_split_records_by_asset():
+ records = [
+ {"symbol": "600000.SH"}, {"symbol": "510300.SH"}, {"symbol": "000001.SH"},
+ ]
+ index, etf, stock = QuoteService._split_records_by_asset(
+ records, {"000001.SH"}, {"510300.SH"},
+ )
+ assert [r["symbol"] for r in index] == ["000001.SH"]
+ assert [r["symbol"] for r in etf] == ["510300.SH"]
+ assert [r["symbol"] for r in stock] == ["600000.SH"]
+ # etf 优先于 index (与 resolve_asset_type 判定顺序一致)
+ index2, etf2, stock2 = QuoteService._split_records_by_asset(
+ [{"symbol": "X"}], {"X"}, {"X"},
+ )
+ assert etf2 and not index2 and not stock2
diff --git a/frontend/src/components/StockIntradayChart.tsx b/frontend/src/components/StockIntradayChart.tsx
index 22afd75..438c699 100644
--- a/frontend/src/components/StockIntradayChart.tsx
+++ b/frontend/src/components/StockIntradayChart.tsx
@@ -48,6 +48,8 @@ export function StockIntradayChart({
// source=none 表示本地无数据且 TickFlow 也拉不到 (停牌/复牌延迟/非交易日)
// 此时不弹"是否获取"询问窗, 只做静态提示, 避免误导用户去拉明知拉不到的数据
const sourceIsNone = minute.data?.source === 'none'
+ // 指数分钟K无本地存储且不支持落库获取 (后端 sync_minute_single 显式拒绝), 不显示获取按钮
+ const isIndex = minute.data?.asset_type === 'index'
useEffect(() => {
setMinuteDismissed(false)
@@ -66,6 +68,9 @@ export function StockIntradayChart({
正在获取分钟K数据…
+ ) : isIndex ? (
+ // 指数: 分钟K仅支持实时读取, 无落库获取入口
+ 指数暂无分钟数据
) : sourceIsNone ? (
// 数据源确认无此日分钟数据 (停牌/复牌延迟等): 静态提示 + 保留重试
diff --git a/frontend/src/components/financials/StockFinancialSearch.tsx b/frontend/src/components/financials/StockFinancialSearch.tsx
index b5f408a..2b739b8 100644
--- a/frontend/src/components/financials/StockFinancialSearch.tsx
+++ b/frontend/src/components/financials/StockFinancialSearch.tsx
@@ -7,6 +7,8 @@ import { QK } from '@/lib/queryKeys'
interface Props {
onSelect: (symbol: string, name: string) => void
+ /** 搜索资产类型, 逗号分隔 (默认 'stock')。如 'stock,index' */
+ assetTypes?: string
}
/**
@@ -14,7 +16,7 @@ interface Props {
* 复用 instrumentSearch 后端(代码 / 名称模糊匹配),单选即跳转该股财务详情。
* 模式对齐 Watchlist.StockSearchBox:useQuery + 外部点击关闭 + 键盘导航。
*/
-export function StockFinancialSearch({ onSelect }: Props) {
+export function StockFinancialSearch({ onSelect, assetTypes }: Props) {
const [query, setQuery] = useState('')
const [open, setOpen] = useState(false)
const [activeIdx, setActiveIdx] = useState(-1)
@@ -22,8 +24,8 @@ export function StockFinancialSearch({ onSelect }: Props) {
const inputRef = useRef(null)
const search = useQuery({
- queryKey: QK.instrumentSearch(query),
- queryFn: () => api.instrumentSearch(query),
+ queryKey: QK.instrumentSearch(query, assetTypes ?? 'stock'),
+ queryFn: () => api.instrumentSearch(query, 20, assetTypes),
enabled: query.trim().length > 0,
staleTime: 30_000,
})
@@ -116,6 +118,9 @@ export function StockFinancialSearch({ onSelect }: Props) {
>
{r.symbol}
{r.name}
+ {r.asset_type === 'index' && (
+ 指数
+ )}
{r.code && {r.code}}
))
diff --git a/frontend/src/components/monitor/RuleEditor.tsx b/frontend/src/components/monitor/RuleEditor.tsx
index 12f8b72..4878885 100644
--- a/frontend/src/components/monitor/RuleEditor.tsx
+++ b/frontend/src/components/monitor/RuleEditor.tsx
@@ -21,7 +21,7 @@ interface Props {
}
const TYPE_DEFAULT_NAME: Record = {
- signal: '个股信号监控', price: '价格监控', market: '市场异动监控', strategy: '策略监控',
+ signal: '信号监控', price: '价格监控', market: '市场异动监控', strategy: '策略监控',
}
const TYPE_ICONS = {
@@ -94,8 +94,8 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
const [symbolQuery, setSymbolQuery] = useState('')
const [strategyQuery, setStrategyQuery] = useState('')
const [strategyCategory, setStrategyCategory] = useState<'all' | 'builtin' | 'custom' | 'ai'>('all')
- // ETF 规则时标的搜索一并搜出 ETF。
- const symbolAssetTypes = assetType === 'etf' ? 'stock,etf' : 'stock'
+ // 标的搜索资产类型: ETF 一并搜股票; 指数只搜指数; 否则只搜股票。
+ const symbolAssetTypes = assetType === 'etf' ? 'stock,etf' : assetType === 'index' ? 'index' : 'stock'
const symbolSearch = useQuery({
queryKey: QK.instrumentSearch(symbolQuery, symbolAssetTypes),
queryFn: () => api.instrumentSearch(symbolQuery, 20, symbolAssetTypes),
@@ -124,7 +124,7 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
if (c.op !== 'truth' && (c.value === null || c.value === undefined)) throw new Error('阈值条件需要数值')
}
}
- if (d.scope === 'symbols' && d.symbols.length === 0) throw new Error('请选择至少一只股票')
+ if (d.scope === 'symbols' && d.symbols.length === 0) throw new Error('请选择至少一只标的')
return api.monitorRuleSave(d)
},
onSuccess: () => {
@@ -183,6 +183,20 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
...SIGNAL_OPTIONS.map(key => ({ key, label: cnSignal(key) })),
...(options.data?.builtin_signals ?? []).filter(option => MONITOR_INTRADAY_SIGNAL_OPTIONS.includes(option.key)),
]
+ // 指数: 隐藏涨跌停/连板类 (指数无这些列) 与分时信号 (无本地分钟K, 会静默不触发)
+ const INDEX_HIDDEN_SIGNALS = (key: string) =>
+ key.includes('limit') || MONITOR_INTRADAY_SIGNAL_OPTIONS.includes(key)
+ const pickerSignals = assetType === 'index'
+ ? monitorBuiltinSignals.filter(o => !INDEX_HIDDEN_SIGNALS(o.key))
+ : monitorBuiltinSignals
+ // 指数: 监控类型仅 signal/price (无涨跌停/策略/封单语义)
+ const visibleTypes = (options.data?.types ?? []).filter(
+ t => assetType !== 'index' || t.key === 'signal' || t.key === 'price',
+ )
+ // 指数: 作用范围仅 symbols (无全市场/板块语义)
+ const visibleScopes = (options.data?.scopes ?? []).filter(
+ s => assetType !== 'index' || s.key === 'symbols',
+ )
const thresholdConds = draft.conditions.filter(c => c.op !== 'truth')
const strategyPresets = strategies.data?.presets ?? []
const normalizedStrategyQuery = strategyQuery.trim().toLowerCase()
@@ -236,7 +250,7 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
signals={selectedSignals}
onChange={onSignalPickerChange}
kind="entry"
- builtinSignals={monitorBuiltinSignals}
+ builtinSignals={pickerSignals}
disabledSignals={intradaySupport?.available === false ? MONITOR_INTRADAY_SIGNAL_OPTIONS : []}
disabledSignalHint={intradaySupport?.reason}
/>
@@ -311,26 +325,33 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
- {/* 资产类型: 股票 / ETF (个股极简模式不显示) */}
+ {/* 资产类型: 股票 / ETF / 指数 (个股极简模式不显示) */}
{!simple && (
资产类型
- {(['stock', 'etf'] as const).map(t => (
+ {(['stock', 'etf', 'index'] as const).map(t => (
))}
@@ -341,7 +362,7 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
监控类型
- {(options.data?.types ?? []).map(t => {
+ {visibleTypes.map(t => {
const Icon = TYPE_ICONS[t.key as keyof typeof TYPE_ICONS] ?? Activity
const active = draft.type === t.key
return (
@@ -384,7 +405,7 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
作用范围
{draft.scope === 'symbols' && (
@@ -400,7 +421,7 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
setSymbolQuery(e.target.value)}
- placeholder="搜索股票..."
+ placeholder="搜索代码或名称..."
className="h-7 w-32 rounded border border-border bg-base pl-6 pr-2 text-[11px] text-foreground focus:outline-none focus:border-accent/50"
/>
@@ -417,7 +438,7 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
)}
- {draft.scope === 'all' &&
对全市场所有股票生效}
+ {draft.scope === 'all' &&
对全市场所有标的生效}
{draft.scope === 'sector' &&
板块精确过滤(开发中,当前等同全市场)}
@@ -447,7 +468,7 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
signals={selectedSignals}
onChange={onSignalPickerChange}
kind="entry"
- builtinSignals={monitorBuiltinSignals}
+ builtinSignals={pickerSignals}
disabledSignals={intradaySupport?.available === false ? MONITOR_INTRADAY_SIGNAL_OPTIONS : []}
disabledSignalHint={intradaySupport?.reason}
/>
@@ -455,7 +476,7 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
{intradaySupport?.available === false
? intradaySupport.reason
- : `分时穿越按已完成的一分钟判断,仅支持指定股票,当前最多监听 ${intradaySupport?.max_symbols ?? 0} 只。`}
+ : `分时穿越按已完成的一分钟判断,仅支持指定标的,当前最多监听 ${intradaySupport?.max_symbols ?? 0} 只。`}
)}
diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts
index 34cd433..9e3a08a 100644
--- a/frontend/src/lib/api.ts
+++ b/frontend/src/lib/api.ts
@@ -522,7 +522,7 @@ export interface MonitorRule {
name: string
enabled: boolean
type: 'strategy' | 'signal' | 'price' | 'market' | 'ladder'
- asset_type?: 'stock' | 'etf'
+ asset_type?: 'stock' | 'etf' | 'index'
scope: 'symbols' | 'all' | 'sector'
symbols: string[]
sector?: string | null
@@ -1302,6 +1302,7 @@ export const api = {
date: string | null
rows: MinuteKlineRow[]
source?: 'local' | 'live' | 'none'
+ asset_type?: 'stock' | 'etf' | 'index'
price_limit?: PriceLimitInfo | null
}>(
`/api/kline/minute?symbol=${encodeURIComponent(symbol)}${date ? `&date=${date}` : ''}`,
@@ -1419,7 +1420,7 @@ export const api = {
: '/api/watchlist/enriched',
),
- screenerStrategies: async (assetType?: 'stock' | 'etf') => {
+ screenerStrategies: async (assetType?: 'stock' | 'etf' | 'index') => {
const data = await request<{ strategies: StrategyDetail[]; load_errors?: StrategyLoadError[] }>(
`/api/strategies?${assetType ? `asset_type=${assetType}&` : ''}timeframe=1d`,
)
@@ -1483,7 +1484,7 @@ export const api = {
stop_loss_pct?: number
max_hold_days?: number
matching?: 'close_t' | 'open_t+1'
- asset_type?: 'stock' | 'etf'
+ asset_type?: 'stock' | 'etf' | 'index'
}) =>
request('/api/backtest/run', {
method: 'POST',
@@ -1503,7 +1504,7 @@ export const api = {
weight?: 'equal' | 'factor_weight'
fees_pct?: number
slippage_bps?: number
- asset_type?: 'stock' | 'etf'
+ asset_type?: 'stock' | 'etf' | 'index'
}) =>
request('/api/backtest/factor/run', {
method: 'POST',
@@ -1527,7 +1528,7 @@ export const api = {
max_positions?: number
initial_capital?: number
position_sizing?: 'equal' | 'score_weight'
- asset_type?: 'stock' | 'etf'
+ asset_type?: 'stock' | 'etf' | 'index'
minute_fill?: boolean
}) =>
request('/api/backtest/strategy/run', {
diff --git a/frontend/src/pages/Monitor.tsx b/frontend/src/pages/Monitor.tsx
index 7e79c71..36a2573 100644
--- a/frontend/src/pages/Monitor.tsx
+++ b/frontend/src/pages/Monitor.tsx
@@ -19,7 +19,7 @@ import { DimensionMembersDialog, type DimensionKind, type DimensionMembersTarget
import { usePreferences } from '@/lib/useSharedQueries'
const TYPE_LABEL: Record = {
- signal: '个股信号', price: '价格/涨跌', market: '市场异动', strategy: '策略监控',
+ signal: '信号', price: '价格/涨跌', market: '市场异动', strategy: '策略监控',
}
/** 严重级别 → 左侧色条 + 图标 */
@@ -337,7 +337,7 @@ function AlertsList({ alertsQuery, confirmClear, setConfirmClear, total, enterTs
) : (
@@ -643,11 +643,11 @@ function RulesList({ rulesQuery, onEdit }: {
) : (
rules.map(r => {
- // 名称截取: "策略监控 · MACD金叉" → "MACD金叉", "个股信号监控 · 300750.SZ" → "个股信号监控"
+ // 名称截取: "策略监控 · MACD金叉" → "MACD金叉", "信号监控 · 300750.SZ" → "信号监控"
const dotIdx = r.name.indexOf(' · ')
const displayName = dotIdx >= 0 ? r.name.slice(dotIdx + 3) : r.name
return (
@@ -672,6 +672,9 @@ function RulesList({ rulesQuery, onEdit }: {
{TYPE_LABEL[r.type]}
+ {r.asset_type === 'index' && (
+
指数
+ )}
{/* 个股类型: 直接显示可点击的代码+名称; 其他类型显示规则名 */}
{r.scope === 'symbols' && r.symbols.length > 0 ? (