排产方案体检仪用Python计算订单延期时长把隐性损失晒在阳光下某精密机械加工厂有 5 台加工中心、每天排 40 个订单。计划员用 PuLP 做排产模型只优化设备利用率最高不检查交付延期。结果方案看起来很美——设备利用率 92%但23% 的订单延期交付客户罚款每月 12 万。后来我写了个延期损失计算器0.5 秒读完排产方案逐单计算延期时长和罚款金额输出延期热力图。厂长看完说原来优化设备利用率代价是每个月白送客户 12 万。—— 参考北京理工大学《运筹学》第 4 章线性规划应用、第 8 章排序与调度问题一、实际应用场景描述排产方案延期损失评估器是任何排产/调度系统上线前的必检项。凡是排了产、交了货、赔了款的地方都是它行业 排产场景 延期痛点 损失量级精密机械 多品种小批量加工 紧急插单打乱计划 客户罚款 5–15 万/月汽车零配件 按序配送JIT 1 个件迟到整条线停 停线损失 10 万/小时电子 EMS 多客户混线生产 高优先级订单被挤 客户流失、索赔医药 批生产质检排队 放行延迟错过船期 整批报废风险食品 短保质期排产 产出晚于最佳发货窗 库存过期损耗化工 连续生产切换 换产拖延致交期违约 合同违约金核心矛盾- 计划员用 PuLP/遗传算法做排产目标通常是设备利用率最高或生产周期最短- 但排产模型里没有延期惩罚或交付窗口约束- 方案出来后没人算到底延期了多少、罚了多少钱- 结果就是设备很忙客户很怒财务很痛。┌──────────────────────────────────────────────────────────────┐│ 排产方案延期损失评估器 · 排产体检仪 ││ ││ 【业务场景】 ││ ┌─────────────────────────────────────────────────────────┐││ │ 输入: 排产方案 订单主数据 │││ │ • 排产方案: 每台设备每段时间加工什么订单 │││ │ • 订单主数据: 下单日期、承诺交期、优先级、罚金单价│││ │ │││ │ 处理管道: │││ │ 1. 解析: 把排产方案还原成每个订单的计划完工时间 │││ │ 2. 比对: 计划完工时间 vs 承诺交期 → 延期天数 │││ │ 3. 计算: 延期天数 × 罚金单价 → 单笔延期损失 │││ │ 4. 统计: 延期订单占比、平均延期、总损失、TOP10 │││ │ │││ │ 输出: │││ │ • 每单延期明细表 │││ │ • 延期统计摘要(总损失/平均延期/延期率) │││ │ • TOP10 高风险延期订单 │││ │ • 可直接喂回PuLP的延期惩罚项系数 │││ └─────────────────────────────────────────────────────────┘││ ││ 【核心矛盾】 ││ • 排产模型: 优化设备利用率/周期, 不管交期 ││ • 客户: 只在乎能不能按时拿到货 ││ • 本程序: 给排产方案做交期体检 — 排产体检仪 │││ ││ 【本程序处理流程】 ││ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐││ │ 解析排产 │──►│ 计算每单 │──►│ 逐单核算 │──►│ 统计汇总 │││ │ 方案 │ │ 计划完工 │ │ 延期损失 │ │ 喂回模型│││ │ │ │ 时间 │ │ 金额 │ │ 系数 │││ └──────────┘ └──────────┘ └──────────┘ └──────────┘│└──────────────────────────────────────────────────────────────┘二、引入痛点含量化对比2.1 现场真实困境某精密机械加工厂计划员原话我们厂有 5 台加工中心每天要排 40 多个订单交期从 3 天到 30 天不等。我用 PuLP 写了个排产模型目标函数是让所有设备的总空闲时间最少——说白了就是把设备塞得越满越好。模型跑得飞快2 秒就出方案设备利用率 92%看着特别漂亮。但执行一周后销售部炸了- 23% 的订单延期交付- 一个德国客户因为延期罚了我们合同额的 5%单笔 3.2 万- 另一个长期客户直接说再延期就换供应商。我回去一看排产表——模型根本不知道这个订单周三必须交它只知道这台机器周四有空。那周我手工对照 200 个订单的交期一条条算延期天数和罚款干了整整 2 天汇总出来一个月隐性罚款 12 万。后来 IT 组写了个 Python 脚本——0.5 秒读完排产方案和订单表自动算出每单延期天数和罚款金额还告诉我如果把这个订单优先级调高能省 8000 元罚款。厂长看完说原来优化设备利用率代价是每个月白送客户 12 万。2.2 人工核算 vs 自动评估量化对比指标 人工核算 Python 自动评估本方案 改善效果200 个订单延期核算 2 天 0.5 秒 -99.99%延期识别准确率 ~85%易漏看 100% 质变延期损失统计 仅总金额 按订单/客户/产线多维 质变延期率 23%事后才知道 排产完即知 前置防控月度隐性罚款 12 万元 方案阶段即规避 消除优化闭环 无算完就完了 延期惩罚系数可喂回 PuLP 质变关键发现排产优化的瓶颈不在求解而在有没有把交期当回事。一旦把延期损失量化、前置、可反馈排产就从设备自嗨变成客户满意 设备高效的双赢。三、核心逻辑讲解大白话版3.1 用大白话解释排产 延期惩罚想象你是外卖平台的调度系统有 5 个骑手、40 个订单- 每个订单都有承诺送达时间——有的 30 分钟有的 60 分钟- 每个骑手都有速度和当前位置- 你的调度算法说让骑手跑最少的空驶里程——听起来很合理对吧但问题来了- 骑手 A 为了少跑空驶把 30 分钟订单和 60 分钟订单串在一起送- 结果 30 分钟那个订单拖到了 50 分钟才送达- 客户给了差评平台罚了骑手 200 块。大白话逻辑1. 排产方案告诉你谁在什么时候干什么2. 承诺交期告诉你客户期望什么时候拿到3. 两者一减完工时间 交期 延期4. 延期 × 罚款单价 真金白银的损失5. 把这个损失加回排产模型的目标函数——下次排产就会怕罚款主动优先保交期。工业现场版- 骑手 设备- 外卖订单 生产订单- 承诺送达时间 承诺交期- 差评罚款 延期违约金- 调度算法 PuLP 排产模型3.2 运筹学模型北理工《运筹学》映射参考北理工《运筹学》第 4 章线性规划应用、第 8 章排序与调度问题带延期惩罚的单机/多机调度模型\begin{aligned}\min \quad Z \sum_{i1}^{n} w_i \cdot \max(0, C_i - d_i) \sum_{j1}^{m} \text{空闲成本}_j \\\text{s.t.} \quad C_i \text{订单 } i \text{ 的完工时间由排产方案决定} \\ d_i \text{订单 } i \text{ 的承诺交期} \\ w_i \text{订单 } i \text{ 的延期惩罚系数元/天} \\ \text{设备能力、先后关系等约束}\end{aligned}关键变量- C_i 订单 i 的实际完工时间从排产方案解析得到- d_i 订单 i 的承诺交期来自订单主数据- \max(0, C_i - d_i) 延期天数负数视为 0- w_i 延期惩罚单价元/天高优先级客户更高。北理工教材要点- 第 4 章 §4.2线性规划在生产计划中的应用- 第 8 章 §8.1单机调度问题 1 \| \| \sum w_i T_i notation- 第 8 章 §8.4带准备时间和交货期的调度- 本程序解决的是从排产方案反算 C_i 再核算 \max(0, C_i - d_i) 这一工程闭环。3.3 如何映射到代码中业务逻辑 Python 代码排产方案ScheduleScheme设备×时段×订单订单主数据dataclass ProductionOrder完工时间解析ScheduleParser.parse_completion_times()延期天数计算DelayCalculator.compute_delay_days()延期损失核算PenaltyEvaluator.calculate_penalties()统计汇总DelayStatistics 类惩罚系数反馈PenaltyCoefficientFeeder 类四、OOP 代码实现精简可运行4.1 项目结构schedule_delay_auditor/├── schedule_delay_auditor.py # 核心代码单文件~350行├── sample_schedule.csv # 示例排产方案├── sample_orders.csv # 示例订单主数据├── README.md # 使用说明└── requirements.txt # 依赖库4.2 完整源代码可直接运行detailssummary/summary排产方案延期损失评估器 · 排产体检仪参考: 北京理工大学《运筹学》第4章线性规划应用、第8章排序与调度问题功能:1. 读取排产方案(CSV/Excel)和订单主数据2. 解析每个订单的计划完工时间3. 逐单计算延期天数 完工时间 - 承诺交期4. 逐单核算延期损失 延期天数 × 罚金单价5. 统计延期率、平均延期、TOP10延期订单6. 输出可喂回PuLP的延期惩罚系数运行:python schedule_delay_auditor.py(需要安装pandas, numpy, pulp)import csvfrom dataclasses import dataclass, fieldfrom typing import List, Dict, Optional, Tuple, Setfrom enum import Enumfrom datetime import datetime, timedeltaimport timeimport numpy as npimport pandas as pdimport pulp# ─── 枚举与常量 ────────────────────────────────────────────────────────────class OrderPriority(Enum):订单优先级URGENT 1 # 紧急(客户要求严, 罚金高)HIGH 2 # 高NORMAL 3 # 正常LOW 4 # 低(可延后)class ScheduleStatus(Enum):排产条目状态PLANNED 已排产DELIVERED 已交付DELAYED 已延期UNPLANNED 未排产# ─── 数据模型 ────────────────────────────────────────────────────────────dataclassclass ProductionOrder:生产订单主数据order_id: strproduct_name: strcustomer: strquantity: floatunit: str 件promised_delivery: datetime # 承诺交期( deadline )priority: OrderPriority OrderPriority.NORMALpenalty_per_day: float 1000.0 # 延期罚金(元/天)processing_time: float 4.0 # 加工时长(小时)early_bonus: float 0.0 # 提前交付奖励(元/天, 可选)propertydef priority_weight(self) - float:优先级权重(用于喂回模型)weights {OrderPriority.URGENT: 5.0, OrderPriority.HIGH: 3.0,OrderPriority.NORMAL: 1.0, OrderPriority.LOW: 0.5}return weights[self.priority]def __str__(self):return (f{self.order_id}({self.product_name}): 客户{self.customer}, f交期{self.promised_delivery.strftime(%m-%d)}, f罚金{self.penalty_per_day:.0f}元/天)dataclassclass ScheduleEntry:排产方案中的一条记录(某设备在某个时段加工某订单)machine_id: strorder_id: strstart_time: datetimeend_time: datetimesequence: int 0 # 同一设备上的加工序号propertydef duration_hours(self) - float:return (self.end_time - self.start_time).total_seconds() / 3600def __str__(self):s self.start_time.strftime(%m-%d %H:%M)e self.end_time.strftime(%m-%d %H:%M)return f{self.machine_id}: {self.order_id} [{s}~{e}]dataclassclass OrderDelayResult:单订单延期核算结果order_id: strpromised_delivery: datetimeplanned_completion: datetimedelay_days: float # 延期天数(负数为提前)penalty_amount: float # 延期罚款(元)status: ScheduleStatuspriority: OrderPrioritypropertydef is_delayed(self) - bool:return self.delay_days 0def __str__(self):flag ⚠️延期 if self.is_delayed else ✅准时return (f{self.order_id}: 交期{self.promised_delivery.strftime(%m-%d)}, f完工{self.planned_completion.strftime(%m-%d)}, f延期{self.delay_days:.1f}天, 罚款{self.penalty_amount:.0f}元 {flag})dataclassclass DelayStatistics:延期统计摘要total_orders: intdelayed_orders: inton_time_orders: inttotal_penalty: floatavg_delay_days: float # 仅统计延期订单max_delay_days: floatdelay_rate: float # 延期率def summary(self) - str:return (f延期统计摘要:\nf • 总订单数: {self.total_orders}\nf • 延期订单: {self.delayed_orders} ({self.delay_rate*100:.1f}%)\nf • 准时订单: {self.on_time_orders}\nf • 总罚款金额: {self.total_penalty:.0f} 元\nf • 平均延期: {self.avg_delay_days:.1f} 天(仅延期单)\nf • 最大延期: {self.max_delay_days:.1f} 天)# ─── 排产方案解析器 ───────────────────────────────────────────────────────class ScheduleParser:排产方案解析器: 还原每个订单的计划完工时间def __init__(self, schedule_entries: List[ScheduleEntry], orders: List[ProductionOrder]):self.entries schedule_entriesself.orders {o.order_id: o for o in orders}def parse_completion_times(self) - Dict[str, datetime]:解析每个订单的计划完工时间逻辑: 一个订单可能在多台设备上分段加工,最终完工时间 该订单所有加工段中最大的end_timecompletion_times: Dict[str, datetime] {}# 按订单分组, 取最晚的end_time作为完工时间for entry in self.entries:oid entry.order_idif oid not in completion_times or entry.end_time completion_times[oid]:completion_times[oid] entry.end_time# 未在排产方案中的订单, 完工时间设为永不(视为严重延期)for oid in self.orders:if oid not in completion_times:# 未排产: 用承诺交期7天作为默认(表示严重拖延)completion_times[oid] self.orders[oid].promised_delivery timedelta(days7)return completion_times# ─── 延期计算器 ───────────────────────────────────────────────────────────class DelayCalculator:延期天数与罚款核算器def __init__(self, orders: List[ProductionOrder], completion_times: Dict[str, datetime]):self.orders {o.order_id: o for o in orders}self.completion_times completion_timesdef compute_all(self) - List[OrderDelayResult]:逐单计算延期结果results: List[OrderDelayResult] []for oid, completion in self.completion_times.items():order self.orders.get(oid)if order is None:continuedeadline order.promised_delivery# 延期天数 (完工 - 交期)换算为天, 提前则记为0(不奖励也不罚)diff_seconds (completion - deadline).total_seconds()delay_days max(0.0, diff_seconds / 86400.0)# 罚款 延期天数 × 罚金单价penalty delay_days * order.penalty_per_day# 状态判定if delay_days 0:status ScheduleStatus.DELAYEDelif oid in [e.order_id for e in []]:status ScheduleStatus.UNPLANNEDelse:status ScheduleStatus.DELIVEREDresults.append(OrderDelayResult(order_idoid,promised_deliverydeadline,planned_completioncompletion,delay_daysdelay_days,penalty_amountpenalty,statusstatus,priorityorder.priority))return results# ─── 延期统计器 ───────────────────────────────────────────────────────────class DelayStatisticsCalculator:延期统计汇总staticmethoddef compute(results: List[OrderDelayResult]) - DelayStatistics:total len(results)delayed [r for r in results if r.is_delayed]on_time total - len(delayed)total_penalty sum(r.penalty_amount for r in delayed)avg_delay np.mean([r.delay_days for r in delayed]) if delayed else 0.0max_delay max((r.delay_days for r in delayed), default0.0)delay_rate len(delayed) / total if total 0 else 0.0return DelayStatistics(total_orderstotal,delayed_orderslen(delayed),on_time_orderson_time,total_penaltytotal_penalty,avg_delay_daysavg_delay,max_delay_daysmax_delay,delay_ratedelay_rate)staticmethoddef top_delayed_orders(results: List[OrderDelayResult], n: int 10) - List[OrderDelayResult]:TOP N 延期最严重的订单delayed [r for r in results if r.is_delayed]delayed.sort(keylambda r: r.penalty_amount, reverseTrue)return delayed[:n]# ─── 惩罚系数反馈器(喂回PuLP) ─────────────────────────────────────────────class PenaltyCoefficientFeeder:把延期惩罚系数喂回排产模型核心思路: 将延期损失转化为排产目标函数中的权重系数,使下次排产时模型自动优先保高罚金订单的交期.def __init__(self, orders: List[ProductionOrder], results: List[OrderDelayResult]):self.orders {o.order_id: o for o in orders}self.results {r.order_id: r for r in results}def build_penalty_coefficients(self) - Dict[str, float]:为每笔订单生成喂回模型的惩罚系数公式: penalty_coeff_i w_i × (1 delay_days_i / deadline_buffer)含义: 已经被排延期的订单, 下次给它更高权重, 逼模型优先它coeffs: Dict[str, float] {}for oid, order in self.orders.items():base order.priority_weight * order.penalty_per_dayresult self.results.get(oid)if result and result.delay_days 0:# 已延期的订单, 系数放大(逼模型重视)buffer_days 3.0 # 交期缓冲amplification 1.0 result.delay_days / buffer_dayscoeffs[oid] base * amplificationelse:coeffs[oid] basereturn coeffsdef to_pulp_objective_term(self, x_vars: Dict[Tuple[str, str], pulp.LpVariable],coeffs: Dict[str, float]) - pulp.LpAffineExpression:将惩罚项拼成PuLP目标函数可加的一项假设 x[(order_id, machine_id)] 1 表示订单分配到该设备,这里简化为: penalty Σ coeff_i × completion_day_i实际使用时需结合具体决策变量结构.term 0for (oid, mid), var in x_vars.items():if oid in coeffs:term coeffs[oid] * varreturn term# ─── 报告生成器 ───────────────────────────────────────────────────────────class DelayReport:延期分析报告生成器staticmethoddef print_summary(stats: DelayStatistics):print(f\n {stats.summary()})staticmethoddef print_top_delayed(top_list: List[OrderDelayResult]):if not top_list:print(\n✅ 没有延期订单, 一切正常!)returnprint(f\n TOP {len(top_list)} 延期最严重的订单:)print(f {订单号:12} {客户:10} {交期:8} {完工:8} {延期(天):10} {罚款(元):12} {优先级})print(f {─*72})for r in top_list:cust r.order_id # 简化显示pri {1:紧急,2:高,3:正常,4:低}[r.priority.value]d_deadline r.promised_delivery.strftime(%m-%d)d_comp r.planned_completion.strftime(%m-%d)print(f {r.order_id:12} {cust:10} {d_deadline:8} {d_comp:8} f{r.delay_days:10.1f} {r.penalty_amount:12.0f} {pri})staticmethoddef save_detail_csv(results: List[OrderDelayResult], path: str delay_detail.csv):data []for r in results:data.append({订单号: r.order_id,承诺交期: r.promised_delivery.strftime(%Y-%m-%d %H:%M),计划完工: r.planned_completion.strftime(%Y-%m-%d %H:%M),延期天数: round(r.delay_days, 2),罚款金额(元): round(r.penalty_amount, 2),状态: r.status.value,优先级: {1:紧急,2:高,3:正常,4:低}[r.priority.value]})pd.DataFrame(data).to_csv(path, indexFalse, encodingutf-8-sig)print(f\n 延期明细已保存到: {path})staticmethoddef save_penalty_coefficients(coeffs: Dict[str, float], path: str penalty_coeffs.csv):data [{订单号: k, 惩罚系数: round(v, 2)} for k, v in sorted(coeffs.items(), keylambda x: -x[1])]pd.DataFrame(data).to_csv(path, indexFalse, encodingutf-8-sig)print(f 惩罚系数已保存到: {path} (可喂回PuLP))# ─── 数据加载器 ───────────────────────────────────────────────────────────class DataLoader:数据加载器staticmethoddef load_orders(csv_path: str None) - List[ProductionOrder]:if csv_path is None:return DataLoader._sample_orders()orders []with open(csv_path, r, encodingutf-8) as f:for row in csv.DictReader(f):orders.append(ProductionOrder(order_idrow[order_id],product_namerow[product_name],customerrow[customer],quantityfloat(row[quantity]),unitrow.get(unit, 件),promised_deliverydatetime.strptime(row[promised_delivery], %Y-%m-%d %H:%M),priorityOrderPriority(int(row.get(priority, 3))),penalty_per_dayfloat(row.get(penalty_per_day, 1000)),processing_timefloat(row.get(processing_time, 4))))return ordersstaticmethoddef load_schedule(csv_path: str None) - List[ScheduleEntry]:if csv_path is None:return DataLoader._sample_schedule()entries []with open(csv_path, r, encodingutf-8) as f:for row in csv.DictReader(f):entries.append(ScheduleEntry(machine_idrow[machine_id],order_idrow[order_id],start_timedatetime.strptime(row[start_time], %Y-%m-%d %H:%M),end_timedatetime.strptime(row[end_time], %Y-%m-%d %H:%M),sequenceint(row.get(sequence, 0))))return entriesstaticmethoddef _sample_orders() - List[ProductionOrder]:内置示例订单(40个, 含紧急/高/正常/低优先级)base datetime(2025, 7, 1, 8, 0)customers [A客户(德), B客户(日), C客户(国内), D客户(美), E客户(国内)]products [齿轮轴, 法兰盘, 轴承座, 联轴节, 泵壳, 阀体, 曲轴, 导轨]priorities [(OrderPriority.URGENT, 5000), # 紧急, 罚金高(OrderPriority.HIGH, 3000),(OrderPriority.NORMAL, 1000),(OrderPriority.LOW, 500),]orders []for i in range(40):cust customers[i % len(customers)]prod products[i % len(products)]pri, pen priorities[i % len(priorities)]# 交期: 3~30天不等deadline base timedelta(days3 (i * 7) % 28, hours(i*3)%24)orders.append(ProductionOrder(order_idfPO{i1:04d},product_nameprod,customercust,quantity10 (i*3)%50,promised_deliverydeadline,prioritypri,penalty_per_daypen,processing_time2.0 (i%5)))return ordersstaticmethoddef _sample_schedule() - List[ScheduleEntry]:内置示例排产方案(5台设备, 40个订单的加工安排)故意制造一些延期: 把部分订单排在交期之后base datetime(2025, 7, 1, 8, 0)machines [fMC{i} for i in range(1, 6)]orders DataLoader._sample_orders()entries []# 简化排产: 每台设备串行加工8个订单, 每个4小时for m_idx, machine in enumerate(machines):machine_orders orders[m_idx*8:(m_idx1)*8]cursor base timedelta(hoursm_idx*2) # 各设备错开启动for seq, order in enumerate(machine_orders):dur order.processing_timestart cursorend利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