guoqiang
/
OrderReallocation-HeavyTrucks-NSGA-2
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 1 Wiki Activity

没报错的一版

This commit is contained in:
Hgq 2025-12-05 23:53:59 +08:00
parent 2e655da16a
commit 0849790f95
8 changed files with 751 additions and 388 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

164
chromosome_utils.py
View File

@ -1,82 +1,122 @@
import random import random
import numpy as np import numpy as np
from data_structures import OrderData, RiskEnterpriseData, SupplierData from data_structures import OrderData, RiskEnterpriseData, SupplierData # 数据结构类
class ChromosomeUtils: class ChromosomeUtils:
"""染色体编码、解码、修复工具类""" """染色体工具类：提供染色体的拆分、合并、修复等功能，确保满足约束条件"""
def __init__(self, order_data: OrderData, risk_data: RiskEnterpriseData, supplier_data: SupplierData): def __init__(self, order_data: OrderData, risk_data: RiskEnterpriseData, supplier_data: SupplierData):
self.order = order_data """
self.risk = risk_data 初始化工具类
self.supplier = supplier_data :param order_data: 订单数据（需求、交货期等）
self.I = order_data.I :param risk_data: 风险企业数据（产能等）
:param supplier_data: 供应商数据（产能、价格等）
"""
self.order = order_data # 订单数据
self.risk = risk_data # 风险企业数据
self.supplier = supplier_data # 供应商数据
self.I = order_data.I # 物料种类数
# 预计算每个物料的可选企业列表（0=风险企业，1+为供应商ID）
self.material_optional_enterprises = self._get_material_optional_enterprises() self.material_optional_enterprises = self._get_material_optional_enterprises()
# 每个物料的可选企业数量
self.material_enterprise_count = [len(ents) for ents in self.material_optional_enterprises] self.material_enterprise_count = [len(ents) for ents in self.material_optional_enterprises]
# 染色体总长度（3层：企业层+能力层+数量层）
self.chromosome_length = 3 * sum(self.material_enterprise_count) self.chromosome_length = 3 * sum(self.material_enterprise_count)
def _get_material_optional_enterprises(self) -> list[list[int]]: def _get_material_optional_enterprises(self) -> list[list[int]]:
"""获取每个物料的可选企业列表：0=风险企业，1~supplier_count=供应商""" """
生成每个物料的可选企业列表（内部方法）
:return: 二维列表，每个子列表为对应物料的可选企业ID
"""
optional = [] optional = []
for i in range(self.I): for i in range(self.I): # 遍历每种物料
ents = [0] # 先加入风险企业 ents = [0] # 先加入风险企业（ID=0）
# 加入可生产该物料的供应商（ID=1+供应商索引）
for j in range(self.supplier.supplier_count): for j in range(self.supplier.supplier_count):
if self.supplier.can_produce[j][i] == 1: if self.supplier.can_produce[j][i] == 1: # 供应商j可生产物料i
ents.append(j + 1) ents.append(j + 1) # 供应商ID为j+1（区分风险企业）
optional.append(ents) optional.append(ents)
return optional return optional
def _split_chromosome(self, chromosome: np.ndarray) -> tuple[np.ndarray, np.ndarray, np.ndarray]: def _split_chromosome(self, chromosome: np.ndarray) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
"""拆分染色体为三层：企业编码层、能力编码层、数量编码层""" """
total_ent_count = sum(self.material_enterprise_count) 拆分染色体为三层
enterprise_layer = chromosome[:total_ent_count] :param chromosome: 完整染色体
capacity_layer = chromosome[total_ent_count:2*total_ent_count] :return: 企业层、能力层、数量层（均为numpy数组）
quantity_layer = chromosome[2*total_ent_count:] """
total_ent_count = sum(self.material_enterprise_count) # 所有物料的企业总数
enterprise_layer = chromosome[:total_ent_count] # 第一层：企业选择（0/1）
capacity_layer = chromosome[total_ent_count:2 * total_ent_count] # 第二层：产能
quantity_layer = chromosome[2 * total_ent_count:] # 第三层：数量
return enterprise_layer, capacity_layer, quantity_layer return enterprise_layer, capacity_layer, quantity_layer
def _merge_chromosome(self, enterprise_layer: np.ndarray, capacity_layer: np.ndarray, def _merge_chromosome(self, enterprise_layer: np.ndarray, capacity_layer: np.ndarray,
quantity_layer: np.ndarray) -> np.ndarray: quantity_layer: np.ndarray) -> np.ndarray:
"""合并三层为完整染色体""" """
合并三层为完整染色体
:param enterprise_layer: 企业层
:param capacity_layer: 能力层
:param quantity_layer: 数量层
:return: 完整染色体
"""
return np.hstack([enterprise_layer, capacity_layer, quantity_layer]) return np.hstack([enterprise_layer, capacity_layer, quantity_layer])
def repair_enterprise_layer(self, enterprise_layer: np.ndarray) -> np.ndarray: def repair_enterprise_layer(self, enterprise_layer: np.ndarray) -> np.ndarray:
"""修复企业编码层：确保每种物料至少有一个企业生产""" """
修复企业层：确保每种物料至少选择一个企业
:param enterprise_layer: 待修复的企业层
:return: 修复后的企业层
"""
repaired = enterprise_layer.copy() repaired = enterprise_layer.copy()
# 计算各物料的企业编码分割点
split_points = np.cumsum(self.material_enterprise_count) split_points = np.cumsum(self.material_enterprise_count)
start = 0 start = 0 # 起始索引
for i in range(self.I): for i in range(self.I): # 遍历每种物料
end = split_points[i] end = split_points[i] # 当前物料的企业编码结束索引
segment = repaired[start:end] segment = repaired[start:end] # 当前物料的企业选择片段
if np.sum(segment) == 0: if np.sum(segment) == 0: # 未选择任何企业时，随机选一个
select_idx = random.randint(0, len(segment)-1) select_idx = random.randint(0, len(segment) - 1)
segment[select_idx] = 1 segment[select_idx] = 1
repaired[start:end] = segment repaired[start:end] = segment
start = end start = end # 更新起始索引
return repaired return repaired
def repair_capacity_layer(self, enterprise_layer: np.ndarray, capacity_layer: np.ndarray) -> np.ndarray: def repair_capacity_layer(self, enterprise_layer: np.ndarray, capacity_layer: np.ndarray) -> np.ndarray:
"""修复能力编码层：满足单物料产能和总产能约束""" """
修复能力层：确保满足单物料产能约束和企业总产能约束
:param enterprise_layer: 企业层（用于确定哪些企业被选中）
:param capacity_layer: 待修复的能力层
:return: 修复后的能力层
"""
repaired = capacity_layer.copy() repaired = capacity_layer.copy()
split_points = np.cumsum(self.material_enterprise_count) split_points = np.cumsum(self.material_enterprise_count)
start = 0 start = 0
# 单物料产能约束
# 1. 修复单物料产能约束（每个企业的单物料产能不超过上限）
for i in range(self.I): for i in range(self.I):
end = split_points[i] end = split_points[i]
ents = self.material_optional_enterprises[i] ents = self.material_optional_enterprises[i] # 可选企业列表
segment = repaired[start:end] segment = repaired[start:end] # 当前物料的产能片段
enterprise_segment = enterprise_layer[start:end] enterprise_segment = enterprise_layer[start:end] # 企业选择状态
for idx, ent in enumerate(ents): for idx, ent in enumerate(ents):
if enterprise_segment[idx] == 1: if enterprise_segment[idx] == 1: # 仅处理被选中的企业
# 确定该企业的单物料最大产能
if ent == 0: if ent == 0:
max_cap = self.risk.C0_i_std[i] max_cap = self.risk.C0_i_std[i] # 风险企业
else: else:
supplier_id = ent - 1 supplier_id = ent - 1
max_cap = self.supplier.Cj_i_std[supplier_id][i] max_cap = self.supplier.Cj_i_std[supplier_id][i] # 供应商
# 若产能不合法，重置为1到max_cap之间的随机值
if segment[idx] <= 0 or segment[idx] > max_cap: if segment[idx] <= 0 or segment[idx] > max_cap:
segment[idx] = random.uniform(1, max_cap) segment[idx] = random.uniform(1, max_cap)
repaired[start:end] = segment repaired[start:end] = segment
start = end start = end
# 总产能约束
enterprise_total_capacity = {} # 2. 修复企业总产能约束（企业所有物料的总产能不超过上限）
# 先计算每个企业的当前总产能
enterprise_total_capacity = {} # {企业ID: 总产能}
start = 0 start = 0
for i in range(self.I): for i in range(self.I):
end = split_points[i] end = split_points[i]
@ -84,20 +124,23 @@ class ChromosomeUtils:
enterprise_segment = enterprise_layer[start:end] enterprise_segment = enterprise_layer[start:end]
capacity_segment = repaired[start:end] capacity_segment = repaired[start:end]
for idx, ent in enumerate(ents): for idx, ent in enumerate(ents):
if enterprise_segment[idx] == 1: if enterprise_segment[idx] == 1: # 仅统计选中的企业
if ent not in enterprise_total_capacity: if ent not in enterprise_total_capacity:
enterprise_total_capacity[ent] = 0 enterprise_total_capacity[ent] = 0
enterprise_total_capacity[ent] += capacity_segment[idx] enterprise_total_capacity[ent] += capacity_segment[idx]
start = end start = end
# 调整超出总产能的企业
# 调整超出总产能上限的企业（按比例缩放）
for ent, total_cap in enterprise_total_capacity.items(): for ent, total_cap in enterprise_total_capacity.items():
# 确定企业的总产能上限
if ent == 0: if ent == 0:
max_total_cap = self.risk.C0_total_max max_total_cap = self.risk.C0_total_max # 风险企业总产能上限
else: else:
supplier_id = ent - 1 supplier_id = ent - 1
max_total_cap = self.supplier.Cj_total_max[supplier_id] max_total_cap = self.supplier.Cj_total_max[supplier_id] # 供应商总产能上限
if total_cap > max_total_cap:
scale = max_total_cap / total_cap if total_cap > max_total_cap: # 超出上限时缩放
scale = max_total_cap / total_cap # 缩放比例
start = 0 start = 0
for i in range(self.I): for i in range(self.I):
end = split_points[i] end = split_points[i]
@ -106,9 +149,10 @@ class ChromosomeUtils:
capacity_segment = repaired[start:end] capacity_segment = repaired[start:end]
for idx, e in enumerate(ents): for idx, e in enumerate(ents):
if e == ent and enterprise_segment[idx] == 1: if e == ent and enterprise_segment[idx] == 1:
capacity_segment[idx] *= scale capacity_segment[idx] *= scale # 按比例缩小
repaired[start:end] = capacity_segment repaired[start:end] = capacity_segment
start = end start = end
return repaired return repaired
def repair_quantity_layer(self, enterprise_layer: np.ndarray, quantity_layer: np.ndarray) -> np.ndarray: def repair_quantity_layer(self, enterprise_layer: np.ndarray, quantity_layer: np.ndarray) -> np.ndarray:
@ -149,11 +193,15 @@ class ChromosomeUtils:
else: else:
scale = qi / current_total scale = qi / current_total
segment[selected_indices] *= scale segment[selected_indices] *= scale
# 步骤3：处理超出最大供应量的情况（修复中文变量名+逻辑） # 步骤3：处理超出最大供应量的情况（添加最大迭代次数限制）
need_adjust = True need_adjust = True
while need_adjust: max_iter = 10 # 最大迭代次数，避免无限循环
iter_count = 0
while need_adjust and iter_count < max_iter: # 增加迭代次数限制
iter_count += 1
need_adjust = False need_adjust = False
total_excess = 0 total_excess = 0
# 截断超出max_q的数量
for idx, ent in zip(selected_indices, selected_ents): for idx, ent in zip(selected_indices, selected_ents):
if ent == 0: if ent == 0:
max_q = qi max_q = qi
@ -165,6 +213,7 @@ class ChromosomeUtils:
total_excess += excess total_excess += excess
segment[idx] = max_q segment[idx] = max_q
need_adjust = True need_adjust = True
# 分配超出量到有剩余的企业
if total_excess > 0: if total_excess > 0:
available_indices = [] available_indices = []
available_max = [] available_max = []
@ -185,22 +234,39 @@ class ChromosomeUtils:
alloc_excess = total_excess * alloc_ratio alloc_excess = total_excess * alloc_ratio
for idx, alloc in zip(available_indices, alloc_excess): for idx, alloc in zip(available_indices, alloc_excess):
segment[idx] += alloc segment[idx] += alloc
else: need_adjust = True # 分配后可能需要再次检查
segment[selected_indices] = np.minimum(segment[selected_indices], else:
[self.supplier.MaxOrder[ent-1][i] if ent !=0 else qi for ent in selected_ents]) # 无剩余容量，强制截断并重新缩放总量
segment[selected_indices] = np.minimum(segment[selected_indices],
[self.supplier.MaxOrder[ent - 1][i] if ent != 0 else qi
for ent in selected_ents])
need_adjust = True
# 检查总量误差，重新缩放
current_total = np.sum(segment[selected_indices]) current_total = np.sum(segment[selected_indices])
if abs(current_total - qi) > 1e-6: if abs(current_total - qi) > 1e-6:
scale = qi / current_total scale = qi / current_total
segment[selected_indices] *= scale segment[selected_indices] *= scale
need_adjust = True need_adjust = True # 缩放后可能需要再次检查
# 最终强制总量等于需求（避免迭代次数到限后仍有误差）
current_total = np.sum(segment[selected_indices])
if abs(current_total - qi) > 1e-6:
scale = qi / current_total
segment[selected_indices] *= scale
repaired[start:end] = segment repaired[start:end] = segment
start = end start = end
return repaired return repaired
def repair_chromosome(self, chromosome: np.ndarray) -> np.ndarray: def repair_chromosome(self, chromosome: np.ndarray) -> np.ndarray:
"""完整修复染色体：企业层→能力层→数量层""" """
完整修复染色体（按顺序修复三层）
:param chromosome: 待修复的染色体
:return: 修复后的染色体
"""
# 拆分三层
enterprise_layer, capacity_layer, quantity_layer = self._split_chromosome(chromosome) enterprise_layer, capacity_layer, quantity_layer = self._split_chromosome(chromosome)
# 依次修复（企业层→能力层→数量层，依赖关系）
enterprise_layer = self.repair_enterprise_layer(enterprise_layer) enterprise_layer = self.repair_enterprise_layer(enterprise_layer)
capacity_layer = self.repair_capacity_layer(enterprise_layer, capacity_layer) capacity_layer = self.repair_capacity_layer(enterprise_layer, capacity_layer)
quantity_layer = self.repair_quantity_layer(enterprise_layer, quantity_layer) quantity_layer = self.repair_quantity_layer(enterprise_layer, quantity_layer)
# 合并修复后的三层
return self._merge_chromosome(enterprise_layer, capacity_layer, quantity_layer) return self._merge_chromosome(enterprise_layer, capacity_layer, quantity_layer)

