最小费用流问题描述
在网络模型中，除了考虑网络上各边的容量以外，通常还需要考虑各边上流动的费用。最小费用流问题就是在给定网络模型中各节点的需求量和供应量的情况下，如何分配流量和路径，使得费用达到最小的问题。
在由个节点的点集合
和
个边的边集合
组成的图
中，当边
上的最大容量
和费用
给定时，最小费用流问题的数学模型可以描述为：
其中代表从节点
到节点
的流量，
和
分别为节点
的后续节点集和前续节点集，
表示从其他节点进入到节点
的总流量，而
表示从节点
流出的流量。
求解最小费用流问题的代表性算法有Successive shortest path algorithm，Primal-dual法和Out of Kilter法等。
最小费用流问题示例
一个最小费用流问题的示例问题如图所示：
MinimumCostFlow_description
遗传算法求解最小费用流问题
个体编码
此处依然采用优先级编码，对各个节点给予优先级；在生成路径时根据优先级挑选路径下一步的节点。
解码
同最大流量问题的解码。
其他遗传算法操作
- 评价函数：解码过程中得到的各边流量和代价的乘积之和
- 育种选择：binary锦标赛选择
- 变异算法：交叉-有序交叉(OX)，突变-乱序突变(Shuffle Index)
- 环境选择：子代完全替换父代(无精英保留)
代码示例
完整代码如下：
## 环境设定
import numpy as np
import matplotlib.pyplot as plt
from deap import base, tools, creator, algorithms
import random
params = {
'font.family': 'serif',
'figure.dpi': 300,
'savefig.dpi': 300,
'font.size': 12,
'legend.fontsize': 'small'
}
plt.rcParams.update(params)
import copy
# --------------------
# 问题定义
creator.create('FitnessMax', base.Fitness, weights=(-1.0,)) # 最小化问题
creator.create('Individual', list, fitness=creator.FitnessMax)
# 个体编码
toolbox = base.Toolbox()
geneLength = 25
toolbox.register('perm', np.random.permutation, geneLength)
toolbox.register('individual', tools.initIterate,
creator.Individual, toolbox.perm)
# 评价函数
# 类似链表，存储每个节点的可行路径，用于解码
nodeDict = {'1': [2,3,4,5,6], '2': [7,8], '3': [7,8,9], '4': [8,9,10], '5': [9,10,11],
'6': [10,11], '7': [13], '8': [12,14], '9': [13,15], '10': [14,16],
'11':[15], '12':[17], '13':[17,18], '14':[18,19], '15':[19,20],
'16':[20,21], '17':[18,22], '18':[19,22,23], '19':[20,23,24],
'20':[23,24], '21':[24], '22':[25], '23':[25], '24':[25]
}
def genPath(ind, nodeDict=nodeDict):
'''输入一个优先度序列之后，返回一条从节点1到节点25的可行路径 '''
path = [1]
endNode = len(ind)
while not path[-1] == endNode:
curNode = path[-1] # 当前所在节点
if nodeDict[str(curNode)]: # 当前节点指向的下一个节点不为空时，到达下一个节点
toBeSelected = nodeDict[str(curNode)] # 获取可以到达的下一个节点列表
else:
return path
priority = np.asarray(ind)[np.asarray(
toBeSelected)-1] # 获取优先级,注意列表的index是0-9
nextNode = toBeSelected[np.argmax(priority)]
path.append(nextNode)
return path
# 存储每条边的剩余容量，用于计算路径流量和更新节点链表
capacityDict = {'1,2': 20, '1,3': 20, '1,4': 20, '1,5': 20, '1,6': 20,
'2,7': 10, '2,8': 8,
'3,7': 6, '3,8': 5, '3,9': 4,
'4,8': 5, '4,9': 8, '4,10': 10,
'5,9': 10, '5,10': 4, '5,11': 10,
'6,10': 10, '6,11': 10,
'7,13': 15,
'8,12': 15, '8,14': 15,
'9,13': 15, '9,15': 15,
'10,14': 15, '10,16': 15,
'11,15': 15,
'12,17': 20,
'13,17': 20, '13,18': 20,
'14,18': 20, '14,19': 20,
'15,19': 20, '15,20': 20,
'16,20': 20, '16,21': 20,
'17,18': 8, '17,22': 25,
'18,19': 8, '18,22': 25, '18,23': 20,
'19,20': 8, '19,23': 10, '19,24': 25,
'20,23': 10, '20,24': 25,
'21,24': 25,
'22,25': 30, '23,25': 30, '24,25':30}
# 存储每条边的代价
costDict = {'1,2': 10, '1,3': 13, '1,4': 32, '1,5': 135, '1,6': 631,
'2,7': 10, '2,8': 13,
'3,7': 10, '3,8': 15, '3,9': 33,
'4,8': 4, '4,9': 15, '4,10': 35,
'5,9': 3, '5,10': 13, '5,11': 33,
'6,10': 7, '6,11': 7,
'7,13': 10,
'8,12': 4, '8,14': 9,
'9,13': 11, '9,15': 12,
'10,14': 9, '10,16': 14,
'11,15': 5,
'12,17': 8,
'13,17': 6, '13,18': 7,
'14,18': 7, '14,19': 7,
'15,19': 5, '15,20': 14,
'16,20': 4, '16,21': 14,
'17,18': 11, '17,22': 11,
'18,19': 5, '18,22': 8, '18,23': 34,
'19,20': 3, '19,23': 10, '19,24': 35,
'20,23': 3, '20,24': 14,
'21,24': 12,
'22,25': 10, '23,25': 2, '24,25':3}
def traceCapacity(path, capacityDict):
''' 获取给定path的最大流量，更新各边容量 '''
pathEdge = list(zip(path[::1], path[1::1]))
keys = []
edgeCapacity = []
for edge in pathEdge:
key = str(edge[0]) + ',' + str(edge[1])
keys.append(key) # 保存edge对应的key
edgeCapacity.append(capacityDict[key]) # 该边对应的剩余容量
pathFlow = min(edgeCapacity) # 路径上的最大流量
# 更新各边的剩余容量
for key in keys:
capacityDict[key] -= pathFlow # 注意这里是原位修改
return pathFlow
def updateNodeDict(capacityDict, nodeDict):
''' 对剩余流量为0的节点，删除节点指向；对于链表指向为空的节点，由于没有下一步可以移动的方向，
从其他所有节点的指向中删除该节点
'''
for edge, capacity in capacityDict.items():
if capacity == 0:
key, toBeDel = str(edge).split(',') # 用来索引节点字典的key，和需要删除的节点toBeDel
if int(toBeDel) in nodeDict[key]:
nodeDict[key].remove(int(toBeDel))
delList = []
for node, nextNode in nodeDict.items():
if not nextNode: # 如果链表指向为空的节点，从其他所有节点的指向中删除该节点
delList.append(node)
for delNode in delList:
for node, nextNode in nodeDict.items():
if delNode in nextNode:
nodeDict[node].remove(delNode)
def calCost(path, pathFlow, costDict):
'''计算给定路径的成本'''
pathEdge = list(zip(path[::1], path[1::1]))
keys = []
edgeCost = []
for edge in pathEdge:
key = str(edge[0]) + ',' + str(edge[1])
keys.append(key) # 保存edge对应的key
edgeCost.append(costDict[key]) # 该边对应的cost
pathCost = sum([eachEdgeCost*pathFlow for eachEdgeCost in edgeCost])
return pathCost
def evaluate(ind, outputPaths=False):
'''评价函数'''
# 初始化所需变量
nodeDictCopy = copy.deepcopy(nodeDict) # 浅复制
capacityDictCopy = copy.deepcopy(capacityDict)
paths = []
pathFlows = []
overallCost = 0
givenFlow = 70 # 需要运送的流量
eps = 1e-5
# 开始循环
while nodeDictCopy['1'] and (abs(givenFlow) > eps):
path = genPath(ind, nodeDictCopy) # 生成路径
# 当路径无法抵达终点，说明经过这个节点已经无法往下走，从所有其他节点的指向中删除该节点
if path[-1] != geneLength:
for node, nextNode in nodeDictCopy.items():
if path[-1] in nextNode:
nodeDictCopy[node].remove(path[-1])
continue
paths.append(path) # 保存路径
pathFlow = traceCapacity(path, capacityDictCopy) # 计算路径最大流量
if givenFlow < pathFlow: # 当剩余流量不能填满该路径的最大流量时，将所有剩余流量分配给该路径
pathFlow = givenFlow
pathFlows.append(pathFlow) # 保存路径的流量
givenFlow -= pathFlow # 更新需要运送的剩余流量
updateNodeDict(capacityDictCopy, nodeDictCopy) # 更新节点链表
# 计算路径上的cost
pathCost = calCost(path, pathFlow, costDict)
overallCost += pathCost
if outputPaths:
return overallCost, paths, pathFlows
return overallCost,
toolbox.register('evaluate', evaluate)
# 迭代数据
stats = tools.Statistics(key=lambda ind:ind.fitness.values)
stats.register('min', np.min)
stats.register('avg', np.mean)
stats.register('std', np.std)
# 生成初始族群
toolbox.popSize = 100
toolbox.register('population', tools.initRepeat, list, toolbox.individual)
pop = toolbox.population(toolbox.popSize)
# 注册工具
toolbox.register('select', tools.selTournament, tournsize=2)
toolbox.register('mate', tools.cxOrdered)
toolbox.register('mutate', tools.mutShuffleIndexes, indpb=0.5)
# --------------------
# 遗传算法参数
toolbox.ngen = 200
toolbox.cxpb = 0.8
toolbox.mutpb = 0.05
# 遗传算法主程序部分
hallOfFame = tools.HallOfFame(maxsize=1)
pop,logbook = algorithms.eaSimple(pop, toolbox, cxpb=toolbox.cxpb, mutpb=toolbox.mutpb,
ngen = toolbox.ngen, stats=stats, halloffame=hallOfFame, verbose=True)
结果如下：
# 输出结果
from pprint import pprint
bestInd = hallOfFame.items[0]
overallCost, paths, pathFlows = eval(bestInd, outputPaths=True)
print('最优解路径为: ')
pprint(paths)
print('各路径上的流量为：'+str(pathFlows))
print('最小费用为: '+str(overallCost))
# 可视化迭代过程
minFit = logbook.select('min')
avgFit = logbook.select('avg')
plt.plot(minFit, 'b-', label='Minimum Fitness')
plt.plot(avgFit, 'r-', label='Average Fitness')
plt.xlabel('# Gen')
plt.ylabel('Fitness')
plt.legend(loc='best')
## 结果：
#最优解路径为:
#[[1, 2, 7, 13, 17, 22, 25],
# [1, 2, 8, 14, 19, 20, 23, 25],
# [1, 3, 7, 13, 17, 22, 25],
# [1, 3, 9, 13, 17, 22, 25],
# [1, 3, 8, 14, 19, 23, 25],
# [1, 4, 9, 13, 17, 22, 25],
# [1, 4, 9, 13, 18, 22, 25],
# [1, 4, 8, 14, 19, 23, 25],
# [1, 4, 8, 12, 17, 22, 25],
# [1, 4, 10, 16, 20, 23, 25],
# [1, 4, 10, 16, 20, 24, 25],
# [1, 5, 9, 13, 18, 19, 23, 25],
# [1, 5, 9, 15, 20, 24, 25],
# [1, 5, 11, 15, 20, 24, 25]]
#各路径上的流量为：[10, 8, 5, 4, 5, 1, 7, 2, 3, 2, 5, 3, 7, 8]
#最小费用为: 6971
MinimumCostProblem_iterations
得到的最优解可视化如下：
MinimumCostFlow_rsl
