vnpy+Tushare Pro：A股历史日线数据导入与后复权回测数据搭建

做量化交易，第一步从来不是写策略，而是把数据这事彻底搞定。我自己入坑这个方向折腾最多的，恰恰不是策略代码，而是怎么把A股历史行情变成一份干净、可靠、能反复使用的本地数据。这篇文章是用vnpy加Tushare Pro这套组合，把沪深股票历史日线数据完整导入vnpy数据库的实战方案，包含可直接复制的完整Python代码。整个流程跑完，本地库里会有一份从指定年份至今、做了后复权处理的标准日线数据，可以直接对接vnpy的CTA回测、组合回测甚至后续实盘模块。不管你是刚接触量化的小白，还是已经写了点策略但一直被数据源折磨的开发者，这套方案都能直接落地。

1. 选型思考：为什么是vnpy加Tushare Pro，而不是其他方案

1.1 数据源对比：Tushare、AkShare、Baostock怎么选

国内能免费或低成本拿到A股历史行情的渠道，主流就这几个：Tushare Pro、AkShare、Baostock。这三个我都用过，各有脾气。

Baostock很良心，接口稳定且完全免费，但数据字段偏老，文档更新也慢。AkShare胜在零门槛，但因为它从东财、新浪等多个网站抓数据，偶尔会出现某天数据缺失、字段单位变化的情况，做长期回测时这点很要命。

Tushare Pro则是另一种风格：接口是标准化的，字段说明文档完整，特别是复权因子有独立接口，这对做量化回测非常关键。它的门槛在积分体系，注册并完善个人信息后能拿到基础积分，大部分行情接口都能覆盖。如果只是拉日线、周线、复权因子这类数据，基础积分足够用。要拉分钟线、资金流等高阶数据才需要更高积分。总的判断是：如果你需要一个能长期稳定维护的研究环境，Tushare Pro最合适；如果只是临时调研一下，用AkShare更方便。我目前的主力数据源就是Tushare Pro，所以下面整套流程都围绕它展开。

1.2 为什么非要套一个vnpy，直接pandas算不行吗

很多人一开始会想：我已经能用Tushare把数据拉下来了，存成CSV或者DataFrame不就行了吗？为什么要引入vnpy这个看似笨重的框架？

我的理解是：单独的CSV文件在策略研究早期够用，但很快就变成噩梦。当你有了几十只股票、跨越多年日线、还要做复权处理、增量更新、统一字段单位时，CSV文件在文件系统里乱成一团，每写一个策略都要先写一段"找数据、读数据、清洗数据"的胶水代码，这些代码还会因为数据格式不统一而反复出bug。

vnpy的价值在于它自带一个数据层抽象。不管底层是SQLite还是MySQL，对外都统一成一套接口：get_database()拿到数据库操作对象，save_bar_data()写入K线，load_bar_data()读取K线。数据在vnpy里被规范成BarData对象，字段统一，回测引擎可以直接消费。我只需要写一次"从Tushare拉数据并转换成BarData"的导入脚本，之后所有策略都从vnpy数据库里读数据，格式永远一致。这就是"数据层"的意义：数据集成一次，以后所有模块复用。

2. 环境准备的三件套：Python版本、vnpy数据库、Tushare凭证

2.1 vnpy安装的版本陷阱

vnpy对Python版本比较挑剔。我实测下来，vnpy 3.x在Python 3.10和3.11上最稳，新版本对3.12的兼容也越来越好，但如果你还在用Python 3.7或3.8，建议先升级。反正量化研究迟早要吃Python新版本的红利，别在解释器版本上卡自己。

安装命令很简单：

pip install vnpy pip install tushare

国内网络环境下，建议加上国内镜像源，不然某些依赖包下载速度非常感人：

pip install vnpy -i https://pypi.tuna.tsinghua.edu.cn/simple pip install tushare -i https://pypi.tuna.tsinghua.edu.cn/simple

