Python 股票 K 线数据质量校验：字段、缺失值、重复行和价格异常

本文记录一个使用 QuantDash 获取股票 K 线数据后继续用 Pandas 做数据质量校验的示例。主要检查必要字段、缺失值、重复行、价格逻辑、成交量异常和交易日期顺序。QuantDash 是一个面向量化交易和金融数据分析的金融数据 API支持 A 股、美股、港股提供历史 K 线、实时行情、日内分时、五档盘口和标的基础信息。它提供 REST API 和 Python SDK返回结果可以接入 Pandas、回测系统、行情监控或数据入库流程。说明本文只是数据接口接入和数据质量校验示例不构成投资建议。实际使用前需要自行验证数据质量、复权方式、交易日历、延迟和权限限制。1. 适用场景这类校验流程适合使用 QuantDash 获取股票 K 线数据将行情数据转换为 Pandas DataFrame在回测、入库或行情监控前做数据验收检查字段缺失、重复数据和价格异常批量标的数据接入后做基础质量报告。2. 安装和初始化安装 SDKpip install quantdash导入并初始化客户端from quantdash import QuantDash qd QuantDash(api_keyyour-key)实际使用时把your-key替换为自己的 API Key。3. 获取 K 线数据下面以600519.SH为例获取日 K 线from quantdash import QuantDash qd QuantDash(api_keyyour-key) df qd.klines.get( 600519.SH, period1d, to_dataframeTrue ) 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_columnsREQUIRED_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(subsetkeys, keepFalse) return { duplicated_rows: int(duplicated.sum()), sample: df[duplicated].head(10) } print(check_duplicates(df))如果只获取单只股票并且数据里没有symbol字段也可以按trade_date检查。7. 检查价格逻辑K 线价格至少应满足几个基本规则open、high、low、close应大于 0high不应小于lowopen应在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(-, , regexFalse) 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