发布于VeighNa社区公众号【vnpy-community】
 
原文作者：VeighNa小助手 | 发布时间：2025-06-15
 
VeighNa 4.0.0 版本新增了面向 AI 量化策略的 vnpy.alpha 模块，为专业量化交易员提供从多因子机器学习（ML）策略研发到实盘交易的一站式解决方案。

在股票多因子策略中，精准的成分股数据至关重要。在 A 股市场，成分股数据不仅能明确投资范围、控制样本特征，还可通过追踪历史成分变动，避免回测中的幸存者偏差。成分股通常流动性较好，可降低实盘执行风险；同时，以指数作为业绩基准，可更客观地评估策略表现。需要注意的是，本文方法主要适用于股票市场，期货等其他资产类别在数据需求和处理流程上有所不同。

本文将演示如何借助迅投研数据服务与 VeighNa，完整下载 A 股指数成分股数据，并实现成分变动跟踪与历史行情获取。

 

环境准备与模块导入

 

from datetime import datetime                           # 日期时间处理
from tqdm import tqdm                                   # 进度条显示
from xtquant import xtdata                              # 迅投研数据API

# vnpy相关模块
from vnpy.trader.database import DB_TZ                  # 数据库时区
from vnpy.trader.datafeed import get_datafeed           # 数据服务接口
from vnpy.trader.constant import Exchange, Interval     # 交易所和时间周期常量
from vnpy.trader.object import HistoryRequest           # 历史数据请求对象
from vnpy.alpha import AlphaLab, logger                 # Alpha研究实验室和日志

上述模块涵盖了数据下载、处理与存储的核心功能。其中，xtquant 是迅投研提供的 Python API，而 vnpy.alpha 则是 VeighNa 专为因子策略投研推出的组件。

 

下载参数配置

 

task_name = "csi300"                # 任务名称，用于标识数据文件夹
index_symbol = "000300.SSE"         # vnpy格式的指数代码
xt_index_symbol = "000300.SH"       # 迅投研格式的指数代码

start_date = "20070101"             # 开始日期
end_date = "20231231"               # 结束日期

intervals = [
    Interval.DAILY,                 # 设置下载日线数据
]

示例中选取沪深 300 指数（csi300）作为研究对象，时间范围为 2007 年初至 2023 年底，数据周期为日线。可以按需扩展至其他指数（如中证 500、创业板指等）或其他周期（如分钟线）。

 

初始化研究环境

 

# 创建投研实验室
lab = AlphaLab(f"./lab/{task_name}")    # 指定数据文件夹

# 初始化数据服务（迅投研数据服务）
datafeed = get_datafeed()               # 获取配置的数据服务
datafeed.init()                         # 初始化数据服务连接

AlphaLab 是 VeighNa 的因子投研流程管理组件，集成了数据管理、模型训练、信号生成和策略回测等完整环节。此处创建其实例，并指定数据存储目录。

 

下载成分股变动数据

 

# 下载历史成分股信息
xtdata.download_sector_data()                                # 下载板块分类数据
xtdata.download_history_data("", "stocklistchange", "", "")  # 下载股票列表变更历史

在获取指数板块信息前，需要先执行 download_sector_data 下载板块分类；而获取板块成分股列表前，同样需通过 download_history_data 下载股票列表变更历史。

首次下载耗时较长，请耐心等待。如需查看进度，可在 xtquant 自动创建的 data/log 目录中查看日志，其中会出现 "download done [xxx/xxx]" 等提示。

 

获取指数成分股变动

 

# 查询交易日历
days = xtdata.get_trading_calendar(market="SZ", start_time=start_date, end_time=end_date)

# 轮询获取指数成分股
index_components = {}
end_datetime = datetime.strptime(end_date, "%Y%m%d")
for i in days:
    dt = datetime.strptime(i, "%Y%m%d")
    if dt > end_datetime:
        continue

    # 获取特定日期的指数成分股
    xt_symbols = xtdata.get_stock_list_in_sector(xt_index_symbol, i)

    # 将迅投研格式转换为vnpy格式
    vt_symbols: list = []
    for xt_symbol in xt_symbols:
        vt_symbol = xt_symbol.replace("SH", "SSE").replace("SZ", "SZSE")
        vt_symbols.append(vt_symbol)

    # 以日期为键，成分股列表为值，存储在字典中
    index_components[dt.strftime("%Y-%m-%d")] = vt_symbols

# 保存到数据中心
lab.save_component_data(index_symbol, index_components)

下载完毕后，先使用 get_trading_calendar 获取交易日历，再对每个交易日调用 get_stock_list_in_sector 取得当天的成分股列表，并将代码从迅投研格式（如 000001.SZ）转换为 VeighNa 格式（如 000001.SZSE）。

最终得到的字典以日期为键、成分股列表为值，并通过 save_component_data 存入 AlphaLab 数据中心，为后续因子计算与策略回测奠定基础。

 

加载成分股列表

 

# 加载指数成分股代码
component_symbols = lab.load_component_symbols(index_symbol, start_date, end_date)

