【vn.py】源码解析之 Dual Thrust 策略

Dual Thrust 策略

Dual Thrust 策略是 Michael Chalek 在80 年代开发的一种通道突破策略，曾经被Future Thruth杂志评为最赚钱的策略之一， Dual Thrust常常用于日内CTA策略或者日间CTA策略。到目前为止，Dual Thrust 仍然在自动化交易系统中排名第二左右。由于其对于多种品种的普适性，所以也被广泛应用于各种股票、期货、货币等市场。

Dual Thrust 策略在做日内策略时的思想很简单，以当日的开盘价为基准设定上下通道进行突破，上下通道是以当日开盘价加减一定比例的范围Range而确定，然后日内突破上轨平空开多，突破下轨平多开空。这个范围Range由前N日的四个价位来确定 Range = Max(HH-LC, HC-LL)，其中HH是前N日的High的最高价，LC是前N日的Close的最低价，HC是前N日Close的最高价，LL是前N日Low的最低价，在计算这个范围指标时有点类似平均真实波动指标（Average True Range，ATR），但是还是有很大区别的毕竟平均真实波动是 J. Welles Wilder 在1978提出的技术指标。

引入了多日的价格可以使得一定时期的范围Range相对稳定，当然也有直接以昨日的范围Range作为今日突破的基准。另外，Dual Thrust策略没有单独的止损，是一个趋势反转系统，也就说当出现反向信号时也就意味着平仓。

所以总结上面所说，上通道是 Open + K1 * Range；下通道是 Open + K2 * Range。需要设定的参数除了N之外，还有作为范围Range比例的系数K1和K2，通常这两个系数并不是一样的，这样也就导致了做多和做空的的触发条件是非对称的。

Dual Thrust 策略源码分析

由于前面已经写过关于CTA策略执行的流程和代码的架构，所以下面将主要围绕策略的参数与变量以及策略的主要执行逻辑进行分析。

1、完整源码

from datetime import time
from vnpy.app.cta_strategy import (
    CtaTemplate,
    StopOrder,
    TickData,
    BarData,
    TradeData,
    OrderData,
    BarGenerator,
    ArrayManager,
)


class DualThrustStrategy(CtaTemplate):
    """"""

    author = "用Python的交易员"

    fixed_size = 1
    k1 = 0.4
    k2 = 0.6

    bars = []

    day_open = 0
    day_high = 0
    day_low = 0

    range = 0
    long_entry = 0
    short_entry = 0
    exit_time = time(hour=14, minute=55)

    long_entered = False
    short_entered = False

    parameters = ["k1", "k2", "fixed_size"]
    variables = ["range", "long_entry", "short_entry", "exit_time"]

    def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
        """"""
        super(DualThrustStrategy, self).__init__(
            cta_engine, strategy_name, vt_symbol, setting
        )

        self.bg = BarGenerator(self.on_bar)
        self.am = ArrayManager()
        self.bars = []

    def on_init(self):
        """
        Callback when strategy is inited.
        """
        self.write_log("策略初始化")
        self.load_bar(10)

    def on_start(self):
        """
        Callback when strategy is started.
        """
        self.write_log("策略启动")

    def on_stop(self):
        """
        Callback when strategy is stopped.
        """
        self.write_log("策略停止")

    def on_tick(self, tick: TickData):
        """
        Callback of new tick data update.
        """
        self.bg.update_tick(tick)

    def on_bar(self, bar: BarData):
        """
        Callback of new bar data update.
        """
        self.cancel_all()

        self.bars.append(bar)
        if len(self.bars) <= 2:
            return
        else:
            self.bars.pop(0)
        last_bar = self.bars[-2]

        if last_bar.datetime.date() != bar.datetime.date():
            if self.day_high:
                self.range = self.day_high - self.day_low
                self.long_entry = bar.open_price + self.k1 * self.range
                self.short_entry = bar.open_price - self.k2 * self.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.range:
            return

        if bar.datetime.time() < self.exit_time:
            if self.pos == 0:
                if bar.close_price > self.day_open:
                    if not self.long_entered:
                        self.buy(self.long_entry, self.fixed_size, stop=True)
                else:
                    if not self.short_entered:
                        self.short(self.short_entry,
                                   self.fixed_size, stop=True)

            elif self.pos > 0:
                self.long_entered = True

                self.sell(self.short_entry, self.fixed_size, stop=True)

                if not self.short_entered:
                    self.short(self.short_entry, self.fixed_size, stop=True)

            elif self.pos < 0:
                self.short_entered = True

                self.cover(self.long_entry, self.fixed_size, stop=True)

                if not self.long_entered:
                    self.buy(self.long_entry, self.fixed_size, stop=True)

        else:
            if self.pos > 0:
                self.sell(bar.close_price * 0.99, abs(self.pos))
            elif self.pos < 0:
                self.cover(bar.close_price * 1.01, abs(self.pos))

        self.put_event()

    def on_order(self, order: OrderData):
        """
        Callback of new order data update.
        """
        pass

    def on_trade(self, trade: TradeData):
        """
        Callback of new trade data update.
        """
        self.put_event()

    def on_stop_order(self, stop_order: StopOrder):
        """
        Callback of stop order update.
        """
        pass

