简介基于Baostock金融数据接口的K线数据自动化下载与本地存储工具面向股票分析师、量化研究者及普通投资者解决多市场指数与A股全量股票历史K线数据获取不便的问题。工具通过Python脚本调用Baostock接口支持上证指数、深证成指、沪深300、创业板指等主要指数的日K线与5分钟K线数据抓取同时覆盖A股所有上市股票的对应周期数据并将结果自动保存至本地便于后续分析。资源包共4个文件以main.py主脚本为核心配套README.md说明文档、说明文件txt及附赠资源docx整体约36KB结构精简适合直接参考或二次修改。已有146人学习下载适用于搭建个人行情数据仓库、策略回测前的数据准备以及入门金融数据接口调用等场景。1. 先把话说清楚这个工具解决的是“要数据时永远差一截”的问题做量化回测或者指数分析的人大多经历过这种场景想验证一个多因子策略打开行情软件发现本地只有日线5分钟数据要手动补补到一半又断网想拉全市场5000多只股票的K线跑了一整夜第二天发现少了几百只更麻烦的是不同来源的数据口径还不一样前复权、后复权、不复权混在一起回测结果根本没法对齐。基于Baostock金融数据接口的这套下载与本地存储方案就是为了把这些零散的痛点收拢成一个可重复执行的工程化流程。它会覆盖上证指数、深证成指、沪深300、创业板指等主要市场指数以及A股全部上市股票的日K线和5分钟K线历史数据最终落到本地目录和压缩包里后续做因子计算、复盘、回测都能直接从本地读取不再依赖网络。这套东西适合谁适合正在做个股或指数策略研究的从业者也适合刚接触量化、不想在数据采集上反复折腾的新手。Baostock本身是免费接口没有授权费也不限制个人研究用途但接口的调用方式偏底层需要自己处理会话、字段、增量更新和容错。这篇文章就把这些脏活讲透从接口原理到可复现的脚本最后到避坑经验一次讲完。2. 原理与选型Baostock的数据口径和K线接口参数2.1 登录会话、交易日历与数据边界Baostock的API是一个Python库使用时先调用bs.login()建立会话结束再bs.logout()。这个会话不需要账号密码但每次脚本运行都应该显式登录和退出不要想着一次登录长期挂着。保持长会话在很多情况下会触发服务端断开连接之后所有查询返回错误码但你的脚本不会主动报错只会返回空DataFrame最后写进本地文件的全是空表这个坑我在实践中碰到过不止一次。交易日历方面Baostock提供query_trade_dates(start_date, end_date)返回字段是calendar_date和is_trading_day。在规划本地存储之前我强烈建议先把交易日历下载下来存一份之后做数据完整性校验全靠它。比如沪深300成分股的日K线缺了某一天你要能判断这一天到底是停牌还是非交易日否则很容易把停牌误判成缺失。数据边界也要提前想清楚。Baostock对单次查询的历史深度有内部限制拉全市场个股的时候不能无限往回追溯。实际处理中我会把起止日期按年分段每段单独查询避免一次请求数据量过大被截断或超时。比如5分钟K线单只股票一年的数据量就有上万行全市场叠加起来对内存和磁盘的压力都不小分段拉取是最稳妥的方案。2.2 日K与5分钟K线的字段差异和频率参数日K线查询用query_history_k_data_plus核心参数是fields、start_date、end_date、frequency和adjustflag。常见字段包括字段含义备注date交易日期格式YYYY-MM-DDcode证券代码带交易所前缀如sh.600000open / high / low / close开高低收浮点数preclose昨收可用于计算收益率volume / amount成交量 / 成交额volume单位是股amount单位是元turn换手率百分比tradestatus交易状态1表示正常交易pctChg涨跌幅百分比isST是否ST1是0否adjustflag复权标记1后复权、2前复权、3不复权adjustflag是很多人在初始化数据时最纠结的参数。做回测通常用前复权因为价格连续做事件分析或分红统计用后复权更合理做技术指标拟合用不复权也能接受。我的习惯是本地分别存一份不复权和一份前复权不复权作为原始底稿前复权作为计算层这样后续切换口径不需要重新下载。5分钟K线调用方式类似frequency5即可。但要注意5分钟数据的字段和日K不同它没有preclose、turn、tradestatus这些日频衍生字段只有date、time、code、open、high、low、close、volume、amount、adjustflag。其中time字段的格式是YYYYMMDDHHMMSSsss毫秒部分通常是000解析时要用datetime.strptime(row[time], %Y%m%d%H%M%S%f)这类方式处理不能直接当字符串截取。2.3 常用指数代码映射与全市场股票列表获取指数代码在Baostock里是带交易所前缀的字符串这个前缀不能省。主要指数映射如下指数名称Baostock代码上证指数sh.000001深证成指sz.399001沪深300sh.000300创业板指sz.399006上证50sh.000016中证500sh.000905注意上证指数是sh.000001和A股第一只股票sh.600000完全不一样创业板指是sz.399006不是sz.300xxx开头的个股。写脚本时建议把这些映射关系放进字典避免每次手敲出错。全市场股票列表通过query_all_stock(day)获取参数day是日期字符串。返回值里有一个tradeStatus字段1表示当日正常交易0表示停牌或退市。如果只取当日可交易股票会漏掉历史退市和长期停牌的标的后面做全市场回测时样本就有偏差。更好的方式是维护一个静态股票池每天增量更新把所有出现过的代码都保存下来再逐个拉历史K线。这样即使某只股票后来退市历史数据也完整。这一点我在后面避坑章节里还会展开。3. 自动化下载与本地存储从指数到全市场股票的落地脚本3.1 目录规划与文件命名约定本地目录规划是我的习惯动作它决定了后面所有脚本读数据时逻辑是否清晰。我常用结构是data/ ├── index/ │ ├── daily/ │ └── minute_5/ ├── stock/ │ ├── daily/ │ └── minute_5/ ├── calendar/ ├── logs/ └── archive/文件名用代码去掉点号的方式命名比如sh_000300.csv、sz_399006.csv。不建议直接用sh.000300.csv因为Windows和Linux对点号的处理虽然一致但后续用shell脚本批量操作时点号可能被某些工具当作扩展名分隔符产生不必要的麻烦。每只股票的日K文件里再固定包含code列即使文件名改了也能从内容追溯。日历文件存在calendar/下文件名就叫trade_calendar.csv字段定义为calendar_date,is_trading_day。这文件只在初始化时下载一次之后每周更新一次即可。3.2 指数日K下载脚本指数日K的下载逻辑最简单代码量也少先把它跑通能快速验证环境是否正常。下面是我实际在用的脚本骨架import baostock as bs import pandas as pd import datetime from pathlib import Path # 指数代码与中文名映射 INDEX_LIST { sh.000001: 上证指数, sz.399001: 深证成指, sh.000300: 沪深300, sz.399006: 创业板指, sh.000016: 上证50, sh.000905: 中证500, } OUTPUT_DIR Path(data/index/daily) OUTPUT_DIR.mkdir(parentsTrue, exist_okTrue) lg bs.login() print(login:, lg.error_code, lg.error_msg) today datetime.date.today().strftime(%Y-%m-%d) fields date,code,open,high,low,close,volume,amount for code, name in INDEX_LIST.items(): rs bs.query_history_k_data_plus( code, fields, start_date1990-01-01, end_datetoday, frequencyd, adjustflag3, ) if rs.error_code ! 0: print(f[error] {name}: {rs.error_msg}) continue df rs.get_data() if df.empty: print(f[empty] {name} no data) continue out_path OUTPUT_DIR / f{code.replace(., _)}.csv df.to_csv(out_path, indexFalse, encodingutf-8-sig) print(f[done] {name}: {len(df)} rows - {out_path}) bs.logout()这个脚本有几处值得说明。adjustflag3表示不复权指数本身没有复权概念所以这里固定填3。start_date从1990年开始目的是让脚本覆盖指数发布前的空区间Baostock对空区间会返回空DataFrame而不是报错所以不用刻意改起始日期。rs.error_code ! 0是判断查询是否成功的标准方式如果接口参数写错这一步能直接打印出原因避免把空数据写进文件。文件用utf-8-sig编码保存这是给Excel用户留的后路。后面用pandas或第三方库读数据时utf-8-sig也能正常读属于兼容性最好的选择之一。如果后面发现某只指数下载的行数和预期差异很大优先检查交易日历而不是怀疑脚本本身。3.3 全市场个股日K下载与增量更新全市场个股的日K下载比指数麻烦在两点一是股票数量多二是需要断点续下。我常写的逻辑是先构建股票代码列表然后逐个查询查询时优先跳过已经存在且非空的文件实现增量更新。import baostock as bs import pandas as pd import datetime import time import os from pathlib import Path DATA_DIR Path(data/stock/daily) DATA_DIR.mkdir(parentsTrue, exist_okTrue) bs.login() # 获取当前交易日可交易股票 today datetime.date.today().strftime(%Y-%m-%d) rs bs.query_all_stock(daytoday) stock_codes [] while rs.next(): row rs.get_row_data() if row[2] 1: # tradeStatus stock_codes.append(row[0]) stock_codes sorted(stock_codes) print(total tradable stocks:, len(stock_codes)) fields date,code,open,high,low,close,preclose,volume,amount,turn,tradestatus,pctChg,isST for i, code in enumerate(stock_codes, start1): out_file DATA_DIR / f{code.replace(., _)}.csv # 增量更新已存在且非空则跳过 if out_file.exists() and out_file.stat().st_size 100: print(f[skip] {code} already exists) continue rs bs.query_history_k_data_plus( code, fields, start_date1990-01-01, end_datetoday, frequencyd, adjustflag2, # 前复权 ) if rs.error_code ! 0: print(f[error] {code}: {rs.error_msg}) time.sleep(1) continue df rs.get_data() if not df.empty: df.to_csv(out_file, indexFalse, encodingutf-8-sig) print(f[{i}/{len(stock_codes)}] {code} {len(df)} rows) else: print(f[empty] {code} no data) time.sleep(0.2) # 控制请求频率 bs.logout()这段代码的增量更新逻辑依赖文件大小大于100字节作为“已下载”的判断条件因为正常股票即使只有几十个交易日CSV文件也会超过1KB。如果某只股票文件存在但下载中断导致只有表头没有数据文件大小会是几十字节就会被重新下载。这个阈值可以根据实际数据量调整不要设成0否则空文件会一直被跳过。全市场股票列表来自query_all_stock它的局限是只返回查询日当天可交易的股票。历史退市和长期停牌个股不会出现在列表里所以如果要建全历史股票池需要另写一个维护脚本定期从历史交易日逐个查询并合并代码集合。不过对于大多数个人研究者先覆盖当前正常交易股票已经足够退市股票的分析属于低频场景。3.4 5分钟K线下载与节流控制5分钟K线下载是整套方案里最容易翻车的部分。量级感要先建立起来日K大约一年240条记录5分钟K一年大约是48倍也就是大约11520条。单只股票还好全市场5000只股票一年的5分钟数据就是几千万行存成CSV要到几十GB。所以5分钟数据的下载策略必须和日K不一样不能一股脑全市场全历史拉取。我的建议是分两步走。第一步确定范围只下载最近N个月或者只下载某个指数池子里的成分股不要一上来就全市场。第二步分段下载每只股票按季度拆分区间每段之间sleep至少0.3秒。import baostock as bs import pandas as pd from pathlib import Path import time import datetime OUTPUT_DIR Path(data/stock/minute_5) OUTPUT_DIR.mkdir(parentsTrue, exist_okTrue) bs.login() code sh.600000 # 示例个股 start 2024-10-01 end 2024-12-31 # 5分钟K线字段 fields date,time,code,open,high,low,close,volume,amount,adjustflag rs bs.query_history_k_data_plus( code, fields, start_datestart, end_dateend, frequency5, adjustflag3, ) if rs.error_code ! 0: print(error:, rs.error_msg) else: df rs.get_data() print(rows:, len(df)) # 时间字段解析 df[trade_time] df[time].apply( lambda x: datetime.datetime.strptime(x, %Y%m%d%H%M%S%f) ) out_path OUTPUT_DIR / f{code.replace(., _)}_2024Q4.csv df.to_csv(out_path, indexFalse, encodingutf-8-sig) print(saved:, out_path) bs.logout()adjustflag3在5分钟数据下基本是唯一正常的选择因为Baostock的5分钟K线不支持前复权计算强行用复权结果会出现数值漂移。如果需要前复权的5分钟数据只能自己用日线复权因子对分钟价格做处理这个动作放到后续数据预处理阶段更合适。另外5分钟数据天然包含集合竞价和收盘集合竞价的记录吗Baostock的时间轴是从9:30开始到11:30结束下午从13:00到15:00但不同数据源对尾盘那根K线的归属有差异所以下载后要主动过滤时间范围避免把不属于连续竞价时段的异常点混进因子计算。具体过滤方法在避坑章节里讲这里先记住“下载不等于干净”。4. 避坑排查从全量下载到增量更新的5个典型问题4.1 现象一第一天下载正常第二天大量股票返回空数据现象描述脚本前一天还能正常拉数据第二天重启后上千只股票返回空DataFrame文件里只有表头。一开始容易怀疑是Baostock接口挂了但其实是自己的调用方式踩了坑。原因分析Baostock的登录会话在多次请求后会被服务端主动断开或者触发限流。断开后继续调用query_history_k_data_plus不会抛异常而是返回error_code非0或空DataFrame。这时候脚本还按正常流程写入文件就会把好数据覆盖成空表。解决办法每500到800次请求主动bs.logout()再bs.login()每次查询后都要检查rs.error_code不是0就跳过并记录日志不要写文件另外把单次下载的股票数量控制在200只以内跑一批批与批之间sleep 1秒。老脚本里bs.login()只调用一次直接跑全市场这种写法在网络波动时非常脆弱。4.2 现象二指数代码一直提示参数错误现象描述用000001或sh000001查询上证指数返回error_msg提示参数错误。原因分析Baostock对证券代码有严格要求必须是小写的交易所前缀加点号加数字比如sh.000001、sz.399001。去掉点、大写字谜、在指数代码后面加.XSHG后缀都会导致查询失败。这个问题经常在大规模写代码时出现因为股票代码通常不带前缀容易下意识复制过来用。解决办法指数代码建一个映射表直接抄不要用手拼字符串的方式生成。个股代码如果要带前缀可以用sh.加6位数字、sz.加0/3开头数字的规则做转换但最好也先打印几个样本核对再批量跑。4.3 现象三5分钟K线的时间轴怎么都对不上交易时段现象描述下载某只股票的5分钟K线后发现数据里出现9:25、9:30、11:30、15:00这些时间点和通达信或同花顺里的K线图对不上要么多一根要么少一根。原因分析不同数据商对5分钟K线的边界处理不统一。有的把9:25集合竞价算作一根有的把11:30的收盘K线和下午13:00的开盘K线合并处理有的则是不加任何过滤直接落盘。Baostock返回的是按自然时间生成的K线轴它对边界K线的处理方式和某些行情软件不一致所以本土数据源反而要对齐。解决办法下载后统一过滤时间只保留9:30到11:30、13:00到15:00区间的K线然后按“每根K线的时间标记为该区间终点”的方式重采样。比如9:30-9:35的数据标记为9:35标准的5分钟周期不会有9:30这个起始标记除非收盘那根特殊。具体过滤代码可以用pandas.between_time实现效率高又能直接对齐周期。4.4 现象四CSV文件用Excel打开乱码用pandas读取报列名不一致现象描述脚本跑完CSV文件在终端用cat看是中文表头用Excel打开全是乱码或者pandas读出来的列名是date但文件里第一行看起来是date第二行却开始错位。原因分析这是典型的编码不一致问题。utf-8编码的中文在Excel里不会被自动识别Excel在Windows下默认用ANSI编码打开CSV所以乱码pandas读取时如果指定encodingutf-8、文件尾却带了BOM列名解析也会出问题。解决办法保存CSV时统一用encodingutf-8-sig它会写入BOM头Excel能正确识别pandas读取时用encodingutf-8-sig也不会报错。另外在所有脚本里固定字段顺序不要依赖列名动态索引否则字段顺序一变之前存的CSV就全废了。4.5 现象五下载到一半进程被杀重启后不知道断在哪现象描述全市场股票下载跑到3000多只时电脑休眠或断电进程被杀。重启脚本后从第一只开始跑之前的进度全丢了。或者是脚本跳过已存在文件但有些文件只写了一半被误判为已完成。原因分析没有记录断点也没有对“完成”状态做严格定义。文件存在但内容不完整大小判断失效文件里只有表头没有数据的空文件也会被认为是已下载。解决办法引入一个done标记文件每只股票写完CSV后额外写一个同名的.done空文件。增量判断只看.done文件是否存在不存在就重新下载。这个方案逻辑简单但非常可靠比单靠文件大小判断要严谨得多。同时脚本启动后先扫描已有.done数量打印“已完成/总数”的比例一眼就能看出进度。5. 存储与校验把海量CSV整理成可回放的本地数据集5.1 交易日历与数据完整性校验本地数据攒多了以后最怕的不是没数据而是数据悄悄缺了一天。K线自己不会告诉你哪根是缺的只有拿交易日历比对才能发现。所以我建议在批量下载完成后加一层基于交易日历的校验逻辑。import pandas as pd from pathlib import Path CAL_FILE Path(data/calendar/trade_calendar.csv) INDEX_FILE Path(data/index/daily/sh_000300.csv) # 读取交易日历 cal pd.read_csv(CAL_FILE) trade_days set(cal.loc[cal[is_trading_day] 1, calendar_date].tolist()) # 读取指数K线 df pd.read_csv(INDEX_FILE) data_days set(df[date].tolist()) # 取交集范围 common_days sorted(trade_days data_days) if common_days: start_day, end_day common_days[0], common_days[-1] expected set(cal.loc[ (cal[is_trading_day] 1) (cal[calendar_date] start_day) (cal[calendar_date] end_day), calendar_date ].tolist()) missing sorted(expected - data_days) print(missing days:, len(missing)) print(missing[:20])这段代码先找到K线数据和交易日历都有记录的范围再在这个范围内查找缺失日期。如果缺失日期集中在早期或者个股停牌期那是正常的如果最近几个月频繁缺失说明增量更新逻辑出了问题。对于指数数据理论上不应该有缺失因为指数是连续计算的一旦发现缺日期优先怀疑下载脚本的跳过逻辑有bug。5.2 收盘价极值校验与复权一致性检查数据不是下载完就能直接用还要做最基础的四价一致性检查。某次我在做因子预处理时发现一只股票某天的low比high还高追溯下去是下载过程中CSV被截断导致的。增加一个校验脚本能提前暴露这类问题。import pandas as pd from pathlib import Path file_path Path(data/stock/daily/sh_600000.csv) df pd.read_csv(file_path) # 四价逻辑检查 bad df[(df[high] df[low]) | (df[close] df[high]) | (df[open] df[low])] print(bad rows:, len(bad)) if not bad.empty: print(bad[[date, open, high, low, close]].head()) # 复权后价格不能为负 neg df[df[close] 0] print(non-positive close:, len(neg))前复权数据在早期可能出现接近0甚至负数的价格这是复权计算的正常现象因为当时股价经过多次分红除权后折算到当前基准。但如果close普遍为负且数量巨大大概率是复权参数用错或字段错位需要重新下载。这个脚本可以作为批量任务每次下载完自动执行只把异常结果打印出来不阻塞流程。5.3 数据压缩归档与按周期目录管理数据量积累到一定程度磁盘占用会开始影响日常使用。CSV文件本身没有压缩5分钟K线全市场一年就要消耗几十GB归档压缩是必须的一步。我通常用Python内置的zipfile做不额外依赖系统命令import zipfile from pathlib import Path SRC_DIR Path(data/stock/daily) ARCHIVE_PATH Path(data/archive/stock_daily_2024.zip) with zipfile.ZipFile(ARCHIVE_PATH, w, zipfile.ZIP_DEFLATED) as zf: for i, csv_file in enumerate(SRC_DIR.glob(*.csv), start1): zf.write(csv_file, arcnamecsv_file.name) if i % 1000 0: print(fpacked {i} files)归档完后用ZipFile.testzip()检查一遍完整性这步别省。压缩后的文件不删除原CSV只是多一份备份等真正要跑大规模回测时直接从压缩包读取比从几千个分散文件里读要快得多。至于按周期管理我的习惯是日K按年归档5分钟K按季度归档文件名带周期范围比如stock_daily_2024.zip、stock_minute5_2024Q4.zip。6. 进阶把下载工具变成日常维护脚本数据下载完成只是第一步真正的价值在于让这个流程自动化地持续运转。定时增量任务我一般安排在收盘后半小时到一小时比如下午4点后因为此时当天数据已经定型避免盘中数据回刷造成本地文件多次重写。定时任务本身用系统自带机制就行不需要额外搭服务。30 16 * * 1-5 cd /path/to/stock_downloader python download_daily.py logs/download.log 21日志要按周滚动不能无限追加进同一个文件否则几个月后排查问题要从几十万行日志里找线索。我习惯在下载脚本里写入每批任务的起始时间、股票总数、成功数、失败数和耗时这些字段后面做稳定性分析时非常有用。有了稳定更新的本地数据接下来就是和策略计算无缝衔接。这里有个细节值得专门说前复权数据在每次分红除权后历史价格都会发生偏移所以如果本地存储的是前复权数据持续累积快一年后早期数据会和新数据不一致需要对整段历史重新执行一次前复权刷新。更省事的做法是本地始终保存不复权原始数据需要前复权时再实时计算这样原始数据永远不做覆盖可回溯性最好。代价是每次计算前多一步复权处理但对于本地数据量来说这个耗时完全可以接受。我早期做下载工具时一股脑把全市场5分钟K线拉到底结果磁盘空间告急中途数据还因为字段乱序混了几个月后来才意识到问题不在接口而在工程习惯。现在我的顺序永远是先定周期再定复权规则最后写存储逻辑。这个习惯帮我省了无数次返工的麻烦希望帮到你。本文还有配套的精品资源点击获取