之前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 现在回来看看代码。其实注释都比较清楚了。注意的几点是
策略类是用字符串格式记录的，然后用eval方法关联类，所以必须引用，虽然编辑器提示未使用
在excel保存setting必须双引号，因为json文件默认只能识别双引号。
批量回测结果会用excel输出，示例就是这样。
默认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 hasnt 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() 最后可以去我的Github下载代码，比较方便
https://github.com/BillyZhangGuoping/VNPY2_BILLY