别让脏数据毁掉回测：用 Pandas + pytest 给 A 股 K 线做 12 项质量验收

原创

用户7083614

发布于 2026-09-05 14:17:41

很多回测的问题，不在策略，而在输入数据。

重复 K 线会把成交量算两遍；未复权价格会制造“暴跌”；把 UTC 当北京时间会让信号错位；停牌日被错误填充后，模型甚至会在不存在的行情上成交。更麻烦的是，这些问题通常不会让程序报错——代码照常运行，净值曲线也照常上涨。

这篇文章给出一套与数据供应商无关的验收方法：先把任意接口返回的数据统一成标准 DataFrame，再执行结构、价格、时间、覆盖率和可追溯性检查，最后用 pytest 把规则固化成回归测试。

本文只讨论数据工程，不构成投资建议。

一、先定义数据契约

不要让每个策略自行猜测字段含义。建议先规定最小 K 线结构：

字段	含义	建议类型
symbol	证券代码，包含交易所后缀	string
timestamp	K 线结束时间	datetime64ns, Asia/Shanghai
open/high/low/close	开高低收	float64
volume	成交量，单位必须固定	float64
amount	成交额，单位必须固定	float64
adjust	none/forward/backward	string
source	数据来源	string
fetched_at	实际获取时间	datetime

其中最容易被忽略的是三个字段：

adjust：不记录复权方式，就无法复现实验。
source：出现差异时，必须知道数据来自哪里。
fetched_at：历史数据可能被供应商修订，获取日期也是研究条件。

二、一个可复用的验证器

下面的代码不依赖具体 SDK。无论数据来自 REST API、CSV、数据库，还是开源库，只要先转换为上述结构，就能使用同一套检查。

from __future__ import annotations

from dataclasses import dataclass
from typing import Iterable

import numpy as np
import pandas as pd


REQUIRED_COLUMNS = {
    "symbol",
    "timestamp",
    "open",
    "high",
    "low",
    "close",
    "volume",
    "amount",
    "adjust",
    "source",
    "fetched_at",
}


@dataclass(frozen=True)
class DataIssue:
    code: str
    message: str
    rows: int = 0


def normalize_timestamp(
    values: pd.Series,
    *,
    source_timezone: str = "Asia/Shanghai",
) -> pd.Series:
    """统一转换为 Asia/Shanghai；禁止默默猜测时区。"""
    parsed = pd.to_datetime(values, errors="coerce")

    if parsed.dt.tz is None:
        parsed = parsed.dt.tz_localize(
            source_timezone,
            ambiguous="raise",
            nonexistent="raise",
        )

    return parsed.dt.tz_convert("Asia/Shanghai")


def validate_klines(
    frame: pd.DataFrame,
    *,
    expected_symbols: Iterable[str] | None = None,
    valid_trade_dates: set[pd.Timestamp] | None = None,
    intraday: bool = False,
) -> list[DataIssue]:
    issues: list[DataIssue] = []
    missing_columns = REQUIRED_COLUMNS - set(frame.columns)
    if missing_columns:
        return [
            DataIssue(
                "missing_columns",
                f"缺少字段: {sorted(missing_columns)}",
            )
        ]

    df = frame.copy()
    df["timestamp"] = normalize_timestamp(df["timestamp"])

    numeric = ["open", "high", "low", "close", "volume", "amount"]
    for column in numeric:
        df[column] = pd.to_numeric(df[column], errors="coerce")

    # 1. 空值与无穷值
    invalid_numeric = df[numeric].isna().any(axis=1) | np.isinf(
        df[numeric]
    ).any(axis=1)
    if invalid_numeric.any():
        issues.append(
            DataIssue("invalid_numeric", "价格或成交字段包含空值/无穷值", int(invalid_numeric.sum()))
        )

    invalid_timestamp = df["timestamp"].isna()
    if invalid_timestamp.any():
        issues.append(
            DataIssue("invalid_timestamp", "时间戳无法解析", int(invalid_timestamp.sum()))
        )

    # 2. 主键重复
    duplicated = df.duplicated(["symbol", "timestamp"], keep=False)
    if duplicated.any():
        issues.append(
            DataIssue("duplicate_bar", "同一证券同一时刻存在重复 K 线", int(duplicated.sum()))
        )

    # 3. OHLC 逻辑关系
    price_columns = df[["open", "high", "low", "close"]]
    bad_ohlc = (
        (df["high"] < price_columns.max(axis=1))
        | (df["low"] > price_columns.min(axis=1))
        | (df["low"] > df["high"])
    )
    if bad_ohlc.any():
        issues.append(
            DataIssue("invalid_ohlc", "OHLC 高低关系不成立", int(bad_ohlc.sum()))
        )

    # 4. 非法价格和成交量
    non_positive_price = (price_columns <= 0).any(axis=1)
    if non_positive_price.any():
        issues.append(
            DataIssue("non_positive_price", "价格小于等于 0", int(non_positive_price.sum()))
        )

    negative_trade = (df[["volume", "amount"]] < 0).any(axis=1)
    if negative_trade.any():
        issues.append(
            DataIssue("negative_trade", "成交量或成交额为负数", int(negative_trade.sum()))
        )

    # 5. 原始返回顺序检查；排序以后再检查会掩盖问题
    out_of_order = 0
    for _, group in df.groupby("symbol", sort=False):
        out_of_order += int((group["timestamp"].diff().dropna() <= pd.Timedelta(0)).sum())
    if out_of_order:
        issues.append(DataIssue("out_of_order", "时间序列重复或非严格递增", out_of_order))

    # 6. 标的覆盖率
    if expected_symbols is not None:
        expected = set(expected_symbols)
        returned = set(df["symbol"].dropna().astype(str))
        missing = sorted(expected - returned)
        unexpected = sorted(returned - expected)
        if missing:
            issues.append(DataIssue("missing_symbols", f"未返回标的: {missing[:20]}", len(missing)))
        if unexpected:
            issues.append(
                DataIssue("unexpected_symbols", f"返回了未请求标的: {unexpected[:20]}", len(unexpected))
            )

    # 7. 交易日检查：生产环境应传入交易所日历，不要只排除周末
    if valid_trade_dates is not None:
        trade_dates = df["timestamp"].dt.normalize().dt.tz_localize(None)
        invalid_date = ~trade_dates.isin(valid_trade_dates)
        if invalid_date.any():
            issues.append(
                DataIssue("non_trade_date", "数据落在交易所非交易日", int(invalid_date.sum()))
            )

    # 8. A 股分钟线时段检查
    if intraday:
        minutes = df["timestamp"].dt.hour * 60 + df["timestamp"].dt.minute
        in_morning = minutes.between(9 * 60 + 30, 11 * 60 + 30)
        in_afternoon = minutes.between(13 * 60, 15 * 60)
        outside_session = ~(in_morning | in_afternoon)
        if outside_session.any():
            issues.append(
                DataIssue("outside_session", "分钟 K 线落在常规交易时段之外", int(outside_session.sum()))
            )

    # 9. 元数据不能混用
    for column in ["adjust", "source"]:
        unique_count = df.groupby("symbol")[column].nunique(dropna=False)
        mixed = unique_count[unique_count > 1]
        if not mixed.empty:
            issues.append(
                DataIssue(
                    f"mixed_{column}",
                    f"同一证券混用了多个 {column}: {mixed.index.tolist()[:20]}",
                    len(mixed),
                )
            )

    return issues

