VNPY中,優(yōu)化參數(shù)也經(jīng)常要批量去做,一個是一組不同策略批量對一個品種優(yōu)化,還有一個策略對應(yīng)不同憑證,下面是源代碼,放在example\CtaBacktesting文件夾下面,主要是參考了原來的優(yōu)化代碼。
# encoding: UTF-8
import pandas as pd
from vnpy.trader.app.ctaStrategy.ctaBacktesting import BacktestingEngine, MINUTE_DB_NAME, OptimizationSetting
from vnpy.trader.app.ctaStrategy.strategy.strategyBollChannel import BollChannelStrategy
class BatchOptimization(object):
def __init__(self):
""
def calculateBacktesting(self,symbollist,strategylist, sort = 'totalNetPnl'):
#填入品種隊列和策略隊列,返回結(jié)果resultlist, 為了輸出方便檢索,加入品種名稱,策略名稱和策略參數(shù)
resultlist = []
for symbol in symbollist:
for strategy in strategylist:
result = self.runBacktesting(symbol,strategy,sort)
#加入品種名稱,策略名稱和策略參數(shù)
if isinstance(result,dict):
#如果返回的是dict,直接加入
result["Symbolname"] = str(symbol["vtSymbol"])
result["strategyname"] = str(strategy[0])
result["strategysetting"] = str(strategy[1])
resultlist.append(result)
else:
# 發(fā)現(xiàn)優(yōu)化回來的是一個包含元組的隊列,元組有三個組成,第一個排策略參數(shù),第二個回測目標(biāo)的值,第三策略參數(shù)全部運(yùn)行結(jié)果。
# 這里我們要的就是第三個,先插入這個dict,在dict插入symbolname,和strategysetting
for resultraw in result:
resultlist.append(resultraw[2])
resultlist[-1]["Symbolname"] = str(symbol["vtSymbol"])
resultlist[-1]["strategysetting"] = str(resultraw[0])
return resultlist
def runBacktesting(self, symbol, strategy, sort):
#寫入測試品種和參數(shù), 返回回測數(shù)據(jù)集包含回測結(jié)果
# 在引擎中創(chuàng)建策略對象
# 創(chuàng)建回測引擎
engine = BacktestingEngine()
# 設(shè)置引擎的回測模式為K線
engine.setBacktestingMode(engine.BAR_MODE)
# 設(shè)置回測用的數(shù)據(jù)起始日期
engine.setStartDate(symbol["StartDate"])
engine.setSlippage(symbol["Slippage"]) # 1跳
engine.setRate(symbol["Rate"]) # 傭金大小
engine.setSize(symbol["Size"]) # 合約大小
engine.setPriceTick(symbol["Slippage"]) # 最小價格變動
engine.setCapital(symbol["Capital"])
# 設(shè)置使用的歷史數(shù)據(jù)庫
engine.setDatabase(MINUTE_DB_NAME, symbol["vtSymbol"])
#調(diào)用優(yōu)化方法,可以集成優(yōu)化測試
setting = OptimizationSetting() # 新建一個優(yōu)化任務(wù)設(shè)置對象
setting.setOptimizeTarget(sort) # 設(shè)置優(yōu)化排序的目標(biāo)是策略凈盈利
print strategy[1]
for settingKey in strategy[1]:
if isinstance(strategy[1][settingKey], tuple):
setting.addParameter(settingKey,strategy[1][settingKey][0],strategy[1][settingKey][1],strategy[1][settingKey][2])
else:
setting.addParameter(settingKey,strategy[1][settingKey])
#
optimizationresult = engine.runParallelOptimization(strategy[0], setting)
engine.output(u'輸出統(tǒng)計數(shù)據(jù)')
# 如果是使用優(yōu)化模式,這里返回的是策略回測的dict的list,如果普通回測就是單個dict
# 如果大于30 ,就返回三十之內(nèi),否則全部
if len(optimizationresult) > 30:
return optimizationresult[:30]
else:
return optimizationresult
def toExcel(self, resultlist, path = "C:\data\datframe.xlsx"):
#按照輸入統(tǒng)計數(shù)據(jù)隊列和路徑,輸出excel,這里不提供新增模式,如果想,可以改
#dft.to_csv(path,index=False,header=True, mode = 'a')
summayKey = resultlist[0].keys()
# summayValue = result.values()
dft = pd.DataFrame(columns=summayKey)
for result_ in resultlist:
new = pd.DataFrame(result_, index=["0"])
dft = dft.append(new,ignore_index=True)
dft.to_excel(path,index=False,header=True)
print "回測統(tǒng)計結(jié)果輸出到" + path
if __name__ == "__main__":
#創(chuàng)建品種隊列,這里可以用json導(dǎo)入,為了方便使用直接寫了。
symbollist = [{
"vtSymbol": 'm1809',
"StartDate": "20180101",
"Slippage": 1,
"Size": 10,
"Rate": 2 / 10000,
"Capital": 10000
},
{
"vtSymbol": 'rb1810',
"StartDate": "20180101",
"Slippage": 1,
"Size": 10,
"Rate": 2 / 10000,
"Capital": 10000
}
]
# 這里是同一個策略,不同參數(shù)的情況,當(dāng)然可以有多個策略和多個參數(shù)組合
Strategylist2 = []
# 策略list,如果是元組,那么就是三個,按照第一個初始,第二個結(jié)束,第三個步進(jìn)
settinglist =[
{'bollWindow': (10,20,2)}]
# 合并一個元組
if settinglist != []:
for para1 in settinglist:
Strategylist2.append((BollChannelStrategy, para1))
NT = BatchOptimization()
resultlist = NT.calculateBacktesting(symbollist,Strategylist2,sort = 'totalNetPnl')
#定義路徑
path = "C:\Project\OptimizationResult.xlsx"
NT.toExcel(resultlist,path)