Python 中最小成本流優化導論

最小成本流優化是指在一個由節點和邊組成的網絡中，最小化流動的運輸成本。節點包括供應源（來源）和需求點（匯），每個節點都有不同的成本和容量限制。其目標是找到從來源到匯的最低成本流動方式，同時遵守所有的容量限制。

應用

最小成本流優化的應用非常廣泛，涵蓋了多個行業和領域。這種方法在物流和供應鏈管理中至關重要，因為它用於最小化運輸成本，同時確保貨物及時送達。在電信領域，它有助於優化數據在網絡中的路由，以減少延遲並提高帶寬利用率。能源部門利用最小成本流優化有效地分配電力，減少損失和運營成本。城市規劃和基礎設施發展也受益於這種優化技術，因為它有助於設計高效的公共交通系統和水分配網絡。

範例

以下是一個簡單的流動優化範例：

上面的圖片展示了一個最小成本流優化問題，包含六個節點和八條邊。節點A和B作為來源，每個的供應量為50單位，而節點E和F作為匯，每個的需求量為40單位。每條邊的最大容量為25單位，變動成本在圖片中顯示。優化的目標是分配每條邊的流量，將所需的單位從節點A和B運送到節點E和F，同時遵守邊的容量限制，並以最低的成本進行。

節點F只能從節點B接收供應。有兩條路徑：直接或通過節點D。直接路徑的成本為2，而通過D的間接路徑的總成本為3。因此，25單位（邊的最大容量）直接從B運送到F。剩下的15單位則通過B-D-F路徑來滿足需求。

目前，從節點B轉移了40單位，剩下的10單位可以運送到節點E。供應節點E的可用路徑包括：A-E和B-E，成本為3，A-C-E的成本為4，B-C-E的成本為5。因此，從A-E運送25單位（受到邊的容量限制），從B-E運送10單位（受到節點B剩餘供應的限制）。為了滿足節點E的40單位需求，還需要通過A-C-E再運送5單位，這樣就沒有流量分配到B-C路徑。

數學公式化

我介紹兩種最小成本流優化的數學公式：

1. LP（線性規劃），僅包含連續變數

2. MILP（混合整數線性規劃），包含連續和離散變數

我使用以下定義：

LP公式化

這個公式僅包含連續的決策變數，意味著它們可以有任何值，只要滿足所有約束條件。決策變數在這裡是所有邊的流量變數x(u, v)。

目標函數描述了應該最小化的成本是如何計算的。在這種情況下，它被定義為流量乘以變動成本，並在所有邊上進行求和：

約束條件是必須滿足的條件，以確保解決方案是有效的，確保流量不超過容量限制。

首先，所有流量必須是非負的，並且不超過邊的容量：

流量守恆約束確保進入一個節點的流量必須等於離開該節點的流量。這些約束適用於所有既不是來源也不是匯的節點：

對於來源和匯節點，流出的流量和流入的流量之差必須小於或等於該節點的供應量：

如果v是來源，則流出的流量減去流入的流量的差值不得超過供應s(v)。如果v是匯節點，我們不允許流入的流量超過流出的流量（對於匯節點，s(v)是負的）。

MILP

此外，MILP公式化除了包含LP公式的連續變數外，還包含只能有特定值的離散變數。離散變數允許限制使用的節點或邊的數量到某些值。它也可以用來引入使用節點或邊的固定成本。在這篇文章中，我展示了如何添加固定成本。需要注意的是，添加離散決策變數會使找到最佳解變得更加困難，因此這種公式應該僅在LP公式不可能的情況下使用。

目標函數定義為：

包含三個項目：所有邊的變動成本、所有邊的固定成本和所有節點的固定成本。

可以分配給邊的最大流量取決於邊的容量、邊的選擇變數和來源節點的選擇變數：

這個方程確保只有當邊的選擇變數和來源節點的選擇變數為1時，流量才能分配到邊上。

流量守恆約束與LP問題相同。

實施

在這一部分，我將解釋如何在Python中實施MILP優化。你可以在這個repo中找到代碼。

庫

