做A股量化回测的第一步从来不是选股策略而是先搞到一份干净、完整、可持续更新的历史行情数据。2019年我刚开始折腾量化的时候市面上的免费数据源就那么几个每一个都藏着不少坑。几年下来我把自己这套“A股历史数据免费下载”的方案反复打磨现在4000多只个股实际A股已经超过5000家了的日线数据可以从零开始一键打包下载每天收盘后还能自动增量更新。这篇文章把完整方案和踩坑记录都写出来希望能让想上车的朋友少走弯路也顺便聊聊那些免费数据源背后最容易被忽略的质量问题。如果你是刚开始做量化回测、学术研究或者单纯想自己维护一份本地行情数据库这篇文章都适合。我会尽量把每个选择的理由讲透而不是只丢给你一段能跑的代码。1. 免费数据源横评为什么我选择“主力辅助备援”的组合拳很多新手一上来就搜“A股数据免费下载”结果发现市面上的方案鱼龙混杂。有的要注册拿token有的接口用两次就被限流有的数据质量堪忧。我这些年把所有主流免费数据源都试了一遍先说结论没有一个数据源是完美的所以你要做的是组合使用而不是单押某一家。1.1 四个主流免费数据源的真实差异先放一个我根据自己的使用经验整理的对比表后面再逐个拆开讲数据源注册门槛稳定性日线字段完整度并发友好度最适合的场景TuShare要token积分高全中长期研究主力Baostock无需注册高全中个人全量下载主力AkShare无需注册中全低辅助信息、实时快照efinance无需注册中全低备援TuShare是老牌数据接口文档完善数据质量高社区也很成熟。但它需要注册账号获取token而且关键接口基本都有积分门槛免费用户想用全量历史K线这类Pro接口得靠持续使用攒积分慢慢解锁。我不能说它不好只是“免费无门槛”这两个条件它都满足不了。如果你愿意花两三个月养积分把它当主力数据源是完全可行的。Baostock是我实际下载全量用下来最顺手的一个也是我在这套方案里的主力数据源。它无需注册调用前直接bs.login()就能登录没有token和积分的概念。覆盖沪深A股/B股、指数、部分港股美股日线/周线/月线/分钟线都有字段也算全。关键是它对个人用户完全免费而且数据更新基本能在当天下午5点左右到位。它唯一的短板是并发太高时连接会不稳定这个后面我会专门讲。AkShare最大的优势是接口极其丰富除了行情还有财报、资金流、龙虎榜等等基本能从一个库里拿到你想要的绝大多数数据。但它本质上是把各大财经网站的公开接口做了一层规范化封装所以上游网页一改版接口就可能直接报错。它很适合用来拿股票列表、交易日历这类“辅助信息”但我不建议拿它做全量历史数据的下载主力——连续大量请求容易被限流而且接口失效的坑会让你排查到怀疑人生。efinance是一个相对小众的开源库性能不错但社区维护力度不稳定接口变动时找不到太多解决方案。我把它放在备援位置偶尔用来交叉验证数据。1.2 我的选型策略与具体分工我现在的方案是三路配合主力下载用 Baostock免费、无token、数据格式规范适合全量历史日线数据的批量拉取。辅助信息用 AkShare用它获取全市场股票代码列表、交易日历甚至每天的实时行情快照做参照。数据校验用腾讯/新浪的公开历史行情接口不需要专门造轮子只要在抽样校验时从第三方渠道交叉比对某几只股票的收盘价确认主力数据源没有出现系统性偏差就够了。这样做的好处是任何一个数据源出问题我都还有另一条路可以快速顶上。比如去年AkShare因为上游页面改版导致股票列表接口报错我直接用Baostock的query_all_stock拿代码池完全没影响下载流程。这种“互备”的思路做数据工程的人应该都懂。另外提醒一句免费数据源即便允许个人免费使用如果你要做商业发布或者对外提供数据服务务必先确认对应数据源的授权条款。个人研究和学术用途一般没问题但商用就是另一回事了。2. 一键打包的流水线设计从代码池到标准落地做全市场的一键打包核心不是“写一个下载函数”而是设计一条完整流水线。我的流水线分四步拿到全市场股票代码池 → 并发拉取历史日线 → 清洗校验 → 标准化存储。任何一步设计得不好后面都会还债。2.1 全市场股票代码池怎么拿代码池是整个下载流程的起点。没有一份完整的股票列表后面所有并发都是空谈。而且这份列表必须随着时间更新因为每年都有新股上市、老股退市。我推荐两种方法获取代码池方法一用Baostock的query_all_stock(day某个交易日)拿到当天沪深所有证券代码然后过滤出纯A股import baostock as bs bs.login() rs bs.query_all_stock(day2024-12-20) print(rs.fields) # 确认返回字段 all_stocks [] while rs.next(): all_stocks.append(rs.get_row_data()) bs.logout()这个方法的问题是返回结果里包含指数、基金等各种品种需要你自己按代码前缀过滤。过滤逻辑大概是沪市主板sh.60科创板sh.68深市主板sz.00创业板sz.30北交所bj.8/4/92。如果你只是想快速拿到“当前全部A股”我更推荐方法二。方法二用AkShare的ak.stock_info_a_code_name()直接返回干净的A股代码和名称列表import akshare as ak df_code ak.stock_info_a_code_name() print(df_code.head()) # 不同版本列名可能不同先打印确认 codes df_code[code].tolist()这个接口返回的code是纯数字字符串比如600000、000001。实际调用Baostock下载时需要按市场前缀转换成sh.600000、sz.000001这种格式。转换逻辑写一块放在代码池获取之后def normalize_to_bs_code(code): code str(code).zfill(6) if code.startswith(6): return fsh.{code} elif code.startswith((0, 3)): return fsz.{code} else: return fbj.{code}A股代码段其实还在微调比如沪深主板整合后新代码段会变所以这段映射逻辑建议不要写死在代码里而是留成配置方便以后增补。代码池不要每次下载都重新拉建议存成JSON缓存起来每周更新一次就够了。这样既能减少对外部接口的依赖也能保证下载脚本能离线跑。2.2 日线数据该下哪些字段单位怎么统一很多人在下载时习惯“全字段一把梭”我建议按需取字段但有几个字段是必须的字段含义为什么必须date交易日期时间索引code股票代码主键之一open/high/low/close开高低收行情四价一切技术指标的基础preclose昨收盘计算涨跌幅、识别除权除息日volume成交量流动性研究必用amount成交额资金流向研究必用turn换手率流动性因子pctChg涨跌幅收益序列直接来源isST是否ST样本筛选时常用用Baostock拉日线时对应接口和字段名是query_history_k_data_plus。完整调用示例后面第四章会给这里先强调一个坑不同数据源的volume单位不一样。有的按“股”有的按“手”。Baostock返回的volume单位是股amount单位是元。如果你同时用了多个数据源做交叉验证务必要先统一单位否则后面计算成交量因子时会出现数量级的错误。2.3 存储层选型CSV、Parquet、SQLite别一上来就上库4000多只股票每只存一个文件。我最早图省事把所有股票塞进一个超大CSV里结果每次读取要等十几秒后来改成每只股票单独存一个小文件速度立刻上来了。下面是几种存储方案的测试对比具体数字会因字段和计算环境略有浮动但量级就是这样存储方案全市场日线约15年估算体积优点缺点单CSV2-4GB通用、Excel能开读取慢定位单票全表扫每票一个CSV2-4GB单票读取快、易排查文件数量多Parquet500MB-1GB读取快、压缩率高依赖pyarrow不便直接查看SQLite1-2GB可SQL查询写入稍慢不如Parquet快对个人研究来说我建议先用“每票一个CSV”。理由很实在交易数据本质是“按股票代码分片”绝大多数场景都是先取某只股票的时间序列再拼接横截面。按票分文件正好匹配这种访问模式而且万一某只股票的数据坏了你只需要重新下载这一只不用全量重来。如果你做因子回测时发现批量读取太慢再把高频率读取的字段转成Parquet或者用pyarrow批量读取同一目录下的所有CSV速度会成倍提升。SQLite适合你已经把数据做成了宽表、想用SQL直接查询去重后的事件数据但作为原始行情存储它的效率不如按票分文件。3. 复权与数据质量陷阱免费数据最容易埋雷的地方如果说代码和存储只是体力活那数据处理阶段的“复权”和“质量清洗”才是真正的分水岭。很多新手拿到数据就开跑结果收益率算出来完全是错的问题多半出在除权除息上。3.1 除权除息如何扭曲你的收益率计算先看一个最简单的例子。某股票昨天收盘价10元今天每股分红0.5元。除息日交易所会把参考价自动调整为9.5元。如果你直接用10元和9.5元算涨跌幅会得到“-5%”的结论但你的总资产并没有减少——你手里多了一笔0.5元的现金分红。也就是说除权除息造成的价格跳空不是市场交易行为而是制度性的价格修正。这种“假跳空”如果不处理收益率序列就会在分红季出现大量虚假的下跌你的策略回测结果会一塌糊涂。送股、转增、配股同样会造成这个问题而且调整公式更复杂。3.2 前复权、后复权、不复权回测到底该用哪个复权分三种很多资料讲得绕我用自己的话概括一下不复权保留真实成交价包含除权跳空。适合看真实盘口、做事件研究。前复权以当前价格为锚把历史价格整体往下调。好处是最近的价格保持在真实价位附近画K线最直观坏处是每次有新数据整个历史序列都会整体变动不适合做跨期比较和长期回测。后复权以上市首日或某个最早基准日的价格为锚把后面的价格往上调。历史序列是固定的不会因为新数据加入而变化适合计算长期收益和做定量回测。我自己的原则很简单回测用后复权看盘用前复权研究分红送配时用不复权复权因子。如果你不想纠结下载时直接用Baostock的adjustflag1拿后复权数据能应付绝大多数回测场景。这里要特别提醒不要自己手搓复权因子除非你真的在做分红事件研究。自己算需要精确的送转比例、分红金额、配股价等信息任何一个数据源在这些字段上口径出错你算出来的因子就会歪掉。老老实实用数据源提供的复权结果然后把精力放在真正影响你策略的数据校验上。3.3 停牌、退市、新股最容易被忽视的样本偏差除了复权数据质量还有三座大山。第一座是停牌。停牌期间没有行情记录所以下载回来的数据天生就是断的。做时间序列分析时你要用交易日历做笛卡尔积把缺失的日期补出来价格填NaN。Baostock 提供了query_trade_dates可以拿全市场交易日列表先把这个列表和个股数据做个对齐。不补齐会怎样很多时间窗计算会把停牌日误当成“没有数据”而跳过导致窗口错位。第二座是新股。注册制下新股上市首日不设涨跌幅限制部分策略在清洗时会把首日数据剔除避免极端波动影响统计结果。你要在代码里明确记录“这个标的是否包含上市首日”至少不要模模糊糊把首日混在正常交易里。第三座也是最坑的——幸存者偏差survivorship bias。假设你在2025年做历史回测用“当前存活的股票列表”去拉历史数据那么已经退市的股票就永远消失在你的样本里了。退市股往往经历了股价崩塌把它们排除掉你的回测结果会显得比真实市场好得多。严谨的做法是用“历史时点的股票列表”或“包含退市股的数据快照”做无偏样本。Baostock 虽然对已退市股票的数据支持不如对存续股那么稳定但早期数据一般还能查到建议你在下载代码池时就把历史上出现过的退市股也纳入考虑。4. 4000只股票的并发下载完整的可运行代码前面讲的都是设计思路这一章直接上实战代码。我的环境是Python 3.10以上版本依赖库为baostock、pandas、akshare、tqdm装好之后按下面的顺序执行即可。4.1 核心下载函数与线程池实现核心下载函数我用Baostock的query_history_k_data_plus按“后复权”方式拉数据这样拿回来就能直接用于回测省去手动折算的麻烦import baostock as bs import pandas as pd import time import random def download_one(code, start1990-01-01, end2024-12-31, retry3): 下载单只股票的后复权日线数据带失败重试。 code 格式sh.600000 / sz.000001 for attempt in range(retry): try: rs bs.query_history_k_data_plus( code, date,code,open,high,low,close,preclose,volume,amount,turn,tradestatus,pctChg,isST, start_datestart, end_dateend, frequencyd, adjustflag1, # 1后复权2前复权3不复权 ) rows [] while (rs.error_code 0) and rs.next(): rows.append(rs.get_row_data()) df pd.DataFrame(rows, columnsrs.fields) return df except Exception as e: wait_time 2 ** attempt random.uniform(0, 1) print(f{code} 第{attempt1}次失败: {e}, {wait_time:.1f}秒后重试) time.sleep(wait_time) return None然后是多线程并发下载import os from concurrent.futures import ThreadPoolExecutor, as_completed from tqdm import tqdm SAVE_DIR ./a_share_daily os.makedirs(SAVE_DIR, exist_okTrue) def download_all(stock_codes, max_workers20): failed [] with ThreadPoolExecutor(max_workersmax_workers) as executor: future_map { executor.submit(download_one, code): code for code in stock_codes } for future in tqdm(as_completed(future_map), totallen(stock_codes), desc下载中): code future_map[future] try: df future.result() if df is not None and not df.empty: df.to_csv(os.path.join(SAVE_DIR, f{code}.csv), indexFalse) else: failed.append(code) except Exception: failed.append(code) return failed这里有一个非常重要的经验Baostock 的并发数不要开太高。我实测过20个并发稳定30个偶尔报连接错误50个以上错误率明显上升。全市场4000多只股票即使每只耗时0.3-0.5秒20并发跑完也只需要3-5分钟完全没必要为了省时间把并发加到极限。如果你想更稳一点可以每个线程单独调用一次bs.login()代价是更慢一些。4.2 断点续传与失败重试机制全量下载最怕中途断网或程序崩溃。如果直接从零开始重新跑前面几小时的进度全废了。所以一定要做断点续传第二次运行的时候跳过本地已经存在的文件只下缺失的。def get_pending_codes(stock_codes, save_dirSAVE_DIR): pending [] for code in stock_codes: path os.path.join(save_dir, f{code}.csv) if not os.path.exists(path) or os.path.getsize(path) 1024: pending.append(code) return pending然后在主流程里加一个循环第一次跑完把failed列表里的代码再次调用download_all最多重试两轮。如果重试之后仍有失败就把失败代码写进failed_codes.json人工去检查是网络问题还是数据源本身对该股票不覆盖。if __name__ __main__: bs.login() stock_codes load_stock_pool(stock_pool.json) # 上一步生成 pending get_pending_codes(stock_codes) print(f待下载股票数: {len(pending)}) failed download_all(pending) for round_no in range(2): if not failed: break print(f第{round_no1}轮重试剩余{len(failed)}只) time.sleep(5) failed download_all(failed) import json with open(failed_codes.json, w) as f: json.dump(failed, f, ensure_asciiFalse, indent2) bs.logout()4.3 数据完整性校验清单下载完不等于结束还要校验。我一般在全量下载后跑一个基础校验脚本检查四件事文件行数是否大于0日期是否递增且没有重复四价是否满足high max(open, close)且low min(open, close)关键字段是否有明显异常值例如价格出现0或负数。def validate_csv(code, path): try: df pd.read_csv(path) if df.empty: return code, 空文件 df[date] pd.to_datetime(df[date]) if not df[date].is_monotonic_increasing: return code, 日期乱序 if df[date].duplicated().any(): return code, 日期重复 if (df[high] df[low]).all() False: return code, 最高价低于最低价 if (df[[open, high, low, close]].min().min()) 0: return code, 存在非正价格 except Exception as e: return code, f解析失败: {e} return code, OK校验报告会告诉我哪些股票有问题修复方式一般是“删除对应文件后重新下载该股”或者结合另一个数据源做交叉核对。5. 增量更新机制让数据每天自动保持新鲜全量下载解决的是“从无到有”但数据工程真正考验人的是“从有到新”。每天收盘后把当天数据追加进本地库这步做不好前面全量下载的成果会越来越陈旧。5.1 增量更新的核心逻辑增量更新的思路特别朴素读本地文件最后一条日期以“最后日期1”为起点去下载合并时去重按日期排序后覆盖写回。def incremental_update(code, end_dateNone): path os.path.join(SAVE_DIR, f{code}.csv) if not os.path.exists(path): # 本地没有直接全量下载 return download_one(code, start1990-01-01, endend_date) df_old pd.read_csv(path) last_date df_old[date].max() next_date (pd.to_datetime(last_date) pd.Timedelta(days1)).strftime(%Y-%m-%d) df_new download_one(code, startnext_date, endend_date) if df_new is None or df_new.empty: return df_old df_merged ( pd.concat([df_old, df_new]) .drop_duplicates(subsetdate) .sort_values(date) .reset_index(dropTrue) ) return df_merged核心细节很多人会忽略停牌当天数据源返回空是正常的不要当成错误。如果某只股票停牌一个月那一个月里它本来就没有行情记录增量更新脚本只需要把“成功返回空”和“下载失败”区分开即可。前者是正常后者才需要报警。另一个细节是合并时的去重。网络抖动可能让你拿到重复数据或者上次运行已经写入了今天的数据但程序崩溃了drop_duplicates能保证无论跑几次都不会重复追加。5.2 定时任务与数据源更新时点的配合增量更新是每天都该跑的任务不建议手动执行。在Linux/macOS上用crontabWindows上用“任务计划程序”。我自己的服务跑在Linux上crontab长这样30 17 * * 1-5 cd /path/to/project /usr/bin/python3 update_daily.py logs/update.log 21时间是周一到周五的17:30。为什么定这个点因为Baostock的当日数据一般下午5点后才更新太早跑会拿不到当天完整数据。不同数据源的更新时点不同有的甚至要到晚上8点后。我建议你拿到一个新数据源后先连续几天在不同时段手动跑一次增量记录数据源真正稳定的更新时间再据此设定定时任务。5.3 数据版本管理与“出问题可溯源”做数据工程最怕“某天数据错了但不知道是哪天开始错的”。我的做法很简单但很有效每天更新前先把上一版的每票CSV目录整体做一次快照只保留最近30天的备份在数据目录下放一个meta.json记录每次更新的数据源、更新时间、数据源接口版本如果发现某段数据有问题可以通过备份快速定位到出错那天再决定是否需要整体重下。这个流程不复杂但对后续研究价值巨大。你不想遇到这种情况写完一篇因子研报复现时发现数据有问题却不知道是下载当天被污染了还是后来修改过文件。有了版本记录排查时间可以从“几周”缩短到“几小时”。我个人这几年最大的体会是免费数据源之间的差异往往不是“有没有数据”而是“数据干净不干净、可持续性如何”。一套成熟的数据流水线比拼的不是某一个技巧多么精妙而是把代码池、复权处理、并发控制、断点续传、增量更新这些环节一个个钉死。做到位之后你会发现每天花在数据维护上的时间其实不到五分钟大部分精力可以安心投入到策略研究里。
