本文记录一个使用 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