為了構建流動網絡，我使用了NetworkX，這是一個優秀的庫（https://networkx.org/），用於處理圖形。有許多有趣的文章展示了NetworkX在處理圖形時的強大和易用性，例如自定義NetworkX圖形、NetworkX：操作子圖的代碼演示、使用NetworkX進行社交網絡分析：簡明介紹。

構建優化時的一個重要方面是確保輸入正確定義。即使是一個小錯誤也可能使問題無法實現或導致意外的解決方案。為了避免這種情況，我使用了Pydantic來驗證用戶輸入，並在最早的階段提出任何問題。這篇文章提供了對Pydantic的易於理解的介紹。

為了將定義的網絡轉換為數學優化問題，我使用了PuLP。這使得以直觀的方式定義所有變數和約束。這個庫還有一個優勢，就是可以以簡單的即插即用方式使用許多不同的求解器。這篇文章提供了對這個庫的良好介紹。

定義節點和邊

以下代碼展示了如何定義節點：

from pydantic import BaseModel, model_validator
from typing import Optional

# 節點和邊的定義
class Node(BaseModel, frozen=True):
“””
網絡節點的類，具有以下屬性：
name: str – 節點名稱
demand: float – 節點需求（如果節點是匯）
supply: float – 節點供應（如果節點是來源）
capacity: float – 從節點流出的最大流量
type: str – 節點類型
x: float – 節點的x坐標
y: float – 節點的y坐標
fixed_cost: float – 選擇節點的成本
“””
name: str
demand: Optional[float] = 0.0
supply: Optional[float] = 0.0
capacity: Optional[float] = float(‘inf’)
type: Optional[str] = None
x: Optional[float] = 0.0
y: Optional[float] = 0.0
fixed_cost: Optional[float] = 0.0

@model_validator(mode=”after”)
def validate(self):
“””
驗證節點定義是否正確
“””
# 檢查需求是否為非負
if self.demand < 0 or self.demand == float('inf'): raise ValueError('需求必須是非負且有限的')
# 檢查供應是否為非負
if self.supply < 0: raise ValueError('供應必須是非負的')
# 檢查容量是否為非負
if self.capacity < 0: raise ValueError('容量必須是非負的')
# 檢查固定成本是否為非負
if self.fixed_cost < 0: raise ValueError('固定成本必須是非負的')
return self

節點通過Node類定義，該類繼承自Pydantic的BaseModel。這使得自動驗證成為可能，確保每當創建新對象時，所有屬性都以正確的數據類型定義。在這種情況下，只有名稱是必需的輸入，所有其他屬性都是可選的，如果未提供，則會分配指定的默認值。通過將“frozen”參數設置為True，我使所有屬性不可變，這意味著它們在對象初始化後不能被更改。

validate方法在對象初始化後執行，並進行更多檢查，以確保提供的值符合預期。具體來說，它檢查需求、供應、容量、變動成本和固定成本是否為非負。此外，它還不允許無限需求，因為這會導致無法實現的優化問題。

這些檢查看起來微不足道，但它們的主要好處是當輸入不正確時，會在最早的階段觸發錯誤。因此，它們防止創建不正確的優化模型。探索為什麼模型無法解決會耗費更多時間，因為有許多因素需要分析，而這種“微不足道”的輸入錯誤可能不是第一個需要調查的方面。

邊的實現如下：

class Edge(BaseModel, frozen=True):
“””兩個節點之間的邊的類，具有以下屬性：
origin: ‘Node’ – 邊的起始節點
destination: ‘Node’ – 邊的終止節點
capacity: float – 通過邊的最大流量
variable_cost: float – 通過邊的每單位流量成本
fixed_cost: float – 選擇邊的成本
“””
origin: Node
destination: Node
capacity: Optional[float] = float(‘inf’)
variable_cost: Optional[float] = 0.0
fixed_cost: Optional[float] = 0.0

