diff --git a/backend/app/api/kline.py b/backend/app/api/kline.py index 2c0a545..c85547c 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]] = {} diff --git a/backend/app/api/watchlist.py b/backend/app/api/watchlist.py index 004c3ef..a75e206 100644 --- a/backend/app/api/watchlist.py +++ b/backend/app/api/watchlist.py @@ -180,8 +180,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() @@ -210,8 +212,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} @@ -225,8 +238,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 1dbbed8..cf2ae35 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,16 @@ 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") + # ---- 指数: 仅有指数监控规则时才 flush 焐热 (无规则零成本) ---- + 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(): + try: + self._repo.flush_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") # ---- 通知 SSE ---- self._broadcast_quote_updated() @@ -725,6 +743,14 @@ class QuoteService: from app.tickflow.client import get_paid_realtime_client symbols = preferences.get_realtime_watchlist_symbols() + # 指数监控规则标的并入轮询 (独立于股票前5名额) + 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 @@ -776,22 +802,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 +830,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 +853,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 用)。""" @@ -1018,6 +1083,13 @@ 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 @@ -1044,6 +1116,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 +1465,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 +1483,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 6b3a359..30f129e 100644 --- a/backend/app/strategy/monitor_rules.py +++ b/backend/app/strategy/monitor_rules.py @@ -106,6 +106,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"): diff --git a/backend/app/tickflow/repository.py b/backend/app/tickflow/repository.py index 267e925..4c104df 100644 --- a/backend/app/tickflow/repository.py +++ b/backend/app/tickflow/repository.py @@ -316,6 +316,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 关键路径, @@ -383,6 +385,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 推后台线程预热") @@ -473,6 +478,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 最新日到内存 + 构建聚合表。 @@ -879,6 +886,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: @@ -952,6 +1003,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: @@ -1130,13 +1187,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: @@ -1924,7 +1981,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 @@ -1941,6 +1998,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] @@ -2006,6 +2066,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_monitor_index.py b/backend/tests/test_monitor_index.py new file mode 100644 index 0000000..fa0db25 --- /dev/null +++ b/backend/tests/test_monitor_index.py @@ -0,0 +1,72 @@ +"""指数监控规则校验测试。""" +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() == {} # 策略结果缓存未被触碰 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_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