做量化交易的人都知道回测系统的核心不是策略有多花哨而是数据有多可靠。如果历史行情数据本身就有问题那么再完美的回测结果也只是“垃圾进垃圾出”。本文从实战出发聊聊如何通过 API 批量拉取历史行情数据并做一套严谨的回测数据清洗流程。这些坑我都踩过。一、为什么历史行情数据这么难搞很多人以为历史行情就是“股票代码日期开高低收成交量”。真上手才发现问题一大堆不同数据源格式不同有的前复权、有的后复权、有的不复权停牌日、除权除息日、涨跌停板数据容易被忽略API 限流、断点续传、数据缺失需要处理国内 A 股、美股、期货的数据格式和规则差异巨大一个合格的量化回测系统必须能从源头保证数据的完整性、一致性、无偏性。二、批量拉取的工程设计2.1 基础思路不要一次性拉全部历史数据更不要写死日期。合理的设计应该是配置股票池 → 判断本地已有数据 → 只拉缺失区间 → 合并去重 → 校验一致性2.2 代码示例带断点续传的批量拉取下面使用 iTick API 获取历史日线数据前复权并实现本地缓存与断点续传。importrequestsimportpandasaspdimporttimefrompathlibimportPath API_TOKENyour_token_here# 替换为实际 TokenBASE_URLhttps://api.itick.orgdefbuild_headers():构造请求头包含 API Token 验证return{token:API_TOKEN,Content-Type:application/json}deffetch_stock_history(stock_code,regionHK,k_type8,start_date20000101,end_date20231231,cache_dir./data/raw): 带缓存的批量拉取自动断点续传 参数说明 stock_code : 股票代码港股示例00700 region : 市场代码HK/US/SZ/SH 等 k_type : K线类型8:日线9:周线10:月线 start_date : 开始日期格式 YYYYMMDD end_date : 结束日期格式 YYYYMMDD cache_dir : 本地缓存目录 Path(cache_dir).mkdir(parentsTrue,exist_okTrue)cache_filePath(cache_dir)/f{stock_code}.parquet# 已有数据则加载仅拉取缺失区间ifcache_file.exists():df_oldpd.read_parquet(cache_file)df_old[trade_date]pd.to_datetime(df_old[trade_date])last_datedf_old[trade_date].max()start_date(last_datepd.Timedelta(days1)).strftime(%Y%m%d)ifstart_dateend_date:returndf_oldprint(f{stock_code}: 本地已有数据至{last_date.date()}开始增量拉取...)else:df_oldpd.DataFrame()# 将日期范围转换为时间戳iTick kType 模式下需通过 et 参数控制截止start_tsint(pd.Timestamp(start_date).timestamp())end_tsint(pd.Timestamp(end_date).timestamp())all_data[]current_end_tsend_ts batch_days100# 每批最多拉取约 100 个交易日whileTrue:# 计算当前批次的起始截止区间基于天数回推batch_start_tsmax(start_ts,current_end_ts-batch_days*86400)params{region:region,code:stock_code,kType:k_type,limit:500,# 每次最多返回 500 根 K 线et:current_end_ts}try:urlf{BASE_URL}/stock/klineresprequests.get(url,headersbuild_headers(),paramsparams,timeout15)ifresp.status_code!200:print(f拉取失败:{stock_code}, 状态码{resp.status_code})time.sleep(2)continuedataresp.json()ifdata.get(code)0anddata.get(data):batch_datadata[data]all_data.extend(batch_data)print(f{stock_code}: 拉取到{len(batch_data)}条数据)# 判断是否还有更早的数据earliest_tsbatch_data[-1].get(t,0)ifbatch_dataelse0ifearliest_tsstart_tsorlen(batch_data)500:breakcurrent_end_tsearliest_ts-86400# 继续拉取更早数据else:print(f拉取失败:{stock_code}, 错误信息:{data.get(msg)})breaktime.sleep(0.5)# 限流控制exceptExceptionase:print(f拉取异常:{stock_code}, 错误:{e})time.sleep(5)continueifnotall_data:returndf_old# 数据转换与合并df_newpd.DataFrame(all_data)# 将时间戳转换为日期df_new[trade_date]pd.to_datetime(df_new[t],units)# 重命名字段为统一格式df_newdf_new.rename(columns{o:open,h:high,l:low,c:close,v:volume})df_newdf_new[[trade_date,open,high,low,close,volume]]df_combinedpd.concat([df_old,df_new],ignore_indexTrue)ifnotdf_old.emptyelsedf_new df_combineddf_combined.drop_duplicates(subset[trade_date]).sort_values(trade_date)df_combined.to_parquet(cache_file,indexFalse)print(f{stock_code}: 数据保存至{cache_file}, 共计{len(df_combined)}条)returndf_combined这个函数做了几件关键的事检查本地缓存Parquet 格式只拉取缺失区间通过limit和分批区间控制拉取量支持大量历史数据的自动分页异常重试与限流睡眠时间戳自动转换为标准化日期字段2.3 多股票并发拉取单线程循环拉取效率较低可使用线程池实现并发但仍需控制并发数以避免 API 限流fromconcurrent.futuresimportThreadPoolExecutor,as_completeddeffetch_batch(stock_list,regionHK,max_workers3): 批量拉取多只股票的历史数据 max_workers: 并发数建议 ≤ 5防止被限流 results{}withThreadPoolExecutor(max_workersmax_workers)asexecutor:futures{executor.submit(fetch_stock_history,code,region):codeforcodeinstock_list}forfutureinas_completed(futures):codefutures[future]try:results[code]future.result()print(f{code}: 拉取完成)exceptExceptionase:print(f{code}: 拉取失败, 错误:{e})returnresults并发数建议不超过5否则容易被数据源封禁。三、回测数据清洗 Checklist拉下来的原始数据离直接用于回测还差好几步。这是我总结的清洗流程每一步都不能省。3.1 时间轴处理# 确保交易日连续无跳空defalign_trading_days(df,trading_calendarNone):df[trade_date]pd.to_datetime(df[trade_date])dfdf.sort_values(trade_date).set_index(trade_date)iftrading_calendarisNone:# 生成完整日历工作日频率full_calendarpd.date_range(startdf.index.min(),enddf.index.max(),freqB)else:full_calendartrading_calendar dfdf.reindex(full_calendar)returndf用工作日频率freqB生成完整日历缺失日期会自动填入 NaN后续再填充或标记。3.2 除权除息与复权统一这是最大的坑很多新手直接用不复权数据做回测结果会发现某天价格突然跳空低开 30%实际上是除权策略却以为是大跌而错误开平仓。最佳实践全程使用前复权qfq数据保持历史价格连续可比。但要注意前复权会导致早期价格出现负数极端分红需要做截断处理# 剔除前复权后的负价格或极小价格dfdf[(df[close]0.01)(df[high]0.01)]3.3 涨跌停板标记回测时如果策略根据信号在涨停价买入实际根本无法成交。需要提前标记# 计算涨跌停价A股主板±10%科创/创业±20%港股无涨跌停板限制defcalc_limit_prices(df,stock_code):# 根据股票代码判断市场ifstock_code.startswith(688)orstock_code.startswith(300):limit_pct0.20# 科创板/创业板elifstock_code.startswith(600)orstock_code.startswith(000):limit_pct0.10# A股主板else:# 港股无涨跌停板限制直接返回df[is_limit_up]Falsedf[is_limit_down]Falsereturndf df[prev_close]df[close].shift(1)df[upper_limit]df[prev_close]*(1limit_pct)df[lower_limit]df[prev_close]*(1-limit_pct)# 标记一字板df[is_limit_up](df[open]df[upper_limit]-0.001)(df[close]df[upper_limit]-0.001)df[is_limit_down](df[open]df[lower_limit]0.001)(df[close]df[lower_limit]0.001)returndf回测执行时遇到is_limit_up且为买入信号应跳过或转换策略。3.4 停牌数据处理停牌期间没有成交不应填充为前一日价格会导致回测出现不合理收益。正确做法# 停牌日成交量应该为0或NaN不做前向填充df[volume]df[volume].fillna(0)# 对于价格字段停牌日保持NaN后续回测引擎遇到NaN应直接跳过该日3.5 数据对齐多股票回测多股票回测时需将所有股票对齐到同一个交易日历defalign_multi_stocks(stock_dfs,trading_days): stock_dfs: dict {code: DataFrame} trading_days: 交易日列表pd.DatetimeIndex aligned{}forcode,dfinstock_dfs.items():df_aligneddf.set_index(trade_date).reindex(trading_days)aligned[code]df_alignedreturnaligned四、数据质量校验清洗完毕后一定要跑一遍自动化校验defvalidate_data(df,stock_code):checks{是否有重复日期:df.index.duplicated().sum()0,是否有空价格:df[[open,high,low,close]].isna().any().any()False,最低价是否高于最高价:(df[low]df[high]).all(),成交量是否非负:(df[volume]0).all(),价格序列是否单调异常:((df[close]-df[close].shift(1)).abs()/df[close].shift(1)0.2).all(),# 除去涨跌停}forname,resultinchecks.items():print(f{stock_code}-{name}:{通过ifresultelse失败})returnall(checks.values())五、存储与版本管理建议格式强烈推荐Parquet或Feather比 CSV 快 10 倍以上且占用空间小。目录结构data/ raw/ # 原始API拉取数据按股票保存 cleaned/ # 清洗后数据已复权、对齐、填充 meta/ # 股票列表、交易日历、除权因子备份版本控制历史数据不要放 Git用DVCData Version Control或直接云存储S3、OSS。六、个人建议永远保留原始拉取数据清洗脚本可重复执行。否则哪天发现清洗逻辑错了你还得全部重拉。不要完美主义。回测数据做不到 100%精确但必须保证无偏性误差在买卖双方随机出现。先验小样本。对某只股票拉 3 年数据手动核对除权除息日、涨跌停日确信流程正确后再批量跑。备胎数据源。核心股票池至少准备两个数据源交叉验证。最后记住一句话回测是用来排除坏策略的不是用来证明好策略的。而这一切的起点就是靠谱的历史行情数据。希望这篇文章能帮你少走弯路。参考文档https://docs.itick.org/websocket/stocksGitHubhttps://github.com/itick-org/