数据质量与校验：缺失、异常、口径

一根 K 线的最高价低于收盘价、某只股票缺失、成交量单位从股变成手，怎样在数据发布前发现，而不是等用户回测得出错误结论后才发现？

• 完整性：该有的数据是否都在，例如证券列表中的股票是否都有对应交易日数据。
• 唯一性：主键是否重复，例如同一代码、同一交易日是否出现两条日 K。
• 有效性：值是否在合法范围，例如价格大于 0、比例在约定范围内、日期格式正确。
• 内在一致性：字段之间的关系是否成立，例如 high 不低于 open 和 close，low 不高于它们。
• 准确性：与权威来源或第二来源是否一致，通常需要抽样或重点字段对账。
• 及时性：是否在承诺时间内到达，时间戳是否对应正确的交易日和时区。
• 口径一致：单位、复权方式、资产范围和字段含义是否前后一致。

规则不通过不等于数据必错。某个市场涨跌幅较大，可能是正常交易制度允许的结果。应把确定性错误设为“拦截”，把可疑但可能真实的结果设为“告警”，并保留复核依据。规则还要带生效日期，因为交易制度和字段口径可能变化。

接入：格式、字段、行数、重复、空值
  ↓
加工后：OHLC 关系、均价范围、跨日衔接、涨跌幅
  ↓
发布后：与另一来源对账、延迟监控、修正追踪

越早发现问题，修复成本越低；但只做接入检查还不够，业务加工后仍可能产生新问题。

import pandas as pd

df = pd.read_csv("bars.csv", dtype={"code": str})
for col in ["open", "high", "low", "close", "volume", "turnover"]:
    df[col] = pd.to_numeric(df[col], errors="coerce")

checks = {
    "duplicate_key": int(df.duplicated(["code", "trade_date"]).sum()),
    "missing_row": int(df.isna().any(axis=1).sum()),
    "bad_ohlc": int((
        (df.high < df[["open", "close"]].max(axis=1)) |
        (df.low > df[["open", "close"]].min(axis=1)) |
        (df[["open", "high", "low", "close"]] <= 0).any(axis=1)
    ).sum()),
}

active = df[df.volume > 0].copy()
active["vwap"] = active.turnover / active.volume
checks["vwap_outside_range"] = int(
    (~active["vwap"].between(active.low, active.high)).sum()
)
print(checks)

成交额 ÷ 成交量得到的均价应落在当日最低价和最高价之间。这条规则既能查异常，也能帮助发现“股”和“手”混用的单位错误。真实生产环境还要明确停牌、无成交、上市首日和不同板块规则等例外。

1. 只查字段类型和空值，不查 OHLC 关系、均价范围等业务关系。
2. 把所有异常都拦截，误删真实的极端行情。
3. 规则写死，不记录生效日期，制度变化后产生大量误报或漏报。
4. 只逐行检查，不和证券列表、交易日历等外部参照比较，因此发现不了整只股票缺失。
5. 只在入库时检查一次，不监控上游事后修正和发布后的对账结果。
6. 修正数据却不保留修正记录，导致用户无法解释历史结果为什么变化。

• 上交所股票上市规则（2025 年 4 月修订）
• 深交所投资者规则资料
• Great Expectations 文档
• Pandera 文档