参数优化

vnpy提供2种参数优化的解决方案：穷举算法、遗传算法

穷举算法

穷举算法原理：

  • 输入需要优化的参数名、优化区间、优化步进，以及优化目标。
  1.     def add_parameter(
  2.         self, name: str, start: float, end: float = None, step: float = None
  3.     ):
  4.         """"""
  5.         if not end and not step:
  6.             self.params[name] = [start]
  7.             return
  8.  
  9.         if start >= end:
  10.             print("参数优化起始点必须小于终止点")
  11.             return
  12.  
  13.         if step <= 0:
  14.             print("参数优化步进必须大于0")
  15.             return
  16.  
  17.         value = start
  18.         value_list = []
  19.  
  20.         while value <= end:
  21.             value_list.append(value)
  22.             value += step
  23.  
  24.         self.params[name] = value_list
  25.  
  26.     def set_target(self, target_name: str):
  27.         """"""
  28.         self.target_name = target_name
  • 形成全局参数组合, 数据结构为[{key: value, key: value}, {key: value, key: value}]。
  1.     def generate_setting(self):
  2.         """"""
  3.         keys = self.params.keys()
  4.         values = self.params.values()
  5.         products = list(product(*values))
  6.  
  7.         settings = []
  8.         for p in products:
  9.             setting = dict(zip(keys, p))
  10.             settings.append(setting)
  11.  
  12.         return settings
  • 遍历全局中的每一个参数组合：遍历的过程即运行一次策略回测，并且返回优化目标数值；然后根据目标数值排序，输出优化结果。
  1.     def run_optimization(self, optimization_setting: OptimizationSetting, output=True):
  2.         """"""
  3.         # Get optimization setting and target
  4.         settings = optimization_setting.generate_setting()
  5.         target_name = optimization_setting.target_name
  6.  
  7.         if not settings:
  8.             self.output("优化参数组合为空，请检查")
  9.             return
  10.  
  11.         if not target_name:
  12.             self.output("优化目标未设置，请检查")
  13.             return
  14.  
  15.         # Use multiprocessing pool for running backtesting with different setting
  16.         pool = multiprocessing.Pool(multiprocessing.cpu_count())
  17.  
  18.         results = []
  19.         for setting in settings:
  20.             result = (pool.apply_async(optimize, (
  21.                 target_name,
  22.                 self.strategy_class,
  23.                 setting,
  24.                 self.vt_symbol,
  25.                 self.interval,
  26.                 self.start,
  27.                 self.rate,
  28.                 self.slippage,
  29.                 self.size,
  30.                 self.pricetick,
  31.                 self.capital,
  32.                 self.end,
  33.                 self.mode
  34.             )))
  35.             results.append(result)
  36.  
  37.         pool.close()
  38.         pool.join()
  39.  
  40.         # Sort results and output
  41.         result_values = [result.get() for result in results]
  42.         result_values.sort(reverse=True, key=lambda result: result[1])
  43.  
  44.         if output:
  45.             for value in result_values:
  46.                 msg = f"参数：{value[0]}, 目标：{value[1]}"
  47.                 self.output(msg)
  48.  
  49.         return result_values

注意：可以使用multiprocessing库来创建多进程实现并行优化。例如：若用户计算机是2核，优化时间为原来1/2；若计算机是10核，优化时间为原来1/10。

穷举算法操作：

  • 点击“参数优化”按钮，会弹出“优化参数配置”窗口，用于设置优化目标（如最大化夏普比率、最大化收益回撤比）和设置需要优化的参数以及优化区间，如图。

https://vnpy-community.oss-cn-shanghai.aliyuncs.com/forum_experience/yazhang/cta_backtester/optimize_setting.png

  • 设置好需要优化的参数后，点击“优化参数配置”窗口下方的“确认”按钮开始进行调用CPU多核进行多进程并行优化，同时日志会输出相关信息。

https://vnpy-community.oss-cn-shanghai.aliyuncs.com/forum_experience/yazhang/cta_backtester/optimize_log.png

  • 点击“优化结果”按钮可以看出优化结果，如图的参数组合是基于目标数值（夏普比率）由高到低的顺序排列的。

https://vnpy-community.oss-cn-shanghai.aliyuncs.com/forum_experience/yazhang/cta_backtester/optimize_result.png

遗传算法

