回测研究

backtesting.py定义了回测引擎，下面主要介绍相关功能函数，以及回测引擎应用示例：

加载策略

把组合策略逻辑，对应所有合约品种，以及参数设置（可在策略文件外修改）载入到回测引擎中。

  1.     def add_strategy(self, strategy_class: type, setting: dict) -> None:
  2.         """"""
  3.         self.strategy = strategy_class(
  4.             self, strategy_class.__name__, copy(self.vt_symbols), setting
  5.         )

载入历史数据

负责载入对应品种的历史数据，大概有4个步骤：

  • 检查起始日期是否小于结束日期；

  • 清除之前载入的历史数据；

  • 有条件地从数据库中选取数据，每次载入30天数据。其筛选标准包括：vt_symbol、 回测开始日期、回测结束日期、K线周期；

  • 载入数据是以迭代方式进行的，数据最终存入self.history_data。

  1.     def load_data(self) -> None:
  2.         """"""
  3.         self.output("开始加载历史数据")
  5.         if not self.end:
  6.             self.end = datetime.now()
  8.         if self.start >= self.end:
  9.             self.output("起始日期必须小于结束日期")
  10.             return
  12.         # Clear previously loaded history data
  13.         self.history_data.clear()
  14.         self.dts.clear()
  16.         # Load 30 days of data each time and allow for progress update
  17.         progress_delta = timedelta(days=30)
  18.         total_delta = self.end - self.start
  19.         interval_delta = INTERVAL_DELTA_MAP[self.interval]
  21.         for vt_symbol in self.vt_symbols:
  22.             start = self.start
  23.             end = self.start + progress_delta
  24.             progress = 0
  26.             data_count = 0
  27.             while start < self.end:
  28.                 end = min(end, self.end)  # Make sure end time stays within set range
  30.                 data = load_bar_data(
  31.                     vt_symbol,
  32.                     self.interval,
  33.                     start,
  34.                     end
  35.                 )
  37.                 for bar in data:
  38.                     self.dts.add(bar.datetime)
  39.                     self.history_data[(bar.datetime, vt_symbol)] = bar
  40.                     data_count += 1
  42.                 progress += progress_delta / total_delta
  43.                 progress = min(progress, 1)
  44.                 progress_bar = "#" * int(progress * 10)
  45.                 self.output(f"{vt_symbol}加载进度：{progress_bar} [{progress:.0%}]")
  47.                 start = end + interval_delta
  48.                 end += (progress_delta + interval_delta)
  50.             self.output(f"{vt_symbol}历史数据加载完成，数据量：{data_count}")
  52.         self.output("所有历史数据加载完成")

撮合成交

载入组合策略以及相关历史数据后，策略会根据最新的数据来计算相关指标。若符合条件会生成交易信号，发出具体委托（buy/sell/short/cover），并且在下一根K线成交。

回测引擎提供限价单撮合成交机制来尽量模仿真实交易环节：

  • 限价单撮合成交：（以买入方向为例）先确定是否发生成交，成交标准为委托价>= 下一根K线的最低价；然后确定成交价格，成交价格为委托价与下一根K线开盘价的最小值。

下面展示在引擎中限价单撮合成交的流程：

  • 确定会撮合成交的价格；

  • 遍历限价单字典中的所有限价单，推送委托进入未成交队列的更新状态；

  • 判断成交状态，若出现成交，推送成交数据和委托数据；

  • 从字典中删除已成交的限价单。

  1.     def cross_limit_order(self) -> None:
  2.         """
  3.         Cross limit order with last bar/tick data.
  4.         """
  5.         for order in list(self.active_limit_orders.values()):
  6.             bar = self.bars[order.vt_symbol]
  8.             long_cross_price = bar.low_price
  9.             short_cross_price = bar.high_price
  10.             long_best_price = bar.open_price
  11.             short_best_price = bar.open_price
  13.             # Push order update with status "not traded" (pending).
  14.             if order.status == Status.SUBMITTING:
  15.                 order.status = Status.NOTTRADED
  16.                 self.strategy.update_order(order)
  18.             # Check whether limit orders can be filled.
  19.             long_cross = (
  20.                 order.direction == Direction.LONG
  21.                 and order.price >= long_cross_price
  22.                 and long_cross_price > 0
  23.             )
  25.             short_cross = (
  26.                 order.direction == Direction.SHORT
  27.                 and order.price <= short_cross_price
  28.                 and short_cross_price > 0
  29.             )
  31.             if not long_cross and not short_cross:
  32.                 continue
  34.             # Push order update with status "all traded" (filled).
  35.             order.traded = order.volume
  36.             order.status = Status.ALLTRADED
  37.             self.strategy.update_order(order)
  39.             self.active_limit_orders.pop(order.vt_orderid)
  41.             # Push trade update
  42.             self.trade_count += 1
  44.             if long_cross:
  45.                 trade_price = min(order.price, long_best_price)
  46.             else:
  47.                 trade_price = max(order.price, short_best_price)
  49.             trade = TradeData(
  50.                 symbol=order.symbol,
  51.                 exchange=order.exchange,
  52.                 orderid=order.orderid,
  53.                 tradeid=str(self.trade_count),
  54.                 direction=order.direction,
  55.                 offset=order.offset,
  56.                 price=trade_price,
  57.                 volume=order.volume,
  58.                 datetime=self.datetime,
  59.                 gateway_name=self.gateway_name,
  60.             )
  62.             self.strategy.update_trade(trade)

