【vn.py】策略实现之R-Breaker策略-白红宇

【vn.py】策略实现之R-Breaker策略

阅读量：2052 次

发布时间：2019-04-28

本文共 9383 字，大约阅读时间需要 31 分钟。

文章目录

写在前面

vnpy中提供了很多CTA策略的源码，前面的文章也对很多策略进行了代码分析。这些工作不仅仅是为了了解那些策略的逻辑，更多的是为以后的自己基于vnpy进行策略提供思路和借鉴。因此，这篇文章将以有名的R-Breaker策略为例，基于vnpy实现这个策略的逻辑并进行回测与参数优化。

R-Breaker

R-Breaker是一种中高频的日内交易策略，这个策略也长期被Future Truth杂志评为最赚钱的策略之一。R-Breaker策略结合了趋势和反转两种交易方式，所以交易机会相对较多，比较适合日内1-Min和5-Min级别的数据。它的交易思想是：

1、根据昨日的开高低收价位计算出今日的六个价位，按照价格高低依次是：突破买入价（Bbreak）、观察卖出价（Ssetup）、反转卖出价（Senter）、反转买入价（Benter）、观察买入价（Bsetup）、突破卖出价（Sbreak），依次作为交易的六个触发价位。其中六个价位的计算方式如下:

Ssetup= High + 0.35 * (Close – Low)

Bsetup= Low – 0.35 * (High – Close)

Senter= 1.07 / 2 * (High + Low) – 0.07 * Low

Benter= 1.07 / 2 * (High + Low) – 0.07 * High

Bbreak= Ssetup+ 0.25 * (Ssetup– Bsetup)

Sbreak = Bsetup– 0.25 * (Ssetup– Bsetup)

2、根据价格的走势满足哪种情况来决定是否开仓：

情况1，如果持空仓，价格超过突破买入价，采取趋势策略，顺势开仓做多；

情况2，如果持有多单（空仓），价格超过观察卖出价，之后跌破反转卖出价，采取反转策略，反手做空（开空）；

情况3，如果持有空单（空仓），价格跌破观察买入价，之后超过反转买入价，采取反转策略，反手做多（开多）；

情况4，如果持空仓，价格跌破突破卖出价，采取趋势策略，顺势开仓做空；

3、设定相应的止盈止损。

4、收盘前平仓。

5、也可以根据昨日价格的波幅进行过滤，如果波幅较小则今日不开仓。同时也可以以其他指标如ATR、CCI、RSI进行过滤。

R-Breaker策略实现以及代码讲解

下面将对R-Breaker策略基于vnpy进行策略实现。

1、策略参数以及变量

R-breaker策略的参数主要是四个乘数，用于控制六个价位之间的距离。

# 定义参数    setup_coef = 0.35    break_coef = 0.25    enter_coef1 = 1.07    enter_coef2 = 0.07    fixed_size = 1    # 定义变量    Bbreak = 0  # 突破买入价    Ssetup = 0  # 观察卖出价    Senter = 0  # 反转卖出价    Benter = 0  # 反转买入价    Bsetup = 0  # 观察买入价    Sbreak = 0  # 突破卖出价    # 昨日开高低收    day_high = 0    day_open = 0    day_close = 0    day_low = 0    exit_time = time(hour=14, minute=55)    # 添加参数和变量名到对应的列表    parameters = ["setup_coef", "break_coef", "enter_coef1", "enter_coef2", "fixed_size"]    variables = ["Bbreak", "Ssetup", "Senter", "Benter", "Bsetup", "Sbreak"]

2、策略执行逻辑

同之前的策略一样，R-breaker也需要加载历史数据对六个价位变量进行初始化。

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.bars = []        self.am = ArrayManager()    def on_init(self):        """        Callback when strategy is inited.        """        self.write_log("策略初始化")        self.load_bar(10)

R-breaker策略的交易逻辑也是写在了on_bar()函数中。一开始取消上一Bar数据中没有交易成功的订单，然后将新的bar数据添加到bars中，目的是为了比较当前bar数据与上一个bar数据之间的日期与时间。

def on_bar(self, bar: BarData):        """        Callback of new bar data update.        """        self.cancel_all()        am = self.am        am.update_bar(bar)        if not am.inited:            return        self.bars.append(bar)        if len(self.bars) <= 2:            return        else:            self.bars.pop(0)        last_bar = self.bars[-2]

下面这段的逻辑就是通过上面获取的上一个bar数据与当前bar数据进行日期比较，从而获取昨日的开高低收价格，从而根据这四个价格来计算用于今日交易的六个价位。如果历史数据加载的还不够，六个价位数据还未计算出，那么就直接return。

if last_bar.datetime.date() != bar.datetime.date():  # 如果是第二天的bar数据            if self.day_open:                self.Bsetup = self.day_low - self.setup_coef*(self.day_high - self.day_close)                self.Ssetup = self.day_high + self.setup_coef*(self.day_close - self.day_low)                self.Benter = (self.enter_coef1/2)*(self.day_high + self.day_low) - self.enter_coef2*self.day_high                self.Senter = (self.enter_coef1/2)*(self.day_high + self.day_low) - self.enter_coef2*self.day_low                self.Bbreak = self.Ssetup + self.break_coef*(self.Ssetup - self.Bsetup)                self.Sbreak = self.Bsetup + self.break_coef*(self.Ssetup - self.Bsetup)            self.day_open = bar.open_price            self.day_high = bar.high_price            self.day_close = bar.close_price            self.day_low = bar.low_price        else:  # 如果是当天的数据            self.day_high = max(self.day_high, bar.high_price)            self.day_low = min(self.day_low, bar.low_price)            self.day_close = bar.close_price         # 如果还未计算六个价格        if not self.Ssetup:            return

