首页 > Web开发 > 详细

在VNPY2的进行CTA批量回测,支持Json和Excel格式导入策略

时间:2021-09-08 22:39:38      阅读:38      评论:0      收藏:0      [点我收藏+]

之前VNPY 1版本中,我的个人代码很多是直接在VNPY库代码直接修改或者增加的。每次VNPY升级就是非常头疼,要做代码对比,在一些可能被更新覆盖的地方再次维护测试。而且因为更新的地方很乱,造成后面生产版本一致停留在VNPY1.92。

 

这次准备不在VNPY的库文件代码上修改,而是像引用NUMPY或者Pandas这样,采用调用继承的方式,把自己的代码和VNPY的库代码隔离;这样即使VNPY升级,个人代码不用太担心,只要简单测试,保证继承引用VNPY的类或方法正常工作就可以了。

 

也是之前VNPY 1版本实现的功能,批量回测,结果Excel导出。这次支持策略参数用Json或Excel导入,同时支持多个策略的组合portfolio收益计算;其实都是VNPY2提供好的,调用而已。只要VNPY2.0 正确安装,历史数据存在,这些代码就可以运行。

 

代码包括这几个文件:

- BatchCTABacktesting.py:批量回测代码文件,在这个代码里面定义和下面个关联文件路径,默认路径都在一个文件夹。

- vtSymbol.json:这个是定义品种交易属性,回测时候从vtSymbol.json文档读取品种的交易属性,比如费率,交易每跳,比率,滑点;这样不用在回测时候维护。示例格式如下;有心的可以改成通配符,这样减少维护量。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
{
  "MA2009.CZCE": {
    "rate"0.0001,
    "slippage"1,
    "size"10,
    "pricetick"1
  },
  "rb2010.SHFE": {
    "rate"0.0001,
    "slippage"1,
    "size"10,
    "pricetick"1
  }
}

