import random import numpy as np from data_structures import OrderData, RiskEnterpriseData, SupplierData, Config from chromosome_utils import ChromosomeUtils from objective_calculator import ObjectiveCalculator from encoder import Encoder from genetic_operators import GeneticOperator from nsga2 import NSGA2 from visualizer import ResultVisualizer def main(): """主函数：执行NSGA-II算法求解多目标优化问题（整数化版本）""" try: # 1. 初始化随机种子（确保结果可复现） random.seed(42) np.random.seed(42) # 2. 初始化数据（订单、风险企业、供应商、算法配置） print("初始化数据结构...") order_data = OrderData() # 订单数据（需求、交货期等，整数） risk_data = RiskEnterpriseData() # 风险企业数据（整数） supplier_data = SupplierData() # 供应商数据（整数） config = Config() # 算法参数配置 # 3. 初始化工具类和算法组件 print("初始化算法组件...") utils = ChromosomeUtils(order_data, risk_data, supplier_data) # 染色体工具（整数化） calculator = ObjectiveCalculator(order_data, risk_data, supplier_data, utils, config) # 目标函数计算器（整数化） encoder = Encoder(config, utils) # 种群初始化编码器（整数化） genetic_op = GeneticOperator(config, utils) # 遗传操作器（整数化） nsga2 = NSGA2(config.pop_size, config.objective_num) # NSGA-II算法实例 visualizer = ResultVisualizer(utils) # 结果可视化工具（适配整数化） # 4. 初始化种群 print("初始化种群（整数化）...") population = encoder.initialize_population() print(f"初始化种群完成，（种群大小，染色体长度）: {population.shape if population.size > 0 else '空'}") # 若种群初始化失败（为空），直接退出 if population.size == 0: print("错误：种群初始化失败，无法继续进化") return # 5. 记录进化过程中的历史数据（整数化） all_objectives = [] # 所有代的目标函数值（整数） convergence_history = [] # 收敛趋势（每代最优前沿的平均目标值，整数） best_front = [] # 最终帕累托前沿解（整数） best_front_objs = [] # 最终帕累托前沿的目标值（整数） no_improve_count = 0 # 无改进计数器（用于早停） prev_best_avg = float('inf') # 上一代的最优平均目标值（整数求和） # 6. 进化主循环 print(f"开始进化（最大代数：{config.max_generations}，早停耐心：{config.early_stop_patience}）...") for generation in range(config.max_generations): try: # 计算当前种群的目标函数值（整数） objectives = [calculator.calculate_objectives(chrom) for chrom in population] all_objectives.extend(objectives) # 记录所有目标值（整数） # 非支配排序，获取当前代的帕累托前沿 ranks, fronts = nsga2.fast_non_dominated_sort(objectives) current_front = fronts[0] if fronts else [] # 第0层为最优前沿 current_front_objs = [objectives[i] for i in current_front] if current_front else [] best_front = population[current_front] if current_front else [] # 更新当前最优前沿解（整数） best_front_objs = current_front_objs # 更新当前最优前沿目标值（整数） # 记录收敛趋势（基于最优前沿的平均目标值，整数） if len(current_front_objs) > 0: avg_cost = sum(obj[0] for obj in current_front_objs) // len(current_front_objs) # 整数平均 avg_tardiness = sum(obj[1] for obj in current_front_objs) // len(current_front_objs) # 整数平均 convergence_history.append((avg_cost, avg_tardiness)) # 判断是否改进（用于早停） current_avg = avg_cost + avg_tardiness # 合并两个目标的平均值（整数） if abs(current_avg - prev_best_avg) < 1e-4: # 变化小于阈值，视为无改进 no_improve_count += 1 else: no_improve_count = 0 # 有改进，重置计数器 prev_best_avg = current_avg else: no_improve_count += 1 # 无前沿解，视为无改进 # 选择操作（锦标赛选择） selected = nsga2.selection(population, objectives) # 校验选择后的种群大小 assert len(selected) == config.pop_size, \ f"选择后的种群大小({len(selected)})与目标大小({config.pop_size})不符" # 交叉操作（两点交叉）- 修复索引越界问题 offspring = [] # 子代种群（整数） selected_len = len(selected) # selected的长度（等于pop_size） i = 0 max_iter = 2 * config.pop_size # 最大迭代次数，避免无限循环 iter_count = 0 while len(offspring) < config.pop_size and iter_count < max_iter: iter_count += 1 # 检查i是否超出selected范围，若超出则重置（循环使用个体） if i >= selected_len: i = 0 # 尝试两两交叉（需i+1在范围内，且满足交叉概率） if i + 1 < selected_len and random.random() < config.crossover_prob: parent1 = selected[i] parent2 = selected[i + 1] child1, child2 = genetic_op.two_point_crossover(parent1, parent2) # 添加child1（若未达到种群大小） if len(offspring) < config.pop_size: offspring.append(child1) # 添加child2（若未达到种群大小） if len(offspring) < config.pop_size: offspring.append(child2) i += 2 # 处理下一对个体 else: # 直接添加当前父代（避免i+1越界） offspring.append(selected[i]) i += 1 # 处理下一个个体（步长改为1，避免快速越界） # 若迭代次数用尽仍未生成足够子代，补充随机个体（健壮性处理，整数） while len(offspring) < config.pop_size: offspring.append(encoder._generate_random_chromosome()) # 整数化随机染色体 # 变异操作（均匀变异，整数化） offspring = [ genetic_op.uniform_mutation(chrom) if random.random() < config.mutation_prob else chrom for chrom in offspring ] offspring = np.array(offspring[:config.pop_size]).astype(int) # 确保子代大小和整数类型 # 合并父代和子代，准备环境选择 combined = np.vstack([population, offspring]).astype(int) # 合并种群（整数） # 计算合并种群的目标函数值（整数） combined_objs = objectives + [calculator.calculate_objectives(chrom) for chrom in offspring] # 环境选择（保留最优的pop_size个个体） population, objectives = nsga2.environmental_selection(combined, combined_objs) # 校验环境选择后的种群大小和整数类型 population = population.astype(int) assert len(population) == config.pop_size, \ f"环境选择后的种群大小({len(population)})与目标大小({config.pop_size})不符" # 早停检查（连续多代无改进则停止） if no_improve_count >= config.early_stop_patience: print(f"早停触发：连续{no_improve_count}代无改进，终止于第{generation}代") break # 每50代打印一次进度 if generation % 50 == 0: front_size = len(current_front) if current_front else 0 print(f"第{generation}代完成，当前最优前沿大小: {front_size}，无改进计数: {no_improve_count}") except Exception as e: print(f"第{generation}代进化出错：{str(e)}，跳过当前代") continue # 7. 结果可视化与输出（整数化） print("进化完成，处理结果（整数化）...") if len(best_front_objs) > 0: visualizer.plot_pareto_front(all_objectives, best_front_objs) # 绘制帕累托前沿（整数） visualizer.plot_convergence(convergence_history) # 绘制收敛趋势（整数） visualizer.print_pareto_solutions(best_front, best_front_objs) # 打印最优解详情（整数） else: print("未找到有效帕累托前沿解") except Exception as e: print(f"程序运行出错：{str(e)}") import traceback traceback.print_exc() # 打印详细错误栈 if __name__ == "__main__": print("程序启动（整数化版本）...") main() print("程序结束")