如何用Python线性规划减少生产调度中的换产次数？

核心需求 #

使用Python PuLP库构建生产调度模型，在满足产能和订单需求的前提下，有效限制换产次数：通过目标函数添加惩罚项，激励模型减少SKU切换操作，暂不考虑订单交期与优先级。

样本数据 #

销售订单样本 #

SKU Quantities  Due_Dates   Priority
0   SKU1    60000   2023-06-17 23:00:00 1
1   SKU2    30000   2023-06-17 23:00:00 1
2   SKU3    30000   2023-06-17 23:00:00 1

产能数据样本 #

每小时产能1000单位，共5天（120小时）：

StartTime   Capacity    DateTime
0   6/15/2023 0:00  1000    2023-06-15 00:00:00
1   6/15/2023 1:00  1000    2023-06-15 01:00:00
2   6/15/2023 2:00  1000    2023-06-15 02:00:00
3   6/15/2023 3:00  1000    2023-06-15 03:00:00
4   6/15/2023 4:00  1000    2023-06-15 04:00:00

现有代码 #

数据读取代码 #

import pandas as pd
from pulp import LpMinimize, LpBinary, LpProblem, LpStatus, lpSum, LpVariable, value
from datetime import datetime, timedelta

order_book = pd.read_csv('./Formation Scheduler Test/order_book.csv')  # SKU, Quantities, Due_Dates, Priority
order_book['Due_Dates'] = pd.to_datetime(order_book['Due_Dates'])  # 确保Due_Dates为日期时间格式

order_book = order_book.sort_values(by=['Due_Dates','Priority'])
order_book = order_book.reset_index(drop=True)
display(order_book)

capacity_data = pd.read_csv('./Formation Scheduler Test/capacity_data.csv')  # 未来x天的小时产能数据
capacity_data['DateTime'] = pd.to_datetime(capacity_data['StartTime'])
capacity_data = capacity_data.sort_values(by=['DateTime']) 
capacity_data = capacity_data.reset_index(drop=True)

PuLP线性规划代码（原版本） #

prob = LpProblem("Production_Scheduling", LpMinimize)

x = LpVariable.dicts("production", ((i, j) for i in order_book.index for j in capacity_data.index), lowBound=0, cat='Integer')
y = LpVariable.dicts("is_producing", ((i, j) for i in order_book.index for j in capacity_data.index), cat='Binary')
z = LpVariable.dicts("changeover", ((i, j) for i in order_book.index for j in range(len(capacity_data.index) - 1)), cat='Binary')

prob += lpSum([y[(i, j)] - y[(i, j+1)] for i in order_book.index for j in range(len(capacity_data.index) - 1)]) 

# 产量不得超过产能
for j in capacity_data.index:
    prob += lpSum([x[(i, j)] for i in order_book.index]) <= capacity_data.loc[j, 'Capacity']


for i in order_book.index:
    prob += lpSum([x[(i, j)] for j in capacity_data.index]) >= 360


# 产量需满足需求
# for i in order_book.index:
#     prob += lpSum([x[(i, j)] for j in capacity_data.index]) >= order_book.loc[i, 'Quantities']
for i in order_book.index:
    prob += lpSum([x[(i, j)] for j in capacity_data.index]) <= order_book.loc[i, 'Quantities']

for i in order_book.index:
    prob += lpSum([x[(i, j)] for j in capacity_data.index]) >= lpSum([order_book.loc[i, 'Quantities'] * y[(i, j)] for j in capacity_data.index])


# 关联x和y变量
for i in order_book.index:
    for j in capacity_data.index:
        prob += x[(i, j)] <= y[(i, j)] * large_number  # 若y[i,j]为0，则x[i,j]必须为0

# 添加换产约束
for i in order_book.index:
    for j in range(len(capacity_data.index) - 1):
        prob += z[(i, j)] >= y[(i, j)] - y[(i, j+1)]
        prob += z[(i, j)] >= y[(i, j+1)] - y[(i, j)]

prob.solve()

结果查看代码 #

for i in order_book.index:
    for j in capacity_data.index:
        if y[(i, j)].varValue > 0:
            print(f"Produce {x[(i, j)].varValue} units of SKU {i} in hour {j}")

生成结果DataFrame代码 #

# 准备调度表和订单状态表
schedule = []
order_status = []