有一个高频安装坑：vnpy依赖TA-Lib这个技术指标库，而TA-Lib在Windows上通过pip直接安装经常失败。如果遇到报错，不要硬刚，去PyPI或者对应Python版本的whl文件仓库下载匹配的安装包本地安装。很多刚入门的人在这里被劝退，其实绕过这个坑后面就顺畅了。

2.2 数据库选型：本地开发从SQLite开始就够了

vnpy 3.x默认使用SQLite存储数据，这对个人量化研究来说是非常合理的选择。SQLite是文件型数据库，零配置、不用单独装服务端，数据就存在一个文件里。你不需要懂数据库运维，不需要记忆账号密码，写完脚本直接跑。

如果你此前用过vnpy并配置了MySQL之类的数据库，get_database()会去连你配置的那个库。第一次跑导入脚本时，建议先确认数据库配置。vnpy的配置一般放在用户目录下的.vntrader/vt_setting.json里，默认就是SQLite。如果你不确定，可以直接删掉这个文件让vnpy重新生成默认配置，或者把database.name显式设为sqlite。

等你的数据量到了几十GB、需要多进程并发读写时，再考虑迁移到MySQL或PostgreSQL，通过安装vnpy_mysql这类插件就能接上。但起步阶段，SQLite绝对够用了。全A股五千多只股票从2010年至今的日线后复权数据，SQLite文件也就几个GB，个人电脑完全扛得住。

2.3 Tushare的Token、积分与限流

Tushare Pro需要注册账号，然后在个人主页的接口Token页面复制你的token串。这个token就是访问凭证，代码里会用到。

积分体系是很多人一开始搞不清楚的地方。简单说：不同的接口对积分有门槛要求，注册并完善资料后能拿到基础积分，daily日线接口、adj_factor复权因子接口这些常用接口的基础积分通常够用。如果调用某个接口时提示权限不足或积分不够，去对应接口的文档页看一下所需积分数值，按官方指引做任务或者开通就行。

限流也是一定要正视的。Tushare对每分钟请求次数有限制，不同积分对应不同频率。导入全A股数据时，一只股票要调用两次接口（一次行情、一次复权因子），五千多只股票就是上万次请求，如果触发限流会被临时拒掉。所以脚本里必须加time.sleep()控制节奏，这个后面代码里会体现。

3. 主流程拆解：从Tushare原始行情到vnpy标准BarData

3.1 核心套路：股票列表、行情、复权因子三个接口怎么配合

整个导入流程可以拆成三个动作：

第一步，用stock_basic接口拿到当前所有A股股票的基础信息。这个接口返回每只股票的ts_code（比如600000.SH）、股票名称、上市日期、所属行业等。量化研究通常只需要上市状态为L（上市）的股票，退市的不碰。

第二步，对每只股票，用daily接口拉取日线行情。这个接口返回开高低收、成交量、成交额、前收盘、涨跌幅等字段。注意两个单位细节：vol成交量单位是手，amount成交额单位是千元，写入vnpy时要换算。

第三步，用adj_factor接口拉取复权因子。这个接口返回每个交易日的复权因子数值，是后续计算前复权或后复权价格的关键。Tushare本身不直接返回复权价，只返回因子，需要我们自己在代码里算。

把行情和因子按ts_code加trade_date合并到一起，这张表就是我们构造BarData的原材料。

3.2 完整导入脚本（可直接复制）

下面这段代码我按vnpy 3.x的接口习惯写的，可以直接保存成Python文件运行。我加足了注释，重点说明哪些地方藏了坑。

