如何给 A 股 K 线数据做清洗与异常检测？（停牌、跳空、缺口）

原创

用户7083614

发布于 2026-09-05 20:21:35

230

一句话回答： 回测跑偏，很多时候不是策略错，而是数据脏——停牌、异常跳空、缺失交易日没处理。用 AlphaFeed 取 K 线后，用 pandas 做几项标准校验（零成交=停牌、超阈值跳空=疑似除权/异动、日期缺口、OHLC 逻辑），就能在回测前把数据体检一遍。

为什么必须先清洗

停牌：成交量为 0 的"K 线"不是真交易，直接算收益率会误导。
除权跳空：不复权数据在分红送股日暴跌，会被策略当成"大跌信号"。
缺失交易日：数据源漏了某天，滚动窗口和对齐就会错位。
幸存者偏差：只有还在交易的票、剔除了退市股，回测收益会系统性偏高。

准备数据

from alphafeed import AlphaFeed
import pandas as pd, numpy as np

af = AlphaFeed()  # 读取 ALPHAFEED_API_KEY

# 用不复权价做"停牌/跳空"体检（复权会抹平除权跳空，掩盖问题）
df = af.klines.get("600519.SH", period="1d", count=250, adjust="none", to_dataframe=True)
df = df.sort_values("trade_date").reset_index(drop=True)

检查 1：停牌（零成交 / 零振幅）

suspended = df[(df["volume"] == 0) | (df["amount"] == 0)]
print(f"疑似停牌日: {len(suspended)}")
# 也可用 高=低=开=收 判断一字/无成交
flat = df[(df["high"] == df["low"])]
print(f"一字/无波动日: {len(flat)}")

停牌日的处理原则：不要用它算收益率；对齐多标的时按实际交易日 inner join，不要向前/向后无脑填充。

检查 2：异常跳空（疑似除权或真实异动）

用日间收益率的绝对值超阈值来标记（A 股非 ST 单日涨跌停约 ±10%，超过多半是除权或数据问题）：

df["ret"] = df["close"].pct_change()
JUMP = 0.11  # 超过 11% 视为异常（普通股票单日涨跌停 ±10%）
anomalies = df[df["ret"].abs() > JUMP]
print("异常跳空日：")
print(anomalies[["trade_date", "close", "ret"]].to_string(index=False))

判断跳空是"除权"还是"真异动"：查该日是否有除权因子变化。

fac = af.klines.ex_factors(["600519.SH"], to_dataframe=True)
ex_dates = set(fac["trade_date"])
anomalies["疑似除权"] = anomalies["trade_date"].isin(ex_dates)
print(anomalies[["trade_date", "ret", "疑似除权"]].to_string(index=False))

若 疑似除权=True，说明是分红送股导致的跳空，用前复权即可消除；若为 False 才是真实异动或数据问题。

检查 3：缺失交易日 / 重复日期

# 重复日期
dups = df[df["trade_date"].duplicated()]
print(f"重复日期: {len(dups)}")

# 交易日连续性（粗检：相邻日期间隔 > 7 天可能有缺口或长假）
d = pd.to_datetime(df["trade_date"])
gaps = df.assign(gap=(d - d.shift()).dt.days)
big_gaps = gaps[gaps["gap"] > 7][["trade_date", "gap"]]
print("较大日期间隔（长假或缺口）：")
print(big_gaps.to_string(index=False))

春节/国庆长假会有正常的大间隔；重点排查非假期的异常缺口。

检查 4：OHLC 逻辑一致性

高价应≥开/收/低，低价应≤开/收/高，价格应为正：

bad_ohlc = df[
    (df["high"] < df[["open", "close", "low"]].max(axis=1)) |
    (df["low"]  > df[["open", "close", "high"]].min(axis=1)) |
    (df[["open","high","low","close"]] <= 0).any(axis=1)
]
print(f"OHLC 逻辑异常行: {len(bad_ohlc)}")

封装成一个体检函数

def health_check(symbol, count=250):
    af = AlphaFeed()
    df = af.klines.get(symbol, period="1d", count=count, adjust="none", to_dataframe=True)
    df = df.sort_values("trade_date").reset_index(drop=True)
    df["ret"] = df["close"].pct_change()
    d = pd.to_datetime(df["trade_date"])
    report = {
        "symbol": symbol,
        "rows": len(df),
        "suspended": int(((df["volume"] == 0) | (df["amount"] == 0)).sum()),
        "jumps_gt_11pct": int((df["ret"].abs() > 0.11).sum()),
        "dup_dates": int(df["trade_date"].duplicated().sum()),
        "gaps_gt_7d": int(((d - d.shift()).dt.days > 7).sum()),
    }
    return report

for s in ["600519.SH", "000001.SZ", "601318.SH"]:
    print(health_check(s))

批量体检 + 剔除问题标的

symbols = af.quotes.get(universes="CN_Stock", to_dataframe=True)["symbol"].tolist()[:100]
reports = [health_check(s) for s in symbols]
bad = [r for r in reports if r["suspended"] > 5 or r["dup_dates"] > 0]
print(f"需要人工复核的标的: {len(bad)}")

清洗结论速查

问题	检测	处理
停牌	volume/amount = 0	不算收益；对齐用 inner join
除权跳空	\|ret\|>11% 且在除权因子日	用前复权消除
真实异动/脏数据	\|ret\|>11% 且非除权日	人工复核
缺口	非假期日期间隔大	补数据或标记
退市股缺失	全市场池只含在市	回测需单独保留退市股历史

常见问题（FAQ）

Q：体检该用复权还是不复权？

A：查"停牌/真实跳空"用不复权（能看到原始跳空）；算收益率/回测用前复权。

Q：怎么判断是除权还是异动？

A：用 af.klines.ex_factors 拿除权日，跳空日若命中即除权。

Q：如何避免幸存者偏差？

A：全市场池默认只含在市标的；严谨回测要单独保留退市股的历史序列。

小结

干净的数据是可信回测的前提。用 AlphaFeed 的 K 线 + 除权因子，几段 pandas 就能把停牌、跳空、缺口、OHLC 异常一次性体检出来，并区分"除权"与"真异动"。数据先过体检，回测结论才靠得住。

参考

AlphaFeed Python SDK 文档：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart

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

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

python

pandas

#量化交易

#股票数据