计算策略盈亏情况

基于收盘价、当日持仓量、合约规模、滑点、手续费率等计算总盈亏与净盈亏，并且其计算结果以DataFrame格式输出，完成基于逐日盯市盈亏统计。

下面展示盈亏情况的计算过程

  • 浮动盈亏 = 持仓量 *（当日收盘价 - 昨日收盘价）* 合约规模

  • 实际盈亏 = 持仓变化量 * （当时收盘价 - 开仓成交价）* 合约规模

  • 总盈亏 = 浮动盈亏 + 实际盈亏

  • 净盈亏 = 总盈亏 - 总手续费 - 总滑点

  1.     def calculate_pnl(
  2.         self,
  3.         pre_close: float,
  4.         start_pos: float,
  5.         size: int,
  6.         rate: float,
  7.         slippage: float
  8.     ) -> None:
  9.         """"""
  10.         # If no pre_close provided on the first day,
  11.         # use value 1 to avoid zero division error
  12.         if pre_close:
  13.             self.pre_close = pre_close
  14.         else:
  15.             self.pre_close = 1
  17.         # Holding pnl is the pnl from holding position at day start
  18.         self.start_pos = start_pos
  19.         self.end_pos = start_pos
  21.         self.holding_pnl = self.start_pos * (self.close_price - self.pre_close) * size
  23.         # Trading pnl is the pnl from new trade during the day
  24.         self.trade_count = len(self.trades)
  26.         for trade in self.trades:
  27.             if trade.direction == Direction.LONG:
  28.                 pos_change = trade.volume
  29.             else:
  30.                 pos_change = -trade.volume
  32.             self.end_pos += pos_change
  34.             turnover = trade.volume * size * trade.price
  36.             self.trading_pnl += pos_change * (self.close_price - trade.price) * size
  37.             self.slippage += trade.volume * size * slippage
  38.             self.turnover += turnover
  39.             self.commission += turnover * rate
  41.         # Net pnl takes account of commission and slippage cost
  42.         self.total_pnl = self.trading_pnl + self.holding_pnl
  43.         self.net_pnl = self.total_pnl - self.commission - self.slippage

计算策略统计指标

calculate_statistics函数是基于逐日盯市盈亏情况（DateFrame格式）来计算衍生指标，如最大回撤、年化收益、盈亏比、夏普比率等。

  1.             df["balance"] = df["net_pnl"].cumsum() + self.capital
  2.             df["return"] = np.log(df["balance"] / df["balance"].shift(1)).fillna(0)
  3.             df["highlevel"] = (
  4.                 df["balance"].rolling(
  5.                     min_periods=1, window=len(df), center=False).max()
  6.             )
  7.             df["drawdown"] = df["balance"] - df["highlevel"]
  8.             df["ddpercent"] = df["drawdown"] / df["highlevel"] * 100
  10.             # Calculate statistics value
  11.             start_date = df.index[0]
  12.             end_date = df.index[-1]
  14.             total_days = len(df)
  15.             profit_days = len(df[df["net_pnl"] > 0])
  16.             loss_days = len(df[df["net_pnl"] < 0])
  18.             end_balance = df["balance"].iloc[-1]
  19.             max_drawdown = df["drawdown"].min()
  20.             max_ddpercent = df["ddpercent"].min()
  21.             max_drawdown_end = df["drawdown"].idxmin()
  23.             if isinstance(max_drawdown_end, date):
  24.                 max_drawdown_start = df["balance"][:max_drawdown_end].idxmax()
  25.                 max_drawdown_duration = (max_drawdown_end - max_drawdown_start).days
  26.             else:
  27.                 max_drawdown_duration = 0
  29.             total_net_pnl = df["net_pnl"].sum()
  30.             daily_net_pnl = total_net_pnl / total_days
  32.             total_commission = df["commission"].sum()
  33.             daily_commission = total_commission / total_days
  35.             total_slippage = df["slippage"].sum()
  36.             daily_slippage = total_slippage / total_days
  38.             total_turnover = df["turnover"].sum()
  39.             daily_turnover = total_turnover / total_days
  41.             total_trade_count = df["trade_count"].sum()
  42.             daily_trade_count = total_trade_count / total_days
  44.             total_return = (end_balance / self.capital - 1) * 100
  45.             annual_return = total_return / total_days * 240
  46.             daily_return = df["return"].mean() * 100
  47.             return_std = df["return"].std() * 100
  49.             pnl_std = df["net_pnl"].std()
  51.             if return_std:
  52.                 # sharpe_ratio = daily_return / return_std * np.sqrt(240)
  53.                 sharpe_ratio = daily_net_pnl / pnl_std * np.sqrt(240)
  54.             else:
  55.                 sharpe_ratio = 0
  57.             return_drawdown_ratio = -total_net_pnl / max_drawdown