# -*- coding: utf-8 -*- """ Tushare Pro -> vnpy 数据库 历史日线导入脚本 首次全量导入A股沪深日线（后复权）示例 """ import time from datetime import datetime import pandas as pd import tushare as ts from vnpy.trader.constant import Exchange, Interval from vnpy.trader.database import get_database from vnpy.trader.object import BarData # 你的Tushare token，注意不要泄露 TOKEN = "你的Token" # 日线数据从哪天开始拉，建议设早一点，保证后有足够的因子区间 START_DATE = "20100101" END_DATE = "20250110" # 每次请求之间休息的秒数，根据你的积分限流调整 SLEEP_SECONDS = 0.5 ts.set_token(TOKEN) pro = ts.pro_api() database = get_database() def to_vnpy_symbol(ts_code: str): """把Tushare的ts_code拆成vnpy的symbol和exchange字段""" code, market = ts_code.split(".") if market == "SH": return code, Exchange.SSE if market == "SZ": return code, Exchange.SZSE # 北交所的股票在老版本vnpy里可能没有对应枚举，先跳过 return None, None def fetch_daily_with_adj(ts_code: str, start: str, end: str, adj_start: str): """ 拉取单只股票的日线行情和复权因子，合并后返回。 adj_start是复权因子的起点，建议用上市日期，保证后复权基准稳定。 """ daily = pro.daily(ts_code=ts_code, start_date=start, end_date=end) adj = pro.adj_factor(ts_code=ts_code, start_date=adj_start, end_date=end) if daily is None or daily.empty or adj is None or adj.empty: return pd.DataFrame(), 1.0 # 记录整个复权因子序列的第一个因子，作为后复权的基准 base_factor = adj["adj_factor"].iloc[0] # 把行情和因子按代码+日期合并 df = pd.merge(daily, adj, on=["ts_code", "trade_date"]) df["trade_date"] = df["trade_date"].astype(str) df = df.sort_values("trade_date").reset_index(drop=True) return df, base_factor def to_bar_list(df: pd.DataFrame, symbol: str, exchange: Exchange, base_factor: float) -> list: """把Tushare的DataFrame逐行转换成vnpy的BarData对象""" bars = [] for _, row in df.iterrows(): dt = datetime.strptime(row["trade_date"], "%Y%m%d") factor = row["adj_factor"] # 后复权价：原始价格乘以因子再除以基准因子 open_price = round(float(row["open"]) * factor / base_factor, 4) high_price = round(float(row["high"]) * factor / base_factor, 4) low_price = round(float(row["low"]) * factor / base_factor, 4) close_price = round(float(row["close"]) * factor / base_factor, 4) bar = BarData( symbol=symbol, exchange=exchange, datetime=dt, interval=Interval.DAILY, open_price=open_price, high_price=high_price, low_price=low_price, close_price=close_price, volume=float(row["vol"]) * 100, # 手转股 turnover=float(row["amount"]) * 1000, # 千元转元 gateway_name="Tushare", ) # vnpy 3.x的BarData支持extra字段，用来存复权因子 bar.extra["adj_factor"] = float(factor) bars.append(bar) return bars def main(): # 1. 获取当前所有A股列表 stock_df = pro.stock_basic( exchange="", list_status="L", fields="ts_code,symbol,name,industry,list_date", ) total = len(stock_df) success = 0 for i, row in stock_df.iterrows(): ts_code = row["ts_code"] symbol, exchange = to_vnpy_symbol(ts_code) if symbol is None: continue # 复权因子从上市日期开始拉，保证后复权基准固定 list_date = str(row["list_date"]).replace("-", "")[:8] start = max(START_DATE, list_date) try: # 删除旧数据，避免重复导入导致数据翻倍 database.delete_bar_data(symbol, exchange, Interval.DAILY) df, base_factor = fetch_daily_with_adj(ts_code, start, END_DATE, list_date) if df.empty: print(f"[{i + 1}/{total}] {ts_code} 无行情数据") continue bars = to_bar_list(df, symbol, exchange, base_factor) database.save_bar_data(bars) success += 1 print(f"[{i + 1}/{total}] {ts_code} 导入 {len(bars)} 根K线") time.sleep(SLEEP_SECONDS) except Exception as e: print(f"[{i + 1}/{total}] {ts_code} 导入异常：{e}") time.sleep(1) print(f"全量导入结束：成功 {success} 只，未导入 {total - success} 只") if __name__ == "__main__": main()

3.3 代码里藏的几个细节

这段代码不是"能跑就行"的水平，里面几个细节是实际踩坑踩出来的。

