diff --git a/backend/app/indicators/pipeline.py b/backend/app/indicators/pipeline.py index dab7777..24ade27 100644 --- a/backend/app/indicators/pipeline.py +++ b/backend/app/indicators/pipeline.py @@ -36,12 +36,16 @@ logger = logging.getLogger(__name__) # ── 自定义信号缓存 ───────────────────────────────────── # 从 data/user_data/custom_signals/*.json 加载并编译为 Polars 表达式。 +# 两套表达式分别用于全量路径 (allow_shift=True, 支持日期偏移条件) +# 和盘中增量热路径 (allow_shift=False, 跳过偏移条件)。 # 模块级缓存:首次调用时加载,invalidate_custom_signals() 后下次重载。 +# 增量路径每秒级执行, 若不缓存则每轮 glob + 读所有 JSON + 重编译表达式。 _custom_signal_exprs: dict[str, pl.Expr] | None = None +_custom_signal_exprs_today: dict[str, pl.Expr] | None = None def _get_custom_signal_exprs() -> dict[str, pl.Expr]: - """懒加载自定义信号表达式(带模块级缓存)。""" + """懒加载自定义信号表达式(带模块级缓存,allow_shift=True)。""" global _custom_signal_exprs if _custom_signal_exprs is None: from app.strategy import custom_signals @@ -54,10 +58,29 @@ def _get_custom_signal_exprs() -> dict[str, pl.Expr]: return _custom_signal_exprs +def _get_custom_signal_exprs_today() -> dict[str, pl.Expr]: + """盘中增量热路径专用 (allow_shift=False, 跳过日期偏移条件)。 + + 与全量版分开缓存:盘中单日快照上 .shift 跨 symbol 语义不正确, + build_expressions(allow_shift=False) 会跳过带偏移的信号, 结果集不同。 + """ + global _custom_signal_exprs_today + if _custom_signal_exprs_today is None: + from app.strategy import custom_signals + try: + sigs = custom_signals.load_all(settings.data_dir) + _custom_signal_exprs_today = custom_signals.build_expressions(sigs, allow_shift=False) + except Exception as e: + logger.warning("custom signals load failed (today): %s", e) + _custom_signal_exprs_today = {} + return _custom_signal_exprs_today + + def invalidate_custom_signals() -> None: """失效自定义信号缓存(保存/删除信号后调用,下次计算重新加载)。""" - global _custom_signal_exprs + global _custom_signal_exprs, _custom_signal_exprs_today _custom_signal_exprs = None + _custom_signal_exprs_today = None # enriched parquet 仅存储的列 (14 列) @@ -1652,14 +1675,10 @@ def compute_enriched_today( df = df.drop([c for c in drop_cols if c in df.columns]) # 自定义信号(日级实时路径同样注入, 但不支持日期偏移条件 → allow_shift=False) + # 复用模块级缓存 _custom_signal_exprs_today: 增量热路径每秒级执行, + # 不缓存则每轮 glob + 读所有 JSON + 重编译表达式。失效由 invalidate_custom_signals 统一管理。 from app.strategy import custom_signals - try: - sigs = custom_signals.load_all(settings.data_dir) - today_exprs = custom_signals.build_expressions(sigs, allow_shift=False) - except Exception as e: - logger.warning("custom signals load failed (today): %s", e) - today_exprs = {} - df = custom_signals.inject(df, today_exprs) + df = custom_signals.inject(df, _get_custom_signal_exprs_today()) # 清理 NaN / Inf float_cols = [c for c in df.columns if df[c].dtype.is_float()] diff --git a/backend/app/tickflow/repository.py b/backend/app/tickflow/repository.py index c2ca060..62ede68 100644 --- a/backend/app/tickflow/repository.py +++ b/backend/app/tickflow/repository.py @@ -1327,6 +1327,23 @@ class KlineRepository: """单股日K查询 — 从14列parquet读取后即时计算指标。""" from datetime import timedelta + # 快路径: 请求的列全是 parquet 直接存储的列 (如迷你蜡烛图只要 OHLCV) → + # scan + 列下推直接返回, 跳过 warmup(150天) 与 _compute_enriched_range 全套指标计算。 + # 仍用 enriched_latest 缓存覆盖最新日 (盘中更准), 只保留请求列。 + # 试探 scan 仅读请求的列; 缺列时回退到下方完整计算路径 (代价仅一次轻量 scan)。 + if columns: + df = self._scan_daily_symbol(symbol, start, end, columns) + if not df.is_empty() and all(c in df.columns for c in columns): + cached, cache_date = self.get_enriched_latest() + if cached is not None and not cached.is_empty() and cache_date: + if start <= cache_date <= end: + cached_part = self._filter_cached(cached, symbol, columns) + if not cached_part.is_empty(): + df = df.filter(pl.col("date") != cache_date) + common_cols = [c for c in df.columns if c in cached_part.columns] + df = pl.concat([df.select(common_cols), cached_part.select(common_cols)]) + return df + # 扩展范围用于指标预热 (MA60 需要 ~60 交易日 ≈ 120 日历日) warmup_start = start - timedelta(days=150) @@ -1381,6 +1398,14 @@ class KlineRepository: """指数日K查询 — 从独立指数 enriched parquet 读取后即时计算通用指标。""" from datetime import timedelta + # 快路径: 若请求的列全部是 parquet 直接存储的列 (如迷你蜡烛图只要 OHLCV), + # 直接 scan + 列下推返回, 跳过 warmup(150天) 与 _compute_index_enriched_range 全套指标计算。 + # 试探 scan 仅读请求的列, 缺列时回退到下方完整计算路径 (代价仅一次轻量 scan)。 + if columns: + df = self._scan_index_daily_symbol(symbol, start, end, columns) + if not df.is_empty() and all(c in df.columns for c in columns): + return df + warmup_start = start - timedelta(days=150) df = self._scan_index_daily_symbol(symbol, warmup_start, end, None) if not df.is_empty(): @@ -1401,6 +1426,13 @@ class KlineRepository: """ETF 日K查询 — 优先读独立 ETF enriched,兼容旧版 index enriched 中的 ETF。""" from datetime import timedelta + # 快路径: 独立 ETF 数据 + 请求列全是 parquet 存储列 → 直接 scan 列下推, 跳过 warmup + compute。 + # 无独立 ETF 数据 (旧版存入 index enriched) 时 df 为空, 自然回退到下方完整路径。 + if columns: + df = self._scan_etf_daily_symbol(symbol, start, end, columns) + if not df.is_empty() and all(c in df.columns for c in columns): + return df + warmup_start = start - timedelta(days=150) df = self._scan_etf_daily_symbol(symbol, warmup_start, end, None) if df.is_empty():