12.7 案例二：量化回测前的数据完整性校验

面向经管学生、研究者与从业者的 AI 智能体设计教材

作者

李学恒、林建浩、严翊歆

发布于

2026-05-11

场景设定：你在做 A 股量化回测，数据源是本地 CSV 文件，包含日频行情（开高低收、成交量）和财务指标。回测脚本对数据格式有严格要求——缺失值、日期断档、列名不一致都会导致回测结果失真，甚至静默产出错误信号。

手动检查数据完整性既费时又容易遗漏。用一个 PreToolUse Hook，在回测脚本启动前自动校验数据文件，问题拦截在源头。

项目目录如下：

quant-backtest/
├── data/
│   ├── daily_prices.csv      # 日频行情数据
│   └── financial_ratios.csv  # 财务指标数据
├── strategies/
│   └── momentum.py           # 动量策略回测脚本
├── scripts/
│   └── validate_data.py      # 数据完整性校验脚本
├── outputs/                   # 回测结果输出
└── CLAUDE.md                  # 项目规则文件

validate_data.py 负责四项检查：CSV 文件是否存在且非空、关键列是否齐全、关键列是否存在缺失值、数据行数是否低于最低阈值。脚本以退出码报告结果：0 表示通过，非 0 表示存在问题，同时将具体错误输出到 stderr。

#!/usr/bin/env python3
"""回测数据完整性校验 - validate_data.py"""
import sys, os, csv, argparse

parser = argparse.ArgumentParser()
parser.add_argument("--data-dir", default="data/")
args = parser.parse_args()

REQUIRED_COLS = {"date", "open", "high", "low", "close", "volume"}
MIN_ROWS = 200  # 至少 200 个交易日
errors = []

if not os.path.isdir(args.data_dir):
    print(f"ERROR: 数据目录 {args.data_dir} 不存在", file=sys.stderr)
    sys.exit(1)

csv_files = [f for f in os.listdir(args.data_dir) if f.endswith(".csv")]
if not csv_files:
    print(f"ERROR: {args.data_dir} 中没有 CSV 文件", file=sys.stderr)
    sys.exit(1)

for fname in csv_files:
    path = os.path.join(args.data_dir, fname)

    with open(path, encoding="utf-8") as fh:
        reader = csv.DictReader(fh)
        cols = set(reader.fieldnames or [])

        # 检查必需列
        missing = REQUIRED_COLS - cols
        if missing:
            errors.append(f"{fname}: 缺少列 {missing}")
            continue

        rows = list(reader)

        # 检查数据量
        if len(rows) == 0:
            errors.append(f"{fname}: 数据为空")
            continue
        if len(rows) < MIN_ROWS:
            errors.append(f"{fname}: 仅 {len(rows)} 行，低于最低阈值 {MIN_ROWS}")

        # 检查关键列缺失值
        for i, row in enumerate(rows, start=2):
            for col in REQUIRED_COLS:
                if not row.get(col, "").strip():
                    errors.append(f"{fname} 第 {i} 行: {col} 列为空")
                    break  # 每行只报一次
            if len(errors) > 20:  # 错误过多时截断
                errors.append("... 错误过多，已截断")
                break

if errors:
    for e in errors:
        print(f"ERROR: {e}", file=sys.stderr)
    sys.exit(1)

print("数据校验通过")

在 .claude/settings.json 中配置 Hook：

{
  "hooks": {
    "PreToolUse": [
      {
        "matcher": "Bash",
        "hooks": [
          {
            "type": "command",
            "command": "python3 scripts/validate_data.py --data-dir data/"
          }
        ]
      }
    ]
  }
}

matcher 设为 Bash，表示每次 Claude Code 准备执行 Bash 命令时触发校验。校验脚本以非零退出码报告问题，Claude Code 会阻断后续操作——回测脚本不会在脏数据上运行。

实际使用时，直接向 Claude Code 发出回测指令：

▶ Claude Code

用 strategies/momentum.py 对 data/daily_prices.csv 做 2023 年全年的动量策略回测，输出净值曲线到 outputs/ 目录

Claude Code 准备执行 python3 strategies/momentum.py 时，Hook 先运行 validate_data.py。如果数据存在问题（例如 2023-06-15 至 2023-06-19 缺失四个交易日），Hook 阻断执行并将错误信息回灌给当前会话。模型收到反馈后，会先尝试修复数据缺口，再重新发起回测。

校验脚本设计原则