第一，后复权基准因子用的是复权因子序列的第一个值，而不是行情数据的第一行。很多教程直接用df["adj_factor"].iloc[0]，这有个隐患：如果复权因子没从上市首日拉全，以后扩展数据区间时基准变了，同一段历史价格就全变了。我把adj_start设为上市日期，就能保证后复权价格在任意时间导入都稳定。

第二，入库前先delete_bar_data。这个操作避免了脚本重复运行时数据翻倍。代价是如果某只股票在删除后导入失败，这只股票暂时会没有数据。但对本地开发来说可以接受，重新跑一遍就行。

第三，限流问题。代码里留了SLEEP_SECONDS参数。全量导入五千多只股票时，每只要打两次接口，让出节奏稳一点，第一次全量导入建议放在晚上跑，睡一觉起来基本就导完了。如果你积分较低、限流更紧，把sleep调大到1秒甚至更多。

第四，vol和amount的单位。Tushare返回的成交量单位是手，成交额单位是千元，而vnpy的BarData里volume和turnover一般习惯用股和元。如果不转换，以后写策略按手数计算时很容易差出100倍，这个bug极其隐蔽。代码里统一转了，不用再管。

4. 复权这事必须单独说：回测失真往往从这里开始

4.1 不复权数据为什么不能直接用

如果你拿Tushare的daily接口原始数据直接回测，遇到股票分红除权的那一天，价格会出现一个看似暴跌的跳空缺口。比如一只股票除权除息日开盘价直接从20元跳到18元，这不是真的暴跌，是分红导致的除权行为。

问题在于，量化策略根本不关心这件事。策略以为跌了10%，触发加仓信号，实际上可能只是因为除权。长期回测下来，这种失真会不断干扰信号计算结果，尤其是对价格敏感的均线类、动量类策略，结果完全失真。

所以量化回测必须使用复权价格。复权的本质就是把历史价格按除权因子调整，让价格序列在除权除息日保持连续，从而真实反映资产的收益变化。

4.2 前复权与后复权怎么选

Tushare的复权因子是个累积系数，用它换算价格的公式是：

• 后复权价 = 原始价 × 当日因子 / 基准日因子
• 前复权价 = 原始价 × 当日因子 / 最新因子

这里的基准不同，直接决定了两种复权的性质。

我个人的选择是：回测一律用后复权。原因在于前复权以最新价格为基准，每次数据更新后，整个历史序列都会被重新调整，今天跑出来的信号和下周跑出来的可能完全不一样，长期研究时很难复现结果。后复权以最早期为基准，一旦定好就稳定了，回测结果可复现，也方便做长周期统计。

有个极端情况需要注意：前复权在长周期高分红股票上，早期价格会被压到接近0甚至理论上的负值。虽然这种情况比较少见，但一旦碰到，策略里的对数收益率计算就直接崩了。后复权不会出现这个问题。

4.3 vnpy里复权数据的落地方式

vnpy本身没有内置复权计算逻辑，它默认数据库里存的就是已经处理好的价格。所以我们要在导入阶段把复权做完，写入数据库的BarData价格就是复权价。

我在第3章的代码里做的是后复权。如果你更习惯用前复权，只需要把to_bar_list里的基准因子改成df["adj_factor"].iloc[-1]（最新因子），其他逻辑不用动。但要做好心理准备：前复权价格每过一段时间就可能整体变化，你的策略信号会跟着历史数据漂移。

这里还想提醒一句：如果你把复权价格写进了数据库，抽样验证的时候千万不要拿不复权的原始行情直接对比。当时的股价20元是真实价格，但你库里存的后复权价格可能是40元或80元，这并不代表数据错了，只是基准不同。对比时要先按因子换算，或者只看价格的变化趋势是否一致。

5. 跑完脚本只是开始：增量更新和数据体检

5.1 增量更新：收盘后只需要拉一段增量

全量导入完成后，你还需要每天或每周把新数据追加进库里，这个动作叫增量更新。增量更新的核心是：先查出库里这只股票最后一根K线的日期，从下一天开始拉就行，没必要把整个历史再拉一遍。

下面这段代码是基于第3章函数的一个增量更新示例，核心思想是用load_bar_data加载最近400天数据，找到最后日期，然后接着拉：

