之前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文档读取品种的交易属性，比如费率，交易每跳，比率，滑点；这样不用在回测时候维护。示例格式如下；有心的可以改成通配符，这样减少维护量。
{
"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策略文件进行批量回测了，或着计算组合收益。示例格式如下：
{
"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是参数，vt_symbol是品种。主要是有时候用excel做策略批量维护或者生成，然后就可以直接批量回测了。
现在回来看看代码。其实注释都比较清楚了。注意的几点是
策略类是用字符串格式记录的，然后用eval方法关联类，所以必须引用，虽然编辑器提示未使用
在excel保存setting必须双引号，因为json文件默认只能识别双引号。
批量回测结果会用excel输出，示例就是这样。
- 默认json导入会计算组合收入，excel不会计算组合收益，可以直接修改代码。
# 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()
最后可以去我的Github下载代码，比较方便