遗传算法原理：

  • 输入需要优化的参数名、优化区间、优化步进，以及优化目标；

  • 形成全局参数组合，该组合的数据结构是列表内镶嵌元组，即[[(key, value), (key, value)] , [(key, value), (key,value)]]，与穷举算法的全局参数组合的数据结构不同。这样做的目的是有利于参数间进行交叉互换和变异。

  1.     def generate_setting_ga(self):
  2.         """""" 
  3.         settings_ga = []
  4.         settings = self.generate_setting()     
  5.         for d in settings:            
  6.             param = [tuple(i) for i in d.items()]
  7.             settings_ga.append(param)
  8.         return settings_ga
  • 形成个体：调用random()函数随机从全局参数组合中获取参数。
  1.         def generate_parameter():
  2.             """"""
  3.             return random.choice(settings)
  • 定义个体变异规则: 即发生变异时，旧的个体完全被新的个体替代。
  1.         def mutate_individual(individual, indpb):
  2.             """"""
  3.             size = len(individual)
  4.             paramlist = generate_parameter()
  5.             for i in range(size):
  6.                 if random.random() < indpb:
  7.                     individual[i] = paramlist[i]
  8.             return individual,
  • 定义评估函数：入参的是个体，即[(key, value), (key, value)]形式的参数组合，然后通过dict()转化成setting字典，然后运行回测，输出目标优化数值，如夏普比率、收益回撤比。(注意，修饰器@lru_cache作用是缓存计算结果，避免遇到相同的输入重复计算，大大降低运行遗传算法的时间)
  1. @lru_cache(maxsize=1000000)
  2. def _ga_optimize(parameter_values: tuple):
  3.     """"""
  4.     setting = dict(parameter_values)
  5.  
  6.     result = optimize(
  7.         ga_target_name,
  8.         ga_strategy_class,
  9.         setting,
  10.         ga_vt_symbol,
  11.         ga_interval,
  12.         ga_start,
  13.         ga_rate,
  14.         ga_slippage,
  15.         ga_size,
  16.         ga_pricetick,
  17.         ga_capital,
  18.         ga_end,
  19.         ga_mode
  20.     )
  21.     return (result[1],)
  22.  
  23. def ga_optimize(parameter_values: list):
  24.     """"""
  25.     return _ga_optimize(tuple(parameter_values))
  • 运行遗传算法：调用deap库的算法引擎来运行遗传算法，其具体流程如下。1）先定义优化方向，如夏普比率最大化；2）然后随机从全局参数组合获取个体，并形成族群；3）对族群内所有个体进行评估（即运行回测），并且剔除表现不好个体；4）剩下的个体会进行交叉或者变异，通过评估和筛选后形成新的族群；（到此为止是完整的一次种群迭代过程）；5）多次迭代后，种群内差异性减少，整体适应性提高，最终输出建议结果。该结果为帕累托解集，可以是1个或者多个参数组合。

注意：由于用到了@lru_cache, 迭代中后期的速度回提高非常多，因为很多重复的输入都避免了再次的回测，直接在内存中查询并且返回计算结果。

  1. from deap import creator, base, tools, algorithms
  2. creator.create("FitnessMax", base.Fitness, weights=(1.0,))
  3. creator.create("Individual", list, fitness=creator.FitnessMax)
  4.         ......
  5.         # Set up genetic algorithem
  6.         toolbox = base.Toolbox() 
  7.         toolbox.register("individual", tools.initIterate, creator.Individual, generate_parameter)                          
  8.         toolbox.register("population", tools.initRepeat, list, toolbox.individual)                                            
  9.         toolbox.register("mate", tools.cxTwoPoint)                                               
  10.         toolbox.register("mutate", mutate_individual, indpb=1)               
  11.         toolbox.register("evaluate", ga_optimize)                                                
  12.         toolbox.register("select", tools.selNSGA2)       
  13.  
  14.         total_size = len(settings)
  15.         pop_size = population_size                      # number of individuals in each generation
  16.         lambda_ = pop_size                              # number of children to produce at each generation
  17.         mu = int(pop_size * 0.8)                        # number of individuals to select for the next generation
  18.  
  19.         cxpb = 0.95         # probability that an offspring is produced by crossover    
  20.         mutpb = 1 - cxpb    # probability that an offspring is produced by mutation
  21.         ngen = ngen_size    # number of generation
  22.  
  23.         pop = toolbox.population(pop_size)      
  24.         hof = tools.ParetoFront()               # end result of pareto front
  25.  
  26.         stats = tools.Statistics(lambda ind: ind.fitness.values)
  27.         np.set_printoptions(suppress=True)
  28.         stats.register("mean", np.mean, axis=0)
  29.         stats.register("std", np.std, axis=0)
  30.         stats.register("min", np.min, axis=0)
  31.         stats.register("max", np.max, axis=0)
  32.  
  33.         algorithms.eaMuPlusLambda(
  34.             pop, 
  35.             toolbox, 
  36.             mu, 
  37.             lambda_, 
  38.             cxpb, 
  39.             mutpb, 
  40.             ngen, 
  41.             stats,
  42.             halloffame=hof
  43.         )
  44.  
  45.         # Return result list
  46.         results = []
  47.  
  48.         for parameter_values in hof:
  49.             setting = dict(parameter_values)
  50.             target_value = ga_optimize(parameter_values)[0]
  51.             results.append((setting, target_value, {}))
  52.  
  53.         return results