策略开发

多合约组合策略模板提供完整的信号生成和委托管理功能，用户可以基于该模板自行开发策略。新策略可以放在用户运行的文件内（推荐），如在c:\users\administrator.vntrader目录下创建strategies文件夹；可以放在根目录下vnpy\app\portfolio_strategy\strategies文件夹内。 注意：策略文件命名是以下划线模式，如pair_trading_strategy.py；而策略类命名采用的是驼峰式，如PairTradingStrategy。

下面通过PairTradingStrategy策略示例，来展示策略开发的具体步骤：

参数设置

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

如针对rb和hc品种，用户可以创建基于PairTradingStrategy的策略示例，如rb_hc_PairTradingStrategy，boll_window可以由20改成30。

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

  1.     price_add = 5
  2.     boll_window = 20
  3.     boll_dev = 2
  4.     fixed_size = 1
  5.     leg1_ratio = 1
  6.     leg2_ratio = 1
  8.     leg1_symbol = ""
  9.     leg2_symbol = ""
  10.     current_spread = 0.0
  11.     boll_mid = 0.0
  12.     boll_down = 0.0
  13.     boll_up = 0.0

类的初始化

初始化：

  • 通过super( )的方法继承StrategyTemplate，在__init__( )函数传入Strategy引擎、策略名称、vt_symbols、参数设置。

  • 调用K线生成模块:通过时间切片来把Tick数据合成1分钟K线数据。

  • 如果需要，可以调用K线时间序列管理模块：基于K线数据，来生成相应的技术指标。

  1.     def __init__(
  2.         self,
  3.         strategy_engine: StrategyEngine,
  4.         strategy_name: str,
  5.         vt_symbols: List[str],
  6.         setting: dict
  7.     ):
  8.         """"""
  9.         super().__init__(strategy_engine, strategy_name, vt_symbols, setting)
  11.         self.bgs: Dict[str, BarGenerator] = {}
  12.         self.targets: Dict[str, int] = {}
  13.         self.last_tick_time: datetime = None
  15.         self.spread_count: int = 0
  16.         self.spread_data: np.array = np.zeros(100)
  18.         # Obtain contract info
  19.         self.leg1_symbol, self.leg2_symbol = vt_symbols
  21.         def on_bar(bar: BarData):
  22.             """"""
  23.             pass
  25.         for vt_symbol in self.vt_symbols:
  26.             self.targets[vt_symbol] = 0
  27.             self.bgs[vt_symbol] = BarGenerator(on_bar)

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

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

注意：函数load_bar(1)，代表策略初始化需要载入1个交易日的历史数据。该历史数据只能是K线数据（on_tick只有实盘中会调用，回测不支持）。在策略初始化时候，会调用K线时间序列管理器计算并缓存相关的计算指标，但是并不触发交易。

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

Tick数据回报

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

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

  1.     def on_tick(self, tick: TickData):
  2.         """
  3.         Callback of new tick data update.
  4.         """
  5.         if (
  6.             self.last_tick_time
  7.             and self.last_tick_time.minute != tick.datetime.minute
  8.         ):
  9.             bars = {}
  10.             for vt_symbol, bg in self.bgs.items():
  11.                 bars[vt_symbol] = bg.generate()
  12.             self.on_bars(bars)
  14.         bg: BarGenerator = self.bgs[tick.vt_symbol]
  15.         bg.update_tick(tick)
  17.         self.last_tick_time = tick.datetime

K线数据回报

在回调函数on_bars中，同时收到所有合约的K线行情推送，并实现核心交易逻辑。调用buy/sell/short/cover/cancel_order等函数来发送交易请求。