2、策略参数与变量

从下面的代码中可以看出超参有三个，范围Range比例的系数 K1、K2 以及仓位数 fixed_size。需要维护的变量有范围 range、上下通道 long_entry、short_entry。由于是日内交易，所以策略中还设定了一个每天的退场时间 exit_time 为14:55。另外还需要注意一下，就是vnpy中的Dual Thrust策略和上面介绍的还是有些区别的，主要在于vnpy中range是由昨日的最高价和最低价确定的，这个后面策略逻辑处就可以看出。总而言之，Dual Thrust 策略的参数还是较少的。

    fixed_size = 1
    k1 = 0.4
    k2 = 0.6

    bars = []

    day_open = 0
    day_high = 0
    day_low = 0

    range = 0
    long_entry = 0
    short_entry = 0
    exit_time = time(hour=14, minute=55)

    long_entered = False
    short_entered = False

    parameters = ["k1", "k2", "fixed_size"]
    variables = ["range", "long_entry", "short_entry", "exit_time"]

3、策略执行逻辑

同前面的Double MA一样，在对策略进行变量初始化时，也需要加载10天的1Min级别数据。然后主要的逻辑也是在on_bar()函数中，也就是说这个策略也是Bar驱动的。
首先，需要先清空所有未成交的委托单，然后记录下上一个bar和当前的bar。

        self.cancel_all()

        self.bars.append(bar)
        if len(self.bars) <= 2:
            return
        else:
            self.bars.pop(0)
        last_bar = self.bars[-2]

如果上一个bar的日期和当前bar的日期不同，则说明当前bar是今天的第一个bar，上一个bar是昨日的最后一个bar，然后通过昨日的最高价和最低价计算range；否则的话，就通过 max(self.day_high, bar.high_price)和min(self.day_low, bar.low_price) 记录每日的bar的最高价和最低价，从而确定range。然后，以K1和K2与range来设置上下通道long_entry和short_entry的大小。

        if last_bar.datetime.date() != bar.datetime.date():
            if self.day_high:
                self.range = self.day_high - self.day_low
                self.long_entry = bar.open_price + self.k1 * self.range
                self.short_entry = bar.open_price - self.k2 * self.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.range:
            return

在得到昨日的范围range之后就可以判断是否满足开仓条件了。如果当前bar的时间在退场时间之前就进行判断是否满足开平仓条件，否则的话就直接以限价单进行平今。

在退场时间之前判断是否满足开平仓条件时：
1、如果不持仓：
并且当前bar的close大于当日开盘价并且没有多头入场，就以上轨道long_entry的价位开一笔停止单。
并且当前bar的close小于当日开盘价并且没有空头入场，就以下轨道short_entry的价位开一笔停止单。
2、如果持有多头，就以下轨道short_entry的价位开一笔停止单来平仓，并且如果没有空头入场的话再以下轨道short_entry的价位开一笔停止单来开仓；如果持有空头，就以上轨道long_entry的价位开一笔停止单来平仓，并且如果没有多头入场的话再以上轨道long_entry的价位开一笔停止单来开仓。

注意上面开出的委托单都是以停止单的形式开出的，停止单也就意味着是需要在bar的内部进行交易撮合，在bar内部价格突破时开仓。vnpy中的Dual Thrust策略有些类似于反复试探的思想，每个bar内部进行撮合交易，发出订单，没有撮合成功则在新的bar接收时撤掉之前发出的订单。毕竟是日内策略，如果还是以1-min的bar基础上进行交易未免交易机会也确实会有些少，而通过停止单的形式则正好可以满足其日内突破的要求。

        if bar.datetime.time() < self.exit_time:
            if self.pos == 0:
                if bar.close_price > self.day_open:
                    if not self.long_entered:
                        self.buy(self.long_entry, self.fixed_size, stop=True)
                else:
                    if not self.short_entered:
                        self.short(self.short_entry,
                                   self.fixed_size, stop=True)
            elif self.pos > 0:
                self.long_entered = True
                self.sell(self.short_entry, self.fixed_size, stop=True)
                if not self.short_entered:
                    self.short(self.short_entry, self.fixed_size, stop=True)
            elif self.pos < 0:
                self.short_entered = True
                self.cover(self.long_entry, self.fixed_size, stop=True)
                if not self.long_entered:
                    self.buy(self.long_entry, self.fixed_size, stop=True)
        else:
            if self.pos > 0:
                self.sell(bar.close_price * 0.99, abs(self.pos))
            elif self.pos < 0:
                self.cover(bar.close_price * 1.01, abs(self.pos))
        self.put_event()

今天的文章【vn.py】源码解析之 Dual Thrust 策略分享到此就结束了，感谢您的阅读。