@model_validator(mode=”after”)
def validate(self):
“””驗證邊的定義是否正確”””
# 檢查節點名稱是否不同
if self.origin.name == self.destination.name: raise ValueError(‘起始和終止名稱必須不同’)
# 檢查容量是否為非負
if self.capacity < 0: raise ValueError('容量必須是非負的')
# 檢查變動成本是否為非負
if self.variable_cost < 0: raise ValueError('變動成本必須是非負的')
# 檢查固定成本是否為非負
if self.fixed_cost < 0: raise ValueError('固定成本必須是非負的')
return self

所需的輸入是起始節點和終止節點對象。此外，可以提供容量、變動成本和固定成本。容量的默認值為無限，這意味著如果未提供容量值，則假設該邊沒有容量限制。驗證確保提供的值為非負，並且起始節點名稱和終止節點名稱不同。

流圖對象的初始化

為了定義流圖並優化流動，我創建了一個名為FlowGraph的新類，該類繼承自NetworkX的DiGraph類。通過這樣做，我可以添加特定於流優化的方法，同時使用DiGraph提供的所有方法：

from networkx import DiGraph
from pulp import LpProblem, LpVariable, LpMinimize, LpStatus

class FlowGraph(DiGraph):
“””
定義和解決最小成本流問題的類
“””
def __init__(self, nodes=[], edges=[]):
“””
初始化FlowGraph對象
:param nodes: 節點列表
:param edges: 邊列表
“””
# 初始化有向圖
super().__init__(None)

# 添加節點和邊
for node in nodes: self.add_node(node)
for edge in edges: self.add_edge(edge)

def add_node(self, node):
“””
將節點添加到圖中
:param node: Node對象
“””
# 檢查節點是否為Node對象
if not isinstance(node, Node): raise ValueError(‘節點必須是Node對象’)
# 將節點添加到圖中
super().add_node(node.name, demand=node.demand, supply=node.supply, capacity=node.capacity, type=node.type,
fixed_cost=node.fixed_cost, x=node.x, y=node.y)

def add_edge(self, edge):
“””
將邊添加到圖中
@param edge: Edge對象
“””
# 檢查邊是否為Edge對象
if not isinstance(edge, Edge): raise ValueError(‘邊必須是Edge對象’)
# 檢查節點是否存在
if not edge.origin.name in super().nodes: self.add_node(edge.origin)
if not edge.destination.name in super().nodes: self.add_node(edge.destination)

# 將邊添加到圖中
super().add_edge(edge.origin.name, edge.destination.name, capacity=edge.capacity,
variable_cost=edge.variable_cost, fixed_cost=edge.fixed_cost)

FlowGraph通過提供節點和邊的列表來初始化。第一步是將父類初始化為空圖。接下來，通過add_node和add_edge方法添加節點和邊。這些方法首先檢查提供的元素是否為Node或Edge對象。如果不是，則會引發錯誤。這確保所有添加到圖中的元素都通過了前一部分的驗證。接下來，這些對象的值被添加到DiGraph對象中。請注意，DiGraph類也使用add_node和add_edge方法來做到這一點。通過使用相同的方法名稱，我覆蓋了這些方法，以確保每當向圖中添加新元素時，必須通過FlowGraph方法添加，這樣可以驗證對象類型。因此，不可能構建一個包含未通過驗證測試的任何元素的圖。

初始化優化問題

以下方法將網絡轉換為優化模型，解決它並檢索優化值。

def min_cost_flow(self, verbose=True):
“””
運行最小成本流優化
@param verbose: bool – 打印優化狀態（默認：True）
@return: 優化狀態
“””
self.verbose = verbose

# 獲取最大流量
self.max_flow = sum(node[‘demand’] for _, node in super().nodes.data() if node[‘demand’] > 0)

start_time = time.time()
# 創建LP問題
self.prob = LpProblem(“FlowGraph.min_cost_flow”, LpMinimize)
# 分配決策變數
self._assign_decision_variables()
# 分配目標函數
self._assign_objective_function()
# 分配約束
self._assign_constraints()
if self.verbose: print(f”模型創建時間: {time.time() – start_time:.2f} s”)

start_time = time.time()
# 解決LP問題
self.prob.solve()
solve_time = time.time() – start_time

# 獲取狀態
status = LpStatus[self.prob.status]