- ctaStrategy.json:定义要批量回测策略,其实和VNPY2默认的CTA策略文件是一样的,这样就可以直接用实盘CTA策略文件进行批量回测了,或着计算组合收益。示例格式如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
{
  "BollChannelStrategy_MA8888.CZCE": {
    "class_name""BollChannelStrategy",
    "vt_symbol""MA8888.CZCE",
    "setting": {
      "boll_window"40,
      "boll_dev"3
    }
  },
  "DoubleMaStrategy2_CTA_rb8888.SHFE": {
    "class_name""DoubleMaStrategy",
    "vt_symbol""rb8888.SHFE",
    "setting": {
      "fast_window"10,
      "slow_window"40
    }
}

- ctaStrategy.xls:用xls格式定义的批量回测数据,示例格式如下;有三列, class_name是策略类, setting是参数,v t_symbol是品种。主要是有时候用excel做策略批量维护或者生成,然后就可以直接批量回测了。

class_name setting vt_symbol
AtrRsiStrategy {"atr_length": 10, "atr_ma_length": 50} MA8888.CZCE
DoubleMaStrategy {"fast_window": 10, "slow_window": 40} rb8888.SHFE

 

现在回来看看代码。其实注释都比较清楚了。注意的几点是

  1.  策略类是用字符串格式记录的,然后用eval方法关联类,所以必须引用,虽然编辑器提示未使用

  2. 在excel保存setting必须双引号,因为json文件默认只能识别双引号。

  3. 批量回测结果会用excel输出,示例就是这样。

  4. 默认json导入会计算组合收入,excel不会计算组合收益,可以直接修改代码。

     

技术分享图片

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
# encoding: UTF-8
import json
import traceback
from datetime import datetime, date
import pandas as pd
from pandas import DataFrame
from vnpy.app.cta_strategy.backtesting import BacktestingEngine
# 策略类是用字符串格式记录的,然后用eval方法关联类,所以必须引用,虽然编辑器提示未使用
from vnpy.app.cta_strategy.strategies.boll_channel_strategy import BollChannelStrategy
from vnpy.app.cta_strategy.strategies.turtle_signal_strategy import TurtleSignalStrategy
from vnpy.app.cta_strategy.strategies.double_ma_strategy import DoubleMaStrategy
class BatchCTABackTest:
   """
   提供批量CTA策略回测,输出结果到excel或pdf,和CTA策略批量优化,输出结果到excel或pdf,
   """
   def __init__(self, vtSymbolconfig="vtSymbol.json", exportpath=".\\"):
      """
      加载配置路径
      """
      config = open(vtSymbolconfig)
      self.setting = json.load(config)
      self.exportpath = exportpath
   def addParameters(self, engine, vt_symbol: str, startDate, endDate, interval="1m", capital=1_000_000):
      """
      从vtSymbol.json文档读取品种的交易属性,比如费率,交易每跳,比率,滑点
      """
      if vt_symbol in self.setting:
         engine.set_parameters(
            vt_symbol=vt_symbol,
            interval=interval,
            start=startDate,
            end=endDate,
            rate=self.setting[vt_symbol]["rate"],
            slippage=self.setting[vt_symbol]["slippage"],
            size=self.setting[vt_symbol]["size"],
            pricetick=self.setting[vt_symbol]["pricetick"],
            capital=capital
         )
      else:
         print("symbol %s hasn‘t be maintained in config file" % vt_symbol)
      return engine
   def runBatchTest(self, strategy_setting, startDate, endDate, portfolio):
      """
      进行回测
      """
      resultDf = DataFrame()
      dfportfolio = None
      for strategy_name, strategy_config in strategy_setting.items():
         engine = BacktestingEngine()
         vt_symbol = strategy_config["vt_symbol"]
         engine = self.addParameters(engine, vt_symbol, startDate, endDate)
         if type(strategy_config["setting"]) is str:
            print(strategy_config["setting"])
            engine.add_strategy(
               eval(strategy_config["class_name"]),
               json.loads(strategy_config["setting"], )
            )
         else:
            engine.add_strategy(
               eval(strategy_config["class_name"]),
               strategy_config["setting"]
            )
         engine.load_data()
         engine.run_backtesting()
         df = engine.calculate_result()
         if portfolio == True:
            if dfportfolio is None:
               dfportfolio = df
            else:
               dfportfolio = dfportfolio + df
         resultDict = engine.calculate_statistics(df, False)
         resultDict["class_name"= strategy_config["class_name"]
         resultDict["setting"= strategy_config["setting"]
         resultDict["vt_symbol"= strategy_config["vt_symbol"]
         resultDf = resultDf.append(resultDict, ignore_index=True)
      if portfolio == True:
         # dfportfolio = dfportfolio.dropna()
         engine = BacktestingEngine()
         engine.calculate_statistics(dfportfolio)
         engine.show_chart(dfportfolio)
      return resultDf
   def runBatchTestJson(self, jsonpath="ctaStrategy.json", startDate=datetime(201971),
                        endDate=datetime(202011), exporpath=None, portfolio=True):
      """
      从ctaStrategy.json去读交易策略和参数,进行回测
      """
      with open(jsonpath, mode="r", encoding="UTF-8") as f:
         strategy_setting = json.load(f)
      resultDf = self.runBatchTest(strategy_setting, startDate, endDate, portfolio)
      self.ResultExcel(resultDf, exporpath)
      return strategy_setting
   def runBatchTestExcecl(self, path="ctaStrategy.xls", startDate=datetime(201971),
                          endDate=datetime(202011), exporpath=None, portfolio=False):
      """
      从ctaStrategy.excel去读交易策略和参数,进行回测
      """
      df = pd.read_excel(path)
      strategy_setting = df.to_dict(orient=‘index‘)
      resultDf = self.runBatchTest(strategy_setting, startDate, endDate, portfolio)
      self.ResultExcel(resultDf, exporpath)
      return strategy_setting
   def ResultExcel(self, result, export=None):
      """
      输出交易结果到excel
      """
      if export != None:
         exportpath = export
      else:
         exportpath = self.exportpath
      try:
         path = exportpath + "CTABatch" + str(date.today()) + "v0.xls"
         result.to_excel(path, index=False)
         print("CTA Batch result is export to %s" % path)
      except:
         print(traceback.format_exc())
      return None
if __name__ == ‘__main__‘:
   bts = BatchCTABackTest()
   bts.runBatchTestJson()

 

最后可以去我的Github下载代码,比较方便

在VNPY2的进行CTA批量回测,支持Json和Excel格式导入策略

原文:https://www.cnblogs.com/chenguopa/p/15239822.html

(0)
(0)
   
举报
评论 一句话评论(0
关于我们 - 联系我们 - 留言反馈 - 联系我们:wmxa8@hotmail.com
© 2014 bubuko.com 版权所有
打开技术之扣,分享程序人生!