统计指标绘图

通过matplotlib绘制4幅图：

  • 资金曲线图

  • 资金回撤图

  • 每日盈亏图

  • 每日盈亏分布图

  1.     def show_chart(self, df: DataFrame = None) -> None:
  2.         """"""
  3.         # Check DataFrame input exterior
  4.         if df is None:
  5.             df = self.daily_df
  7.         # Check for init DataFrame
  8.         if df is None:
  9.             return
  11.         fig = make_subplots(
  12.             rows=4,
  13.             cols=1,
  14.             subplot_titles=["Balance", "Drawdown", "Daily Pnl", "Pnl Distribution"],
  15.             vertical_spacing=0.06
  16.         )
  18.         balance_line = go.Scatter(
  19.             x=df.index,
  20.             y=df["balance"],
  21.             mode="lines",
  22.             name="Balance"
  23.         )
  24.         drawdown_scatter = go.Scatter(
  25.             x=df.index,
  26.             y=df["drawdown"],
  27.             fillcolor="red",
  28.             fill='tozeroy',
  29.             mode="lines",
  30.             name="Drawdown"
  31.         )
  32.         pnl_bar = go.Bar(y=df["net_pnl"], name="Daily Pnl")
  33.         pnl_histogram = go.Histogram(x=df["net_pnl"], nbinsx=100, name="Days")
  35.         fig.add_trace(balance_line, row=1, col=1)
  36.         fig.add_trace(drawdown_scatter, row=2, col=1)
  37.         fig.add_trace(pnl_bar, row=3, col=1)
  38.         fig.add_trace(pnl_histogram, row=4, col=1)
  40.         fig.update_layout(height=1000, width=1000)
  41.         fig.show()    def show_chart(self, df: DataFrame = None):
  42.         """"""
  43.         if not df:
  44.             df = self.daily_df
  46.         if df is None:
  47.             return
  49.         plt.figure(figsize=(10, 16))
  51.         balance_plot = plt.subplot(4, 1, 1)
  52.         balance_plot.set_title("Balance")
  53.         df["balance"].plot(legend=True)
  55.         drawdown_plot = plt.subplot(4, 1, 2)
  56.         drawdown_plot.set_title("Drawdown")
  57.         drawdown_plot.fill_between(range(len(df)), df["drawdown"].values)
  59.         pnl_plot = plt.subplot(4, 1, 3)
  60.         pnl_plot.set_title("Daily Pnl")
  61.         df["net_pnl"].plot(kind="bar", legend=False, grid=False, xticks=[])
  63.         distribution_plot = plt.subplot(4, 1, 4)
  64.         distribution_plot.set_title("Daily Pnl Distribution")
  65.         df["net_pnl"].hist(bins=50)
  67.         plt.show()

效果如下：

https://user-images.githubusercontent.com/11263900/93662305-b8737e00-fa91-11ea-8c83-70ca30e58dc4.png回测结果

https://user-images.githubusercontent.com/11263900/93662331-ec4ea380-fa91-11ea-9e93-d61f90d0d282.png回测图

策略回测示例

  • 导入回测引擎和组合策略

  • 设置回测相关参数，如：品种、K线周期、回测开始和结束日期、手续费、滑点、合约规模、起始资金

  • 载入策略和数据到引擎中，运行回测。

  • 计算基于逐日统计盈利情况，计算统计指标，统计指标绘图。

  1. from datetime import datetime
  2. from importlib import reload
  4. import vnpy.app.portfolio_strategy
  5. reload(vnpy.app.portfolio_strategy)
  7. from vnpy.app.portfolio_strategy import BacktestingEngine
  8. from vnpy.trader.constant import Interval
  10. import vnpy.app.portfolio_strategy.strategies.pair_trading_strategy as stg
  11. reload(stg)
  12. from vnpy.app.portfolio_strategy.strategies.pair_trading_strategy import PairTradingStrategy
  14. engine = BacktestingEngine()
  15. engine.set_parameters(
  16.     vt_symbols=["y888.DCE", "p888.DCE"],
  17.     interval=Interval.MINUTE,
  18.     start=datetime(2019, 1, 1),
  19.     end=datetime(2020, 4, 30),
  20.     rates={
  21.         "y888.DCE": 0/10000,
  22.         "p888.DCE": 0/10000
  23.     },
  24.     slippages={
  25.         "y888.DCE": 0,
  26.         "p888.DCE": 0
  27.     },
  28.     sizes={
  29.         "y888.DCE": 10,
  30.         "p888.DCE": 10
  31.     },
  32.     priceticks={
  33.         "y888.DCE": 1,
  34.         "p888.DCE": 1
  35.     },
  36.     capital=1_000_000,
  37. )
  39. setting = {
  40.     "boll_window": 20,
  41.     "boll_dev": 1,
  42. }
  43. engine.add_strategy(PairTradingStrategy, setting)
  45. engine.load_data()
  46. engine.run_backtesting()
  47. df = engine.calculate_result()
  48. engine.calculate_statistics()
  49. engine.show_chart()