if verbose:
# 打印優化狀態
if status == ‘Optimal’:
# 獲取目標值
objective = self.prob.objective.value()
print(f”找到最佳解：{objective:.2f}，耗時 {solve_time:.2f} s”)
else:
print(f”優化狀態：{status}，耗時 {solve_time:.2f} s”)

# 分配變數值
self._assign_variable_values(status==’Optimal’)

return status

Pulp的LpProblem被初始化，常數LpMinimize將其定義為最小化問題——這意味著它應該最小化目標函數的值。在接下來的行中，所有決策變數都被初始化，目標函數以及所有約束都被定義。這些方法將在接下來的部分中解釋。

接下來，問題被解決，在這一步中，所有決策變數的最佳值被確定。隨後檢索優化的狀態。當狀態為“Optimal”時，表示找到了一個最佳解，其他狀態包括“Infeasible”（無法滿足所有約束）、“Unbounded”（目標函數可以有任意低的值）和“Undefined”，表示問題定義不完整。如果未找到最佳解，則需要檢查問題定義。

最後，檢索所有變數的優化值並分配給相應的節點和邊。

定義決策變數

所有決策變數在以下方法中初始化：

def _assign_variable_values(self, opt_found):
“””
如果找到最佳解，則分配決策變數值，否則設置為None
@param opt_found: bool – 是否找到最佳解
“””
# 分配邊的值
for _, _, edge in super().edges.data():
# 初始化值
edge[‘flow’] = None
edge[‘selected’] = None
# 檢查是否找到最佳解
if opt_found and edge[‘flow_var’] is not None:
edge[‘flow’] = edge[‘flow_var’].varValue

if edge[‘selection_var’] is not None:
edge[‘selected’] = edge[‘selection_var’].varValue

# 分配節點的值
for _, node in super().nodes.data():
# 初始化值
node[‘selected’] = None
if opt_found:
# 檢查節點是否有選擇變數
if node[‘selection_var’] is not None:
node[‘selected’] = node[‘selection_var’].varValue

首先，它遍歷所有邊，並在邊的容量大於0時分配連續決策變數。此外，如果邊的固定成本大於0，則還定義了一個二元決策變數。接下來，它遍歷所有節點，並將二元決策變數分配給具有固定成本的節點。最後，計算並打印連續和二元決策變數的總數。

定義目標

在所有決策變數初始化後，可以定義目標函數：

def _assign_objective_function(self):
“””
定義目標函數
“””
objective = 0

# 添加邊的成本
for _, _, edge in super().edges.data():
if edge[‘selection_var’] is not None: objective += edge[‘selection_var’] * edge[‘fixed_cost’]
if edge[‘flow_var’] is not None: objective += edge[‘flow_var’] * edge[‘variable_cost’]

# 添加節點的成本
for _, node in super().nodes.data():
# 添加節點選擇成本
if node[‘selection_var’] is not None: objective += node[‘selection_var’] * node[‘fixed_cost’]

self.prob += objective, ‘目標’,

目標初始化為0。然後，對於每條邊，如果邊有選擇變數，則添加固定成本；如果邊有流量變數，則添加變動成本。對於所有具有選擇變數的節點，固定成本也被添加到目標中。在方法的最後，目標被添加到LP對象中。

定義約束

所有約束在以下方法中定義：

def _assign_constraints(self):
“””
定義約束
“””
# 約束數量計數
constr_count = 0
# 為具有固定成本的邊添加容量約束
for origin_name, destination_name, edge in super().edges.data():
# 獲取容量
capacity = edge[‘capacity’] if edge[‘capacity’] < float('inf') else self.max_flow
rhs = capacity
if edge[‘selection_var’] is not None: rhs *= edge[‘selection_var’]
self.prob += edge[‘flow_var’] <= rhs, f"capacity_{origin_name}-{destination_name}",
constr_count += 1

# 獲取起始節點
origin_node = super().nodes[origin_name]
# 檢查起始節點是否有選擇變數
if origin_node[‘selection_var’] is not None:
rhs = capacity * origin_node[‘selection_var’]
self.prob += (edge[‘flow_var’] <= rhs, f"node_selection_{origin_name}-{destination_name}",)
constr_count += 1

