本文记录一个使用 QuantDash 获取股票 K 线数据后，继续用 Pandas 做数据质量校验的示例。主要检查必要字段、缺失值、重复行、价格逻辑、成交量异常和交易日期顺序。

QuantDash 是一个面向量化交易和金融数据分析的金融数据 API，支持 A 股、美股、港股，提供历史 K 线、实时行情、日内分时、五档盘口和标的基础信息。它提供 REST API 和 Python SDK，返回结果可以接入 Pandas、回测系统、行情监控或数据入库流程。

说明：本文只是数据接口接入和数据质量校验示例，不构成投资建议。实际使用前需要自行验证数据质量、复权方式、交易日历、延迟和权限限制。

1. 适用场景

这类校验流程适合：

• 使用 QuantDash 获取股票 K 线数据；

• 将行情数据转换为 Pandas DataFrame；

• 在回测、入库或行情监控前做数据验收；

• 检查字段缺失、重复数据和价格异常；

• 批量标的数据接入后做基础质量报告。

2. 安装和初始化

安装 SDK：

pip install quantdash

导入并初始化客户端：

from quantdash import QuantDash

qd = QuantDash(api_key="your-key")

实际使用时，把 your-key 替换为自己的 API Key。

3. 获取 K 线数据

下面以 600519.SH 为例获取日 K 线：

from quantdash import QuantDash

qd = QuantDash(api_key="your-key")

df = qd.klines.get(
    "600519.SH",
    period="1d",
    to_dataframe=True
)

print(df.head())
print(df.columns)

后续示例假设数据里包含这些字段：

symbol, trade_date, open, high, low, close, volume

如果实际字段名不同，可以先做一层字段映射，再进入校验流程。

4. 检查必要字段

先确认后续分析依赖的字段是否存在：

REQUIRED_COLUMNS = [
    "trade_date", "open", "high", "low", "close", "volume"
]

def check_required_columns(df, required_columns=REQUIRED_COLUMNS):
    missing = [col for col in required_columns if col not in df.columns]
    return {
        "passed": len(missing) == 0,
        "missing_columns": missing
    }

print(check_required_columns(df))

字段缺失时，不建议继续计算指标或进入回测。

5. 检查缺失值

def check_missing_values(df):
    missing_count = df.isna().sum()
    missing_rate = df.isna().mean()
    return {
        "missing_count": missing_count.to_dict(),
        "missing_rate": missing_rate.to_dict()
    }

print(check_missing_values(df))

缺失值不一定都代表错误。比如停牌、无成交或非交易时段，都可能导致部分字段为空。这里先统计缺失情况，后续再结合业务规则处理。

6. 检查重复数据

对日 K 线来说，同一个标的在同一个交易日通常只应有一条记录。

def check_duplicates(df, keys=("symbol", "trade_date")):
    keys = [key for key in keys if key in df.columns]
    duplicated = df.duplicated(subset=keys, keep=False)
    return {
        "duplicated_rows": int(duplicated.sum()),
        "sample": df[duplicated].head(10)
    }

print(check_duplicates(df))

如果只获取单只股票，并且数据里没有 symbol 字段，也可以按 trade_date 检查。

7. 检查价格逻辑

K 线价格至少应满足几个基本规则：

• open、high、low、close 应大于 0；

• high 不应小于 low；

• open 应在 low 和 high 之间；

• close 应在 low 和 high 之间。

def check_price_logic(df):
    errors = {}

    price_cols = ["open", "high", "low", "close"]
    for col in price_cols:
        if col in df.columns:
            errors[f"{col}_non_positive"] = int((df[col] <= 0).sum())

    if {"high", "low"}.issubset(df.columns):
        errors["high_lt_low"] = int((df["high"] < df["low"]).sum())

    if {"open", "high", "low"}.issubset(df.columns):
        errors["open_out_of_range"] = int(
            ((df["open"] > df["high"]) | (df["open"] < df["low"])).sum()
        )

    if {"close", "high", "low"}.issubset(df.columns):
        errors["close_out_of_range"] = int(
            ((df["close"] > df["high"]) | (df["close"] < df["low"])).sum()
        )

    return errors

print(check_price_logic(df))

这个函数可以快速发现明显异常，比如价格为 0、最高价低于最低价等。

8. 检查成交量

def check_volume(df):
    if "volume" not in df.columns:
        return {"volume_missing": True}

    return {
        "negative_volume": int((df["volume"] < 0).sum()),
        "zero_volume": int((df["volume"] == 0).sum())
    }

print(check_volume(df))

成交量为 0 不一定是错误，需要结合停牌、无成交、市场类型和数据频率判断。

9. 检查日期顺序

交易日不能简单按自然日判断，因为节假日和周末都不是交易日。这里先做一个基础检查：日期是否倒序、唯一日期数量是否合理。

def check_trade_date_order(df):
    data = df.copy()
    data["trade_date"] = data["trade_date"].astype(str)
    data["trade_date_parsed"] = data["trade_date"].str.replace("-", "", regex=False)

    sorted_dates = data["trade_date_parsed"].sort_values()
    is_sorted = list(data["trade_date_parsed"]) == list(sorted_dates)

    return {
        "rows": len(data),
        "unique_dates": data["trade_date_parsed"].nunique(),
        "is_sorted": is_sorted
    }

print(check_trade_date_order(df))

如果要做更严格的检查，可以再接入正式交易日历，对缺失交易日进行比对。

10. 生成质量报告

把前面的检查组合起来：

def build_quality_report(df):
    return {
        "required_columns": check_required_columns(df),
        "missing_values": check_missing_values(df),
        "duplicates": check_duplicates(df),
        "price_logic": check_price_logic(df),
        "volume": check_volume(df),
        "trade_date_order": check_trade_date_order(df)
    }

report = build_quality_report(df)

for name, result in report.items():
    print("==", name, "==")
    print(result)

如果是批量标的数据，可以按 symbol 分组生成报告：

def build_reports_by_symbol(df):
    reports = {}
    for symbol, group in df.groupby("symbol"):
        reports[symbol] = build_quality_report(group)
    return reports

这样可以定位是哪只标的出现字段缺失、日期断档或价格异常。

11. 小结

使用 QuantDash 获取行情数据后，不建议直接进入策略或回测。更稳的流程是：

QuantDash API -> DataFrame -> 数据质量校验 -> 清洗/入库 -> 分析/回测

基础校验至少应覆盖：

• 必要字段；

• 缺失值；

• 重复行；

• 价格逻辑；

• 成交量异常；

• 日期顺序；

• 批量标的分组报告。

数据质量问题往往比模型问题更隐蔽。把这些校验函数沉淀下来，后面无论做回测、行情监控还是数据入库，都会更容易排查问题。

相关链接：

• 参考文档：QuantDash 简介 - QuantDash