简介面向金融数据分析与量化研究场景这份工具包基于Baostock金融数据接口实现股票与指数K线数据的自动化下载与本地存储。资源覆盖上证指数、深证成指、沪深300、创业板指等主要市场指数以及A股全部上市股票的日K线与5分钟高频历史数据可大幅提升数据获取效率降低人工整理成本适合金融分析师、量化研究者及有一定编程基础的投资爱好者使用。压缩包共4个文件包括Python脚本、Markdown说明文档、文本说明文件和Word附赠文档总大小36KB脚本负责自动抓取数据文档提供接口调用与字段说明整体轻量易用。当前已有146人学习使用者可获得完整下载脚本、清晰的操作指南以及附赠的补充资料能够按需调整并快速搭建本地K线数据库。1. 把A股K线一键落到本地这份Baostock工具包解决的是最烦的那步做量化回测的人最烦的不是写策略而是每天都在数据获取上浪费时间。Baostock这个免费接口可以拉取上证指数、深证成指、沪深300、创业板指等主要市场指数的日K和5分钟K线也能覆盖A股市场的股票历史行情。配合自动化的下载脚本和本地存储方案数据能直接落到CSV文件给回测引擎读取。这份资源把登录、取数、落盘、增量更新的流程都整理成了可改参运行的脚本适合想用历史K线做策略验证、技术指标研究或数据备份的开发者直接上手。2. Baostock接口选型为什么免费接口反而适合自动化回测2.1 选它的核心理由登录无状态、接口直白、不怕token失效做数据下载脚本最怕的就是第三方平台的token过期要重新申请、积分不够被限流。Baostock的登录方式很特别bs.login()不需要注册账号也能拿到一个匿名会话返回结果里的error_code为0就表示登录成功。它没有web端那一套复杂的鉴权流程非常适合写进定时任务、无人值守的批量脚本里。核心取数接口是query_history_k_data_plus一个函数就把股票、指数、日线、分钟线全部覆盖。参数走的是start_date end_date的区间模式对增量更新很友好——你想补最近一个月的数据直接把开始日期写成上个月末就行不需要自己维护游标或者翻页逻辑。返回的结果集是一个带有next()方法的游标对象遍历方式和读取文件流类似写起来很顺手。接口的免费策略对个人量化爱好者来说基本够用。相比那些需要付费才能拉全历史数据的平台Baostock在日K和部分分钟线上的数据完整性并不差而且对调用频率的限制相对宽松。只要程序里做好sleep限速跑全市场几千只股票的日K线是没问题的。最让我满意的一点是数据字段返回的是纯字符串列表配合pandas.DataFrame(rows, columnsrs.fields)一行就能转成结构化表格清洗成本极低。2.2 频率、复权与字段入库之前先想清楚这三件事下载K线之前有三个参数直接决定数据能不能用于回测分别是frequency、adjustflag和字段列表。frequency取值d是日线5m是5分钟线。日线数据历史范围长适合策略回测5分钟数据量是日线的几十倍适合做日内策略但下载时间长、存储占用大两者最好分开跑脚本。adjustflag是复权选项——1后复权、2前复权、3不复权。做技术指标计算比如均线、MACD、布林带一定要用复权数据否则除权除息日会出现跳空假信号。做因子分析或分红研究才考虑用不复权。字段选择上推荐一个通用组合字段含义备注date交易日日线格式2024-01-15time交易时间仅分钟线有值格式14:35:00code证券代码带交易所前缀如sh.600000open/high/low/close开高低收字符串需自行转floatvolume成交量股数留意单位amount成交额元pctChg涨跌幅百分比数值turn换手率仅股票有值指数为空指数数据和股票数据在turn字段上存在差异全市场统一入库时建议保留同一套字段列表避免不同文件列数不一致后面合并数据时还要做列对齐。2.3 代码命名规则先看懂前缀再写循环Baostock的代码格式是「交易所前缀 数字代码」。上证指数是sh.000001深证成指是sz.399001沪深300是sh.000300创业板指是sz.399006。个股方面上交所股票是sh.600xxx、sh.601xxx、sh.688xxx深交所是sz.000xxx、sz.002xxx、sz.300xxx北交所是bj.8xxxxx。手动拼代码容易出错好在接口提供了query_all_stock(day)方法传入一个交易日就能返回当天所有在市交易的证券代码和名称。返回的code字段已经带好了前缀可以直接塞进query_history_k_data_plus作为参数不需要自己判断交易所。批量下载脚本的核心逻辑就是「先取列表再遍历拉K线」这个链路我接下来会在第3章完整展开。3. 复现批量下载登录、拉指数、拉全市场股票的完整脚本3.1 环境准备与最小可用代码先把依赖装上。建议用Python 3.8以上版本配合pandas处理数据。装好依赖后第一步永远是登录拉一个标的验证连通性不要一上来就跑全市场。pip install baostock pandas下面这段是验证连通性的最小代码也是后续所有下载任务的基础骨架import baostock as bs import pandas as pd # 1. 登录不需要任何凭证 lg bs.login() print(login:, lg.error_code, lg.error_msg) # 2. 取上证指数日K rs bs.query_history_k_data_plus( sh.000001, date,code,open,high,low,close,preclose,volume,amount, start_date2020-01-01, end_date2024-12-31, frequencyd, adjustflag2, # 前复权 ) # 3. 遍历游标收集数据 rows [] while (rs.error_code 0) and rs.next(): rows.append(rs.get_row_data()) # 4. 转DataFrame并落盘 df pd.DataFrame(rows, columnsrs.fields) print(df.head()) df.to_csv(sh000001_d.csv, indexFalse) # 5. 退出登录释放连接 bs.logout()这段代码的逻辑分五步登录获取会话、调用查询接口、用while next()游标方式逐行读取返回结果、将字符串行列表组装成DataFrame、最后存成CSV并退出登录。rs.error_code是判断查询是否成功的核心标识只有等于0才说明数据返回正常rs.fields保存的是当前查询的字段名列表用它做列名永远不会跟字段参数写错顺序。参数上需要注意start_date和end_date格式必须严格是YYYY-MM-DDpandas解析时少零补零都容易出错adjustflag2用的是前复权如果拿不准该用哪种在第5章我会给出选型建议。3.2 拉取全市场股票列表拿到基础标的清单单标的验证通过后就要解决「下载哪些股票」的问题。手工维护一份代码列表很不现实推荐每天或每月用query_all_stock拉一次当时在市交易的标的清单既包含全部A股代码也能顺带看到交易状态。import baostock as bs import pandas as pd bs.login() # 指定一个交易日获取当天所有在市交易的证券 rs bs.query_all_stock(day2024-12-31) stock_list [] while (rs.error_code 0) and rs.next(): row rs.get_row_data() # row[0]是带前缀的代码row[1]是交易状态row[2]是名称 if row[1] 1: # 只保留正常交易状态 stock_list.append({code: row[0], code_name: row[2]}) df_stocks pd.DataFrame(stock_list) df_stocks.to_csv(stock_list_20241231.csv, indexFalse, encodingutf-8-sig) print(valid stocks:, len(df_stocks)) bs.logout()query_all_stock返回的每一行有三个字段证券代码、交易状态、证券名称。tradeStatus等于1表示正常交易退市或长期停牌的标的标记为0过滤逻辑就写在循环里。将列表存成CSV的好处是后续断点续传时可以直接读文件重新开始不用再请求一次接口。这份列表文件就是第八次跑全市场下载时的「任务清单」。需要说明的是列表里的代码已经带交易所前缀比如sh.600519、sz.000001后续下载K线时直接作为第一个参数传入即可千万不要自己再拼一次前缀很容易出现sh.sh.600519这种低级错误。3.3 全市场批量下载带进度、带限速、带错误记录有了任务清单接下来就是全市场批量下载。这里我给出的做法是单标的封装成函数循环遍历列表每个标的之间做小延时失败标的写入日志后继续绝不中断。import time import random import os import baostock as bs import pandas as pd DATA_DIR output/daily LOG_FILE output/download_errors.log os.makedirs(DATA_DIR, exist_okTrue) FIELDS date,code,open,high,low,close,preclose,volume,amount,pctChg,turn def download_one(code, start_date, end_date, adjustflag2): 下载单只股票日K成功返回行数失败返回0 rs bs.query_history_k_data_plus( code, FIELDS, start_datestart_date, end_dateend_date, frequencyd, adjustflagadjustflag, ) if rs.error_code ! 0: print(query error:, code, rs.error_msg) return 0 rows [] while rs.next(): rows.append(rs.get_row_data()) if rows: df pd.DataFrame(rows, columnsrs.fields) # 统一用代码命名文件方便后续快速读取 safe_code code.replace(., _) df.to_csv(os.path.join(DATA_DIR, f{safe_code}.csv), indexFalse, encodingutf-8-sig) return len(rows) # 登录一次 lg bs.login() print(login:, lg.error_code, lg.error_msg) # 读任务清单 tasks pd.read_csv(stock_list_20241231.csv) # 遍历下载 errors [] for i, row in tasks.iterrows(): code row[code] try: n download_one(code, 2020-01-01, 2024-12-31) if n 0: errors.append(code) # 每200个标的打印一次进度 if (i 1) % 200 0: print(fprogress: {i1}/{len(tasks)}, last{code}, rows{n}) except Exception as e: errors.append(code) print(exception:, code, e) # 随机延时避免短时间请求过于密集 time.sleep(random.uniform(0.05, 0.2)) # 记录失败清单 with open(LOG_FILE, w) as f: f.write(\n.join(errors)) bs.logout()这个批量脚本的设计要点有三个一是单标的下载函数独立封装出错了只影响当前标的不会让整个循环崩溃二是进度打印以200个为间隔跑全市场时能直观看到剩余时间不用心里没底三是失败清单单独落盘等全部跑完后再针对error.log里的代码单独重试效率比中途打断重跑高得多。延时参数random.uniform(0.05, 0.2)是我常用的值。延时太小容易触发服务端限流太大则全市场几千只股票要跑好几个小时。实际跑的时候如果发现连续报错优先加大延时而不是反复重试——这是数据采集类脚本的通病越急越容易被限制。4. 本地存储与增量更新让历史数据只下载一次的落盘方案4.1 目录结构按数据类型分目录别把几千个CSV混在一起全市场数据下载完成后目录组织决定了后面写策略时读数据的效率。我习惯按「数据类型/代码」两级目录来组织日线和分钟线分开存储指数和个股分开存储元数据单独放一层。output/ ├── daily/ # 个股日K线 │ ├── sh_600000.csv │ └── sz_000001.csv ├── index/ # 指数日K线 │ ├── sh_000001.csv │ ├── sz_399001.csv │ ├── sh_000300.csv │ └── sz_399006.csv ├── minute5/ # 5分钟K线 │ └── sh_600000.csv ├── meta/ │ └── download_log.json └── stock_list_20241231.csv文件名里的点号替换成下划线是为了避免不同工具解析文件路径时把.sh当成扩展名。目录分层的核心价值是回测时可以直接按目录批量读文件比如只读output/index下的指数数据做大盘择时不用遍历几千个个股文件。download_log.json记录每次下载的起始日期、结束日期、股票数量和耗时方便下次增量的时间断点判断。这个文件不是必须的但强烈建议维护尤其在增量更新时能帮你快速定位「上次跑到哪了」。4.2 增量更新实现先读本地最新日期再只拉缺口区间如果每次更新都把全历史重拉一遍既浪费带宽又容易被接口限流。正确做法是先读本地CSV的最后一条日期从那个日期开始往后拉然后拼接去重。import os from datetime import datetime import baostock as bs import pandas as pd DATA_DIR output/daily def get_local_max_date(code): 读取本地CSV的最后交易日期文件不存在返回None path os.path.join(DATA_DIR, f{code}.csv) if not os.path.exists(path): return None df pd.read_csv(path, usecols[date]) return df[date].max() # 日期是YYYY-MM-DD字符串可直接比较 def incremental_download(code): local_max get_local_max_date(code) today datetime.now().strftime(%Y-%m-%d) # 本地数据已是最新直接跳过 if local_max and local_max today: return False start local_max if local_max else 2015-01-01 rs bs.query_history_k_data_plus( code, date,code,open,high,low,close,preclose,volume,amount,pctChg,turn, start_datestart, end_datetoday, frequencyd, adjustflag2, ) if rs.error_code ! 0: raise RuntimeError(f{code} query failed: {rs.error_msg}) rows [] while rs.next(): rows.append(rs.get_row_data()) if not rows: return False new_df pd.DataFrame(rows, columnsrs.fields) path os.path.join(DATA_DIR, f{code}.csv) if local_max is not None: old_df pd.read_csv(path) # 拼接后按日期去重保留最新数据 merged pd.concat([old_df, new_df], ignore_indexTrue) merged merged.drop_duplicates(subset[date], keeplast) merged merged.sort_values(date).reset_index(dropTrue) merged.to_csv(path, indexFalse, encodingutf-8-sig) else: new_df.to_csv(path, indexFalse, encodingutf-8-sig) return True这段代码的逻辑核心是「用字符串日期做比较」。YYYY-MM-DD格式下字符串比较和日期比较结果一致所以不需要转成datetime对象。drop_duplicates(subset[date], keeplast)保证了即使起始日期和本地已有数据重叠最终入库数据也不会有重复K线。增量更新的边界条件有两个第一如果本地最新日期恰好等于今天说明数据已经是最新直接返回。第二如果文件不存在则从2015-01-01拉全量历史。这里的起始日期按你自己的需求改想拉更长历史就改成2000-01-01。还要注意增量更新最好在收盘后执行盘中拉到的当日K线只有未收盘的临时数据混入历史库会污染回测结果。4.3 失败重试与日志把断点续传做成标配能力跑全市场增量时网络抖动、接口超时几乎必然发生。我的经验是不要指望一次跑完而是把「失败重试」和「进度记录」做进脚本里。import time import json def run_with_retry(code, max_retry3, wait5): for attempt in range(1, max_retry 1): try: incremental_download(code) return True except Exception as e: print(fattempt {attempt}/{max_retry} failed: {code}, {e}) time.sleep(wait * attempt) # 退避等待5s, 10s, 15s return False # 读取上次失败的清单继续跑 fail_list [] if os.path.exists(meta/failed.json): with open(meta/failed.json) as f: fail_list json.load(f) for code in fail_list: ok run_with_retry(code) if not ok: with open(meta/failed_again.json, a) as f: f.write(code \n)重试采用线性退避策略第一次失败等5秒第二次等10秒第三次等15秒。三次仍然失败就直接跳过记录到failed_again.json。这种「本轮跳过、下一轮再试」的方式比单个标的上死磕更符合全市场批量的节奏。进度记录方面只要meta/failed.json始终保留未完成标的脚本重启后可以从断点继续不需要重新扫描全部文件。5. Baostock避坑手册六个最常被问到的异常现场5.1 跑着跑着突然全部报错日志提示尚未登录现象是脚本前几百个标的正常后面所有请求开始返回尚未登录或连接异常。原因是Baostock的登录会话有有效期长任务跑半小时以上可能自动掉线并非代码写错。解决的常见做法是分段处理比如每处理300个标的执行一次bs.logout()再bs.login()重新建立会话。for i, row in tasks.iterrows(): if i % 300 0: bs.logout() lg bs.login() print(relogin at, i, lg.error_code)5.2 数据拉回来发现日期不连续以为漏数据现象是某只股票CSV文件里日K日期跳过了好几天比如1月10号之后直接跳到1月15号。原因大概率是该股票在1月11日到14日处于停牌状态停牌期间Baostock不返回K线数据这不是接口故障。遇到这种情况不要急着补数据先通过query_all_stock的tradeStatus字段确认停牌状态或者对照交易日历文件检查缺失日期。5.3 同一天的前复权价格换了起始日期就变了这是最容易让人崩溃的坑。我用start_date2020-01-01拉的前复权数据和用start_date2022-01-01拉的数据在同一交易日的收盘价居然不同。原因是前复权价格是根据指定区间的除权除息事件回溯调整的起始日期不同复权基准就不同历史价格自然对不上。解决方案是如果要做跨年度增量更新存储时统一用后复权adjustflag1它的复权基准固定在上市首日价格序列不会因为增量区间改变而漂移。这是血泪经验回测数据一定要保证「同一标的价格序列自洽」。5.4 5分钟数据下载卡死进度半天不动现象是下载5分钟K线时脚本长时间无响应看起来像死机。原因是5分钟数据量远大于日线一只股票一年就有约11520条记录全市场几千万条数据在同一个请求里返回网络和内存都扛不住。解决方式是分时间片下载比如每次只拉一个季度拉完一批落盘再拉下一批同时把延时加大到random.uniform(0.5, 1.0)。5分钟数据建议独立脚本执行别跟日线放一起跑。# 分季度拉取5分钟数据避免单次请求数据量过大 periods [ (2024-01-01, 2024-03-31), (2024-04-01, 2024-06-30), (2024-07-01, 2024-09-30), (2024-10-01, 2024-12-31), ] for start, end in periods: rs bs.query_history_k_data_plus( code, date,time,code,open,high,low,close,volume,amount, start_datestart, end_dateend, frequency5m, adjustflag2, ) # ... 收集并落盘 time.sleep(1)5.5 CSV文件用Excel打开中文乱码现象是CSV文件用pandas读取正常但Excel打开后字段名和股票名称全是乱码。原因是Baostock返回的字段名是英文不受影响但如果你把code_name写入CSVpandas默认的UTF-8编码在Excel里不兼容。解决方式是写文件时统一指定encodingutf-8-sigBOM头会让Excel自动识别UTF-8编码。从第3章的代码开始我已经全部用了sig后缀直接照抄就不会踩。5.6 全市场下载太慢一个循环几小时跑不完现象是几千只股票全部串行下载预估耗时好几个小时感觉效率太低。原因是网络请求本身有往返延迟单标的虽然只要几十毫秒但串行累积起来就是天文数字。改进方向有两个一是用ThreadPoolExecutor做多线程并发但要注意Baostock对同一账户的并发连接数有限制建议线程数控制在5以内二是缩小全量范围只下载自己策略需要的板块或指数成分股不必每次都把全市场拉一遍。常见的做法是用4线程并发加0.1秒延时整体耗时能缩短到串行的三分之一左右。并发方案涉及会话管理逻辑比串行复杂新手阶段还是先跑串行跑顺了再优化。6. 数据校验走一遍用三行断言给本地K线库兜底6.1 拉完数据先过校验脚本别急着喂给策略数据下载完成不等于数据可用。我见过太多因为CSV里混入空行、价格为零、日期重复而导致回测结果严重失真的事故。我的习惯是每次下载或增量更新后强制跑一遍校验脚本把明显的数据异常拦截在策略之外。import pandas as pd import os DATA_DIR output/daily def verify_file(code): path os.path.join(DATA_DIR, code.replace(., _) .csv) if not os.path.exists(path): return False df pd.read_csv(path) # 断言1非空文件 assert len(df) 0, f{code} is empty # 断言2收盘价全部为正 assert (df[close].astype(float) 0).all(), f{code} has non-positive close # 断言3日期严格递增且无重复 assert df[date].is_monotonic_increasing, f{code} dates not sorted assert not df[date].duplicated().any(), f{code} duplicated dates return True # 跑最近下载过的前200只股票 tasks pd.read_csv(stock_list_20241231.csv).head(200) failed [] for row in tasks.iterrows(): code row[1][code] try: verify_file(code) except AssertionError as e: failed.append(str(e)) print(failed:, failed)三个断言分别解决三类典型问题空文件说明下载被跳过、非正价格说明字段里有脏数据、日期乱序说明拼接时排序出了问题。如果你发现is_monotonic_increasing报错优先检查增量更新的sort_values(date)是否执行——我一度把这个排序放在drop_duplicates之前导致去重后顺序被破坏翻车过一次。6.2 把下载和校验串成一条命令定时任务直接调用校验脚本写好后建议和下载脚本放进同一条命令链里先下载后校验任何一环失败都让任务报错退出。python download_daily.py python verify_kline.py校验通过后再生成一份meta/verify_report.json记录校验时间、文件数和失败项后续想在策略里判断数据新鲜度直接读这个报告就行不用每次重新扫描全部CSV。从那以后我每次拿到新下载的K线数据不管是从这套资源脚本跑出来的还是手动补的都会强制先走一遍校验流程再让策略去读。这个习惯帮我挡掉过好几回数值为0的坏数据混进回测的情况也让「数据有没有问题」从玄学变成了一个可以确认的步骤。希望帮到你。本文还有配套的精品资源点击获取