校验脚本应当只做判断、不做修复。修复逻辑交给模型根据上下文决定——是插值补全、跳过缺失区间，还是更换数据源，每次情况不同。Hook 的职责是拦截和报告，不是自作主张。

--- title: "12.7 案例二：量化回测前的数据完整性校验" --- 场景设定：你在做 A 股量化回测，数据源是本地 CSV 文件，包含日频行情（开高低收、成交量）和财务指标。回测脚本对数据格式有严格要求——缺失值、日期断档、列名不一致都会导致回测结果失真，甚至静默产出错误信号。手动检查数据完整性既费时又容易遗漏。用一个 `PreToolUse` Hook，在回测脚本启动前自动校验数据文件，问题拦截在源头。项目目录如下： ```txt quant-backtest/ ├── data/ │ ├── daily_prices.csv # 日频行情数据 │ └── financial_ratios.csv # 财务指标数据 ├── strategies/ │ └── momentum.py # 动量策略回测脚本 ├── scripts/ │ └── validate_data.py # 数据完整性校验脚本 ├── outputs/ # 回测结果输出 └── CLAUDE.md # 项目规则文件 ``` `validate_data.py` 负责四项检查：CSV 文件是否存在且非空、关键列是否齐全、关键列是否存在缺失值、数据行数是否低于最低阈值。脚本以退出码报告结果：0 表示通过，非 0 表示存在问题，同时将具体错误输出到 stderr。 ```python #!/usr/bin/env python3 """回测数据完整性校验 - validate_data.py""" import sys, os, csv, argparse parser = argparse.ArgumentParser() parser.add_argument("--data-dir", default="data/") args = parser.parse_args() REQUIRED_COLS = {"date", "open", "high", "low", "close", "volume"} MIN_ROWS = 200 # 至少 200 个交易日 errors = [] if not os.path.isdir(args.data_dir): print(f"ERROR: 数据目录 {args.data_dir} 不存在", file=sys.stderr) sys.exit(1) csv_files = [f for f in os.listdir(args.data_dir) if f.endswith(".csv")] if not csv_files: print(f"ERROR: {args.data_dir} 中没有 CSV 文件", file=sys.stderr) sys.exit(1) for fname in csv_files: path = os.path.join(args.data_dir, fname) with open(path, encoding="utf-8") as fh: reader = csv.DictReader(fh) cols = set(reader.fieldnames or []) # 检查必需列 missing = REQUIRED_COLS - cols if missing: errors.append(f"{fname}: 缺少列 {missing}") continue rows = list(reader) # 检查数据量 if len(rows) == 0: errors.append(f"{fname}: 数据为空") continue if len(rows) < MIN_ROWS: errors.append(f"{fname}: 仅 {len(rows)} 行，低于最低阈值 {MIN_ROWS}") # 检查关键列缺失值 for i, row in enumerate(rows, start=2): for col in REQUIRED_COLS: if not row.get(col, "").strip(): errors.append(f"{fname} 第 {i} 行: {col} 列为空") break # 每行只报一次 if len(errors) > 20: # 错误过多时截断 errors.append("... 错误过多，已截断") break if errors: for e in errors: print(f"ERROR: {e}", file=sys.stderr) sys.exit(1) print("数据校验通过") ``` 在 `.claude/settings.json` 中配置 Hook： ```json { "hooks": { "PreToolUse": [ { "matcher": "Bash", "hooks": [ { "type": "command", "command": "python3 scripts/validate_data.py --data-dir data/" } ] } ] } } ``` `matcher` 设为 `Bash`，表示每次 Claude Code 准备执行 Bash 命令时触发校验。校验脚本以非零退出码报告问题，Claude Code 会阻断后续操作——回测脚本不会在脏数据上运行。实际使用时，直接向 Claude Code 发出回测指令： ```opencode 用 strategies/momentum.py 对 data/daily_prices.csv 做 2023 年全年的动量策略回测，输出净值曲线到 outputs/ 目录 ``` Claude Code 准备执行 `python3 strategies/momentum.py` 时，Hook 先运行 `validate_data.py`。如果数据存在问题（例如 2023-06-15 至 2023-06-19 缺失四个交易日），Hook 阻断执行并将错误信息回灌给当前会话。模型收到反馈后，会先尝试修复数据缺口，再重新发起回测。 ::: {.callout-tip} ## 校验脚本设计原则校验脚本应当只做判断、不做修复。修复逻辑交给模型根据上下文决定——是插值补全、跳过缺失区间，还是更换数据源，每次情况不同。Hook 的职责是拦截和报告，不是自作主张。 :::

其他格式