fix(indicators): 盘中递推窗口按交易日计数并核对 K 线根数, 次新股与长假前后不再用残缺窗口出值

_build_live_agg 按自然日 90 天切历史, 再用 tail(N) 取 MA/BOLL 部分和、60 日极值与
N 日前收盘, 从不核对窗口里实际有几根 K 线:

- 春节/国庆长假前后 90 个自然日只有 57~59 个交易日, 全市场盘中 MA60、60 日动量
  用残缺窗口算出, 与盘后全量不一致;
- 次新股历史不足 N 根时 tail(N) 返回全部已有 K 线, 盘中 MA60 = (部分和 + 今收) / 60
  等残缺值照样出值 (BOLL 下轨甚至为负), 盘后全量 rolling/shift 窗口不满为空。

修复:
- 窗口起点取「自然日 90 天」与「最近 60 个交易日」较早者; 缓存路径用缓存日期计数,
  降级路径多读 60 个自然日后计数。平常窗口不变 (不缩小)。
- _window_len 改为窗口内实际 K 线根数 (原先算了但无人使用), compute_enriched_today
  按各指标所需历史根数置空, 与全量 rolling(窗口满才出值) / shift(N) 同口径。

Closes #333

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

backend/app/indicators/pipeline.py

@@ -2097,6 +2097,24 @@ def compute_enriched_today(
           .alias("annual_vol_20d"),
     )
 
+    # ---- 窗口不满置空 ----
+    # live_agg 的部分和 / 极值 / N 日前收盘用 tail(N) 取, 历史 K 线不足 N 根 (次新股) 时
+    # 取到的是残缺窗口; 全量 rolling_*(N) / shift(N) 窗口不满为空, 按窗口内实际根数同口径置空。
+    min_history_bars = {
+        "ma5": 4, "ma10": 9, "ma20": 19, "ma30": 29, "ma60": 59,
+        "vol_ma5": 4, "vol_ma10": 9, "vol_ratio_5d": 5,
+        "boll_upper": 19, "boll_lower": 19, "high_60d": 59, "low_60d": 59,
+        "momentum_3d": 3, "momentum_5d": 5, "momentum_10d": 10,
+        "momentum_20d": 20, "momentum_30d": 30, "momentum_60d": 60,
+        "annual_vol_20d": 20,
+    }
+    if "_window_len" in df.columns:
+        df = df.with_columns([
+            pl.when(pl.col("_window_len") >= need).then(pl.col(column)).otherwise(None).alias(column)
+            for column, need in min_history_bars.items()
+            if column in df.columns
+        ])
+
     # ---- 信号 (需要昨天的指标值判断交叉) ----
     if not prev_enriched.is_empty():
         sig_prev = prev_enriched.select(

backend/app/tickflow/repository.py

@@ -71,6 +71,22 @@ def enriched_dirname(asset_type: str) -> str:
     return "kline_etf_enriched" if asset_type == "etf" else "kline_daily_enriched"
 
 
+# 盘中递推状态的最长窗口 (交易日): MA60 部分和 tail(59)、60 日动量 tail(60)
+_LIVE_AGG_WINDOW_BARS = 60
+
+
+def _live_agg_window_start(dates: pl.Series, latest: date, calendar_start: date) -> date:
+    """盘中递推历史窗口起点: 自然日起点与「最近 60 个交易日」起点取较早者。
+
+    自然日 90 天通常含 62~65 个交易日, 但春节/国庆长假前后只有 57~59 个,
+    tail(59)/tail(60) 会取到残缺窗口 (与 get_enriched_history 按交易日计数同理)。
+    """
+    trading = dates.filter(dates <= latest).unique().sort()
+    if trading.len() >= _LIVE_AGG_WINDOW_BARS:
+        return min(calendar_start, trading[-_LIVE_AGG_WINDOW_BARS])
+    return calendar_start
+
+
 def _last_available_rows(df: pl.DataFrame, cutoff: date) -> pl.DataFrame:
     """从已按 symbol/date 排序的数据中取每只标的最后一条有效状态。"""
     if df.is_empty():
@@ -765,8 +781,9 @@ class KlineRepository:
                 needed = [c for c in base_cols if c in hist_all.columns]
                 step = time.perf_counter()
                 logger.info("live agg step start: slice history cache")
+                window_start = _live_agg_window_start(hist_all["date"], latest, start_60d)
                 df_hist = hist_all.select(needed).filter(
-                    (pl.col("date") >= start_60d) & (pl.col("date") <= latest)
+                    (pl.col("date") >= window_start) & (pl.col("date") <= latest)
                 ).sort(["symbol", "date"])
                 logger.info("live agg step done: slice history cache rows=%d (%.2fs)", len(df_hist), time.perf_counter() - step)
 
@@ -936,7 +953,9 @@ class KlineRepository:
                 pl.col("low").tail(8).min().alias("_kdj_8d_low"),
                 pl.col("high").tail(8).max().alias("_kdj_8d_high"),
 
-                pl.col("close").tail(59).len().alias("_window_len"),
+                # 窗口内实际 K 线根数 (≤ 窗口天数): compute_enriched_today 据此把
+                # 历史不足的窗口指标置空, 与全量 rolling(窗口满才出值) 同口径
+                pl.len().alias("_window_len"),
             ])
         )
 
