期权策略投研中的难点（已探讨） 搭建本地化期权数据库（本文主题） 期权策略的开发和回测 实盘期权策略交易运维
构建并持续维护一个全面的期权历史合约信息数据库，确保每个交易日的合约信息准确无误；
为每个期权品种准备包含所有历史合约（包括当前和过往合约）的K线数据库，以便进行完整的历史数据回放。
更新迅投研本地缓存中的历史合约信息表；
更新VeighNa Elite数据库中的期权合约数据； 更新VeighNa Elite数据库中的期权K线数据。
from multiprocessing import Processfrom datetime import datetimefrom vnpy.trader.database import BarOverviewfrom vnpy.trader.datafeed import get_datafeedfrom vnpy.trader.object import ContractData, BarData, HistoryRequestfrom vnpy.trader.constant import Exchange, Product, OptionType, Intervalfrom vnpy.trader.setting import SETTINGSfrom elite_database import EliteDatabase# 配置迅投研数据服务SETTINGS["datafeed.name"] = "xt"SETTINGS["datafeed.username"] = "token"SETTINGS["datafeed.password"] = ""
# 交易所映射关系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}
def update_history_data() -> None:"""更新历史合约信息"""# 在子进程中加载xtquantfrom xtquant.xtdata import download_history_data# 初始化数据服务datafeed = get_datafeed()datafeed.init()# 下载历史合约信息download_history_data("", "historycontract")print("xtquant历史合约信息下载完成")
def update_contract_data(sector_name: str) -> None:"""更新合约数据"""# 在子进程中加载xtquantfrom 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.CALLelif type_str == "P":option_type = OptionType.PUToption_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线（已到期的不会再发生变化） 从本地已有数据的末端开始向后更新（避免重复下载已存在的部分）
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:continuevt_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].datetimeend_dt: datetime = bars[-1].datetimemsg: str = f"{contract.vt_symbol}数据更新成功，{start_dt} - {end_dt}"print(msg)
if __name__ == "__main__":# 使用子进程更新历史合约信息process: Process = Process(target=update_history_data)process.start()process.join() # 等待子进程执行完成# 更新合约信息update_contract_data("中金所")update_contract_data("过期中金所")# 更新历史数据update_bar_data()
