策略开发

CTA策略模板提供完整的信号生成和委托管理功能,用户可以基于该模板自行开发策略。新策略可以放在根目录下vnpy\app\cta_strategy\strategies文件夹内,也可以放在用户运行的文件内(VN Station模式)。注意:策略文件命名是以下划线模式,如boll_channel_strategy.py;而策略类命名采用的是驼峰式,如BollChannelStrategy。

下面通过BollChannelStrategy策略示例,来展示策略开发的具体步骤:

参数设置

定义策略参数并且初始化策略变量。策略参数为策略类的公有属性,用户可以通过创建新的实例来调用或者改变策略参数。

如针对rb1905品种,用户可以创建基于BollChannelStrategy的策略示例,如RB_BollChannelStrategy,boll_window可以由18改成30。

创建策略实例的方法有效地实现了一个策略跑多个品种,并且其策略参数可以通过品种的特征进行调整。

  1. boll_window = 18
  2. boll_dev = 3.4
  3. cci_window = 10
  4. atr_window = 30
  5. sl_multiplier = 5.2
  6. fixed_size = 1
  7.  
  8. boll_up = 0
  9. boll_down = 0
  10. cci_value = 0
  11. atr_value = 0
  12.  
  13. intra_trade_high = 0
  14. intra_trade_low = 0
  15. long_stop = 0
  16. short_stop = 0

类的初始化

初始化分3步:

  • 通过super( )的方法继承CTA策略模板,在init( )函数传入CTA引擎、策略名称、vt_symbol、参数设置。
  • 调用K线生成模块:通过时间切片来把Tick数据合成1分钟K线数据,然后更大的时间周期数据,如15分钟K线。
  • 调用K线时间序列管理模块:基于K线数据,如1分钟、15分钟,来生成相应的技术指标。
  1. def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
  2. """"""
  3. super(BollChannelStrategy, self).__init__(
  4. cta_engine, strategy_name, vt_symbol, setting
  5. )
  6.  
  7. self.bg = BarGenerator(self.on_bar, 15, self.on_15min_bar)
  8. self.am = ArrayManager()

策略的初始化、启动、停止

通过“CTA策略”组件的相关功能按钮实现。

注意:函数load_bar(10),代表策略初始化需要载入10个交易日的历史数据。该历史数据可以是Tick数据,也可以是K线数据。

  1. def on_init(self):
  2. """
  3. Callback when strategy is inited.
  4. """
  5. self.write_log("策略初始化")
  6. self.load_bar(10)
  7.  
  8. def on_start(self):
  9. """
  10. Callback when strategy is started.
  11. """
  12. self.write_log("策略启动")
  13.  
  14. def on_stop(self):
  15. """
  16. Callback when strategy is stopped.
  17. """
  18. self.write_log("策略停止")

Tick数据回报

策略订阅某品种合约行情,交易所会推送Tick数据到该策略上。

由于BollChannelStrategy是基于15分钟K线来生成交易信号的,故收到Tick数据后,需要用到K线生成模块里面的update_tick函数,通过时间切片的方法,聚合成1分钟K线数据,并且推送到on_bar函数。

  1. def on_tick(self, tick: TickData):
  2. """
  3. Callback of new tick data update.
  4. """
  5. self.bg.update_tick(tick)

K线数据回报

收到推送过来的1分钟K线数据后,通过K线生成模块里面的update_bar函数,以分钟切片的方法,合成15分钟K线数据,并且推送到on_15min_bar函数。

  1. def on_bar(self, bar: BarData):
  2. """
  3. Callback of new bar data update.
  4. """
  5. self.bg.update_bar(bar)

15分钟K线数据回报

负责CTA信号的生成,由3部分组成:

  • 清空未成交委托:为了防止之前下的单子在上一个15分钟没有成交,但是下一个15分钟可能已经调整了价格,就用cancel_all()方法立刻撤销之前未成交的所有委托,保证策略在当前这15分钟开始时的整个状态是清晰和唯一的。
  • 调用K线时间序列管理模块:基于最新的15分钟K线数据来计算相应计算指标,如布林带通道上下轨、CCI指标、ATR指标
  • 信号计算:通过持仓的判断以及结合CCI指标、布林带通道、ATR指标在通道突破点挂出停止单委托(buy/sell),同时设置离场点(short/cover)。注意:CTA策略具有低胜率和高盈亏比的特定:在难以提升胜率的情况下,研究提高策略盈亏比有利于策略盈利水平的上升。
  1. def on_15min_bar(self, bar: BarData):
  2. """"""
  3. self.cancel_all()
  4.  
  5. am = self.am
  6. am.update_bar(bar)
  7. if not am.inited:
  8. return
  9.  
  10. self.boll_up, self.boll_down = am.boll(self.boll_window, self.boll_dev)
  11. self.cci_value = am.cci(self.cci_window)
  12. self.atr_value = am.atr(self.atr_window)
  13.  
  14. if self.pos == 0:
  15. self.intra_trade_high = bar.high_price
  16. self.intra_trade_low = bar.low_price
  17.  
  18. if self.cci_value > 0:
  19. self.buy(self.boll_up, self.fixed_size, True)
  20. elif self.cci_value < 0:
  21. self.short(self.boll_down, self.fixed_size, True)
  22.  
  23. elif self.pos > 0:
  24. self.intra_trade_high = max(self.intra_trade_high, bar.high_price)
  25. self.intra_trade_low = bar.low_price
  26.  
  27. self.long_stop = self.intra_trade_high - self.atr_value * self.sl_multiplier
  28. self.sell(self.long_stop, abs(self.pos), True)
  29.  
  30. elif self.pos < 0:
  31. self.intra_trade_high = bar.high_price
  32. self.intra_trade_low = min(self.intra_trade_low, bar.low_price)
  33.  
  34. self.short_stop = self.intra_trade_low + self.atr_value * self.sl_multiplier
  35. self.cover(self.short_stop, abs(self.pos), True)
  36.  
  37. self.put_event()

委托回报、成交回报、停止单回报

在策略中可以直接pass,其具体逻辑应用交给回测/实盘引擎负责。

  1. def on_order(self, order: OrderData):
  2. """
  3. Callback of new order data update.
  4. """
  5. pass
  6.  
  7. def on_trade(self, trade: TradeData):
  8. """
  9. Callback of new trade data update.
  10. """
  11. self.put_event()
  12.  
  13. def on_stop_order(self, stop_order: StopOrder):
  14. """
  15. Callback of stop order update.
  16. """
  17. pass