Quant_Code/1.交易策略/3.期权策略/AdvancedSpreadStrategy_dualthrust2.py

from vnpy.trader.constant import Interval
from vnpy.trader.utility import ArrayManager, BarGenerator, load_json, save_json
from vnpy.trader.object import TickData, BarData

from elite_optionstrategy import (
    StrategyTemplate,
    Variable,
    Parameter,
    PortfolioData,
    ChainData,
    OptionData,
    OptionBarGenerator,
)

from datetime import time

class AdvancedSpreadStrategy(StrategyTemplate):
    """基于DualThrust信号做空符合价差的策略"""

    author: str = "用Python的交易员"

    option_portfolio: str = Parameter("MO")  # 期权产品代码
    underlying_symbol: str = Parameter("IMJQ00.CFFEX")  # 标的合约代码
    k1: float = Parameter(0.3)  # DualThrust参数k1
    k2: float = Parameter(0.9)  # DualThrust参数k2
    risk_level: int = Parameter(40)  # 开仓风险度
    percent_add: float = Parameter(0.02)  # 委托超价比例
    otm_level: int = Parameter(0)  # 做空期权档位
    leg1_ratio: int = Parameter(4)  # 顺势腿的比例
    leg2_ratio: int = Parameter(1)  # 逆势腿的比例

    dt_signal: int = Variable(0)  # 当前信号多空
    trading_size: int = Variable(1)  # 当前交易数量
    atm_strike: float = Variable(0)  # 当前平值行权价

    def on_init(self) -> None:
        """策略初始化"""
        self.write_log("策略初始化")

        # K线截面合成器
        self.obg: OptionBarGenerator = OptionBarGenerator(self.on_bars)

        # 订阅行情
        self.subscribe_options(self.option_portfolio)
        self.subscribe_data(self.underlying_symbol)

        # 标的信号对象
        self.factor: DualThrustFactor = DualThrustFactor(self.underlying_symbol, self.k1, self.k2)

        # 加载标的历史数据初始化
        bars: list[BarData] = self.load_bars(self.underlying_symbol, 40, Interval.MINUTE)
        for bar in bars:
            self.factor.update_bar(bar)

        # 缓存文件名称
        self.data_filename: str = f"{self.name}_data.json"

    def on_start(self) -> None:
        """策略启动"""
        self.write_log("策略启动")

        data: dict = load_json(self.data_filename)
        self.dt_signal = data.get("dt_signal", 0)

    def on_stop(self) -> None:
        """策略停止"""
        self.write_log("策略停止")

        data: dict = {"dt_signal": self.dt_signal}
        save_json(self.data_filename, data)

    def on_tick(self, tick: TickData) -> None:
        """Tick推送"""
        self.obg.update_tick(tick)

    def on_bars(self, bars: dict[str, BarData]) -> None:
        """K线推送"""
        # 回测首先计算标的信号
        underlying_bar: BarData = bars.pop(self.underlying_symbol, None)
        if underlying_bar:
            self.factor.update_bar(underlying_bar)

        # 获取期权组合对象
        portfolio: PortfolioData = self.get_portfolio(self.option_portfolio)

        # 更新最新期权价格到组合
        price_data: dict[str, float] = {}
        for bar in bars.values():
            price_data[bar.vt_symbol] = bar.close_price

        portfolio.update_price(price_data)

        # 获取当月期权链
        front_chain: ChainData = portfolio.get_chain_by_level(0)
        if not front_chain:
            self.write_log("无法获取当月期权链，请检查是否正确添加了期权合约")
            return

        # 计算平值期权
        front_chain.calculate_atm()
        self.atm_strike = front_chain.atm_strike

        # 获取当前DualThrust多空信号
        dt_signal: int = self.factor.get_signal()

        # 如果DualThrust多头信号，且尚未做多
        if dt_signal > 0 and self.dt_signal <= 0:
            # 做空Put
            call: OptionData = front_chain.get_option_by_level(cp=1, level=self.otm_level)
            put: OptionData = front_chain.get_option_by_level(cp=-1, level=self.otm_level)

            if call and put:
                # 清空之前的目标
                self.clear_targets()

                atr_value: float = self.factor.get_atr()
                if atr_value:
                    # self.trading_size = int(self.risk_level / atr_value)
                    self.trading_size = 1 # max(self.trading_size, 1)
                else:
                    self.trading_size = 1

                self.set_target(put.vt_symbol, -self.trading_size * self.leg1_ratio)
                self.set_target(call.vt_symbol, -self.trading_size * self.leg2_ratio)
        # 如果DualThrust空头信号，且尚未做空
        elif dt_signal < 0 and self.dt_signal >= 0:
            # 做空Call
            call: OptionData = front_chain.get_option_by_level(cp=1, level=self.otm_level)
            put: OptionData = front_chain.get_option_by_level(cp=-1, level=self.otm_level)

            if call and put:
                # 清空之前的目标
                self.clear_targets()

                atr_value: float = self.factor.get_atr()
                if atr_value:
                    # self.trading_size = int(self.risk_level / atr_value)
                    self.trading_size = 1 #max(self.trading_size, 1)
                else:
                    self.trading_size = 1

                self.set_target(call.vt_symbol, -self.trading_size * self.leg1_ratio)
                self.set_target(put.vt_symbol, -self.trading_size * self.leg2_ratio)

        # 缓存DualThrust多空信号
        self.dt_signal = dt_signal

        # 执行具体的委托交易
        self.execute_trading(price_data, self.percent_add)

        # 推送UI事件更新
        self.put_event()