80
data_structures.py
View File

@ -1,99 +1,113 @@
# 数据结构定义 # 数据结构定义：存储订单、企业、供应商数据及算法配置
class OrderData: class OrderData:
"""订单数据类""" """订单数据类：存储物料需求、交货期、成本等信息"""
def __init__(self): def __init__(self):
self.I = 5 # 物料种类数 self.I = 5 # 物料种类数
self.Q = [250, 300, 200, 350, 280] # 各物料待生产量 self.Q = [250, 300, 200, 350, 280] # 各物料的需求数量
self.Dd = 12 # 需求交货期（时长） self.Dd = 12 # 需求交货期（单位：时间）
self.P0 = [50.0, 80.0, 60.0, 70.0, 90.0] # 风险企业单位采购价 self.P0 = [50.0, 80.0, 60.0, 70.0, 90.0] # 风险企业的单位采购价
self.T0 = [5.0, 8.0, 6.0, 7.0, 9.0] # 风险企业单位运输成本 self.T0 = [5.0, 8.0, 6.0, 7.0, 9.0] # 风险企业的单位运输成本
self.transport_speed = 10.0 # 运输速度（单位/时间） self.transport_speed = 10.0 # 运输速度（单位：距离/时间）
class RiskEnterpriseData: class RiskEnterpriseData:
"""风险企业数据类（单物料产能约束+总产能约束）""" """风险企业数据类：存储风险企业的产能、距离等信息"""
def __init__(self): def __init__(self):
self.I = 5 # 物料种类数（与订单一致） self.I = 5 # 物料种类数（与订单一致）
self.C0_i_std = [40.0, 50.0, 35.0, 45.0, 48.0] # 单物料单位时间标准产能 self.C0_i_std = [40.0, 50.0, 35.0, 45.0, 48.0] # 单物料的单位时间标准产能
self.C0_total_max = 100.0 # 总产能上限 self.C0_total_max = 100.0 # 总产能上限（单位时间）
self.distance = 10.0 # 位置距离 self.distance = 10.0 # 与需求点的距离
class SupplierData: class SupplierData:
"""供应商数据类（按约束文档定义）""" """供应商数据类：存储各供应商的产能、价格、距离等信息"""
def __init__(self, I=5): def __init__(self, I=5):
self.I = I self.I = I # 物料种类数
self.supplier_count = 4 self.supplier_count = 4 # 供应商数量
self.names = ["S0", "S1", "S2", "S3"] self.names = ["S0", "S1", "S2", "S3"] # 供应商名称
# 能否生产某物料矩阵 (supplier_count × I)
# 能否生产某物料的矩阵（supplier_count × I），1=能生产，0=不能
self.can_produce = [ self.can_produce = [
[1, 1, 1, 1, 1], [1, 1, 1, 1, 1],
[1, 0, 1, 0, 1], [1, 0, 1, 0, 1],
[0, 1, 0, 1, 0], [0, 1, 0, 1, 0],
[0, 0, 1, 1, 1] [0, 0, 1, 1, 1]
] ]
# 单物料单位时间标准生产能力
# 单物料单位时间标准产能（supplier_count × I）
self.Cj_i_std = [ self.Cj_i_std = [
[20.0, 18.0, 15.0, 22.0, 25.0], [20.0, 18.0, 15.0, 22.0, 25.0],
[25.0, 0.0, 30.0, 0.0, 28.0], [25.0, 0.0, 30.0, 0.0, 28.0],
[0.0, 22.0, 0.0, 35.0, 0.0], [0.0, 22.0, 0.0, 35.0, 0.0],
[0.0, 0.0, 20.0, 30.0, 22.0] [0.0, 0.0, 20.0, 30.0, 22.0]
] ]
# 供应商单位时间最大总生产能力
# 供应商单位时间的最大总产能（supplier_count）
self.Cj_total_max = [120.0, 110.0, 100.0, 95.0] self.Cj_total_max = [120.0, 110.0, 100.0, 95.0]
# 最小起订量
# 最小起订量（supplier_count × I）
self.MinOrder = [ self.MinOrder = [
[20.0, 20.0, 15.0, 25.0, 20.0], [20.0, 20.0, 15.0, 25.0, 20.0],
[30.0, 0.0, 25.0, 0.0, 30.0], [30.0, 0.0, 25.0, 0.0, 30.0],
[0.0, 25.0, 0.0, 30.0, 0.0], [0.0, 25.0, 0.0, 30.0, 0.0],
[0.0, 0.0, 20.0, 35.0, 25.0] [0.0, 0.0, 20.0, 35.0, 25.0]
] ]
# 最大供应量
# 最大供应量（supplier_count × I）
self.MaxOrder = [ self.MaxOrder = [
[100.0, 150.0, 80.0, 120.0, 130.0], [100.0, 150.0, 80.0, 120.0, 130.0],
[120.0, 0.0, 100.0, 0.0, 110.0], [120.0, 0.0, 100.0, 0.0, 110.0],
[0.0, 140.0, 0.0, 150.0, 0.0], [0.0, 140.0, 0.0, 150.0, 0.0],
[0.0, 0.0, 90.0, 130.0, 100.0] [0.0, 0.0, 90.0, 130.0, 100.0]
] ]
# 单位采购价格
# 单位采购价格（supplier_count × I）
self.P_ij = [ self.P_ij = [
[60.0, 85.0, 70.0, 80.0, 100.0], [60.0, 85.0, 70.0, 80.0, 100.0],
[65.0, 0.0, 75.0, 0.0, 105.0], [65.0, 0.0, 75.0, 0.0, 105.0],
[0.0, 90.0, 0.0, 85.0, 0.0], [0.0, 90.0, 0.0, 85.0, 0.0],
[0.0, 0.0, 78.0, 88.0, 98.0] [0.0, 0.0, 78.0, 88.0, 98.0]
] ]
# 单位运输成本
# 单位运输成本（supplier_count × I）
self.T_ij = [ self.T_ij = [
[7.0, 9.0, 8.0, 10.0, 12.0], [7.0, 9.0, 8.0, 10.0, 12.0],
[6.0, 0.0, 9.0, 0.0, 11.0], [6.0, 0.0, 9.0, 0.0, 11.0],
[0.0, 10.0, 0.0, 12.0, 0.0], [0.0, 10.0, 0.0, 12.0, 0.0],
[0.0, 0.0, 10.0, 13.0, 14.0] [0.0, 0.0, 10.0, 13.0, 14.0]
] ]
# 供应商位置距离
# 供应商与需求点的距离（supplier_count）
self.distance = [45.0, 35.0, 60.0, 50.0] self.distance = [45.0, 35.0, 60.0, 50.0]
class Config: class Config:
"""算法参数配置类""" """算法参数配置类：存储NSGA-II的各类参数"""
def __init__(self): def __init__(self):
# 种群参数 # 种群参数
self.pop_size = 50 self.pop_size = 50 # 种群大小
self.N1_ratio = 0.2 # 变更成本最小种群比例 self.N1_ratio = 0.2 # 优先成本的种群比例
self.N2_ratio = 0.2 # 交付延期最短种群比例 self.N2_ratio = 0.2 # 优先延期的种群比例
self.N3_ratio = 0.3 # 基于风险企业种群比例 self.N3_ratio = 0.3 # 强制风险企业的种群比例
self.N4_ratio = 0.3 # 随机种群比例 self.N4_ratio = 0.3 # 随机种群比例
# 遗传操作参数 # 遗传操作参数
self.crossover_prob = 0.8 self.crossover_prob = 0.8 # 交叉概率
self.mutation_prob = 0.3 self.mutation_prob = 0.3 # 变异概率
self.max_generations = 100 self.max_generations = 100 # 最大进化代数
# 惩罚系数 # 惩罚系数
self.delta = 1.3 # 变更惩罚系数 self.delta = 1.3 # 变更惩罚系数
self.gamma = 0.8 # 提前交付惩罚系数 self.gamma = 0.8 # 提前交付惩罚系数
# 早停参数 # 早停参数
self.early_stop_patience = 50 self.early_stop_patience = 50 # 连续多少代无改进则早停
# 目标函数数量 # 目标函数数量
self.objective_num = 2 self.objective_num = 2 # 双目标（成本+延期）