@@ -963,11 +982,14 @@ class KlineRepository:
 
     def _build_live_agg_from_parquet(self, latest: date, start_60d: date) -> tuple[pl.DataFrame, pl.DataFrame]:
         """降级路径: 从 parquet 读取数据并计算指标 (当 _enriched_history_cache 不可用时)。"""
+        from datetime import timedelta
+
         from app.indicators.pipeline import compute_indicators
 
+        # 多读一段自然日, 再按交易日计数确定窗口起点 (长假前后 90 个自然日不足 60 个交易日)
         lf = (
             scan_enriched_parquet(self._enriched_glob)
-            .filter(pl.col("date") >= start_60d)
+            .filter(pl.col("date") >= start_60d - timedelta(days=60))
             .filter(pl.col("date") <= latest)
             .sort(["symbol", "date"])
         )
@@ -980,6 +1002,8 @@ class KlineRepository:
 
         if df_hist.is_empty():
             return df_hist, pl.DataFrame()
+        window_start = _live_agg_window_start(df_hist["date"], latest, start_60d)
+        df_hist = df_hist.filter(pl.col("date") >= window_start)
 
         df_with_indicators = compute_indicators(df_hist)
 

backend/tests/test_live_agg_window_bars.py

@@ -0,0 +1,142 @@
+"""盘中递推状态 (live_agg) 的窗口必须真有足够的交易日 K 线。
+
+_build_live_agg 按「自然日 90 天」切历史再 tail(59)/tail(60) 取部分和、N 日前收盘:
+
+- 春节/国庆长假前后, 90 个自然日只有 57~59 个交易日, 全市场的盘中 MA60、
+  60 日动量用残缺窗口算出, 与盘后全量 (rolling_mean(60) / shift(60)) 不一致;
+- 次新股历史不足窗口长度时, tail(N) 取到的是全部已有 K 线, 盘中 MA60 等于
+  (已有收盘价之和 + 今收) / 60 这类残缺值, 而盘后全量在窗口不满时为空。
+"""
+from __future__ import annotations
+
+import math
+from datetime import date, timedelta
+
+import polars as pl
+import pytest
+
+from app.indicators.pipeline import compute_enriched_today, compute_indicators
+from app.tickflow.repository import DataStore, KlineRepository, _live_agg_window_start
+
+TODAY = date(2025, 3, 3)
+# 2025 春节休市 1/28~2/4 + 元旦: 2024-12-03 ~ 2025-02-28 只有 58 个交易日
+CLOSED = {date(2025, 1, 1)} | {date(2025, 1, 28) + timedelta(days=i) for i in range(8)}
+
+VALUE_COLUMNS = (
+    "ma5", "ma10", "ma20", "ma30", "ma60",
+    "momentum_5d", "momentum_10d", "momentum_20d", "momentum_30d", "momentum_60d",
+    "vol_ma5", "vol_ma10", "vol_ratio_5d",
+)
+# 以下列在窗口不满时全量为空; 数值口径另有独立问题, 这里只核对空与非空
+NULLNESS_COLUMNS = ("high_60d", "low_60d", "boll_upper", "boll_lower", "annual_vol_20d")
+
+
+def _history_days() -> list[date]:
+    days: list[date] = []
+    d = TODAY - timedelta(days=299)
+    while d < TODAY:
+        if d.weekday() < 5 and d not in CLOSED:
+            days.append(d)
+        d += timedelta(days=1)
+    return days
+
+
+def _bars() -> pl.DataFrame:
+    days = _history_days()
+    rows = []
+    listings = {
+        "600001.SH": days,       # 老股: 约 200 个历史交易日
+        "000002.SZ": days,
+        "001003.SZ": days[-12:],  # 次新股: 12 个历史交易日
+        "001004.SZ": days[-40:],  # 次新股: 40 个历史交易日
+    }
+    for n, (symbol, history) in enumerate(listings.items()):
+        for i, day in enumerate([*history, TODAY]):
+            close = round(10.0 + n + 0.6 * math.sin(i * 0.37 + n) + 0.004 * i, 2)
+            volume = 1000.0 + 300.0 * math.cos(i * 0.51 + n)
+            rows.append((symbol, day, close, close + 0.08, close - 0.08, close, volume))
+    return pl.DataFrame(
+        rows,
+        schema=["symbol", "date", "open", "high", "low", "close", "volume"],
+        orient="row",
+    ).with_columns(
+        (pl.col("close") * pl.col("volume")).alias("amount"),
+        pl.col("close").alias("raw_close"),
+        pl.col("high").alias("raw_high"),
+        pl.col("low").alias("raw_low"),
+    ).sort(["symbol", "date"])
+
+
+def _incremental(repo: KlineRepository, bars: pl.DataFrame, latest: date) -> pl.DataFrame:
+    history_indicators = compute_indicators(bars.filter(pl.col("date") < TODAY))
+    repo._build_live_agg(latest)
+    return compute_enriched_today(
+        repo.get_live_agg(),
+        history_indicators.filter(pl.col("date") == latest),
+        bars.filter(pl.col("date") == TODAY).drop("raw_close", "raw_high", "raw_low"),
+        None,
+    ).sort("symbol")
+
+
+def _assert_matches_full(incremental: pl.DataFrame, bars: pl.DataFrame) -> None:
+    full = compute_indicators(bars).filter(pl.col("date") == TODAY).sort("symbol")
+    assert incremental["symbol"].to_list() == full["symbol"].to_list()
+    mismatches: list[str] = []
+    for column in (*VALUE_COLUMNS, *NULLNESS_COLUMNS):
+        pairs = zip(
+            full["symbol"].to_list(), incremental[column].to_list(), full[column].to_list(),
+            strict=True,
+        )
+        for symbol, got, want in pairs:
+            if (got is None) != (want is None):
+                mismatches.append(f"{symbol} {column}: 盘中={got} 全量={want}")
+            elif column in VALUE_COLUMNS and got is not None and got != pytest.approx(want, rel=1e-9):
+                mismatches.append(f"{symbol} {column}: 盘中={got:.6f} 全量={want:.6f}")
+    assert mismatches == []
+
+
+def test_fixture_holiday_window_has_fewer_than_60_trading_days():
+    days = _history_days()
+    latest = days[-1]
+    in_90_calendar_days = [d for d in days if d >= latest - timedelta(days=90)]
+    assert len(in_90_calendar_days) < 60  # 构造前提: 长假让 90 个自然日不足 60 个交易日
+
+
+def test_window_start_counts_trading_days():
+    days = _history_days()
+    latest = days[-1]
+    calendar_start = latest - timedelta(days=90)
+    dates = pl.Series("date", days)
+
+    # 长假: 自然日起点只含 58 个交易日 → 前移到第 60 个交易日
+    assert _live_agg_window_start(dates, latest, calendar_start) == days[-60]
+    assert _live_agg_window_start(dates, latest, calendar_start) < calendar_start
+    # 平常: 自然日起点已覆盖 60 个交易日 → 保持自然日起点 (窗口不缩小)
+    assert _live_agg_window_start(dates, latest, days[-65]) == days[-65]
+    # 本地不足 60 个交易日 → 退回自然日起点
+    assert _live_agg_window_start(dates.tail(30), latest, calendar_start) == calendar_start
+
+
+def test_live_agg_from_history_cache_matches_full(tmp_path):
+    bars = _bars()
+    history = bars.filter(pl.col("date") < TODAY)
+    repo = KlineRepository(DataStore(tmp_path))
+    repo._enriched_history_cache = compute_indicators(history)
+
+    incremental = _incremental(repo, bars, history["date"].max())
+
+    _assert_matches_full(incremental, bars)
+
+
+def test_live_agg_from_parquet_fallback_matches_full(tmp_path):
+    bars = _bars()
+    history = bars.filter(pl.col("date") < TODAY)
+    target = tmp_path / "kline_daily_enriched" / "all"
+    target.mkdir(parents=True)
+    history.write_parquet(target / "part.parquet")
+    repo = KlineRepository(DataStore(tmp_path))
+    assert repo._enriched_history_cache is None  # 走 _build_live_agg_from_parquet 降级路径
+
+    incremental = _incremental(repo, bars, history["date"].max())
+
+    _assert_matches_full(incremental, bars)