Merge pull request #342 from kevin9327/fix/live-fallback-window-trading-days

fix(quote): ETF/指数盘中 enriched 回退路径的历史窗口按交易日计数, 长假后 MA60 等不再为空

backend/app/services/quote_service.py

@@ -1754,6 +1754,7 @@ class QuoteService:
             if not use_incremental:
                 from datetime import timedelta
                 from app.indicators.pipeline import compute_enriched
+                from app.tickflow.repository import _live_agg_window_start
 
                 logger.info("enriched 全量计算 (live_agg=%s, 上次日期=%s)",
                             "ok" if not live_agg.is_empty() else "空", prev_date)
@@ -1764,7 +1765,8 @@ class QuoteService:
                 ohlcv_cols = ["symbol", "date", "open", "high", "low", "close", "volume", "amount", "quote_ts"]
                 hist_df = guarded_collect(
                     scan_daily_parquet(daily_glob)
-                    .filter(pl.col("date") >= cutoff)
+                    # 多读一段自然日, 再按交易日计数确定窗口起点 (长假前后 90 个自然日不足 60 个交易日)
+                    .filter(pl.col("date") >= cutoff - timedelta(days=60))
                     .sort(["symbol", "date"]),
                     priority="background",
                 )
@@ -1773,6 +1775,9 @@ class QuoteService:
 
                 hist_cols = [c for c in ohlcv_cols if c in hist_df.columns]
                 hist_df = hist_df.select(hist_cols).filter(pl.col("date") != today)
+                # 与股票盘中递推窗口同口径: 至少覆盖 60 个交易日, 否则 MA60/60 日极值/60 日动量为空
+                window_start = _live_agg_window_start(hist_df["date"], today, cutoff)
+                hist_df = hist_df.filter(pl.col("date") >= window_start)
                 daily_ohlcv = daily_df.select([c for c in ohlcv_cols if c in daily_df.columns])
                 full_df = pl.concat([hist_df, daily_ohlcv], how="diagonal_relaxed")
                 full_df = full_df.sort(["symbol", "date"])

backend/tests/test_live_fallback_window_bars.py

@@ -0,0 +1,101 @@
+"""ETF / 指数盘中 enriched (全量回退路径) 的历史窗口必须真有 60 个交易日。
+
+ETF 与指数没有 live_agg, 每轮实时行情都走 _flush_live_enriched 的全量回退:
+读近「90 个自然日」的日K + 今日行情重算指标。春节/国庆长假后 90 个自然日只有
+56~59 个交易日, rolling_mean(60) / rolling_max(60) / shift(60) 窗口不满, 全部
+ETF 与指数盘中 MA60、60 日极值、60 日动量为空 (创60日新高/新低、MA20 上穿 MA60
+等信号一并失效), 而盘后缓存 (_refresh_etf_enriched / _refresh_index_enriched,
+300 个自然日) 有值。股票盘中递推窗口已按交易日计数 (_live_agg_window_start)。
+"""
+from __future__ import annotations
+
+import math
+from datetime import date, timedelta
+from pathlib import Path
+from types import SimpleNamespace
+
+import polars as pl
+import pytest
+
+import app.services.quote_service as qs_module
+from app.indicators.pipeline import compute_enriched
+from app.services.quote_service import QuoteService
+
+TODAY = date(2026, 2, 26)
+# 2026 元旦 + 春节休市 (2/16~2/23): 2025-11-28 ~ 2026-02-25 只有 57 个交易日
+CLOSED = {date(2026, 1, 1), date(2026, 1, 2)} | {date(2026, 2, 16) + timedelta(days=i) for i in range(8)}
+SYMBOLS = ("510300.SH", "159915.SZ")
+WINDOW_COLUMNS = ("ma60", "high_60d", "low_60d", "momentum_60d")
+TABLES = {"etf": "kline_etf_daily", "index": "kline_index_daily"}
+
+
+def _trading_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:
+    rows = []
+    for n, symbol in enumerate(SYMBOLS):
+        for i, day in enumerate(_trading_days()):
+            close = round(4.0 + n + 0.3 * math.sin(i * 0.23 + n) + 0.002 * i, 3)
+            rows.append((symbol, day, close, close + 0.02, close - 0.02, close, 1e6 + 1e4 * i, 4e6))
+    return pl.DataFrame(
+        rows,
+        schema=["symbol", "date", "open", "high", "low", "close", "volume", "amount"],
+        orient="row",
+    ).sort(["symbol", "date"])
+
+
+class _StubRepo:
+    def __init__(self, data_dir: Path) -> None:
+        self.store = SimpleNamespace(data_dir=data_dir)
+        self.flushed: dict[str, pl.DataFrame] = {}
+
+    def get_enriched_latest_asset(self, asset_type: str):
+        return pl.DataFrame(), None
+
+    def flush_live_enriched_asset(self, asset_type: str, df: pl.DataFrame) -> None:
+        self.flushed[asset_type] = df
+
+    def merge_live_enriched_asset(self, asset_type: str, df: pl.DataFrame) -> None:
+        self.flushed[asset_type] = df
+
+
+def test_fixture_holiday_window_has_fewer_than_60_trading_days():
+    days = _trading_days()
+    history = [d for d in days if TODAY - timedelta(days=90) <= d < TODAY]
+    assert len(history) < 60  # 构造前提: 长假让 90 个自然日不足 60 个交易日
+
+
+@pytest.mark.parametrize("asset_type", ["etf", "index"])
+def test_fallback_window_counts_trading_days(tmp_path, monkeypatch, asset_type):
+    bars = _bars()
+    table = tmp_path / TABLES[asset_type]
+    for part in bars.filter(pl.col("date") < TODAY).partition_by("date"):
+        out = table / f"date={part['date'][0].isoformat()}" / "part.parquet"
+        out.parent.mkdir(parents=True)
+        part.write_parquet(out)
+
+    monkeypatch.setattr(qs_module, "cn_today", lambda: TODAY)
+    qs = QuoteService()
+    repo = _StubRepo(tmp_path)
+    qs._repo = repo
+    qs._flush_live_enriched(
+        bars.filter(pl.col("date") == TODAY),
+        None,
+        asset_type=asset_type,
+        merge=asset_type == "index",
+    )
+
+    live = repo.flushed[asset_type].sort("symbol")
+    full = compute_enriched(bars).filter(pl.col("date") == TODAY).sort("symbol")
+    assert live["symbol"].to_list() == full["symbol"].to_list()
+    for column in WINDOW_COLUMNS:
+        assert full[column].null_count() == 0
+        assert live[column].to_list() == pytest.approx(full[column].to_list(), rel=1e-12), column