简介基于Baostock金融数据接口的K线数据自动化下载与本地存储工具面向金融分析师、量化研究者及普通投资者解决多市场指数与全A股历史行情数据批量获取的痛点。工具覆盖上证指数、深证成指、沪深300、创业板指等主要指数同时支持A股所有上市股票的日K线与5分钟K线数据自动拉取和本地存储大幅提升数据获取效率。资源共4个文件包含Python脚本、Markdown文档、TXT说明及DOCX附赠文档压缩包整体仅36KB轻量易用脚本实现自动下载流程文档提供使用说明与补充资料。目前已有146人学习下载尤其适合需要搭建本地行情数据库的初学者和中小团队。借助这套工具无需深厚编程背景即可获得连续、一致的日线与高频数据为后续分析、回测及研究提供可靠基础。1. 本地K线库我为什么选 Baostock 来做全市场自动下载做量化回测的第一步永远是准备数据而不是写策略。之前我在一个开源回测框架里跑指数增强想法需要沪深300成分股近五年的日K免费数据源要么按条数收费要么单次只能导出一小段手工导了一下午还漏了几只。后来换成 Baostock 接口写了几十行脚本把主要指数、全市场个股的日K和5分钟K线全部拉到了本地后续回测再也不受制于人。这个标题描述的就是这套自动化下载与本地存储工具的完整落地过程接口边界、目录设计、指数下载、全市场个股、5分钟线、常见坑最后给一版校验和增量更新方案。适合正在搭个人行情库、做技术分析或策略回测的开发者新手能跟着跑通熟手可以直接拿去改。2. 先摸清 Baostock 的接口边界频率、复权与字段确认2.1 Baostock 能提供什么频率、复权、字段与登录会话Baostock 是一个免费开源的 Python 行情接口包名就是 baostock。常见做法是先 import baostock as bs然后 bs.login() 建立会话之后用 query_history_k_data_plus 查询K线最后 bs.logout() 收尾。这个接口覆盖 A 股和主要指数的历史行情本身不限制本地存储方式所以很适合做“自动化下载 本地存储”这类工具。我建议先把接口边界搞清楚再写工具。它支持日K、周K、月K以及5分钟、15分钟、30分钟、60分钟分钟线。日线字段里有 date、code、open、high、low、close、volume、amount、preclose、turn、tradestatus 等分钟线则是 time、code、open、high、low、close、volume、amount。注意一个关键细节接口返回的所有值都是字符串空值可能是空字符串或 None需要自己在代码里统一处理这一点直接影响后续脚本的健壮性。复权参数 adjustflag 是另一个关键点1 表示后复权2 表示前复权3 表示不复权。指数我一般都用不复权因为指数点位本身就是市场原始表现个股做短周期回测用前复权做长周期研究用后复权避免未来价格影响历史值。这里没有绝对正确只有“符合你的回测假设”。还有一个容易被忽略的边界登录会话是有寿命的。长时间大批量查询后连接不会一直稳这是免费接口常见的限流手段。我在做全市场下载时策略是“分段登录、及时重连”这个放到后面的避坑章节展开。Baostock 还提供 query_all_stock 拿证券清单、query_trade_dates 拿交易日历这两组接口正好是本地K线库的左右手。这里把频率和字段先做一个速查表后面写脚本时直接对照频率时间字段常用字段典型用途ddatedate, code, open, high, low, close, volume, amount日线回测、技术分析w / mdate同上长周期趋势研究5 / 15 / 30 / 60timetime, code, open, high, low, close, volume, amount日内策略、分钟级信号2.2 本地存储选型parquet 与 CSV 在千万行场景下的差距为什么标题里会强调“本地存储”因为下载只是第一步怎么存直接影响回测时的读取效率。我的经验是避免“把所有数据塞进一个 CSV”尤其是5分钟线全市场几千万行一次性读进内存光解析就够你喝一壶。parquet 是列式存储读取时可以只加载需要的列文件体积也明显小于 CSV 原文。如果你只是验证几个指数CSV 完全够用一旦进入全市场日K甚至分钟线建议直接上 parquet。为了照顾数据调试可以单独保留一个 logs 目录放异常记录原始行情文件保持紧凑。parquet 依赖 pyarrow安装时顺手装上即可读取用 pd.read_parquet 就能接回 pandas。目录结构我习惯按“市场/标的”分层像这样data/ ├── daily/ │ ├── index/ │ │ └── sh.000001.parquet │ └── stock/ │ └── sz.000001.parquet ├── m5/ │ └── stock/ │ └── sz.000001.parquet └── logs/ └── download.log指数和股票分目录存放天然隔离了“指数代码与个股代码撞车”的问题。文件名直接带交易所前缀比如 sh.000001 和 sz.000001就算后面要做跨市场合并也不会混淆。parquet 和 CSV 的取舍用一张表说清楚维度CSVParquet文件体积大数字按文本存储列式压缩通常小一截读取速度全表扫描解析越大的文件越慢按列读取局部加载很快额外依赖无pyarrow可直接打开排查可以文本编辑器就能看需要专门工具或 pandas我的建议是本地调试阶段可以先落 CSV确认逻辑没问题后再转 parquet。正式的全量任务直接写 parquet省得后面还要换格式重跑一遍。2.3 用探测脚本确认接口返回的时间范围与字段顺序在正式全量下载前我习惯先跑一个“探测脚本”拿一只指数和一只股票各拉两条记录打印字段名和数据内容确认字段顺序、时间格式、空值表现。这个动作只要几分钟但能省掉后面的大把排查时间。import baostock as bs import pandas as pd lg bs.login() # 拉上证指数最近一段日K先看字段结构与数据形态 rs bs.query_history_k_data_plus( sh.000001, date,code,open,high,low,close,preclose,volume,amount, start_date2024-05-01, end_date2024-05-10, frequencyd, adjustflag3 ) print(error_code:, rs.error_code, error_msg:, rs.error_msg) print(fields:, rs.get_fields()) rows [] while rs.next(): rows.append(rs.get_row_data()) print(pd.DataFrame(rows, columnsrs.get_fields())) bs.logout()这段代码的作用是验证四件事查询接口本身是否可用、返回字段顺序和文档是否一致、行数据是不是纯字符串、空值长什么样。如果 rs.error_code 不是 0说明参数或网络有问题不要急着做全量下载。get_fields() 拿到的字段顺序正好可以作为后续生成 DataFrame 的列名避免手写列名顺序错位。参数说明start_date 和 end_date 的格式是 YYYY-MM-DDfrequencyd 表示日Kadjustflag3 表示不复权。code 必须带交易所前缀sh. 开头是上交所sz. 开头是深交所。经过这轮探测再写批量脚本会少很多返工也能确认接口当前实际能拉到的最早日期和你预期是否一致。提示探测脚本的结果只代表当前登录会话下的接口状态。免费数据源偶尔会有临时调整正式任务前重新跑一次探测是值得养成的习惯。3. 主要指数日K线批量下载指数代码对照与脚本落地3.1 指数代码对照表上证、深成、沪深300、创业板指都怎么传标题里点名的几个指数代码其实有固定规律上证指数是 sh.000001深证成指是 sz.399001沪深300 是 sh.000300创业板指是 sz.399006。刚接触的人很容易把上证指数错写成 sh.000001 之外的东西或者把创业板指当成 300 开头代码一错数据就全错。常见主要指数代码我整理成表可以直接抄指数名称代码备注上证指数sh.000001沪市综合指数深证成指sz.399001深市成分指数沪深300sh.000300沪深两市成分创业板指sz.399006创业板综合上证50sh.000016沪市大盘蓝筹中证500sh.000905中小盘成分代码前缀 sh./sz. 不能丢并且 sh.000001 是上证指数sz.000001 是深市一只个股代码两者完全不同。这个“撞车”问题在第6章会专门展开但在设计存储时先把目录分开、文件名保留前缀就能从根源上避免覆盖。另外提醒一点指数代码不是一成不变的新指数会发布老指数也有可能调整采样规则。我一般把指数清单放在脚本顶部的一个字典里新增指数时只改一处不要散落在多个函数里。3.2 指数日K线下载脚本登录、遍历与落盘指数数量少下载脚本可以写得比较直接。下面的脚本会登录 Baostock遍历指数代码字典逐只拉日K并写入 parquet最后统一退出登录。import baostock as bs import pandas as pd import os INDEX_CODES { sh.000001: 上证指数, sz.399001: 深证成指, sh.000300: 沪深300, sz.399006: 创业板指, sh.000016: 上证50, sh.000905: 中证500, } FIELDS_INDEX date,code,open,high,low,close,volume,amount def download_index_daily(start_date, end_date, out_dirdata/daily/index): lg bs.login() for code, name in INDEX_CODES.items(): rs bs.query_history_k_data_plus( code, FIELDS_INDEX, start_datestart_date, end_dateend_date, frequencyd, adjustflag3 ) rows [] while rs.error_code 0 and rs.next(): rows.append(rs.get_row_data()) df pd.DataFrame(rows, columnsFIELDS_INDEX.split(,)) for col in df.columns: if col not in (date, code): df[col] pd.to_numeric(df[col], errorscoerce) os.makedirs(out_dir, exist_okTrue) file_path os.path.join(out_dir, code .parquet) df.to_parquet(file_path, indexFalse) print(f{name} {len(df)} rows - {file_path}) bs.logout() download_index_daily(2015-01-01, 2024-12-31)逻辑说明先登录一次然后遍历指数字典逐只查询。这里的字段列表和前面探测脚本保持一致固定写死是因为已经验证过字段顺序行数据是字符串所以用 pd.to_numeric 把数值列批量转成数字转换失败的地方变成 NaN。每只指数下载完立即 to_parquet 写盘即使中途出错已经完成的指数文件也保留在磁盘上。参数说明frequencyd 表示日Kadjustflag3 表示不复权。out_dir 是输出目录文件名用 code .parquet这里的 code 自带 sh./sz. 前缀所以上证指数和深证成指不会重名。start_date 和 end_date 按你的研究区间来不需要一次拉太早数据范围不够时再向前扩区间重新拉一遍即可。3.3 adjustflag 复权参数与字段裁剪为什么指数选不复权写脚本时最容易纠结的是复权参数。我的原则很简单指数用不复权个股做回测用前复权做超长周期研究用后复权。复权参数速查如下参数值含义适用场景1后复权长周期收益研究2前复权策略回测、技术指标计算3不复权指数、行情原始形态指数本身没有除权除息缺口复权反而会引入不存在的价格跳变所以指数任务统一用 3。字段裁剪方面我只取 date, code, open, high, low, close, volume, amount 这八个字段。preclose、turn、tradestatus 这些字段对指数研究用处不大少存一个字段意味着 parquet 文件更小、读取更快。如果需要前收盘价完全可以在 DataFrame 里用 shift(1) 自己生成没必要把每个字段都堆进存储文件。还有一个小细节指数成交量字段返回的单位和个股不同做跨市场合并时建议统一口径。如果只是单指数分析这个字段不用太纠结。4. A股全市场个股日K线下载股票清单、断点续传与增量更新4.1 全市场股票清单怎么抓完整历史交易日并集去重A 股“所有上市股票”这个目标比听起来麻烦。如果你只在当前日期调用一次 query_all_stock拿到的只有现在还挂牌的股票那些已经退市的标的会从清单里消失但它们的历史K线同样是回测中可能要用到的数据。更稳的做法是取多个历史交易日的证券清单做并集。query_all_stock 的入参是一个日期返回该日期还在列表里的证券代码和名称。我一般选近五年的年度最后一个交易日分别拉取后把 code 并入同一个集合再排除指数代码得到“曾上市股票池”。代码大致如下import baostock as bs INDEX_BLACKLIST { sh.000001, sz.399001, sh.000300, sz.399006, sh.000016, sh.000905 } YEARS [2019-12-31, 2020-12-31, 2021-12-31, 2022-12-31, 2023-12-31] def fetch_stock_list(years, blacklist): code_set set() lg bs.login() for day in years: rs bs.query_all_stock(dayday) while rs.error_code 0 and rs.next(): row rs.get_row_data() code row[0] # 形如 sh.600000 / sz.000001 if code in blacklist: continue code_set.add(code) bs.logout() return sorted(code_set) stock_codes fetch_stock_list(YEARS, INDEX_BLACKLIST) print(stock count:, len(stock_codes))逻辑说明每个历史日期调用一次 query_all_stock返回的是那一天的证券列表其中包括当时还在交易、后来退市的股票。多日清单并集后去重能覆盖绝大部分曾上市标的。blacklist 里放的是指数代码避免把指数混入个股下载任务。参数说明years 列表可以按需调整间隔拉长能覆盖更多退市股票但也会多几次查询blacklist 建议保留主要指数代码。这里有个小提醒query_all_stock 的 tradeStatus 字段在不同年份可能取值不同我一般只取 code不依赖该字段做过滤。4.2 全市场日K下载脚本边下边写、进度文件与自动重连全市场股票有几千只日K下载是个长任务。第一次做的时候我用了一个最朴素的办法把所有结果先放进一个大列表跑完再统一写盘。结果跑到三分之一内存快爆了而且中途断一次网就得全部重来。后来改成“边下边写、进度记录、分段登录”三件套再没翻过车。import baostock as bs import pandas as pd import os, json, time FIELDS_STOCK date,code,open,high,low,close,volume,amount BATCH_SIZE 300 # 每下载300只重新登录一次 OUT_DIR data/daily/stock PROGRESS_FILE progress_daily.json def load_progress(path): if os.path.exists(path): with open(path, r, encodingutf-8) as f: return set(json.load(f)) return set() def save_progress(path, done): with open(path, w, encodingutf-8) as f: json.dump(sorted(done), f) def log_skip(code, reason): with open(skip.log, a, encodingutf-8) as f: f.write(f{code}\t{reason}\n) def download_stock_daily(code_list, start_date, end_date, adjustflag2): done load_progress(PROGRESS_FILE) pending [c for c in code_list if c not in done] print(pending:, len(pending)) lg bs.login() for i, code in enumerate(pending): try: rs bs.query_history_k_data_plus( code, FIELDS_STOCK, start_datestart_date, end_dateend_date, frequencyd, adjustflagadjustflag ) rows [] while rs.error_code 0 and rs.next(): rows.append(rs.get_row_data()) if len(rows) 0: log_skip(code, empty) done.add(code) save_progress(PROGRESS_FILE, done) continue df pd.DataFrame(rows, columnsFIELDS_STOCK.split(,)) for col in df.columns: if col not in (date, code): df[col] pd.to_numeric(df[col], errorscoerce) os.makedirs(OUT_DIR, exist_okTrue) df.to_parquet(os.path.join(OUT_DIR, code .parquet), indexFalse) done.add(code) if (i 1) % BATCH_SIZE 0: save_progress(PROGRESS_FILE, done) bs.logout() time.sleep(1) lg bs.login() except Exception as e: log_skip(code, ferror: {e}) time.sleep(0.5) bs.logout() save_progress(PROGRESS_FILE, done)逻辑说明每只股票下载完立即写盘即使进程崩溃也只丢当前一只progress_daily.json 记录已完成代码重启时直接跳过已完成部分这就是断点续传。BATCH_SIZE 等于 300 时每下载 300 只主动 logout 再重新 login避免长连接被服务端断开同时配合 sleep(1) 降低请求频率。参数说明adjustflag2 表示前复权适合大部分回测场景如果你要复刻原始行情改成 3。start_date 和 end_date 控制历史范围。空数据写入 skip.log也许是停牌期也许是代码本身有问题后面校验时统一核对。进度文件建议放在脚本目录而不是数据目录避免和数据文件混在一起。提示首次全量下载时start_date 可以设为你研究所需的最早日期不用从 1990 年拉。数据范围不够再向前扩区间比一次拉过量数据稳得多。4.3 全量与增量分离首次补历史、每日只拉尾盘全量下载完成后并不代表任务结束。每天收盘后需要做增量更新把当天的新K线补充进本地库。增量更新的思路和全量不同先用最近一个交易日调用 query_all_stock 拿到当前上市清单然后对每只股票读取本地 parquet 最后一行的日期把 last_date 1 作为 start_date去查询新的数据。import pandas as pd import os def incremental_update(latest_day): codes fetch_stock_list([latest_day], INDEX_BLACKLIST) for code in codes: path os.path.join(OUT_DIR, code .parquet) if not os.path.exists(path): start 2015-01-01 # 本地没有文件走全量逻辑 else: df pd.read_parquet(path, columns[date]) last_date pd.to_datetime(df[date].iloc[-1]) start (last_date pd.Timedelta(days1)).strftime(%Y-%m-%d) # 这里继续调用 download_stock_daily 的单只逻辑 update_one_stock(code, start, latest_day)逻辑说明增量更新的核心是“接口幂等”。重复拉同一区间的数据不会产生副作用最多是覆盖写一遍所以不需要复杂的锁机制。对每只股票先定位本地最后一条数据的日期再往后拉本地已经有文件的部分不会被重复追加。参数说明latest_day 取最近一个交易日格式 YYYY-MM-DD。增量更新通常几分钟就能跑完比每天全量重拉省时得多。需要注意新上市股票没有本地文件会走全量分支从预设的最早日开始拉。5. 5分钟K线下载的差异化处理字段、时间范围探测与落盘5.1 接口差异frequency5 与日K的返回字段5分钟K线和日K在 Baostock 接口上的差异比很多人想象的大。第一是频率参数不同日K是 frequencyd5分钟线是 frequency5第二是时间字段不同日K叫 date分钟线叫 time值长这样 2024-01-04 09:30:00第三是字段范围不同日K的 preclose、turn、tradestatus 等字段在分钟线接口上不一定都会返回。我一般把5分钟K线的字段固定为time, code, open, high, low, close, volume, amount。写脚本时字段列表不要直接用日K那套否则 rs.next() 取出来的行数据列数对不上DataFrame 创建就会报错。另外要意识到一个问题5分钟数据不是所有标的都有完整历史。老股票可能覆盖时间长一些次新股只有上市后的分钟数据有的冷门标的甚至一大段时间查不到。这种差异不是 bug而是数据源的客观边界需要先探测再用。5.2 先探测再全量确认5分钟数据最早可用范围千万不要直接对全市场跑5分钟K线的全量下载因为有些标的的数据范围和你预期差得很远跑完才发现大量空文件白白浪费时间。我的习惯是先写一个探测函数对几只代表性标的确认最早可用时间再决定全量任务的 start_date。import baostock as bs import pandas as pd from datetime import datetime, timedelta M5_FIELDS time,code,open,high,low,close,volume,amount def probe_m5_earliest(code, start_candidate2015-01-01): end datetime.now().strftime(%Y-%m-%d) current start_candidate result None for _ in range(12): rs bs.query_history_k_data_plus( code, M5_FIELDS, start_datecurrent, end_dateend, frequency5, adjustflag3 ) times [] while rs.error_code 0 and rs.next(): times.append(rs.get_row_data()[0]) if times: result (current, min(times)) break current (pd.to_datetime(current) - pd.DateOffset(years1)).strftime(%Y-%m-%d) return result lg bs.login() print(probe_m5_earliest(sh.000001)) print(probe_m5_earliest(sz.000001)) bs.logout()逻辑说明从 2015 年往前试探如果查询返回空就把 start_date 再向前推一年直到能找到数据为止。返回结果是一个元组第二项是该区间内最早的一条 time这样你能直观看到这个标的5分钟数据的起点。注意 start_candidate 不要设到 1990 年免费数据源的分钟线通常不会覆盖那么早。参数说明frequency5 是5分钟线adjustflag 这里用 3 做探测因为探测只需要看是否存在数据不需要关心复权。循环上限 12 次也就是覆盖 12 年的前移足够实际使用。对指数和几个活跃个股分别探测得到的结果就能指导全量任务的 start_date。5.3 时间戳标准化与去重分钟线与日线合并的基础5分钟数据落盘后第一件事是清洗。常见问题有两个同一个 time 出现重复行以及 time 格式在不同标的不完全一致。清洗逻辑不能省不然后续合并日线和分钟线时会有一堆坑。def normalize_time(value): if value is None or str(value).strip() : return pd.NaT value str(value).strip() if in value: return pd.to_datetime(value) return pd.to_datetime(value 15:00:00) def clean_m5(df): df[time] df[time].map(normalize_time) df df.dropna(subset[time]) df df.drop_duplicates(subset[time], keeplast) df df.sort_values(time).reset_index(dropTrue) return df逻辑说明normalize_time 把两种时间格式统一成 datetime。日K的 date 是 2024-01-04没有时分秒也可以复用这个函数给它补一个收盘时间 15:00:00这样两个 DataFrame 的 time 字段类型完全一致concat 或 merge 时不会因为格式不同而失败。drop_duplicates 专门处理分钟线重复行keeplast 表示保留最后一条。参数说明subset 限定按 time 列去重避免其他列的值干扰判断。sort_values 是为了让文件里的数据按时间升序排列方便后续增量更新直接读最后一行。这一步做完5分钟K线文件才算干净。6. 避坑Baostock 下载K线常见的5类问题与排查6.1 指数与个股代码撞车000001 到底是谁现象拉全市场清单时发现 sh.000001 和 sz.000001 同时存在如果不加区分直接写文件名某些逻辑会把上证指数的数据覆盖到一只深市个股的文件上回测结果完全错乱。原因指数代码和个股代码共用数字段sh.000001 是上证指数sz.000001 是深市的一只个股代码。只看数字后缀根本无法区分必须带上交易所前缀。解决存储文件名统一用完整 code带 sh./sz. 前缀目录上把 index 和 stock 分开。清洗代码时用前缀判断所属市场sh.6、sh.68、sz.0、sz.3 开头的是个股指数代码单独维护一个黑名单集合。这个动作是地基地基歪了后面全是歪的。6.2 全量跑到一半连接假死分段登录与进度文件现象全量下载跑了几百只股票后rs.error_code 开始报错或者查询直接卡住不动日志里出现连续的错误记录。重启脚本重新跑之前下过的又得再来一遍。原因免费接口对长连接和请求频率有限制长时间高并发查询会让连接进入不稳定状态。我之前一度以为是玄学后来把间隔和重连加上后问题基本消失。解决第一每下载 300 只左右主动 logout 再重新 login中间 sleep(1) 降频第二进度文件必须在每只股票写盘后立即更新脚本重启时加载进度跳过已完成部分第三异常不要裸抛统一用 try/except 捕获并写 skip.log。血泪经验进度文件不能省它比任何重试机制都管用。6.3 前复权价格会漂移同一个历史K线今天和上周拉的不一样现象某只股票用前复权拉历史日K上周保存的文件和今天重新拉的同一天数据收盘价对不上回测结果也跟着变。原因前复权以当前价格作为基准一旦股票发生除权除息所有历史价格都会重新调整。所以你拉取时间不同同一段历史的前复权价格就可能不同。这不是接口数据错误而是复权算法本身的特性。解决短周期回测用前复权问题不大长周期研究建议用后复权后复权价格只随行情变化不会因为新除权事件而整体平移。如果一定要用前复权又要稳定复现把“数据拉取日期”作为元数据记下并在换数据源时重新对齐。本地库的更新任务里前复权全量重拉的成本比想象高量力而行。6.4 5分钟的 time 字段与日线的 date 字段拼不到一起现象把日K和5分钟K线 concat 成一个大 DataFrame时间列对不上排序也乱某些行 date 是字符串某些行 time 是 Timestamp。原因日K的时间字段叫 date值是 2024-01-045分钟线的时间字段叫 time值是 2024-01-04 09:30:00。字段名不同、类型不同、精度不同直接拼接自然出问题。解决在清洗阶段统一调用 normalize_time把所有时间列转成 datetime指定列名统一叫 dt 或 time。日K补一个收盘时间分钟线保留自己的时分秒这样两条线可以按同一把时间尺对齐。这个坑在“多频率混合回测”时几乎是必踩的先处理再入库。6.5 停牌期返回空 DataFrame是常态不是异常现象某只股票本地日K文件中间缺了一大段看起来像下载失败但查 skip.log 却找不到 error 记录。原因停牌期间没有K线成交query_history_k_data_plus 在这个区间返回空是正常表现。只要股票没有退市复牌后K线就会继续出现中间的空白不需要填充。解决空返回先写入 skip.log理由标为 empty不要当错误重试。完整性校验时缺失日期要和停牌公告交叉核对不能直接判定文件损坏。反过来如果全市场几千只股票在同一段日期集体缺失那才需要怀疑是接口或参数问题而不是个股停牌。7. 进阶本地K线库的完整性校验与回测接入技巧K线库建好之后校验是值得认真做的一步。用 Baostock 的 query_trade_dates 拉一段交易日历然后逐个比对本地文件的日期集合缺失比例异常的就列出来。这个动作能在你真刀真枪跑策略之前把数据层的坑提前排掉。def check_missing_days(codes, start_date, end_date): rs bs.query_trade_dates(start_datestart_date, end_dateend_date) trading_days [] while rs.error_code 0 and rs.next(): row rs.get_row_data() if row[1] 1: # 1 表示交易日 trading_days.append(row[0]) for code in codes: path os.path.join(data/daily/stock, code .parquet) if not os.path.exists(path): print(code, MISSING FILE) continue df pd.read_parquet(path, columns[date]) got set(pd.to_datetime(df[date]).astype(str)) missing sorted(set(trading_days) - got) if len(missing) 5: print(code, missing, len(missing), missing[:10])逻辑说明交易日历是校验的“标尺”每只股票应有的交易日数量一目了然。停牌会造成少量缺失属于正常所以阈值设为大于 5 天才提示如果一只股票常年有几十天缺失多半是下载任务中跳过了要回到全量脚本重拉。校验脚本不要每天跑每周跑一次就够。压缩存储也值得做parquet 默认压缩已经不错写入时指定 compressionzstd 能进一步压体积对5分钟这种千万行级别的文件收益明显。读取时只取需要的列避免全表加载。最后是把本地文件读成标准 OHLCV 给回测框架用def load_ohlcv(code, freqd, startNone, endNone): root data/daily if freq d else data/m5 path os.path.join(root, stock, code .parquet) df pd.read_parquet(path) time_col date if date in df.columns else time df[time_col] pd.to_datetime(df[time_col]) df df.sort_values(time_col) if start: df df[df[time_col] start] if end: df df[df[time_col] end] return df.set_index(time_col)这个函数把所有存储细节封装掉了。回测策略只需要传一个 code 和频率拿到的是排序好的 OHLCV 数据本地库和策略层就彻底解耦。后续换数据源时只要改 load_ohlcv 内部实现策略代码一行都不用动。我自己第一次做全市场下载时就是吃了没写进度文件的亏几小时的数据因为一次断网全部白跑后来把“边下边写、进度记录、定时校验”当成默认习惯再没出过类似事故。K线库这件事一次性投入做扎实后面每次回测都能省回十倍时间。希望帮到你。本文还有配套的精品资源点击获取