Shulou(Shulou.com)06/02 Report--

This article will explain in detail how to write a kind of GeneticOptimizeStrategy for VNPY strategy genetic algorithm optimization, the content of the article is of high quality, so the editor shares it for you to do a reference. I hope you will have a certain understanding of the relevant knowledge after reading this article.

Wrote a class, GeneticOptimizeStrategy

Parameterlist dictionary; simplifies the problem of calling parameters and defining the corresponding random range. As long as the policy parameter name and the corresponding random range are defined in Parameterlist, the meta-ancestor of two parameters is a random number between two, call random.uniform (), the meta-ancestor of three parameters is the beginning and end and intermediate step, and call random.randrange (). If it is an array, it is randomly selected in the middle of the array.

Symbollist dictionary to maintain back-test varieties and data

The poptoExcel method outputs an Excel, including parameters and value; parameters that can be called directly. Merge the same items at the same time. The effect is as follows

The source code can also be found in my GitHub

-

Found that there was an error in multithreading.

CPickle.PicklingError: Can't pickle: attribute lookup builtin.instanceme

After searching, it is found that it is a multi-process problem in python2.7. Pool.map cannot bind methods in the class.

Put the evaluate method outside the class to make a static method binding, although it solves the pickle, but in the case of multithreading, there is an error in the correspondence between the name and value of the policy parameter.

The way the parameters are assigned is also changed, from

[value1,value2,value3...] Become [{key1:value1}, {key2,value2}, {key3,value3}...], so. It can meet the requirements of crossover and mutation. The algorithm of generating random DNA needs to be changed.

Def parameter_generate (self):''according to the set start value, end value and step Randomly generate policy parameters to be optimized:''parameter_list = [] for key, value in self.parameterlist.items (): if isinstance (value, tuple): if len (value) = = 3: parameter_list.append ({key:random.randrange (value [0], value [1]) Value [2])}) elif len (value) = = 2: parameter_list.append ({key:random.uniform (value [0], value [1])}) elif isinstance (value List): parameter_list.append ({key:random.choice (value)}) else: parameter_list.append ({key:value}) return parameter_list

Parameter_list is a format similar to vnpy optimize, but with enhancements. The strategy_avg return becomes [{key1:value1}, {key2,value2}, {key3,value3}...], and the list format containing dictionaries can also satisfy crossover and mutation methods.

In the evolutionary screening method object_func, use the following traversal to put [{key1:value1}, {key2,value2}, {key3,value3}.] Change back to {key1,value1,key2,value2...} so that you can enter the back test.

Setting = {} for item in range (len (strategy_avg)): setting.update (strategy_ avg [item])

Finally, because the evolutionary screening method object_func is placed outside the class, but it is necessary to input some backtest parameters, such as variety, date, etc., there are two ways to achieve this. One is to change the incoming individual list to Tuple list, to [(individual, parameterPackage)..] List like this, but you have to modify algorithms.eaMuPlusLambda, which is more troublesome.

Another is to enhance individual by adding this set of backtest parameters as attributes. However, multithreading should also be noted that you have to put parameterPackage in the class as a static property, otherwise you will be prompted that parameterPackage is empty. We have not found a more suitable way to deal with it.

Creator.create ("Individual", list, fitness=creator.FitnessMulti, parameterPackage=parameterPackage)

The complete new code is as follows

