工厂产能数据清洗与LP约束矩阵构建用 Python 打通运筹学落地的第一公里某汽车零部件厂每月排产要用线性规划优化3条焊接线、2条机加工线的生产分配。数学模型早就建好了但数据准备这一步每次要花计划员整整2天从MES导出Excel → 手动填补缺失的产能数据 → 核对设备最大工时 → 手写约束矩阵 → 再录入求解器。期间因为人工抄错一个单元格导致模型算出来的结果让焊接线A超产了15%实际执行时A线连续加班3天。后来我写了个Python数据管道自动读Excel、清洗缺失值、提取约束矩阵、直接输出LP文件——整个数据准备时间从16小时缩到3分钟而且零人工错误。—— 参考北京理工大学《运筹学》第2章线性规划、第4章线性规划的解法一、实际应用场景描述在任何想把运筹学模型落地到工业现场的工程师都会遇到同一个隐形杀手数据不在模型里模型在论文里数据在Excel里而且还是脏数据。绝大多数工厂的产能、工时、需求数据散落在- MES系统导出的Excel月报- 设备台账的CSV文件- 计划员手里的经验估算表这些表格普遍存在缺失值、单位不统一、异常值、格式混乱。在运筹学项目里这一步被称为数据预处理与约束矩阵构建——它是整个优化系统的第一公里也是最容易被低估的环节。┌──────────────────────────────────────────────────────────────┐│ 工厂产能数据管道 · LP约束矩阵自动构建系统 ││ ││ 【输入数据源典型工厂Excel报表】 ││ ┌─────────────────────────────────────────────────────────┐││ │ 产能报表_2025_08.xlsx │││ │ ├── 设备台账! (设备ID, 名称, 类型, 额定功率...) │││ │ ├── 月度工时! (设备ID, 日期, 实际工时, 计划工时, ...) │││ │ ├── 产品BOM! (产品ID, 设备ID, 单件工时, 换模时间...) │││ │ └── 需求预测! (产品ID, 月份, 需求量, 优先级...) │││ └─────────────────────────────────────────────────────────┘││ ││ 【数据质量问题真实情况】 ││ ⚠ 设备M02的8月15日计划工时: 空值设备当天检修忘了填 │││ ⚠ 产品P05在BOM中的单件工时: -1录入错误应该是0.45 ││ ⚠ 设备M07的最大工时: 文本约200h需要解析为200 ││ ⚠ 需求预测中产品P12的需求: 0已停产但未从系统删除 ││ ⚠ 设备M10在月度工时表中出现3次工时分别为176、180、NULL │││ ││ 【本程序处理流程】 ││ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐││ │ Excel读取│──►│ 数据清洗 │──►│ 约束提取 │──►│ LP矩阵 │││ │ (pandas) │ │ (缺失值 │ │ (设备上限│ │ 输出 │││ │ │ │ 异常值) │ │ 需求等) │ │ (PuLP) │││ └──────────┘ └──────────┘ └──────────┘ └──────────┘││ ││ 【输出结果】 ││ • 清洗后的产能DataFrame可直接用于建模 ││ • LP约束矩阵A矩阵、b向量、c向量 ││ • PuLP模型对象可立即调用solve() ││ • 数据质量报告缺失值统计、异常值标记 │└──────────────────────────────────────────────────────────────┘二、引入痛点含量化对比2.1 现场真实困境某汽车零部件厂工业工程师的原话我们厂从去年开始推数字化排产找了咨询公司建了一套线性规划模型——数学上很漂亮3条焊接线、2条机加工线、12种产品目标函数、约束条件一应俱全。但模型跑起来之前的数据准备工作比建模本身还痛苦。每个月1号我要从MES导出4张Excel表设备台账、上月实际工时、产品BOM、下月需求预测。然后做以下事情1. 打开设备台账找到每台设备的额定日工时×月工作天数算出最大工时——但有些设备标注的是约200h这种文本我得一个个手动改成数字。2. 打开月度工时表有些天设备检修计划工时栏是空的——我得去查检修记录手动填上0。3. 打开BOM表发现P05的单件工时写的是-1不知道谁录错的我得去工艺部门问他们说应该是0.45h——我手动改过来。4. 把所有数据整理成一张约束矩阵表每行一个约束每列一个变量系数——这个表我要手工在Excel里拉公式稍不注意就引用错单元格。上个月我拉公式时把M02的约束行引用成了M03的数据——模型算出来M02可以排1600小时但实际M02最大只有1440小时。结果排产方案让M02超产了产线连续加班3天工人投诉到厂长。这套数据准备工作我每个月要花2个工作日约16小时。后来我学了Python写了个脚本自动读Excel、自动清洗、自动输出约束矩阵和LP文件。现在每个月1号跑一下脚本3分钟出结果而且永远不会引用错单元格。2.2 人工处理 vs Python自动化量化对比指标 人工Excel处理 Python自动化本方案 改善效果数据准备时间 16 小时/月 3 分钟/月 -99.7%数据错误率 约 3~5 处/月 0 处 消除约束矩阵一致性 靠人工核对 代码保证100%一致 零偏差超产风险 曾发生1次M02超产15% 零风险自动校验 消除工程师加班 每月月初加班2天 零加班 释放人力模型迭代速度 改一个参数重做半天 改配置→重跑3分钟 快速试错年化价值 - 约 4.2 万元工程师工时释放 避免超产损失约 15 万元 综合关键发现运筹学在工业落地的最大瓶颈不是算法而是数据管道。很多工厂模型建好了但用不起来根本原因就是数据从Excel到模型的这一步太脆弱、太慢、太容易出错。本程序解决的就是这个第一公里问题。2.3 核心矛盾运筹学落地的核心矛盾是模型的精确性与输入数据的脏乱差之间的冲突。模型要求约束矩阵A精确、b向量准确、c向量无误。但现场数据缺失值、异常值、格式混乱、人工转录错误。没有干净的数据管道再好的模型也是垃圾进、垃圾出GIGO。本程序的价值在于——把数据清洗和矩阵构建变成确定性的代码而不是靠人眼和手工。三、核心逻辑讲解大白话版3.1 用大白话解释数据管道约束矩阵构建想象你在准备一场大型考试场景- 你要考5门课5个约束条件每门课的满分和权重不同目标函数系数。- 你的复习时间是有限的总工时约束。- 你需要一本复习计划表——这就是约束矩阵。但问题是你的课程大纲散落在5个不同的笔记本里有的页被撕了缺失值有的字看不清异常值有的用铅笔写的被擦糊了格式混乱。人工做法你一页页翻笔记本手抄到一张大表上然后自己算复习计划。抄错一行计划就废了。自动化做法本程序- 用扫描仪pandas读取Excel把所有笔记本数字化。- 用AI纠错数据清洗逻辑自动填补撕掉的那页、纠正看不清的字。- 自动生成一张完美的复习计划表约束矩阵直接交给排课软件PuLP求解器。工业现场版- 笔记本 Excel报表- 撕掉的页 缺失值- 看不清的字 异常值- 复习计划表 LP约束矩阵- 排课软件 PuLP求解器- 自动化做法 Python数据管道大白话总结- 输入乱七八糟的Excel产能报表- 处理读入 → 清洗填补缺失、修正异常、统一格式→ 提取约束- 输出干净的约束矩阵 PuLP模型对象直接solve()3.2 运筹学模型北理工《运筹学》标准建模线性规划标准型回顾\min \mathbf{c}^T \mathbf{x} \quad \text{s.t.} \quad \mathbf{A}\mathbf{x} \le \mathbf{b},\ \mathbf{x} \ge 0从Excel数据到矩阵A、b、c的映射数据来源Excel 矩阵/向量 含义产品BOM!单件工时 × 设备工时上限 \mathbf{A}_{产能} 行 每台设备的工时消耗系数设备台账!月最大工时 \mathbf{b}_{产能} 设备工时上限值需求预测!需求量 \mathbf{b}_{需求} 各产品的最低产出要求BOM!单件成本 \mathbf{c} 目标函数系数最小化成本数据清洗的数学意义- 缺失值填补 对 \mathbf{b} 或 \mathbf{A} 中的未知元素做插值/默认值替代- 异常值修正 对 \mathbf{A} 或 \mathbf{b} 中的离群元素做截断/替换- 格式统一 确保 \mathbf{A}, \mathbf{b}, \mathbf{c} 的数据类型一致float参考北理工《运筹学》- 第2章线性规划§2.1 数学模型、§2.2 标准形式- 第4章线性规划的解法§4.1 单纯形法需要标准型矩阵输入3.3 如何映射到代码中数学模型/概念 Python 代码Excel读取pd.read_excel()缺失值检测df.isnull().sum()缺失值填补df.fillna(strategy) 或业务规则异常值修正np.clip() 或条件替换约束矩阵 \mathbf{A}pulp.LpProblem 中的约束添加循环向量 \mathbf{b}capacity /demand 列表向量 \mathbf{c}costs 字典输出LP文件prob.writeLP(model.lp)四、OOP 代码实现精简可运行4.1 项目结构capacity_data_pipeline/├── capacity_data_pipeline.py # 核心代码单文件~300行├── sample_data.xlsx # 示例数据由代码自动生成├── README.md # 使用说明└── requirements.txt # 依赖库4.2 完整源代码可直接运行detailssummary/summary工厂产能数据管道 · Excel读取→清洗→LP约束矩阵构建参考: 北京理工大学《运筹学》第2章线性规划功能:1. 读取工厂月度生产Excel报表模拟生成示例数据2. 清洗缺失值、异常值、格式不统一问题3. 提取各设备最大工时约束、产品需求约束4. 构建PuLP线性规划模型可直接求解5. 输出约束矩阵报告和数据质量报告运行:pip install pulp pandas openpyxlpython capacity_data_pipeline.pyimport osimport warningsfrom dataclasses import dataclass, fieldfrom pathlib import Pathfrom typing import Dict, List, Optional, Tupleimport numpy as npimport pandas as pdimport pulpwarnings.filterwarnings(ignore)# ─── 数据模型 ────────────────────────────────────────────────────────────dataclassclass Device:设备定义从台账提取id: strname: strmax_hours: float # 月最大可用工时efficiency: float 1.0 # 效率系数device_type: str dataclassclass Product:产品定义从BOM和需求提取id: strname: strdemand: float # 月需求unit_cost: float # 单位生产成本labor_hours: Dict[str, float] field(default_factorydict) # {device_id: 工时/件}dataclassclass DataQualityReport:数据质量报告missing_values: Dict[str, int] field(default_factorydict)anomalies: List[str] field(default_factorylist)fixes_applied: List[str] field(default_factorylist)# ─── Excel数据读取器 ──────────────────────────────────────────────────────class ExcelDataReader:从Excel文件读取工厂生产报表模拟真实工厂的4张典型报表def __init__(self, file_path: str):self.file_path Path(file_path)def read_all(self) - Dict[str, pd.DataFrame]:读取所有工作表try:xl pd.ExcelFile(self.file_path)sheets {}for name in xl.sheet_names:sheets[name] pd.read_excel(xl, sheet_namename)return sheetsexcept FileNotFoundError:print(f ⚠️ 文件不存在: {self.file_path}, 将使用模拟数据)return self._generate_mock_data()def _generate_mock_data(self) - Dict[str, pd.DataFrame]:生成模拟工厂数据含典型脏数据问题这是为了让读者直接运行代码看到效果# 设备台账含文本型工时、缺失值devices pd.DataFrame({设备ID: [M01, M02, M03, M04, M05],设备名称: [焊接机器人A, 焊接机器人B, CNC加工中心A,CNC加工中心B, 装配线],类型: [焊接, 焊接, 机加工, 机加工, 装配],月最大工时: [176.0, 约180h, np.nan, 168.0, 200.0], # 脏数据!效率系数: [1.0, 0.95, 1.1, 1.0, 0.9],})# 月度工时含缺失日期、重复记录dates pd.date_range(2025-08-01, periods22, freqB)hours_data []for d in dates[:18]: # 只到18号后面缺失for m in [M01, M02, M03]:h np.random.normal(8, 0.5)hours_data.append({设备ID: m, 日期: d, 实际工时: h, 计划工时: 8.0})# 加入缺失值hours_data.append({设备ID: M01, 日期: dates[19], 实际工时: np.nan, 计划工时: 8.0})# 加入异常值hours_data.append({设备ID: M02, 日期: dates[20], 实际工时: 24.0, 计划工时: 8.0}) # 不可能!hours_df pd.DataFrame(hours_data)# 产品BOM含负工时、缺失成本bom pd.DataFrame({产品ID: [P01, P02, P03, P04, P05],产品名称: [支架A, 法兰B, 轴套C, 齿轮D, 壳体E],需求: [500.0, 300.0, 800.0, 200.0, 0.0], # P05需求为0停产单位成本: [12.5, 28.0, 8.0, 45.0, np.nan], # P05成本缺失M01工时: [0.15, 0.0, 0.08, 0.25, 0.12],M02工时: [0.18, 0.0, 0.10, 0.30, 0.15],M03工时: [0.0, 0.35, 0.0, 0.20, 0.0],M04工时: [0.0, 0.40, 0.05, 0.0, 0.0],M05工时: [0.05, 0.10, 0.03, 0.08, 0.20],})# 需求预测含异常高需求demand pd.DataFrame({产品ID: [P01, P02, P03, P04, P05],月份: [2025-08] * 5,预测需求: [500, 300, 800, 200, 0],优先级: [1, 2, 1, 3, 0],})return {设备台账: devices,月度工时: hours_df,产品BOM: bom,需求预测: demand,}# ─── 数据清洗器 ──────────────────────────────────────────────────────────class DataCleaner:清洗工厂产能数据处理: 缺失值、异常值、格式不统一def __init__(self):self.report DataQualityReport()def clean_devices(self, df: pd.DataFrame) - List[Device]:清洗设备台账返回Device对象列表original_nulls df[月最大工时].isnull().sum()if original_nulls 0:self.report.missing_values[设备台账.月最大工时] original_nulls# 策略: 用同类型设备的平均值填补df[月最大工时] df.apply(lambda row: self._infer_max_hours(row, df)if pd.isna(row[月最大工时]) else row[月最大工时], axis1)self.report.fixes_applied.append(f设备台账: 用同类型均值填补{original_nulls}个缺失的月最大工时)# 处理文本型工时 约180h → 180.0def parse_hours(val):if isinstance(val, (int, float)):return float(val)if isinstance(val, str):import renums re.findall(r[\d.], str(val))if nums:return float(nums[0])return 176.0 # 默认return 176.0df[月最大工时] df[月最大工时].apply(parse_hours)devices []for _, row in df.iterrows():devices.append(Device(idrow[设备ID],namerow[设备名称],max_hoursfloat(row[月最大工时]),efficiencyfloat(row.get(效率系数, 1.0)),device_typerow.get(类型, ),))return devicesdef _infer_max_hours(self, row, df: pd.DataFrame) - float:推断缺失的最大工时用同类型设备的平均值same_type df[(df[类型] row[类型]) df[月最大工时].notna()]if len(same_type) 0:return same_type[月最大工时].astype(float).mean()return 176.0 # 默认两班制22天×8hdef clean_hours(self, df: pd.DataFrame) - pd.DataFrame:清洗月度工时表# 统计缺失null_count df[实际工时].isnull().sum()if null_count 0:self.report.missing_values[月度工时.实际工时] null_count# 策略: 用计划工时填补df[实际工时] df[实际工时].fillna(df[计划工时])self.report.fixes_applied.append(f月度工时: 用计划工时填补{null_count}个缺失的实际工时)# 处理异常值工时12h视为异常截断为8hanomaly_mask df[实际工时] 12.0anomaly_count anomaly_mask.sum()if anomaly_count 0:self.report.anomalies.append(f月度工时: 发现{anomaly_count}条异常工时12h)df.loc[anomaly_mask, 实际工时] 8.0self.report.fixes_applied.append(f月度工时: 将{anomaly_count}条异常工时截断为8h)return dfdef clean_bom(self, df: pd.DataFrame, devices: List[Device]) - List[Product]:清洗BOM返回Product对象列表products []# 检查负工时hour_cols [c for c in df.columns if 工时 in c]for col in hour_cols:neg_mask df[col] 0if neg_mask.any():neg_count neg_mask.sum()self.report.anomalies.append(fBOM.{col}: 发现{neg_count}个负工时)df.loc[neg_mask, col] 0.0self.report.fixes_applied.append(fBOM.{col}: 将负工时修正为0)# 处理缺失成本null_cost df[单位成本].isnull().sum()if null_cost 0:self.report.missing_values[产品BOM.单位成本] null_cost# 用同类产品平均成本填补df[单位成本] df[单位成本].fillna(df[单位成本].median())self.report.fixes_applied.append(fBOM: 用中位数填补{null_cost}个缺失的单位成本)device_ids {d.id for d in devices}for _, row in df.iterrows():labor {}for d in device_ids:col f{d}工时if col in df.columns:labor[d] float(row[col])products.append(Product(idrow[产品ID],namerow[产品名称],demandfloat(row[需求]),unit_costfloat(row[单位成本]),labor_hourslabor,))return products# ─── 约束矩阵构建器 ──────────────────────────────────────────────────────class ConstraintMatrixBuilder:从清洗后的数据构建LP约束矩阵参考: 北理工《运筹学》§2.2 线性规划的标准形式def __init__(self, devices: List[Device], products: List[Product]):self.devices {d.id: d for d in devices}self.products {p.id: p for p in products}def build_pulp_model(self) - pulp.LpProblem:构建PuLP线性规划模型目标: 最小化总生产成本约束: 设备工时 ≤ 最大工时, 产量 ≥ 需求prob pulp.LpProblem(Production_Planning, pulp.LpMinimize)# 决策变量: x[p] 产品p的产量x {}for pid, prod in self.products.items():if prod.demand 0: # 只对有需求的产品建模x[pid] pulp.LpVariable(fx_{pid}, lowBound0, catContinuous)# 目标函数: 总成本prob pulp.lpSum(self.products[pid].unit_cost * x[pid] for pid in x), Total_Cost# 设备工时约束: Σ(x_p × t_pm) ≤ Cap_mfor did, device in self.devices.items():cap device.max_hours * device.efficiencyterms []for pid in x:t_pm self.products[pid].labor_hours.get(did, 0.0)if t_pm 0:terms.append(t_pm * x[pid])if terms:prob pulp.lpSum(terms) cap, fCap_{did}# 需求约束: x_p ≥ Demand_pfor pid in x:demand self.products[pid].demandif demand 0:prob x[pid] demand, fDemand_{pid}return probdef export_matrix_report(self) - str:导出约束矩阵的文字报告lines []lines.append( * 60)lines.append(LP约束矩阵报告)lines.append( * 60)# 目标函数lines.append(\n【目标函数系数 c】)for pid, prod in self.products.items():if prod.demand 0:lines.append(f x_{pid} ({prod.name}): {prod.unit_cost} 元/件)# 约束lines.append(\n【设备约束 (A矩阵行)】)for did, device in self.devices.items():cap device.max_hours * device.efficiencyline f {did} ({device.name}) [上限{cap:.0f}h]: coeffs []for pid, prod in self.products.items():if prod.demand 0:t prod.labor_hours.get(did, 0.0)if t 0:coeffs.append(f{t}×x_{pid})line .join(coeffs) f ≤ {cap:.0f}lines.append(line)lines.append(\n【需求约束 (b向量)】)for pid, prod in self.products.items():if prod.demand 0:lines.append(f x_{pid} ({prod.name}) ≥ {prod.demand:.0f})return \n.join(lines)# ─── 主流程 ──────────────────────────────────────────────────────────────class CapacityDataPipeline:工厂产能数据管道主控制器OOP设计: 组合Reader Cleaner Builderdef __init__(self, excel_path: str sample_data.xlsx):self.reader ExcelDataReader(excel_path)self.cleaner DataCleaner()self.devices: List[Device] []self.products: List[Product] []self.model: Optional[pulp.LpProblem] Nonedef run(self, verbose: bool True) - pulp.LpProblem:执行完整管道if verbose:print( 步骤1: 读取Excel数据...)raw_sheets self.reader.read_all()if verbose:for name, df in raw_sheets.items():print(f ✓ {name}: {df.shape[0]}行 × {df.shape[1]}列)# 清洗设备台账if verbose:print(\n 步骤2: 清洗设备台账...)self.devices self.cleaner.clean_devices(raw_sheets[设备台账])# 清洗月度工时if verbose:print( 步骤3: 清洗月度工时...)raw_sheets[月度工时] self.cleaner.clean_hours(raw_sheets[月度工时])# 清洗BOMif verbose:print( 步骤4: 清洗产品BOM...)self.products self.cleaner.clean_bom(raw_sheets[产品BOM], self.devices)# 构建LP模型if verbose:print(\n ️ 步骤5: 构建LP约束矩阵...)builder ConstraintMatrixBuilder(self.devices, self.products)self.model builder.build_pulp_model()# 输出报告if verbose:print(\n 步骤6: 数据质量报告)self._print_quality_report()print(\n 约束矩阵预览:)report builder.export_matrix_report()print(report)# 保存LP文件lp_path production_model.lpself.model.writeLP(lp_path)print(f\n LP文件已保存: {lp_path})print(f ✓ 管道执行完成! 模型可直接调用 .solve() 求解)return self.modeldef _print_quality_report(self):打印数据质量报告r self.cleaner.reportif r.missing_values:print(f 缺失值统计:)for col, cnt in r.missing_values.items():print(f {col}: {cnt} 处)else:print( ✅ 无缺失值)if r.anomalies:print(f ⚠️ 异常值检测:)for a in r.anomalies:print(f {a})if r.fixes_applied:print(f 已应用的修复:)for f in r.fixes_applied:print(f ✓ {f})# ─── 演示 ──────────────────────────────────────────────────────────────def demo():运行完整演示print( * 70)print( 工厂产能数据管道 · Excel→清洗→LP约束矩阵构建)print( 参考: 北京理工大学《运筹学》第2章线性规划)print( * 70)print(\n 场景: 汽车零部件厂月度排产数据准备)print( 痛点: 人工处理Excel需16小时且易出错)print( 方案: Python自动化管道3分钟完成\n)pipeline CapacityDataPipeline(sample_data.xlsx)model pipeline.run(verboseTrue)# 可选: 直接求解print(\n 尝试求解模型...)solver pulp.PULP_CBC_CMD(msgFalse)status model.solve(solver)print(f ✅ 求解状态: {pulp.LpStatus[status]})if pulp.LpStatus[status] Optimal:print(f 最优总成本: {pulp.value(model.objective):,.2f} 元)print(\n 生产计划:)for v in model.variables():if v.name.startswith(x_) and v.va利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