搞量化这两年我印象最深的就是数据问题。刚入门那会儿我在网上翻遍了各种免费的A股历史行情数据不是要注册就是要积分有些网站还得自己写爬虫解析页面费时费力不说数据质量也没保证。后来圈里朋友提到baostock我用Python试了一下一个下午就把全市场几千只股票的历史日线数据拉到了本地。从那以后凡有人问我A股历史行情数据用什么我都先推荐baostock。这篇内容适合谁呢主要是刚接触量化分析、想用Python批量下载多只股票历史数据的朋友尤其是被各种收费数据接口劝退的入门选手。文章会把整个流程理顺——从环境准备、单只股票请求、批量下载再到常见错误排查思路一次讲清楚。你看完之后应该能写出一套自己可复用的数据下载脚本不用再去别处拼凑零散的代码片段。1. 为什么我推荐baostock免费行情数据的选型思路1.1 数据源横向对比baostock、Tushare、AkShare我在决定数据方案之前把市面上常见的几个免费接口挨个试了一遍。Tushare是老牌了文档也全但关键的历史行情接口现在普遍需要积分积分得靠充值或者社区贡献攒对刚入门的人来说门槛有点高。AkShare走的是爬取各财经网站公开数据的路子覆盖面广但网站一旦改版解析就崩稳定性是它最大的痛点。baostock是另一条路它有自己独立的数据服务器Python库相当于一个客户端通过专门协议去请求数据不需要登录token不需要积分我用下来大半年日线级别的数据稳定性和完整度都过关。对比维度baostockTushareAkShare是否需要token不需要Pro需要积分和token不需要数据稳定度高自建服务器高受目标网站影响A股日/周/月线支持支持支持分钟线支持5/15/30/60分钟支持支持财务和估值数据支持常用指标更全更全接入成本极低中低选型的时候还有一个很容易忽略的点baostock的查询接口是专门为批量场景设计的。它不像AkShare那样要模拟浏览器请求也不像Tushare老接口那样有每分钟调用次数限制批量拉几千只股票的时候省心很多。而且数据落地之后是规规矩矩的表格结构pandas直接就能处理省了清洗的功夫。1.2 baostock的边界能拿什么数据、不能拿什么数据任何免费工具都有边界baostock也不例外。我用下来感觉最大的限制是它覆盖的是沪深A股、指数、中小板、创业板等常规证券B股、港股、美股是没有的复权因子和停牌数据可以通过字段自己处理但不会送你财务三大报表的完整明细分钟线数据虽然能拿到但一次请求的数据量比日线大很多工程上要对请求区间做拆分。不过这些限制放在量化入门阶段完全够用。如果你要做的是日线级别的选股、回测、因子分析baostock提供的数据字段已经相当充裕了——行情、成交量、换手率、市盈率、市净率、总市值都有。真到了要跑多市场多资产策略的时候再考虑别的数据源也不迟。2. 环境准备与登录检查先把地基打牢2.1 安装与Python环境问题先说环境。baostock是个纯Python库pip就能装pip install baostock pandaspandas虽然不直接参与baostock的请求过程但没有它你从baostock取回来的数据就没法高效处理所以建议一起装上。我遇到过不少新手卡在安装这一步pip install之后import还是报ModuleNotFoundError十有八九是电脑里有多个Python环境pip装到了一个解释器上而你的脚本用的是另一个解释器在跑。怎么排查这个问题Windows下我建议直接在命令行确认一下python --version pip --version看这两个命令显示的Python路径是否指向同一个环境。如果用的是PyCharm在右下角解释器设置里选对虚拟环境如果用的是VS Code按CtrlShiftP打开命令面板输入Python: Select Interpreter选你已经装好baostock的那个环境。这类问题好查就是特别磨人。2.2 登录验证error_code就是你的健康检查baostock的使用套路很固定先登录取数据最后登出。第一次连的时候我建议单独跑一个小脚本import baostock as bs lg bs.login() print(error_code:, lg.error_code) print(error_msg:, lg.error_msg) # 验证完成后执行 bs.logout()如果看到error_code是0说明连接正常。这可不是可有可无的步骤批量下载前做一次登录检查能避免后面所有请求都因为一个网络问题白白失败。这里有个容易出错的细节baostock返回的error_code是字符串0不是整数0。判断的时候别拿数字去比我第一次就没留意害得排查了半天。另外登录只需要成功一次整个进程里的后续请求都会复用这个会话不需要每查一只股票就login/logout一次。我在实践里见过有人循环里反复登录登出速度慢不说网络抖动时还容易直接断连。3. 单只股票下载先把核心API的每个参数吃透3.1 query_history_k_data_plus完整示例baostock取历史K线的核心函数是query_history_k_data_plus。第一步我建议只跑通一只股票比如用浦发银行做测试import baostock as bs import pandas as pd lg bs.login() rs bs.query_history_k_data_plus( sh.600000, date,code,open,high,low,close,preclose,volume,amount,turn,pctChg, start_date2023-01-01, end_date2023-12-31, frequencyd, adjustflag2 ) print(error_code:, rs.error_code) print(error_msg:, rs.error_msg) data_list [] while (rs.error_code 0) rs.next(): data_list.append(rs.get_row_data()) df pd.DataFrame(data_list, columnsrs.fields) print(df.head()) bs.logout()我刚开始用baostock的时候拿到df就直接去做运算结果发现close加open怎么算都不对后来才意识到baostock返回的所有字段都是字符串类型不是数字。上面这段df.head()看着是数字其实每个单元格都是str。所以做分析之前必须做类型转换for col in [open, high, low, close, preclose, volume, amount, turn, pctChg]: df[col] pd.to_numeric(df[col], errorscoerce)这是一个非常容易踩的暗坑尤其从Tushare切过来的朋友会特别不习惯——那边默认返回数值类型换到baostock之后计算的均值、收益率全变成拼接字符串了要排查一晚才能反应过来。提示baostock返回的DataFrame里所有字段初始都是字符串做任何数值运算前先统一用to_numeric做类型转换。3.2 字段选择与复权逻辑字段有多有少建议只取自己用得上的。做策略回测最常用的组合是行情加换手date, code, open, high, low, close, preclose, volume, amount, turn, pctChg。如果不管三七二十一全塞进来比如peTTM、pbMRQ这类估值字段单只股票看不出来但批量拉全市场时数据量和内存压力都会成倍增加。adjustflag参数非常关键它控制复权方式1后复权2前复权3不复权回测建议用前复权这样你在回测日看到的股价跟行情软件上的前复权K线图一致如果要做严格的长期收益计算最好同时拉一份后复权数据配合使用。新手最容易犯的错是把不复权数据直接拿去做长周期回测遇到除权除息收益率曲线会出现莫名其妙的坑。记住一个简单的判断标准策略持仓周期短、很少跨除权日用前复权够用做长周期研究复权逻辑要想清楚。baostock还支持分钟线数据frequency参数换成5、15、30、60分别对应5分钟、15分钟、30分钟、60分钟K线。分钟线的数据量比日线大几个量级一只股票一年的5分钟数据就可能上万行批量拉取时建议按季度切分区间避免单次请求数据量过大导致超时或内存压力。3.3 数据正确性验证拿到数据后我习惯先做两步验证。第一步是看行数和日期范围用df[date].min()和df[date].max()比对请求区间如果发现少了几个月大概率是停牌或者股票当时还没上市。第二步是随机挑一个日期去行情软件上比对收盘价比对的时候注意复权方式要一致。我见过有人拿前复权数据跟非复权K线图对比价格对不上就怀疑是baostock的问题其实是两个数据的口径本来就不一样。停牌问题在批量下载时格外突出。A股停牌股票在查询区间内会少数据行这是正常现象不要当成接口故障。但你要在保存数据的逻辑里对“空结果”和“失败”做区分否则后面做全市场分析时会漏掉一批股票而不自知。4. 批量下载多只股票从能用变成好用的工程化方案4.1 循环结构、断点续传和异常隔离批量下载多只股票最朴素的写法就是for循环里套上面的单只查询代码。但只写一个for循环是远远不够的实战中一定会遇到一两个请求超时、某只股票数据异常的情况。我现在的做法是给每只股票套上try-except单独隔离异常不管哪只失败都不影响全局同时把失败记录写进日志文件。断点续传也很有用。我把每只股票存成单独的CSV文件文件名就用股票代码加交易所前缀如果文件已存在就直接跳过。这样脚本拉到一半被中断了重跑一次会自动跳过已经成功的股票不用从头再拉一遍。千万不要把所有股票塞进同一个DataFrame最后一次性写文件数据量一大内存就先爆了断点续传也没法做。下面是我日常在用的批量下载脚本框架import baostock as bs import pandas as pd import os import time OUTPUT_DIR ./stock_history FAIL_LOG ./failed_stocks.txt FIELDS date,code,open,high,low,close,preclose,volume,amount,turn,pctChg START_DATE 2015-01-01 END_DATE 2024-12-31 ADJUST 2 def download_stock(code: str) - int: filepath os.path.join(OUTPUT_DIR, f{code}.csv) if os.path.exists(filepath): print(f[跳过] {code} 已存在) return 0 rs bs.query_history_k_data_plus( code, FIELDS, start_dateSTART_DATE, end_dateEND_DATE, frequencyd, adjustflagADJUST ) if rs.error_code ! 0: raise RuntimeError(f接口返回错误 {rs.error_code}: {rs.error_msg}) rows [] while rs.next(): rows.append(rs.get_row_data()) if not rows: raise RuntimeError(f{code} 返回空数据) df pd.DataFrame(rows, columnsrs.fields) for col in [open, high, low, close, preclose, volume, amount, turn, pctChg]: df[col] pd.to_numeric(df[col], errorscoerce) df.to_csv(filepath, indexFalse, encodingutf-8-sig) print(f[完成] {code} 共 {len(df)} 行) return len(df) def main(): os.makedirs(OUTPUT_DIR, exist_okTrue) lg bs.login() if lg.error_code ! 0: print(登录失败:, lg.error_msg) return stock_list [ sh.600000, sz.000001, sz.300750, sh.601318, sz.000858, ] failed [] start_ts time.time() for code in stock_list: try: download_stock(code) except Exception as e: print(f[失败] {code}: {e}) failed.append(code) with open(FAIL_LOG, a, encodingutf-8) as f: f.write(f{code}\t{str(e)}\n) time.sleep(0.3) print(f本次运行结束成功 {len(stock_list) - len(failed)} 只失败 {len(failed)} 只耗时 {time.time() - start_ts:.1f}s) bs.logout() if __name__ __main__: main()这套脚本我用了挺长时间核心思路就三条异常隔离、断点续传、失败留痕。你用了之后会遇到一种很典型的场景——第一次跑完失败列表里有七八只股票你只需要再跑一次脚本会自动跳过已经成功的只补下载剩下的非常舒服。4.2 输出格式选择与数据分库每只股票一个CSV文件的方案对单只股票分析最友好pd.read_csv直接加载就行。但如果你要做的是全市场选股得一次读几千个文件再纵向拼接。这种情况下我建议再维护一个“合并库”用增量追加的方式把每天的数据更新进去。文件格式上CSV通用、可读、谁都能打开但全市场几千只股票的日线数据叠起来有几百MBCSV的读写效率和存储空间都不占优。如果想要更高效的存储Parquet是很好的选择读写速度快得多还能保留数据类型信息。不过对多数入门者来说还是建议从CSV开始等真的感觉到IO成了瓶颈再迁移到Parquet不迟。工程上的原则永远是别提前优化先跑通再改。4.3 用query_all_stock自动获取股票列表告别手动维护手动维护股票代码列表不是长久之计沪深两市的股票数量一直在变而且你手动维护的列表很可能漏掉新股。应该用baostock自己查rs bs.query_all_stock(day2024-12-31) stock_list [] while rs.next(): stock_list.append(rs.get_row_data())拿到的每一行是[code, tradeStatus, code_name]比如[sh.600000, 1, 浦发银行]code字段自带sh/sz前缀直接就能用到下载循环里不用再手动加前缀。不过做全市场拉取之前提醒一句几千只股票每只间隔0.3秒光是请求耗时就要一二十分钟这还不算数据落盘时间。等不及的话可以先过滤掉停牌股票或者等用到第6章里讲的多进程方案再提速。5. 常见错误排查链路一个一个来这一部分我按“现象、排查过程、根因、修复”的顺序拆开讲方便你对照定位。都是批量任务里高频出现的真问题。5.1 查询结果为空但没报错现象代码没有任何异常rs.error_code也是0但循环里一行数据都读不到。排查过程我先怀疑是接口参数问题于是打印rs.fields确认字段没毛病再打印start_date和end_date发现都对。绕了一圈之后试着把股票代码从600000改成sh.600000数据马上就出来了。根因baostock的股票代码必须带交易所前缀sh代表上交所sz代表深交所。六个纯数字的代码是查不出任何数据的关键是它还不报错就安静地返回空结果特别容易让人摸不着头脑。修复代码列表导入后做一次格式化。如果是从query_all_stock拿到的本身就带前缀不会踩坑。如果是从别的地方拿的纯数字代码自己处理一下def normalize_code(code: str) - str: code code.strip() if code.startswith((sh., sz.)): return code if code.startswith(6): return sh. code if code.startswith((0, 3)): return sz. code raise ValueError(f无法判断交易所前缀: {code})5.2 error_code非0参数格式导致的惨案现象rs.error_code返回的不是0error_msg里经常能看到参数错误之类的提示。排查过程把参数逐个往回查发现日期传的是20240101而不是2024-01-01。baostock要求的日期格式是YYYY-MM-DD差一个横杠它都不认。根因baostock对字符串参数的格式校验非常严格。日期必须用短横线连接frequency参数传的是字符串d、w、madjustflag传的是字符串1、2、3。这些细节在从老项目搬代码的时候特别容易踩中。修复外部传参统一做一层清洗from datetime import datetime def format_date(dt_str: str) - str: dt datetime.strptime(dt_str, %Y%m%d) return dt.strftime(%Y-%m-%d)5.3 批量循环中断异常处理不彻底现象批量脚本跑到中途突然崩溃退出报了一堆traceback。之前下载好的股票都在后面的一只都没拉。排查过程看traceback发现是在某只股票的while循环里抛了异常可能是网络连接中断也可能是某行数据在类型转换时抛错。由于最外层没有合理的异常边界整个进程就这么退了。根因批量下载必须在单只股票的粒度上做异常隔离业务异常和网络异常要分开处理并且把失败信息记录下来而不是让一个异常把整个任务拖垮。修复就像第4章示例代码里写的那样每只股票的处理都套上try-except同时写失败日志文件。这里有个容易忽略的点如果失败是网络抖动这类瞬时问题重跑一次通常就能解决所以失败日志里一定要记录时间方便判断是不是集中性故障。5.4 内存溢出与查询过慢现象全市场批量拉日线时内存一路飙升数据还没落盘进程就被操作系统杀掉了。排查过程代码review的时候发现批量脚本把每只股票的DataFrame都append到一个全局list里等全部拉完再统一to_csv。几千只股票的数据全攒在内存里不爆才怪。另外查询区间拉到了2000年每只股票几千行字段又取了一长串数据量自然爆炸。根因数据累积没有及时落盘同时请求范围过大、字段过多。baostock查询接口是把结果先整体传回本地再逐行读取的时间跨度越大、字段越多单次请求的传输时间和内存消耗就越高。修复改成每只股票取完就立刻写CSV然后释放引用。全市场批量拉取时可以按年份切片一年一年查或者按板块分批查每批之间sleep几秒既降低内存压力也给接口留出喘息空间。5.5 文件乱码问题现象用Excel打开下载好的CSV中文内容变成乱码。排查过程最开始我用的是encodingutf-8后来有同事说打不开。用记事本打开没乱码但Excel显示乱码换成encodingutf-8-sig之后Excel就正常了。根因UTF-8不带BOM时Excel默认按ANSI解析中文环境下会出现乱码。这不算baostock的问题是CSV文件跨软件兼容性的老毛病。修复保存CSV时固定用encodingutf-8-sig这个方案能同时兼容Excel、记事本和Python读取。后续读回DataFrame的时候也用encodingutf-8-sig或者直接让pandas自动推断不要两边编码不一致。6. 进阶增量更新、多进程和一些性能优化6.1 增量更新每天只拉新数据全量重拉的效率太低了尤其每天要更新一次数据的时候。我现在的做法是本地CSV保存的已经是历史数据增量更新时先读本地文件的最后日期然后把查询的start_date设为最后日期加一天拉到新数据后跟原始DataFrame拼接统一覆盖写回。下面是思路示意def load_last_date(filepath: str) - str: if not os.path.exists(filepath): return 1990-01-01 df pd.read_csv(filepath, usecols[date]) return str(df[date].max()) def incremental_update(code: str): filepath os.path.join(OUTPUT_DIR, f{code}.csv) last_date load_last_date(filepath) if last_date END_DATE: return # 以last_date为新的start_date拉取新增数据后合并去重拼接的时候要注意baostock可能返回重复日期的数据用drop_duplicates(subsetdate, keeplast)做一次收尾保证本地数据干净。6.2 性能优化与多进程方案baostock的登录会话是基于进程的直接开线程池容易出问题。要提速最简单稳妥的方式是进程隔离把股票列表切成若干份每个子进程各自bs.login()之后处理自己那份互不干扰。我自己常用multiprocessing.Pool来实现全市场股票列表切4个进程时间大概能压缩到单进程的三分之一左右。这里有个非常容易踩的坑子进程的函数里要重新调用bs.login()不能复用父进程的登录状态。如果只在主进程login然后丢给子进程去query会发现子进程里的接口全部报错。另外机器性能一般的话也别开太多进程。baostock服务端对请求频率是有限制的进程开太多反而容易被限流。我日常跑4个进程的时候每只股票之间仍然会加0.2秒的sleep稳定优先。6.3 可扩展的API交易日历、基本面数据批量下载历史行情数据只是一个开始。baostock还有query_trade_dates可以拉交易日历做停牌判断和策略的时间轴对齐非常有用query_stock_basic可以查股票基本信息配合query_all_stock就能过滤掉ST股或次新股减少无效下载query_profit_data、query_operation_data这些接口还能拉到财务指标。后续要做因子分析这些都能扩展出来。我个人的体会是先把行情数据的批量下载管道一次性搭好之后要接什么数据都只是往同一个管道里加一小段而已。关键是文件命名、字段规范、异常处理这套框架要在一开始就搭对后面会省下大量时间。