161
encoder.py
View File

@ -1,114 +1,171 @@
import random import random
import numpy as np import numpy as np
from data_structures import Config from data_structures import Config # 配置参数类
from chromosome_utils import ChromosomeUtils from chromosome_utils import ChromosomeUtils # 染色体工具类
from objective_calculator import ObjectiveCalculator from objective_calculator import ObjectiveCalculator # 目标函数计算器
from nsga2 import NSGA2 from nsga2 import NSGA2 # NSGA-II算法类
class Encoder: class Encoder:
"""种群初始化编码器""" """种群初始化编码器：生成初始种群，包含多种初始化策略"""
def __init__(self, config: Config, utils: ChromosomeUtils): def __init__(self, config: Config, utils: ChromosomeUtils):
"""
初始化编码器
:param config: 算法配置（如种群大小、各策略比例等）
:param utils: 染色体工具类实例
"""
self.config = config self.config = config
self.utils = utils self.utils = utils
self.pop_size = config.pop_size self.pop_size = config.pop_size # 种群总大小
self.N1 = int(config.N1_ratio * self.pop_size)
self.N2 = int(config.N2_ratio * self.pop_size) # 按比例分配四种初始化策略的个体数量
self.N3 = int(config.N3_ratio * self.pop_size) self.N1 = int(config.N1_ratio * self.pop_size) # 优先最小化成本的个体数
self.N4 = self.pop_size - self.N1 - self.N2 - self.N3 self.N2 = int(config.N2_ratio * self.pop_size) # 优先最小化延期的个体数
self.N3 = int(config.N3_ratio * self.pop_size) # 强制选择风险企业的个体数
self.N4 = self.pop_size - self.N1 - self.N2 - self.N3 # 随机生成的个体数
def _generate_random_chromosome(self, force_risk_enterprise: bool = False) -> np.ndarray: def _generate_random_chromosome(self, force_risk_enterprise: bool = False) -> np.ndarray:
"""生成随机染色体""" """
enterprise_layer = [] 生成随机染色体（内部方法）
capacity_layer = [] :param force_risk_enterprise: 是否强制选择风险企业（用于N3策略）
quantity_layer = [] :return: 生成的染色体（经过修复）
for i in range(self.utils.I): """
ent_count = self.utils.material_enterprise_count[i] enterprise_layer = [] # 企业选择层（0/1）
ents = self.utils.material_optional_enterprises[i] capacity_layer = [] # 产能层
# 企业层 quantity_layer = [] # 数量层
e_genes = np.zeros(ent_count)
select_count = random.randint(1, ent_count) for i in range(self.utils.I): # 遍历每种物料
select_indices = random.sample(range(ent_count), select_count) ent_count = self.utils.material_enterprise_count[i] # 当前物料的可选企业数量
e_genes[select_indices] = 1 ents = self.utils.material_optional_enterprises[i] # 可选企业列表（0为风险企业）
# 1. 生成企业层（随机选择至少1个企业）
e_genes = np.zeros(ent_count) # 初始化企业选择为0（未选择）
select_count = random.randint(1, ent_count) # 随机选择1到ent_count个企业
select_indices = random.sample(range(ent_count), select_count) # 随机选择索引
e_genes[select_indices] = 1 # 标记选中的企业
if force_risk_enterprise: if force_risk_enterprise:
e_genes[0] = 1 e_genes[0] = 1 # 强制选择风险企业（索引0）
enterprise_layer.extend(e_genes) enterprise_layer.extend(e_genes)
# 能力层
# 2. 生成能力层（为选中的企业随机分配产能）
c_genes = np.zeros(ent_count) c_genes = np.zeros(ent_count)
for idx, ent in enumerate(ents): for idx, ent in enumerate(ents):
if e_genes[idx] == 1: if e_genes[idx] == 1: # 仅为选中的企业分配产能
if ent == 0: if ent == 0:
max_cap = self.utils.risk.C0_i_std[i] # 修复：self.utils.risk max_cap = self.utils.risk.C0_i_std[i] # 风险企业的最大产能
else: else:
supplier_id = ent - 1 supplier_id = ent - 1
max_cap = self.utils.supplier.Cj_i_std[supplier_id][i] # 修复：self.utils.supplier max_cap = self.utils.supplier.Cj_i_std[supplier_id][i] # 供应商的最大产能
c_genes[idx] = random.uniform(1, max_cap) c_genes[idx] = random.uniform(1, max_cap) # 随机产能
capacity_layer.extend(c_genes) capacity_layer.extend(c_genes)
# 数量层
# 3. 生成数量层（为选中的企业随机分配数量）
q_genes = np.zeros(ent_count) q_genes = np.zeros(ent_count)
for idx, ent in enumerate(ents): for idx, ent in enumerate(ents):
if e_genes[idx] == 1: if e_genes[idx] == 1: # 仅为选中的企业分配数量
if ent == 0: if ent == 0:
max_q = self.utils.order.Q[i] # 修复：self.utils.order max_q = self.utils.order.Q[i] # 风险企业最多分配全部需求
else: else:
supplier_id = ent - 1 supplier_id = ent - 1
max_q = self.utils.supplier.MaxOrder[supplier_id][i] # 修复：self.utils.supplier max_q = self.utils.supplier.MaxOrder[supplier_id][i] # 供应商的最大供应量
q_genes[idx] = random.uniform(1, max_q) q_genes[idx] = random.uniform(1, max_q) # 随机数量
quantity_layer.extend(q_genes) quantity_layer.extend(q_genes)
# 合并三层并修复染色体（确保满足所有约束）
chromosome = self.utils._merge_chromosome( chromosome = self.utils._merge_chromosome(
np.array(enterprise_layer), np.array(capacity_layer), np.array(quantity_layer) np.array(enterprise_layer), np.array(capacity_layer), np.array(quantity_layer)
) )
return self.utils.repair_chromosome(chromosome) return self.utils.repair_chromosome(chromosome)
def initialize_population(self) -> np.ndarray: def initialize_population(self) -> np.ndarray:
"""初始化种群：N1+N2+N3+N4""" """
pop1 = self._initialize_by_objective(self.N1, "cost") 初始化完整种群（四种策略组合）
pop2 = self._initialize_by_objective(self.N2, "tardiness") :return: 初始化后的种群（numpy数组）
pop3 = self._initialize_by_risk_enterprise(self.N3) """
pop4 = self._initialize_random(self.N4) # 按四种策略生成子种群
# 处理空数组情况 pop1 = self._initialize_by_objective(self.N1, "cost") # 优先成本
pop2 = self._initialize_by_objective(self.N2, "tardiness") # 优先延期
pop3 = self._initialize_by_risk_enterprise(self.N3) # 强制风险企业
pop4 = self._initialize_random(self.N4) # 随机生成
# 过滤空数组（避免合并时报错）
population_list = [p for p in [pop1, pop2, pop3, pop4] if len(p) > 0] population_list = [p for p in [pop1, pop2, pop3, pop4] if len(p) > 0]
if len(population_list) == 0: if len(population_list) == 0: # 所有子种群都为空时返回空数组
return np.array([]) return np.array([])
# 合并子种群并打乱顺序
population = np.vstack(population_list) population = np.vstack(population_list)
np.random.shuffle(population) np.random.shuffle(population)
return population[:self.pop_size] return population[:self.pop_size] # 确保种群大小正确
def _initialize_by_objective(self, count: int, objective_type: str) -> np.ndarray: def _initialize_by_objective(self, count: int, objective_type: str) -> np.ndarray:
"""基于目标函数初始化""" """
if count <= 0: 基于目标函数初始化（生成候选解后选择最优的count个）
:param count: 需生成的个体数量
:param objective_type: 优化目标（"cost"或"tardiness"）
:return: 筛选后的子种群
"""
if count <= 0: # 数量为0时返回空数组
return np.array([]) return np.array([])
candidate_count = max(2*count, 20)
# 生成候选解（数量为count的2倍或至少20个，确保有足够选择空间）
candidate_count = max(2 * count, 20)
candidates = [self._generate_random_chromosome() for _ in range(candidate_count)] candidates = [self._generate_random_chromosome() for _ in range(candidate_count)]
calculator = ObjectiveCalculator(self.utils.order, self.utils.risk, self.utils.supplier, self.utils, self.config) # 修复：self.utils的属性
# 计算候选解的目标函数值
calculator = ObjectiveCalculator(self.utils.order, self.utils.risk, self.utils.supplier, self.utils,
self.config)
objectives = [calculator.calculate_objectives(chrom) for chrom in candidates] objectives = [calculator.calculate_objectives(chrom) for chrom in candidates]
# 按目标函数排序并选择前count个
if objective_type == "cost": if objective_type == "cost":
# 按成本升序排序（成本越小越优）
sorted_indices = sorted(range(candidate_count), key=lambda x: objectives[x][0]) sorted_indices = sorted(range(candidate_count), key=lambda x: objectives[x][0])
else: else:
# 按延期升序排序（延期越小越优）
sorted_indices = sorted(range(candidate_count), key=lambda x: objectives[x][1]) sorted_indices = sorted(range(candidate_count), key=lambda x: objectives[x][1])
return np.array([candidates[i] for i in sorted_indices[:count]]) return np.array([candidates[i] for i in sorted_indices[:count]])
def _initialize_by_risk_enterprise(self, count: int) -> np.ndarray: def _initialize_by_risk_enterprise(self, count: int) -> np.ndarray:
"""基于风险企业初始化""" """
基于风险企业初始化（强制选择风险企业，用NSGA-II筛选）
:param count: 需生成的个体数量
:return: 筛选后的子种群
"""
if count <= 0: if count <= 0:
return np.array([]) return np.array([])
candidate_count = max(2*count, 20)
# 生成强制选择风险企业的候选解
candidate_count = max(2 * count, 20)
candidates = [self._generate_random_chromosome(force_risk_enterprise=True) for _ in range(candidate_count)] candidates = [self._generate_random_chromosome(force_risk_enterprise=True) for _ in range(candidate_count)]
calculator = ObjectiveCalculator(self.utils.order, self.utils.risk, self.utils.supplier, self.utils, self.config) # 修复：self.utils的属性
# 计算目标函数并进行非支配排序
calculator = ObjectiveCalculator(self.utils.order, self.utils.risk, self.utils.supplier, self.utils,
self.config)
objectives = [calculator.calculate_objectives(chrom) for chrom in candidates] objectives = [calculator.calculate_objectives(chrom) for chrom in candidates]
nsga2 = NSGA2(candidate_count, 2) nsga2 = NSGA2(candidate_count, 2) # 2个目标函数
ranks, fronts = nsga2.fast_non_dominated_sort(objectives) ranks, fronts = nsga2.fast_non_dominated_sort(objectives) # 非支配排序
# 从帕累托前沿开始选择，直到满足数量
selected = [] selected = []
for front in fronts: for front in fronts:
if len(selected) + len(front) <= count: if len(selected) + len(front) <= count:
selected.extend([candidates[i] for i in front]) selected.extend([candidates[i] for i in front])
else: else:
selected.extend([candidates[i] for i in front[:count-len(selected)]]) # 前沿数量超过剩余需求时，取部分
selected.extend([candidates[i] for i in front[:count - len(selected)]])
break break
return np.array(selected) return np.array(selected)
def _initialize_random(self, count: int) -> np.ndarray: def _initialize_random(self, count: int) -> np.ndarray:
"""随机初始化""" """
随机初始化（直接生成count个随机染色体）
:param count: 需生成的个体数量
:return: 随机子种群
"""
if count <= 0: if count <= 0:
return np.array([]) return np.array([])
return np.array([self._generate_random_chromosome() for _ in range(count)]) return np.array([self._generate_random_chromosome() for _ in range(count)])