下面的部分就是根据当前bar数据在六个价位组成的区域来决定发出什么样的订单。因为R-breaker策略是在盘中触发的，所以我们这里发出的订单都是停止单，目的也是为了更好地在价格达到指定条件时触发交易。

1、如果当前bar最高价格超过观察卖出价，并且收盘价也超过了观察卖出价，那么就以突破买入价发出一笔停止单做多。

2、如果当前bar的最高价超过了观察卖出价，并且收盘价回落到了观察卖出价之下，那么就以反转卖出价发出一笔停止单做空。

3、如果当前bar的最低价跌破观察卖出价，并且收盘价也在观察卖出价之下，那么就以突破卖出价发出一笔停止单做空。

4、如果当前bar的最低价跌破观察卖出价，并且收盘价在观察卖出价之上，那么就以反转买入价发出一笔停止单做多。

在每日交易结束之前平仓。

# 如果在交易时间内        if bar.datetime.time() < self.exit_time:             if self.pos == 0:                 if bar.high_price > self.Ssetup and bar.close_price > self.Ssetup:                     self.buy(self.Bbreak, self.fixed_size, stop=True)                 elif bar.high_price > self.Ssetup and bar.close_price < self.Ssetup:                     self.short(self.Senter, self.fixed_size, stop=True)                 elif bar.low_price < self.Bsetup and bar.close_price < self.Bsetup:                     self.short(self.Sbreak, self.fixed_size, stop=True)                 elif bar.low_price < self.Bsetup and bar.close_price > self.Bsetup:                     self.buy(self.Benter, 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()

3、完整源码

from datetime import timefrom vnpy.app.cta_strategy import (    CtaTemplate,    StopOrder,    TickData,    BarData,    TradeData,    OrderData,    BarGenerator,    ArrayManager,)class RBreakStrategy(CtaTemplate):    """R break策略"""    # 策略作者    author = "Frankie"    # 定义参数    setup_coef = 0.35    break_coef = 0.25    enter_coef1 = 1.07    enter_coef2 = 0.07    fixed_size = 1    # 定义变量    Bbreak = 0  # 突破买入价    Ssetup = 0  # 观察卖出价    Senter = 0  # 反转卖出价    Benter = 0  # 反转买入价    Bsetup = 0  # 观察买入价    Sbreak = 0  # 突破卖出价    # 昨日开高低收    day_high = 0    day_open = 0    day_close = 0    day_low = 0    exit_time = time(hour=14, minute=55)    # 添加参数和变量名到对应的列表    parameters = ["setup_coef", "break_coef", "enter_coef1", "enter_coef2", "fixed_size"]    variables = ["Bbreak", "Ssetup", "Senter", "Benter", "Bsetup", "Sbreak"]    def __init__(self, cta_engine, strategy_name, vt_symbol, setting):        """"""        super().__init__(cta_engine, strategy_name, vt_symbol, setting)        self.bg = BarGenerator(self.on_bar)        self.am = ArrayManager()        self.bars = []    def on_init(self):        self.write_log("策略初始化")        self.load_bar(10)    def on_start(self):        self.write_log("策略启动")        self.put_event()    def on_stop(self):        self.write_log("策略停止")        self.put_event()    def on_tick(self, tick: TickData):        self.bg.update_tick(tick)    def on_bar(self, bar: BarData):        self.cancel_all()        am = self.am        am.update_bar(bar)        if not am.inited:            return        self.bars.append(bar)        if len(self.bars) <= 2:            return        else:            self.bars.pop(0)        last_bar = self.bars[-2]  # 上一个bar数据        # ------------ 计算前一天的开高低收 ----------------        if last_bar.datetime.date() != bar.datetime.date():  # 如果是第二天的bar数据            if self.day_open:                self.Bsetup = self.day_low - self.setup_coef*(self.day_high - self.day_close)                self.Ssetup = self.day_high + self.setup_coef*(self.day_close - self.day_low)                self.Benter = (self.enter_coef1/2)*(self.day_high + self.day_low) - self.enter_coef2*self.day_high                self.Senter = (self.enter_coef1/2)*(self.day_high + self.day_low) - self.enter_coef2*self.day_low                self.Bbreak = self.Ssetup + self.break_coef*(self.Ssetup - self.Bsetup)                self.Sbreak = self.Bsetup + self.break_coef*(self.Ssetup - self.Bsetup)            self.day_open = bar.open_price            self.day_high = bar.high_price            self.day_close = bar.close_price            self.day_low = bar.low_price        else:  # 如果是当天的数据            self.day_high = max(self.day_high, bar.high_price)            self.day_low = min(self.day_low, bar.low_price)        # 如果还未计算六个价格        if not self.Ssetup:            return        # 如果在交易时间内        if bar.datetime.time() < self.exit_time:             if self.pos == 0:                 if bar.high_price > self.Ssetup and bar.close_price > self.Ssetup:                     self.buy(self.Bbreak, self.fixed_size, stop=True)                 elif bar.high_price > self.Ssetup and bar.close_price < self.Ssetup:                     self.short(self.Senter, self.fixed_size, stop=True)                 elif bar.low_price < self.Bsetup and bar.close_price < self.Bsetup:                     self.short(self.Sbreak, self.fixed_size, stop=True)                 elif bar.low_price < self.Bsetup and bar.close_price > self.Bsetup:                     self.buy(self.Benter, 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):        """        通过该函数收到委托状态更新推送。        """        pass    def on_trade(self, trade: TradeData):        """        通过该函数收到成交推送。        """        # 成交后策略逻辑仓位发生变化，需要通知界面更新。        self.put_event()    def on_stop_order(self, stop_order: StopOrder):        """        通过该函数收到本地停止单推送。        """        pass