fix(data): akshare 三表下载鲁棒性 — 原子写 + --repair (G4)

write_parquet_and_marker 改原子写(tmp→os.replace→marker) kill 不产残缺 parquet; 新增 --repair 只重取 missing/empty/corrupt 忽略 marker(周度补漏不必等财报季 --force 全量); is_parquet_healthy 辅助。top_holders 修复无回归, 106 tests。

scripts/data_platform/akshare_static_download.py

@@ -386,20 +386,63 @@ def load_done_units(data_type: str) -> set:
     return done
 
 
+# 空 parquet 阈值: 文件 <1KB 视为 empty/corrupt (有效 parquet 哪怕 0 行也 >1KB
+# 因为有 schema/metadata; 真正 0 字节或几十字节的肯定是异常).
+EMPTY_PARQUET_MIN_BYTES = 1024
+
+
+def is_parquet_healthy(parquet_path: Path) -> bool:
+    """判断 parquet 是否值得保留 (有数据)。返 True 健康 / False 需重取。
+
+    unhealthy 条件 (任一):
+      - 文件不存在 (orphan marker / 被删)
+      - size < EMPTY_PARQUET_MIN_BYTES (残缺或 0 字节)
+      - pd.read_parquet 抛异常 (corrupt magic byte 等)
+      - 读后 df.empty (空数据, 如北交所 akshare 不覆盖)
+
+    --repair 模式下用此函数决定是否忽略 marker 强制重取。
+    """
+    try:
+        if not parquet_path.exists():
+            return False
+        if parquet_path.stat().st_size < EMPTY_PARQUET_MIN_BYTES:
+            return False
+        df = pd.read_parquet(parquet_path)
+        return not df.empty
+    except Exception:
+        return False
+
+
 def write_parquet_and_marker(
     df: pd.DataFrame,
     parquet_path: Path,
 ) -> bool:
-    """写 parquet + marker。返 True 成功 / False 失败。"""
+    """原子写 parquet + marker。返 True 成功 / False 失败。
+
+    原子语义: 先写 .parquet.tmp → os.replace 到正式路径 → 写 marker。
+    进程被 kill / 断电时:
+      - to_parquet 中断: 只留 .tmp (正式路径未触碰, 旧版本数据保留)
+      - os.replace 中断: 同上 (replace 是原子操作, 要么完成要么没发生)
+      - marker 未写: 下次非 --force 会重试 (marker 是断点续传真相源)
+    失败时清理 .tmp 残骸 (replace 失败时 .tmp 还在; 成功后 .tmp 已消失)。
+    """
+    tmp_path = parquet_path.with_suffix(parquet_path.suffix + ".tmp")
     try:
         parquet_path.parent.mkdir(parents=True, exist_ok=True)
-        df.to_parquet(parquet_path, index=False)
+        df.to_parquet(tmp_path, index=False)
+        os.replace(tmp_path, parquet_path)
+        # marker 仅在 replace 成功后写 (replace 是 Linux/Windows 上的原子操作)
         marker_path_for(parquet_path).write_text(
             datetime.datetime.now().isoformat()
         )
         return True
     except Exception as e:
         logger.error("写入 %s 失败: %s", parquet_path, e)
+        # 清理 .tmp 残骸 (replace 失败时它还在; 成功后它已消失)
+        try:
+            tmp_path.unlink(missing_ok=True)
+        except OSError:
+            pass
         return False
 
 
@@ -612,6 +655,7 @@ def download_one_unit(
     unit_id: str,
     fetch_fn: Callable[[], pd.DataFrame],
     force: bool,
+    repair: bool = False,
 ) -> Tuple[str, int]:
     """通用单 unit 下载: 拉 df → 写 parquet + marker。
 
@@ -619,12 +663,24 @@ def download_one_unit(
     各 fetch_xxx 已把 None 转空 df; 这里 df 永远非 None 但可能空)。
 
     返 (status, rows), status ∈ {'ok', 'skipped', 'empty', 'failed'}。
+
+    跳过策略 (优先级从高到低):
+      - force=True: 总是重取 (忽略 marker 与 repair)
+      - repair=True: marker 在但 parquet unhealthy (missing/empty/corrupt) 仍重取;
+                     marker 在且 parquet 健康 → skip
+      - 默认 (force=repair=False): marker 在 → skip (空文件也标 done, 避免全量重跑)
     """
     parquet_path = parquet_path_per_unit(data_type, unit_id)
     marker_path = marker_path_for(parquet_path)
 
     if not force and marker_path.exists():
