简介沪深股票历史日线数据资源覆盖沪深两市全部股票自上市日至2022年1月10日的每日行情。数据字段除开盘价、收盘价、最高价、最低价、振幅、成交量、成交额、换手率外还预计算了MACD、CCI以及同花顺手机版多空指标可直接用于判断价格趋势、识别超买超卖区域和辅助买卖决策对技术分析爱好者、金融研究员及量化交易者都有实用价值。整包仅1个SQL文件共1个文件压缩后约424.52MB利用SQL可灵活完成区间统计、指标筛选与回测数据提取免去自行爬取清洗的繁琐步骤。已有605人学习下载数据结构清晰便于回溯不同市场周期下的个股表现也可用于构建多因子模型或验证经典技术指标的有效性适合需要全量历史A股数据做策略复盘或模型训练的开发者直接使用。1. 沪深股票全量日线回补从接口选型到数据校验的一次性到位拿到“沪深股票历史以来到2022-01-10的全部日线数据”这个需求第一反应不是写代码而是先想清楚一件事你要的是“能跑通的脚本”还是“能用于因子回测的可靠数据集”。这两者的差距非常大。2022年1月10日这个截止点意味着你需要从交易所最早的数据上证1990年12月19日、深证1991年4月3日一直拉到指定日期跨越30多年、5000多只股票、每只股票几千条K线总量在千万级记录。这个量级如果用公共接口逐日请求按每秒3次的频率限制纯串行下载需要跑十几个小时而且大概率中途断掉。更麻烦的是不同数据源对“历史以来”的定义不同有的只给最近三年有的复权因子缺失有的停牌日直接留空洞。把这些问题拆开看核心就三个选对数据源、设计断点续传的回补流程、验证数据完整性和复权一致性。这篇文章把这套方案完整讲透。2. 数据源选型与全量历史数据的获取机制2.1 为什么免费接口里baostock最适合全量回补市面上的A股日线数据接口看似很多但能做全量历史回补的窗口其实很窄。你搜“沪深股票历史数据接口”最先跳出来的是tushare、akshare、baostock三个。tushare的pro接口需要积分门槛基础积分只能取最近几年的数据历史全量需要较高积分或者付费对于一次性回补需求不划算。akshare的底层多为爬虫实现东方财富、新浪财经这些源经常改版爬虫稳定性没法保证而且历史数据最早只到2000年左右再往前就断了。baostock是专门为量化研究设计的免费接口不需要token不限积分数据从交易所成立日就开始覆盖而且是批量返回——一次查询直接给一整只股票的全历史日线不用自己按日期翻页。这就是为什么做全量回补时baostock是首选。2.2 baostock的登录校验与数据工厂模式baostock的使用方式比较特殊它不是函数式调用而是先登录拿到会话再通过query_history_k_data_plus这个工厂方法取数据。登录动作会做本机校验和网络握手返回的结果码0表示成功10001001表示网络错误10001002表示密码错误。虽然它是免费接口但登录态是有时效的长时间运行需要定期检查连接状态或者干脆每次批量任务重新登录。import baostock as bs import pandas as pd from datetime import datetime # 登录获取会话 lg bs.login() if lg.error_code ! 0: print(f登录失败: {lg.error_msg}) raise SystemExit(1) # 查询平安银行全历史日线 rs bs.query_history_k_data_plus( sz.000001, date,code,open,high,low,close,preclose,volume,amount,adjustflag,turn,tradestatus, start_date1991-04-03, end_date2022-01-10, frequencyd, adjustflag3 )这段代码里的adjustflag参数决定了复权行为3表示不复权1表示后复权2表示前复权。全量回补时这里建议取3不复权因为后面做因子计算时需要自己控制前复权/后复权逻辑依赖接口的复权结果容易踩坑。tradestatus字段是关键1表示正常交易0表示停牌回补时不能只按日期切分还得按交易状态过滤。2.3 交易日历的坑为什么不能按自然日顺序拉取直接按股票代码遍历拉取是最直觉的做法但这里有一个隐藏的效率问题。baostock的query_history_k_data_plus是按股票维度返回所有历史K线不是按日维度。也就是说每调用一次就拿回一整只股票的全量数据循环5000次就能覆盖全市场。这个设计非常适合全量回补但前提是你得先拿到完整的股票列表。常见的做法是用query_stock_basic拉取全部证券代码及状态然后过滤出股票类型再逐只回补。# 获取全部证券代码 rs bs.query_stock_basic() stock_list [] while (rs.error_code 0) rs.next(): row rs.get_row_data() stock_list.append({ code: row[0], # 证券代码 code_name: row[1], # 证券名称 type: row[3], # 1:股票 2:指数 3:其它 status: row[4] # 1:上市 0:退市 }) df_stocks pd.DataFrame(stock_list) # 只保留股票同时包含已退市的保证历史数据完整 df_stocks df_stocks[df_stocks[type] 1] print(f股票总数: {len(df_stocks)})这里一定要注意status字段不要过滤掉退市股。沪深股票的历史全量数据里退市股是必须包含的否则做幸存者偏差研究时数据就是残缺的。baostock对退市股的历史数据仍然正常返回只是最新的交易状态是退市。在拉取时你会发现有些退市股的上市日期非常早比如最早的一批“老八股”它们的日线数据从1990年底就开始了。2.4 增量字段的设计一次回补长期受益全量回补不应该是用完一次就扔的脚本。2022年1月10日之后的每个交易日你可能还需要增量追加数据。所以存储格式在一开始就要设计好。我建议用Parquet按年分区存储而不是直接存CSV原因有三个一是千万级行数的CSV加载要几十秒Parquet只要几秒二是Parquet自带列式压缩磁盘占用只有CSV的三分之一左右三是后续做增量更新时可以直接用pyarrow的dataset API做分区写入效率高很多。字段设计上至少要保留date、code、open、high、low、close、volume、amount、adjustflag、turn、tradestatus这11个字段其中adjustflag存当时取数用的复权标记这是事后校验数据一致性的重要依据。import pyarrow as pa import pyarrow.parquet as pq from pathlib import Path # 按年分区的写入逻辑 def save_daily_data(df: pd.DataFrame, base_dir: str ./stock_data): df[year] pd.to_datetime(df[date]).dt.year for year, group in df.groupby(year): out_path Path(base_dir) / fyear{year} out_path.mkdir(parentsTrue, exist_okTrue) # 单只股票追加写入使用行组追加模式 pq.write_to_dataset( group, root_pathbase_dir, partition_cols[year], existing_data_behavioroverwrite_or_ignore )overwrite_or_ignore这个参数很关键它允许在已有的分区上追加新数据而不覆盖旧文件。不过要特别注意这个参数是按文件粒度去重的如果你的增量数据跨多个交易日后又回头去补前几天的数据就会产生重复行。所以更稳妥的做法是把增量数据的日期范围严格控制在上一次回补的截止日之后。3. 全量回补的完整流程断点续传、频率控制与异常重试3.1 基于已回补记录的状态表全量回补几十万次接口调用中途必然出问题网络断、接口限流、进程被杀。不设计断点续传机制的回补脚本就是一次性脚本重跑时要全部重新拉。正确的做法是先建一张回补状态表记录每只股票的代码、最后成功回补的日期、失败次数和错误信息每次启动时先检查这张表跳过已经完成的只重试失败的。这张状态表本身也可以和K线数据放在同一个Parquet目录下用一个小型CSV文件维护就行。# 初始化回补状态目录 mkdir -p ./stock_data/checkpoint touch ./stock_data/checkpoint/stock_progress.csv状态表的最小字段是code、last_date、status、retry_count、last_error。last_date用于增量补数据时判断从哪天开始拉status标记done、failed、partial三种状态retry_count控制最大重试次数last_error记录最近一次失败原因方便排查。3.2 主循环逐股回补与自适应限速主循环的逻辑比想象中简单但几个细节决定成败。首先是查询接口的start_date和end_date全量回补时start_date直接填1990-12-19上证最早交易日end_date填2022-01-10不逐日请求一次拉全。其次是频率控制baostock对并发没有严格限制但单进程连续高频请求会触发服务端的限流表现为返回码为10002003请求过于频繁。最稳的做法是加入自适应限速正常请求间隔0.1秒如果遇到限流指数退避到1秒、2秒、4秒直到成功。import time import random def fetch_stock_history(bs, code: str, start: str, end: str, max_retry: int 5): retry_count 0 wait_time 0.1 # 正常间隔 while retry_count max_retry: rs bs.query_history_k_data_plus( code, date,code,open,high,low,close,preclose,volume,amount,adjustflag,turn,tradestatus, start_datestart, end_dateend, frequencyd, adjustflag3 ) if rs.error_code 0: # 成功读取数据 rows [] while rs.next(): rows.append(rs.get_row_data()) df pd.DataFrame(rows, columnsrs.fields) # 只保留正常交易日 df df[df[tradestatus] 1] return df elif rs.error_code 10002003: # 限流指数退避 retry_count 1 time.sleep(wait_time * (2 ** retry_count) random.uniform(0, 0.5)) else: # 其他网络错误等待后重试 retry_count 1 time.sleep(2 retry_count) return None这段代码的处理逻辑是成功时一次性把所有行取干净rs.next()循环要走到返回False为止否则会导致后续查询结果串数据。tradestatus 1的过滤是为了剔除停牌日停牌日接口返回的空值行如果不处理后续计算收益时会当成0处理造成严重误差。这里有个容易被忽略的细节baostock的K线数据里也有tradestatus为0的行这些行不是没有数据而是当天停牌所以交易字段为空如果你直接用dropna()清掉就会误删整个日期行导致时间轴断裂——虽然很低级但很多人第一次做都会踩。3.3 全市场并行与进程级容错单线程逐只回补到2022年1月10日对5000多只股票来说假设单只股票返回平均120毫秒大概需要10分钟。这个速度其实已经可以接受不需要上多线程。但如果想跑得更快可以用concurrent.futures做线程池但要注意baostock的session不是线程安全的每个线程必须创建独立的连接。常见做法是用threading.local存线程专属的session或者干脆用多进程每个进程独立登录。如果你选择多进程我建议每500只股票一个子进程进程内用ThreadPoolExecutor开4个线程各线程独立baostock登录。整体速度可以压到2分钟左右。但代价是代码复杂度上升调试难度增大。我先给一个稳妥的单线程方案因为对多数一次性回补来说10分钟的耗时是可以接受的。import pandas as pd from pathlib import Path def backfill_all_stocks(df_stocks, checkpoint_file: str, end_date: str 2022-01-10): # 读取已有进度 if Path(checkpoint_file).exists(): df_ckpt pd.read_csv(checkpoint_file, dtype{code: str}) else: df_ckpt pd.DataFrame(columns[code, last_date, status, retry_count, last_error]) for _, row in df_stocks.iterrows(): code row[code] # 如果已经回补完成跳过 existing df_ckpt[df_ckpt[code] code] if not existing.empty and existing.iloc[0][status] done: continue df fetch_stock_history(bs, code, 1990-12-19, end_date) if df is not None and len(df) 0: save_daily_data(df) # 更新进度 update_checkpoint(df_ckpt, code, df[date].max(), done, 0, ) else: update_checkpoint(df_ckpt, code, , failed, 1, query return empty)这里每次循环拿到的df是整个股票的历史需要及时写入磁盘释放内存。你可能会问一只股票30年的日线也就8000行左右内存压力不大但5000只全量回补如果一次性攒在内存里不写盘最后阶段会占用几百MB甚至上GB。渐进写盘还有个好处是中断后已写入的数据立即可用不必等全部跑完。3.4 增量更新的时间边界全量回补完成之后2022-01-10之后的新交易日怎么追加这是另一个常见需求场景数据需要每天更新。增量更新的起点是每只股票在状态表里的last_date 1个交易日终点是今天。但注意baostock的接口只接受YYYY-MM-DD字符串不接受datetime对象而且start_date必须是交易日如果填了非交易日接口不会报错而是返回空结果。所以增量更新的标准做法是先用query_trade_dates拿到最近的交易日历再以last_date的下一个交易日作为起点。def get_trade_dates_after(bs, last_date: str, end_date: str): rs bs.query_trade_dates(start_datelast_date, end_dateend_date) dates [] while (rs.error_code 0) rs.next(): row rs.get_row_data() # 第3个字段是日历日期第4个字段是是否交易日 dates.append(row[1]) return dates # 增量更新示例找到last_date之后的下一个交易日 from datetime import datetime, timedelta last_date 2022-01-07 next_date datetime.strptime(last_date, %Y-%m-%d) timedelta(days1) # 用query_trade_dates确认next_date是否为交易日如果不是顺延增量更新的频率控制可以更激进因为每天只需要拉几千只股票当天的数据接口压力小。但你如果用的是同一套查询函数建议保留原来的限速策略保证在极端情况下比如一次性回补7天没跑也能稳定完成。4. 复权处理的三种选择与数据质量校验方法4.1 前复权、后复权、不复权的适用场景标题里的“全部日线数据”没有指定复权方式但实际应用中你必须在取数时就决策因为这直接影响后续策略表现。不复权数据是用来算真实成交额和成交量的任何涉及到金额绝对值的统计比如流动性分析都只能用不复权数据。前复权数据适合技术分析因为它保留了最近的价格绝对值K线形态和均线系统看起来直观但前复权数据的历史价格是动态变化的每次新的除权除息事件发生历史价格都会整体变动所以在回测中反复加载前复权数据会导致价格不一致。后复权数据不存在这个问题它把所有历史价格按复权因子统一调整序列连续且唯一适合做长期收益计算和因子回测。4.2 用不复权数据手动计算复权因子的完整代码因为baostock的adjustflag3是不复权数据需要自己实现复权因子的计算。常见做法是用preclose昨收和close今收的关系来推算除权因子。这个逻辑很简单正常情况下close / preclose接近当日收益率但如果发生除权除息这个比值会偏离正常波动范围偏离的比例就是复权因子。注意这个计算只对除权除息日之后的序列有效前复权需要从最新日期往回推导后复权需要从最早日期往未来推导。import numpy as np def calc_adjust_factor(df: pd.DataFrame) - pd.DataFrame: df df.sort_values(date).reset_index(dropTrue) # 理论收益率当日收盘 / 昨收 df[ret_raw] df[close] / df[preclose] # 实际区间最大涨跌幅过滤涨停/跌停之外的视为除权 mask (df[ret_raw] 1.11) | (df[ret_raw] 0.89) df[adjust_factor_raw] np.where(mask, df[ret_raw], 1.0) # 从最早日期累计后复权因子 df[adjust_factor_post] df[adjust_factor_raw].cumprod() # 前复权因子 后复权因子 / 最新后复权因子 latest_factor df[adjust_factor_post].iloc[-1] df[adjust_factor_pre] df[adjust_factor_post] / latest_factor # 生成复权价格 df[close_adj_pre] df[close] * df[adjust_factor_pre] df[close_adj_post] df[close] * df[adjust_factor_post] return df这段代码的处理精度对大多数场景够用但有个明显的坑正常涨停/跌停日的close/preclose也可能超过1.1或低于0.9特别是创业板注册制之后20%的涨跌幅限制会把正常的涨跌误判为除权。所以这个逻辑需要配合turn换手率或者成交量变化来二次过滤。一个简单的修正办法是如果当日成交量没有异常放大就不视为除权日。4.3 数据完整性校验用同源和异源数据交叉验证全量回补完成后数据质量校验是不可省的一步。校验分两个维度一是完整性检查每只股票的日线数量和时间连续性二是一致性用另一数据源抽取部分股票做收盘价对比容差设为0.5%四舍五入到小数点后两位导致的误差范围。# 检查每只股票是否存在跳空缺口停牌日除外 def validate_continuous(df: pd.DataFrame, trade_dates: set) - list: missing_dates [] for date in df[date]: if date not in trade_dates: missing_dates.append(date) return missing_dates # 抽样与akshare对比 # import akshare as ak # df_ak ak.stock_zh_a_hist(symbol000001, perioddaily, # start_date20200101, end_date20220110, adjust) # 对比close列允许0.5%误差对比时注意两个细节一是抽样不能只看头部股票要按代码分布随机抽50只覆盖不同上市年份和不同行业二是截止日期2022-01-10当天的数据在收盘前是无法取到的所以校验样本范围要往前挪一天从2022-01-07及之前取。如果发现某些股票在baostock和akshare中的收盘价不一致优先检查该股票是否发生过送转或配股因为两个数据源对复权因子的处理可能不同。4.4 退市股与ST股的规范化处理另一个容易被忽略的坑是ST和退市股。ST股有5%的涨跌幅限制但这一信息不会直接出现在K线数据里只能通过股票名称code_name字段判断。如果你在全量数据里看到某只股票的某一天涨幅超过5%那不是数据错误而是该股票在那一天还不叫ST或者刚摘帽。真正的数据错误是某只股票连续多日涨跌幅恒为0但tradestatus显示为1这通常是该股票长期停牌没有及时更新交易状态。处理方式是在校验逻辑里加一条若同股票连续N天比如30天成交量为0但状态为正常则标记为可疑数据人工复核。数据校验完成后整个数据集就可以作为后续研究的基础了。不要只在本地存一份建议同时导出一份CSV格式的备份放到另一块磁盘或对象存储上因为Parquet虽然读取快但如果你后续要用R、SPSS或者其他不支持Parquet的工具读取就会面临格式转换的问题。CSV备份的代价是占用存储空间翻倍但考虑到单份全量日线数据压缩后也就2GB左右这个代价值得付。整个回补流程跑通后你会得到一个完整的、时间连续的、字段齐全的沪深A股日线数据集它可以支撑绝大多数量化研究和数据分析工作。本文还有配套的精品资源点击获取
沪深股票全量日线回补实战：从baostock选型到数据校验
简介沪深股票历史日线数据资源覆盖沪深两市全部股票自上市日至2022年1月10日的每日行情。数据字段除开盘价、收盘价、最高价、最低价、振幅、成交量、成交额、换手率外还预计算了MACD、CCI以及同花顺手机版多空指标可直接用于判断价格趋势、识别超买超卖区域和辅助买卖决策对技术分析爱好者、金融研究员及量化交易者都有实用价值。整包仅1个SQL文件共1个文件压缩后约424.52MB利用SQL可灵活完成区间统计、指标筛选与回测数据提取免去自行爬取清洗的繁琐步骤。已有605人学习下载数据结构清晰便于回溯不同市场周期下的个股表现也可用于构建多因子模型或验证经典技术指标的有效性适合需要全量历史A股数据做策略复盘或模型训练的开发者直接使用。1. 沪深股票全量日线回补从接口选型到数据校验的一次性到位拿到“沪深股票历史以来到2022-01-10的全部日线数据”这个需求第一反应不是写代码而是先想清楚一件事你要的是“能跑通的脚本”还是“能用于因子回测的可靠数据集”。这两者的差距非常大。2022年1月10日这个截止点意味着你需要从交易所最早的数据上证1990年12月19日、深证1991年4月3日一直拉到指定日期跨越30多年、5000多只股票、每只股票几千条K线总量在千万级记录。这个量级如果用公共接口逐日请求按每秒3次的频率限制纯串行下载需要跑十几个小时而且大概率中途断掉。更麻烦的是不同数据源对“历史以来”的定义不同有的只给最近三年有的复权因子缺失有的停牌日直接留空洞。把这些问题拆开看核心就三个选对数据源、设计断点续传的回补流程、验证数据完整性和复权一致性。这篇文章把这套方案完整讲透。2. 数据源选型与全量历史数据的获取机制2.1 为什么免费接口里baostock最适合全量回补市面上的A股日线数据接口看似很多但能做全量历史回补的窗口其实很窄。你搜“沪深股票历史数据接口”最先跳出来的是tushare、akshare、baostock三个。tushare的pro接口需要积分门槛基础积分只能取最近几年的数据历史全量需要较高积分或者付费对于一次性回补需求不划算。akshare的底层多为爬虫实现东方财富、新浪财经这些源经常改版爬虫稳定性没法保证而且历史数据最早只到2000年左右再往前就断了。baostock是专门为量化研究设计的免费接口不需要token不限积分数据从交易所成立日就开始覆盖而且是批量返回——一次查询直接给一整只股票的全历史日线不用自己按日期翻页。这就是为什么做全量回补时baostock是首选。2.2 baostock的登录校验与数据工厂模式baostock的使用方式比较特殊它不是函数式调用而是先登录拿到会话再通过query_history_k_data_plus这个工厂方法取数据。登录动作会做本机校验和网络握手返回的结果码0表示成功10001001表示网络错误10001002表示密码错误。虽然它是免费接口但登录态是有时效的长时间运行需要定期检查连接状态或者干脆每次批量任务重新登录。import baostock as bs import pandas as pd from datetime import datetime # 登录获取会话 lg bs.login() if lg.error_code ! 0: print(f登录失败: {lg.error_msg}) raise SystemExit(1) # 查询平安银行全历史日线 rs bs.query_history_k_data_plus( sz.000001, date,code,open,high,low,close,preclose,volume,amount,adjustflag,turn,tradestatus, start_date1991-04-03, end_date2022-01-10, frequencyd, adjustflag3 )这段代码里的adjustflag参数决定了复权行为3表示不复权1表示后复权2表示前复权。全量回补时这里建议取3不复权因为后面做因子计算时需要自己控制前复权/后复权逻辑依赖接口的复权结果容易踩坑。tradestatus字段是关键1表示正常交易0表示停牌回补时不能只按日期切分还得按交易状态过滤。2.3 交易日历的坑为什么不能按自然日顺序拉取直接按股票代码遍历拉取是最直觉的做法但这里有一个隐藏的效率问题。baostock的query_history_k_data_plus是按股票维度返回所有历史K线不是按日维度。也就是说每调用一次就拿回一整只股票的全量数据循环5000次就能覆盖全市场。这个设计非常适合全量回补但前提是你得先拿到完整的股票列表。常见的做法是用query_stock_basic拉取全部证券代码及状态然后过滤出股票类型再逐只回补。# 获取全部证券代码 rs bs.query_stock_basic() stock_list [] while (rs.error_code 0) rs.next(): row rs.get_row_data() stock_list.append({ code: row[0], # 证券代码 code_name: row[1], # 证券名称 type: row[3], # 1:股票 2:指数 3:其它 status: row[4] # 1:上市 0:退市 }) df_stocks pd.DataFrame(stock_list) # 只保留股票同时包含已退市的保证历史数据完整 df_stocks df_stocks[df_stocks[type] 1] print(f股票总数: {len(df_stocks)})这里一定要注意status字段不要过滤掉退市股。沪深股票的历史全量数据里退市股是必须包含的否则做幸存者偏差研究时数据就是残缺的。baostock对退市股的历史数据仍然正常返回只是最新的交易状态是退市。在拉取时你会发现有些退市股的上市日期非常早比如最早的一批“老八股”它们的日线数据从1990年底就开始了。2.4 增量字段的设计一次回补长期受益全量回补不应该是用完一次就扔的脚本。2022年1月10日之后的每个交易日你可能还需要增量追加数据。所以存储格式在一开始就要设计好。我建议用Parquet按年分区存储而不是直接存CSV原因有三个一是千万级行数的CSV加载要几十秒Parquet只要几秒二是Parquet自带列式压缩磁盘占用只有CSV的三分之一左右三是后续做增量更新时可以直接用pyarrow的dataset API做分区写入效率高很多。字段设计上至少要保留date、code、open、high、low、close、volume、amount、adjustflag、turn、tradestatus这11个字段其中adjustflag存当时取数用的复权标记这是事后校验数据一致性的重要依据。import pyarrow as pa import pyarrow.parquet as pq from pathlib import Path # 按年分区的写入逻辑 def save_daily_data(df: pd.DataFrame, base_dir: str ./stock_data): df[year] pd.to_datetime(df[date]).dt.year for year, group in df.groupby(year): out_path Path(base_dir) / fyear{year} out_path.mkdir(parentsTrue, exist_okTrue) # 单只股票追加写入使用行组追加模式 pq.write_to_dataset( group, root_pathbase_dir, partition_cols[year], existing_data_behavioroverwrite_or_ignore )overwrite_or_ignore这个参数很关键它允许在已有的分区上追加新数据而不覆盖旧文件。不过要特别注意这个参数是按文件粒度去重的如果你的增量数据跨多个交易日后又回头去补前几天的数据就会产生重复行。所以更稳妥的做法是把增量数据的日期范围严格控制在上一次回补的截止日之后。3. 全量回补的完整流程断点续传、频率控制与异常重试3.1 基于已回补记录的状态表全量回补几十万次接口调用中途必然出问题网络断、接口限流、进程被杀。不设计断点续传机制的回补脚本就是一次性脚本重跑时要全部重新拉。正确的做法是先建一张回补状态表记录每只股票的代码、最后成功回补的日期、失败次数和错误信息每次启动时先检查这张表跳过已经完成的只重试失败的。这张状态表本身也可以和K线数据放在同一个Parquet目录下用一个小型CSV文件维护就行。# 初始化回补状态目录 mkdir -p ./stock_data/checkpoint touch ./stock_data/checkpoint/stock_progress.csv状态表的最小字段是code、last_date、status、retry_count、last_error。last_date用于增量补数据时判断从哪天开始拉status标记done、failed、partial三种状态retry_count控制最大重试次数last_error记录最近一次失败原因方便排查。3.2 主循环逐股回补与自适应限速主循环的逻辑比想象中简单但几个细节决定成败。首先是查询接口的start_date和end_date全量回补时start_date直接填1990-12-19上证最早交易日end_date填2022-01-10不逐日请求一次拉全。其次是频率控制baostock对并发没有严格限制但单进程连续高频请求会触发服务端的限流表现为返回码为10002003请求过于频繁。最稳的做法是加入自适应限速正常请求间隔0.1秒如果遇到限流指数退避到1秒、2秒、4秒直到成功。import time import random def fetch_stock_history(bs, code: str, start: str, end: str, max_retry: int 5): retry_count 0 wait_time 0.1 # 正常间隔 while retry_count max_retry: rs bs.query_history_k_data_plus( code, date,code,open,high,low,close,preclose,volume,amount,adjustflag,turn,tradestatus, start_datestart, end_dateend, frequencyd, adjustflag3 ) if rs.error_code 0: # 成功读取数据 rows [] while rs.next(): rows.append(rs.get_row_data()) df pd.DataFrame(rows, columnsrs.fields) # 只保留正常交易日 df df[df[tradestatus] 1] return df elif rs.error_code 10002003: # 限流指数退避 retry_count 1 time.sleep(wait_time * (2 ** retry_count) random.uniform(0, 0.5)) else: # 其他网络错误等待后重试 retry_count 1 time.sleep(2 retry_count) return None这段代码的处理逻辑是成功时一次性把所有行取干净rs.next()循环要走到返回False为止否则会导致后续查询结果串数据。tradestatus 1的过滤是为了剔除停牌日停牌日接口返回的空值行如果不处理后续计算收益时会当成0处理造成严重误差。这里有个容易被忽略的细节baostock的K线数据里也有tradestatus为0的行这些行不是没有数据而是当天停牌所以交易字段为空如果你直接用dropna()清掉就会误删整个日期行导致时间轴断裂——虽然很低级但很多人第一次做都会踩。3.3 全市场并行与进程级容错单线程逐只回补到2022年1月10日对5000多只股票来说假设单只股票返回平均120毫秒大概需要10分钟。这个速度其实已经可以接受不需要上多线程。但如果想跑得更快可以用concurrent.futures做线程池但要注意baostock的session不是线程安全的每个线程必须创建独立的连接。常见做法是用threading.local存线程专属的session或者干脆用多进程每个进程独立登录。如果你选择多进程我建议每500只股票一个子进程进程内用ThreadPoolExecutor开4个线程各线程独立baostock登录。整体速度可以压到2分钟左右。但代价是代码复杂度上升调试难度增大。我先给一个稳妥的单线程方案因为对多数一次性回补来说10分钟的耗时是可以接受的。import pandas as pd from pathlib import Path def backfill_all_stocks(df_stocks, checkpoint_file: str, end_date: str 2022-01-10): # 读取已有进度 if Path(checkpoint_file).exists(): df_ckpt pd.read_csv(checkpoint_file, dtype{code: str}) else: df_ckpt pd.DataFrame(columns[code, last_date, status, retry_count, last_error]) for _, row in df_stocks.iterrows(): code row[code] # 如果已经回补完成跳过 existing df_ckpt[df_ckpt[code] code] if not existing.empty and existing.iloc[0][status] done: continue df fetch_stock_history(bs, code, 1990-12-19, end_date) if df is not None and len(df) 0: save_daily_data(df) # 更新进度 update_checkpoint(df_ckpt, code, df[date].max(), done, 0, ) else: update_checkpoint(df_ckpt, code, , failed, 1, query return empty)这里每次循环拿到的df是整个股票的历史需要及时写入磁盘释放内存。你可能会问一只股票30年的日线也就8000行左右内存压力不大但5000只全量回补如果一次性攒在内存里不写盘最后阶段会占用几百MB甚至上GB。渐进写盘还有个好处是中断后已写入的数据立即可用不必等全部跑完。3.4 增量更新的时间边界全量回补完成之后2022-01-10之后的新交易日怎么追加这是另一个常见需求场景数据需要每天更新。增量更新的起点是每只股票在状态表里的last_date 1个交易日终点是今天。但注意baostock的接口只接受YYYY-MM-DD字符串不接受datetime对象而且start_date必须是交易日如果填了非交易日接口不会报错而是返回空结果。所以增量更新的标准做法是先用query_trade_dates拿到最近的交易日历再以last_date的下一个交易日作为起点。def get_trade_dates_after(bs, last_date: str, end_date: str): rs bs.query_trade_dates(start_datelast_date, end_dateend_date) dates [] while (rs.error_code 0) rs.next(): row rs.get_row_data() # 第3个字段是日历日期第4个字段是是否交易日 dates.append(row[1]) return dates # 增量更新示例找到last_date之后的下一个交易日 from datetime import datetime, timedelta last_date 2022-01-07 next_date datetime.strptime(last_date, %Y-%m-%d) timedelta(days1) # 用query_trade_dates确认next_date是否为交易日如果不是顺延增量更新的频率控制可以更激进因为每天只需要拉几千只股票当天的数据接口压力小。但你如果用的是同一套查询函数建议保留原来的限速策略保证在极端情况下比如一次性回补7天没跑也能稳定完成。4. 复权处理的三种选择与数据质量校验方法4.1 前复权、后复权、不复权的适用场景标题里的“全部日线数据”没有指定复权方式但实际应用中你必须在取数时就决策因为这直接影响后续策略表现。不复权数据是用来算真实成交额和成交量的任何涉及到金额绝对值的统计比如流动性分析都只能用不复权数据。前复权数据适合技术分析因为它保留了最近的价格绝对值K线形态和均线系统看起来直观但前复权数据的历史价格是动态变化的每次新的除权除息事件发生历史价格都会整体变动所以在回测中反复加载前复权数据会导致价格不一致。后复权数据不存在这个问题它把所有历史价格按复权因子统一调整序列连续且唯一适合做长期收益计算和因子回测。4.2 用不复权数据手动计算复权因子的完整代码因为baostock的adjustflag3是不复权数据需要自己实现复权因子的计算。常见做法是用preclose昨收和close今收的关系来推算除权因子。这个逻辑很简单正常情况下close / preclose接近当日收益率但如果发生除权除息这个比值会偏离正常波动范围偏离的比例就是复权因子。注意这个计算只对除权除息日之后的序列有效前复权需要从最新日期往回推导后复权需要从最早日期往未来推导。import numpy as np def calc_adjust_factor(df: pd.DataFrame) - pd.DataFrame: df df.sort_values(date).reset_index(dropTrue) # 理论收益率当日收盘 / 昨收 df[ret_raw] df[close] / df[preclose] # 实际区间最大涨跌幅过滤涨停/跌停之外的视为除权 mask (df[ret_raw] 1.11) | (df[ret_raw] 0.89) df[adjust_factor_raw] np.where(mask, df[ret_raw], 1.0) # 从最早日期累计后复权因子 df[adjust_factor_post] df[adjust_factor_raw].cumprod() # 前复权因子 后复权因子 / 最新后复权因子 latest_factor df[adjust_factor_post].iloc[-1] df[adjust_factor_pre] df[adjust_factor_post] / latest_factor # 生成复权价格 df[close_adj_pre] df[close] * df[adjust_factor_pre] df[close_adj_post] df[close] * df[adjust_factor_post] return df这段代码的处理精度对大多数场景够用但有个明显的坑正常涨停/跌停日的close/preclose也可能超过1.1或低于0.9特别是创业板注册制之后20%的涨跌幅限制会把正常的涨跌误判为除权。所以这个逻辑需要配合turn换手率或者成交量变化来二次过滤。一个简单的修正办法是如果当日成交量没有异常放大就不视为除权日。4.3 数据完整性校验用同源和异源数据交叉验证全量回补完成后数据质量校验是不可省的一步。校验分两个维度一是完整性检查每只股票的日线数量和时间连续性二是一致性用另一数据源抽取部分股票做收盘价对比容差设为0.5%四舍五入到小数点后两位导致的误差范围。# 检查每只股票是否存在跳空缺口停牌日除外 def validate_continuous(df: pd.DataFrame, trade_dates: set) - list: missing_dates [] for date in df[date]: if date not in trade_dates: missing_dates.append(date) return missing_dates # 抽样与akshare对比 # import akshare as ak # df_ak ak.stock_zh_a_hist(symbol000001, perioddaily, # start_date20200101, end_date20220110, adjust) # 对比close列允许0.5%误差对比时注意两个细节一是抽样不能只看头部股票要按代码分布随机抽50只覆盖不同上市年份和不同行业二是截止日期2022-01-10当天的数据在收盘前是无法取到的所以校验样本范围要往前挪一天从2022-01-07及之前取。如果发现某些股票在baostock和akshare中的收盘价不一致优先检查该股票是否发生过送转或配股因为两个数据源对复权因子的处理可能不同。4.4 退市股与ST股的规范化处理另一个容易被忽略的坑是ST和退市股。ST股有5%的涨跌幅限制但这一信息不会直接出现在K线数据里只能通过股票名称code_name字段判断。如果你在全量数据里看到某只股票的某一天涨幅超过5%那不是数据错误而是该股票在那一天还不叫ST或者刚摘帽。真正的数据错误是某只股票连续多日涨跌幅恒为0但tradestatus显示为1这通常是该股票长期停牌没有及时更新交易状态。处理方式是在校验逻辑里加一条若同股票连续N天比如30天成交量为0但状态为正常则标记为可疑数据人工复核。数据校验完成后整个数据集就可以作为后续研究的基础了。不要只在本地存一份建议同时导出一份CSV格式的备份放到另一块磁盘或对象存储上因为Parquet虽然读取快但如果你后续要用R、SPSS或者其他不支持Parquet的工具读取就会面临格式转换的问题。CSV备份的代价是占用存储空间翻倍但考虑到单份全量日线数据压缩后也就2GB左右这个代价值得付。整个回补流程跑通后你会得到一个完整的、时间连续的、字段齐全的沪深A股日线数据集它可以支撑绝大多数量化研究和数据分析工作。本文还有配套的精品资源点击获取
