Python量化投研实战：用baostock批量下载股票历史数据全指南

在量化投研和数据挖掘的工作流里最烦人的往往不是写策略而是搞数据。刚开始接触股票数据分析时我为了凑齐一个像样的股票池日线数据手动从各种网站复制粘贴或者在 Excel 里折腾半天最后数据还对不齐、有缺失那叫一个痛苦。后来接触到 baostock 这个免费、开源的 Python 证券数据接口库情况才彻底改观。它不需要复杂的 Token 认证数据来源于证券宝官方整理的市场公开信息接口稳定、字段规范特别适合用 Python 做批量下载和历史数据回溯。这篇文章就把我实际使用 baostock 批量下载多只股票历史数据的完整思路、核心代码和踩坑记录整理出来希望能帮你少走弯路。1. 内容整体设计与思路拆解在动手写代码之前得先把“批量下载历史数据”这件事拆解清楚想清楚方案再动手能避免后面很多返工。1.1 批量下载技术选型对比为什么建议选 baostock市面上能获取股票历史数据的 Python 库不少比如 akshare、tushare、yfinance还有 baostock。我简单做过一轮对比仅代表个人使用体验特性baostocktushareakshareyfinance是否需要注册 Token不需要免费登录需要且有积分限制不需要不需要数据稳定性较好接口变动不频繁稳定但高频数据权限收费依赖上游网页结构偶尔失效依赖 Yahoo 接口时有中断历史数据完整性日线/周线/月线较全全面但部分需高积分较全美股为主A 股数据缺失较多字段规范化字段命名统一规范字段随接口变化英文为主上手门槛低官方文档清晰中低低从我自己的使用体验来看如果你主要做A 股日线、周线、月线级别的历史数据回溯baostock 是最省心的选择。它没有积分墙不用频繁维护登录态接口返回的字段也足够规范。虽然它在分钟线、实时行情方面支持有限但做批量历史下载它非常合适。1.2 批量下载的核心设计思路批量下载多只股票表面上是“for 循环里查数据”但实际操作中要考虑几个关键点否则容易翻车第一个是股票池管理。得先明确你要下载哪些股票。按行业分按指数成分股分还是自定义一篮子股票如果不做这一步脚本就不知道该遍历哪些代码。我在实际项目中通常会把股票池放在一个 CSV 文件里列名就叫symbol每一行是一个股票代码比如sh.600000或sz.000001。这样脚本只需要读文件后续要加股票、删股票都很方便不用改代码。第二个是数据落地格式。baostock 返回的数据是 DataFrame 格式默认的字段都是英文字段名比如code、open、high、low、close、volume、amount等。批量下载时建议把每只股票单独存成一个 CSV 文件文件名带上股票代码和日期范围比如sh.600000_daily_2015-01-01_2023-12-31.csv。这样做的好处是后期做单股回测、因子计算时可以直接按股票代码索引不用每次从一个大文件里做筛选I/O 压力小很多。第三个是进程管理。baostock 的接口本身是同步阻塞的如果股票数量特别多比如上千只纯 for 循环速度很慢。但 baostock 官方不推荐短时间高频调用接口所以我在实践中倾向于用“单进程 多线程”的方式做 I/O 等待优化而不是盲目上多进程。这里面的度得把握好后面我会专门提到。1.3 数据下载的整体流程框架整个流程可以用一句话概括准备股票池 - 逐个查询 - 数据清洗 - 本地落地 - 校验完整性。准备阶段的核心是确定股票代码列表和起止日期。查询阶段用的核心接口是 baostock 的query_history_k_data_plus需要传入股票代码、日期范围、频率日/周/月和复权方式前复权、后复权、不复权。数据清洗阶段需要把Date转成时间类型把数值字段从字符串转成浮点数处理缺失值。落地阶段就是把清洗后的 DataFrame 存成 CSV。校验阶段要检查文件是否生成、是否有空文件、数据行数是否符合预期。2. 核心细节解析与实操要点理解了整体流程现在进入需要动手的部分。这一节我把 baostock 的核心接口逻辑以及几个容易出错的细节单独拿出来讲。2.1 baostock 核心接口逻辑从登录到数据解析baostock 的使用方式比较固定常规步骤是导入模块并登录lg bs.login()成功后会返回一个结果对象里面包含error_code和error_msg。登录时如果本机网络环境不好可能会报error_code10002这通常是网络问题多试一次或者检查代理设置。调用query_history_k_data_plus拉取 K 线数据。遍历返回结果rs用rs.get_row_data()逐行取数据。用rs.fields拿到字段名列表配合get_row_data()转成 DataFrame。退出登录bs.logout()。代码框架大概是这样后面会给出完整脚本import baostock as bs import pandas as pd import datetime # 登录 lg bs.login() print(login respond error_code: lg.error_code) print(login respond error_msg: lg.error_msg) # 查询数据 rs bs.query_history_k_data_plus( sh.600000, date,code,open,high,low,close,preclose,volume,amount,adjustflag,turn,pctChg, start_date2015-01-01, end_date2023-12-31, frequencyd, adjustflag2 ) print(query respond error_code: rs.error_code) print(query respond error_msg: rs.error_msg) # 解析数据 data_list [] while (rs.error_code 0) and rs.next(): data_list.append(rs.get_row_data()) # 转 DataFrame result pd.DataFrame(data_list, columnsrs.fields) # 退出登录 bs.logout()注意看query_history_k_data_plus里的fields参数是可以自定义的这能让返回的数据更精简。比如做纯价格分析就不需要turn和pctChg换手率和涨跌幅能减少内存占用和传输量。2.2 复权方式的深度理解前复权、后复权、不复权baostock 的adjustflag参数有三种取值1后复权把历史价格按最新股本和分红除权因子调整真实反映“如果一直持股当前市值是多少”。2前复权把最新价格保持不变调整历史价格适合看趋势和技术指标。3不复权直接用原始成交价。这里有个容易踩坑的地方做历史回测时如果用不复权数据计算收益率遇到除权除息日会出现巨大的价格跳空导致收益计算异常。我的实践经验是如果做长周期策略回测建议使用后复权数据因为它的价格序列是连续的且不会受最新价格影响如果做短周期技术分析比如 MA、MACD可以用前复权数据。但是要注意前复权数据是动态变化的最新的除权除息事件会改变历史价格所以如果你需要保存一份稳定的历史数据用于研究复盘后复权是更稳妥的选择。2.3 字段含义与数据清洗的关键点baostock 返回的字段里有几个需要特别说明date交易日期格式是YYYY-MM-DD这个字段在存 CSV 时最好原样保留为字符串方便 sort。code股票代码带交易所前缀比如sh.600000、sz.000001。open、high、low、close都是字符串类型需要astype(float)转换。volume成交量单位是股注意不是手。amount成交额单位是元。turn换手率这里是一个字符串百分比数值比如0.85表示 0.85%不是 85%。pctChg涨跌幅同样是个百分数值。isST是否 ST 股这个字段在 baostock 中有时有、有时没有得看接口版本。数据清洗环节我通常会做这么几件事先把空值NaN或者空字符串过滤掉再把数值列统一转成float如果有除零或极端值还要做一轮 Winsorize 处理。但要注意baostock 返回的空值往往就是空字符串直接用pd.to_numeric会报错所以清洗的时候要先把空字符串替换成np.nan。2.4 股票代码格式的坑baostock 的股票代码格式和我们在行情软件里看到的 6 位纯数字代码不一样。比如浦发银行行情软件显示600000但 baostock 要求传sh.600000。这个格式如果不注意最常见的结果是query_history_k_data_plus返回的rs.error_code是10001提示参数错误。解决办法有两个一个是在构建股票池时直接保存带前缀的代码另一个是在脚本里写一个转换函数根据数字代码前两位判断是沪市还是深市再拼上前缀。我的习惯是在股票池 csv 里直接存标准格式这样省去转换逻辑脚本更简洁。这里给一个小工具函数如果你只有 6 位数字代码可以用它自动补全def format_bs_code(code: str) - str: code code.strip() if code.startswith((sh., sz.)): return code if code.startswith(6): return fsh.{code} if code.startswith((0, 3)): return fsz.{code} if code.startswith((4, 8)): return fbj.{code} raise ValueError(fUnrecognized code: {code})2.5 时间跨度的细节处理baostock 单次查询的时间跨度虽然比较宽但我在实测中发现如果一次查询 10 年的日线数据偶尔会出现数据不完整或响应时间过长的情况。稳妥的做法是按年份分段查询然后拼接。比如你要下载 2015 年至今的数据可以分成 2015、2016、2017... 每年查一次再用pd.concat合并。这样做还有一个额外的好处如果某个年份查询失败可以单独重试不用把整个查询重新跑一遍。3. 实操过程与核心环节实现理论说多了没用直接进入实战。这一节我给出一个完整的、可以拿来就用的批量下载脚本并逐步解释关键环节是怎么实现的。3.1 准备股票池文件我建议把股票池放在一个stock_pool.csv文件里格式如下symbol sh.600000 sh.600036 sz.000001 sz.000002如果你需要沪深 300 的全部成分股之前可能要从第三方网站爬或者用 baostock 自带的指数成分接口。baostock 有query_hs300_stocks和query_zz500_stocks接口可以直接拿到沪深 300 和中证 500 的成分股列表。不过我通常还是维护一个自定义的 CSV因为成分股会定期调整而我的研究股票池是相对固定的。3.2 批量下载脚本完整实现以下脚本是我实际在用的一个简化版本功能包括读取股票池、分段查询、异常重试、落地 CSV、打印进度。注意我把登录操作放在了循环外因为频繁登录容易被服务端风控而且也没有必要。import baostock as bs import pandas as pd import datetime import os import time from tqdm import tqdm # 字段列表按需取用 FIELDS date,code,open,high,low,close,preclose,volume,amount,adjustflag,turn,pctChg FREQUENCY d ADJUSTFLAG 2 # 前复权 # 股票池 STOCK_POOL_FILE stock_pool.csv OUTPUT_DIR daily_data START_DATE 2015-01-01 END_DATE 2024-12-31 def init_output_dir(path: str): if not os.path.exists(path): os.makedirs(path) def query_single_stock(bs_instance, symbol: str, start: str, end: str, retry: int 3): 查询单只股票历史数据带重试机制。 返回 DataFrame。 for attempt in range(retry): rs bs_instance.query_history_k_data_plus( symbol, FIELDS, start_datestart, end_dateend, frequencyFREQUENCY, adjustflagADJUSTFLAG ) if rs.error_code 0: data_list [] while rs.next(): data_list.append(rs.get_row_data()) df pd.DataFrame(data_list, columnsrs.fields) # 清洗 if df.empty: return df numeric_cols [open, high, low, close, preclose, volume, amount, turn, pctChg] for col in numeric_cols: df[col] pd.to_numeric(df[col], errorscoerce) df[date] pd.to_datetime(df[date]) df df.sort_values(date).reset_index(dropTrue) return df else: print(f[WARN] {symbol} 查询失败({rs.error_code}: {rs.error_msg}), 第{attempt1}次重试) time.sleep(2) return pd.DataFrame() def main(): # 1. 读取股票池 stock_pool pd.read_csv(STOCK_POOL_FILE, dtype{symbol: str}) symbols stock_pool[symbol].tolist() # 2. 初始化输出目录 init_output_dir(OUTPUT_DIR) # 3. 登录只登录一次 lg bs.login() if lg.error_code ! 0: print(f登录失败: {lg.error_msg}) return print(登录成功) # 4. 逐只下载 failed_symbols [] for symbol in tqdm(symbols, descDownloading): file_name f{symbol}_daily_{START_DATE}_{END_DATE}.csv file_path os.path.join(OUTPUT_DIR, file_name) # 如果文件已存在且非空可跳过断点续传思路 if os.path.exists(file_path) and os.path.getsize(file_path) 0: print(f[SKIP] {symbol} 已存在跳过) continue df query_single_stock(bs, symbol, START_DATE, END_DATE) if df.empty: print(f[FAIL] {symbol} 无数据) failed_symbols.append(symbol) continue # 5. 保存 CSV df.to_csv(file_path, indexFalse, encodingutf-8-sig) time.sleep(0.5) # 控制请求频率 # 6. 退出 bs.logout() print(f全部完成。失败股票: {failed_symbols}) if __name__ __main__: main()简单解释一下这个脚本里几个重要的设计字段中的adjustflag同时出现在数据里。这样做是有意为之因为adjustflag字段值能够告诉你这份数据用的是哪种复权方式方便后续做数据校验避免把前复权数据和后复权数据混用。断点续传思路。脚本开始时检查文件是否已经存在如果存在且非空就跳过。这个设计在实际操作中太重要了因为下载几百只股票时难免中途断网或程序崩溃有了这个机制重新运行脚本就能接着下载不用从头再来。time.sleep(0.5)控制请求频率。baostock 虽然免费但不代表可以无限并发猛刷。我在实践中的经验是500ms 的间隔已经比较稳妥既能保证速度又不容易触发服务端限制。如果频繁遇到10002网络错误或10004连接中断可以把间隔调整到 1 秒以上。3.3 分段查询逻辑应对大跨度时间数据刚才提到分段查询更稳妥这里给出一个具体实现。核心思路是把START_DATE到END_DATE按年份切分成多个区间然后在循环里依次查询并合并。def split_date_range(start: str, end: str): 按年份切分时间区间。 start_dt datetime.datetime.strptime(start, %Y-%m-%d) end_dt datetime.datetime.strptime(end, %Y-%m-%d) ranges [] current start_dt while current end_dt: year_end current.replace(month12, day31) if year_end end_dt: year_end end_dt ranges.append((current.strftime(%Y-%m-%d), year_end.strftime(%Y-%m-%d))) current datetime.datetime(year_end.year 1, 1, 1) return ranges def query_single_stock_yearly(bs_instance, symbol: str, start: str, end: str): 分段查询并拼接。 frames [] for s, e in split_date_range(start, end): df_year query_single_stock(bs_instance, symbol, s, e, retry3) if not df_year.empty: frames.append(df_year) if not frames: return pd.DataFrame() return pd.concat(frames, ignore_indexTrue).drop_duplicates(subset[date], keeplast)这个函数里对拼接结果做了drop_duplicates(subset[date])防止两个年份区间因时间边界重复导致数据行重复。实测下来这种方式比一次性查询 10 年数据要稳定很多出现数据截断的概率大幅降低。3.4 数据校验下载完了不等于万事大吉数据下载完之后建议立刻做一轮完整性校验。我最常做的校验有三个交易日数量校验。A 股一年大约有 240~250 个交易日如果你下载的是 2019 年 1 月到 2023 年 12 月共 5 年正常应该有 1200 多行。如果某只股票只有 800 行那就要小心了可能是上市时间晚次新股、长期停牌也可能是下载时漏了数据。我通常写一个简单的统计函数打印每只股票的数据行数和日期范围。日期连续性校验。检查是否有缺失的交易日。最简单的方式是找到该区间内所有实际交易日可以用任意一只正常股票作基准或者用上证指数来对然后对比目标股票的日期是否覆盖了这些交易日。如果缺失太多就得重新下载。极端值校验。检查high是否小于low、close是否超出high/low的范围。这些都是数据源异常时会出现的低级错误但一旦出现对后续回测结果的影响是连锁性的。4. 常见问题与排查技巧实录这一节是整篇文章里我最想让你认真看的部分。baostock 虽然好用但坑也不少。我把自己实际踩过的坑和排查思路整理成清单遇到问题可以对照着看。4.1 baostock 常见错误码对照表错误码含义常见原因我的处理办法10001参数错误股票代码格式不对或字段拼写错误检查symbol是否带sh./sz.前缀检查fields是否在官方文档列表内10002网络错误本地网络波动、代理干扰检查网络连接关闭系统代理重试10003登录失败用户名密码错误免费用户一般不用账号确认 baostock 是否需要账号重新调用bs.login()10004数据不存在查询了尚未上市的时间段先用query_stock_basic查上市日期再设查询区间10005频率或复权参数错误frequency或adjustflag超出范围确认frequency只能为d、w、madjustflag只能为1、2、310006数据量过大单次查询时间跨度过大改用分段查询按年份切分10007用户未登录logout后还在查询检查代码里是否提前调用了bs.logout()注意baostock 返回的错误码本身只有在rs.error_code或者lg.error_code里能看到。很多人习惯只看返回的 DataFrame结果数据不对也不知道去哪排查。我的习惯是每次调用接口后都打印一下error_code和error_msg这样问题一出现就能立刻定位。即使error_code 0error_msg有时也会有一些 warning值得扫一眼。4.2 批量下载时数据缺失停牌股和次新股批量下载时最容易被忽略的就是停牌股和次新股。停牌股在停牌期间没有交易数据baostock 会直接跳过这些日期所以 DataFrame 里的日期是不连续的这是正常现象。但如果下载的是指数成分股池某个股票可能因为长时间停牌导致数据行数比别家少一大截。这时就不要盲目补数据而应该先去查一下它的停牌公告确认是合理缺失还是下载不完整。次新股就是上市时间晚于你设定的START_DATE。比如你想下载 2015 年至今的数据但是某只股票 2021 年才上市那 baostock 只会返回 2021 年之后的数据。这种情况下不能简单判断“数据缺失”而是要把上市日期纳入考虑。可以用bs.query_stock_basic(codestock_code)查询上市日期和退市日期然后在校验逻辑里做适配。4.3 循环查询时频繁登录退出引起的连接问题很多人会把bs.login()和bs.logout()放到循环体内部每下载一只股票就重新登录一次。我一开始也这么干后来发现这样不仅慢而且容易触发服务端的连接频率限制导致后面的查询全部失败。正确做法是登录一次循环查询最后退出。如果会话中途断开比如网络重启可以捕获异常后重新登录然后接着从断点续传。这里有一个简单的连接检测方式就是查询任意一只指数数据如果返回的error_code ! 0说明连接已经断了需要重新bs.login()。4.4 空值处理和数据类型转换的坑baostock 返回的字段几乎全是字符串。有些字段在遇到停牌或数据缺失时返回的是空字符串而不是NaN。这时候直接做df[close].astype(float)会抛ValueError。所以我建议清洗时先用pd.to_numeric(col, errorscoerce)把无法转换的内容变成NaN然后再统一处理。还有一个细节是turn换手率和pctChg涨跌幅在 baostock 里是“数值百分比”类型比如pctChg3.25表示涨了 3.25%不是 0.0325。如果你后续要计算收益率千万别忘了除以 100。4.5 大文件合并时的性能和内存问题如果你要下载几千只股票并且把所有结果合并成一个parquet或feather文件内存可能会爆掉。我的经验是分批合并比如每下载 100 只股票合并一次或者直接保持每只股票一个文件的结构。另外一个性能优化技巧是在读取 CSV 时把date列直接解析为日期类型同时把数值列的类型在read_csv时指定好减少内存占用。df pd.read_csv(sh.600000_daily.csv, parse_dates[date], dtype{open: float32, high: float32, low: float32, close: float32})4.6 网络环境导致的连接重置baostock 的服务器在国内如果你所在网络环境有代理或者防火墙很容易出现10002错误。排查思路先关掉系统代理试一次再检查防火墙是否拦截了baostock的通信端口。如果仍有问题可以在代码里设置超时重试实在不行就换一个时间再跑。我遇到过白天高峰期频繁10002晚上 10 点之后跑同样的脚本就非常顺畅的情况。5. 进阶优化多进程并发与数据落地格式升级基础版脚本能解决日常需求但如果股票池是上千只又想尽可能快地拿到数据那就要考虑并发和存储格式的优化了。5.1 多进程安全使用的边界前面我说过 baostock 不建议高频调用但实际测试下来适度的并发还是能明显提速的。我采用的方案是concurrent.futures.ThreadPoolExecutor控制最大线程数在 4~6 个每个线程内部串行处理一个股票池分片。from concurrent.futures import ThreadPoolExecutor, as_completed MAX_WORKERS 4 def download_one(symbol): # 注意这里需要重新登录因为 baostock session 不是线程安全的 lg bs.login() try: df query_single_stock_yearly(bs, symbol, START_DATE, END_DATE) if df.empty: return symbol, False df.to_csv(os.path.join(OUTPUT_DIR, f{symbol}.csv), indexFalse) return symbol, True finally: bs.logout() with ThreadPoolExecutor(max_workersMAX_WORKERS) as executor: futures {executor.submit(download_one, sym): sym for sym in symbols} for future in as_completed(futures): sym futures[future] try: symbol, status future.result() except Exception as e: print(f[ERROR] {sym}: {e})这里有个重要的注意点baostock 的 session登录态并不是线程安全的。所以我在download_one里做了线程内重新登录和退出的操作避免多个线程共享同一个连接产生不可预知的问题。不过要提醒一句并发调高之后被服务端临时限制的概率也会增加。我的经验值是 4 个线程已经够用追求极致速度可以调到 6但超过 8 就可能触发异常。如果你跑大任务最好在脚本里给每个线程分配一个合理的休眠时间比如每次查询完time.sleep(0.3)整体速度并不会慢太多但稳定性提升明显。5.2 Parquet 存储从 CSV 到分析型格式当数据量积累到一定程度CSV 的存储效率和读取速度就会成为瓶颈。我后期把所有日线数据统一转成了 Parquet 格式按股票代码分区存储配合 pandas 读取非常流畅。Parquet 的好处是列式存储、压缩率高而且保留了 dtypes 信息读取时不需要再做类型转换。如果你也想转 Parquet直接用 pandas 即可df.to_parquet(sh.600000_daily.parquet, indexFalse)不过要注意pandas 读写 Parquet 需要安装 pyarrow 或 fastparquet 库你可以根据自己的环境选择。pyarrow 更主流推荐优先使用。5.3 增量更新让数据保持“新鲜”历史数据下载完成后还有一件事经常遇到过了一个月你又需要更新数据。如果重新下载整个历史区间虽然 baostock 响应很快但没必要浪费资源。更好的方案是增量更新。思路是读取本地已有数据的最新日期再从这个日期的后一天开始查询到今天然后追加到原来的 DataFrame 中最后去重保存。def incremental_update(symbol: str, csv_path: str): if not os.path.exists(csv_path): return query_single_stock_yearly(bs, symbol, START_DATE, END_DATE) old_df pd.read_csv(csv_path, parse_dates[date]) last_date old_df[date].max() start_inc (last_date datetime.timedelta(days1)).strftime(%Y-%m-%d) end_inc datetime.date.today().strftime(%Y-%m-%d) if start_inc end_inc: print(f{symbol} 已是最新数据) return old_df new_df query_single_stock_yearly(bs, symbol, start_inc, end_inc) if new_df.empty: return old_df combined pd.concat([old_df, new_df], ignore_indexTrue) combined combined.drop_duplicates(subset[date], keeplast).sort_values(date) combined.to_csv(csv_path, indexFalse, encodingutf-8-sig) return combined这个函数我封装得比较简练但思路很实用。有了它每周或者每个月跑一次增量脚本数据就能一直保持最新而不用每次都下载全量数据。写在最后的一点体会用 baostock 做批量下载说难不难说简单也不简单。难的不是调接口本身而是如何把一个“能用”的脚本变得更“可靠、可维护、可复用”。我从一开始手动复制粘贴到后来写出上面这套带断点续传、分段查询、增量更新的流程中间踩的坑基本都写在文章里了。如果让我给你一个最核心的建议我会说永远不要高估一次下载的完整性也不要低估数据质量对后续分析的影响。每一次任务跑完多花两分钟做一下行数校验和日期连续性校验你会省下后面大量的返工时间。希望这篇实战文章能帮你顺利跑通批量下载流程。如果你在实际操作中还遇到了其他奇奇怪怪的问题欢迎按着错误码去排查很多问题其实都是参数格式或者网络环境导致的冷静下来一步步排查很快就能解决。