网硕互联帮助中心本文你将得到什么：
痛点：策略亏钱，第一嫌疑人不是策略
很多人写完策略就急着回测，回测结果一难看就开始调参数。但更常见的真相是：你喂给回测引擎的数据本身就是坏的——停牌没剔除、除权没复权、接口偶发返回了几行脏数据，策略在垃圾上跑，参数调得再好也是白调。
量化圈的老话是"垃圾进，垃圾出"。数据清洗是这个系列的主题，而清洗的第一步不是洗，是体检：先搞清楚数据到底哪里脏，再决定怎么洗。
这篇先给你一套最小体检器：四项检查，五十行内搞定，跑一遍就知道你的 K 线有没有病。
环境准备
pip install mairui pandas numpy
第一步：拉数据，再"故意弄脏"它
体检器不能只报喜。为了验证它真的能抓到问题，我们先用确定性注入器往干净数据里造 4 类典型缺陷（可复现、可验证），再让体检器去抓：
- 缺 bar：删掉某天的记录（接口漏发、本地落库失败）；
- 重复行：同一天出现两行（重试导致的重复写入）；
- OHLC 逻辑错误：最高价低于最低价（字段错位、解析错误）；
- 价格跳变：单日 +35%（主板涨跌停只有 10%，这是典型的脏数据）。
from mairui import Client
import pandas as pd
import numpy as np
import random
client = Client("LICENCE-66D8-9F96-0C7F0FBCD073") # 演示证书，请换成你自己的正式证书
def wiggle(bars, seed):
"""对样例数据做确定性小幅扰动，模拟真实行情起伏（同种子同结果）。"""
rng = random.Random(seed)
out = []
for i, b in enumerate(bars):
k = 1.0 + (rng.random() – 0.5) * 0.10
nb = dict(b)
for f in ("o", "h", "l", "c"):
nb[f] = round(float(nb[f]) * k, 2)
out.append(nb)
return out
def inject_defects(bars):
"""确定性注入 4 类缺陷，顺序从尾到头避免索引漂移。"""
data = [dict(b) for b in bars]
data[40]["c"] = round(float(data[40]["c"]) * 1.35, 2) # 价格跳变
data[20]["h"], data[20]["l"] = data[20]["l"], data[20]["h"] # h < l
del data[30] # 缺 bar
data.insert(11, dict(data[10])) # 重复行
return data
SEED = 20260926 + 600519
bars = client.stock_history(code="600519", period="d", dividend="f")
bars = wiggle(bars, seed=SEED)
bad_bars = inject_defects(bars)
df = pd.DataFrame(bad_bars).rename(columns={"t": "date"}).sort_values("date").reset_index(drop=True)
for col in ("o", "h", "l", "c", "v"):
df[col] = df[col].astype(float)
说明：这里用的是公开演示证书，为演示方便明文写入；实际项目请换成你自己的正式证书（申请配置见 SDK 文档 https://github.com/MaiRuiApi/mairui ），换证后代码无需修改，注入器也只是验证体检器用的教学工具。
第二步：四项体检
def check_ohlc(df):
"""OHLC 逻辑校验：h>=l、h>=max(o,c)、l<=min(o,c)、价格全为正。"""
rules = {
"h<l": df["h"] < df["l"],
"h<max(o,c)": df["h"] < df[["o", "c"]].max(axis=1),
"l>min(o,c)": df["l"] > df[["o", "c"]].min(axis=1),
"价格<=0": (df[["o", "h", "l", "c"]] <= 0).any(axis=1),
}
mask = pd.Series(False, index=df.index)
hit_rule = pd.Series("", index=df.index, dtype=object)
for name, m in rules.items():
newly = m & ~mask
hit_rule[newly] = name
mask = mask | m
return df[mask].copy(), hit_rule[mask]
def check_duplicates(df):
return df[df["date"].duplicated(keep=False)]
def check_jump(df, threshold=0.11):
"""主板单日涨跌停 10%，留 1% 余量。ST 股与不同板块要单独调阈值。"""
ret = df["c"].pct_change().abs()
return df[ret > threshold].copy(), ret
def check_missing(df):
"""按工作日间隙检测缺 bar。1~3 个工作日=疑似缺 bar；>=4 疑似节假日窗口。"""
dates = pd.to_datetime(df["date"]).values.astype("datetime64[D]")
rows = []
for i in range(1, len(dates)):
gap = int(np.busday_count(dates[i – 1] + np.timedelta64(1, "D"), dates[i]))
if gap > 0:
kind = "疑似节假日窗口(需交易日历核对)" if gap >= 4 else "疑似缺bar"
rows.append({"prev_date": str(dates[i – 1]), "next_date": str(dates[i]),
"missing_busdays": gap, "judge": kind})
return pd.DataFrame(rows)
bad_ohlc, ohlc_rules = check_ohlc(df)
bad_dup = check_duplicates(df)
bad_jump, rets = check_jump(df)
bad_gap = check_missing(df)
report = pd.DataFrame({
"检查项": ["OHLC 逻辑", "重复行", "价格跳变", "工作日间隙"],
"检出数": [len(bad_ohlc), len(bad_dup), len(bad_jump), len(bad_gap)],
})
print(report.to_string(index=False))
report.to_csv("checkup_report.csv", index=False, encoding="utf-8-sig")
运行结果
输出示例（演示证样例数据 + 确定性注入，仅供演示）：
检查项 检出数
OHLC 逻辑 2
重复行 2
价格跳变 2
工作日间隙 2
明细：
- OHLC 违例 2 处：2025-10-21: 违反 h<l（o=10.27 h=10.26 l=10.34 c=10.3）；2025-11-18: 违反 h<max(o,c)（o=10.83 h=10.87 l=10.75 c=14.54）——第二处就是被注入 +35% 跳变的那行，close 14.54 高过了当天最高价，一条坏数据同时踩了两条规则；
- 重复行：2025-09-29 出现 2 次；
- 价格跳变 2 处：2025-11-18: 11.05 -> 14.54（+31.6%），以及次日回落 14.54 -> 11.71（+19.5%）——跳变会把相邻两天都标记出来；
- 工作日间隙 2 处：2025-09-30 -> 2025-10-09 缺 6 个工作日（疑似节假日窗口）（这是 2025 年国庆休市，正常）；2025-11-03 -> 2025-11-05 缺 1 个工作日（疑似缺bar）（这正是被删掉的那条）。
对照组：对未注入的原始数据跑同样的体检，OHLC/重复/跳变全部 0 检出，唯一提示就是国庆休市窗口——体检器没有把正常休市当缺 bar 误报。
结果怎么读（诚实的部分）
常见坑
小结与下篇预告
- 数据清洗的第一步是体检：OHLC 逻辑、重复行、价格跳变、缺 bar 四项检查 + 交易日历兜底；
- 体检器本身也要自检：用确定性注入器造坏数据验证检出能力，本文 4 类缺陷全检出、对照组零误伤；
- 体检报告能告诉你"哪里脏"，但脏数据怎么处理有讲究：极端值到底是该删还是该留？下一篇讲异常值检测与处理三法（3σ / IQR / 滚动中位数），以及"极端行情保留、数据错误剔除"的判断标准。
免责声明
本文所有内容仅用于技术学习与交流，不构成任何投资建议。股市有风险，投资需谨慎。
评论前必须登录！
注册