信号的生成，由3部分组成：

  • 清空未成交委托：为了防止之前下的单子在上一个5分钟没有成交，但是下一个5分钟可能已经调整了价格，就用cancel_all()方法立刻撤销之前未成交的所有委托，保证策略在当前这5分钟开始时的整个状态是清晰和唯一的。

  • 时间过滤：这里有用(bar.datetime.minute+1) % 5的逻辑来达到每5分钟推送一次的逻辑。

  • 指标计算：基于最新的5分钟K线数据来计算相应计算指标，如价差的布林带通道上下轨。

  • 信号计算：通过持仓的判断以及结合布林带通道上下轨，在通道突破点以及离场点设置目标仓位。

  • 仓位管理：基于目标仓位和现在持仓的差别来计算本次下单的数量并挂出委托（buy/sell)，同时设置离场点(short/cover)。

  1.     def on_bars(self, bars: Dict[str, BarData]):
  2.         """"""
  3.         self.cancel_all()
  5.         # Return if one leg data is missing
  6.         if self.leg1_symbol not in bars or self.leg2_symbol not in bars:
  7.             return
  9.         # Calculate current spread
  10.         leg1_bar = bars[self.leg1_symbol]
  11.         leg2_bar = bars[self.leg2_symbol]
  13.         # Filter time only run every 5 minutes
  14.         if (leg1_bar.datetime.minute + 1) % 5:
  15.             return
  17.         self.current_spread = (
  18.             leg1_bar.close_price * self.leg1_ratio - leg2_bar.close_price * self.leg2_ratio
  19.         )
  21.         # Update to spread array
  22.         self.spread_data[:-1] = self.spread_data[1:]
  23.         self.spread_data[-1] = self.current_spread
  25.         self.spread_count += 1
  26.         if self.spread_count <= self.boll_window:
  27.             return
  29.         # Calculate boll value
  30.         buf: np.array = self.spread_data[-self.boll_window:]
  32.         std = buf.std()
  33.         self.boll_mid = buf.mean()
  34.         self.boll_up = self.boll_mid + self.boll_dev * std
  35.         self.boll_down = self.boll_mid - self.boll_dev * std
  37.         # Calculate new target position
  38.         leg1_pos = self.get_pos(self.leg1_symbol)
  40.         if not leg1_pos:
  41.             if self.current_spread >= self.boll_up:
  42.                 self.targets[self.leg1_symbol] = -1
  43.                 self.targets[self.leg2_symbol] = 1
  44.             elif self.current_spread <= self.boll_down:
  45.                 self.targets[self.leg1_symbol] = 1
  46.                 self.targets[self.leg2_symbol] = -1
  47.         elif leg1_pos > 0:
  48.             if self.current_spread >= self.boll_mid:
  49.                 self.targets[self.leg1_symbol] = 0
  50.                 self.targets[self.leg2_symbol] = 0
  51.         else:
  52.             if self.current_spread <= self.boll_mid:
  53.                 self.targets[self.leg1_symbol] = 0
  54.                 self.targets[self.leg2_symbol] = 0
  56.         # Execute orders
  57.         for vt_symbol in self.vt_symbols:
  58.             target_pos = self.targets[vt_symbol]
  59.             current_pos = self.get_pos(vt_symbol)
  61.             pos_diff = target_pos - current_pos
  62.             volume = abs(pos_diff)
  63.             bar = bars[vt_symbol]
  65.             if pos_diff > 0:
  66.                 price = bar.close_price + self.price_add
  68.                 if current_pos < 0:
  69.                     self.cover(vt_symbol, price, volume)
  70.                 else:
  71.                     self.buy(vt_symbol, price, volume)
  72.             elif pos_diff < 0:
  73.                 price = bar.close_price - self.price_add
  75.                 if current_pos > 0:
  76.                     self.sell(vt_symbol, price, volume)
  77.                 else:
  78.                     self.short(vt_symbol, price, volume)
  80.         self.put_event()

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

在于组合策略中需要对多合约同时下单交易，在回测时无法判断某一段K线内部每个合约委托成交的先后时间顺序，因此无法提供on_order和on_trade获取委托成交推送，而只能在每次on_bars回调时通过get_pos和get_order来进行相关的状态查询。

同时组合策略模块只支持限价单交易，不提供停止单功能（StopOrder）。

  1.     def send_order(
  2.         self,
  3.         vt_symbol: str,
  4.         direction: Direction,
  5.         offset: Offset,
  6.         price: float,
  7.         volume: float,
  8.         lock: bool = False
  9.     ) -> List[str]:
  10.         """
  11.         Send a new order.
  12.         """
  13.         if self.trading:
  14.             vt_orderids = self.strategy_engine.send_order(
  15.                 self, vt_symbol, direction, offset, price, volume, lock
  16.             )
  17.             return vt_orderids
  18.         else:
  19.             return []