# encoding: UTF-8 "" shows how to perform parameter optimization. " From _ _ future__ import divisionfrom _ future__ import print_functionfrom vnpy.trader.app.ctaStrategy.ctaBacktesting import BacktestingEngine, MINUTE_DB_NAME, OptimizationSettingfrom vnpy.trader.app.ctaStrategy.strategy.strategyBollChannel import BollChannelStrategyimport randomimport numpy as npfrom deap import creator, base, tools, algorithmsimport multiprocessingimport time, datetimeimport pandas as pddef object_func (strategy_avgTuple): "" this function is the optimization objective function, according to the randomly generated policy parameters Two result metrics are automatically returned after running the backtest: payback ratio and Sharp ratio "" strategy_avg = strategy_avgTuple paraSet = strategy_avgTuple.parameterPackage symbol = paraSet ["symbol"] strategy = paraSet ["strategy"] # create the backtest engine object engine = BacktestingEngine () # set the data used in the backtest engine.setBacktestingMode (engine.BAR_MODE) # set the engine's return The test mode is K-line engine.setDatabase ("VnTrader_1Min_Db") Symbol ["vtSymbol"]) # set the historical database engine.setStartDate (symbol ["StartDate"]) # set the data start date for the back test engine.setEndDate (symbol ["EndDate"]) # set the data start date for the back test # configure the backtest engine parameter engine.setSlippage (symbol ["Slippage"]) # 1 hop engine.setRate (symbol ["Rate"]) # Commission size engine.setSize (symbol ["Size"]) # contract size engine.setPriceTick (symbol ["Slippage"]) # minimum price change engine.setCapital (symbol ["Capital"]) setting = {} for item in range (len (strategy_avg)): setting.update (strategy_ [item]) engine.clearBacktestingResult () # loading strategy engine.initStrategy (strategy Setting) # run the backtest Return the specified result metric engine.runBacktesting () # run backtest # Daily backtest # engine.calculateDailyResult () backresult = engine.calculateBacktestingResult () try: capital = round (backresult ['capital'], 3) # Revenue withdrawal ratio profitLossRatio = round (backresult [' profitLossRatio'], 3) # Sharp ratio # Sharp ratio sharpeRatio = round (backresult ['sharpeRatio']) 3) except Exception, e: print ("Error:% s,% s"% (str (Exception), str (e)) sharpeRatio = 0 profitLossRatio = 0 # earnings withdrawal ratio averageWinning = 0 # Sharp ratio # Sharp ratio capital = 0 return capital, sharpeRatio ProfitLossRatioclass GeneticOptimizeStrategy (object): Strategy = BollChannelStrategy Symbollist = {"vtSymbol": 'rb0000', "StartDate": "20140601", "EndDate": "20141101", "Slippage": 1, "Size": 10, "Rate": 2 / 10000.0 "Capital": 10000} Parameterlist = {'bollWindow':, # Boolean band window' bollDev':, # Bollinger band channel threshold 'slMultiplier':,' barMins': } parameterPackage = {"symbol": Symbollist, "strategy": Strategy} #-def _ init__ (self, Strategy, Symbollist Parameterlist): self.strategy = Strategy self.symbol = Symbollist self.parameterlist = Parameterlist self.parameterPackage = {"strategy": self.strategy, "symbol": self.symbol} creator.create ("FitnessMulti", base.Fitness, weights= (1.0,1.0,1.0)) # 1.0 to find the maximum -1.0 find the minimum creator.create ("Individual", list, fitness=creator.FitnessMulti, parameterPackage=parameterPackage) #-def parameter_generate (self):''according to the starting value of the setting Termination value and step Randomly generate policy parameters to be optimized:''parameter_list = [] for key, value in self.parameterlist.items (): if isinstance (value, tuple): if len (value) = = 3: parameter_list.append ({key:random.randrange (value [0], value [1]) Value [2])}) elif len (value) = = 2: parameter_list.append ({key:random.uniform (value [0], value [1])}) elif isinstance (value List): parameter_list.append ({key:random.choice (value)}) else: parameter_list.append ({key:value}) return parameter_list def mutArrayGroup (self, individual, parameterlist, indpb): size = len (individual) paralist = parameterlist () for i in xrange (size): if random.random ()

