做量化分析或者数据研究的人十有八九都遇到过同一个尴尬好不容易写好了策略逻辑结果卡在行情数据获取上。商业数据源一年几万块免费接口要么限制多、要么不稳定网上能找到的现成数据包又往往是裁剪过的拿来做回测心里总是不踏实。我自己也在这个坑里折腾了很久直到后来全面转向 baostock才算是把“数据从哪来”这个问题彻底解决了。这篇文章就把我实际使用的完整方案分享出来从环境准备、批量下载多只股票历史数据的核心代码到数据落盘、断点续传、增量更新这些工程化细节最后是把高频踩的坑和错误排查整理成一份速查表。如果你正在用 Python 做股票数据分析、写量化策略回测或者只是需要一份干净的A股历史行情数据做研究这篇文章应该能帮你少走不少弯路。1. baostock 到底能做什么为什么我推荐它先花点时间把 baostock 这个东西说清楚。它本质上是一个免费、开源的 Python 第三方库专门用来获取 A 股市场的历史行情数据。和很多需要注册、申请 token、甚至要付费的同类接口不同baostock 最大的特点就是联网后可以直接调用不需要任何身份认证也没有明显的调用频率限制。1.1 能拿到哪些数据baostock 覆盖的数据范围对于做日线级别的分析来说是相当够用的。股票历史日线行情开高低收、成交量、成交额、换手率、市盈率、市净率、市销率、市现率等数据可以追溯到 1990 年 A 股开市。指数历史行情上证指数、深证成指、沪深300、中证500 等主流指数都能直接拿。股票基本面数据季度/年度的盈利能力、运营能力、成长能力、偿债能力等财务指标。复权因子这是做量化回测时极其重要的数据baostock 直接提供省去了自己计算前复权、后复权的麻烦。交易日历每年每月的交易日明细做日期对齐时很实用。我用这个库做过一次历史回测从 2015 年到现在将近 3000 个交易日、几十只股票的数据中途没有被封、没有断流整体稳定性比预期好很多。1.2 和其他数据源的对比很多朋友会问既然有 tushare、akshare 这些库为什么还要用 baostock我个人的使用感受对比如下。对比项baostocktushareakshare是否需要注册/token不需要需要积分制部分接口需要历史数据完整性覆盖1990年至今积分越高权限越多数据源多质量参差复权因子直接提供完整高积分可获取部分提供调用频率限制目前无明显限制积分限制易限流数据源变动影响大数据格式稳定性稳定字段固定稳定经常变动需跟进更新tushare 在社区知名度最高但它对高积分的要求确实挡住了不少新手。akshare 的优点是数据源丰富缺点是某些接口的数据源经常改版代码写着写着就可能失效需要持续维护。baostock 刚好在两者之间找到了一个平衡点代码逻辑稳定、数据格式固定、使用完全免费。说实话如果你只需要 A 股的日线数据和基本面数据做研究baostock 完全够用没必要去折腾那些需要积分和 token 的方案。2. 环境准备安装 baostock 并跑通第一次查询先说一下环境要求。baostock 是基于 Python 的库官方支持 Python 2.7 和 3.x推荐直接用 Python 3.6 以上版本。整个安装和验证过程非常简单基本不用配置什么额外的依赖。2.1 安装命令Windows、macOS 或者 Linux 环境都一样直接用 pip 安装pip install baostock -i https://pypi.tuna.tsinghua.edu.cn/simple这里用清华镜像源主要是为了下载速度快。如果你的网络环境访问 PyPI 官方源没问题直接用pip install baostock也可以。安装完以后可以在 Python 交互环境里验证一下是否导入成功import baostock as bs print(bs.__version__)能正常输出版本号说明安装成功。我遇到过一些特殊情况比如环境里同时存在多个 Python 版本pip 安装到了 3.x 但命令行运行的是 2.x导致模块找不到。这种情况建议用python -m pip install baostock来安装确保装到当前 Python 环境对应的 pip 里。2.2 登录与登出机制baostock 的使用模式是“登录 - 操作 - 登出”。首次使用需要调用login()建立一个会话所有查询操作都在这个会话中进行操作完成后调用logout()释放连接。import baostock as bs # 登录参数留空则使用匿名账户 lg bs.login() print(登录状态:, lg.error_code, lg.error_msg) # 这里写你的业务代码 # ... # 登出 lg bs.logout()很多新手会忽略logout()短时间跑一两次脚本可能感觉不到问题但如果跑批量任务脚本频繁启动、登录连接不释放进程可能会越跑越慢。2.3 第一个查询实例先跑通一个最简单的查询以贵州茅台为例获取它在 2023 年全年的日线数据。import baostock as bs import pandas as pd # 登录 lg bs.login() # 查询历史K线数据 rs bs.query_history_k_data_plus( sh.600519, date,code,open,high,low,close,preclose,volume,amount,pctChg, start_date2023-01-01, end_date2023-12-31, frequencyd, adjustflag2 ) # 将结果转换为 DataFrame data_list [] while (rs.error_code 0) and rs.next(): data_list.append(rs.get_row_data()) result pd.DataFrame(data_list, columnsrs.fields) print(result.head()) # 登出 lg logout()注意这里的股票代码格式baostock 要求带交易所前缀sh.代表上交所sz.代表深交所。一只股票代码同时存在上海和深圳时这种事情确实有通过前缀区分才不会有歧义。adjustflag参数是复权设置这个比较关键参数值含义1后复权2前复权3不复权做技术分析、算指标比如 MACD、均线一般用前复权数据避免除权除息造成的价格跳空计算真实收益率和回测建议用后复权数据更能反映真实收益看原始交易盘口就用不复权。这个参数踩坑率极高很多朋友拿到的数据没有算复权导致后面指标计算错得离谱追了半天 bug 才发现问题出在数据源。3. 批量下载多只股票的核心实现方案跑通单只股票以后批量下载本质上就是加一层循环但这里有几个细节如果不处理好很容易被“看似无关紧要的坑”拖死。3.1 定义股票池和日期范围首先定义一个股票代码列表。可以是自己手工整理的名单也可以是从某个选股策略筛选出的结果。关键是格式要统一都带上交易所前缀。# 示例股票池几只不同行业的代表性股票 stock_list [ sh.600000, # 浦发银行 sh.600036, # 招商银行 sz.000001, # 平安银行 sz.000333, # 美的集团 sh.601318, # 中国平安 sz.300750, # 宁德时代 sh.600519, # 贵州茅台 sz.000858, # 五粮液 sh.601012, # 隆基绿能 sz.002594, # 比亚迪 ]日期范围我一般用start_date和end_date两个变量集中管理方便后面做增量更新。start_date 2020-01-01 end_date 2024-12-313.2 循环查询与 DataFrame 拼接最直接的批量写法是遍历股票池把每只股票的数据先存到一个临时变量最后统一拼接。这里要提个醒query_history_k_data_plus返回的结果是游标式的需要通过while rs.next()逐行读取不能直接像 pandas 那样一把梭。import baostock as bs import pandas as pd import time lg bs.login() # 定义需要获取的字段 fields date,code,open,high,low,close,preclose,volume,amount,pctChg all_data [] for code in stock_list: print(f正在获取: {code}) try: rs bs.query_history_k_data_plus( code, fields, start_datestart_date, end_dateend_date, frequencyd, adjustflag2 ) if rs.error_code ! 0: print(f查询失败: {code}, 错误码: {rs.error_code}, 错误信息: {rs.error_msg}) continue data_list [] while rs.next(): data_list.append(rs.get_row_data()) if not data_list: print(f警告: {code} 在指定日期范围内没有数据) continue df pd.DataFrame(data_list, columnsrs.fields) all_data.append(df) print(f完成: {code}, 共 {len(df)} 条记录) except Exception as e: print(f获取 {code} 失败: {e}) continue # 控制请求节奏避免对服务器造成压力 time.sleep(0.1) # 拼接所有数据 if all_data: result pd.concat(all_data, ignore_indexTrue) print(f全部完成共获取 {len(result)} 条记录) else: print(未获取到任何数据) bs.logout()实际跑起来10 只股票、5 年日线数据总耗时大概 10 秒左右速度是可以接受的。循环里加time.sleep(0.1)是我自己的习惯。虽然 baostock 目前没有严格的频率限制但我们在做批量任务时保持适度的请求间隔是基本的礼貌也能降低被服务端临时限制的风险。3.3 保存数据CSV 还是按股票拆文件批量下载的目的通常是为了后续分析所以数据要落盘。对散户级别的研究来说to_csv是最省事的方案。我通常会按股票代码分别存成一个 CSV 文件而不是所有数据挤在一个大文件里。这样后续加载某只股票的数据时只需读对应文件效率更高。import os output_dir stock_data os.makedirs(output_dir, exist_okTrue) for code in stock_list: # 从 sh.600000 形式转为规范文件名 filename code.replace(., _) .csv filepath os.path.join(output_dir, filename) # 如果之前已经下载过跳过简单断点续传 if os.path.exists(filepath): print(f文件已存在跳过: {filename}) continue # 具体查询逻辑同上这里只展示保存部分 df.to_csv(filepath, indexFalse, encodingutf-8-sig) print(f已保存: {filepath})这里有一个小细节encoding参数我习惯用utf-8-sig这样用 Excel 直接打开 CSV 时不会出现中文乱码。如果你后续主要用 pandas 读取用默认的utf-8也行。3.4 将数据写入 SQLite 统一管理股票数量一旦上百拆成上百个 CSV 文件后查询和管理的成本就上升了。这种场景我更推荐用 SQLite。它是 Python 内置支持的单文件数据库不用额外安装服务非常适合个人量化研究这种中等规模的数据。import sqlite3 conn sqlite3.connect(stock_data.db) cursor conn.cursor() # 创建行情表如果表已存在则忽略 cursor.execute( CREATE TABLE IF NOT EXISTS daily_kline ( date TEXT, code TEXT, open REAL, high REAL, low REAL, close REAL, preclose REAL, volume REAL, amount REAL, pctChg REAL, PRIMARY KEY (date, code) ) ) # 把 DataFrame 写入 SQLite for code in stock_list: # 假设 df 是当前股票的数据 df.to_sql(daily_kline, conn, if_existsappend, indexFalse) conn.commit() conn.close()SQLite 方案的好处有两方面。第一不用维护一堆文件一个.db文件全搞定第二可以直接用 SQL 语句做跨股票、跨日期的条件查询比如“找出 2023 年所有股票的成交量前 100 名”一条 SQL 就出来了用 CSV 就得先把所有文件读一遍再内存计算。4. 工程化升级增量更新与日志记录批量下载能跑通只是第一步。你真把自己的策略用起来之后就会发现数据是每天都在变的。每天手动全量跑一遍下载脚本既浪费时间又容易出错。所以工程化升级里最重要的一件事就是做增量更新。4.1 增量更新的两种思路增量更新的核心思路是“只拿新增的数据别重复拉旧数据”。我实际用过比较顺手的两种方案。方案一按日期段更新。脚本启动时先查数据库里当前股票已有的最大日期然后从最大日期的第二天开始拉到今天。def get_last_date(conn, code): 查询某只股票在数据库里的最大日期 cursor conn.cursor() cursor.execute(SELECT MAX(date) FROM daily_kline WHERE code ?, (code,)) result cursor.fetchone() return result[0] if result and result[0] else None last_date get_last_date(conn, sh.600000) if last_date: start_date (pd.Timestamp(last_date) pd.Timedelta(days1)).strftime(%Y-%m-%d) else: start_date 2020-01-01这种做法的好处是逻辑简单不依赖 baostock 的任何特殊接口纯靠本地数据状态驱动。方案二用交易日历判断。baostock 提供query_trade_dates接口可以拿到所有交易日。我们可以一次性拿到当年的交易日列表然后只拉那些本地库里还不存在的日期。rs bs.query_trade_dates(start_date2025-01-01, end_date2025-12-31)这个接口返回两个字段calendar_date和is_trading_day。经过本地过滤后剩下的才是真正需要补的数据日期。两种方案各有优势我自己的做法是初期数据量小、结构不复杂时用方案一沉淀出一个完整的增量更新框架后切换到方案二可以精确控制补数范围避免某些极端停牌日拉出来空数据。4.2 日志记录与失败重试批量下载几百只股票时完全一次跑通基本不可能。偶发超时、重试、数据为空的情况总会出现。这时候如果没有日志很难知道到底哪些股票成功了、哪些失败了、失败原因是什么。我习惯用一个简单的日志函数把输出同时写到控制台和文件里。import datetime def log(msg): timestamp datetime.datetime.now().strftime(%Y-%m-%d %H:%M:%S) line f[{timestamp}] {msg} print(line) with open(download.log, a, encodingutf-8) as f: f.write(line \n)配合一个失败重试机制对查询失败的股票最多重试 3 次每次间隔 2 秒。如果 3 次都失败就把代码记到一个failed_list里全部跑完之后统一打出来方便人工排查。max_retry 3 failed_list [] for code in stock_list: for attempt in range(max_retry): try: # 查询逻辑... break # 成功后跳出重试循环 except Exception as e: log(f{code} 第 {attempt1} 次尝试失败: {e}) if attempt max_retry - 1: failed_list.append(code) time.sleep(2)跑完以后看到failed_list不是空列表千万不要无视。一定要回去确认这些股票到底是代码格式错了还是已经退市、改名。很多时候所谓的“数据缺失”其实是股票代码变更导致的。4.3 多线程加速的利与弊有些朋友觉得每次都一只只下载太慢想用多线程来加速。我的建议是这个操作要谨慎。baostock 的登录态和 session 管理在某些情况下并不能完全保证线程安全。我实测下来多线程并发请求时偶发会触发连接重置或者数据错乱。如果真要用多线程比较稳妥的做法是只对“登录 → 查询 → 登出”这个单元做进程隔离而不是简单地在同一个 session 里开多线程。或者更直接一点拆成多个脚本每个脚本负责一部分股票每个脚本单独登录、单独运行最后再把结果合并。我的真实感受是对个人研究和中小型策略来说单线程加 0.1 秒延时已经完全够用。100 只股票、5 年数据慢一点的电脑也就两三分钟跑完。与其冒险用多线程去抢那几十秒的时间不如把稳定性放在第一位。毕竟数据下载这种任务跑挂了重来一次浪费的时间远比省下的那点时间多。5. 常见错误与排查技巧速查表以下是这几年用 baostock 过程中我和身边朋友踩过的坑。整理成了一份速查表基本覆盖了日常使用的高频问题。错误现象可能原因解决办法AttributeError: module baostock has no attribute query_history_k_data_plus安装的 baostock 版本过旧或没有正确安装升级到最新版pip install --upgrade baostock登录失败控制台显示login failed网络不通或本机存在代理干扰检查网络临时关闭系统代理后重试查询返回结果为空但没有报错股票代码格式错误少了前缀或该股票已经退市、停牌检查代码是否为sh.或sz.开头验证股票代码是否真实存在字段名错误报错unknown fields查询字符串里写了不存在的字段名核对字段列表date,code,open,high,low,close,preclose,volume,amount,pctChg是常用且稳妥的组合数据里出现大量的nan或空字符串某些字段本身没有值如停牌日无成交量或复权因子数据缺失在数据处理时做fillna(0)或dropna()根据分析需求决定处理方式数据条数和预期不符少了某几天停牌导致的交易日无数据或 baostock 数据源自身不包含部分极端行情用查询到的交易日历比对确认差异是真正的数据缺失还是非交易日多线程并发时连接重置baostock 的 session 不是完全线程安全的避免在同一个进程里开多线程请求改为多进程划分股票池每个进程独立登录RecursionError或内存消耗过大一次查询的股票数量太多、时间跨度太长返回数据量过大分批次查询每只股票单独查询并按时间分段及时释放变量必要时用 SQLite 增量存储5.1 容易被忽略的数据质量问题复权和停牌这里想单独强调一个很多人没注意到的问题复权因子的缺失会在跨日期计算收益时造成严重偏差。举个例子一只股票在某天实施了高送转每 10 股转增 10 股如果你用的是不复权数据当天股价会从 100 元直接跳到 50 元看起来像暴跌了 50%。但实际对持仓者来说股票数量翻倍总资产没有任何变化。如果回测时用了不复权数据策略会被这种假的暴跌信号干扰最终结果完全不可用。baostock 提供了adjustflag参数来解决这个问题。如果你在查询时选择的adjustflag2前复权它返回的价格已经把历史价格按最新股本做了调整避免了除权除息造成的跳空。这个细节是决定量化回测结果是否可信的关键之一。停牌的情况也需要注意。A 股市场个股临时停牌很常见停牌期间没有交易数据。baostock 查询结果是直接跳过这些日期的所以从2024-01-01到2024-12-31按交易日历应该约 242 个交易日但某只股票可能只有 230 条记录少了 12 天。这 12 天里部分是真的停牌部分可能是数据源自身的缺失。怎么区分用 baostock 的query_trade_dates拿到完整的交易日列表和实际数据日期做set的差集差异部分再人工判断是停牌还是缺失。这一招在做全市场轮动策略时特别重要能帮你避免后期因为个别股票数据问题导致无法复现策略结果。5.2 环境迁移时的坑如果你在本地跑通了想部署到服务器或另一台电脑上最大的坑就是环境不一致。这里提供一个快速自检清单Python 版本是否为 3.6 以上baostock 在新版本 Python 下兼容性更好。是否安装了 pandasbaostock 本身不强制依赖 pandas但实际使用中基本离不开它。是否在正确的虚拟环境里安装如果用了 conda 或 venv确保切换到目标环境后再安装和运行。系统时间是否正确baostock 的登录和某些操作依赖系统时间时间偏差过大会导致校验失败。这些看起来都是小事但服务器部署时最容易在环境上卡壳。建议直接把环境依赖写入requirements.txtbaostock0.8.8 pandas1.3.0然后在目标机器上执行pip install -r requirements.txt一次性把环境搭好省得到处找人查错。6. 数据进阶处理从原始行情到可用于策略的数据下载完历史数据只完成了整个流程的一半。原始数据要变成可以输入到策略模型里的干净数据中间还需要几个标准处理步骤。这部分虽然不是 baostock 的功能但却是批量下载后必须面对的问题。6.1 字段类型转换与缺失值处理baostock 返回的数据默认全是字符串类型。你直接对open列做加减乘除会发现报错或者计算结果不对。所以第一步一定是要把数值字段转成float把日期字段转成datetime。# 假设 df 是 baostock 返回的 DataFrame numeric_cols [open, high, low, close, preclose, volume, amount, pctChg] for col in numeric_cols: df[col] pd.to_numeric(df[col], errorscoerce) df[date] pd.to_datetime(df[date])errorscoerce的作用是转换出错时置为NaN而不是直接报错退出。这样即使某一行数据格式有问题也不会中断整个流程。对于NaN值我的处理习惯是如果停牌导致的成交量缺失填充为 0如果是价格类字段出现NaN则优先用前一日价格填充极少数情况下连续多日缺失直接把对应日期从样本里剔除避免引入失真数据。6.2 添加技术指标对历史行情数据做技术分析时常用的指标如 MA、MACD、RSI、KDJ 等都可以直接用 pandas 计算不需要额外安装 TA-Lib 这种重量级库。以 5 日均线和 20 日均线为例df[ma5] df[close].rolling(window5).mean() df[ma20] df[close].rolling(window20).mean()MACD 的实现稍微复杂一点但用 pandas 的ewm函数也能搞定# 计算 MACD ema12 df[close].ewm(span12, adjustFalse).mean() ema26 df[close].ewm(span26, adjustFalse).mean() df[dif] ema12 - ema26 df[dea] df[dif].ewm(span9, adjustFalse).mean() df[macd] (df[dif] - df[dea]) * 2如果你用的是前复权数据计算的指标在除权除息日附近依然连续、平滑反过来用不复权数据指标在除权日会有明显突变策略信号可能完全失真。6.3 多股票数据的合并与截面处理做多股票分析时往往需要把数据整理成“宽表”格式每一行是一个交易日每一列是一只股票的某个字段。pandas 的pivot函数是干这个的利器。# 假设 df_all 是包含多只股票的日线数据字段包括 date, code, close # 转换为以日期为行、股票代码为列的形式 close_pivot df_all.pivot(indexdate, columnscode, valuesclose)这种格式做截面分析特别方便比如每天对所有股票的涨幅排序、计算每日收益率的中位数一行代码就能完成。我自己的经验是数据处理环节最容易出错的地方不是代码逻辑而是对“这只股票现在叫什么、代码是什么”这类信息的变化缺乏追踪。股票改名、退市、重新上市都会在批量下载时制造一些诡异的问题。所以我的股票池管理一直坚持一个原则代码列表定期核对不在一个脚本里“一劳永逸”。7. 后续可以怎么扩展批量下载数据这个能力打通之后你可以做的事情就非常丰富了。多因子选股把所有股票的财务数据和质量数据整合到一起构建因子库做因子筛选和打分。策略回测基于下载的日线数据和复权因子实现简单的均线策略、突破策略或者更复杂的多因子模型验证自己的交易思路。数据可视化结合matplotlib或plotly把股价走势、成交量变化、均线系统画出来辅助做直观判断。每日定时更新配合系统定时任务Windows 的任务计划程序或 Linux 的 crontab让脚本每天收盘后自动拉取最新数据形成一套完整的个人量化数据中台。我个人的体会是baostock 作为数据层工具最大的价值在于让你以极低的成本拿到干净、完整、稳定的历史行情数据从而把精力真正投入到“策略逻辑”和“数据分析”这些核心技术问题上而不是反复纠结数据源是否可靠。把上面分享的批量下载、增量更新、错误排查这套流程跑通你就能拥有一个属于自己的、可持续更新的 A 股行情数据库。补充一个长期使用下来最重要的心得批量任务一定要做成“可重复执行、可中断续跑”的形式千万别追求一次性脚本。加一个文件存在性判断、加一个失败重试、加一条日志记录前期只花十几分钟后期能帮你省下几十个小时的排查时间。