三、为什么不能“发现缺口就自动向前填充”

时间序列里看到缺失值，很多人第一反应是：

df = df.ffill()

对行情数据来说，这可能制造不存在的价格：

股票上市前被填入价格；
停牌期间被当成可以成交；
午休被补成连续分钟线；
涨跌停时被假设能够按目标价成交；
不同市场节假日错位后出现伪数据。

更安全的顺序是：

用交易所日历生成预期时间轴；
区分停牌、未上市、接口失败和真正缺失；
保留缺失原因字段；
由策略明确决定是否允许填充；
成交模拟单独处理“有价格但不可成交”的情况。

缺失不是一种状态，而是多种业务事件的表面结果。

四、把规则写成 pytest 回归测试

只在 Notebook 里打印警告不够。数据接口或 SDK 升级后，旧问题可能再次出现。应把验收规则放进 CI。

import pandas as pd


def test_bad_bars_are_detected():
    frame = pd.DataFrame(
        {
            "symbol": ["600519.SH", "600519.SH", "600519.SH"],
            "timestamp": [
                "2026-09-01 09:31:00",
                "2026-09-01 09:32:00",
                "2026-09-01 09:32:00",  # 重复
            ],
            "open": [100.0, 101.0, 101.0],
            "high": [101.0, 100.0, 102.0],  # 第二行 high < open
            "low": [99.0, 100.0, 100.0],
            "close": [100.5, 100.5, 101.5],
            "volume": [1000, -1, 1200],      # 负成交量
            "amount": [100500, 100500, 121800],
            "adjust": ["none", "none", "none"],
            "source": ["demo", "demo", "demo"],
            "fetched_at": ["2026-09-01 16:00:00"] * 3,
        }
    )

    issues = validate_klines(frame, intraday=True)
    codes = {issue.code for issue in issues}

    assert "duplicate_bar" in codes
    assert "invalid_ohlc" in codes
    assert "negative_trade" in codes
    assert "out_of_order" in codes

真实项目还应增加几类固定样本：

除权除息日：比较不复权、前复权、后复权；
停牌与复牌日：验证缺失和成交约束；
新股上市日：防止上市前出现数据；
北交所、ETF、指数：验证代码和字段语义；
跨年和节假日：验证交易日历；
接口超时、429、5xx：验证重试后是否产生重复数据。

五、12 项上线检查清单

在数据进入回测或生产任务前，至少确认：

必需字段齐全；
时间戳可解析且时区明确；
symbol + timestamp 唯一；
OHLC 逻辑成立；
价格为正；
成交量和成交额非负；
每个标的时间严格递增；
请求标的与返回标的一致；
日期属于交易所日历；
分钟线位于正确交易时段；
复权方式和数据源没有在序列中混用；
数据来源、SDK 版本和获取时间可追溯。

这 12 项仍然只是底线。它们能发现结构性错误，却不能证明价格一定正确。对于关键研究，还需要抽样和第二个有授权的数据源交叉核对。

六、数据质量报告应和回测结果一起保存

建议每次任务同时输出：

run_id
strategy_version
data_source
data_version
adjustment
requested_symbols
returned_symbols
start_time / end_time
issue_counts
raw_data_hash
created_at

这样当净值异常或供应商修订历史数据时，才能回答：当时究竟使用了哪一版数据？

可复现回测不是“代码还能运行”，而是代码、参数、数据语义和输入版本都能被重新定位。

总结

量化研究最危险的数据问题，往往不会抛出异常。它只会悄悄改变信号、成交价格和最终收益。

可靠的流程应该是：

接口响应
→ 标准化 DataFrame
→ 数据契约验证
→ 交易日历与业务规则检查
→ 保存质量报告和原始数据哈希
→ 因子与回测

先验证数据，再讨论策略。相比继续增加指标和参数，这通常是提高回测可信度最划算的一步。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

pandas