107
genetic_operators.py
View File

@ -1,79 +1,118 @@
import random import random
import numpy as np import numpy as np
from data_structures import Config from data_structures import Config # 算法配置参数类
from chromosome_utils import ChromosomeUtils from chromosome_utils import ChromosomeUtils # 染色体工具类
class GeneticOperator: class GeneticOperator:
"""遗传操作：交叉、变异""" """遗传操作类：实现交叉和变异操作，用于产生新个体"""
def __init__(self, config: Config, utils: ChromosomeUtils): def __init__(self, config: Config, utils: ChromosomeUtils):
self.config = config """
self.utils = utils 初始化遗传操作器
:param config: 算法配置参数（如交叉概率、变异概率等）
:param utils: 染色体工具类实例（提供染色体修复等功能）
"""
self.config = config # 配置参数
self.utils = utils # 染色体工具
def two_point_crossover(self, parent1: np.ndarray, parent2: np.ndarray) -> tuple[np.ndarray, np.ndarray]: def two_point_crossover(self, parent1: np.ndarray, parent2: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
"""两点交叉""" """
length = self.utils.chromosome_length 两点交叉：在染色体上随机选择两个点，交换两点之间的基因片段
if length < 2: :param parent1: 父代染色体1
:param parent2: 父代染色体2
:return: 两个子代染色体（经过修复）
"""
length = self.utils.chromosome_length # 染色体总长度
if length < 2: # 染色体长度不足2时无法交叉，直接返回父代副本
return parent1.copy(), parent2.copy() return parent1.copy(), parent2.copy()
point1 = random.randint(0, length//2)
point2 = random.randint(point1+1, length-1) # 随机选择两个交叉点（point1 < point2）
point1 = random.randint(0, length // 2)
point2 = random.randint(point1 + 1, length - 1)
# 复制父代基因并交换片段
child1 = parent1.copy() child1 = parent1.copy()
child2 = parent2.copy() child2 = parent2.copy()
child1[point1:point2] = parent2[point1:point2] child1[point1:point2] = parent2[point1:point2] # 交换point1到point2之间的基因
child2[point1:point2] = parent1[point1:point2] child2[point1:point2] = parent1[point1:point2]
# 修复染色体（确保满足约束条件）
child1 = self.utils.repair_chromosome(child1) child1 = self.utils.repair_chromosome(child1)
child2 = self.utils.repair_chromosome(child2) child2 = self.utils.repair_chromosome(child2)
return child1, child2 return child1, child2
def uniform_mutation(self, chromosome: np.ndarray) -> np.ndarray: def uniform_mutation(self, chromosome: np.ndarray) -> np.ndarray:
"""均匀变异：遍历三层每个基因""" """
mutated = chromosome.copy() 均匀变异：对染色体的三层基因（企业层、能力层、数量层）进行随机变异
:param chromosome: 待变异的染色体
:return: 变异后的染色体（经过修复）
"""
mutated = chromosome.copy() # 复制原始染色体，避免直接修改
# 拆分染色体为三层
enterprise_layer, capacity_layer, quantity_layer = self.utils._split_chromosome(mutated) enterprise_layer, capacity_layer, quantity_layer = self.utils._split_chromosome(mutated)
# 计算各物料的企业编码在染色体中的分割点（用于分层处理）
split_points = np.cumsum(self.utils.material_enterprise_count) split_points = np.cumsum(self.utils.material_enterprise_count)
# 企业层变异（0/1翻转）
start = 0 # 1. 企业层变异（0/1翻转，即是否选择该企业）
for i in range(self.utils.I): start = 0 # 起始索引
end = split_points[i] for i in range(self.utils.I): # 遍历每种物料
end = split_points[i] # 当前物料的企业编码结束索引
# 遍历该物料的所有可选企业
for idx in range(start, end): for idx in range(start, end):
# 以设定的变异概率进行翻转
if random.random() < self.config.mutation_prob: if random.random() < self.config.mutation_prob:
enterprise_layer[idx] = 1 - enterprise_layer[idx] enterprise_layer[idx] = 1 - enterprise_layer[idx] # 0→1或1→0
start = end start = end # 更新起始索引为下一个物料
# 修复企业层（确保每种物料至少选择一个企业）
enterprise_layer = self.utils.repair_enterprise_layer(enterprise_layer) enterprise_layer = self.utils.repair_enterprise_layer(enterprise_layer)
# 能力层变异
# 2. 能力层变异（调整选中企业的产能）
start = 0 start = 0
for i in range(self.utils.I): for i in range(self.utils.I):
end = split_points[i] end = split_points[i]
ents = self.utils.material_optional_enterprises[i] ents = self.utils.material_optional_enterprises[i] # 当前物料的可选企业列表
e_segment = enterprise_layer[start:end] e_segment = enterprise_layer[start:end] # 当前物料的企业选择状态
for idx in range(start, end): for idx in range(start, end):
if random.random() < self.config.mutation_prob and e_segment[idx-start] == 1: # 仅对被选中的企业（e_segment[idx-start]==1）进行变异
ent = ents[idx-start] if random.random() < self.config.mutation_prob and e_segment[idx - start] == 1:
ent = ents[idx - start] # 企业ID（0为风险企业，1+为供应商）
# 确定该企业的最大产能
if ent == 0: if ent == 0:
max_cap = self.utils.risk.C0_i_std[i] max_cap = self.utils.risk.C0_i_std[i] # 风险企业的单物料产能
else: else:
supplier_id = ent - 1 supplier_id = ent - 1 # 供应商索引（转换为0基）
max_cap = self.utils.supplier.Cj_i_std[supplier_id][i] max_cap = self.utils.supplier.Cj_i_std[supplier_id][i] # 供应商的单物料产能
# 随机生成1到max_cap之间的新产能
capacity_layer[idx] = random.uniform(1, max_cap) capacity_layer[idx] = random.uniform(1, max_cap)
start = end start = end
# 修复能力层（确保不超过企业总产能约束）
capacity_layer = self.utils.repair_capacity_layer(enterprise_layer, capacity_layer) capacity_layer = self.utils.repair_capacity_layer(enterprise_layer, capacity_layer)
# 数量层变异
# 3. 数量层变异（调整选中企业的生产数量）
start = 0 start = 0
for i in range(self.utils.I): for i in range(self.utils.I):
end = split_points[i] end = split_points[i]
ents = self.utils.material_optional_enterprises[i] ents = self.utils.material_optional_enterprises[i]
e_segment = enterprise_layer[start:end] e_segment = enterprise_layer[start:end]
for idx in range(start, end): for idx in range(start, end):
if random.random() < self.config.mutation_prob and e_segment[idx-start] == 1: # 仅对被选中的企业进行变异
ent = ents[idx-start] if random.random() < self.config.mutation_prob and e_segment[idx - start] == 1:
qi = self.utils.order.Q[i] ent = ents[idx - start]
qi = self.utils.order.Q[i] # 当前物料的总需求
# 确定该企业的最大可分配数量
if ent == 0: if ent == 0:
max_q = qi max_q = qi # 风险企业最多分配全部需求
else: else:
supplier_id = ent - 1 supplier_id = ent - 1
max_q = self.supplier.MaxOrder[supplier_id][i] max_q = self.utils.supplier.MaxOrder[supplier_id][i] # 供应商的最大供应量
# 随机生成1到max_q之间的新数量
quantity_layer[idx] = random.uniform(1, max_q) quantity_layer[idx] = random.uniform(1, max_q)
start = end start = end
# 修复数量层（确保总数量满足需求，且不超过企业最大供应）
quantity_layer = self.utils.repair_quantity_layer(enterprise_layer, quantity_layer) quantity_layer = self.utils.repair_quantity_layer(enterprise_layer, quantity_layer)
# 合并并修复
# 合并三层并返回
mutated = self.utils._merge_chromosome(enterprise_layer, capacity_layer, quantity_layer) mutated = self.utils._merge_chromosome(enterprise_layer, capacity_layer, quantity_layer)
return mutated return mutated

240
main.py
View File

@ -1,5 +1,6 @@
import random import random
import numpy as np import numpy as np
# 导入数据结构和工具类（根据实际项目结构调整导入路径）
from data_structures import OrderData, RiskEnterpriseData, SupplierData, Config from data_structures import OrderData, RiskEnterpriseData, SupplierData, Config
from chromosome_utils import ChromosomeUtils from chromosome_utils import ChromosomeUtils
from objective_calculator import ObjectiveCalculator from objective_calculator import ObjectiveCalculator
@ -10,105 +11,162 @@ from visualizer import ResultVisualizer
def main(): def main():
# 初始化数据 """主函数：执行NSGA-II算法求解多目标优化问题"""
order_data = OrderData() try:
risk_data = RiskEnterpriseData() # 1. 初始化随机种子（确保结果可复现）
supplier_data = SupplierData() random.seed(42)
config = Config() np.random.seed(42)
# 初始化工具类 # 2. 初始化数据（订单、风险企业、供应商、算法配置）
utils = ChromosomeUtils(order_data, risk_data, supplier_data) print("初始化数据结构...")
calculator = ObjectiveCalculator(order_data, risk_data, supplier_data, utils, config) order_data = OrderData() # 订单数据（需求、交货期等）
encoder = Encoder(config, utils) risk_data = RiskEnterpriseData() # 风险企业数据
genetic_op = GeneticOperator(config, utils) supplier_data = SupplierData() # 供应商数据
nsga2 = NSGA2(config.pop_size, config.objective_num) config = Config() # 算法参数配置
visualizer = ResultVisualizer(utils)
# 初始化种群 # 3. 初始化工具类和算法组件
population = encoder.initialize_population() print("初始化算法组件...")
print(f"初始化种群完成，种群大小: {population.shape}") 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. 初始化种群
all_objectives = [] print("初始化种群...")
convergence_history = [] population = encoder.initialize_population()
best_front = [] print(f"初始化种群完成，种群大小: {population.shape if population.size > 0 else '空'}")
best_front_objs = [] # 若种群初始化失败（为空），直接退出
no_improve_count = 0 if population.size == 0:
prev_best_avg = float('inf') print("错误：种群初始化失败，无法继续进化")
return
# 进化过程 # 5. 记录进化过程中的历史数据
for generation in range(config.max_generations): all_objectives = [] # 所有代的目标函数值
# 计算目标函数 convergence_history = [] # 收敛趋势（每代最优前沿的平均目标值）
objectives = [calculator.calculate_objectives(chrom) for chrom in population] best_front = [] # 最终帕累托前沿解
all_objectives.extend(objectives) best_front_objs = [] # 最终帕累托前沿的目标值
no_improve_count = 0 # 无改进计数器（用于早停）
prev_best_avg = float('inf') # 上一代的最优平均目标值
# 记录当前代的最优前沿 # 6. 进化主循环
ranks, fronts = nsga2.fast_non_dominated_sort(objectives) print(f"开始进化（最大代数：{config.max_generations}，早停耐心：{config.early_stop_patience}）...")
current_front = fronts[0] for generation in range(config.max_generations):
current_front_objs = [objectives[i] for i in current_front] try:
best_front = population[current_front] # 计算当前种群的目标函数值
best_front_objs = current_front_objs objectives = [calculator.calculate_objectives(chrom) for chrom in population]
all_objectives.extend(objectives) # 记录所有目标值
# 收敛判断（基于前沿平均目标值） # 非支配排序，获取当前代的帕累托前沿
if len(current_front_objs) > 0: ranks, fronts = nsga2.fast_non_dominated_sort(objectives)
avg_cost = sum(obj[0] for obj in current_front_objs) / len(current_front_objs) current_front = fronts[0] if fronts else [] # 第0层为最优前沿
avg_tardiness = sum(obj[1] for obj in current_front_objs) / len(current_front_objs) current_front_objs = [objectives[i] for i in current_front] if current_front else []
convergence_history.append((avg_cost, avg_tardiness)) best_front = population[current_front] if current_front else [] # 更新当前最优前沿解
current_avg = avg_cost + avg_tardiness best_front_objs = current_front_objs # 更新当前最优前沿目标值
if abs(current_avg - prev_best_avg) < 1e-4:
no_improve_count += 1 # 记录收敛趋势（基于最优前沿的平均目标值）
else: if len(current_front_objs) > 0:
no_improve_count = 0 avg_cost = sum(obj[0] for obj in current_front_objs) / len(current_front_objs)
prev_best_avg = current_avg 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]) # 确保子代大小严格等于pop_size
# 合并父代和子代，准备环境选择
combined = np.vstack([population, offspring]) # 合并种群
# 计算合并种群的目标函数值
combined_objs = objectives + [calculator.calculate_objectives(chrom) for chrom in offspring]
# 环境选择（保留最优的pop_size个个体）
population, objectives = nsga2.environmental_selection(combined, combined_objs)
# 校验环境选择后的种群大小
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: else:
no_improve_count += 1 print("未找到有效帕累托前沿解")
# 选择（锦标赛选择） except Exception as e:
selected = nsga2.selection(population, objectives) print(f"程序运行出错：{str(e)}")
import traceback
# 交叉 traceback.print_exc() # 打印详细错误栈
offspring = []
i = 0
while len(offspring) < config.pop_size:
if i+1 < config.pop_size and random.random() < config.crossover_prob:
parent1 = selected[i]
parent2 = selected[i+1]
child1, child2 = genetic_op.two_point_crossover(parent1, parent2)
offspring.append(child1)
if len(offspring) < config.pop_size:
offspring.append(child2)
else:
offspring.append(selected[i])
i += 2
# 变异
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])
# 合并父代和子代
combined = np.vstack([population, offspring])
combined_objs = objectives + [calculator.calculate_objectives(chrom) for chrom in offspring]
# 环境选择（核心：NSGA-II的环境选择）
population, objectives = nsga2.environmental_selection(combined, combined_objs)
# 早停检查
if no_improve_count >= config.early_stop_patience:
print(f"早停于第 {generation} 代")
break
if generation % 50 == 0:
print(f"第 {generation} 代完成，当前最优前沿大小: {len(current_front)}")
# 结果可视化与打印
print("进化完成，绘制结果...")
visualizer.plot_pareto_front(all_objectives, best_front_objs)
visualizer.plot_convergence(convergence_history)
visualizer.print_pareto_solutions(best_front, best_front_objs)
if __name__ == "__main__": if __name__ == "__main__":
random.seed(42) print("程序启动...")
np.random.seed(42) main()
main() print("程序结束")

