python的运筹学工业场景模拟第八十四篇:读取零件BOM物料清单，拆解产品物料消耗，生成物料消耗系数矩阵。

BOM“翻译机”用Python拆解物料清单把“层级结构”变成“消耗系数矩阵”“某工程机械厂有 3 大系列产品、87 种自制件、近 300 种原材料。计划员每月做主生产计划MPS需要把产品产量换算成原材料需求。但 BOM物料清单是多层树状结构整机 → 部件 → 零件 → 原材料。人工一层层拆解算 3 天还经常漏算、错算。上个月因为漏算轴承钢需求导致关键零件停产 2 天损失 42 万。后来我写了个BOM 拆解器0.8 秒读完 87 种自制件的 BOM自动生成 3 大系列 × 300 种原材料的消耗系数矩阵直接喂给 PuLP 做线性规划。厂长说‘原来不是计划员不行是我们缺个能把 BOM 翻译成数学语言的工具。’”—— 参考北京理工大学《运筹学》第 4 章“运输与存储问题”、第 7 章“网络流与图论”一、实际应用场景描述BOM 拆解 → 物料消耗系数矩阵生成器是任何涉及生产计划、物料需求、成本优化的“数据翻译工具”。凡是“产品由零件组成、零件由材料构成”的地方都是它行业 生产场景 BOM 痛点 优化风险工程机械 整机装配 多层BOM5~7层 原材料缺料停产汽车制造 整车生产 数万种零部件 MRP运算失真电子装配 PCBA制造 元器件替代关系复杂 采购计划偏差家电生产 整机总装 自制件外协件混合 产能分配失衡航空航天 型号研制 构型管理严格 物料追溯困难化工生产 配方管理 原料配比精确 成本失控核心矛盾- ERP 里的 BOM 是“树状层级结构”父项-子项适合工程管理和生产领料- 线性规划PuLP/Scipy需要的是“消耗系数矩阵”产品→原材料一行一个产品一列一种材料- 人工拆解多层 BOM 耗时、易错导致物料需求计算失真- 没有准确的消耗系数矩阵生产计划优化就是“空中楼阁”。┌──────────────────────────────────────────────────────────────┐│ BOM拆解器 · 物料消耗系数矩阵生成器 · BOM翻译机 ││ ││ 【业务场景】 ││ ┌─────────────────────────────────────────────────────────┐││ │ 输入: 企业BOM物料清单(Excel/CSV) │││ │ • 物料主数据(编码、名称、类型、单位) │││ │ • BOM结构(父项编码、子项编码、用量、损耗率) │││ │ • 产品工艺路线(可选, 用于区分工序用料) │││ │ │││ │ 处理管道: │││ │ 1. 构建BOM树: 解析父子关系, 建立物料层级结构 │││ │ 2. 递归拆解: 从最终产品出发, 逐层展开到原材料 │││ │ 3. 汇总消耗: 累计各层级用量, 计算总消耗系数 │││ │ 4. 生成矩阵: 输出“产品×原材料”的消耗系数矩阵 │││ │ │││ │ 输出: │││ │ • 物料消耗系数矩阵(可直接用于PuLP线性规划) │││ │ • 各层级物料需求明细(BOM展开表) │││ │ • 虚拟件/中间件的消耗路径分析 │││ └─────────────────────────────────────────────────────────┘││ ││ 【核心矛盾】 ││ • 计划员: 想把产品产量换算成原材料需求 │││ • BOM: 是树状层级结构, 不是矩阵 │││ • 线性规划: 需要消耗系数矩阵A, 使得 Ax b (需求向量) │││ • 本程序: 把BOM树翻译成系数矩阵 — BOM翻译机 │││ ││ 【本程序处理流程】 ││ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐││ │ 构建BOM │──►│ 递归拆解 │──►│ 汇总消耗 │──►│ 生成系数 │││ │ 树(图结 │ │ (深度优先 │ │ (累计各 │ │ 矩阵(产品│││ │ 构) │ │ 搜索) │ │ 层级用量)│ │ ×原材料) │││ └──────────┘ └──────────┘ └──────────┘ └──────────┘│└──────────────────────────────────────────────────────────────┘二、引入痛点含量化对比2.1 现场真实困境某工程机械厂生产计划主管原话“我们厂主要生产3 大系列挖掘机20 吨级、30 吨级、40 吨级共 8 种主流机型。每台挖掘机有 3000 多个零件其中 87 种是自制的结构件、液压件等其余是外购件。每月做主生产计划MPS时我需要把8 种机型的计划产量换算成近 300 种原材料的需求量钢板、型材、轴承钢、液压油等。但 BOM 是多层树状结构- 第 0 层整机如 20 吨挖掘机- 第 1 层发动机总成、液压系统、结构件总成- 第 2 层缸体、活塞、连杆、阀块- 第 3 层毛坯、锻件、铸件- 第 4 层钢板、圆钢、型材等原材料人工拆解 BOM 的过程计划员拿着 Excel 版的 BOM 表从整机开始一层层往下找把每一层的用量手动乘起来再汇总到原材料。比如- 1 台整机需要 1 个液压系统- 1 个液压系统需要 2 个阀块- 1 个阀块需要 5 公斤轴承钢- 所以 1 台整机需要 1×2×5 10 公斤轴承钢。这个过程非常痛苦- 87 种自制件每层都要算算完 3 天- 经常漏算中间层级比如忘了算“锻件”这一层- 损耗率容易算错比如轴承钢利用率 85%但忘了乘这个系数- BOM 变更后要重新算一遍又得 3 天。上个月因为漏算了一种高强度螺栓的轴承钢需求导致关键零件停产 2 天生产线停了 2 条直接损失 42 万。厂长在月度经营会上说‘你们计划部 10 个人算个物料需求要 3 天还算错这效率怎么跟得上市场’后来 IT 组写了个 Python 脚本——0.8 秒读完 87 种自制件的 BOM自动递归拆解到原材料生成 8 种机型 × 300 种原材料的消耗系数矩阵直接喂给 PuLP 做线性规划优化。现在改 BOM 或调产量1 秒出结果。”2.2 人工拆解 vs 自动拆解量化对比指标 人工拆解 Python 自动拆解本方案 改善效果87 种自制件 BOM 拆解 3 天 0.8 秒 -99.99%物料需求计算准确率 ~92%易漏算 100% 质变BOM 变更响应时间 3 天 1 秒 -99.99%计划员工作量 10 人 × 3 天 1 人 × 1 秒 -99.9%停产损失 42 万/次 0 元 消除主生产计划MPS周期 每月一次 每日/实时 质变关键发现生产计划优化的瓶颈不在“求解”而在“BOM 数据的结构化”。一旦 BOM 被翻译成消耗系数矩阵PuLP 就能秒级算出最优物料需求。三、核心逻辑讲解大白话版3.1 用大白话解释“BOM 拆解与消耗系数矩阵”想象你要开个汉堡店卖 3 种汉堡牛肉堡、鸡肉堡、鱼堡。每种汉堡都由面包、肉饼、蔬菜、酱料组成。而肉饼又是由肉馅、淀粉、调料做的面包是由面粉、酵母、水做的。你的 BOM 表物料清单可能是这样的父项 子项 用量 单位牛肉堡 面包 1 个牛肉堡 牛肉饼 1 个牛肉堡 生菜 2 片牛肉堡 番茄酱 10 克牛肉饼 牛肉馅 100 克牛肉饼 淀粉 10 克牛肉饼 盐 2 克面包 面粉 50 克面包 酵母 1 克面包 水 30 克现在问题来了如果你明天要卖 100 个牛肉堡需要多少面粉人工算法你先算“1 个牛肉堡需要 1 个面包”再算“1 个面包需要 50 克面粉”所以 100 个牛肉堡需要 100 × 1 × 50 5000 克面粉。这叫“一层层往下算”。但如果有 10 层 BOM像工程机械那样人工算就头大了“牛肉堡 → 牛肉饼 → 牛肉馅 → 牛肉 → 牛 → 饲料 → 玉米 → …”。大白话逻辑1. BOM 是“树”根是最終产品叶子是原材料中间是零件/部件2. 拆解就是“从根到叶走一遍”把每一层的用量乘起来3. 消耗系数矩阵就是“把树压扁成表格”行是产品列是原材料格子是“做 1 个产品需要多少这种材料”4. 有了这个矩阵PuLP 就能秒算物料需求“产量向量 × 系数矩阵 原材料需求向量”。工业现场版- 汉堡店 工程机械厂- 汉堡 挖掘机最终产品- 面包/肉饼 自制件部件/零件- 面粉/牛肉 原材料- 消耗系数矩阵 产品与原材料的数量关系3.2 运筹学模型北理工《运筹学》映射参考北理工《运筹学》第 4 章“运输与存储问题”、第 7 章“网络流与图论”物料需求计划MRP的线性规划模型\begin{aligned}\min \quad Z \sum_{j1}^{m} c_j \cdot x_j \quad \text{(最小化原材料采购成本)} \\\text{s.t.} \quad \sum_{j1}^{m} a_{ij} \cdot x_j \ge d_i, \quad i1,\dots,n \quad \text{(产品需求满足)} \\ x_j \le S_j, \quad j1,\dots,m \quad \text{(原材料供应上限)} \\ x_j \ge 0, \quad j1,\dots,m\end{aligned}其中消耗系数矩阵 A (a_{ij}) 是本程序的核心输出- a_{ij} 生产 1 单位产品 i 需要消耗原材料 j 的数量- x_j 原材料 j 的采购量- d_i 产品 i 的需求量。BOM 拆解的数学本质有向无环图DAG的拓扑遍历。- 节点物料最终产品、自制件、原材料- 边BOM 关系父项 → 子项权重为用量- 拆解过程从根节点出发沿边进行深度优先搜索DFS累计路径上的权重乘积。北理工教材要点- 第 4 章 §4.1运输问题的数学模型消耗系数矩阵类似运输问题的“单位运价表”- 第 4 章 §4.2存储问题中的物料平衡约束 \sum a_{ij}x_j d_i - 第 7 章 §7.1图与网络的基本概念BOM 是有向图- 第 7 章 §7.2最小生成树与最短路径BOM 拆解类似最短路径问题- 本程序解决的是“从 BOM 树状结构生成消耗系数矩阵 A ”问题。3.3 如何映射到代码中业务逻辑 Python 代码物料主数据dataclass MaterialBOM 结构dataclass BOMItemBOM 树BOMTree 类基于有向图递归拆解BOMExploder.explode()DFS 算法消耗系数计算CoefficientCalculator.calculate()矩阵生成MatrixGenerator.generate_consumption_matrix()四、OOP 代码实现精简可运行4.1 项目结构bom_exploder/├── bom_exploder.py # 核心代码单文件~320行├── sample_bom_structures.csv # 示例BOM结构表├── sample_materials.csv # 示例物料主数据├── README.md # 使用说明└── requirements.txt # 依赖库4.2 完整源代码可直接运行detailssummary/summaryBOM拆解器 → 物料消耗系数矩阵生成器 · BOM翻译机参考: 北京理工大学《运筹学》第4章运输与存储问题、第7章网络流与图论功能:1. 读取物料主数据和BOM结构表(CSV格式)2. 构建BOM树(有向无环图)3. 递归拆解BOM, 从最终产品展开到原材料4. 计算累计消耗系数(考虑损耗率)5. 生成产品×原材料的消耗系数矩阵(可直接用于PuLP)运行:python bom_exploder.py(需要安装pandas, numpy, pulp)import csvimport jsonfrom dataclasses import dataclass, fieldfrom typing import List, Dict, Optional, Tuple, Setfrom enum import Enumfrom collections import defaultdict, dequeimport timeimport numpy as npimport pandas as pdimport pulp# ─── 枚举与常量 ────────────────────────────────────────────────────────────class MaterialType(Enum):物料类型FINISHED_GOOD 成品 # 最终产品SEMI_FINISHED 半成品 # 自制件/部件RAW_MATERIAL 原材料 # 原材料PURCHASED 外购件 # 外购零件VIRTUAL 虚拟件 # 用于BOM管理的虚拟物料class BOMStatus(Enum):BOM状态ACTIVE 生效OBSOLETE 失效PENDING 待审批# ─── 数据模型 ────────────────────────────────────────────────────────────dataclassclass Material:物料主数据material_code: strmaterial_name: strmaterial_type: MaterialTypeunit: str # 计量单位scrap_rate: float 0.0 # 损耗率(0~1)lead_time: int 0 # 采购/生产提前期(天)safety_stock: float 0.0 # 安全库存propertydef is_end_product(self) - bool:是否为最终产品return self.material_type MaterialType.FINISHED_GOODpropertydef is_raw_material(self) - bool:是否为原材料return self.material_type MaterialType.RAW_MATERIALpropertydef is_intermediate(self) - bool:是否为中间品(自制件/半成品)return self.material_type in [MaterialType.SEMI_FINISHED, MaterialType.VIRTUAL]def __str__(self):return f{self.material_code}({self.material_name}): {self.material_type.value}, {self.unit}dataclassclass BOMItem:BOM结构项(父子关系)parent_code: strchild_code: strquantity: float # 单耗(父项生产1单位, 需要子项多少单位)scrap_rate: float 0.0 # 额外损耗率(覆盖物料主数据)effective_date: Optional[str] Nonestatus: BOMStatus BOMStatus.ACTIVEpropertydef total_quantity(self) - float:总用量(含损耗)total_scrap max(self.scrap_rate, 0)return self.quantity * (1 total_scrap)def __str__(self):return f{self.parent_code} - {self.child_code}: {self.quantity}{ ( str(self.scrap_rate*100) %) if self.scrap_rate 0 else }dataclassclass ConsumptionCoefficient:消耗系数(产品→原材料)product_code: strmaterial_code: strcoefficient: float # 生产1单位产品消耗的原材料数量path: List[str] field(default_factorylist) # 消耗路径(用于追溯)def __str__(self):path_str → .join(self.path) if self.path else 直接return f{self.product_code} → {self.material_code}: {self.coefficient:.4f} ({path_str})dataclassclass BOMExplosionResult:BOM展开结果product_code: strtotal_quantity: float 1.0 # 产品产量material_requirements: Dict[str, float] field(default_factorydict) # 原材料需求intermediate_requirements: Dict[str, float] field(default_factorydict) # 中间品需求consumption_coefficients: List[ConsumptionCoefficient] field(default_factorylist)explosion_time: float 0.0 # 展开耗时(秒)def __str__(self):return (f产品{self.product_code}(产量{self.total_quantity}): f需{len(self.material_requirements)}种原材料, f{len(self.intermediate_requirements)}种中间品, f展开耗时{self.explosion_time*1000:.1f}ms)# ─── BOM树构建器 ──────────────────────────────────────────────────────────class BOMTree:BOM树(有向无环图)def __init__(self):self.materials: Dict[str, Material] {}self.bom_items: List[BOMItem] []self.children_map: Dict[str, List[BOMItem]] defaultdict(list)self.parent_map: Dict[str, List[BOMItem]] defaultdict(list)self._topological_order: List[str] []def add_material(self, material: Material):添加物料self.materials[material.material_code] materialdef add_bom_item(self, bom_item: BOMItem):添加BOM项if bom_item.status ! BOMStatus.ACTIVE:returnself.bom_items.append(bom_item)self.children_map[bom_item.parent_code].append(bom_item)self.parent_map[bom_item.child_code].append(bom_item)def build(self):构建BOM树, 进行拓扑排序# 检查是否有环if self._has_cycle():raise ValueError(BOM结构存在循环依赖!)# 拓扑排序self._topological_order self._topological_sort()def _has_cycle(self) - bool:检测是否有环(DFS)visited set()rec_stack set()def dfs(code: str) - bool:visited.add(code)rec_stack.add(code)for bom_item in self.children_map.get(code, []):if bom_item.child_code not in visited:if dfs(bom_item.child_code):return Trueelif bom_item.child_code in rec_stack:return Truerec_stack.remove(code)return Falsefor material_code in self.materials:if material_code not in visited:if dfs(material_code):return Truereturn Falsedef _topological_sort(self) - List[str]:拓扑排序(Kahn算法)in_degree defaultdict(int)# 计算入度for bom_item in self.bom_items:in_degree[bom_item.child_code] 1# 初始化队列(入度为0的节点, 通常是原材料)queue deque([code for code in self.materials if in_degree[code] 0])topo_order []while queue:code queue.popleft()topo_order.append(code)for bom_item in self.children_map.get(code, []):child_code bom_item.child_codein_degree[child_code] - 1if in_degree[child_code] 0:queue.append(child_code)if len(topo_order) ! len(self.materials):raise ValueError(BOM结构存在循环依赖, 无法进行拓扑排序!)return topo_orderdef get_children(self, parent_code: str) - List[BOMItem]:获取子项return self.children_map.get(parent_code, [])def get_parents(self, child_code: str) - List[BOMItem]:获取父项return self.parent_map.get(child_code, [])def get_root_products(self) - List[str]:获取根节点(最终产品, 无父项)roots []for code in self.materials:if not self.parent_map.get(code):if self.materials[code].is_end_product:roots.append(code)return rootsdef __str__(self):return fBOM树: {len(self.materials)}种物料, {len(self.bom_items)}条BOM关系# ─── BOM拆解器 ───────────────────────────────────────────────────────────class BOMExploder:BOM拆解器(递归展开)def __init__(self, bom_tree: BOMTree):self.bom_tree bom_treeself.coefficients: Dict[Tuple[str, str], float] defaultdict(float)self.paths: Dict[Tuple[str, str], List[str]] defaultdict(list)def explode(self, product_code: str, quantity: float 1.0) - BOMExplosionResult:拆解单个产品BOMstart_time time.perf_counter()# 清空之前的系数self.coefficients.clear()self.paths.clear()# 递归拆解self._explode_recursive(product_code, quantity, [product_code])# 汇总结果result BOMExplosionResult(product_codeproduct_code,total_quantityquantity,explosion_timetime.perf_counter() - start_time)# 分离原材料和中间品for (prod, mat), coef in self.coefficients.items():if prod ! product_code:continuematerial self.bom_tree.materials.get(mat)if material:if material.is_raw_material:result.material_requirements[mat] coef * quantityelif material.is_intermediate:result.intermediate_requirements[mat] coef * quantity# 添加消耗系数result.consumption_coefficients.append(ConsumptionCoefficient(product_codeprod,material_codemat,coefficientcoef,pathself.paths.get((prod, mat), [])))return resultdef _explode_recursive(self, current_code: str, current_quantity: float, path: List[str]):递归拆解(DFS)material self.bom_tree.materials.get(current_code)# 如果是原材料, 记录消耗系数if material and material.is_raw_material:product_code path[0]self.coefficients[(product_code, current_code)] current_quantity# 记录路径if (product_code, current_code) not in self.paths:self.paths[(product_code, current_code)] path.copy()return# 如果是虚拟件, 不记录, 继续拆解if material and material.material_type MaterialType.VIRTUAL:# 虚拟件不消耗资源, 直接传递数量for bom_item in self.bom_tree.get_children(current_code):child_quantity current_quantity * bom_item.total_quantityself._explode_recursive(bom_item.child_code, child_quantity, path)return# 如果是中间品或成品, 继续拆解子项children self.bom_tree.get_children(current_code)if not children:# 没有子项, 可能是外购件或原材料(但未标记为原材料)if material and material.material_type MaterialType.PURCHASED:product_code path[0]self.coefficients[(product_code, current_code)] current_quantityif (product_code, current_code) not in self.paths:self.paths[(product_code, current_code)] path.copy()return# 拆解子项for bom_item in children:child_quantity current_quantity * bom_item.total_quantitynew_path path [bom_item.child_code]self._explode_recursive(bom_item.child_code, child_quantity, new_path)def explode_all_products(self, product_quantities: Dict[str, float] None) - Dict[str, BOMExplosionResult]:拆解所有最终产品results {}root_products self.bom_tree.get_root_products()if product_quantities:# 只拆解指定的产品for product_code, quantity in product_quantities.items():if product_code in self.bom_tree.materials:results[product_code] self.explode(product_code, quantity)else:# 拆解所有根产品(默认产量1)for product_code in root_products:results[product_code] self.explode(product_code, 1.0)return results# ─── 消耗系数矩阵生成器 ──────────────────────────────────────────────────class ConsumptionMatrixGenerator:消耗系数矩阵生成器def __init__(self, bom_tree: BOMTree):self.bom_tree bom_treedef generate(self, explosion_results: Dict[str, BOMExplosionResult]) - Tuple[np.ndarray, List[str], List[str]]:生成消耗系数矩阵 A (产品 × 原材料)# 获取所有产品和原材料product_codes sorted(explosion_results.keys())material_codes set()for result in explosion_results.values():material_codes.update(result.material_requirements.keys())material_codes sorted(material_codes)# 初始化矩阵n_products len(product_codes)n_materials len(material_codes)matrix np.zeros((n_products, n_materials))# 填充矩阵product_index {code: i for i, code in enumerate(product_codes)}material_index {code: j for j, code in enumerate(material_codes)}for product_code, result in explosion_results.items():i product_index[product_code]for material_code, quantity in result.material_requirements.items():j material_index[material_code]# 系数是单位产品消耗量(总需求/产量)matrix[i, j] quantity / result.total_quantityreturn matrix, product_codes, material_codesdef generate_pulp_code(self, matrix: np.ndarray, product_codes: List[str], material_codes: List[str]) - str:生成PuLP可用的代码code_lines []code_lines.append(# 物料消耗系数矩阵(由bom_exploder自动生成))code_lines.append(f# 生成时间: {time.strftime(%Y-%m-%d %H:%M:%S)})code_lines.append(f# 产品数量: {len(product_codes)}, 原材料数量: {len(material_codes)})code_lines.append()code_lines.append(import pulp)code_lines.append(import numpy as np)code_lines.append()code_lines.append(# 产品列表)code_lines.append(fproducts {product_codes})code_lines.append()code_lines.append(# 原材料列表)code_lines.append(fmaterials {material_codes})code_lines.append()code_lines.append(# 消耗系数矩阵 A (产品 × 原材料))code_lines.append(A np.array([)for i, row in enumerate(matrix):row_str [ , .join([f{val:.6f} for val in row]) ]if i len(matrix) - 1:row_str ,code_lines.append(row_str)code_lines.append(]))code_lines.append()code_lines.append(# 示例: 在PuLP模型中使用)code_lines.append(# 决策变量: 原材料采购量)code_lines.append(x pulp.LpVariable.dicts(material, materials, lowBound0))code_lines.append(# 约束: 物料需求满足)code_lines.append(for i, product in enumerate(products):)code_lines.append( prob pulp.lpSum([A[i, j] * x[materials[j]] for j in range(len(materials))]) demand[product])return \n.join(code_lines)def save_matrix(self, matrix: np.ndarray, product_codes: List[str], material_codes: List[str], path: str利用AI解决实际问题。如果你觉得这个工具好用欢迎关注长安牧笛