diff --git a/backend/app/indicators/pipeline.py b/backend/app/indicators/pipeline.py index 3a80d00..d625c1f 100644 --- a/backend/app/indicators/pipeline.py +++ b/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) # ================================================================ diff --git a/backend/app/services/data_integrity.py b/backend/app/services/data_integrity.py index 8361b3e..d0b0cae 100644 --- a/backend/app/services/data_integrity.py +++ b/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)) diff --git a/backend/app/services/quote_service.py b/backend/app/services/quote_service.py index 812cfbb..ce69ac5 100644 --- a/backend/app/services/quote_service.py +++ b/backend/app/services/quote_service.py @@ -827,6 +827,11 @@ class QuoteService: 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"): diff --git a/backend/tests/test_data_integrity.py b/backend/tests/test_data_integrity.py index f9e43c0..07eebbd 100644 --- a/backend/tests/test_data_integrity.py +++ b/backend/tests/test_data_integrity.py @@ -107,6 +107,73 @@ 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) @@ -114,6 +181,41 @@ def test_today_partition_is_never_flagged(tmp_path): 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) @@ -275,8 +377,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( @@ -299,7 +411,7 @@ 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 == {} @@ -330,8 +442,15 @@ def test_realtime_gate_blocks_when_no_local_data(tmp_path, monkeypatch): 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( @@ -350,11 +469,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(