total_demand = total_supply = 0
# 添加流量守恆約束
for node_name, node in super().nodes.data():
# 聚合進入和流出的流量
in_flow = 0
for _, _, edge in super().in_edges(node_name, data=True):
if edge[‘flow_var’] is not None: in_flow += edge[‘flow_var’]

out_flow = 0
for _, _, edge in super().out_edges(node_name, data=True):
if edge[‘flow_var’] is not None: out_flow += edge[‘flow_var’]

# 添加節點容量約束
if node[‘capacity’] < float('inf'):
self.prob += out_flow <= node['capacity'], f"node_capacity_{node_name}",
constr_count += 1

# 檢查節點類型
if node[‘demand’] == node[‘supply’]:
# 轉運節點：進流量 = 出流量
self.prob += in_flow == out_flow, f”flow_balance_{node_name}”,
else:
# 進流量 – 出流量 >= 需求 – 供應
rhs = node[‘demand’] – node[‘supply’]
self.prob += in_flow – out_flow >= rhs, f”flow_balance_{node_name}”,
constr_count += 1

# 更新總需求和供應
total_demand += node[‘demand’]
total_supply += node[‘supply’]

if self.verbose:
print(f”約束數量: {constr_count}”)
print(f”總供應: {total_supply}, 總需求: {total_demand}”)

首先，為每條邊定義容量約束。如果邊有選擇變數，則容量與該變數相乘。如果沒有容量限制（容量設置為無限），但有選擇變數，則該選擇變數與通過聚合所有節點的需求計算的最大流相乘。如果邊的起始節點有選擇變數，則添加額外的約束，這意味著流量只能從該節點流出，如果選擇變數設置為1。

接下來，為所有節點定義流量守恆約束。為此，計算節點的總進流和出流。獲取所有進出邊可以通過使用DiGraph類的in_edges和out_edges方法輕鬆完成。如果節點有容量限制，則最大出流將受到該值的約束。對於流量守恆，需要檢查節點是來源、匯還是轉運節點（需求等於供應）。在第一種情況下，進流量和出流量之間的差必須大於或等於需求和供應之間的差，而在後者的情況下，進流量和出流量必須相等。

總約束數量在方法的最後進行計數並打印。

檢索優化值

運行優化後，可以使用以下方法檢索優化的變數值：

def _assign_variable_values(self, opt_found):
“””
如果找到最佳解，則分配決策變數值，否則設置為None
@param opt_found: bool – 是否找到最佳解
“””
# 分配邊的值
for _, _, edge in super().edges.data():
# 初始化值
edge[‘flow’] = None
edge[‘selected’] = None
# 檢查是否找到最佳解
if opt_found and edge[‘flow_var’] is not None:
edge[‘flow’] = edge[‘flow_var’].varValue

if edge[‘selection_var’] is not None:
edge[‘selected’] = edge[‘selection_var’].varValue

# 分配節點的值
for _, node in super().nodes.data():
# 初始化值
node[‘selected’] = None
if opt_found:
# 檢查節點是否有選擇變數
if node[‘selection_var’] is not None:
node[‘selected’] = node[‘selection_var’].varValue

這個方法遍歷所有邊和節點，檢查決策變數是否已被分配，並通過varValue將決策變數值添加到相應的邊或節點。

示範

為了演示如何應用流優化，我創建了一個供應鏈網絡，由2個工廠、4個配送中心（DC）和15個市場組成。所有由工廠生產的貨物必須通過一個配送中心，然後才能送達市場。

節點屬性被定義：

範圍意味著生成均勻分佈的隨機數來分配這些屬性。由於工廠和DC有固定成本，因此優化還需要決定應該選擇哪些實體。

在所有工廠和DC之間以及所有DC和市場之間生成邊。邊的變動成本計算為起始節點和終止節點之間的歐幾里得距離。從工廠到DC的邊的容量設置為350，而從DC到市場的邊的容量設置為100。

以下代碼展示了如何定義網絡以及如何運行優化：

