简介：面向金融数据分析与量化研究场景，这份工具包基于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, columns=rs.fields)一行就能转成结构化表格，清洗成本极低。
2.2 频率、复权与字段：入库之前先想清楚这三件事
下载K线之前，有三个参数直接决定数据能不能用于回测，分别是frequency、adjustflag和字段列表。
frequency取值"d"是日线，"5m"是5分钟线。日线数据历史范围长，适合策略回测；5分钟数据量是日线的几十倍，适合做日内策略，但下载时间长、存储占用大，两者最好分开跑脚本。adjustflag是复权选项——"1"后复权、"2"前复权、"3"不复权。做技术指标计算，比如均线、MACD、布林带，一定要用复权数据，否则除权除息日会出现跳空假信号。做因子分析或分红研究才考虑用不复权。
字段选择上，推荐一个通用组合：
| 字段 | 含义 | 备注 |
|---|---|---|
| date | 交易日 | 日线格式2024-01-15 |
| time | 交易时间 | 仅分钟线有值，格式14:35:00 |
| code | 证券代码 | 带交易所前缀，如sh.600000 |
| open/high/low/close | 开高低收 | 字符串，需自行转float |
| volume | 成交量 | 股数，留意单位 |
| 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_date="2020-01-01", end_date="2024-12-31", frequency="d", adjustflag="2", # 前复权 ) # 3. 遍历游标，收集数据 rows = [] while (rs.error_code == "0") and rs.next(): rows.append(rs.get_row_data()) # 4. 转DataFrame并落盘 df = pd.DataFrame(rows, columns=rs.fields) print(df.head()) df.to_csv("sh000001_d.csv", index=False) # 5. 退出登录，释放连接 bs.logout()这段代码的逻辑分五步：登录获取会话、调用查询接口、用while + next()游标方式逐行读取返回结果、将字符串行列表组装成DataFrame、最后存成CSV并退出登录。rs.error_code是判断查询是否成功的核心标识，只有等于"0"才说明数据返回正常；rs.fields保存的是当前查询的字段名列表，用它做列名永远不会跟字段参数写错顺序。
参数上需要注意：start_date和end_date格式必须严格是YYYY-MM-DD，pandas解析时少零补零都容易出错；adjustflag="2"用的是前复权，如果拿不准该用哪种，在第5章我会给出选型建议。
3.2 拉取全市场股票列表：拿到基础标的清单
单标的验证通过后，就要解决「下载哪些股票」的问题。手工维护一份代码列表很不现实，推荐每天或每月用query_all_stock拉一次当时在市交易的标的清单，既包含全部A股代码，也能顺带看到交易状态。
import baostock as bs import pandas as pd bs.login() # 指定一个交易日，获取当天所有在市交易的证券 rs = bs.query_all_stock(day="2024-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", index=False, encoding="utf-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_ok=True) FIELDS = "date,code,open,high,low,close,preclose,volume,amount,pctChg,turn" def download_one(code, start_date, end_date, adjustflag="2"): """下载单只股票日K，成功返回行数，失败返回0""" rs = bs.query_history_k_data_plus( code, FIELDS, start_date=start_date, end_date=end_date, frequency="d", adjustflag=adjustflag, ) 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, columns=rs.fields) # 统一用代码命名文件，方便后续快速读取 safe_code = code.replace(".", "_") df.to_csv(os.path.join(DATA_DIR, f"{safe_code}.csv"), index=False, encoding="utf-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(f"progress: {i+1}/{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_date=start, end_date=today, frequency="d", adjustflag="2", ) 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, columns=rs.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_index=True) merged = merged.drop_duplicates(subset=["date"], keep="last") merged = merged.sort_values("date").reset_index(drop=True) merged.to_csv(path, index=False, encoding="utf-8-sig") else: new_df.to_csv(path, index=False, encoding="utf-8-sig") return True这段代码的逻辑核心是「用字符串日期做比较」。YYYY-MM-DD格式下，字符串比较和日期比较结果一致，所以不需要转成datetime对象。drop_duplicates(subset=["date"], keep="last")保证了即使起始日期和本地已有数据重叠，最终入库数据也不会有重复K线。
增量更新的边界条件有两个：第一，如果本地最新日期恰好等于今天，说明数据已经是最新，直接返回。第二，如果文件不存在，则从2015-01-01拉全量历史。这里的起始日期按你自己的需求改，想拉更长历史就改成2000-01-01。还要注意，增量更新最好在收盘后执行，盘中拉到的当日K线只有未收盘的临时数据，混入历史库会污染回测结果。
4.3 失败重试与日志：把断点续传做成标配能力
跑全市场增量时，网络抖动、接口超时几乎必然发生。我的经验是不要指望一次跑完，而是把「失败重试」和「进度记录」做进脚本里。
import time import json def run_with_retry(code, max_retry=3, wait=5): for attempt in range(1, max_retry + 1): try: incremental_download(code) return True except Exception as e: print(f"attempt {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_date="2020-01-01"拉的前复权数据，和用start_date="2022-01-01"拉的数据，在同一交易日的收盘价居然不同。原因是前复权价格是根据指定区间的除权除息事件回溯调整的，起始日期不同，复权基准就不同，历史价格自然对不上。解决方案是：如果要做跨年度增量更新，存储时统一用后复权adjustflag="1"，它的复权基准固定在上市首日，价格序列不会因为增量区间改变而漂移。这是血泪经验，回测数据一定要保证「同一标的价格序列自洽」。
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_date=start, end_date=end, frequency="5m", adjustflag="2", ) # ... 收集并落盘 time.sleep(1)5.5 CSV文件用Excel打开中文乱码
现象是CSV文件用pandas读取正常，但Excel打开后字段名和股票名称全是乱码。原因是Baostock返回的字段名是英文不受影响，但如果你把code_name写入CSV，pandas默认的UTF-8编码在Excel里不兼容。解决方式是写文件时统一指定encoding="utf-8-sig"，BOM头会让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的坏数据混进回测的情况，也让「数据有没有问题」从玄学变成了一个可以确认的步骤。希望帮到你。
本文还有配套的精品资源，点击获取