for i in order_book.index:
    order_produced = 0
    order_late = 0  # 跟踪逾期产量
    for j in capacity_data.index:
        if x[(i,j)].varValue > 0:
            schedule.append([capacity_data.loc[j, 'DateTime'], order_book.loc[i, 'SKU'], x[(i, j)].varValue])
            order_produced += x[(i,j)].varValue
            if capacity_data.loc[j, 'DateTime'] > order_book.loc[i, 'Due_Dates']:
                order_late += x[(i,j)].varValue 
        
    order_status.append([
        order_book.loc[i, 'SKU'],
        order_book.loc[i, 'Quantities'],
        order_produced,
        order_late,
        max(0, order_book.loc[i, 'Quantities'] - order_produced),
        x,
        order_book.loc[i, 'Due_Dates']  # 添加原始交期
    ])


schedule_df = pd.DataFrame(schedule, columns=['DateTime', 'SKU', 'Quantity'])

问题分析与修正方案 #

原代码无法有效限制换产次数的核心问题：

1. 目标函数定义错误：原目标函数未针对换产次数设置惩罚，无法引导模型减少切换。
2. 换产变量定义偏差：原z变量跟踪单个SKU的生产状态变化，而非整条生产线的SKU切换。
3. 缺少单小时生产唯一性约束：未限制每个小时只能生产一个SKU，导致换产逻辑混乱。
4. 需求满足约束错误：原约束设置为产量≤需求，无法保证订单全部完成。

修正后的PuLP代码 #

prob = LpProblem("Production_Scheduling", LpMinimize)

# 定义变量
x = LpVariable.dicts("production", ((i, j) for i in order_book.index for j in capacity_data.index), lowBound=0, cat='Integer')
y = LpVariable.dicts("is_producing", ((i, j) for i in order_book.index for j in capacity_data.index), cat='Binary')
# 重新定义换产变量：z[j]表示第j小时到j+1小时是否换产（1=换产，0=不换产）
z = LpVariable.dicts("changeover", (j for j in range(len(capacity_data.index)-1)), cat='Binary')

# 设置惩罚系数：每次换产的惩罚值，可根据实际成本调整
penalty_cost = 100
# 目标函数：最小化总换产惩罚 + 保证需求满足（这里以换产惩罚为核心目标）
prob += lpSum([z[j] * penalty_cost for j in range(len(capacity_data.index)-1)])

# 1. 产能约束：每小时总产量不超过该小时产能
for j in capacity_data.index:
    prob += lpSum([x[(i, j)] for i in order_book.index]) <= capacity_data.loc[j, 'Capacity']

# 2. 需求满足约束：每个SKU的总产量等于订单需求
for i in order_book.index:
    prob += lpSum([x[(i, j)] for j in capacity_data.index]) == order_book.loc[i, 'Quantities']

# 3. 关联x和y变量：若y[i,j]=0，则x[i,j]必须为0；若y[i,j]=1，x[i,j]可在产能范围内
large_number = 10**6  # 定义足够大的数，需大于最大单小时产能和单SKU需求
for i in order_book.index:
    for j in capacity_data.index:
        prob += x[(i, j)] <= y[(i, j)] * large_number
        prob += x[(i, j)] <= capacity_data.loc[j, 'Capacity']  # 单个SKU每小时产量不超过产能

# 4. 单小时生产唯一性约束：每个小时最多生产一个SKU（单生产线场景）
for j in capacity_data.index:
    prob += lpSum([y[(i, j)] for i in order_book.index]) <= 1

# 5. 换产约束：相邻小时生产不同SKU时，触发换产变量z[j]=1
for j in range(len(capacity_data.index)-1):
    for i in order_book.index:
        for k in order_book.index:
            if i != k:
                # 如果j小时生产i，j+1小时生产k，则z[j]必须为1
                prob += y[(i, j)] + y[(k, j+1)] <= 1 + z[j]

# 求解模型
prob.solve()

# 输出求解状态
print(f"求解状态：{LpStatus[prob.status]}")

修正说明 #

1. 目标函数：以换产次数乘以惩罚系数作为最小化目标，让模型优先选择换产少的方案。
2. 换产变量：全局跟踪相邻小时的生产线切换，而非单个SKU的状态变化。
3. 单小时约束：确保每小时仅生产一个SKU，符合实际生产逻辑，换产定义清晰。
4. 需求约束：使用==保证订单全部完成，满足核心需求。
5. 大M约束：通过逻辑约束关联相邻小时的生产状态，准确触发换产惩罚。

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