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

﹏ヽ暗。殇╰゛Y 2023-01-22 13:54 46阅读 0赞

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

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

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

代码包括这几个文件：

[\- BatchCTABacktesting.py：][VNPY 1]批量回测代码文件，在这个代码里面定义和下面个关联文件路径，默认路径都在一个文件夹。

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

{ 
      "MA2009.CZCE": { 
        "rate": 0.0001,
        "slippage": 1,
        "size": 10,
        "pricetick": 1
      },
      "rb2010.SHFE": { 
        "rate": 0.0001,
        "slippage": 1,
        "size": 10,
        "pricetick": 1
      }
    }

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

{ 
      "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：][VNPY 1]用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

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

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

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

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

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

![在这里插入图片描述][20210520100952275.png]

# 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(2019, 7, 1),
                            endDate=datetime(2020, 1, 1), 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(2019, 7, 1),
                              endDate=datetime(2020, 1, 1), 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()

来自 “ 开源世界 ” ，链接：[http://ym.baisou.ltd/post/577.html][http_ym.baisou.ltd_post_577.html]，如需转载，请注明出处，否则将追究法律责任。

[VNPY 1]: http://github.crmeb.net/u/defu
[20210520100952275.png]: /images/20221020/783c5832a9f44866955703144edb5524.png
[http_ym.baisou.ltd_post_577.html]: http://ym.baisou.ltd/post/577.html