VeighNa量化社区
你的开源社区量化交易平台 | vn.py | vnpy
Member
avatar
加入于:
帖子: 5379
声望: 327

发布于VeighNa社区公众号【vnpy-community】
 
原文作者: 陈晓优 | 发布时间:2024-05-31

 

确认数据需求

 

本文是《聊聊期权量化》系列的第二篇,让我们先回顾一下本系列计划探讨的主题:

  1. 期权策略投研中的难点(已探讨)
  2. 搭建本地化期权数据库(本文主题)
  3. 期权策略的开发和回测
  4. 实盘期权策略交易运维

在上一篇文章中,我们探讨了期权策略回测过程中潜在的主要难点,并强调了数据管理的重要性。具体而言,我们需要:

  • 构建并持续维护一个全面的期权历史合约信息数据库,确保每个交易日的合约信息准确无误;
  • 为每个期权品种准备包含所有历史合约(包括当前和过往合约)的K线数据库,

  • 以便进行完整的历史数据回放。

 

迅投研数据准备

 

在明确了需求之后,我们决定采用【迅投研】数据服务来构建我们的本地期权投研数据库。对于迅投研的基本使用方法,可以参考我们公众号之前发布的《基于迅投研的量化数据自运维下载更新》一文。

整个期权数据的下载流程可以简化为以下三个关键步骤:

  1. 更新迅投研本地缓存中的历史合约信息表;
  2. 更新VeighNa Elite数据库中的期权合约数据;
  3. 更新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元/年的价格覆盖股票、期权、期货等多种数据,在同类产品中性价比还是颇具竞争力。

你对于期权回测数据的准备策略的开发有什么疑问,或者希望在后续文章中看到的内容?欢迎在评论区留言告诉我们!!!

 

免责声明

文章中的信息或观点仅供参考,作者不对其准确性或完整性做出任何保证。读者应以其独立判断做出投资决策,作者不对因使用本报告的内容而引致的损失承担任何责任。

 

Member
avatar
加入于:
帖子: 1
声望: 0

您好,运行代码示例代码的时候,报错【RuntimeError: 该模块必须在VeighNa Trader Elite版中使用】,

Traceback (most recent call last):
File "e:\VNPYSCRIPT\test.py", line 11, in <module>
from elite_database import EliteDatabase
File "E:\vnElite\lib\site-packages\elite_database__init__.py", line 8, in <module elite_database>
RuntimeError: 该模块必须在VeighNa Trader Elite版中使用

我用的是官网下的veighNA elite 仿真模拟版,是说这个版本不能用EliteDatabase吗,必须买正式版吗?

Member
avatar
加入于:
帖子: 5379
声望: 327

jupyter脚本也要在elite_lab仿真模拟上跑,不能用其他python环境

Member
avatar
加入于:
帖子: 1
声望: 0

为什么我运行上述程序之后并没有获取到历史数据,而是直接开始运行了后续未到期合约的获取?(第一次运行)
description

Member
avatar
加入于:
帖子: 5379
声望: 327

从你的输出来看是因为你没有获取到过期合约。可以打印一下看看get_stock_list_in_sector(“过期中金所”)是不是没查到合约

Member
avatar
加入于:
帖子: 7
声望: 0

请问extendinfo 报key error,如何处理?谢谢!

update_contract_data("上证期权")
option_strike=data["ExtendInfo"]["OptExercisePrice"],
KeyError: 'ExtendInfo'

Member
avatar
加入于:
帖子: 119
声望: 2

花钱如流血 wrote:

请问extendinfo 报key error,如何处理?谢谢!

update_contract_data("上证期权")
option_strike=data["ExtendInfo"]["OptExercisePrice"],
KeyError: 'ExtendInfo'
把这个删了["ExtendInfo"],迅投那边改了

© 2015-2022 上海韦纳软件科技有限公司
备案服务号:沪ICP备18006526号

沪公网安备 31011502017034号

【用户协议】
【隐私政策】
【免责条款】