尧图网页设计 
QLIB 这套平台我用了快两年一直觉得它很像那种“主业很强但生活自理能力堪忧”的同学回测效率高因子库全机器学习流水线也是现成的但唯独数据这件事它默认你得自己搞定。我去年年底搭本地A股研究环境时最折腾的竟然不是模型而是“每天收盘后怎么把最新行情更新进QLIB”。当时在社区里搜了一圈要么是讲QLIB怎么跑官方数据集的要么是单独讲tushare怎么拉数据的很少有人把这两者怎么接起来、每天怎么维持更新讲透。这篇文章就我目前在用的“QLIB tushare每日行情更新”方案完整复盘一遍内容包括数据格式设计、复权方式选择、全量初始化和增量更新的完整代码以及我踩过的一堆坑。目标是让你拿到手就能搭出一条自动更新的数据管道而不是看完还是一头雾水。1. 先把“更新每日行情”这件事拆开看1.1 QLIB 不是“带数据的回测框架”它只负责计算QLIB 是微软开源的一个量化投资平台核心优势是把特征工程、模型训练、回测、实盘交易这整条链路打通。但它有个很鲜明的特点数据默认由使用者自己准备。官方仓库里的示例数据主要是为了让新用户跑通流程而你一旦接入自己的行情源就必须面对它的数据格式约定。QLIB 的数据存储分三层calendars/day.txt交易日历一行一个日期格式是YYYY-MM-DDinstruments/all.txt股票池列表记录每只代码的上市日期、退市日期features/instrument.bin每只股票的二进制特征矩阵列对应字段行对应日历中的交易日。这种设计在读取时非常高效因为模型训练阶段会大量随机访问历史数据二进制文件比 CSV 少一次解析开销。代价就是所有外部数据都需要先切成规范格式再用官方脚本转换成.bin。一开始我也嫌麻烦后来对这个结构反而有了好感一旦数据按 QLIB 的约定存好后面跑 Alpha158、做滚动训练、切换因子集都不用再操心数据层的问题。反过来如果数据更新没做好哪怕因子代码写得再对出来的结果也是脏的。所以“更新每日行情”这件事本质上不是“把今天的K线下载下来”就完了而是要维护好三个文件体系日历要追加、股票池要更新、特征矩阵要补上最新一行。这三个动作缺一个第二天跑策略时都会出事。下面我会按这个思路一步步展开。1.2 为什么数据管道选 tushare 而不是别的方式市面上能拿A股日线的数据源不少baostock、akshare、tushare我都试过。baostock免费且稳定但字段偏少复权因子要自己处理akshare接口多但上游经常改页面结构维护成本偏高tushare虽然需要积分但pro_bar这个接口对日线行情、复权、复权因子的支持是最省心的。我选择 tushare 的核心原因有三点接口稳定字段规范。日线级别的 OHLCV 在pro_bar里都是现成的字段名和 QLIB 的约定差距很小只有vol和volume这种小差异。复权因子可以直接拿。tushare 算好了adj_factor不管你用前复权还是后复权都不用自己推算除权除息。交易日历完整。做增量更新时最怕的就是“今天是不是交易日”判断错trade_cal接口直接告诉你哪天开市哪天休市。你可能觉得“直接去财经网站下载历史数据再转CSV”也行但那种方式的问题是无法自动化每天收盘后手动下载、手动命名、手动合并坚持一周就会崩溃。tushare 的接口方式虽然要写点代码但一次性搭好之后每天跑脚本就好这是长期研究最舒服的状态。2. 动手前必须想清楚的三件事2.1 数据源接口怎么选pro_bar、daily、trade_cal 分工不同tushare 的接口很多但做每日行情更新时真正高频用到的就那么几个。我列一张表把用途和注意事项写清楚接口作用注意事项ts.pro_bar获取日线行情支持复权核心接口返回ts_code、trade_date、open、high、low、close、vol、amount等pro.daily获取未复权日线pro_bar底层就是它一般不需要直接调pro.trade_cal获取交易日历判停方法is_open1才是交易日pro.stock_basic获取股票基础信息用来生成 QLIB 的instruments/all.txt注意区分上市和退市状态pro.adj_factor获取复权因子如果你要自己算复权价用原始价乘以因子即可ts.pro_bar是最省事的入口。它内部会帮你把除权除息数据算好常见的调用方式是这样的import tushare as ts pro ts.pro_api(你的token) df ts.pro_bar( ts_code600000.SH, adjhfq, start_date20240101, end_date20241231 )adj参数有三档不填就是未复权qfq表示前复权hfq表示后复权。这块怎么选我放到 2.3 节专门讲因为它直接影响历史数据的稳定性。2.2 QLIB 对输入文件的目录和列名要求QLIB 官方提供了一个数据转换脚本dump_bin.py它负责把 CSV 转成二进制。但脚本对 CSV 的目录结构是有明确要求的。你在准备数据之前得先搭好这个目录csv_data/ ├── calendars/ │ └── day.txt # 交易日列表每行一个日期形如 2024-01-02 ├── instruments/ │ └── all.txt # 股票列表代码 上市日期 退市日期 └── features/ ├── SH600000.csv # 每只股票一个 CSV文件名是 QLIB 风格的代码 ├── SZ000001.csv └── ...这里有个容易踩坑的地方dump_bin.py在dump_all模式下会去features目录下找所有以.csv结尾的文件然后根据文件名确定股票代码。文件名必须是SH600000这种格式也就是“交易所前缀 6位数字”tushare 返回的是600000.SH这种格式需要转换。CSV 内部格式也有要求。第一列必须是日期列名建议叫date格式为%Y-%m-%d后面的列是要入库的特征字段。比如date,open,high,low,close,volume 2024-01-02,10.5,10.8,10.4,10.6,100000 2024-01-03,10.6,10.9,10.5,10.8,120000dump_bin.py转出来的.bin文件会按calendars/day.txt的日历对齐。也就是说如果某只股票 CSV 里缺少某个交易日的记录那它在.bin里对应的那一行就是 NaN。2.3 tushare 如何计算复权QLIB 归档时应该用哪种复权这个问题是新手最容易翻车的环节。先说 tushare 那边是怎么处理的。tushare 的复权因子adj_factor是一个随交易日变化的因子序列。如果你拿到原始价和复权因子后复权价的计算公式是这样的后复权价 原始价 * 对应日期的复权因子前复权价则是在后复权价的基础上再除以最新一期的复权因子前复权价 原始价 * 复权因子 / 最新复权因子你可以用代码验证一下ts.pro_bar就算你不传adj返回的也是未复权价传了hfq返回的就是原始价乘以因子之后的结果。这两种方式最终数值是一致的区别只是 tushare 帮你算好了省去你手动 merge 因子表。那问题来了QLIB 入库时应该用哪种我个人强烈建议用后复权hfq不要用前复权qfq。原因是前复权有一个天生缺陷每当有新数据进来最新复权因子可能变化然后整段历史价格都会被重算。这意味着你今天训练好的模型特征明天可能因为数据更新悄悄变了实验就不可复现了。后复权则没有这个问题只要除权除息事件不发生改变历史价格一旦入库就固定下来。唯一的缺点是后复权价和当前市价差得很远但这个缺点对 QLIB 几乎没影响因子计算绝大多数都在用收益率、均线、波动率这些对单调缩放不敏感的量绝对价格水平本来也不重要。我见过有人把未复权价格直接入库然后用复权因子在 QLIB 里做动态调整这样更精确但实现难度高对多数场景没必要。后复权是从“实现简单、结果稳定、误差可接受”三个维度综合下来的最优解。3. 实战完整跑通“拉取 - 清洗 - 入库”流程3.1 首次全量初始化把历史行情完整拉下来初始化阶段的目标是把某个股票池从起始日期到当前日期的全部日线数据拉下来整理成 QLIB 要的 CSV 目录结构。先说股票池。一般研究环境主要关心当前上市状态的股票用stock_basic接口可以拿到pro ts.pro_api(你的token) # 获取当前上市状态的股票列表 stock_list pro.stock_basic( exchange, # 空表示沪深两市 list_statusL, # L上市D退市P暂停上市 fieldsts_code,symbol,name,area,industry,list_date ) print(stock_list.shape)list_statusL拿到的就是当前正常上市的股票。注意如果要做长期回测还要考虑历史上有过上市但后来退市的股票否则回测到 2018 年时会漏掉一堆当时还活着的股票。实际使用中我建议全量初始化时用list_statusL,D,P把所有状态都拉进来然后在instruments/all.txt里用上市日期和退市日期去约束每只股票的有效区间。接着写一个拉取日线的函数def ts_code_to_qlib(ts_code): 将 tushare 的 600000.SH 转成 QLIB 的 SH600000 code, exchange ts_code.split(.) return (SH if exchange SH else SZ) code def download_stock_daily(ts_code, start_date, end_date): 拉取单只股票的后复权日线并整理成 QLIB CSV 格式 df ts.pro_bar( ts_codets_code, adjhfq, start_datestart_date, end_dateend_date ) if df is None or df.empty: return None df df.rename(columns{vol: volume}) df[date] pd.to_datetime( df[trade_date], format%Y%m%d ).dt.strftime(%Y-%m-%d) df df[[date, open, high, low, close, volume]] df df.sort_values(date) return df这里有几个细节需要说明pro_bar的ts_code参数要传600000.SH这种格式不能传600000adjhfq是后复权trade_date返回的是20240102这种 8 位字符串要转成2024-01-02因为 QLIB 的日历格式要求带横线vol要重命名为volumeQLIB 默认字段名里没有vol这种缩写。历史区间怎么选如果你做日频因子研究建议从 2007 年或 2010 年开始太早的数据一方面复权因子可能缺失另一方面对当前市场的参考价值有限。我自己常用20050101作为起点因为 tushare 对这个时间之后的数据覆盖最完整。把所有股票循环一遍写成 CSVimport os import time output_dir csv_data/features os.makedirs(output_dir, exist_okTrue) for idx, row in stock_list.iterrows(): ts_code row[ts_code] qlib_code ts_code_to_qlib(ts_code) file_path os.path.join(output_dir, f{qlib_code}.csv) if os.path.exists(file_path): continue df download_stock_daily(ts_code, 20050101, 20241231) if df is not None: df.to_csv(file_path, indexFalse) # 控制访问频率避免触发接口限流 time.sleep(0.2)这个循环看起来简单但真跑起来你会发现问题几千只股票一只一只拉总共要几分钟到十几分钟中间只要限流一次就断掉。所以完整脚本里我会加一个“断点续传”的逻辑也就是if os.path.exists(file_path): continue。这个逻辑在初始化中断后重新执行时已经拉过的股票就会被跳过节省大量时间。3.2 用 dump_bin 把 CSV 转成 QLIB 的二进制格式CSV 准备完毕之后接下来要生成 QLIB 数据目录。先创建日历文件和 instruments 文件。日历文件csv_data/calendars/day.txt的内容是所有交易日按从早到晚排序一行一个日期。这个文件可以从 tushare 的trade_cal取得cal pro.trade_cal( exchangeSSE, # 上交所沪深通常保持一致 start_date20050101, end_date20241231, is_open1 # 只取交易日 ) cal cal.sort_values(cal_date) with open(csv_data/calendars/day.txt, w) as f: for d in cal[cal_date]: f.write(f{d[:4]}-{d[4:6]}-{d[6:8]}\n)instruments 文件csv_data/instruments/all.txt的格式是每行两列或三列用 Tab 分隔。第一列是 QLIB 格式代码第二列是上市日期第三列是退市日期没有退市的可以写一个未来日期或者写你数据末尾的日期。QLIB 在使用时会读取这个文件决定每只股票什么时候进入、什么时候退出。with open(csv_data/instruments/all.txt, w) as f: for idx, row in stock_list.iterrows(): ts_code row[ts_code] qlib_code ts_code_to_qlib(ts_code) list_date row[list_date] # 转成 YYYY-MM-DD 格式退市日期用数据末尾日期代替 list_date_fmt f{list_date[:4]}-{list_date[4:6]}-{list_date[6:8]} f.write(f{qlib_code}\t{list_date_fmt}\t2024-12-31\n)之后调用官方脚本dump_bin.py。这个脚本在 QLIB 仓库的scripts/目录下直接下载就有python scripts/dump_bin.py dump_all \ --csv_path csv_data \ --qlib_dir qlib_data \ --include_fields open,high,low,close,volume参数含义dump_all全量转储模式会扫描csv_path下面所有 CSV逐个生成.bin--qlib_dir输出目录QLIB 之后就用这个目录作为provider_uri--include_fields指定要保留的字段多个字段用逗号分隔顺序可以自己定但建议按照open,high,low,close,volume排列。跑完之后qlib_data目录下会出现和csv_data一样的结构只不过features目录下的文件从.csv变成了.bin。你可以随机挑一只股票用 QLIB 读一下验证import qlib from qlib.data import D qlib.init(provider_uriqlib_data) data D.features( [SH600000], [$open, $high, $low, $close, $volume], start_time2024-01-01, end_time2024-12-31, freqday ) print(data.tail())如果输出正常说明数据入库成功。3.3 每日增量更新脚本的核心写法全量初始化只需要跑一次之后每天要做的是增量更新。增量更新的核心思路是只处理最新的交易日数据把新行追加到 CSV 末尾再跑dump_update更新.bin。先判断今天是不是交易日from datetime import datetime today datetime.now().strftime(%Y%m%d) cal pro.trade_cal(exchangeSSE, start_datetoday, end_datetoday) if cal.empty or cal[is_open].iloc[0] 0: print(f{today} 不是交易日跳过更新) exit(0)然后拉取所有股票当天的数据追加到每只股票对应的 CSV 里stock_list pro.stock_basic(exchange, list_statusL, fieldsts_code) for idx, row in stock_list.iterrows(): ts_code row[ts_code] qlib_code ts_code_to_qlib(ts_code) file_path os.path.join(output_dir, f{qlib_code}.csv) # 拉当日数据 df download_stock_daily(ts_code, today, today) if df is None or df.empty: # 停牌或当日无数据跳过 continue # 读取历史 CSV old pd.read_csv(file_path, dtype{date: str}) # 避免重复追加同一行 if str(old[date].iloc[-1]) today_fmt: continue merged pd.concat([old, df], ignore_indexTrue) merged merged.drop_duplicates(subset[date]).sort_values(date) merged.to_csv(file_path, indexFalse)这里有一个很重要的细节追加前必须检查 CSV 最后一行的日期。如果上次更新已经成功而这次脚本因为某种原因重复执行会导致同一行数据被写两次。drop_duplicates可以兜底但更保险的做法是直接对比日期大于等于当前交易日就不追加。增量更新完 CSV 后还要更新calendars/day.txt。如果今天是交易日就把今天的日期追加进去如果还没有的话cal_file csv_data/calendars/day.txt with open(cal_file, r) as f: existing_days set(line.strip() for line in f) today_fmt f{today[:4]}-{today[4:6]}-{today[6:8]} if today_fmt not in existing_days: with open(cal_file, a) as f: f.write(today_fmt \n)最后跑dump_updatepython scripts/dump_bin.py dump_update \ --csv_path csv_data \ --qlib_dir qlib_data \ --include_fields open,high,low,close,volumedump_update和dump_all的区别在于它只会对已经存在的.bin文件追加新日期数据不会重新生成全部文件。这样每次更新只需要处理有新增记录的股票速度快很多。如果你中途改了字段配置或者怀疑数据有错可以偶尔强制跑一次dump_all全量重建。4. 每日更新阶段躲不开的四个经典问题4.1 日期类型不一致导致日历对不上这个问题我撞上过不止一次。QLIB 的day.txt里日期格式是2024-01-02字符串类型。而你在 CSV 里如果用了datetime类型或者用了20240102这种整型dump_bin在比对日期时就会出错轻则报 Warning重则直接生成错位的日历。解决方式很粗暴所有日期在写入 CSV 和day.txt之前统一转成字符串并调用strftime(%Y-%m-%d)不要混用类型。在读取已有 CSV 时也要指定dtype{date: str}避免 pandas 自动推断成别的类型。4.2 停牌股票让 dump_update 直接报错增量更新时最容易遇到的运行时错误就是股票停牌。比如某只股票今天停牌tushare 返回空数据那这只股票的 CSV 末尾就不会追加新行。到这里为止没问题问题出在dump_update的日期对齐上如果day.txt已经加入了今天但某只股票的.bin文件没有今天的记录QLIB 就会认为这只股票在今天的特征全是 NaN。如果只是个别股票这个行为可以接受模型训练时一般会自动 dropna。但如果大量股票停牌比如春节前最后一天dump_update可能因为“CSV 末尾日期与日历文件中最后一个交易日不一致”而直接跳过某些文件或报错。我现在的处理策略是更新day.txt时只把“实际有股票交易的交易日”追加进去而不是简单地把当天塞进去。更简单的方法是用上一个交易日做判断如果今天不是交易日干脆不追加如果今天是交易日但某些股票停牌就让它们在.bin里表现为 NaN不要试图去补一条假数据。4.3 接口限流让全量拉取中途卡死tushare 的积分制度决定了它有访问频率限制。积分越低每分钟能请求的次数越少。你在全量初始化时要对几千只股票循环调pro_bar一旦限流接口会返回一段错误提示而不是异常退出如果你没有对返回结果做校验就会把None写进 CSV最后导致dump_bin读取失败。我的做法是写一个带重试的包装函数def download_stock_daily_with_retry(ts_code, start_date, end_date, retries3): for attempt in range(retries): try: df ts.pro_bar( ts_codets_code, adjhfq, start_datestart_date, end_dateend_date ) if df is not None and not df.empty: return df time.sleep(0.5) except Exception as e: print(f{ts_code} 第 {attempt1} 次请求失败: {e}) time.sleep(1) return None再加上一个经验值pro_bar之间至少间隔 0.2 秒积分不够的间隔拉到 0.5 秒。宁可慢一点也不要中断后从头再来。4.4 字段名不一致特征矩阵默默缺列还有一个隐蔽问题字段名不一致。tushare 的日线接口返回vol但 QLIB 的特征字段如果叫volume那dump_bin会把vol当做一个独立字段存进去。结果就是你读特征时用$volume拿不到数据因为.bin里存的列名是$vol。这个问题不会报错只会让某个特征列全部是 NaN而且模型的日志里不一定能立刻发现。我的建议是在写入 CSV 之前统一把 tushare 字段映射成 QLIB 字段vol - volume、amount - amount保持原名在dump_bin的--include_fields里明确写清楚最终字段名入库后马上读一次验证用上节提到的D.features检查列名和数值是否正常。5. 一套可以直接抄作业的完整更新脚本5.1 脚本整体逻辑和工作模式为了让你少走弯路我把上面的代码整合成一个完整脚本支持两种模式init和update。init模式做全量初始化update模式做每日增量。import os import time import argparse import pandas as pd import tushare as ts from datetime import datetime TOKEN 你的tushare_token CSV_ROOT csv_data QLIB_ROOT qlib_data START_DATE 20050101 ts.set_token(TOKEN) pro ts.pro_api() def ts_code_to_qlib(ts_code): code, exchange ts_code.split(.) return (SH if exchange SH else SZ) code def download_stock_daily(ts_code, start_date, end_date, retries3): for attempt in range(retries): try: df ts.pro_bar( ts_codets_code, adjhfq, start_datestart_date, end_dateend_date ) if df is not None and not df.empty: df df.rename(columns{vol: volume}) df[date] pd.to_datetime( df[trade_date], format%Y%m%d ).dt.strftime(%Y-%m-%d) df df[[date, open, high, low, close, volume]] return df.sort_values(date) time.sleep(0.5) except Exception as e: print(f{ts_code} 请求失败: {e}) time.sleep(1) return None def init_all(): os.makedirs(os.path.join(CSV_ROOT, features), exist_okTrue) stock_list pro.stock_basic( exchange, list_statusL,D,P, fieldsts_code,symbol,name,area,industry,list_date ) # 全量拉取日线 for idx, row in stock_list.iterrows(): ts_code row[ts_code] qlib_code ts_code_to_qlib(ts_code) file_path os.path.join(CSV_ROOT, features, f{qlib_code}.csv) if os.path.exists(file_path): continue df download_stock_daily(ts_code, START_DATE, datetime.now().strftime(%Y%m%d)) if df is not None: df.to_csv(file_path, indexFalse) print(f[{idx}/{len(stock_list)}] {qlib_code} 完成) time.sleep(0.2) # 写日历文件 cal pro.trade_cal( exchangeSSE, start_dateSTART_DATE, end_datedatetime.now().strftime(%Y%m%d), is_open1 ).sort_values(cal_date) with open(os.path.join(CSV_ROOT, calendars, day.txt), w) as f: for d in cal[cal_date]: f.write(f{d[:4]}-{d[4:6]}-{d[6:8]}\n) # 写 instruments 文件 with open(os.path.join(CSV_ROOT, instruments, all.txt), w) as f: for idx, row in stock_list.iterrows(): ts_code row[ts_code] qlib_code ts_code_to_qlib(ts_code) list_date row[list_date] if pd.isna(list_date): continue list_date str(int(list_date)) list_date_fmt f{list_date[:4]}-{list_date[4:6]}-{list_date[6:8]} end_fmt datetime.now().strftime(%Y-%m-%d) f.write(f{qlib_code}\t{list_date_fmt}\t{end_fmt}\n) def update_daily(): today datetime.now().strftime(%Y%m%d) today_fmt f{today[:4]}-{today[4:6]}-{today[6:8]} cal pro.trade_cal(exchangeSSE, start_datetoday, end_datetoday) if cal.empty or cal[is_open].iloc[0] 0: print(f{today} 不是交易日跳过) return stock_list pro.stock_basic( exchange, list_statusL, fieldsts_code ) features_dir os.path.join(CSV_ROOT, features) os.makedirs(features_dir, exist_okTrue) for idx, row in stock_list.iterrows(): ts_code row[ts_code] qlib_code ts_code_to_qlib(ts_code) file_path os.path.join(features_dir, f{qlib_code}.csv) df download_stock_daily(ts_code, today, today) if df is None or df.empty: continue if os.path.exists(file_path): old pd.read_csv(file_path, dtype{date: str}) last_date old[date].iloc[-1] if last_date today_fmt: continue merged pd.concat([old, df], ignore_indexTrue) else: merged df merged merged.drop_duplicates(subset[date]).sort_values(date) merged.to_csv(file_path, indexFalse) if idx % 100 0: print(f已处理 {idx} 只股票) time.sleep(0.2) # 追加日历 cal_file os.path.join(CSV_ROOT, calendars, day.txt) with open(cal_file, r) as f: existing_days set(line.strip() for line in f) if today_fmt not in existing_days: with open(cal_file, a) as f: f.write(today_fmt \n) if __name__ __main__: parser argparse.ArgumentParser() parser.add_argument(--mode, choices[init, update], defaultupdate) args parser.parse_args() if args.mode init: init_all() else: update_daily()用法# 第一次全量初始化 python qlib_tushare_updater.py --mode init # 之后每天跑一次 python qlib_tushare_updater.py --mode update # 更新完 CSV 后执行二进制转储 python scripts/dump_bin.py dump_update \ --csv_path csv_data \ --qlib_dir qlib_data \ --include_fields open,high,low,close,volume5.2 定时任务怎么配Linux 上我用的是 crontab0 18 * * 1-5 cd /home/user/qlib_project python qlib_tushare_updater.py --mode update logs/update.log 21意思是每周一到周五下午 6 点执行一次周末不跑。为什么定在 18 点A 股 15 点收盘tushare 官方一般 16 点到 17 点之间会完成当天数据入库18 点拉取基本不会碰到数据缺失。Windows 上可以把同样命令放到“任务计划程序”里每天收盘后触发一次。关键点是日志一定要保留。哪天策略结果不对第一件事就是翻日志看前一天数据到底更新成功没有。5.3 更新完怎么自查数据更新完不能直接跑模型先花两分钟验证一下。我会做三件事第一检查 CSV 末尾日期是否等于最新交易日tail -1 csv_data/features/SH600000.csv第二用 QLIB 读取最后几行特征确认.bin同步成功import qlib from qlib.data import D qlib.init(provider_uriqlib_data) df D.features( [SH600000], [$close, $volume], start_time2025-01-01, end_time2025-02-01, freqday ) print(df.tail())第三顺手对比 tushare 最新交易日的某只股票收盘价和 QLIB 读出来的收盘价两边必须一致。这一步能同时抓出复权问题和更新遗漏。用这套流程跑了半年我最大的体会有两条一是“数据管道稳定”比“数据管道炫酷”重要得多裸编码、日志、重试机制这些看似土味的代码才是保证每天安心下班的关键二是复权方式选定之后不要随便改哪怕你觉得“前复权更能反映真实价格”一旦切换整个历史特征都会变策略结论全部要重做。QLIB 和 tushare 这套组合只要把更新机制理顺就能把精力真正放到因子和模型上而不是跟数据较劲。 QLIB数据更新实战：用tushare构建A股每日行情自动管道
QLIB 这套平台我用了快两年一直觉得它很像那种“主业很强但生活自理能力堪忧”的同学回测效率高因子库全机器学习流水线也是现成的但唯独数据这件事它默认你得自己搞定。我去年年底搭本地A股研究环境时最折腾的竟然不是模型而是“每天收盘后怎么把最新行情更新进QLIB”。当时在社区里搜了一圈要么是讲QLIB怎么跑官方数据集的要么是单独讲tushare怎么拉数据的很少有人把这两者怎么接起来、每天怎么维持更新讲透。这篇文章就我目前在用的“QLIB tushare每日行情更新”方案完整复盘一遍内容包括数据格式设计、复权方式选择、全量初始化和增量更新的完整代码以及我踩过的一堆坑。目标是让你拿到手就能搭出一条自动更新的数据管道而不是看完还是一头雾水。1. 先把“更新每日行情”这件事拆开看1.1 QLIB 不是“带数据的回测框架”它只负责计算QLIB 是微软开源的一个量化投资平台核心优势是把特征工程、模型训练、回测、实盘交易这整条链路打通。但它有个很鲜明的特点数据默认由使用者自己准备。官方仓库里的示例数据主要是为了让新用户跑通流程而你一旦接入自己的行情源就必须面对它的数据格式约定。QLIB 的数据存储分三层calendars/day.txt交易日历一行一个日期格式是YYYY-MM-DDinstruments/all.txt股票池列表记录每只代码的上市日期、退市日期features/instrument.bin每只股票的二进制特征矩阵列对应字段行对应日历中的交易日。这种设计在读取时非常高效因为模型训练阶段会大量随机访问历史数据二进制文件比 CSV 少一次解析开销。代价就是所有外部数据都需要先切成规范格式再用官方脚本转换成.bin。一开始我也嫌麻烦后来对这个结构反而有了好感一旦数据按 QLIB 的约定存好后面跑 Alpha158、做滚动训练、切换因子集都不用再操心数据层的问题。反过来如果数据更新没做好哪怕因子代码写得再对出来的结果也是脏的。所以“更新每日行情”这件事本质上不是“把今天的K线下载下来”就完了而是要维护好三个文件体系日历要追加、股票池要更新、特征矩阵要补上最新一行。这三个动作缺一个第二天跑策略时都会出事。下面我会按这个思路一步步展开。1.2 为什么数据管道选 tushare 而不是别的方式市面上能拿A股日线的数据源不少baostock、akshare、tushare我都试过。baostock免费且稳定但字段偏少复权因子要自己处理akshare接口多但上游经常改页面结构维护成本偏高tushare虽然需要积分但pro_bar这个接口对日线行情、复权、复权因子的支持是最省心的。我选择 tushare 的核心原因有三点接口稳定字段规范。日线级别的 OHLCV 在pro_bar里都是现成的字段名和 QLIB 的约定差距很小只有vol和volume这种小差异。复权因子可以直接拿。tushare 算好了adj_factor不管你用前复权还是后复权都不用自己推算除权除息。交易日历完整。做增量更新时最怕的就是“今天是不是交易日”判断错trade_cal接口直接告诉你哪天开市哪天休市。你可能觉得“直接去财经网站下载历史数据再转CSV”也行但那种方式的问题是无法自动化每天收盘后手动下载、手动命名、手动合并坚持一周就会崩溃。tushare 的接口方式虽然要写点代码但一次性搭好之后每天跑脚本就好这是长期研究最舒服的状态。2. 动手前必须想清楚的三件事2.1 数据源接口怎么选pro_bar、daily、trade_cal 分工不同tushare 的接口很多但做每日行情更新时真正高频用到的就那么几个。我列一张表把用途和注意事项写清楚接口作用注意事项ts.pro_bar获取日线行情支持复权核心接口返回ts_code、trade_date、open、high、low、close、vol、amount等pro.daily获取未复权日线pro_bar底层就是它一般不需要直接调pro.trade_cal获取交易日历判停方法is_open1才是交易日pro.stock_basic获取股票基础信息用来生成 QLIB 的instruments/all.txt注意区分上市和退市状态pro.adj_factor获取复权因子如果你要自己算复权价用原始价乘以因子即可ts.pro_bar是最省事的入口。它内部会帮你把除权除息数据算好常见的调用方式是这样的import tushare as ts pro ts.pro_api(你的token) df ts.pro_bar( ts_code600000.SH, adjhfq, start_date20240101, end_date20241231 )adj参数有三档不填就是未复权qfq表示前复权hfq表示后复权。这块怎么选我放到 2.3 节专门讲因为它直接影响历史数据的稳定性。2.2 QLIB 对输入文件的目录和列名要求QLIB 官方提供了一个数据转换脚本dump_bin.py它负责把 CSV 转成二进制。但脚本对 CSV 的目录结构是有明确要求的。你在准备数据之前得先搭好这个目录csv_data/ ├── calendars/ │ └── day.txt # 交易日列表每行一个日期形如 2024-01-02 ├── instruments/ │ └── all.txt # 股票列表代码 上市日期 退市日期 └── features/ ├── SH600000.csv # 每只股票一个 CSV文件名是 QLIB 风格的代码 ├── SZ000001.csv └── ...这里有个容易踩坑的地方dump_bin.py在dump_all模式下会去features目录下找所有以.csv结尾的文件然后根据文件名确定股票代码。文件名必须是SH600000这种格式也就是“交易所前缀 6位数字”tushare 返回的是600000.SH这种格式需要转换。CSV 内部格式也有要求。第一列必须是日期列名建议叫date格式为%Y-%m-%d后面的列是要入库的特征字段。比如date,open,high,low,close,volume 2024-01-02,10.5,10.8,10.4,10.6,100000 2024-01-03,10.6,10.9,10.5,10.8,120000dump_bin.py转出来的.bin文件会按calendars/day.txt的日历对齐。也就是说如果某只股票 CSV 里缺少某个交易日的记录那它在.bin里对应的那一行就是 NaN。2.3 tushare 如何计算复权QLIB 归档时应该用哪种复权这个问题是新手最容易翻车的环节。先说 tushare 那边是怎么处理的。tushare 的复权因子adj_factor是一个随交易日变化的因子序列。如果你拿到原始价和复权因子后复权价的计算公式是这样的后复权价 原始价 * 对应日期的复权因子前复权价则是在后复权价的基础上再除以最新一期的复权因子前复权价 原始价 * 复权因子 / 最新复权因子你可以用代码验证一下ts.pro_bar就算你不传adj返回的也是未复权价传了hfq返回的就是原始价乘以因子之后的结果。这两种方式最终数值是一致的区别只是 tushare 帮你算好了省去你手动 merge 因子表。那问题来了QLIB 入库时应该用哪种我个人强烈建议用后复权hfq不要用前复权qfq。原因是前复权有一个天生缺陷每当有新数据进来最新复权因子可能变化然后整段历史价格都会被重算。这意味着你今天训练好的模型特征明天可能因为数据更新悄悄变了实验就不可复现了。后复权则没有这个问题只要除权除息事件不发生改变历史价格一旦入库就固定下来。唯一的缺点是后复权价和当前市价差得很远但这个缺点对 QLIB 几乎没影响因子计算绝大多数都在用收益率、均线、波动率这些对单调缩放不敏感的量绝对价格水平本来也不重要。我见过有人把未复权价格直接入库然后用复权因子在 QLIB 里做动态调整这样更精确但实现难度高对多数场景没必要。后复权是从“实现简单、结果稳定、误差可接受”三个维度综合下来的最优解。3. 实战完整跑通“拉取 - 清洗 - 入库”流程3.1 首次全量初始化把历史行情完整拉下来初始化阶段的目标是把某个股票池从起始日期到当前日期的全部日线数据拉下来整理成 QLIB 要的 CSV 目录结构。先说股票池。一般研究环境主要关心当前上市状态的股票用stock_basic接口可以拿到pro ts.pro_api(你的token) # 获取当前上市状态的股票列表 stock_list pro.stock_basic( exchange, # 空表示沪深两市 list_statusL, # L上市D退市P暂停上市 fieldsts_code,symbol,name,area,industry,list_date ) print(stock_list.shape)list_statusL拿到的就是当前正常上市的股票。注意如果要做长期回测还要考虑历史上有过上市但后来退市的股票否则回测到 2018 年时会漏掉一堆当时还活着的股票。实际使用中我建议全量初始化时用list_statusL,D,P把所有状态都拉进来然后在instruments/all.txt里用上市日期和退市日期去约束每只股票的有效区间。接着写一个拉取日线的函数def ts_code_to_qlib(ts_code): 将 tushare 的 600000.SH 转成 QLIB 的 SH600000 code, exchange ts_code.split(.) return (SH if exchange SH else SZ) code def download_stock_daily(ts_code, start_date, end_date): 拉取单只股票的后复权日线并整理成 QLIB CSV 格式 df ts.pro_bar( ts_codets_code, adjhfq, start_datestart_date, end_dateend_date ) if df is None or df.empty: return None df df.rename(columns{vol: volume}) df[date] pd.to_datetime( df[trade_date], format%Y%m%d ).dt.strftime(%Y-%m-%d) df df[[date, open, high, low, close, volume]] df df.sort_values(date) return df这里有几个细节需要说明pro_bar的ts_code参数要传600000.SH这种格式不能传600000adjhfq是后复权trade_date返回的是20240102这种 8 位字符串要转成2024-01-02因为 QLIB 的日历格式要求带横线vol要重命名为volumeQLIB 默认字段名里没有vol这种缩写。历史区间怎么选如果你做日频因子研究建议从 2007 年或 2010 年开始太早的数据一方面复权因子可能缺失另一方面对当前市场的参考价值有限。我自己常用20050101作为起点因为 tushare 对这个时间之后的数据覆盖最完整。把所有股票循环一遍写成 CSVimport os import time output_dir csv_data/features os.makedirs(output_dir, exist_okTrue) for idx, row in stock_list.iterrows(): ts_code row[ts_code] qlib_code ts_code_to_qlib(ts_code) file_path os.path.join(output_dir, f{qlib_code}.csv) if os.path.exists(file_path): continue df download_stock_daily(ts_code, 20050101, 20241231) if df is not None: df.to_csv(file_path, indexFalse) # 控制访问频率避免触发接口限流 time.sleep(0.2)这个循环看起来简单但真跑起来你会发现问题几千只股票一只一只拉总共要几分钟到十几分钟中间只要限流一次就断掉。所以完整脚本里我会加一个“断点续传”的逻辑也就是if os.path.exists(file_path): continue。这个逻辑在初始化中断后重新执行时已经拉过的股票就会被跳过节省大量时间。3.2 用 dump_bin 把 CSV 转成 QLIB 的二进制格式CSV 准备完毕之后接下来要生成 QLIB 数据目录。先创建日历文件和 instruments 文件。日历文件csv_data/calendars/day.txt的内容是所有交易日按从早到晚排序一行一个日期。这个文件可以从 tushare 的trade_cal取得cal pro.trade_cal( exchangeSSE, # 上交所沪深通常保持一致 start_date20050101, end_date20241231, is_open1 # 只取交易日 ) cal cal.sort_values(cal_date) with open(csv_data/calendars/day.txt, w) as f: for d in cal[cal_date]: f.write(f{d[:4]}-{d[4:6]}-{d[6:8]}\n)instruments 文件csv_data/instruments/all.txt的格式是每行两列或三列用 Tab 分隔。第一列是 QLIB 格式代码第二列是上市日期第三列是退市日期没有退市的可以写一个未来日期或者写你数据末尾的日期。QLIB 在使用时会读取这个文件决定每只股票什么时候进入、什么时候退出。with open(csv_data/instruments/all.txt, w) as f: for idx, row in stock_list.iterrows(): ts_code row[ts_code] qlib_code ts_code_to_qlib(ts_code) list_date row[list_date] # 转成 YYYY-MM-DD 格式退市日期用数据末尾日期代替 list_date_fmt f{list_date[:4]}-{list_date[4:6]}-{list_date[6:8]} f.write(f{qlib_code}\t{list_date_fmt}\t2024-12-31\n)之后调用官方脚本dump_bin.py。这个脚本在 QLIB 仓库的scripts/目录下直接下载就有python scripts/dump_bin.py dump_all \ --csv_path csv_data \ --qlib_dir qlib_data \ --include_fields open,high,low,close,volume参数含义dump_all全量转储模式会扫描csv_path下面所有 CSV逐个生成.bin--qlib_dir输出目录QLIB 之后就用这个目录作为provider_uri--include_fields指定要保留的字段多个字段用逗号分隔顺序可以自己定但建议按照open,high,low,close,volume排列。跑完之后qlib_data目录下会出现和csv_data一样的结构只不过features目录下的文件从.csv变成了.bin。你可以随机挑一只股票用 QLIB 读一下验证import qlib from qlib.data import D qlib.init(provider_uriqlib_data) data D.features( [SH600000], [$open, $high, $low, $close, $volume], start_time2024-01-01, end_time2024-12-31, freqday ) print(data.tail())如果输出正常说明数据入库成功。3.3 每日增量更新脚本的核心写法全量初始化只需要跑一次之后每天要做的是增量更新。增量更新的核心思路是只处理最新的交易日数据把新行追加到 CSV 末尾再跑dump_update更新.bin。先判断今天是不是交易日from datetime import datetime today datetime.now().strftime(%Y%m%d) cal pro.trade_cal(exchangeSSE, start_datetoday, end_datetoday) if cal.empty or cal[is_open].iloc[0] 0: print(f{today} 不是交易日跳过更新) exit(0)然后拉取所有股票当天的数据追加到每只股票对应的 CSV 里stock_list pro.stock_basic(exchange, list_statusL, fieldsts_code) for idx, row in stock_list.iterrows(): ts_code row[ts_code] qlib_code ts_code_to_qlib(ts_code) file_path os.path.join(output_dir, f{qlib_code}.csv) # 拉当日数据 df download_stock_daily(ts_code, today, today) if df is None or df.empty: # 停牌或当日无数据跳过 continue # 读取历史 CSV old pd.read_csv(file_path, dtype{date: str}) # 避免重复追加同一行 if str(old[date].iloc[-1]) today_fmt: continue merged pd.concat([old, df], ignore_indexTrue) merged merged.drop_duplicates(subset[date]).sort_values(date) merged.to_csv(file_path, indexFalse)这里有一个很重要的细节追加前必须检查 CSV 最后一行的日期。如果上次更新已经成功而这次脚本因为某种原因重复执行会导致同一行数据被写两次。drop_duplicates可以兜底但更保险的做法是直接对比日期大于等于当前交易日就不追加。增量更新完 CSV 后还要更新calendars/day.txt。如果今天是交易日就把今天的日期追加进去如果还没有的话cal_file csv_data/calendars/day.txt with open(cal_file, r) as f: existing_days set(line.strip() for line in f) today_fmt f{today[:4]}-{today[4:6]}-{today[6:8]} if today_fmt not in existing_days: with open(cal_file, a) as f: f.write(today_fmt \n)最后跑dump_updatepython scripts/dump_bin.py dump_update \ --csv_path csv_data \ --qlib_dir qlib_data \ --include_fields open,high,low,close,volumedump_update和dump_all的区别在于它只会对已经存在的.bin文件追加新日期数据不会重新生成全部文件。这样每次更新只需要处理有新增记录的股票速度快很多。如果你中途改了字段配置或者怀疑数据有错可以偶尔强制跑一次dump_all全量重建。4. 每日更新阶段躲不开的四个经典问题4.1 日期类型不一致导致日历对不上这个问题我撞上过不止一次。QLIB 的day.txt里日期格式是2024-01-02字符串类型。而你在 CSV 里如果用了datetime类型或者用了20240102这种整型dump_bin在比对日期时就会出错轻则报 Warning重则直接生成错位的日历。解决方式很粗暴所有日期在写入 CSV 和day.txt之前统一转成字符串并调用strftime(%Y-%m-%d)不要混用类型。在读取已有 CSV 时也要指定dtype{date: str}避免 pandas 自动推断成别的类型。4.2 停牌股票让 dump_update 直接报错增量更新时最容易遇到的运行时错误就是股票停牌。比如某只股票今天停牌tushare 返回空数据那这只股票的 CSV 末尾就不会追加新行。到这里为止没问题问题出在dump_update的日期对齐上如果day.txt已经加入了今天但某只股票的.bin文件没有今天的记录QLIB 就会认为这只股票在今天的特征全是 NaN。如果只是个别股票这个行为可以接受模型训练时一般会自动 dropna。但如果大量股票停牌比如春节前最后一天dump_update可能因为“CSV 末尾日期与日历文件中最后一个交易日不一致”而直接跳过某些文件或报错。我现在的处理策略是更新day.txt时只把“实际有股票交易的交易日”追加进去而不是简单地把当天塞进去。更简单的方法是用上一个交易日做判断如果今天不是交易日干脆不追加如果今天是交易日但某些股票停牌就让它们在.bin里表现为 NaN不要试图去补一条假数据。4.3 接口限流让全量拉取中途卡死tushare 的积分制度决定了它有访问频率限制。积分越低每分钟能请求的次数越少。你在全量初始化时要对几千只股票循环调pro_bar一旦限流接口会返回一段错误提示而不是异常退出如果你没有对返回结果做校验就会把None写进 CSV最后导致dump_bin读取失败。我的做法是写一个带重试的包装函数def download_stock_daily_with_retry(ts_code, start_date, end_date, retries3): for attempt in range(retries): try: df ts.pro_bar( ts_codets_code, adjhfq, start_datestart_date, end_dateend_date ) if df is not None and not df.empty: return df time.sleep(0.5) except Exception as e: print(f{ts_code} 第 {attempt1} 次请求失败: {e}) time.sleep(1) return None再加上一个经验值pro_bar之间至少间隔 0.2 秒积分不够的间隔拉到 0.5 秒。宁可慢一点也不要中断后从头再来。4.4 字段名不一致特征矩阵默默缺列还有一个隐蔽问题字段名不一致。tushare 的日线接口返回vol但 QLIB 的特征字段如果叫volume那dump_bin会把vol当做一个独立字段存进去。结果就是你读特征时用$volume拿不到数据因为.bin里存的列名是$vol。这个问题不会报错只会让某个特征列全部是 NaN而且模型的日志里不一定能立刻发现。我的建议是在写入 CSV 之前统一把 tushare 字段映射成 QLIB 字段vol - volume、amount - amount保持原名在dump_bin的--include_fields里明确写清楚最终字段名入库后马上读一次验证用上节提到的D.features检查列名和数值是否正常。5. 一套可以直接抄作业的完整更新脚本5.1 脚本整体逻辑和工作模式为了让你少走弯路我把上面的代码整合成一个完整脚本支持两种模式init和update。init模式做全量初始化update模式做每日增量。import os import time import argparse import pandas as pd import tushare as ts from datetime import datetime TOKEN 你的tushare_token CSV_ROOT csv_data QLIB_ROOT qlib_data START_DATE 20050101 ts.set_token(TOKEN) pro ts.pro_api() def ts_code_to_qlib(ts_code): code, exchange ts_code.split(.) return (SH if exchange SH else SZ) code def download_stock_daily(ts_code, start_date, end_date, retries3): for attempt in range(retries): try: df ts.pro_bar( ts_codets_code, adjhfq, start_datestart_date, end_dateend_date ) if df is not None and not df.empty: df df.rename(columns{vol: volume}) df[date] pd.to_datetime( df[trade_date], format%Y%m%d ).dt.strftime(%Y-%m-%d) df df[[date, open, high, low, close, volume]] return df.sort_values(date) time.sleep(0.5) except Exception as e: print(f{ts_code} 请求失败: {e}) time.sleep(1) return None def init_all(): os.makedirs(os.path.join(CSV_ROOT, features), exist_okTrue) stock_list pro.stock_basic( exchange, list_statusL,D,P, fieldsts_code,symbol,name,area,industry,list_date ) # 全量拉取日线 for idx, row in stock_list.iterrows(): ts_code row[ts_code] qlib_code ts_code_to_qlib(ts_code) file_path os.path.join(CSV_ROOT, features, f{qlib_code}.csv) if os.path.exists(file_path): continue df download_stock_daily(ts_code, START_DATE, datetime.now().strftime(%Y%m%d)) if df is not None: df.to_csv(file_path, indexFalse) print(f[{idx}/{len(stock_list)}] {qlib_code} 完成) time.sleep(0.2) # 写日历文件 cal pro.trade_cal( exchangeSSE, start_dateSTART_DATE, end_datedatetime.now().strftime(%Y%m%d), is_open1 ).sort_values(cal_date) with open(os.path.join(CSV_ROOT, calendars, day.txt), w) as f: for d in cal[cal_date]: f.write(f{d[:4]}-{d[4:6]}-{d[6:8]}\n) # 写 instruments 文件 with open(os.path.join(CSV_ROOT, instruments, all.txt), w) as f: for idx, row in stock_list.iterrows(): ts_code row[ts_code] qlib_code ts_code_to_qlib(ts_code) list_date row[list_date] if pd.isna(list_date): continue list_date str(int(list_date)) list_date_fmt f{list_date[:4]}-{list_date[4:6]}-{list_date[6:8]} end_fmt datetime.now().strftime(%Y-%m-%d) f.write(f{qlib_code}\t{list_date_fmt}\t{end_fmt}\n) def update_daily(): today datetime.now().strftime(%Y%m%d) today_fmt f{today[:4]}-{today[4:6]}-{today[6:8]} cal pro.trade_cal(exchangeSSE, start_datetoday, end_datetoday) if cal.empty or cal[is_open].iloc[0] 0: print(f{today} 不是交易日跳过) return stock_list pro.stock_basic( exchange, list_statusL, fieldsts_code ) features_dir os.path.join(CSV_ROOT, features) os.makedirs(features_dir, exist_okTrue) for idx, row in stock_list.iterrows(): ts_code row[ts_code] qlib_code ts_code_to_qlib(ts_code) file_path os.path.join(features_dir, f{qlib_code}.csv) df download_stock_daily(ts_code, today, today) if df is None or df.empty: continue if os.path.exists(file_path): old pd.read_csv(file_path, dtype{date: str}) last_date old[date].iloc[-1] if last_date today_fmt: continue merged pd.concat([old, df], ignore_indexTrue) else: merged df merged merged.drop_duplicates(subset[date]).sort_values(date) merged.to_csv(file_path, indexFalse) if idx % 100 0: print(f已处理 {idx} 只股票) time.sleep(0.2) # 追加日历 cal_file os.path.join(CSV_ROOT, calendars, day.txt) with open(cal_file, r) as f: existing_days set(line.strip() for line in f) if today_fmt not in existing_days: with open(cal_file, a) as f: f.write(today_fmt \n) if __name__ __main__: parser argparse.ArgumentParser() parser.add_argument(--mode, choices[init, update], defaultupdate) args parser.parse_args() if args.mode init: init_all() else: update_daily()用法# 第一次全量初始化 python qlib_tushare_updater.py --mode init # 之后每天跑一次 python qlib_tushare_updater.py --mode update # 更新完 CSV 后执行二进制转储 python scripts/dump_bin.py dump_update \ --csv_path csv_data \ --qlib_dir qlib_data \ --include_fields open,high,low,close,volume5.2 定时任务怎么配Linux 上我用的是 crontab0 18 * * 1-5 cd /home/user/qlib_project python qlib_tushare_updater.py --mode update logs/update.log 21意思是每周一到周五下午 6 点执行一次周末不跑。为什么定在 18 点A 股 15 点收盘tushare 官方一般 16 点到 17 点之间会完成当天数据入库18 点拉取基本不会碰到数据缺失。Windows 上可以把同样命令放到“任务计划程序”里每天收盘后触发一次。关键点是日志一定要保留。哪天策略结果不对第一件事就是翻日志看前一天数据到底更新成功没有。5.3 更新完怎么自查数据更新完不能直接跑模型先花两分钟验证一下。我会做三件事第一检查 CSV 末尾日期是否等于最新交易日tail -1 csv_data/features/SH600000.csv第二用 QLIB 读取最后几行特征确认.bin同步成功import qlib from qlib.data import D qlib.init(provider_uriqlib_data) df D.features( [SH600000], [$close, $volume], start_time2025-01-01, end_time2025-02-01, freqday ) print(df.tail())第三顺手对比 tushare 最新交易日的某只股票收盘价和 QLIB 读出来的收盘价两边必须一致。这一步能同时抓出复权问题和更新遗漏。用这套流程跑了半年我最大的体会有两条一是“数据管道稳定”比“数据管道炫酷”重要得多裸编码、日志、重试机制这些看似土味的代码才是保证每天安心下班的关键二是复权方式选定之后不要随便改哪怕你觉得“前复权更能反映真实价格”一旦切换整个历史特征都会变策略结论全部要重做。QLIB 和 tushare 这套组合只要把更新机制理顺就能把精力真正放到因子和模型上而不是跟数据较劲。