From f325a9676ccce1ce5c9a7a4bc12c4279143f8340 Mon Sep 17 00:00:00 2001 From: 0112020179 Date: Tue, 8 Sep 2026 23:49:55 +0800 Subject: [PATCH] =?UTF-8?q?fix(pipeline):=20=E9=81=BF=E5=85=8D=E5=81=9C?= =?UTF-8?q?=E7=89=8C=E5=88=86=E5=8C=BA=E5=8F=8D=E5=A4=8D=E8=AF=AF=E5=88=A0?= =?UTF-8?q?=E9=87=8D=E7=AE=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/jobs/daily_pipeline.py | 50 +++--- .../tests/test_enriched_partial_partition.py | 166 +++++++++++++++++- 2 files changed, 188 insertions(+), 28 deletions(-) diff --git a/backend/app/jobs/daily_pipeline.py b/backend/app/jobs/daily_pipeline.py index fdc38f5..3c3e70f 100644 --- a/backend/app/jobs/daily_pipeline.py +++ b/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), diff --git a/backend/tests/test_enriched_partial_partition.py b/backend/tests/test_enriched_partial_partition.py index 2eff00f..844cf0c 100644 --- a/backend/tests/test_enriched_partial_partition.py +++ b/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