发布于VeighNa社区公众号【vnpy-community】
原文作者： 陈晓优 | 发布时间：2024-05-31
确认数据需求
本文是《聊聊期权量化》系列的第二篇，让我们先回顾一下本系列计划探讨的主题：
- 期权策略投研中的难点（已探讨）
- 搭建本地化期权数据库（本文主题）
- 期权策略的开发和回测
- 实盘期权策略交易运维
在上一篇文章中，我们探讨了期权策略回测过程中潜在的主要难点，并强调了数据管理的重要性。具体而言，我们需要：
- 构建并持续维护一个全面的期权历史合约信息数据库，确保每个交易日的合约信息准确无误；
为每个期权品种准备包含所有历史合约（包括当前和过往合约）的K线数据库，
以便进行完整的历史数据回放。
迅投研数据准备
在明确了需求之后，我们决定采用【迅投研】数据服务来构建我们的本地期权投研数据库。对于迅投研的基本使用方法，可以参考我们公众号之前发布的《基于迅投研的量化数据自运维下载更新》一文。
整个期权数据的下载流程可以简化为以下三个关键步骤：
- 更新迅投研本地缓存中的历史合约信息表；
- 更新VeighNa Elite数据库中的期权合约数据；
- 更新VeighNa Elite数据库中的期权K线数据。
本文中所提供的代码，需要通过VeighNa Elite Lab交互式开发环境来运行。首先在全局配置SETTINGS中填入迅投研的账号token：
from multiprocessing import Process
from datetime import datetime
from vnpy.trader.database import BarOverview
from vnpy.trader.datafeed import get_datafeed
from vnpy.trader.object import ContractData, BarData, HistoryRequest
from vnpy.trader.constant import Exchange, Product, OptionType, Interval
from vnpy.trader.setting import SETTINGS
from elite_database import EliteDatabase
# 配置迅投研数据服务
SETTINGS["datafeed.name"] = "xt"
SETTINGS["datafeed.username"] = "token"
SETTINGS["datafeed.password"] = ""
由于迅投研和VeighNa在交易所命名规则上存在差异，因此我们需要对两边的交易所进行对应关系映射：
# 交易所映射关系
EXCHANGE_XT2VT = {
"SH": Exchange.SSE,
"SZ": Exchange.SZSE,
"BJ": Exchange.BSE,
"SF": Exchange.SHFE,
"IF": Exchange.CFFEX,
"INE": Exchange.INE,
"DF": Exchange.DCE,
"ZF": Exchange.CZCE,
"GF": Exchange.GFEX
}
迅投研的客户端库xtquant仅在每次启动时会从硬盘文件中读取历史合约信息表，且在后续的进程运行期间不会自动读取更新。为了确保历史合约信息表的更新能够即时反映，这里需要使用multiprocessing多进程库在子进程中执行update_history_data函数：
def update_history_data() -> None:
"""更新历史合约信息"""
# 在子进程中加载xtquant
from xtquant.xtdata import download_history_data
# 初始化数据服务
datafeed = get_datafeed()
datafeed.init()
# 下载历史合约信息
download_history_data("", "historycontract")
print("xtquant历史合约信息下载完成")
下一步将进行一个关键步骤——下载并更新期权合约数据，这一步骤对于后续的期权投研和回测至关重要。在处理期权合约对象ContractData时，特别要注意所有以option_为前缀的字段，确保这些字段信息准确无误且完整，避免任何错误或遗漏：
def update_contract_data(sector_name: str) -> None:
"""更新合约数据"""
# 在子进程中加载xtquant
from xtquant.xtdata import (
get_stock_list_in_sector,
get_instrument_detail
)
# 初始化数据服务
datafeed = get_datafeed()
datafeed.init()
# 查询中金所历史合约代码
vt_symbols: list[str] = get_stock_list_in_sector(sector_name)
# 遍历列表查询合约信息
contracts: list[ContractData] = []
for xt_symbol in vt_symbols:
# 拆分XT代码
symbol, xt_exchange = xt_symbol.split(".")
# 筛选期权合约合约
if "-" in symbol:
data: dict = get_instrument_detail(xt_symbol, True)
type_str = data["InstrumentID"].split("-")[1]
if type_str == "C":
option_type = OptionType.CALL
elif type_str == "P":
option_type = OptionType.PUT
option_underlying: str = data["InstrumentID"].split("-")[0]
contract: ContractData = ContractData(
symbol=data["InstrumentID"],
exchange=EXCHANGE_XT2VT[xt_exchange.replace("O", "")],
name=data["InstrumentName"],
product=Product.OPTION,
size=data["VolumeMultiple"],
pricetick=data["PriceTick"],
min_volume=data["MinLimitOrderVolume"],
option_strike=data["ExtendInfo"]["OptExercisePrice"],
option_listed=datetime.strptime(data["OpenDate"], "%Y%m%d"),
option_expiry=datetime.strptime(data["ExpireDate"], "%Y%m%d"),
option_underlying=option_underlying,
option_portfolio=data["ProductID"],
option_index=str(data["ExtendInfo"]["OptExercisePrice"]),
option_type=option_type,
gateway_name="XT"
)
contracts.append(contract)
# 保存合约信息到数据库
database: EliteDatabase = EliteDatabase()
database.save_contract_data(contracts)
print("合约信息更新成功", len(contracts))
鉴于期权K线数据总量巨大且涉及众多合约，加之数据服务每日提供的流量有限，因此除了初次下载采用全量模式外，后续更新应采用增量模式：
- 仅更新那些尚未到期的期权合约的K线（已到期的不会再发生变化）
- 从本地已有数据的末端开始向后更新（避免重复下载已存在的部分）
以下代码中，默认的查询开始时间参数start设置为2018-1-1，可以根据自己的实际需求进行调整（通常来说已能满足需求）：
def update_bar_data() -> None:
"""更新K线数据"""
# 初始化数据服务
datafeed = get_datafeed()
datafeed.init()
# 获取当前时间戳
now: datetime = datetime.now()
# 获取合约信息
database: EliteDatabase = EliteDatabase()
contracts: list[ContractData] = database.load_contract_data()
# 获取数据汇总
data: list[BarOverview] = database.get_bar_overview()
overviews: dict[str, BarOverview] = {}
for o in data:
# 只保留分钟线数据
if o.interval != Interval.MINUTE:
continue
vt_symbol: str = f"{o.symbol}.{o.exchange.value}"
overviews[vt_symbol] = o
# 遍历所有合约信息
for contract in contracts:
# 如果没有到期时间，则跳过
if not contract.option_expiry:
continue
# 查询数据汇总
overview: BarOverview = overviews.get(contract.vt_symbol, None)
# 如果已经到期，则跳过
if overview and contract.option_expiry < now:
continue
# 初始化查询开始的时间
start: datetime = datetime(2018, 1, 1)
# 实现增量查询
if overview:
start = overview.end
# 执行数据查询和更新入库
req: HistoryRequest = HistoryRequest(
symbol=contract.symbol,
exchange=contract.exchange,
start=start,
end=datetime.now(),
interval=Interval.MINUTE
)
bars: list[BarData] = datafeed.query_bar_history(req)
if bars:
database.save_bar_data(bars)
start_dt: datetime = bars[0].datetime
end_dt: datetime = bars[-1].datetime
msg: str = f"{contract.vt_symbol}数据更新成功，{start_dt} - {end_dt}"
print(msg)
最后，在脚本的主入口处（或在Jupyter Notebook中），按照既定顺序依次调用并执行上述所有函数：
if __name__ == "__main__":
# 使用子进程更新历史合约信息
process: Process = Process(target=update_history_data)
process.start()
process.join() # 等待子进程执行完成
# 更新合约信息
update_contract_data("中金所")
update_contract_data("过期中金所")
# 更新历史数据
update_bar_data()
需要注意的是，目前迅投研的14天试用权限已不再提供期权数据，需要购买正式权限后才能下载使用，不过其4978元/年的价格覆盖股票、期权、期货等多种数据，在同类产品中性价比还是颇具竞争力。
你对于期权回测数据的准备策略的开发有什么疑问，或者希望在后续文章中看到的内容？欢迎在评论区留言告诉我们！！！
免责声明
文章中的信息或观点仅供参考，作者不对其准确性或完整性做出任何保证。读者应以其独立判断做出投资决策，作者不对因使用本报告的内容而引致的损失承担任何责任。