# 定義節點
factories = [Node(name=f’工廠 {i}’, supply=700, type=”工廠”, fixed_cost=100, x=random.uniform(0, 2),
y=random.uniform(0, 1)) for i in range(2)]
dcs = [Node(name=f’DC {i}’, fixed_cost=25, capacity=500, type=”DC”, x=random.uniform(0, 2),
y=random.uniform(0, 1)) for i in range(4)]
markets = [Node(name=f’市場 {i}’, demand=random.randint(1, 100), type=”市場”, x=random.uniform(0, 2),
y=random.uniform(0, 1)) for i in range(15)]

# 定義邊
edges = []
# 工廠到DC
for factory in factories:
for dc in dcs:
distance = ((factory.x – dc.x)**2 + (factory.y – dc.y)**2)**0.5
edges.append(Edge(origin=factory, destination=dc, capacity=350, variable_cost=distance))

# DC到市場
for dc in dcs:
for market in markets:
distance = ((dc.x – market.x)**2 + (dc.y – market.y)**2)**0.5
edges.append(Edge(origin=dc, destination=market, capacity=100, variable_cost=distance))

# 創建FlowGraph
G = FlowGraph(edges=edges)

G.min_cost_flow()

流優化的輸出如下：

變數類型：68個連續變數，6個二元變數
約束數量：161
總供應：1400.0，總需求：909.0
模型創建時間：0.00 s
找到最佳解：1334.88，耗時0.23 s

該問題由68個連續變數組成，這些是邊的流量變數，還有6個二元決策變數，這些是工廠和DC的選擇變數。總共有161個約束，這些約束由邊和節點的容量約束、節點選擇約束（邊只能在起始節點被選擇的情況下有流量）和流量守恆約束組成。下一行顯示總供應為1400，這高於總需求909（如果需求高於供應，則問題將無法實現）。由於這是一個小型優化問題，定義優化模型的時間少於0.01秒。最後一行顯示，在0.23秒內找到了一個目標值為1335的最佳解。

此外，在這篇文章中描述的代碼中，我還添加了兩個可視化優化解決方案的方法。這些方法的代碼也可以在repo中找到。

所有節點根據其各自的x和y坐標定位。節點和邊的大小與流動的總量相關。邊的顏色表示其利用率（流量與容量的比率）。虛線表示沒有流量分配的邊。

在最佳解中，兩個工廠都被選中，因為一個工廠的最大供應量為700，而總需求為909。然而，只有4個DC中的3個被使用（DC 0未被選中）。

一般來說，圖表顯示工廠向最近的DC供應，而DC則向最近的市場供應。然而，這一觀察有一些例外：工廠0也向DC 3供應，儘管工廠1更近。這是由於邊的容量限制，只允許每條邊最多移動350單位。然而，DC 3最近的市場需求略高，因此工廠0向DC 3移動額外的單位以滿足該需求。儘管市場9最接近DC 3，但它由DC 2供應。這是因為DC 3需要工廠0的額外供應來供應這個市場，而從工廠0經過DC 3的總距離長於從工廠0經過DC 2的距離，因此市場9通過後者路徑供應。

另一種可視化結果的方法是通過桑基圖，這種圖專注於可視化邊的流量：

顏色表示邊的利用率，最低利用率為綠色，變為黃色和紅色表示最高利用率。這個圖很好地顯示了每個節點和邊的流量。它突出了從工廠0到DC 3的流量，還顯示市場13由DC 2和DC 1供應。

總結

最小成本流優化可以在許多領域中成為非常有用的工具，例如物流、運輸、電信、能源部門等。應用這種優化時，重要的是將物理系統轉換為由節點和邊組成的數學圖形。這應該以儘可能少的離散（例如二元）決策變數進行，因為這些變數會顯著增加找到最佳解的難度。通過結合Python的NetworkX、Pulp和Pydantic庫，我構建了一個流優化類，這個類直觀易用，同時遵循通用公式，允許在許多不同的用例中應用。圖形和流圖對於理解優化器找到的解決方案非常有幫助。

如果沒有其他說明，所有圖片均由作者創建。