150
nsga2.py
View File

@ -3,119 +3,181 @@ import numpy as np
class NSGA2: class NSGA2:
"""NSGA-II算法类：实现多目标优化的非支配排序、选择、环境选择等核心操作"""
def __init__(self, pop_size, objective_num): def __init__(self, pop_size, objective_num):
self.pop_size = pop_size """
self.objective_num = objective_num 初始化NSGA-II
:param pop_size: 种群大小
:param objective_num: 目标函数数量
"""
self.pop_size = pop_size # 种群大小
self.objective_num = objective_num # 目标数量
def fast_non_dominated_sort(self, objective_values): def fast_non_dominated_sort(self, objective_values):
"""标准快速非支配排序实现""" """
快速非支配排序：将个体按支配关系分层（前沿）
:param objective_values: 所有个体的目标函数值列表
:return: ranks（每个个体的层级）和 fronts（每层的个体索引列表）
"""
pop_size = len(objective_values) pop_size = len(objective_values)
domination_count = [0] * pop_size domination_count = [0] * pop_size # 每个个体被支配的次数
dominated_solutions = [[] for _ in range(pop_size)] dominated_solutions = [[] for _ in range(pop_size)] # 每个个体支配的个体列表
ranks = [0] * pop_size ranks = [0] * pop_size # 每个个体的层级（0为最优）
front = [[]] # 第0层 front = [[]] # 第0层前沿（初始为空）
# 计算支配关系 # 1. 计算支配关系
for p in range(pop_size): for p in range(pop_size):
for q in range(pop_size): for q in range(pop_size):
if p == q: if p == q:
continue continue # 个体不与自身比较
# p支配q？ # 判断p是否支配q
if self._dominates(objective_values[p], objective_values[q]): if self._dominates(objective_values[p], objective_values[q]):
dominated_solutions[p].append(q) dominated_solutions[p].append(q) # p支配q，记录q
# q支配p？ # 判断q是否支配p
elif self._dominates(objective_values[q], objective_values[p]): elif self._dominates(objective_values[q], objective_values[p]):
domination_count[p] += 1 domination_count[p] += 1 # p被q支配，计数+1
# 若p未被任何个体支配，加入第0层前沿
if domination_count[p] == 0: if domination_count[p] == 0:
ranks[p] = 0 ranks[p] = 0
front[0].append(p) front[0].append(p)
# 处理后续层 # 2. 生成后续层级的前沿
i = 0 i = 0
while len(front[i]) > 0: while len(front[i]) > 0: # 当前层非空时继续
next_front = [] next_front = [] # 下一层前沿
for p in front[i]: for p in front[i]: # 遍历当前层的每个个体
for q in dominated_solutions[p]: for q in dominated_solutions[p]: # p支配的个体q
domination_count[q] -= 1 domination_count[q] -= 1 # q的被支配计数-1
if domination_count[q] == 0: if domination_count[q] == 0: # q不再被任何个体支配
ranks[q] = i + 1 ranks[q] = i + 1 # 层级为当前层+1
next_front.append(q) next_front.append(q) # 加入下一层
i += 1 i += 1
front.append(next_front) front.append(next_front) # 追加下一层
return ranks, front[:-1] # 返回rank数组和各层front列表 return ranks, front[:-1] # 返回层级和非空前沿
def _dominates(self, a, b): def _dominates(self, a, b):
"""判断a是否支配b：a的所有目标≤b，且至少一个目标<b""" """
all_le = all(a[i] <= b[i] for i in range(self.objective_num)) 判断个体a是否支配个体b
any_lt = any(a[i] < b[i] for i in range(self.objective_num)) 支配条件：a的所有目标函数值≤b，且至少有一个目标函数值<b
:param a: 个体a的目标值
:param b: 个体b的目标值
:return: 布尔值（a支配b则为True）
"""
all_le = all(a[i] <= b[i] for i in range(self.objective_num)) # 所有目标a≤b
any_lt = any(a[i] < b[i] for i in range(self.objective_num)) # 至少一个目标a<b
return all_le and any_lt return all_le and any_lt
def crowding_distance(self, objective_values, front): def crowding_distance(self, objective_values, front):
"""计算拥挤度""" """
计算前沿中个体的拥挤度（衡量个体在前沿中的分散程度）
:param objective_values: 所有个体的目标值
:param front: 当前前沿的个体索引列表
:return: 每个个体的拥挤度（与front顺序对应）
"""
pop_size = len(front) pop_size = len(front)
distance = [0.0] * pop_size distance = [0.0] * pop_size # 拥挤度初始化
if pop_size <= 1: if pop_size <= 1: # 前沿只有1个个体时，拥挤度为无穷大
return distance return distance
# 按每个目标排序
# 对每个目标函数计算拥挤度
for m in range(self.objective_num): for m in range(self.objective_num):
# 按目标m的值对前沿个体排序
sorted_indices = sorted(range(pop_size), key=lambda i: objective_values[front[i]][m]) sorted_indices = sorted(range(pop_size), key=lambda i: objective_values[front[i]][m])
# 目标m的最大值和最小值
max_val = objective_values[front[sorted_indices[-1]]][m] max_val = objective_values[front[sorted_indices[-1]]][m]
min_val = objective_values[front[sorted_indices[0]]][m] min_val = objective_values[front[sorted_indices[0]]][m]
if max_val - min_val == 0:
if max_val - min_val == 0: # 目标值无差异，无需计算
continue continue
# 边界点设为无穷大
# 边界个体（最小和最大）的拥挤度设为无穷大（确保被保留）
distance[sorted_indices[0]] = float('inf') distance[sorted_indices[0]] = float('inf')
distance[sorted_indices[-1]] = float('inf') distance[sorted_indices[-1]] = float('inf')
# 中间点计算距离
# 中间个体的拥挤度：相邻个体的目标值差除以目标值范围
for i in range(1, pop_size - 1): for i in range(1, pop_size - 1):
distance[sorted_indices[i]] += ( distance[sorted_indices[i]] += (
objective_values[front[sorted_indices[i + 1]]][m] - objective_values[front[sorted_indices[i + 1]]][m] -
objective_values[front[sorted_indices[i - 1]]][m] objective_values[front[sorted_indices[i - 1]]][m]
) / (max_val - min_val) ) / (max_val - min_val)
return distance return distance
def selection(self, population, objective_values): def selection(self, population, objective_values):
"""锦标赛选择（2选1）""" """
锦标赛选择：从种群中选择父代（保留较优个体）
:param population: 当前种群
:param objective_values: 种群的目标函数值
:return: 选择后的父代种群（与原种群大小相同）
"""
pop_size = len(population) pop_size = len(population)
# 计算每个个体的层级和拥挤度
ranks, _ = self.fast_non_dominated_sort(objective_values) ranks, _ = self.fast_non_dominated_sort(objective_values)
distances = self._calculate_all_crowding_distances(objective_values, ranks) distances = self._calculate_all_crowding_distances(objective_values, ranks)
selected = [] selected = []
for _ in range(pop_size): for _ in range(pop_size):
# 随机选择两个个体进行锦标赛
i = random.randint(0, pop_size - 1) i = random.randint(0, pop_size - 1)
j = random.randint(0, pop_size - 1) j = random.randint(0, pop_size - 1)
# 选择层级低（更优）或同层级拥挤度高（更分散）的个体
if ranks[i] < ranks[j] or (ranks[i] == ranks[j] and distances[i] >= distances[j]): if ranks[i] < ranks[j] or (ranks[i] == ranks[j] and distances[i] >= distances[j]):
selected.append(population[i]) selected.append(population[i])
else: else:
selected.append(population[j]) selected.append(population[j])
return np.array(selected) return np.array(selected)
def _calculate_all_crowding_distances(self, objective_values, ranks): def _calculate_all_crowding_distances(self, objective_values, ranks):
"""计算所有个体的拥挤度""" """
max_rank = max(ranks) if ranks else 0 计算所有个体的拥挤度（内部方法）
all_distances = [0.0] * len(objective_values) :param objective_values: 目标函数值
:param ranks: 每个个体的层级
:return: 所有个体的拥挤度列表
"""
max_rank = max(ranks) if ranks else 0 # 最大层级
all_distances = [0.0] * len(objective_values) # 所有个体的拥挤度
# 按层级计算拥挤度
for r in range(max_rank + 1): for r in range(max_rank + 1):
front = [i for i in range(len(objective_values)) if ranks[i] == r] front = [i for i in range(len(objective_values)) if ranks[i] == r] # 当前层级的个体
if len(front) <= 1: if len(front) <= 1: # 单个个体拥挤度为无穷大
for i in front: for i in front:
all_distances[i] = float('inf') all_distances[i] = float('inf')
continue continue
# 计算当前层级的拥挤度
distances = self.crowding_distance(objective_values, front) distances = self.crowding_distance(objective_values, front)
# 映射到全局索引
for i, idx in enumerate(front): for i, idx in enumerate(front):
all_distances[idx] = distances[i] all_distances[idx] = distances[i]
return all_distances return all_distances
def environmental_selection(self, population, objective_values): def environmental_selection(self, population, objective_values):
"""环境选择：从合并种群中选择pop_size个个体""" """
环境选择：从合并的父代和子代中选择最优的pop_size个个体
:param population: 合并的种群（父代+子代）
:param objective_values: 合并种群的目标函数值
:return: 选择后的种群和对应的目标值
"""
# 非支配排序
ranks, fronts = self.fast_non_dominated_sort(objective_values) ranks, fronts = self.fast_non_dominated_sort(objective_values)
selected = [] selected = [] # 选中的个体索引
# 按层级从优到劣选择，直到接近种群大小
for front in fronts: for front in fronts:
if len(selected) + len(front) <= self.pop_size: if len(selected) + len(front) <= self.pop_size:
selected.extend(front) selected.extend(front) # 整层加入
else: else:
# 计算当前front的拥挤度，选择剩余数量的个体 # 当前层无法全加入时，按拥挤度选择剩余个体
distances = self.crowding_distance(objective_values, front) distances = self.crowding_distance(objective_values, front)
# 按拥挤度降序排序（优先选择分散的个体）
sorted_front = [x for _, x in sorted(zip(distances, front), reverse=True)] sorted_front = [x for _, x in sorted(zip(distances, front), reverse=True)]
# 选择剩余所需数量
selected.extend(sorted_front[:self.pop_size - len(selected)]) selected.extend(sorted_front[:self.pop_size - len(selected)])
break break
# 返回选中的个体和对应的目标值
return population[selected], [objective_values[i] for i in selected] return population[selected], [objective_values[i] for i in selected]

