PYGAD并行处理异常：每代从多进程切换至单进程，求原因与解决方案

我正在使用PYGAD优化量化回测策略的参数，目前遇到一个问题：算法每一代都会从多进程模式切换为单进程模式。此前该功能运行完全正常，在将mutation_type改为'adaptive'以尝试提升适应度后，似乎就干扰了并行处理。此外，我在本地运行遗传算法的同时，还在AWS实例上运行其他代码，但我认为这不会影响本地进程。我的设备是搭载M1 Max芯片（10核：8性能核+2能效核）的MacBook Pro，求原因或解决方案。

相关代码 #

def fitness_func(ga_instance, solution, solution_idx):
    # Transform the solution array into a dictionary of parameters
    try:
        
        params = {
            'rsi_periods' : int(solution[0]),
            'macd1': int(solution[1]),
            'macd2': int(solution[2]),
            'macdsig': int(solution[3]),
            'rsi_high': int(solution[4]),
            'stop_loss' : float(solution[5]),
            'take_profit' : float(solution[6]),
            'rsi_low': int(solution[7]),
            'rsi_mid' : int(solution[8])
        }

        # Create a Backtrader instance and add your strategy with the given parameters
        cerebro = bt.Cerebro()
        cerebro.broker.setcommission(commission=0.0075)
        cerebro.addstrategy(RsiMacdStrategy, **params)
        cerebro.broker.setcash(2500.0)
        data = get_data('DOT')
        cerebro.adddata(data)
        # Run the backtest and get the final portfolio value
        results = cerebro.run()
        final_value = results[0].broker.get_value()
        global finished_counter
        finished_counter += 1
        print(f"Finished {finished_counter}")
        # The final portfolio value is used as the fitness score for the genetic algorithm
        return final_value
    except Exception as e:
        print(f"Error in fitness function: {e}")
        raise e


def on_generation(ga_instance):
    # This function is called after each generation of the genetic algorithm
    # You can use it to track progress or print debug information
    try: 
        # Get the best solution from the current generation
        best_solution, best_solution_fitness, best_solution_idx = ga_instance.best_solution()

        # Print the generation number, best fitness score, and best solution
        print(f"Generation: {ga_instance.generations_completed}, Best fitness: {best_solution_fitness}, Best solution: {best_solution}")

        # Write the best solution and fitness score of the current generation to the CSV file
        with open('/Users/.../Desktop/.../Coding/Sentiment Analysis/code/backTesting/results/dotPYGAD.csv', 'a') as f:
            writer = csv.writer(f)
            writer.writerow(list(best_solution) + [best_solution_fitness])

    except Exception as e:
        print(f"Error in on_generation function: {e}")
        raise e


def run_backtest():
    # Define the parameters for your genetic algorithm here
    # Each item in param_types and param_ranges corresponds to a parameter of RsiMacdStrategy
    param_types = [int, int, int, int, int, float, float, int, int]
    param_ranges = [{"low" : 5, "high" : 35}, {"low": 5, "high": 25}, {"low": 10, "high": 45}, {"low": 4, "high": 12}, {"low": 65, "high": 90},
                    {"low": .04, "high" : .1}, {"low": .075, "high" : .2}, {"low" : 35, "high" : 49}, {"low" : 51, "high" : 65}]

    # Initialize the genetic algorithm with the parameters and fitness function
    ga_instance = pygad.GA(num_generations=15,
                            num_parents_mating=30,
                            fitness_func=fitness_func,
                            sol_per_pop=100,
                            num_genes=len(param_types),
                            gene_type=param_types,
                            gene_space=param_ranges,
                            parent_selection_type="rank",
                            crossover_type="single_point",
                            mutation_type="adaptive",
                            mutation_probability=[.6, .2],
                            keep_parents=1,
                            parallel_processing=["process", 8],
                            suppress_warnings=True,
                            keep_elitism=1,
                            on_generation=on_generation)  # Call on_generation after each generation

    # Run the genetic algorithm
    ga_instance.run()
    solution, solution_fitness, solution_idx = ga_instance.best_solution()
    print(f"Best solution: {solution}, Fitness: {solution_fitness}")

可能的解决方案 #

• 版本兼容性检查：自适应突变的实现可能与部分PYGAD版本的多进程逻辑冲突，建议升级到最新稳定版，或回退至之前能正常运行的版本验证问题是否消失。
• 调整自适应突变参数：当前mutation_probability=[.6, .2]的设置可能触发了单进程降级逻辑，尝试改为单值（如mutation_probability=0.4）或调整范围，观察多进程模式是否恢复。
• 适配M1芯片多进程机制：M1芯片的macOS默认使用spawn启动进程，PYGAD的自适应突变逻辑可能未正确适配。在代码开头添加以下代码（需放在if __name__ == '__main__':块内）：

  import multiprocessing
  multiprocessing.set_start_method('spawn')
  

• 替换全局变量：fitness函数中使用的global finished_counter在多进程环境下存在共享问题，可能导致进程异常降级。改用multiprocessing.Value实现进程安全的计数器：

  # 在run_backtest前初始化
  from multiprocessing import Value
  finished_counter = Value('i', 0)
  
  # 在fitness_func中修改
  with finished_counter.get_lock():
      finished_counter.value += 1
      print(f"Finished {finished_counter.value}")
  

• 调整并行进程数：尝试降低parallel_processing中的进程数（如设为4），避免M1芯片的进程调度压力触发单进程 fallback。

内容的提问来源于stack exchange，提问作者thecodeman