-        return "skipped", 0
+        if repair and not is_parquet_healthy(parquet_path):
+            logger.info(
+                "[%s] %s --repair: parquet unhealthy (missing/empty/corrupt), 重取",
+                data_type, unit_id,
+            )
+        else:
+            return "skipped", 0
 
     try:
         df = fetch_fn()
@@ -661,11 +717,20 @@ def run_one_type(
     done_set = load_done_units(data_type)
     if args.force:
         todo = [(uid, fn) for uid, fn in units]
+    elif getattr(args, "repair", False):
+        # repair 模式: 只处理 missing/empty/corrupt 的 unit
+        # (marker 在但 parquet unhealthy → 重取; marker 在且健康 → 跳过)
+        todo = [
+            (uid, fn) for uid, fn in units
+            if not (uid in done_set
+                    and is_parquet_healthy(parquet_path_per_unit(data_type, uid)))
+        ]
     else:
         todo = [(uid, fn) for uid, fn in units if uid not in done_set]
     logger.info(
-        "[%s] 待处理 %d (已完成 %d, 总 %d)",
+        "[%s] 待处理 %d (已完成 %d, 总 %d%s)",
         data_type, len(todo), len(done_set), len(units),
+        ", repair 模式" if getattr(args, "repair", False) else "",
     )
 
     stats = {"ok": 0, "skipped": 0, "empty": 0, "failed": 0, "rows": 0}
@@ -676,7 +741,10 @@ def run_one_type(
 
     for i, (uid, fn) in enumerate(todo):
         try:
-            status, rows = download_one_unit(data_type, uid, fn, args.force)
+            status, rows = download_one_unit(
+                data_type, uid, fn, args.force,
+                repair=getattr(args, "repair", False),
+            )
         except Exception as e:
             status, rows = "failed", 0
             logger.debug("[%s] %s 异常: %s", data_type, uid, e)
@@ -897,6 +965,11 @@ def parse_args() -> argparse.Namespace:
         help="限制处理股票数 (per-stock 类生效), 测试用",
     )
     p.add_argument("--force", action="store_true", help="强制重下, 忽略 marker")
+    p.add_argument(
+        "--repair", action="store_true",
+        help="只重取 missing/empty(size<1KB 或 df.empty)/corrupt 的 parquet, "
+             "忽略其 marker。适合周度补漏, 不必等财报季 --force 全量重跑",
+    )
     return p.parse_args()
 
 

tests/data_platform/test_g4_atomic_and_repair.py

@@ -0,0 +1,444 @@
+# -*- coding: utf-8 -*-
+"""G4 鲁棒性测试: 原子写 + --repair 模式 (task: 防复发 parquet 空/损坏)。
+
+针对两处根因:
+  1. write_parquet_and_marker 非原子写 → kill 中断留残缺 parquet (magic byte 错).
+  2. 空 df 也写 marker → 空文件永久占位 (下次非 --force 跳过, 永不重试).
+
+修复策略:
+  1. 原子写: 先写 .parquet.tmp → os.replace → 写 marker. 失败时清 tmp,
+     正式 parquet 不被污染.
+  2. --repair 模式: 扫已有 parquet, 对 missing / empty / corrupt 强制重取,
+     忽略 marker. 正常模式 (无 --repair) marker 语义不变 (空文件仍标 done).
+"""
+import argparse
+import os
+import sys
+from pathlib import Path
+from unittest.mock import patch
+
+import pandas as pd
+import pytest
+
+_SCRIPT_DIR = os.path.join(os.path.dirname(__file__), "..", "..", "scripts", "data_platform")
+_SCRIPT_DIR = os.path.abspath(_SCRIPT_DIR)
+if _SCRIPT_DIR not in sys.path:
+    sys.path.insert(0, _SCRIPT_DIR)
+
+import akshare_static_download as mod  # noqa: E402
+
+
+# ======================== 原子写测试 ========================
+
+class TestAtomicWrite:
+    """write_parquet_and_marker 原子写: 失败时正式 parquet 不被污染。"""
+
+    def test_successful_write_creates_parquet_and_marker(self, tmp_path):
+        """正常写: parquet + marker 都生成, 无 tmp 残骸。"""
+        pq = tmp_path / "balance" / "600519.SH_balance.parquet"
+        df = pd.DataFrame({"col": [1, 2, 3]})
+
+        ok = mod.write_parquet_and_marker(df, pq)
+
+        assert ok is True
+        assert pq.exists()
+        assert mod.marker_path_for(pq).exists()
+        # 无 tmp 残骸
+        assert not pq.with_suffix(pq.suffix + ".tmp").exists()
+        # 内容正确
+        df_back = pd.read_parquet(pq)
+        assert df_back["col"].tolist() == [1, 2, 3]
+
+    def test_interrupt_does_not_corrupt_existing_parquet(self, tmp_path):
+        """模拟 to_parquet 中途抛异常 (kill/断电): 正式 parquet 保持原内容。
+
+        场景: 已有旧版本 parquet, 重写时 to_parquet 抛异常 → 旧版本应保留,
+        不留残缺文件, 不写 marker。
+        """
+        pq = tmp_path / "balance" / "600519.SH_balance.parquet"
+        pq.parent.mkdir(parents=True, exist_ok=True)
+        # 写旧版本 (模拟 "上次成功的数据")
+        original_df = pd.DataFrame({"a": [1, 2]})
+        original_df.to_parquet(pq, index=False)
+
+        # 模拟 to_parquet 失败 (kill 中断 / 磁盘满 / 等)
+        new_df = pd.DataFrame({"a": [10, 20]})
+        with patch.object(pd.DataFrame, "to_parquet",
+                          side_effect=RuntimeError("simulated kill")):
+            ok = mod.write_parquet_and_marker(new_df, pq)
+
+        assert ok is False
+        # 关键断言: 正式 parquet 保持旧内容 (未被覆写)
+        assert pq.exists()
+        df_back = pd.read_parquet(pq)
+        assert df_back["a"].tolist() == [1, 2]
+        # 无 tmp 残骸 (失败路径应清理)
+        assert not pq.with_suffix(pq.suffix + ".tmp").exists()
+
+    def test_interrupt_from_scratch_leaves_no_formal_parquet(self, tmp_path):
+        """首次写入 (正式 parquet 不存在) 时中断: 不留任何正式 parquet。
+
+        场景: 简化版 — 重点是失败路径不写 marker。
+        (注: 我们的实现是先写 .tmp 再 os.replace 到正式路径, 所以 to_parquet
+        失败时正式路径根本不存在, 这一点由 os.replace 语义保证; 这里只测 marker。)
+        """
+        pq = tmp_path / "new" / "fresh.parquet"
+        pq.parent.mkdir(parents=True, exist_ok=True)
+
+        new_df = pd.DataFrame({"a": [1]})
+        with patch.object(pd.DataFrame, "to_parquet",
+                          side_effect=RuntimeError("simulated kill")):
+            ok = mod.write_parquet_and_marker(new_df, pq)
+
+        assert ok is False
+        # 失败时 marker 必须未写 (否则下次会跳过 = 永久污染)
+        assert not mod.marker_path_for(pq).exists()
+        # tmp 应被清理 (失败路径清理逻辑)
+        assert not pq.with_suffix(pq.suffix + ".tmp").exists()
+
+    def test_marker_only_written_after_replace_success(self, tmp_path):
+        """marker 仅在 os.replace 成功后写。
+
+        用 spy 替换 os.replace, 让它失败 → marker 不应被写。
+        """
+        pq = tmp_path / "balance" / "000001.SZ_balance.parquet"
+        df = pd.DataFrame({"col": [1]})
+
+        with patch.object(mod.os, "replace",
+                          side_effect=OSError("simulated replace failure")):
+            ok = mod.write_parquet_and_marker(df, pq)
+
+        assert ok is False
+        # marker 必须未写
+        assert not mod.marker_path_for(pq).exists()
+        # 正式 parquet 必须不存在 (replace 失败 = 未抵达)
+        assert not pq.exists()
+
+
+# ======================== is_parquet_healthy 工具 ========================
+
+class TestIsParquetHealthy:
+    """is_parquet_healthy: 判断 parquet 是否值得保留 (非空/非损坏)。"""
+
+    def test_missing_file_is_unhealthy(self, tmp_path):
+        pq = tmp_path / "no_such.parquet"
+        assert mod.is_parquet_healthy(pq) is False
+
+    def test_empty_content_is_unhealthy(self, tmp_path):
+        """df.empty 的 parquet (如北交所 akshare 返空) → unhealthy."""
+        pq = tmp_path / "empty.parquet"
+        pd.DataFrame({"col": []}).to_parquet(pq, index=False)
+        assert mod.is_parquet_healthy(pq) is False
+
+    def test_empty_with_columns_is_unhealthy(self, tmp_path):
+        """有列名但 0 行 (如 _safe_top_10_em 返的 schema-only df) → unhealthy."""
+        pq = tmp_path / "empty_cols.parquet"
+        pd.DataFrame(columns=["A", "B", "C"]).to_parquet(pq, index=False)
+        assert mod.is_parquet_healthy(pq) is False
+
+    def test_tiny_file_is_unhealthy(self, tmp_path):
+        """size < 1KB → unhealthy (空 df 写出的 parquet 或残缺文件)."""
+        pq = tmp_path / "tiny.parquet"
+        pq.write_bytes(b"x" * 100)  # 100B 垃圾
+        assert mod.is_parquet_healthy(pq) is False
+
+    def test_corrupt_file_is_unhealthy(self, tmp_path):
+        """read_parquet 抛异常 → unhealthy."""
+        pq = tmp_path / "corrupt.parquet"
+        # 写 2KB 垃圾 (超过 1KB 阈值, 但内容不是 parquet)
+        pq.write_bytes(b"not a parquet file " * 200)
+        assert mod.is_parquet_healthy(pq) is False
+
+    def test_healthy_parquet_with_data(self, tmp_path):
+        """有数据的正常 parquet → healthy."""
+        pq = tmp_path / "healthy.parquet"
+        pd.DataFrame({"col": [1, 2, 3]}).to_parquet(pq, index=False)
+        assert mod.is_parquet_healthy(pq) is True
+
+
+# ======================== --repair 模式: download_one_unit ========================
+
+class TestDownloadOneUnitRepairMode:
+    """download_one_unit 在 repair 模式下的行为。"""
+
+    def _setup_outdir(self, monkeypatch, tmp_path):
+        monkeypatch.setattr(mod, "OUT_DIR", tmp_path)
+
+    def test_repair_refetches_empty_parquet(self, tmp_path, monkeypatch):
+        """empty parquet 在 repair 模式下被重取。"""
+        self._setup_outdir(monkeypatch, tmp_path)
+        data_type = "balance"
+        unit_id = "600519.SH_balance"
+        pq = mod.parquet_path_per_unit(data_type, unit_id)
+        pq.parent.mkdir(parents=True, exist_ok=True)
+        # 写空 parquet (df.empty) + marker (模拟 "上次跑完, 标 done")
+        pd.DataFrame({"col": []}).to_parquet(pq, index=False)
+        mod.marker_path_for(pq).write_text("2026-07-01T00:00:00")
+
+        fetch_fn = lambda: pd.DataFrame({"col": [1, 2, 3]})
+
+        status, rows = mod.download_one_unit(
+            data_type, unit_id, fetch_fn, force=False, repair=True)
+
+        assert status == "ok"
+        assert rows == 3
+        df_back = pd.read_parquet(pq)
+        assert df_back["col"].tolist() == [1, 2, 3]
+
+    def test_repair_refetches_corrupt_parquet(self, tmp_path, monkeypatch):
+        """corrupt parquet (read 抛异常) 在 repair 模式下被重取。"""
+        self._setup_outdir(monkeypatch, tmp_path)
+        data_type = "balance"
+        unit_id = "600519.SH_corrupt"
+        pq = mod.parquet_path_per_unit(data_type, unit_id)
+        pq.parent.mkdir(parents=True, exist_ok=True)
+        # 写 corrupt parquet (2KB 垃圾) + marker
+        pq.write_bytes(b"not parquet " * 200)
+        mod.marker_path_for(pq).write_text("2026-07-01T00:00:00")
+
+        fetch_fn = lambda: pd.DataFrame({"col": [1]})
+
+        status, rows = mod.download_one_unit(
+            data_type, unit_id, fetch_fn, force=False, repair=True)
+
+        assert status == "ok"
+        assert rows == 1
+        # parquet 被覆写为有效内容
+        df_back = pd.read_parquet(pq)
+        assert df_back["col"].tolist() == [1]
+
+    def test_repair_refetches_missing_parquet(self, tmp_path, monkeypatch):
+        """只有 marker, parquet 不存在 → repair 模式重取。"""
+        self._setup_outdir(monkeypatch, tmp_path)
+        data_type = "balance"
+        unit_id = "600519.SH_missing"
+        pq = mod.parquet_path_per_unit(data_type, unit_id)
+        pq.parent.mkdir(parents=True, exist_ok=True)
+        mod.marker_path_for(pq).write_text("2026-07-01T00:00:00")
+
+        fetch_fn = lambda: pd.DataFrame({"col": [1]})
+
+        status, rows = mod.download_one_unit(
+            data_type, unit_id, fetch_fn, force=False, repair=True)
+
+        assert status == "ok"
+        assert pq.exists()
+
+    def test_repair_skips_healthy_parquet(self, tmp_path, monkeypatch):
+        """healthy parquet (marker + 健康) 在 repair 模式下仍 skip。"""
+        self._setup_outdir(monkeypatch, tmp_path)
+        data_type = "balance"
+        unit_id = "600519.SH_balance"
+        pq = mod.parquet_path_per_unit(data_type, unit_id)
+        pq.parent.mkdir(parents=True, exist_ok=True)
+        pd.DataFrame({"col": [1, 2, 3]}).to_parquet(pq, index=False)
+        mod.marker_path_for(pq).write_text("2026-07-01T00:00:00")
+
+        fetch_called = {"n": 0}
+        def fetch_fn():
+            fetch_called["n"] += 1
+            return pd.DataFrame({"col": [99]})
+
+        status, rows = mod.download_one_unit(
+            data_type, unit_id, fetch_fn, force=False, repair=True)
+
+        assert status == "skipped"
+        assert fetch_called["n"] == 0
+        df_back = pd.read_parquet(pq)
+        assert df_back["col"].tolist() == [1, 2, 3]
+
+    def test_repair_refetches_unit_without_marker(self, tmp_path, monkeypatch):
+        """没 marker (从未跑过) 的 unit, repair 模式也重取 (与正常模式行为一致)."""
+        self._setup_outdir(monkeypatch, tmp_path)
+        data_type = "balance"
+        unit_id = "600519.SH_nomark"
+        pq = mod.parquet_path_per_unit(data_type, unit_id)
+        pq.parent.mkdir(parents=True, exist_ok=True)
+
+        fetch_fn = lambda: pd.DataFrame({"col": [1]})
+
+        status, rows = mod.download_one_unit(
+            data_type, unit_id, fetch_fn, force=False, repair=True)
+
+        assert status == "ok"
+        assert pq.exists()
+        assert mod.marker_path_for(pq).exists()
+
+
+class TestDownloadOneUnitNormalModeNoRegression:
+    """回归测试: 正常模式 (无 --repair) marker 语义不变。"""
+
+    def _setup_outdir(self, monkeypatch, tmp_path):
+        monkeypatch.setattr(mod, "OUT_DIR", tmp_path)
+
+    def test_normal_mode_skips_empty_parquet_with_marker(self, tmp_path, monkeypatch):
+        """正常模式: empty parquet 但 marker 在 → skip (避免全量重跑)."""
+        self._setup_outdir(monkeypatch, tmp_path)
+        data_type = "balance"
+        unit_id = "600519.SH_balance"
+        pq = mod.parquet_path_per_unit(data_type, unit_id)
+        pq.parent.mkdir(parents=True, exist_ok=True)
+        pd.DataFrame({"col": []}).to_parquet(pq, index=False)
+        mod.marker_path_for(pq).write_text("2026-07-01T00:00:00")
+
+        fetch_called = {"n": 0}
+        def fetch_fn():
+            fetch_called["n"] += 1
+            return pd.DataFrame({"col": [99]})
+
+        status, rows = mod.download_one_unit(
+            data_type, unit_id, fetch_fn, force=False, repair=False)
+
+        assert status == "skipped"
+        assert fetch_called["n"] == 0
+
+    def test_force_mode_refetches_regardless_of_health(self, tmp_path, monkeypatch):
+        """--force 总是重取, 不管 repair 与否 (force 优先级最高)."""
+        self._setup_outdir(monkeypatch, tmp_path)
+        data_type = "balance"
+        unit_id = "600519.SH_force"
+        pq = mod.parquet_path_per_unit(data_type, unit_id)
+        pq.parent.mkdir(parents=True, exist_ok=True)
+        pd.DataFrame({"col": [1, 2, 3]}).to_parquet(pq, index=False)
+        mod.marker_path_for(pq).write_text("2026-07-01T00:00:00")
+
+        fetch_fn = lambda: pd.DataFrame({"col": [99]})
+
+        status, rows = mod.download_one_unit(
+            data_type, unit_id, fetch_fn, force=True, repair=False)
+
+        assert status == "ok"
+        df_back = pd.read_parquet(pq)
+        assert df_back["col"].tolist() == [99]
+
+
+# ======================== run_one_type: repair 模式过滤 ========================
+
+class TestRunOneTypeRepairFilter:
+    """run_one_type 在 repair 模式下: 只把 unhealthy units 放入 todo。"""
+
+    def _setup_outdir(self, monkeypatch, tmp_path):
+        monkeypatch.setattr(mod, "OUT_DIR", tmp_path)
+        monkeypatch.setattr(mod, "AK_INTERVAL", 0)  # 加速
+
+    def _write_unit(self, data_type, unit_id, df):
+        pq = mod.parquet_path_per_unit(data_type, unit_id)
+        pq.parent.mkdir(parents=True, exist_ok=True)
+        df.to_parquet(pq, index=False)
+        mod.marker_path_for(pq).write_text("2026-07-01T00:00:00")
+        return pq
+
+    def test_repair_filters_healthy_keeps_unhealthy(self, tmp_path, monkeypatch):
+        """run_one_type repair: healthy 不进 todo, empty/missing/corrupt 重取。
+
+        注: 健康的 unit 在 todo 构造阶段就被过滤掉 (不进 download_one_unit),
+        所以 stats['skipped'] 始终是 0 (该计数器只在 unit 级别 check 时递增,
+        见 download_one_unit 的 skipped 返回)。验证重点: ok=3, 文件被覆写。
+        """
+        self._setup_outdir(monkeypatch, tmp_path)
+        data_type = "balance"
+        healthy_id = "000001.SZ_balance"
+        empty_id = "000002.SZ_balance"
+        missing_id = "000003.SZ_balance"
+        corrupt_id = "000004.SZ_balance"
+
+        self._write_unit(data_type, healthy_id, pd.DataFrame({"col": [1, 2, 3]}))
+        self._write_unit(data_type, empty_id, pd.DataFrame({"col": []}))
+        # missing: 只写 marker, 不写 parquet
+        mp = mod.marker_path_for(
+            mod.parquet_path_per_unit(data_type, missing_id))
+        mp.parent.mkdir(parents=True, exist_ok=True)
+        mp.write_text("2026-07-01T00:00:00")
+        # corrupt: 写垃圾 + marker
+        cp = mod.parquet_path_per_unit(data_type, corrupt_id)
+        cp.parent.mkdir(parents=True, exist_ok=True)
+        cp.write_bytes(b"garbage " * 500)
+        mod.marker_path_for(cp).write_text("2026-07-01T00:00:00")
+
+        units = [
+            (healthy_id, lambda: pd.DataFrame({"col": [99]})),
+            (empty_id, lambda: pd.DataFrame({"col": [99]})),
+            (missing_id, lambda: pd.DataFrame({"col": [99]})),
+            (corrupt_id, lambda: pd.DataFrame({"col": [99]})),
+        ]
+        args = argparse.Namespace(force=False, repair=True)
+
+        stats, circuit = mod.run_one_type(data_type, units, args)
+
+        assert circuit is False
+        # 3 个 unhealthy 重取成功; healthy 不进 todo (filtered out)
+        assert stats["ok"] == 3, f"expected 3 ok, got {stats}"
+        assert stats["failed"] == 0
+        # healthy parquet 内容未被改 (没被覆写成 [99])
+        healthy_pq = mod.parquet_path_per_unit(data_type, healthy_id)
+        df_back = pd.read_parquet(healthy_pq)
+        assert df_back["col"].tolist() == [1, 2, 3]
+        # empty/missing/corrupt 都被覆写为新数据
+        for uid in (empty_id, missing_id, corrupt_id):
+            df_back = pd.read_parquet(mod.parquet_path_per_unit(data_type, uid))
+            assert df_back["col"].tolist() == [99]
+
+    def test_normal_mode_no_regression_filter(self, tmp_path, monkeypatch):
+        """回归: 正常模式 (无 repair) 4 个都有 marker → 全部过滤, 0 ok。
+
+        注: done_set 过滤在 todo 构造阶段, units 都不进 todo, 所以 0 ok。
+        skipped 计数器保持 0 (该计数器只在 unit 级 check 时递增)。
+        """
+        self._setup_outdir(monkeypatch, tmp_path)
+        data_type = "balance"
+        ids = [f"00000{i}.SZ_balance" for i in range(4)]
+        for uid in ids:
+            self._write_unit(data_type, uid, pd.DataFrame({"col": [1]}))
+
+        fetch_called = {"n": 0}
+        def fetch_fn():
+            fetch_called["n"] += 1
+            return pd.DataFrame({"col": [99]})
+
+        units = [(uid, fetch_fn) for uid in ids]
+        args = argparse.Namespace(force=False, repair=False)
+
+        stats, circuit = mod.run_one_type(data_type, units, args)
+
+        assert circuit is False
+        # 关键: 没有任何 fetch 被调用 (marker 在的不重取)
+        assert fetch_called["n"] == 0
+        assert stats["ok"] == 0
+        assert stats["failed"] == 0
+        # 旧数据保持不变
+        for uid in ids:
+            df_back = pd.read_parquet(mod.parquet_path_per_unit(data_type, uid))
+            assert df_back["col"].tolist() == [1]
+
+
+# ======================== CLI --repair flag ========================
+
+class TestRepairCliFlag:
+    """parse_args 接受 --repair flag."""
+
+    def test_repair_default_false(self):
+        args = mod.parse_args.__wrapped__(["--types", "balance"]) \
+            if hasattr(mod.parse_args, "__wrapped__") \
+            else _parse_args_helper(["--types", "balance"])
+        assert args.repair is False
+
+    def test_repair_flag_true(self):
+        args = _parse_args_helper(["--types", "balance", "--repair"])
+        assert args.repair is True
+
+    def test_force_and_repair_can_coexist(self):
+        """--force --repair 不冲突 (force 实际优先, 但 CLI 允许同时传)."""
+        args = _parse_args_helper(["--types", "balance", "--force", "--repair"])
+        assert args.force is True
+        assert args.repair is True
+
+
+def _parse_args_helper(argv):
+    """parse_args 接 argv list, 不依赖 sys.argv (pytest 内 sys.argv 是 pytest 的)."""
+    import sys
+    old = sys.argv
+    try:
+        sys.argv = ["prog"] + argv
+        return mod.parse_args()
+    finally:
+        sys.argv = old