fix(pipeline): 避免停牌分区反复误删重算

backend/app/jobs/daily_pipeline.py

@@ -20,9 +20,10 @@ from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from apscheduler.triggers.cron import CronTrigger
 from apscheduler.triggers.interval import IntervalTrigger
 
-from app.indicators.pipeline import run_pipeline
 from app.config import settings
-from app.services import index_sync, instrument_sync, kline_sync, preferences as _prefs
+from app.indicators.pipeline import filter_halt_days, run_pipeline
+from app.services import index_sync, instrument_sync, kline_sync
+from app.services import preferences as _prefs
 from app.tickflow.capabilities import Cap, CapabilitySet
 from app.tickflow.pools import DEMO_SYMBOLS, get_pool
 from app.tickflow.repository import KlineRepository
@@ -37,31 +38,35 @@ def _prune_partial_enriched_partitions(daily_dir: Path, enriched_dir: Path) -> l
 
     自选实时路径会在全市场 enriched 生成前提前创建当日分区 (只有几只自选),
     仅按日期目录计数比较会把它误判为完整分区而跳过计算, 造成日K缺失与
-    均线错误。同日 enriched 行数 < daily 行数即判定为部分分区: 删除后
-    run_pipeline(new_dates_only=True) 会把它们当"新日期"全市场补齐, 与
-    data_integrity.prune_enriched_partitions 的修复语义一致。
+    均线错误。按与加工相同的停牌过滤口径检查 symbol 覆盖, 不能直接比较
+    行数: 正常剔除停牌记录会让 enriched 少行, 导致每次管道都删除重算。
+    删除后 run_pipeline(new_dates_only=True) 会把它们当"新日期"全市场补齐。
     daily 同日分区不存在 (今日日K尚未同步) 时不处理, 留给当日正常流程。
     """
     import shutil
 
-    import pyarrow.parquet as pq
-
-    def _rows(part_dir: Path) -> int:
-        total = 0
-        for f in part_dir.glob("*.parquet"):
-            try:
-                total += pq.ParquetFile(f).metadata.num_rows
-            except Exception:  # noqa: BLE001
-                return -1  # 不可读 → 不动, 交给既有完整性检查兜底
-        return total
-
     pruned: list[str] = []
     for part in enriched_dir.glob("date=*"):
         daily_part = daily_dir / part.stem
         if not daily_part.exists():
             continue
-        e_rows, d_rows = _rows(part), _rows(daily_part)
-        if e_rows >= 0 and d_rows > 0 and e_rows < d_rows:
+        try:
+            expected: set[str] = set()
+            # 每次只读单文件的停牌判定列, 不加载全历史或指标宽表。
+            for path in daily_part.glob("*.parquet"):
+                schema = pl.read_parquet_schema(path)
+                if not {"symbol", "open", "high"}.issubset(schema):
+                    raise ValueError("daily 缺少 symbol/open/high, 无法判断有效标的覆盖")
+                columns = [c for c in ("symbol", "open", "high", "volume", "amount") if c in schema]
+                daily = pl.read_parquet(path, columns=columns)
+                expected.update(filter_halt_days(daily)["symbol"].drop_nulls().to_list())
+            actual: set[str] = set()
+            for path in part.glob("*.parquet"):
+                actual.update(pl.read_parquet(path, columns=["symbol"])["symbol"].drop_nulls().to_list())
+        except Exception as e:
+            logger.warning("enriched 覆盖检查跳过 %s, 保留分区: %s", part.name, e)
+            continue
+        if expected - actual:
             shutil.rmtree(part, ignore_errors=True)
             pruned.append(part.stem.split("=")[1])
     return pruned
@@ -421,7 +426,7 @@ def run_now(
     #     - 首次 (enriched 目录不存在) → 全量
     #     - 往前扩展历史 (新日期 < enriched 已有最早日期) → 全量
     #       前面的除权因子会改变累积因子链,影响后面所有日期的复权价格
-    #     - 往后新增日期 (新日期 > enriched 已有最晚日期)
+    #     - 往后新增日期或已有历史区间内的缺口
     #       → 增量补新区块(所有标的) + 受除权影响个股全日期重算
     #     - 无新日期 + 有新除权因子 → 增量: 只重算受影响个股的全部日期
     #     - 无新日期 + 无变化 → 跳过
@@ -456,15 +461,14 @@ def run_now(
         daily_dates = sorted(d.stem.split("=")[1] for d in daily_dir.glob("date=*"))
         enriched_dates = sorted(d.stem.split("=")[1] for d in enriched_dir.glob("date=*"))
         earliest_enriched = enriched_dates[0]
-        latest_enriched = enriched_dates[-1]
         new_dates = set(daily_dates) - set(enriched_dates)
         if new_dates:
             # 有新日期早于 enriched 最早日期 → 往前扩展
             if any(d < earliest_enriched for d in new_dates):
                 backward_extension = True
-            # 有新日期晚于 enriched 最晚日期 → 往后新增
-            if any(d > latest_enriched for d in new_dates):
-                forward_incremental = True
+            # 包含中间被删的异常分区; 没有新增末日也必须补算。
+            # 往前扩展仍由下方优先走全量分支。
+            forward_incremental = True
 
     def _enriched_batch_progress(cur: int, tot: int) -> None:
         emit("compute_enriched", 65 + int(23 * cur / tot),

backend/tests/test_enriched_partial_partition.py

@@ -3,23 +3,35 @@
 自选实时路径 (merge_live_enriched_asset) 会在全市场 enriched 生成前提前
 创建当日分区 (只有几只自选); 仅按日期目录计数比较会把它误判为完整分区
 而跳过计算, 造成日K连续缺失与均线错误。_prune_partial_enriched_partitions
-按同日 daily/enriched 行数比较, 发现部分分区即删除, 让增量重算按"新日期"
-全市场补齐。
+按过滤停牌后的标的集合检查覆盖, 避免正确过滤的停牌记录导致反复删除重算。
 """
 from __future__ import annotations
 
 from datetime import date
 from pathlib import Path
