修复同步后停牌残留反复触发完整性修复

backend/app/indicators/pipeline.py

@@ -972,11 +972,16 @@ def filter_halt_days(df: pl.DataFrame) -> pl.DataFrame:
 
     停牌日的 open/high 必然为 0 (无集合竞价)。注意 close 可能被数据源
     填充为前收盘价而非 0, 因此不能用 "OHLC 全零" 判断, 否则会漏过这类
-    停牌记录 (如 *ST 撤销风险警示的停牌日), 污染 MA/ATR 等指标。
+    停牌记录 (如 *ST 撤销风险警示的停牌日), 污染 MA/ATR 等指标。旧版实时
+    落盘还会先把 open/high=0 填成 close, 对这类历史数据用零成交量和零成交额
+    作为兼容判据。
     """
     if df.is_empty() or "open" not in df.columns or "high" not in df.columns:
         return df
-    return df.filter(~((pl.col("open") == 0) & (pl.col("high") == 0)))
+    halted = (pl.col("open") == 0) & (pl.col("high") == 0)
+    if "volume" in df.columns and "amount" in df.columns:
+        halted = halted | ((pl.col("volume") == 0) & (pl.col("amount") == 0))
+    return df.filter(~halted)
 
 
 # ================================================================

backend/app/services/data_integrity.py

@@ -8,6 +8,7 @@
 - null              → batch 拉取 / 盘后计算写入的权威历史 → 完整
 - d < 今天 且 时刻 < d 15:00 → 盘中快照 (停机前实时写的) → 坏
 - d < 今天 且 时刻 ≥ d 15:00 → 尾盘定版 (close_final) → 完整
+- batch 权威行中仅夹杂少量零成交实时行 → 停牌残留 → 忽略
 - d == 今天         → 实时更新中, 属正常, 不校验
 - 分区缺失的工作日  → 缺口 (工作日近似; 节假日误报的代价是一次空范围拉取,
   merge-upsert 空写, 无害)
@@ -108,6 +109,62 @@ def _is_snapshot(day: date, quote_ts_ms: int | None) -> bool:
     return ts.date() == day and ts.time() < CLOSE_CUTOFF
 
 
+def _partition_is_snapshot(day: date, part_dir: Path, quote_ts_max_ms: int | None) -> bool:
+    """判断整个分区是否仍是盘中快照, 而非同步后遗留的停牌实时行。
+
+    batch 行用 null quote_ts 标识权威历史。实时轮询曾把停牌股票的 09:15、
+    零成交记录写入分区; 后续 batch 会过滤停牌日, merge-upsert 因而留下这些
+    孤立行。若分区已有 batch 行, 且当日收盘前的实时行全部零成交, 则它们不应
+    让整个分区反复进入修复。整分区都是实时行时仍按快照处理, 包括盘前零成交。
+    """
+    if not _is_snapshot(day, quote_ts_max_ms):
+        return False
+
+    start_ms = int(datetime.combine(day, dt_time.min, tzinfo=CN_TZ).timestamp() * 1000)
+    cutoff_ms = int(datetime.combine(day, CLOSE_CUTOFF, tzinfo=CN_TZ).timestamp() * 1000)
+    authoritative_rows = 0
+    suspicious_rows = 0
+
+    for path in sorted(part_dir.glob("*.parquet")):
+        try:
+            schema = pl.read_parquet_schema(path)
+            if "quote_ts" not in schema:
+                continue
+            columns = [
+                name for name in ("quote_ts", "volume", "amount")
+                if name in schema
+            ]
+            frame = pl.read_parquet(path, columns=columns).with_columns(
+                pl.col("quote_ts").cast(pl.Int64, strict=False),
+            )
+            authoritative_rows += frame["quote_ts"].null_count()
+            suspicious = frame.filter(
+                pl.col("quote_ts").is_between(start_ms, cutoff_ms, closed="left")
+            )
+            if suspicious.is_empty():
+                continue
+            suspicious_rows += suspicious.height
+
+            activity_columns = [
+                name for name in ("volume", "amount") if name in suspicious.columns
+            ]
+            if not activity_columns:
+                return True
+            has_activity = suspicious.select(
+                pl.any_horizontal(
+                    pl.col(name).cast(pl.Float64, strict=False).fill_null(0) > 0
+                    for name in activity_columns
+                ).any()
+            ).item()
+            if has_activity:
+                return True
+        except Exception as e:
+            logger.debug("snapshot residue scan skipped %s: %s", path, e)
+            return True
+
+    return suspicious_rows > 0 and authoritative_rows <= suspicious_rows
+
+
 def _candidate_days(today: date, lookback_days: int) -> list[date]:
     """最近 lookback_days 自然日内、严格早于今天的工作日 (节假日近似, 误报无害)。"""
     days: list[date] = []
@@ -157,7 +214,7 @@ def scan_recent_integrity(
                 continue
             part_dir = base / f"date={day.isoformat()}"
             quote_ts = _quote_ts_max_ms(part_dir)
-            if _is_snapshot(day, quote_ts):
+            if _partition_is_snapshot(day, part_dir, quote_ts):
                 issues.append(IntegrityIssue(day=day, table=table, kind="snapshot"))
 
     issues.sort(key=lambda i: (i.day, i.table))

backend/app/services/quote_service.py

@@ -935,6 +935,11 @@ def _build_daily(records: list[dict]) -> pl.DataFrame:
         result = df.select(select_exprs).with_columns(
             pl.lit(cn_today()).cast(pl.Date).alias("date"),
         )
+        # 停牌/尚无集合竞价的记录 open/high 均为 0。必须在下方用 close 填充前
+        # 过滤, 否则零成交行会被伪装成有效日K, 并在 batch 同步后作为实时残留
+        # 反复触发历史完整性修复。
+        from app.indicators.pipeline import filter_halt_days
+        result = filter_halt_days(result)
         # 修复: API 在非交易时段可能返回 open/high/low=0 或 null,
         # 导致蜡烛从 0 开始。用 close 填充这些异常值。
         for col in ("open", "high", "low"):

backend/tests/test_data_integrity.py

@@ -107,13 +107,115 @@ def test_final_snapshot_after_close_is_clean(tmp_path):
     assert scan_recent_integrity(tmp_path, today=TODAY) == []
 
 
+def test_zero_volume_live_residue_amid_batch_rows_is_clean(tmp_path):
+    """盘后 batch 已覆盖主体数据时, 单个停牌实时残留不能误判整个分区。"""
+    part = tmp_path / "kline_daily" / f"date={FRIDAY.isoformat()}"
+    part.mkdir(parents=True)
+    pl.DataFrame({
+        "symbol": ["600001.SH", "600002.SH", "600003.SH", "600004.SH"],
+        "date": [FRIDAY] * 4,
+        "open": [10.0, 9.8, 12.0, 8.0],
+        "high": [10.2, 9.8, 12.2, 8.1],
+        "low": [9.9, 9.8, 11.9, 7.9],
+        "close": [10.1, 9.8, 12.1, 8.0],
+        "volume": [1000.0, 0.0, 1200.0, 800.0],
+        "amount": [10100.0, 0.0, 14520.0, 6400.0],
+        "quote_ts": [None, _ts_ms(FRIDAY, time(9, 15)), None, None],
+    }).write_parquet(part / "part.parquet")
+    _write_daily_partition(tmp_path, "kline_daily", TODAY, _ts_ms(TODAY, time(10, 0)))
+
+    assert scan_recent_integrity(tmp_path, today=TODAY) == []
+
+
+def test_mostly_zero_preopen_rows_with_one_batch_row_is_flagged(tmp_path):
+    """少量 batch 行不能掩盖占主体的盘前实时快照。"""
+    part = tmp_path / "kline_daily" / f"date={FRIDAY.isoformat()}"
+    part.mkdir(parents=True)
+    preopen_ts = _ts_ms(FRIDAY, time(9, 15))
+    pl.DataFrame({
+        "symbol": ["600001.SH", "600002.SH", "600003.SH", "600004.SH"],
+        "date": [FRIDAY] * 4,
+        "open": [10.0, 20.0, 30.0, 40.0],
+        "high": [10.0, 20.0, 30.0, 40.1],
+        "low": [10.0, 20.0, 30.0, 39.9],
+        "close": [10.0, 20.0, 30.0, 40.0],
+        "volume": [0.0, 0.0, 0.0, 100.0],
+        "amount": [0.0, 0.0, 0.0, 4000.0],
+        "quote_ts": [preopen_ts, preopen_ts, preopen_ts, None],
+    }).write_parquet(part / "part.parquet")
+    _write_daily_partition(tmp_path, "kline_daily", TODAY, _ts_ms(TODAY, time(10, 0)))
+
+    issues = scan_recent_integrity(tmp_path, today=TODAY)
+    assert [(i.day, i.table, i.kind) for i in issues] == [
+        (FRIDAY, "kline_daily", "snapshot"),
+    ]
+
+
+def test_all_zero_preopen_live_partition_is_still_flagged(tmp_path):
+    """整分区都是盘前实时数据时仍须修复, 不能因零成交而放过。"""
+    part = tmp_path / "kline_daily" / f"date={FRIDAY.isoformat()}"
+    part.mkdir(parents=True)
+    pl.DataFrame({
+        "symbol": ["600001.SH", "600002.SH"],
+        "date": [FRIDAY, FRIDAY],
+        "open": [10.0, 20.0],
+        "high": [10.0, 20.0],
+        "low": [10.0, 20.0],
+        "close": [10.0, 20.0],
+        "volume": [0.0, 0.0],
+        "amount": [0.0, 0.0],
+        "quote_ts": [_ts_ms(FRIDAY, time(9, 15))] * 2,
+    }).write_parquet(part / "part.parquet")
+    _write_daily_partition(tmp_path, "kline_daily", TODAY, _ts_ms(TODAY, time(10, 0)))
+
+    issues = scan_recent_integrity(tmp_path, today=TODAY)
+    assert [(i.day, i.table, i.kind) for i in issues] == [
+        (FRIDAY, "kline_daily", "snapshot"),
+    ]
+
+
 def test_today_partition_is_never_flagged(tmp_path):
     # 今天的盘中 quote_ts 属正常实时更新
     _write_daily_partition(tmp_path, "kline_daily", FRIDAY, None)
     _write_daily_partition(tmp_path, "kline_daily", TODAY, _ts_ms(TODAY, time(9, 45)))
     assert scan_recent_integrity(tmp_path, today=TODAY) == []
 
 
+def test_realtime_daily_builder_drops_halted_rows_before_zero_fill():
+    from app.services.quote_service import QuoteService
+
+    result = QuoteService._build_daily([
+        {
+            "symbol": "600001.SH", "last_price": 10.0,
+            "open": 9.9, "high": 10.1, "low": 9.8,
+            "volume": 1000.0, "amount": 10000.0,
+            "timestamp": _ts_ms(TODAY, time(10, 0)),
+        },
+        {
+            "symbol": "600002.SH", "last_price": 20.0,
+            "open": 0.0, "high": 0.0, "low": 0.0,
+            "volume": 0.0, "amount": 0.0,
+            "timestamp": _ts_ms(TODAY, time(9, 15)),
+        },
+    ])
+
+    assert result["symbol"].to_list() == ["600001.SH"]
+
+
+def test_halt_filter_drops_legacy_zero_volume_row_after_ohlc_fill():
+    from app.indicators.pipeline import filter_halt_days
+
+    result = filter_halt_days(pl.DataFrame({
+        "symbol": ["600001.SH", "600002.SH"],
+        "open": [10.0, 20.0],
+        "high": [10.2, 20.0],
+        "volume": [1000.0, 0.0],
+        "amount": [10100.0, 0.0],
+    }))
+
+    assert result["symbol"].to_list() == ["600001.SH"]
+
+
 def test_missing_tail_day_flagged(tmp_path):
     # 周四有数据, 周五(工作日)整天停机缺失, 今天周一启动
     _write_daily_partition(tmp_path, "kline_daily", THURSDAY, None)
@@ -270,8 +372,18 @@ def test_realtime_gate_blocks_on_snapshot_and_launches_repair(tmp_path, monkeypa
     from app.api import settings as settings_api
     from app.services import data_integrity
 
-    _write_daily_partition(tmp_path, "kline_daily", FRIDAY, _ts_ms(FRIDAY, time(11, 58)))
-    _write_daily_partition(tmp_path, "kline_daily", TODAY, _ts_ms(TODAY, time(10, 0)))
+    real_today = datetime.now(CN_TZ).date()
+    snapshot_day = real_today - timedelta(days=1)
+    while snapshot_day.weekday() >= 5:
+        snapshot_day -= timedelta(days=1)
+    _write_daily_partition(
+        tmp_path, "kline_daily", snapshot_day,
+        _ts_ms(snapshot_day, time(11, 58)),
+    )
+    _write_daily_partition(
+        tmp_path, "kline_daily", real_today,
+        _ts_ms(real_today, time(10, 0)),
+    )
 
     launched = []
     monkeypatch.setattr(
@@ -294,15 +406,22 @@ def test_realtime_gate_blocks_on_snapshot_and_launches_repair(tmp_path, monkeypa
     assert "盘中快照" in exc_info.value.detail
     assert "job-x" in exc_info.value.detail
     # 修复任务以最早坏日为起点, 且实时行情未被开启
-    assert launched == [(FRIDAY, "realtime_gate")]
+    assert launched == [(snapshot_day, "realtime_gate")]
     assert saved == {}
 
 
 def test_realtime_gate_allows_clean_data(tmp_path, monkeypatch):
     from app.api import settings as settings_api
 
-    _write_daily_partition(tmp_path, "kline_daily", FRIDAY, None)
-    _write_daily_partition(tmp_path, "kline_daily", TODAY, _ts_ms(TODAY, time(10, 0)))
+    real_today = datetime.now(CN_TZ).date()
+    previous_day = real_today - timedelta(days=1)
+    while previous_day.weekday() >= 5:
+        previous_day -= timedelta(days=1)
+    _write_daily_partition(tmp_path, "kline_daily", previous_day, None)
+    _write_daily_partition(
+        tmp_path, "kline_daily", real_today,
+        _ts_ms(real_today, time(10, 0)),
+    )
 
     saved = {}
     monkeypatch.setattr(
@@ -321,11 +440,15 @@ def test_realtime_gate_allows_clean_data(tmp_path, monkeypatch):
 def test_realtime_gate_ignores_old_issues_beyond_window(tmp_path, monkeypatch):
     from app.api import settings as settings_api
 
-    old_day = TODAY - timedelta(days=AUTO_REPAIR_MAX_LAG_DAYS + 1)
+    real_today = datetime.now(CN_TZ).date()
+    old_day = real_today - timedelta(days=AUTO_REPAIR_MAX_LAG_DAYS + 1)
     while old_day.weekday() >= 5:
         old_day -= timedelta(days=1)
     _write_daily_partition(tmp_path, "kline_daily", old_day, _ts_ms(old_day, time(11, 58)))
-    _write_daily_partition(tmp_path, "kline_daily", TODAY, _ts_ms(TODAY, time(10, 0)))
+    _write_daily_partition(
+        tmp_path, "kline_daily", real_today,
+        _ts_ms(real_today, time(10, 0)),
+    )
 
     saved = {}
     monkeypatch.setattr(