perf(indicators,kline): 热路径信号缓存 + 日K快路径跳过全量指标

三处性能优化 (后端内部, 不改 API 契约):

1. 增量热路径自定义信号缓存 (pipeline.py)
   compute_enriched_today 每秒级执行, 原每轮 glob + 读所有 JSON + 重编译
   表达式。新增 _custom_signal_exprs_today 模块级缓存 (allow_shift=False),
   与全量版分开缓存, invalidate_custom_signals 统一失效。

2. ETF/指数迷你蜡烛图跳过全量指标 (repository.py)
   get_index_daily/get_etf_daily 增加快路径: 请求列全是 parquet 存储列时
   scan + 列下推直接返回, 跳过 150 天 warmup + 全套指标计算。缺列回退完整路径。

3. 单股日K跳过全量指标 (repository.py)
   get_daily 同样加快路径并保留缓存覆盖最新日。采用最保守的列下推方案,
   不引入 needed 裁剪, 避免依赖闭包漏算金融数据的风险。

验证: pytest 全套 582 passed; Ruff 无新增告警; 快路径功能验证通过。

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()]

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():