+from types import SimpleNamespace
 
 import polars as pl
+import pytest
 
+from app.indicators import pipeline
+from app.jobs import daily_pipeline
 from app.jobs.daily_pipeline import _prune_partial_enriched_partitions
+from app.services import data_integrity, preferences
+from app.tickflow.capabilities import CapabilitySet
 
 
 def _write_partition(base: Path, day: str, symbols: list[str]) -> None:
     part = base / f"date={day}"
     part.mkdir(parents=True, exist_ok=True)
-    pl.DataFrame({"symbol": symbols, "close": [1.0] * len(symbols)}).write_parquet(
+    pl.DataFrame({
+        "symbol": symbols,
+        "open": [1.0] * len(symbols),
+        "high": [1.0] * len(symbols),
+        "close": [1.0] * len(symbols),
+        "volume": [100.0] * len(symbols),
+        "amount": [100.0] * len(symbols),
+    }).write_parquet(
         part / "part.parquet"
     )
 
@@ -56,7 +68,151 @@ def test_enriched_date_without_daily_is_left_alone(tmp_path) -> None:
     # 今日日K尚未同步时, 实时创建的当日分区留给当日正常流程处理
     daily = tmp_path / "kline_daily"
     enriched = tmp_path / "kline_daily_enriched"
-    _write_partition(enriched, str(date.today()), ["only_watchlist"])
+    day = "2026-09-07"
+    _write_partition(enriched, day, ["only_watchlist"])
 
     assert _prune_partial_enriched_partitions(daily, enriched) == []
-    assert (enriched / f"date={date.today()}").exists()
+    assert (enriched / f"date={day}").exists()
+
+
+@pytest.mark.parametrize("legacy_halt", [False, True])
+def test_computed_partition_with_halt_is_preserved_on_repeated_checks(tmp_path, monkeypatch, legacy_halt):
+    monkeypatch.setattr(pipeline, "_custom_signal_exprs", {})
+    daily = tmp_path / "kline_daily"
+    enriched = tmp_path / "kline_daily_enriched"
+    day = "2026-09-07"
+    _write_partition(daily, day, ["600001.SH", "600002.SH"])
+    raw_path = daily / f"date={day}" / "part.parquet"
+    raw = pl.read_parquet(raw_path).with_columns(
+        pl.lit(date.fromisoformat(day)).alias("date"),
+        pl.lit(1.0).alias("low"),
+        *[
+            pl.when(pl.col("symbol") == "600002.SH").then(0.0).otherwise(pl.col(c)).alias(c)
+            for c in (["volume", "amount"] if legacy_halt else ["open", "high", "volume", "amount"])
+        ],
+    )
+    raw.write_parquet(raw_path)
+    computed = pipeline._select_storage_cols(pipeline.compute_enriched(raw))
+    assert computed["symbol"].to_list() == ["600001.SH"]
+    target = enriched / f"date={day}" / "part.parquet"
+    target.parent.mkdir(parents=True)
+    computed.write_parquet(target)
+    original = target.read_bytes()
+
+    for _ in range(2):
+        assert _prune_partial_enriched_partitions(daily, enriched) == []
+        assert target.read_bytes() == original
+
+
+@pytest.mark.parametrize("actual", [["a"], ["a", "a"], ["a", "extra"], ["a", "extra", "other"]])
+def test_missing_active_symbol_is_not_hidden_by_counts(tmp_path, actual):
+    daily = tmp_path / "daily"
+    enriched = tmp_path / "enriched"
+    _write_partition(daily, "2026-09-07", ["a", "b"])
+    _write_partition(enriched, "2026-09-07", actual)
+
+    assert _prune_partial_enriched_partitions(daily, enriched) == ["2026-09-07"]
+
+
+def test_duplicate_daily_rows_do_not_require_duplicate_enriched_rows(tmp_path):
+    daily = tmp_path / "daily"
+    enriched = tmp_path / "enriched"
+    _write_partition(daily, "2026-09-07", ["a", "a", "b"])
+    _write_partition(enriched, "2026-09-07", ["a", "b"])
+
+    assert _prune_partial_enriched_partitions(daily, enriched) == []
+
+
+@pytest.mark.parametrize("missing", ["symbol", "open", "high"])
+def test_missing_daily_required_column_does_not_delete(tmp_path, missing):
+    daily = tmp_path / "daily"
+    enriched = tmp_path / "enriched"
+    _write_partition(daily, "2026-09-07", ["a", "b"])
+    _write_partition(enriched, "2026-09-07", ["a"])
+    raw_path = daily / "date=2026-09-07" / "part.parquet"
+    pl.read_parquet(raw_path).drop(missing).write_parquet(raw_path)
+
+    assert _prune_partial_enriched_partitions(daily, enriched) == []
+    assert (enriched / "date=2026-09-07" / "part.parquet").exists()
+
+
+@pytest.mark.parametrize("broken_kind", ["daily", "enriched"])
+def test_unreadable_partition_does_not_delete(tmp_path, broken_kind):
+    daily = tmp_path / "daily"
+    enriched = tmp_path / "enriched"
+    _write_partition(daily, "2026-09-07", ["a", "b"])
+    _write_partition(enriched, "2026-09-07", ["a"])
+    (tmp_path / broken_kind / "date=2026-09-07" / "broken.parquet").write_bytes(b"broken")
+
+    assert _prune_partial_enriched_partitions(daily, enriched) == []
+    assert (enriched / "date=2026-09-07" / "part.parquet").exists()
+
+
+def test_partition_files_are_checked_together(tmp_path):
+    daily = tmp_path / "daily"
+    enriched = tmp_path / "enriched"
+    _write_partition(daily, "2026-09-07", ["a", "b"])
+    _write_partition(enriched, "2026-09-07", ["a"])
+    target = enriched / "date=2026-09-07" / "second.parquet"
+    pl.DataFrame({"symbol": ["b"]}).write_parquet(target)
+
+    assert _prune_partial_enriched_partitions(daily, enriched) == []
+
+
+@pytest.mark.parametrize("missing", ["volume", "amount"])
+def test_legacy_daily_without_optional_halt_column(tmp_path, missing):
+    daily = tmp_path / "daily"
+    enriched = tmp_path / "enriched"
+    _write_partition(daily, "2026-09-07", ["a", "b"])
+    _write_partition(enriched, "2026-09-07", ["a", "b"])
+    path = daily / "date=2026-09-07" / "part.parquet"
+    pl.read_parquet(path).drop(missing).write_parquet(path)
+    assert _prune_partial_enriched_partitions(daily, enriched) == []
+
+
+def test_all_halted_partition_does_not_require_active_rows(tmp_path):
+    daily = tmp_path / "daily"
+    enriched = tmp_path / "enriched"
+    _write_partition(daily, "2026-09-07", ["a", "b"])
+    _write_partition(enriched, "2026-09-07", ["a"])
+    path = daily / "date=2026-09-07" / "part.parquet"
+    pl.read_parquet(path).with_columns(pl.lit(0.0).alias("open"), pl.lit(0.0).alias("high")).write_parquet(path)
+    assert _prune_partial_enriched_partitions(daily, enriched) == []
+
+
+def test_pruned_interior_date_is_rebuilt_without_new_daily_or_factors(tmp_path, monkeypatch):
+    daily = tmp_path / "kline_daily"
+    enriched = tmp_path / "kline_daily_enriched"
+    dates = ["2026-09-01", "2026-09-02", "2026-09-03"]
+    for day in dates:
+        _write_partition(daily, day, ["a", "b"])
+        _write_partition(enriched, day, ["a"] if day == dates[1] else ["a", "b"])
+
+    monkeypatch.setattr(daily_pipeline.settings, "data_dir", tmp_path)
+    monkeypatch.setattr(preferences, "load", lambda: {
+        "pipeline_pull_a_share": False, "pipeline_regime_enabled": False,
+        "minute_sync_enabled": False, "adj_factor_provider": "tickflow",
+    })
+    monkeypatch.setattr(daily_pipeline.instrument_sync, "sync_instruments", lambda *_: 0)
+    monkeypatch.setattr(daily_pipeline, "_resolve_universe", lambda *_: [])
+    monkeypatch.setattr(daily_pipeline, "_invalidate", lambda *_: None)
+    monkeypatch.setattr(daily_pipeline, "_refresh_single_view", lambda *_: None)
+    monkeypatch.setattr(daily_pipeline, "_refresh_views", lambda *_: None)
+    monkeypatch.setattr(data_integrity, "scan_recent_integrity", lambda *_, **__: [])
+    calls = []
+
+    def rebuild(**kwargs):
+        calls.append(kwargs)
+        assert kwargs["new_dates_only"] is True
+        assert kwargs["symbols"] is None
+        _write_partition(enriched, dates[1], ["a", "b"])
+        return 2
+
+    monkeypatch.setattr(daily_pipeline, "run_pipeline", rebuild)
+    repo = SimpleNamespace(store=SimpleNamespace(data_dir=tmp_path),
+                           latest_daily_date=lambda: date.fromisoformat(dates[-1]))
+    daily_pipeline.run_now(repo, CapabilitySet(set()))
+    assert len(calls) == 1
+    assert pl.read_parquet(enriched / f"date={dates[1]}" / "part.parquet")["symbol"].to_list() == ["a", "b"]
+    daily_pipeline.run_now(repo, CapabilitySet(set()))
+    assert len(calls) == 1