< indpb: individual[i] = paralist[i] return individual, def optimize(self): # 设置优化方向：最大化收益回撤比，最大化夏普比率 toolbox = base.Toolbox() # Toolbox是deap库内置的工具箱，里面包含遗传算法中所用到的各种函数 pool = multiprocessing.Pool(processes=(multiprocessing.cpu_count()-1)) toolbox.register("map", pool.map) # 初始化 toolbox.register("individual", tools.initIterate, creator.Individual, self.parameter_generate) # 注册个体：随机生成的策略参数parameter_generate() toolbox.register("population", tools.initRepeat, list, toolbox.individual) # 注册种群：个体形成种群 toolbox.register("mate", tools.cxTwoPoint) # 注册交叉：两点交叉 toolbox.register("mutate", self.mutArrayGroup, parameterlist=self.parameter_generate, indpb=0.6) # 注册变异：随机生成一定区间内的整数 toolbox.register("evaluate", object_func) # 注册评估：优化目标函数object_func() toolbox.register("select", tools.selNSGA2) # 注册选择:NSGA-II(带精英策略的非支配排序的遗传算法) # 遗传算法参数设置 MU = 8 # 设置每一代选择的个体数 LAMBDA = 5 # 设置每一代产生的子女数 pop = toolbox.population(20) # 设置族群里面的个体数量 CXPB, MUTPB, NGEN = 0.5, 0.3, 10 # 分别为种群内部个体的交叉概率、变异概率、产生种群代数 hof = tools.ParetoFront() # 解的集合：帕累托前沿(非占优最优集) # 解的集合的描述统计信息 # 集合内平均值，标准差，最小值，最大值可以体现集合的收敛程度 # 收敛程度低可以增加算法的迭代次数 stats = tools.Statistics(lambda ind: ind.fitness.values) np.set_printoptions(suppress=True) # 对numpy默认输出的科学计数法转换 stats.register("mean", np.mean, axis=0) # 统计目标优化函数结果的平均值 stats.register("std", np.std, axis=0) # 统计目标优化函数结果的标准差 stats.register("min", np.min, axis=0) # 统计目标优化函数结果的最小值 stats.register("max", np.max, axis=0) # 统计目标优化函数结果的最大值 # 运行算法 algorithms.eaMuPlusLambda(pop, toolbox, MU, LAMBDA, CXPB, MUTPB, NGEN, stats, halloffame=hof, verbose=True) # esMuPlusLambda是一种基于(μ+λ)选择策略的多目标优化分段遗传算法 return pop def poptoExcel(self, pop, number = 1000, path = "C:/data/"): #按照输入统计数据队列和路径，输出excel，这里不提供新增模式，如果想，可以改 #dft.to_csv(path,index=False,header=True, mode = 'a') path = path + self.strategy.className + "_" + self.symbol[ "vtSymbol"] + str(datetime.date.today())+ ".xls" summayKey = ["StrategyParameter","TestValues"] best_ind = tools.selBest(pop, number) dft = pd.DataFrame(columns=summayKey) for i in range(0,len(best_ind)-1): if i == 0: # new = pd.DataFrame([{"StrategyParameter":self.complieString(best_ind[i])},{"TestValues":best_ind[i].fitness.values}], index=["0"]) dft = dft.append([{"StrategyParameter":self.complieString(best_ind[i]),"TestValues":best_ind[i].fitness.values}], ignore_index=True) elif str(best_ind[i-1]) == (str(best_ind[i])): pass else: #new = pd.DataFrame({"StrategyParameter":self.complieString(best_ind[i]),"TestValues":best_ind[i].fitness.values}, index=["0"]) dft = dft.append([{"StrategyParameter":self.complieString(best_ind[i]),"TestValues":best_ind[i].fitness.values}], ignore_index=True) dft.to_excel(path,index=False,header=True) print("回测统计结果输出到" + path) def complieString(self,individual): setting = {} for item in range(len(individual)): setting.update(individual[item]) return str(setting)if __name__ == "__main__": Strategy = BollChannelStrategy Symbollist ={ "vtSymbol": 'rb0000', "StartDate": "20140601", "EndDate": "20141101", "Slippage": 1, "Size": 10, "Rate": 2 / 10000.0, "Capital": 10000 } Parameterlist = { 'bollWindow': (10,50,1), #布林带窗口 'bollDev': (2,10,1), #布林带通道阈值 'slMultiplier':(3,6), 'barMins':[2,3,5,10,15,20], } parameterPackage = { "symbol":Symbollist, "parameterlist":Parameterlist, "strategy":Strategy } GE = GeneticOptimizeStrategy(Strategy,Symbollist,Parameterlist) GE.poptoExcel(GE.optimize()) print("-- End of (successful) evolution --") -------------------------------------------------------------------------------------------------------------

On the genetic algorithm for VNPY strategy optimization how to write a class GeneticOptimizeStrategy to share here, I hope the above content can be of some help to everyone, can learn more knowledge. If you think the article is good, you can share it for more people to see.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.