136
objective_calculator.py
View File

@ -4,100 +4,118 @@ from chromosome_utils import ChromosomeUtils
class ObjectiveCalculator: class ObjectiveCalculator:
"""目标函数计算器：变更成本C + 交付延期T""" """目标函数计算器：计算双目标值（变更成本 + 交付延期）"""
def __init__(self, order_data: OrderData, risk_data: RiskEnterpriseData, supplier_data: SupplierData, def __init__(self, order_data: OrderData, risk_data: RiskEnterpriseData, supplier_data: SupplierData,
utils: ChromosomeUtils, config: Config): utils: ChromosomeUtils, config: Config):
"""
初始化计算器
:param order_data: 订单数据
:param risk_data: 风险企业数据
:param supplier_data: 供应商数据
:param utils: 染色体工具类
:param config: 算法配置（惩罚系数等）
"""
self.order = order_data self.order = order_data
self.risk = risk_data self.risk = risk_data
self.supplier = supplier_data self.supplier = supplier_data
self.utils = utils self.utils = utils
self.config = config self.config = config
# 预计算物料企业编码的分割点（提高效率）
self.split_points = np.cumsum(utils.material_enterprise_count) self.split_points = np.cumsum(utils.material_enterprise_count)
def calculate_objectives(self, chromosome: np.ndarray) -> tuple[float, float]: def calculate_objectives(self, chromosome: np.ndarray) -> tuple[float, float]:
"""计算双目标值：(变更成本C, 交付延期T)""" """
计算双目标值
:param chromosome: 染色体（解）
:return: (变更成本C, 交付延期T)
"""
# 拆分染色体为三层
enterprise_layer, capacity_layer, quantity_layer = self.utils._split_chromosome(chromosome) enterprise_layer, capacity_layer, quantity_layer = self.utils._split_chromosome(chromosome)
# 修复：传入capacity_layer参数 # 计算变更成本和交付延期
C = self._calculate_change_cost(enterprise_layer, capacity_layer, quantity_layer) C = self._calculate_change_cost(enterprise_layer, capacity_layer, quantity_layer)
T = self._calculate_tardiness(enterprise_layer, capacity_layer, quantity_layer) T = self._calculate_tardiness(enterprise_layer, capacity_layer, quantity_layer)
return C, T return C, T
# 修复：添加capacity_layer参数 def _calculate_change_cost(self, enterprise_layer: np.ndarray, capacity_layer: np.ndarray,
def _calculate_change_cost(self, enterprise_layer: np.ndarray, capacity_layer: np.ndarray, quantity_layer: np.ndarray) -> float: quantity_layer: np.ndarray) -> float:
"""计算变更成本C = C1 + C2 + C3 + C4""" """
C1 = 0.0 # 变更惩罚成本 计算变更成本 C = C1 + C2 + C3 + C4
C2 = 0.0 # 采购变更成本 - C1: 变更惩罚成本（使用供应商替代风险企业的惩罚）
C3 = 0.0 # 运输变更成本 - C2: 采购成本差异（变更后 - 原始）
C4 = 0.0 # 提前交付惩罚成本 - C3: 运输成本差异（变更后 - 原始）
- C4: 提前交付惩罚成本
"""
C1 = 0.0
C2 = 0.0
C3 = 0.0
C4 = 0.0
# 原始成本（全部由风险企业生产） # 原始成本（全部由风险企业生产时的成本）
original_purchase_cost = sum(self.order.Q[i] * self.order.P0[i] for i in range(self.order.I)) original_purchase_cost = sum(self.order.Q[i] * self.order.P0[i] for i in range(self.order.I))
original_transport_cost = sum(self.order.Q[i] * self.order.T0[i] for i in range(self.order.I)) original_transport_cost = sum(self.order.Q[i] * self.order.T0[i] for i in range(self.order.I))
# 变更后成本 # 变更后成本（当前解的成本）
new_purchase_cost = 0.0 new_purchase_cost = 0.0
new_transport_cost = 0.0 new_transport_cost = 0.0
risk_production = np.zeros(self.order.I) # 风险企业生产数量 risk_production = np.zeros(self.order.I) # 风险企业生产的数量
supplier_production = np.zeros(self.order.I) # 供应商生产数量 supplier_production = np.zeros(self.order.I) # 供应商生产的数量
start = 0 start = 0
for i in range(self.order.I): for i in range(self.order.I): # 遍历每种物料
end = self.split_points[i] end = self.split_points[i]
ents = self.utils.material_optional_enterprises[i] ents = self.utils.material_optional_enterprises[i] # 可选企业
e_segment = enterprise_layer[start:end] e_segment = enterprise_layer[start:end] # 企业选择状态
q_segment = quantity_layer[start:end] q_segment = quantity_layer[start:end] # 数量分配
for idx, ent in enumerate(ents): for idx, ent in enumerate(ents):
if e_segment[idx] == 1: if e_segment[idx] == 1: # 仅处理被选中的企业
q = q_segment[idx] q = q_segment[idx] # 分配的数量
if ent == 0: if ent == 0: # 风险企业
# 风险企业
risk_production[i] += q risk_production[i] += q
new_purchase_cost += q * self.order.P0[i] new_purchase_cost += q * self.order.P0[i] # 采购成本
new_transport_cost += q * self.order.T0[i] new_transport_cost += q * self.order.T0[i] # 运输成本
else: else: # 供应商
# 供应商
supplier_id = ent - 1 supplier_id = ent - 1
supplier_production[i] += q supplier_production[i] += q
new_purchase_cost += q * self.supplier.P_ij[supplier_id][i] new_purchase_cost += q * self.supplier.P_ij[supplier_id][i] # 采购成本
new_transport_cost += q * self.supplier.T_ij[supplier_id][i] new_transport_cost += q * self.supplier.T_ij[supplier_id][i] # 运输成本
start = end start = end
# 计算C1：变更惩罚成本 # 计算C1：变更惩罚成本（对供应商生产的部分收取惩罚）
for i in range(self.order.I): for i in range(self.order.I):
# 惩罚系数×供应商生产数量×（风险企业的单位采购+运输成本）
C1 += self.config.delta * supplier_production[i] * (self.order.P0[i] + self.order.T0[i]) C1 += self.config.delta * supplier_production[i] * (self.order.P0[i] + self.order.T0[i])
# 计算C2：采购变更成本 # 计算C2：采购成本差异（变更后 - 原始）
C2 = new_purchase_cost - original_purchase_cost C2 = new_purchase_cost - original_purchase_cost
# 计算C3：运输变更成本 # 计算C3：运输成本差异（变更后 - 原始）
C3 = new_transport_cost - original_transport_cost C3 = new_transport_cost - original_transport_cost
# 计算C4：提前交付惩罚成本 # 计算C4：提前交付惩罚成本（若实际交货期早于需求交货期）
# 修复：传入capacity_layer参数
actual_delivery_time = self._calculate_actual_delivery_time(enterprise_layer, capacity_layer, quantity_layer) actual_delivery_time = self._calculate_actual_delivery_time(enterprise_layer, capacity_layer, quantity_layer)
if actual_delivery_time < self.order.Dd: if actual_delivery_time < self.order.Dd:
# 计算风险企业和各供应商的交货时间 # 计算风险企业和供应商的交货时间
risk_delivery = [] risk_delivery = [] # 风险企业的各物料交货时间
supplier_deliveries = {} supplier_deliveries = {} # {供应商ID: 各物料交货时间}
start = 0 start = 0
for i in range(self.order.I): for i in range(self.order.I):
end = self.split_points[i] end = self.split_points[i]
ents = self.utils.material_optional_enterprises[i] ents = self.utils.material_optional_enterprises[i]
e_segment = enterprise_layer[start:end] e_segment = enterprise_layer[start:end]
c_segment = capacity_layer[start:end] c_segment = capacity_layer[start:end] # 产能（用于计算生产时间）
q_segment = quantity_layer[start:end] q_segment = quantity_layer[start:end] # 数量
for idx, ent in enumerate(ents): for idx, ent in enumerate(ents):
if e_segment[idx] == 1: if e_segment[idx] == 1:
q = q_segment[idx] q = q_segment[idx]
c = c_segment[idx] c = c_segment[idx]
if c == 0: # 生产时间 = 数量 / 产能（产能为0时按0处理）
production_time = 0 production_time = q / c if c != 0 else 0
else:
production_time = q / c
# 运输时间 = 距离 / 运输速度
if ent == 0: if ent == 0:
transport_time = self.risk.distance / self.order.transport_speed transport_time = self.risk.distance / self.order.transport_speed
risk_delivery.append(production_time + transport_time) risk_delivery.append(production_time + transport_time)
@ -109,48 +127,58 @@ class ObjectiveCalculator:
supplier_deliveries[supplier_id].append(production_time + transport_time) supplier_deliveries[supplier_id].append(production_time + transport_time)
start = end start = end
# 风险企业的最大交货时间（取最长）
D0 = max(risk_delivery) if risk_delivery else 0 D0 = max(risk_delivery) if risk_delivery else 0
# 所有供应商的最大交货时间之和
Dj_sum = sum(max(times) for times in supplier_deliveries.values()) if supplier_deliveries else 0 Dj_sum = sum(max(times) for times in supplier_deliveries.values()) if supplier_deliveries else 0
# 提前交付惩罚 = 惩罚系数 ×（需求交货期 - 风险企业交货时间 + 供应商总交货时间）
C4 = self.config.gamma * ((self.order.Dd - D0) + Dj_sum) C4 = self.config.gamma * ((self.order.Dd - D0) + Dj_sum)
return C1 + C2 + C3 + C4 return C1 + C2 + C3 + C4
def _calculate_actual_delivery_time(self, enterprise_layer: np.ndarray, capacity_layer: np.ndarray, def _calculate_actual_delivery_time(self, enterprise_layer: np.ndarray, capacity_layer: np.ndarray,
quantity_layer: np.ndarray) -> float: quantity_layer: np.ndarray) -> float:
"""计算实际交货期：所有企业中最长的生产+运输时间""" """
max_time = 0.0 计算实际交货期：所有企业中最长的（生产时间 + 运输时间）
:return: 实际交货期
"""
max_time = 0.0 # 最大时间（实际交货期）
start = 0 start = 0
for i in range(self.order.I): for i in range(self.order.I):
end = self.split_points[i] end = self.split_points[i]
ents = self.utils.material_optional_enterprises[i] ents = self.utils.material_optional_enterprises[i]
e_segment = enterprise_layer[start:end] e_segment = enterprise_layer[start:end]
c_segment = capacity_layer[start:end] c_segment = capacity_layer[start:end] # 产能
q_segment = quantity_layer[start:end] q_segment = quantity_layer[start:end] # 数量
for idx, ent in enumerate(ents): for idx, ent in enumerate(ents):
if e_segment[idx] == 1: if e_segment[idx] == 1: # 仅处理选中的企业
q = q_segment[idx] q = q_segment[idx]
c = c_segment[idx] c = c_segment[idx]
if c == 0: # 生产时间 = 数量 / 产能（产能为0时按0处理）
production_time = 0 production_time = q / c if c != 0 else 0
else:
production_time = q / c
# 计算运输时间 # 运输时间
if ent == 0: if ent == 0:
transport_time = self.risk.distance / self.order.transport_speed transport_time = self.risk.distance / self.order.transport_speed
else: else:
supplier_id = ent - 1 supplier_id = ent - 1
transport_time = self.supplier.distance[supplier_id] / self.order.transport_speed transport_time = self.supplier.distance[supplier_id] / self.order.transport_speed
# 总时间 = 生产时间 + 运输时间
total_time = production_time + transport_time total_time = production_time + transport_time
if total_time > max_time: if total_time > max_time:
max_time = total_time max_time = total_time # 更新最大时间
start = end start = end
return max_time return max_time
def _calculate_tardiness(self, enterprise_layer: np.ndarray, capacity_layer: np.ndarray, def _calculate_tardiness(self, enterprise_layer: np.ndarray, capacity_layer: np.ndarray,
quantity_layer: np.ndarray) -> float: quantity_layer: np.ndarray) -> float:
"""计算交付延期T = max(0, 实际交货期 - 需求交货期)""" """
计算交付延期：max(0, 实际交货期 - 需求交货期)
:return: 延期时间（非负）
"""
actual_delivery = self._calculate_actual_delivery_time(enterprise_layer, capacity_layer, quantity_layer) actual_delivery = self._calculate_actual_delivery_time(enterprise_layer, capacity_layer, quantity_layer)
return max(0.0, actual_delivery - self.order.Dd) return max(0.0, actual_delivery - self.order.Dd)

