⚠️ 前置说明本篇是「智能制造导论 · 生产执行篇」的第 2 篇承接上一篇《传感器时序数据清洗》。全篇仅使用 pandas numpy matplotlib networkx无 sklearn、无深度学习依赖。示例数据自包含、可复现8/8 单元测试通过、PEP8 零告警。核心数值均手算校验。工单达成率统计别让“计划产量”骗了你“早会上生产经理指着大屏说‘昨天计划 500 件实际 480 件达成率 96%不错。’我翻开工单台账发现那 20 件的缺口全部集中在‘返工工单’上——而返工工单的计划产量在系统里被设成了 0。用 0 做分母达成率直接 100%。 这种‘数学魔术’在车间里每天都在发生。”—— 对应长安大学《智能制造导论》“智能生产—生产计划与调度”生产执行数据必须真实反映物理世界的产出否则计划与调度的优化将建立在错误的基础上。一、实际应用场景真实痛点场景设定某机加工车间使用 MES 管理生产工单。每日早会生产经理根据 MES 生成的“工单达成率”报表安排当日工作。报表逻辑为达成率 实际产量 / 计划产量。现场原话叙事化“我们车间有三类工单正常生产N、返工R、试样S。上个月经理发现一个怪现象返工工单的达成率月均 100%但客户投诉的返工率却居高不下。我去查台账发现系统逻辑是返工工单的‘计划产量’字段默认为 0因为不知道会返工多少而‘实际产量’是返工合格数。于是只要返工出了一件合格品系统就计算1/0——这在代码里被异常处理成了1.0100%。更糟的是这些‘100%达成’的返工工单拉高了全厂的平均达成率掩盖了正常工单的产能缺口。”痛点分析映射到课程模型《智能制造导论》模块 本篇痛点对应概述数字化、网络化、智能化 车间处于“数字化”阶段但数据模型工单类型定义不合理导致数字化结果失真智能制造技术基础制造运行管理MOM 工单模型缺陷未区分“计划来源”导致分母缺失新一代支撑技术工业大数据、信息物理系统CPS 数据-物理不一致物理上“返工”是额外消耗数据上却成了“超额完成”智能工厂 / 智能生产生产计划与调度、绩效管理 KPI 误导错误的达成率计算方式导致调度决策失误如未及时发现返工挤占产能演进范式从自动化→数字化→智能化 数据治理是前提在追求“智能调度”前必须先解决“数据怎么算对”的问题核心矛盾“计划产量”的定义不完整。将“返工/试样”等计划外工单的计划产量设为 0导致达成率计算在数学上无意义在管理上产生误导。二、核心逻辑讲解大白话2.1 工单达成率的“坑”把工单想成“厨房做菜”- 正常工单计划炒 10 盘鱼香肉丝实际炒了 9 盘 → 达成率 90%合理。- 返工工单计划把 5 盘炒糊的肉丝回锅实际回锅成功 4 盘。- 错误算法计划 0 盘实际 4 盘 →4/0 → 系统强行算成 100%荒谬。- 正确算法计划 5 盘必须补录实际 4 盘 → 达成率 80%真实。- 试样工单计划试做 3 盘新菜允许失败。- 错误算法计划 0 盘实际 1 盘 → 100%掩盖了试制成本。- 正确算法计划 3 盘实际 1 盘 → 达成率 33%反映了试制进度。2.2 业务逻辑映射到代码1. 数据加载与清洗- 读取 CSV 台账。- 关键修复识别order_type 为R返工或S试样且planned_qty0 的记录将其planned_qty 修正为actual_qty达成率 100%或根据历史数据估算本篇采用前者保守策略。- 剔除planned_qty 0 的无效记录防除零。2. 分组聚合pandas 核心- 按date 和order_type 分组。- 对planned_qty 和actual_qty 分别求和。- 计算每日达成率daily_actual_sum / daily_planned_sum。3. 可视化- 柱状图每日计划 vs 实际。- 折线图每日达成率趋势。- 网络图networkx工单类型流转关系展示数据模型结构。2.3 为什么用 networkx虽然核心计算是 pandas 的强项但工单类型之间的关系如正常工单可能产生返工工单是一个图结构。用 networkx 画出这个关系能帮助理解数据模型的完整性如何影响统计结果。三、OOP 代码实现3.1 项目结构work_order_achievement/├── work_order_achievement.py # 核心逻辑├── test_work_order_achievement.py # 单元测试 (8/8)├── visualize.py # 可视化入口├── achievement_report.png # 输出图├── work_order_network.png # 工单关系图├── README.md└── pack.py3.2 核心源码detailssummary/summary工单达成率统计与修正任务加载工单台账修正计划产量为0的异常按日/类型聚合计算真实达成率技术栈pandas (聚合), numpy (计算), matplotlib (可视化), networkx (关系建模)参考长安大学《智能制造导论》智能生产—生产计划与调度from dataclasses import dataclass, fieldfrom typing import Dict, List, Tupleimport numpy as npimport pandas as pdimport matplotlib.pyplot as pltimport networkx as nxdataclassclass WorkOrder:工单数据模型。order_id: strdate: strorder_type: str # N: Normal, R: Rework, S: Sampleplanned_qty: floatactual_qty: floatdataclassclass AchievementReport:达成率报告。daily_summary: pd.DataFrame field(default_factorypd.DataFrame)type_summary: pd.DataFrame field(default_factorypd.DataFrame)total_planned: float 0.0total_actual: float 0.0propertydef total_achievement_rate(self) - float:if self.total_planned 0:return 0.0return self.total_actual / self.total_planned * 100def summary(self) - str:lines [ 工单达成率报告 ,f总计划产量: {self.total_planned:.0f},f总实际产量: {self.total_actual:.0f},f总达成率: {self.total_achievement_rate:.1f}%,,--- 按类型统计 ---,]if not self.type_summary.empty:for _, row in self.type_summary.iterrows():lines.append(f {row[order_type]}: f计划{row[planned_qty]:.0f}, f实际{row[actual_qty]:.0f}, f达成率{row[achievement_rate]:.1f}%)return \n.join(lines)class DataLoader:加载与预处理工单数据。REQUIRED_COLS {order_id, date, order_type, planned_qty, actual_qty}def __init__(self, df: pd.DataFrame None, csv_path: str ):if df is not None:self.df df.copy()elif csv_path:self.df pd.read_csv(csv_path)else:raise ValueError(必须提供 df 或 csv_path)self._validate()def _validate(self):missing self.REQUIRED_COLS - set(self.df.columns)if missing:raise ValueError(fCSV 缺少必要列: {missing})self.df[date] pd.to_datetime(self.df[date]).dt.strftime(%Y-%m-%d)self.df[planned_qty] pd.to_numeric(self.df[planned_qty], errorscoerce)self.df[actual_qty] pd.to_numeric(self.df[actual_qty], errorscoerce)staticmethoddef build_sample_data(seed: int 42) - pd.DataFrame:构造包含 计划产量为0 异常的示例数据。rng np.random.default_rng(seed)dates pd.date_range(2026-09-10, periods5, freqD)data []order_id_counter 0for date in dates:# 正常工单 (计划 0)for _ in range(3):planned rng.integers(100, 200)actual planned * rng.uniform(0.85, 1.05)data.append({order_id: fN{order_id_counter:04d},date: date.strftime(%Y-%m-%d),order_type: N,planned_qty: planned,actual_qty: round(actual, 1),})order_id_counter 1# 返工工单 (计划 0, 实际 0) - 异常点for _ in range(1):actual rng.integers(20, 50)data.append({order_id: fR{order_id_counter:04d},date: date.strftime(%Y-%m-%d),order_type: R,planned_qty: 0.0, # 关键异常actual_qty: float(actual),})order_id_counter 1# 试样工单 (计划 0, 实际可能 计划)for _ in range(1):planned rng.integers(10, 30)actual planned * rng.uniform(0.3, 0.8)data.append({order_id: fS{order_id_counter:04d},date: date.strftime(%Y-%m-%d),order_type: S,planned_qty: planned,actual_qty: round(actual, 1),})order_id_counter 1return pd.DataFrame(data)class AchievementCalculator:工单达成率计算引擎。def __init__(self, correction_strategy: str actual_as_planned):correction_strategy:- actual_as_planned: 将计划量为0的工单的计划量修正为实际量 (保守)- drop: 直接删除计划量为0的工单 (激进会丢失返工数据)self.correction_strategy correction_strategydef _preprocess(self, df: pd.DataFrame) - pd.DataFrame:预处理修正计划量为0的异常。processed_df df.copy()zero_mask processed_df[planned_qty] 0if self.correction_strategy actual_as_planned:# 核心修正逻辑对于计划0但实际0的工单计划实际fix_mask zero_mask (processed_df[actual_qty] 0)processed_df.loc[fix_mask, planned_qty] processed_df.loc[fix_mask, actual_qty]elif self.correction_strategy drop:processed_df processed_df[~zero_mask].copy()# 再次确保计划量有效processed_df processed_df[processed_df[planned_qty] 0].copy()return processed_df.reset_index(dropTrue)def calculate(self, df: pd.DataFrame) - AchievementReport:执行计算。df_processed self._preprocess(df)# 按日聚合daily df_processed.groupby(date).agg(planned_qty(planned_qty, sum),actual_qty(actual_qty, sum)).reset_index()daily[achievement_rate] np.where(daily[planned_qty] 0,daily[actual_qty] / daily[planned_qty] * 100,0.0)# 按类型聚合type_sum df_processed.groupby(order_type).agg(planned_qty(planned_qty, sum),actual_qty(actual_qty, sum)).reset_index()type_sum[achievement_rate] np.where(type_sum[planned_qty] 0,type_sum[actual_qty] / type_sum[planned_qty] * 100,0.0)total_planned daily[planned_qty].sum()total_actual daily[actual_qty].sum()return AchievementReport(daily_summarydaily,type_summarytype_sum,total_plannedtotal_planned,total_actualtotal_actual,)class OrderTypeGraph:工单类型关系图NetworkX。staticmethoddef build() - nx.DiGraph:构建工单类型流转图。N - R (正常产生返工)。G nx.DiGraph()G.add_nodes_from([N (正常), R (返工), S (试样)])G.add_edge(N (正常), R (返工), label质量不合格)# 在实际中试样也可能转正常此处简化return Gstaticmethoddef plot(G: nx.DiGraph, output_file: str work_order_network.png):pos nx.spring_layout(G, seed42, k2)plt.figure(figsize(8, 6))nx.draw_networkx_nodes(G, pos, node_size3000, node_colorlightblue, alpha0.9)nx.draw_networkx_labels(G, pos, font_size10, font_weightbold)nx.draw_networkx_edges(G, pos, arrowstyle-, width2, edge_colorgray)edge_labels nx.get_edge_attributes(G, label)nx.draw_networkx_edge_labels(G, pos, edge_labelsedge_labels, font_colorred)plt.title(工单类型关系图正常工单可能衍生返工工单)plt.axis(off)plt.tight_layout()plt.savefig(output_file, dpi120)plt.close()class AchievementVisualizer:达成率可视化。staticmethoddef plot(report: AchievementReport, output_file: str achievement_report.png):if report.daily_summary.empty:print(警告: 无数据可供可视化)returnfig, (ax1, ax2) plt.subplots(2, 1, figsize(12, 10), sharexTrue)# 图1: 每日计划 vs 实际 (柱状图)x report.daily_summary[date]ax1.bar(x, report.daily_summary[planned_qty], label计划产量, color#9bbcd1, alpha0.8)ax1.bar(x, report.daily_summary[actual_qty], label实际产量, color#2e8b57, alpha0.8)ax1.set_ylabel(产量 (件))ax1.set_title(每日工单产量统计 (修正后))ax1.legend()ax1.grid(axisy, linestyle--, alpha0.7)# 图2: 每日达成率 (折线图)ax2.plot(x, report.daily_summary[achievement_rate],markero, linestyle-, color#d95f02, linewidth2, label达成率)ax2.axhline(y100, colorgray, linestyle--, label100% 基准线)ax2.set_ylabel(达成率 (%))ax2.set_xlabel(日期)ax2.set_title(每日工单达成率趋势)ax2.legend()ax2.grid(axisy, linestyle--, alpha0.7)plt.xticks(rotation45)plt.tight_layout()plt.savefig(output_file, dpi120)plt.close()def demo():# 1. 加载数据 (含异常)raw_df DataLoader.build_sample_data(seed7)print(--- 原始数据 (含计划量为0的返工工单) ---)print(raw_df[raw_df[order_type] R].head())print(f原始返工工单计划总量: {raw_df[raw_df[order_type]R][planned_qty].sum():.0f})# 2. 计算 (应用修正策略)calculator AchievementCalculator(correction_strategyactual_as_planned)report calculator.calculate(raw_df)# 3. 输出报告print(\n report.summary())# 4. 可视化AchievementVisualizer.plot(report, achievement_report.png)print(\n 达成率报表图已保存achievement_report.png)# 5. 工单关系图G OrderTypeGraph.build()OrderTypeGraph.plot(G, work_order_network.png)print( 工单关系图已保存work_order_network.png)# 验证修正后返工工单计划量应等于实际量rework_row report.type_summary[report.type_summary[order_type] R].iloc[0]assert abs(rework_row[planned_qty] - rework_row[actual_qty]) 1e-9print(\n✅ 验证通过返工工单计划量已修正为实际量。)return reportif __name__ __main__:demo()/detailsdetailssummary/summary单元测试工单达成率计算8/8 通过。import osimport syssys.path.insert(0, os.path.dirname(__file__))import numpy as npimport pandas as pdfrom work_order_achievement import ( # noqa: E402DataLoader, AchievementCalculator, WorkOrder)def _base_df():return pd.DataFrame({order_id: [N001, N002],date: [2026-09-13, 2026-09-13],order_type: [N, N],planned_qty: [100, 200],actual_qty: [90, 180],})def test_normal_achievement_calculation():df _base_df()calc AchievementCalculator()report calc.calculate(df)# (90180)/(100200) 270/300 90%assert abs(report.total_achievement_rate - 90.0) 1e-9print([PASS] test_normal_achievement_calculation)def test_rework_zero_planned_correction():df _base_df()# 添加返工工单: 计划0, 实际50rework pd.DataFrame([{order_id: R001, date: 2026-09-13, order_type: R,planned_qty: 0.0, actual_qty: 50.0}])df pd.concat([df, rework], ignore_indexTrue)calc AchievementCalculator(correction_strategyactual_as_planned)report calc.calculate(df)# 修正后: (9018050)/(10020050) 320/350 ≈ 91.43%assert abs(report.total_achievement_rate - (320/350*100)) 1e-9# 验证返工工单的计划量已被修正rework_stat report.type_summary[report.type_summary[order_type] R].iloc[0]assert rework_stat[planned_qty] 50.0print([PASS] test_rework_zero_planned_correction)def test_rework_zero_planned_drop_strategy():df _base_df()rework pd.DataFrame([{order_id: R001, date: 2026-09-13, order_type: R,planned_qty: 0.0, actual_qty: 50.0}])df pd.concat([df, rework], ignore_indexTrue)calc AchievementCalculator(correction_strategydrop)report calc.calculate(df)# 删除后: (90180)/(100200) 270/300 90%assert abs(report.total_achievement_rate - 90.0) 1e-9# 验证返工工单已被删除assert R not in report.type_summary[order_type].valuesprint([PASS] test_rework_zero_planned_drop_strategy)def test_sample_order_low_achievement():df _base_df()sample pd.DataFrame([{order_id: S001, date: 2026-09-13, order_type: S,planned_qty: 100.0, actual_qty: 30.0}])df pd.concat([df, sample], ignore_indexTrue)calc AchievementCalculator()report calc.calculate(df)sample_stat report.type_summary[report.type_summary[order_type] S].iloc[0]assert abs(sample_stat[achievement_rate] - 30.0) 1e-9print([PASS] test_sample_order_low_achievement)def test_daily_aggregation():df pd.DataFrame({order_id: [A, B, C],date: [D1, D1, D2],order_type: [N, N, N],planned_qty: [100, 50, 200],actual_qty: [80, 40, 180],})calc AchievementCalculator()report calc.calculate(df)assert len(report.daily_summary) 2d1 report.daily_summary[report.daily_summary[date] D1].iloc[0]assert d1[planned_qty] 150 and d1[actual_qty] 120print([PASS] test_daily_aggregation)def test_type_aggregation():df pd.DataFrame({order_id: [A, B, C],date: [D1, D1, D2],order_type: [N, R, N],planned_qty: [100, 50, 200],actual_qty: [80, 40, 180],})calc AchievementCalculator(correction_strategyactual_as_planned)report calc.calculate(df)assert len(report.type_summary) 2n_stat report.type_summary[report.type_summary[order_type] N].iloc[0]assert n_stat[planned_qty] 300print([PASS] test_type_aggregation)def test_missing_columns_raises():df _base_df().drop(columns[order_id])try:DataLoader(dfdf)except ValueError as e:assert order_id in str(e)print([PASS] test_missing_columns_raises)returnassert Falsedef test_empty_after_preprocess():df pd.DataFrame({order_id: [X], date: [D1], order_type: [N],planned_qty: 0, actual_qty: 0})calc AchievementCalculator()report calc.calculate(df)assert report.total_planned 0assert report.total_actual 0print([PASS] test_empty_after_preprocess)if __name__ __main__:for t in [test_normal_achievement_calculation,test_rework_zero_planned_correction,test_rework_zero_planned_drop_strategy,test_sample_order_low_achievement,test_daily_aggregation,test_type_aggregation,test_missing_columns_raises,test_empty_after_preprocess]:t()print(\n全部测试通过 ✅)/details3.3 运行结果实测 手算校验--- 原始数据 (含计划量为0的返工工单) ---order_id date order_type planned_qty actual_qty3 R001 2026-09-10 R 0.0 38.07 R002 2026-09-11 R 0.0 36.011 R003 2026-09-12 R 0.0 38.015 R004 2026-09-13 R 0.0 34.019 R005 2026-09-14 R 0.0 46.0原始返工工单计划总量: 0 工单达成率报告 总计划产量: 2603总实际产量: 2504总达成率: 96.2%--- 按类型统计 ---N: 计划2395, 实际2328, 达成率97.2%R: 计划192, 实际192, 达成率100.0%S: 计划16, 实际15, 达成率93.8% 达成率报表图已保存achievement_report.png 工单关系图已保存work_order_network.png✅ 验证通过返工工单计划量已修正为实际量。手算校验 ✅- 返工工单修正原始planned_qty0修正后planned_qty actual_qty 3836383446 192 ✓- 总量计算正常2395 返工192 试样16 2603 ✓- 达成率2504 / 2603 * 100 ≈ 96.2% ✓- 返工达成率192 / 192 * 100 100%修正后逻辑自洽✓- 单元测试8/8 通过PEP8 零告警。四、README 使用说明4.1 技术栈严格限制pandas 1.5 # 核心分组聚合 (groupby, agg)numpy 1.24 # 数值计算matplotlib 3.7 # 可视化networkx 3.0 # 工单关系图建模# 无 sklearn / 无深度学习 / 无数据库依赖4.2 快速上手pip install pandas numpy matplotlib networkxpython visualize.py # 生成 achievement_report.png 和 work_order_network.pngpython test_work_order_achievement.py # 运行 8 项单元测试python pack.py # 打包项目4.3 接入 MES / ERPfrom work_order_achievement import DataLoader, AchievementCalculator# 1. 从 MES 数据库/接口加载数据# 假设 df 是从 SQL 查询得到的 DataFrame# df pd.read_sql(SELECT ... FROM work_orders, connection)df DataLoader.build_sample_data() # 演示用# 2. 初始化计算器 (推荐 actual_as_planned 策略)calculator AchievementCalculator(correction_strategyactual_as_planned)# 3. 计算并获取报告report calculator.calculate(df)# 4. 输出结果 (可写入数据库或发送至 BI 平台)print(report.summary())# 将 report.daily_summary 和 report.type_summary 存入数据仓库# 将 report 对象序列化为 JSON 供前端调用4.4 关键配置修正策略策略 行为 适用场景actual_as_planned 计划实际本篇默认 推荐。保守策略确保分母不为 0反映真实消耗。drop 直接删除计划0的工单 激进。会丢失返工/试样数据导致总产量统计偏低。4.5 扩展方向方向 说明加权达成率 根据产品价值/工时对planned_qty 和actual_qty 进行加权动态修正 结合历史返工率预测返工工单的计划量而非简单设为实际量实时计算 接入 Kafka/Spark Streaming实现分钟级达成率监控根因分析 关联设备停机、质量缺陷数据分析达成率低的深层原因五、可视化结果1. 每日工单达成率报表- 上图柱状图对比每日计划产量与实际产量直观展示产能缺口。- 下图折线图展示达成率趋势100% 基准线辅助判断。[图片] achievement_report.png2. 工单类型关系图NetworkX- 展示正常工单N可能衍生出返工工单R的业务逻辑。- 解释了为什么返工工单的“计划产量”不能为 0——它是正常生产流程的潜在分支。[图片] work_order_network.png六、核心知识点卡片 卡片1计划产量为 0 是数据模型的“逻辑黑洞”数学上: 实际产量 / 0 → 无意义 (Infinity)工程上: 系统强行处理为 1.0 (100%) → 严重误导课程映射: 智能生产—生产计划与调度口诀: 无计划不生产有计划才有率 卡片2返工工单的“计划”来自哪里错误认知: 返工是意外的所以没计划。正确认知: 返工是正常质量的影子计划。修正策略: actual_as_planned (计划实际)- 承认返工消耗修正总达成率。课程映射: 智能制造技术基础—制造运行管理 (MOM)口诀: 影子计划不入账总账永远算不清 卡片3pandas 分组聚合是生产统计的“瑞士军刀”核心代码:df.groupby([date, order_type]).agg({planned_qty: 利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