from datetime import datetime, timedelta def update_symbol(ts_code: str): symbol, exchange = to_vnpy_symbol(ts_code) if symbol is None: return # 载入最近400天，找到库里最后一根K线的日期 recent_bars = database.load_bar_data( symbol, exchange, Interval.DAILY, start=datetime.now() - timedelta(days=400), end=datetime.now(), ) if recent_bars: last_dt = recent_bars[-1].datetime start = (last_dt + timedelta(days=1)).strftime("%Y%m%d") else: # 库里没数据，回到全量导入流程 start = "20100101" df, base_factor = fetch_daily_with_adj(ts_code, start, END_DATE, list_date) if df.empty: print(f"{ts_code} 没有新增数据") return new_bars = to_bar_list(df, symbol, exchange, base_factor) database.save_bar_data(new_bars) print(f"{ts_code} 增量更新 {len(new_bars)} 根K线")

为什么用400天而不是一年？因为A股一年大概有250个交易日，但自然日跨度接近365天，停牌久的股票可能一年没几根K线。用400个自然日能基本保证覆盖最近的交易日，同时数据量不算大，行情接口也不会被频繁调用。

注意我没修改base_factor的逻辑：每次更新时复权因子仍然从上市日期拉，保证基准固定。这一点非常重要，否则增量更新后历史价格会整体漂移。

5.2 数据体检的土办法

数据导入后一定要做验证，别直接拿去跑策略。我每次导入完会用下面这几个土办法检查：

第一步，抽查个股。随便挑一只股票，比如平安银行000001.SZ，从库里读取最近一年的日线，核对最后日期是不是最新的交易日，K线数量是不是合理。

def inspect_stock(ts_code: str, last_days: int = 250): symbol, exchange = to_vnpy_symbol(ts_code) if symbol is None: return start = datetime.now() - timedelta(days=last_days) bars = database.load_bar_data( symbol, exchange, Interval.DAILY, start=start, end=datetime.now() ) print(f"{ts_code}: 近{last_days}天共有{len(bars)}根K线") if bars: print("最新bar:", bars[-1].datetime, "收盘:", bars[-1].close_price) print("最早bar:", bars[0].datetime, "收盘:", bars[0].close_price)

第二步，看除权日前后是否平滑。选一只近期除过权的股票，查看它除权日前后几天的后复权价格序列。如果价格在除权日附近出现异常跳变，说明复权逻辑可能有问题。

第三步，用异常过滤条件粗扫数据。在写入BarData前，检查是否出现high < max(open, close)或low > min(open, close)这种逻辑错误。Tushare原始数据一般没问题，但复权后的浮点计算偶尔会产生极小的精度异常，提前拦掉总比回测时暴雷好。

# 在to_bar_list之前对DataFrame做一次检查 invalid = df[ (df["high"] < df[["open", "close"]].max(axis=1)) | (df["low"] > df[["open", "close"]].min(axis=1)) ] if not invalid.empty: print(f"发现OHLC逻辑异常，共{len(invalid)}行")

第四步，关注全市场股票数量。Tushare的stock_basic返回当前上市股票数，A股近几年在五千只上下。导入完成后可以简单统计库里有多少个symbol，如果数量跟stock_basic差异过大，说明有批量导入失败，回查日志。

5.3 我自己的数据维护习惯

最后分享一个实操习惯。我每次全量导入后，都会把导入日志保留下来，不只是屏幕打印，而是把每只股票的symbol、导入K线数量、最后一根K线日期写到一个文本文件里。过几天增量更新时，我会抽查其中几只，确认最后日期更新到了最新交易日，数量在合理范围。

量化这行有个老话：数据错误是最隐蔽的敌人。它不会让你的程序报错，只会让你的回测曲线莫名其妙地崩塌。把数据导入和更新这套流程固定下来，后面所有策略研究的底气就不一样了。等你回头发现某次策略回测结果异常时，第一个念头不是怀疑策略写错了，而是先去体检数据，这个排查顺序能帮你省掉大量无效时间。