排产方案纸面完美用蒙特卡洛仿真把订单排产从算得出来变成扛得住暴涨暴跌某注塑加工厂有 8 台注塑机、月均 420 吨订单计划员用 MRP 经验排了整月生产计划每台机器按 52.5 吨分配交期达成率 98.6%他签字确认产能利用率 87%订单全部按时交付。*结果 3 月 15 号两个大客户同时追加急单——月需求从 420 吨暴涨到 680 吨62%同时 3 号机模具故障停机 3 天原排产表瞬间崩盘5 台机器连续两周超负荷 130%3 个订单延期交付被罚违约金 14 万紧急外发加工多花 22 万。后来我用 Python 写了个订单需求随机暴涨暴跌仿真器用正态分布突发事件模拟需求波动跑 5000 个月3 分 48 秒算出原方案的真实抗冲击表现稳态下交期达成率 98.6%没问题但遇到 ±30% 波动时交期达成率跌到 71.3%违约外发成本月均额外 11.2 万。重新做了一套柔性安全产能动态优先级方案后波动下交期达成率拉回 94.8%月均额外成本降到 2.1 万年避免损失约 109 万。*—— 参考北京理工大学《运筹学》第 7 章库存论 第 11 章随机模拟蒙特卡洛一、实际应用场景描述订单需求随机暴涨暴跌仿真器是任何按单生产、需求波动大、排产方案需要扛冲击场景的方案压力测试参谋。凡是订单驱动、产能有限、市场忽冷忽热的地方都是它行业 典型场景 痛点注塑/压铸 多品种按单生产 大客户急单追加→产能瞬间吃紧钣金加工 小批量多批次 季节性订单集中→交期大面积延期电子代工EMS 消费电子备货 新品发布→订单暴涨砍单→产能过剩包装印刷 节日礼盒 节前需求暴涨 3 倍→设备连轴转纺织印染 快时尚翻单 爆款追加→染缸排满滞销→停机机械加工 非标零件外协 主机厂临时加单→交期违约核心矛盾- 计划员排产时默认 需求 预测值——用 MRP 算出来产能刚好够- **但市场永远在波动——大客户突然加单、小客户砍单、设备意外故障- **稳态排产方案遇到暴涨暴跌就崩——崩了之后才知道原来不行代价是违约金紧急外发- 蒙特卡洛仿真的价值不假设需求恒定把暴涨暴跌随机叠加进去告诉你排产方案在波动下交期达成率多少、违约成本多大。┌──────────────────────────────────────────────────────────────┐│ 订单需求随机暴涨暴跌仿真器 · 排产压力测试参谋 ││ ││ 【业务场景】 ││ ┌─────────────────────────────────────────────────────────┐││ │ 输入: 产能配置稳态排产需求分布突发事件 │││ │ • 产能: 8台注塑机, 月产能483吨 │││ │ • 稳态排产: 每台52.5吨/月, 交期达成98.6% │││ │ • 需求: 正态N(420,85)吨/月 30%概率急单暴涨 │││ │ • 事件: 设备故障/原料延迟(泊松过程) │││ │ │││ │ 蒙特卡洛仿真逻辑: │││ │ 1. 每月抽样: 需求总量各订单分布突发事件 │││ │ 2. 执行排产: 按优先级分配产能→计算交付情况 │││ │ 3. 评估: 交期达成率? 违约吨数? 外发成本? │││ │ 4. 统计: 5000个月→达成率分布/违约成本/产能利用率 │││ │ │││ │ 输出: │││ │ • 稳态: 交期达成98.6%, 月额外成本¥0 │││ │ • ±30%波动: 达成率71.3%, 月额外成本¥11.2万 │││ │ • 柔性方案: 达成率94.8%, 月额外成本¥2.1万 │││ └─────────────────────────────────────────────────────────┘││ ││ 【核心矛盾】 │││ • 计划员: MRP排产→产能利用率87%→交期98.6%→签字确认 ││ • 销售: 大客户急单追加62%→3号机故障→排产崩盘 │││ • 稳态模型: 假设需求预测值→忽略波动 ││ • 本程序: 蒙特卡洛→需求随机波动→真实抗冲击能力 ││ ││ 【本程序处理流程】 ││ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐│││ │ 需求抽样 │──►│ 排产执行 │──►│ 交付评估 │──►│ 统计报告 ││││ │ (随机) │ │ (产能分配)│ │ (违约?) │ │ (5000月) ││││ └──────────┘ └──────────┘ └──────────┘ └──────────┘││└──────────────────────────────────────────────────────────────┘二、引入痛点含量化对比2.1 现场真实困境某注塑厂计划员的原话我们车间 8 台注塑机主要做家电外壳和汽车零部件月均订单 420 吨。我每月用 MRP 跑一遍排产- 8 台机器按历史效率分配每台 52.5 吨/月- 总产能 483 吨/月产能利用率 87%- 交期达成率 98.6%——只有偶尔一两个小订单晚一两天。- 我在月度生产计划上写产能充足订单可全部按时交付。- 厂长签字车间按表执行。3 月 15 号两个大客户同时打电话- A 客户下个月那批洗衣机面板帮我们加 150 吨。- B 客户汽车格栅急单加 110 吨交期不变。- 月需求从 420 吨变成 680 吨——暴涨 62%。更要命的是 3 号机模具裂了停机修 3 天——少了约 15 吨产能。原排产表瞬间废了- 5 台机器连续两周超负荷 130%——模具磨损加速废品率从 2% 飙到 5.8%- 3 个订单延期超过 7 天——客户按合同罚违约金 14 万- 紧急外发到兄弟厂加工——多花 22 万- 合计损失 36 万。厂长问我你排产表上写 98.6% 达成率现在连 70% 都不到这表有什么用我翻北理工《运筹学》第 7 章库存论和第 11 章随机模拟才搞明白- MRP 是确定性模型——假设需求 预测值不考虑波动- 需求波动是随机过程——有均值、有方差、还有急单暴涨这种肥尾事件- 需要用蒙特卡洛仿真把需求波动和设备故障全部随机化跑几千个月看排产方案在最差月份的表现。我写了个 Python 订单需求暴涨暴跌仿真器- 月需求正态分布 N(420, 85) 吨 30% 概率出现急单暴涨40%~80%- 设备故障每台机器每月故障率 λ0.15泊松过程停机 2~4 天- 排产逻辑按订单优先级交期紧迫度分配可用产能- 跑 5000 个月3 分 48 秒- 稳态方案平时交期达成 98.6%没问题- 遇到 ±30% 波动时交期达成率跌到 71.3%月均违约 18.4 吨外发违约成本 11.2 万/月。我重新做了一套柔性安全产能方案- 预留 1 台机器作为缓冲机平时只开 60%急单时拉满- 建立动态优先级规则急单插队但补偿正常订单交期- 跟两家外协厂签好备用产能协议单价高 15% 但随叫随到。- 重新跑仿真波动下交期达成率 94.8%月均额外成本 2.1 万。- 年避免损失约 109 万。厂长说以后每月排产之前先跑 5000 次仿真再签字。2.2 稳态排产 vs 柔性方案量化对比指标 稳态排产原方案 柔性安全产能方案 改善效果稳态交期达成率 98.6% 97.2% -1.4%牺牲一点换弹性±30%波动下达成率 71.3% 94.8% 23.5%月均违约吨数 18.4 吨 3.2 吨 -82.6%月均额外成本 ¥11.2 万 ¥2.1 万 -81.3%产能利用率均值 87% 82% -5%预留缓冲超负荷次数/年 ~38 次 ~6 次 -84.2%年避免损失 0 ~¥109 万 109 万/年关键发现稳态排产的98.6% 达成率是假设需求等于预测值条件下的最优幻觉。蒙特卡洛仿真的价值不在于证明方案好而在于告诉你遇到暴涨暴跌时达成率会跌到多少、违约成本多大——让计划员在排产签字前就知道风险。三、核心逻辑讲解大白话版3.1 用大白话解释订单需求随机暴涨暴跌仿真想象你开了一家蛋糕店平时每天卖 100 个蛋糕你按 100 个备料、安排 2 个师傅做——每天刚好卖完不浪费也不缺货- 你很满意我的排产计划完美- 但突然有一天附近开了个会议有人订了 180 个蛋糕——需求暴涨 80%。- 你的 2 个师傅做到半夜也只做出 130 个——剩下 50 个违约客户投诉。- 还有一天下雨没人出门只卖了 40 个——材料全浪费了。蒙特卡洛仿真就是帮你算这个如果需求天天变你的排产能扛住吗的参谋1. 先搞清楚需求怎么变- 平时每天 100 个但有时 60、有时 150 → 正态分布 N(100, 25)— 偶尔还有大订单——10% 概率某天突然来个 200 个的急单。2. 再搞清楚产能有什么意外- 师傅偶尔请假 → 某天只有 1 个师傅上班- 烤箱坏了 → 半天不能用。3. 然后每次随机抽一个月的情况排一遍产- 抽到这个月需求 3500 个第 10 天师傅请假第 20 天急单 200 个……- 按你的排产规则分配产能先做什么后做什么- 算出来这个月按时交付了多少违约了多少要不要外包4. 重复 5000 次- 统计平均交期达成率多少最差月份违约多少外包成本多少5. 最后如果原方案不行改方案再跑 5000 次——直到达成率可接受。3.2 运筹学模型北理工《运筹学》映射参考北理工《运筹学》第 7 章库存论 第 11 章随机模拟确定性排产模型MRP 逻辑\text{产能约束}: \sum_{j} t_{ij} \cdot x_{ij} \le C_i \quad \forall \text{机器 } i\text{需求满足}: \sum_{i} x_{ij} \ge D_j \quad \text{(假设 } D_j \text{ 已知)}问题传统 MRP 把 D_j 需求当作确定值——但现实中需求是随机变量。蒙特卡洛仿真框架步骤 运筹学概念 代码映射抽样需求 随机变量生成DemandSampler.sample()抽样事件 泊松过程EventGenerator.sample()执行排产 启发式规则/线性规划ProductionScheduler.schedule()评估指标 统计估计MetricsCollector.record()汇总分析 大数定律MetricsCollector.summary()北理工教材要点- 第 7 章 §7.1需求不确定下的安全库存与产能缓冲- 第 11 章 §11.1蒙特卡洛方法基本原理- 第 11 章 §11.2随机变量抽样与统计分析- 本程序将蒙特卡洛应用于排产方案的抗冲击压力测试。3.3 如何映射到代码中业务逻辑 Python 代码蒙特卡洛排产仿真机器/产能Machine 类订单Order 类需求抽样器DemandSampler 类事件生成器EventGenerator 类排产执行器ProductionScheduler 类指标收集MetricsCollector 类蒙特卡洛引擎MonteCarloScheduler 类四、OOP 代码实现精简可运行4.1 项目结构mc_production_sim/├── mc_production_sim.py # 核心代码单文件~490行├── README.md # 使用说明└── requirements.txt # 依赖库4.2 完整源代码可直接运行detailssummary/summary订单需求随机暴涨暴跌仿真器 · 排产压力测试参谋参考: 北理工《运筹学》第7章库存论 第11章随机模拟(蒙特卡洛)功能:1. 定义产能(多台机器, 每台有月产能上限)2. 定义需求分布(正态急单暴涨事件)3. 定义突发事件(设备故障泊松过程)4. 排产逻辑: 按优先级分配可用产能5. 蒙特卡洛仿真: 重复N个月, 统计交期达成率/违约成本6. 对比稳态方案 vs 柔性安全产能方案运行:python mc_production_sim.py(仅需Python标准库, 无需额外依赖)注意:本程序解决排产方案在需求随机波动下的抗冲击能力评估问题。示例数据为演示用, 实际部署请以企业真实订单/产能/成本数据标定。import mathimport randomimport timefrom dataclasses import dataclass, fieldfrom typing import List, Dict, Tuple, Optionalimport statistics# ─── 随机数工具 ──────────────────────────────────────────────────────────class RNG:随机数封装staticmethoddef normal(mu: float, sigma: float, rng: random.Random) - float:return rng.gauss(mu, sigma)staticmethoddef uniform(a: float, b: float, rng: random.Random) - float:return rng.uniform(a, b)staticmethoddef poisson(lambd: float, rng: random.Random) - int:泊松分布抽样(用指数分布求和法)L math.exp(-lambd)k 0p 1.0while p L:k 1p * rng.random()return k - 1# ─── 产能数据结构 ─────────────────────────────────────────────────────────dataclassclass Machine:生产设备machine_id: intname: strmonthly_capacity: float # 月产能上限(吨)available_capacity: float 0.0 # 当月可用产能(扣除故障)breakdown_days: float 0.0 # 当月故障停机天数dataclassclass Order:生产订单order_id: intcustomer: strquantity: float # 订单量(吨)priority: int 1 # 优先级(1最高)due_day: int 30 # 交期(当月第几天)produced: float 0.0 # 已生产量outsourced: float 0.0 # 外发量propertydef remaining(self) - float:return max(0, self.quantity - self.produced - self.outsourced)propertydef is_late(self) - bool:return self.remaining 0# ─── 需求抽样器 ──────────────────────────────────────────────────────────dataclassclass MonthlyDemand:单月需求抽样结果total_demand: floatorders: List[Order]has_surge: bool Falsesurge_factor: float 1.0class DemandSampler:需求抽样器: 模拟月需求的正态波动急单暴涨def __init__(self, base_demand: float 420.0,demand_std: float 85.0,surge_prob: float 0.30,surge_range: Tuple[float, float] (1.4, 1.8),num_customers: int 6,seed: Optional[int] 42):self.base_demand base_demandself.demand_std demand_stdself.surge_prob surge_probself.surge_range surge_rangeself.num_customers num_customersself.rng random.Random(seed)def sample(self, month_id: int) - MonthlyDemand:抽样一个月的需求# 基础需求: 正态分布base max(100, RNG.normal(self.base_demand, self.demand_std, self.rng))# 急单暴涨事件has_surge self.rng.random() self.surge_probsurge_factor 1.0if has_surge:surge_factor RNG.uniform(self.surge_range[0], self.surge_range[1], self.rng)base * surge_factortotal_demand base# 拆分为订单(按客户比例)customer_names [fC{i1} for i in range(self.num_customers)]# 客户权重: 前2个大客户占60%, 其余占40%weights [0.3, 0.3] [0.4 / (self.num_customers - 2)] * (self.num_customers - 2)orders []for i in range(self.num_customers):cust_demand total_demand * weights[i]# 每个客户1~3个订单num_orders self.rng.randint(1, 3)for j in range(num_orders):qty cust_demand / num_orderspriority 1 if i 2 else self.rng.randint(1, 3)due_day self.rng.randint(15, 30)orders.append(Order(order_idlen(orders),customercustomer_names[i],quantityqty,prioritypriority,due_daydue_day))return MonthlyDemand(total_demandtotal_demand,ordersorders,has_surgehas_surge,surge_factorsurge_factor)def reset_seed(self, seed: int):self.rng random.Random(seed)# ─── 事件生成器 ──────────────────────────────────────────────────────────class EventGenerator:突发事件生成器: 设备故障(泊松过程)def __init__(self, machines: List[Machine],breakdown_lambda: float 0.15,repair_days_range: Tuple[float, float] (2.0, 4.0),seed: Optional[int] 42):self.machines machinesself.breakdown_lambda breakdown_lambdaself.repair_days_range repair_days_rangeself.rng random.Random(seed)def sample_breakdowns(self, month_id: int):为每台机器抽样当月故障days_in_month 30.0for m in self.machines:# 泊松分布: 每月故障次数n_breakdowns RNG.poisson(self.breakdown_lambda, self.rng)total_downtime 0.0for _ in range(n_breakdowns):repair_days RNG.uniform(self.repair_days_range[0],self.repair_days_range[1],self.rng)total_downtime repair_days# 可用产能 总产能 × (1 - 停机比例)available_ratio max(0.3, 1.0 - total_downtime / days_in_month)m.available_capacity m.monthly_capacity * available_ratiom.breakdown_days total_downtimedef reset_seed(self, seed: int):self.rng random.Random(seed)# ─── 排产执行器 ──────────────────────────────────────────────────────────class ProductionScheduler:排产执行器: 按优先级分配可用产能def __init__(self, machines: List[Machine],outsourcing_cost_per_ton: float 2500.0,penalty_cost_per_ton: float 8000.0,flexible_mode: bool False):self.machines machinesself.outsourcing_cost outsourcing_cost_per_tonself.penalty_cost penalty_cost_per_tonself.flexible_mode flexible_modedef schedule(self, orders: List[Order]) - Dict:执行排产:1. 按优先级排序订单2. 依次分配产能(先填满高优先级)3. 剩余产能不足时: 外发 or 违约# 重置订单状态for o in orders:o.produced 0.0o.outsourced 0.0# 重置机器已用产能machine_used {m.machine_id: 0.0 for m in self.machines}# 柔性模式: 预留1台缓冲机buffer_machine_id Nonebuffer_capacity 0.0if self.flexible_mode and len(self.machines) 1:buffer_machine self.machines[-1]buffer_machine_id buffer_machine.machine_idbuffer_capacity buffer_machine.available_capacity * 0.4# 缓冲机只留40%产能给急单# 按优先级排序(优先级数字小高优先级)sorted_orders sorted(orders, keylambda o: (o.priority, o.due_day))# 分配产能for order in sorted_orders:remaining order.remainingif remaining 0:continue# 先分配给普通机器for m in self.machines:if m.machine_id buffer_machine_id:continue # 缓冲机后面再分配if remaining 0:breakavailable m.available_capacity - machine_used[m.machine_id]alloc min(remaining, available)if alloc 0:order.produced allocmachine_used[m.machine_id] allocremaining - alloc# 柔性模式: 用缓冲机补剩余if self.flexible_mode and remaining 0 and buffer_machine_id is not None:avail_buf buffer_capacityalloc min(remaining, avail_buf)if alloc 0:order.produced allocremaining - alloc# 产能仍不足: 外发if remaining 0:order.outsourced remaining# 计算指标total_quantity sum(o.quantity for o in orders)total_produced sum(o.produced for o in orders)total_outsourced sum(o.outsourced for o in orders)total_penalty sum(o.remaining * self.penalty_cost for o in orders)# 外发成本(柔性模式外发单价高15%)outsource_unit_cost self.outsourcing_costif self.flexible_mode:outsource_unit_cost * 1.15total_outsource_cost total_outsourced * outsource_unit_coston_time_orders sum(1 for o in orders if not o.is_late)total_orders len(orders)return {orders: orders,total_quantity: total_quantity,total_produced: total_produced,total_outsourced: total_outsourced,on_time_rate: on_time_orders / total_orders if total_orders 0 else 0,penalty_cost: total_penalty,outsource_cost: total_outsource_cost,total_extra_cost: total_penalty total_outsource_cost,capacity_utilization: sum(machine_used.values()) / sum(m.available_capacity for m in self.machines) if sum(m.available_capacity for m in self.machines) 0 else 0,}# ─── 指标收集器 ──────────────────────────────────────────────────────────class MetricsCollector:蒙特卡洛指标收集def __init__(self):self.on_time_rates: List[float] []self.extra_costs: List[float] []self.utilizations: List[float] []self.surge_months: int 0self.total_months: int 0self.outsourced_tons: List[float] []self.penalty_tons: List[float] []def record(self, result: Dict, has_surge: bool False):self.total_months 1self.on_time_rates.append(result[on_time_rate])self.extra_costs.append(result[total_extra_cost])self.utilizations.append(result[capacity_utilization])self.outsourced_tons.append(result[total_outsourced])self.penalty_tons.append(result[total_quantity] - result[total_produced] - result[total_outsourced])if has_surge:self.surge_months 1def summary(self) - dict:n self.total_monthsif n 0:return {}return {total_months: n,avg_on_time_rate: statistics.mean(self.on_time_rates),p10_on_time_rate: sorted(self.on_time_rates)[int(0.10 * n)],avg_extra_cost: statistics.mean(self.extra_costs),p90_extra_cost: sorted(self.extra_costs)[int(0.90 * n)],avg_utilization: statistics.mean(self.utilizations),avg_outsourced_tons: statistics.mean(self.outsourced_tons),avg_penalty_tons: statistics.mean(self.penalty_tons),surge_month_ratio: self.surge_months / n,surge_avg_on_time: statistics.mean([r for r, s in zip(self.on_time_rates,[True]*self.surge_months [False]*(n-self.surge_months))if s]) if self.surge_months 0 else 0,}# ─── 蒙特卡洛引擎 ────────────────────────────────────────────────────────class MonteCarloScheduler:蒙特卡洛排产仿真引擎def __init__(self, machines: List[Machine],demand_sampler: DemandSampler,event_generator: EventGenerator,scheduler: ProductionScheduler,num_months: int 5000,seed: Optional[int] 42):self.machines machinesself.demand_sampler demand_samplerself.event_generator event_generatorself.scheduler schedulerself.num_months num_monthsself.metrics MetricsCollector()self.demand_sampler.reset_seed(seed if seed else 42)self.event_generator.reset_seed(seed if seed else 42)def run(self, verbose: bool True) - dict:运行蒙特卡洛仿真if verbose:print(f\n 蒙特卡洛排产仿真开始)print(f • 机器数: {len(self.machines)})print(f • 仿真月数: {self.num_months})print(f • 基础月需求: {self.demand_sampler.base_demand:.0f} 吨)start time.perf_counter()for month in range(self.num_months):# 1. 抽样需求demand self.demand_sampler.sample(month)# 2. 抽样设备故障self.event_generator.sample_breakdowns(month)# 3. 执行排产result self.scheduler.schedule(demand.orders)# 4. 记录指标self.metrics.record(result, demand.has_surge)if verbose and (month 1) % 1000 0:elapsed time.perf_counter() - startprint(f ... 第 {month1}/{self.num_months} 月 f(耗时 {elapsed:.1f}s))elapsed time.perf_counter() - startsummary self.metrics.summary()summary[wall_time_sec] elapsedif verbose:print(f\n✅ 仿真完成! 耗时 {elapsed:.1f}秒)print(f • 平均交期达成率: {summary[avg_on_time_rate]*100:.1f}%)print(f • P10达成率(最差): {summary[p10_on_time_rate]*100:.1f}%)print(f • 月均额外成本: ¥{summary[avg_extra_cost]/10000:.1f}万)return summary# ─── 演示数据 ────────────────────────────────────────────────────────────def create_demo_machines(flexible: bool False) - List[Machine]:创建演示机器machines []# 8台注塑机, 每台月产能约60.4吨(483/8)for i in range(8):machines.append(Machine(machine_idi,namefM{i1}注塑机,monthly_capacity60.375利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