class DualThrustFactor:
    """标的物DualThrust因子（基于DualThrust输出多空信号）"""

    def __init__(self, vt_symbol: str, k1: float, k2: float) -> None:
        """构造函数"""
        self.vt_symbol: str = vt_symbol
        self.k1: float = k1
        self.k2: float = k2

        self.bg: BarGenerator = BarGenerator(self.update_bar, 30, self.update_window_bar)
        self.am: ArrayManager = ArrayManager(10)

        self.day_open = 0
        self.day_high = 0
        self.day_low = 0
        self.day_range = 0
        self.long_entry = 0
        self.short_entry = 0
        self.exit_time = time(hour=14, minute=55)

        self.long_entered = False
        self.short_entered = False

        self.signal: int = 0
        self.bars: list[BarData] = []  # 手动维护最近两根K线

    def update_tick(self, tick: TickData) -> None:
        """Tick更新"""
        self.bg.update_tick(tick)

    def update_bar(self, bar: BarData) -> None:
        """K线更新"""
        self.bg.update_bar(bar)

    def update_window_bar(self, bar: BarData) -> None:
        """K线更新"""
        self.am.update_bar(bar)
        if not self.am.inited:
            return

        # 维护最近两根K线
        self.bars.append(bar)
        if len(self.bars) > 2:
            self.bars.pop(0)

        # 确保至少有两根K线用于日期比较
        if len(self.bars) < 2:
            return

        last_bar = self.bars[-2]
        # bar = self.bars[-1]

        # 检查是否新的一天
        if last_bar.datetime.date() != bar.datetime.date():
            if self.day_high:
                self.day_range = self.day_high - self.day_low
                self.long_entry = bar.open_price + self.k1 * self.day_range
                self.short_entry = bar.open_price - self.k2 * self.day_range

            self.day_open = bar.open_price
            self.day_high = bar.high_price
            self.day_low = bar.low_price

            self.long_entered = False
            self.short_entered = False
        else:
            self.day_high = max(self.day_high, bar.high_price)
            self.day_low = min(self.day_low, bar.low_price)

        if not self.day_range:
            return

        # 生成信号逻辑
        if bar.datetime.time() < self.exit_time:
            if bar.close_price > self.day_open:
                self.signal = 1
            else:
                self.signal = -1
        else:
            self.signal = 0

    def get_signal(self) -> int:
        """获取当前多空信号"""
        return self.signal

    def get_atr(self) -> float:
        """获取ATR风险度"""
        if self.am.inited:
            return self.am.atr(14)
        else:
            return 0