调用 load_component_symbols 即可从数据中心加载成分股数据。该方法会在指定日期区间内自动去重，返回所有历史成分股的列表，便于后续行情数据下载。

 

下载历史行情数据

 

# 加载指数成分股代码（去重后的所有历史成分股）
component_symbols = lab.load_component_symbols(index_symbol, start_date, end_date)

# 转换时间格式
start = datetime.strptime(start_date, "%Y%m%d")
start.replace(tzinfo=DB_TZ)  # 设置时区

end = datetime.strptime(end_date, "%Y%m%d")
end.replace(tzinfo=DB_TZ)    # 设置时区

# 除了成分股，还要下载指数数据
task_symbols = component_symbols + [index_symbol]

# 遍历下载数据，使用tqdm显示进度
for vt_symbol in tqdm(task_symbols):
    symbol, exchange_str = vt_symbol.split(".")  # 分离代码和交易所

    for interval in intervals:
        # 创建历史数据请求
        req = HistoryRequest(
            symbol=symbol, 
            exchange=Exchange(exchange_str), 
            start=start, 
            end=end, 
            interval=interval
        )

        # 查询历史数据
        bars = datafeed.query_bar_history(req)

        # 保存数据到数据库
        if bars:
            lab.save_bar_data(bars)
        else:
            logger(interval, vt_symbol)  # 记录下载失败的情况

拿到成分股列表后，需要进一步下载它们的K线历史行情数据。上述代码将去重后的成分股与指数本身合并为下载任务，并使用 tqdm 展示进度。遍历过程中，通过 query_bar_history 拉取历史行情并存储于 AlphaLab 数据库；若查询失败，则记录日志以便后续排查。

 

配置回测参数

 

# 添加回测参数配置
for vt_symbol in component_symbols:
    lab.add_contract_setting(
        vt_symbol,
        long_rate=5/10000,      # 多头费率：万分之五
        short_rate=10/10000,    # 空头费率：万分之十
        size=1,                 # 合约乘数：1
        pricetick=0.0001,       # 价格变动：0.0001
    )

为每只成分股配置回测参数。交易成本的合理设置对回测结果的准确性至关重要。本示例在 A 股市场中采用万分之五的多头费率和万分之十的空头费率，综合考虑了佣金、印花税及滑点。

 

完整代码

 

在文章的结尾还是老规矩附上完整的程序代码：

# 加载模块
from datetime import datetime

from tqdm import tqdm
from xtquant import xtdata

from vnpy.trader.database import DB_TZ
from vnpy.trader.datafeed import get_datafeed
from vnpy.trader.constant import Exchange, Interval
from vnpy.trader.object import HistoryRequest

from vnpy.alpha import AlphaLab, logger


# 设置下载参数
task_name = "csi300"
index_symbol = "000300.SSE"
xt_index_symbol = "000300.SH"

start_date = "20070101"
end_date = "20231231"

intervals = [
    Interval.DAILY,
]

# 创建投研实验室
lab = AlphaLab(f"./lab/{task_name}")    # 指定数据文件夹

# 初始化数据服务（这里配置使用的迅投研）
datafeed = get_datafeed()
datafeed.init()

# 下载历史成分股信息
xtdata.download_sector_data()

xtdata.download_history_data("", "stocklistchange", "", "")

# 查询交易日历
days = xtdata.get_trading_calendar(market="SZ", start_time=start_date, end_time=end_date)

# 轮询获取指数成本股
index_components = {}
end_datetime = datetime.strptime(end_date, "%Y%m%d")
for i in days:
    dt = datetime.strptime(i, "%Y%m%d")
    if dt > end_datetime:
        continue

    xt_symbols = xtdata.get_stock_list_in_sector(xt_index_symbol, i)

    vt_symbols: list = []
    for xt_symbol in xt_symbols:
        vt_symbol = xt_symbol.replace("SH", "SSE").replace("SZ", "SZSE")
        vt_symbols.append(vt_symbol)

    index_components[dt.strftime("%Y-%m-%d")] = vt_symbols

# 保存到数据中心
lab.save_component_data(index_symbol, index_components)

# 加载指数成分股代码
component_symbols = lab.load_component_symbols(index_symbol, start_date, end_date)

# 转换时间格式
start = datetime.strptime(start_date, "%Y%m%d")
start.replace(tzinfo=DB_TZ)

end = datetime.strptime(end_date, "%Y%m%d")
end.replace(tzinfo=DB_TZ)

# 除了成分股，还要下载指数数据
task_symbols = component_symbols + [index_symbol]

# 遍历下载数据
for vt_symbol in tqdm(task_symbols):
    symbol, exchange_str = vt_symbol.split(".")

    for interval in intervals:
        req = HistoryRequest(symbol, Exchange(exchange_str), start, end, interval)
        bars = datafeed.query_bar_history(req)

        if bars:
            lab.save_bar_data(bars)
        else:
            logger(interval, vt_symbol)

# 添加回测参数配置
for vt_symbol in component_symbols:
    lab.add_contract_setting(
        vt_symbol,
        long_rate=5/10000,
        short_rate=10/10000,
        size=1,
        pricetick=0.0001,
    )