101
visualizer.py
View File

@ -1,93 +1,132 @@
import matplotlib.pyplot as plt import matplotlib.pyplot as plt
import numpy as np import numpy as np
from chromosome_utils import ChromosomeUtils from chromosome_utils import ChromosomeUtils # 染色体工具类
class ResultVisualizer: class ResultVisualizer:
"""结果可视化工具""" """结果可视化工具类：绘制帕累托前沿、收敛曲线，打印最优解详情"""
def __init__(self, utils: ChromosomeUtils): def __init__(self, utils: ChromosomeUtils):
"""
初始化可视化工具
:param utils: 染色体工具类实例（用于解析染色体）
"""
self.utils = utils self.utils = utils
self.figsize = (12, 8) self.figsize = (12, 8) # 图表大小
# 企业名称列表（风险企业+供应商）
self.supplier_names = ["RiskEnterprise"] + self.utils.supplier.names self.supplier_names = ["RiskEnterprise"] + self.utils.supplier.names
def plot_pareto_front(self, all_objectives: list[tuple[float, float]], def plot_pareto_front(self, all_objectives: list[tuple[float, float]],
non_dominated_objectives: list[tuple[float, float]]): non_dominated_objectives: list[tuple[float, float]]):
"""绘制帕累托前沿""" """
绘制帕累托前沿（所有解 vs 帕累托最优解）
:param all_objectives: 所有解的目标值
:param non_dominated_objectives: 帕累托最优解的目标值
"""
plt.figure(figsize=self.figsize) plt.figure(figsize=self.figsize)
# 绘制所有解（蓝色点）
if all_objectives: if all_objectives:
c_all = [obj[0] for obj in all_objectives] c_all = [obj[0] for obj in all_objectives] # 成本
t_all = [obj[1] for obj in all_objectives] t_all = [obj[1] for obj in all_objectives] # 延期
plt.scatter(c_all, t_all, color='blue', alpha=0.3, label='All Solutions', zorder=1) plt.scatter(c_all, t_all, color='blue', alpha=0.3, label='All Solutions', zorder=1)
# 绘制帕累托前沿（红色点）
if non_dominated_objectives: if non_dominated_objectives:
c_nd = [obj[0] for obj in non_dominated_objectives] c_nd = [obj[0] for obj in non_dominated_objectives]
t_nd = [obj[1] for obj in non_dominated_objectives] t_nd = [obj[1] for obj in non_dominated_objectives]
plt.scatter(c_nd, t_nd, color='red', s=50, label='Pareto Front', zorder=5) plt.scatter(c_nd, t_nd, color='red', s=50, label='Pareto Front', zorder=5)
# 图表设置
plt.title('Pareto Front: Change Cost vs Tardiness') plt.title('Pareto Front: Change Cost vs Tardiness')
plt.xlabel('Change Cost') plt.xlabel('Change Cost')
plt.ylabel('Tardiness') plt.ylabel('Tardiness')
plt.legend() plt.legend()
plt.grid(True, alpha=0.5) plt.grid(True, alpha=0.5)
plt.savefig('pareto_front.png', dpi=300, bbox_inches='tight') plt.savefig('pareto_front.png', dpi=300, bbox_inches='tight') # 保存图片
plt.show() plt.show()
def plot_convergence(self, convergence_history: list[tuple[float, float]]): def plot_convergence(self, convergence_history: list[tuple[float, float]]):
"""绘制收敛趋势图""" """
if len(convergence_history) == 0: 绘制收敛趋势图（每代最优前沿的平均成本和延期）
:param convergence_history: 收敛历史数据（(平均成本, 平均延期)）
"""
if len(convergence_history) == 0: # 无数据时不绘制
return return
plt.figure(figsize=self.figsize) plt.figure(figsize=self.figsize)
generations = list(range(len(convergence_history))) generations = list(range(len(convergence_history))) # 代数
costs = [h[0] for h in convergence_history] costs = [h[0] for h in convergence_history] # 平均成本
tardiness = [h[1] for h in convergence_history] tardiness = [h[1] for h in convergence_history] # 平均延期
# 子图1：平均成本趋势
plt.subplot(2, 1, 1) plt.subplot(2, 1, 1)
plt.plot(generations, costs, 'b-') plt.plot(generations, costs, 'b-')
plt.title('Convergence History') plt.title('Convergence History')
plt.ylabel('Average Change Cost') plt.ylabel('Average Change Cost')
plt.grid(True, alpha=0.5) plt.grid(True, alpha=0.5)
# 子图2：平均延期趋势
plt.subplot(2, 1, 2) plt.subplot(2, 1, 2)
plt.plot(generations, tardiness, 'r-') plt.plot(generations, tardiness, 'r-')
plt.xlabel('Generation') plt.xlabel('Generation')
plt.ylabel('Average Tardiness') plt.ylabel('Average Tardiness')
plt.grid(True, alpha=0.5) plt.grid(True, alpha=0.5)
plt.tight_layout()
plt.savefig('convergence_history.png', dpi=300, bbox_inches='tight') plt.tight_layout() # 调整布局
plt.savefig('convergence_history.png', dpi=300, bbox_inches='tight') # 保存图片
plt.show() plt.show()
def print_solution_details(self, solution: np.ndarray, objectives: tuple[float, float]): def print_solution_details(self, solution: np.ndarray, objectives: tuple[float, float]):
"""打印单个解的详细信息（含数量校验）""" """
打印单个解的详细信息（企业选择、产能、数量等）
:param solution: 染色体（解）
:param objectives: 该解的目标值（成本, 延期）
"""
# 拆分染色体为三层
enterprise_layer, capacity_layer, quantity_layer = self.utils._split_chromosome(solution) enterprise_layer, capacity_layer, quantity_layer = self.utils._split_chromosome(solution)
split_points = np.cumsum(self.utils.material_enterprise_count) split_points = np.cumsum(self.utils.material_enterprise_count) # 分割点
print(f"\n变更成本: {objectives[0]:.2f}, 交付延期: {objectives[1]:.2f}") print(f"\n变更成本: {objectives[0]:.2f}, 交付延期: {objectives[1]:.2f}")
print("=" * 80) print("=" * 80)
total_q_check = [] total_q_check = [] # 检查各物料的总数量是否满足需求
start = 0 start = 0
for i in range(self.utils.I): for i in range(self.utils.I): # 遍历每种物料
end = split_points[i] end = split_points[i]
ents = self.utils.material_optional_enterprises[i] ents = self.utils.material_optional_enterprises[i] # 可选企业
e_segment = enterprise_layer[start:end] e_segment = enterprise_layer[start:end] # 企业选择状态
c_segment = capacity_layer[start:end] c_segment = capacity_layer[start:end] # 产能
q_segment = quantity_layer[start:end] q_segment = quantity_layer[start:end] # 数量
demand_q = self.utils.order.Q[i]
print(f"物料 {i} - 需求数量: {demand_q}, 分配总量: {np.sum(q_segment[e_segment==1]):.2f}") demand_q = self.utils.order.Q[i] # 需求数量
total_q_check.append(abs(np.sum(q_segment[e_segment==1]) - demand_q) < 1e-2) allocated_q = np.sum(q_segment[e_segment == 1]) # 分配的总数量
print(f"物料 {i} - 需求数量: {demand_q}, 分配总量: {allocated_q:.2f}")
total_q_check.append(abs(allocated_q - demand_q) < 1e-2) # 检查是否满足需求
print(f" 选择的企业及其分配:") print(f" 选择的企业及其分配:")
for idx, ent in enumerate(ents): for idx, ent in enumerate(ents):
if e_segment[idx] == 1: if e_segment[idx] == 1: # 仅打印被选中的企业
ent_name = self.supplier_names[ent] ent_name = self.supplier_names[ent] # 企业名称
cap = c_segment[idx] cap = c_segment[idx] # 产能
qty = q_segment[idx] qty = q_segment[idx] # 数量
print(f" 企业: {ent_name}, 产能: {cap:.2f}, 分配数量: {qty:.2f}") print(f" 企业: {ent_name}, 产能: {cap:.2f}, 分配数量: {qty:.2f}")
start = end start = end
print("-" * 80) print("-" * 80)
# 验证数量约束是否满足
if all(total_q_check): if all(total_q_check):
print("✅ 所有物料数量满足需求约束") print("✅ 所有物料数量满足需求约束")
else: else:
print("❌ 部分物料数量未满足需求约束") print("❌ 部分物料数量未满足需求约束")
def print_pareto_solutions(self, population: np.ndarray, objectives: list[tuple[float, float]]): def print_pareto_solutions(self, population: np.ndarray, objectives: list[tuple[float, float]]):
"""打印帕累托前沿解的详细信息""" """
打印帕累托前沿解的详细信息（最多打印前5个）
:param population: 帕累托前沿解的种群
:param objectives: 对应的目标值列表
"""
print("\n" + "=" * 100) print("\n" + "=" * 100)
print("帕累托前沿解详细方案") print("帕累托前沿解详细方案")
print("=" * 100) print("=" * 100)
# 打印前5个解（或所有解，取较少者）
for i in range(min(5, len(population))): for i in range(min(5, len(population))):
print(f"\n===== 最优解 {i+1} =====") print(f"\n===== 最优解 {i + 1} =====")
self.print_solution_details(population[i], objectives[i]) self.print_solution_details(population[i], objectives[i])