企业官网建设流程全解析

工单-质检-能耗三表关联：用Python算出每批产品的"能耗账"和"质量账"

周三上午，生产部的小李拿着三张Excel表，一脸愁容地走进办公室。

"哥，帮个忙。"小李把三张表摊在桌上，"厂长昨天开会问我：'咱们不同工单生产的产品，能耗差别大吗？不良率跟能耗有没有关系？'我看着这三张表，头都大了。"

我扫了一眼：第一张是工单表，记录了工单号、产品型号、计划数量、实际产出；第二张是质检表，记录了工单号、检测数量、不良数量、不良类型；第三张是能耗表，记录了工单号、总耗电量（kWh）。

"这不就是三张表关联一下，算两个指标吗？"我说。

"你以为简单？"小李苦笑，"工单表有156行，质检表有203行——因为一个工单可能分多次质检，能耗表有148行——有些工单还没录能耗。我想把三张表按工单号对齐，算每个工单的单位产品能耗和不良率，结果VLOOKUP一拉，要么对不上，要么重复计算。"

"这就是多表关联的问题。"我打开编辑器，"用 pandas 的

"merge"，三行代码搞定。"

import pandas as pd

# 1. 读取三张表

orders = pd.read_csv("orders.csv")

quality = pd.read_csv("quality.csv")

energy = pd.read_csv("energy.csv")

# 2. 质检表按工单聚合（一个工单多次质检，求和）

q_agg = quality.groupby("order_id").agg(

total_inspected=("inspected", "sum"),

total_defects=("defects", "sum")

).reset_index()

q_agg["defect_rate"] = q_agg["total_defects"] / q_agg["total_inspected"]

# 3. 三表关联

merged = orders.merge(q_agg, on="order_id", how="inner")

merged = merged.merge(energy[["order_id", "energy_kwh"]], on="order_id", how="inner")

# 4. 计算单位产品能耗

merged["energy_per_unit"] = merged["energy_kwh"] / merged["actual_output"]

print(merged[["order_id", "energy_per_unit", "defect_rate"]].head())

"就这些？"小李瞪大了眼睛。

"核心逻辑就这些。"我运行了一下，屏幕上跳出了结果：

order_id energy_per_unit defect_rate

0 WO-1001 2.34 0.0123

1 WO-1002 3.87 0.0341

2 WO-1003 2.18 0.0089

"你看，"我指着屏幕，"WO-1002的单位能耗是WO-1003的1.77倍，不良率也高出近3倍。这说明能耗高的工单，质量也可能有问题——可能是设备参数没调好，空转时间长，既费电又出废品。"

小李把结果截图发给了厂长，附了一句："以前我只知道'电用了多少、废品有多少'，现在我知道'每批产品的能耗和质量到底什么关系'。这一次三表关联，帮我们看见了'看不见的工单画像'。"

一、实际应用场景（真实痛点）

场景设定：制造企业的生产管理涉及多张数据表——工单表记录生产任务和产出，质检表记录检测结果，能耗表记录各工单的电耗。生产管理人员需要将三张表关联，计算每个工单的单位产品能耗（kWh/件）和不良率（%），以评估不同工单的生产效率和质量水平，发现高能耗、高不良率的异常工单，为工艺优化提供数据支撑。面对不同粒度的数据和一对多的关联关系，手工处理极易出错。

现场原话（叙事化）：

"我们车间有句老话：'电费是老板的，废品是自己的。'"小李说，"但到底哪些工单既费电又出废品，以前没人算过。因为数据散在三张表里，工单表在MES里，质检表在QMS里，能耗表在能源管理系统里。每次要分析，得分别导出再手工拼。"

"那你们的BI系统不能自动关联吗？"我问。

"有BI，但IT说'要开发报表得走流程，排期到下个月'。"小李摊手，"厂长明天就要看结果，我等不了。"

"所以你要的是离线多表关联分析——把三张表按工单号对齐，算单位能耗和不良率，再找异常。"

"对。而且不只是算出来。"小李补充，"我还想看趋势——不同产品型号的单位能耗是不是不一样？同一型号不同批次的不良率波动大不大？能耗和不良率之间有没有相关性？如果高能耗的工单不良率也高，那说明问题出在设备参数上，而不是原材料。"

核心矛盾："生产管理需要量化每批产品的能耗效率和质量水平以优化工艺"与"多源数据分散在不同系统中，缺乏快速关联与综合分析能力"之间的冲突。需要一个"三表关联分析程序"，自动对齐工单、质检、能耗数据，计算关键指标并挖掘关联规律。

二、痛点分析（映射到长安大学《智能制造导论》课程模型）

《智能制造导论》模块 本篇痛点对应

概述：生产管理、制造执行 工单管理：工单是生产执行的基本单元，关联产出、质量、能耗数据。

智能制造技术基础：制造过程数据采集 数据关联：工单、质检、能耗数据来自不同环节，需要关联分析。

新一代支撑技术：工业大数据（多源数据融合） 数据融合：将分散的工单、质量、能耗数据按工单号关联，形成统一视图。

智能工厂与智能生产：能效管理、质量管理 综合指标：单位产品能耗和不良率是衡量生产效率和质量的核心KPI。

演进范式：手工台账 → 单表统计 → 多表关联分析 → 实时综合看板 从"三张表各看各的"到"一次关联看清全局"，实现生产数据的深度融合。

一句话总结：我们需要构建一个"工单-质检-能耗三表关联分析程序"，通过多表关联计算每个工单的单位产品能耗和不良率，评估生产效率和质量水平，发现异常工单。

三、核心逻辑讲解（大白话）

3.1 问题本质：把三表关联看成"拼拼图"

把三表关联和拼拼图的关系，想象成"把三块碎片拼成一幅完整的画"：

* 工单表 = 拼图底板：每个工单是一块底板，上面有工单号、产品型号、计划数量、实际产出。

* 质检表 = 碎片A：记录每个工单的检测结果。注意：一个工单可能检测了多次（不同班次、不同批次），所以质检表中有多个行对应同一个工单号——这是"一对多"关系。

* 能耗表 = 碎片B：记录每个工单的耗电量。通常一个工单一条记录，但也可能因为数据采集方式而有重复或缺失。

* 关联键 = 拼图的卡口：三张表都有

"order_id"（工单号），这就是把它们拼在一起的"卡口"。

* 聚合 = 把碎片A压扁：质检表一对多，需要按工单号分组，把多次检测的数量和不良数加起来，变成一对一。

* merge = 拼合：用

"pd.merge()" 把三张表按工单号拼在一起，形成一张"大宽表"。

* 计算指标 = 在拼好的图上写字：用产出数量算出单位能耗，用检测数量算出不良率。

工业应用：

* 输入：三张CSV——

"orders.csv"（工单表）、

"quality.csv"（质检表）、

"energy.csv"（能耗表）。

* 聚合：

"quality.groupby("order_id").sum()" 合并同一工单的多次质检。

* 关联：

"orders.merge(q_agg, on="order_id")" 左连接工单和质检，

"merge(energy, on="order_id")" 再关联能耗。

* 计算：

"energy_kwh / actual_output" = 单位产品能耗；

"total_defects / total_inspected" = 不良率。

* 分析：按产品型号分组看平均能耗和不良率，用散点图看能耗与不良率的相关性。

3.2 业务逻辑 → 代码映射

读取三张表

│

▼ DataLoader.load()

数据加载：

1. pd.read_csv() 读取工单、质检、能耗表

2. 数据清洗（去重、缺失值处理）

│

▼ QualityAggregator.aggregate()

质检数据聚合：

1. groupby("order_id") 按工单汇总检测数和不良数

2. 计算不良率

│

▼ DataMerger.merge()

三表关联：

1. orders.merge(q_agg) 工单 + 质检

2. merged.merge(energy) + 能耗

3. 处理未匹配的记录（外连接标记）

│

▼ MetricsCalculator.calculate()

指标计算：

1. 单位产品能耗 = 总能耗 / 实际产出

2. 不良率 = 不良数 / 检测数

3. 按产品型号分组统计

│

▼ Visualizer.plot()

可视化：

1. 散点图：能耗 vs 不良率（看相关性）

2. 柱状图：各产品型号的平均单位能耗

3. 柱状图：各产品型号的平均不良率

4. 箱线图：各型号能耗分布

│

▼ ReportGenerator.generate_report()

生成报告：

1. 异常工单标记（高能耗或高不良率）

2. 综合排名

3. 优化建议

3.3 为什么用

"merge" 而不是直接用Excel VLOOKUP？

*

"merge"：pandas 的向量化关联操作，支持内连接、左连接、右连接、外连接。一行代码完成关联，自动处理重复键，速度快，可复现。

* VLOOKUP：Excel 的单列查找，遇到一对多关系会只返回第一条匹配，需要手动处理重复值。大数据量时卡顿，且公式易错。

* 工程选择：本例用

"merge" 做多表关联，配合

"groupby" 处理一对多关系，确保数据完整性。

3.4 如何处理"质检表一对多"的问题？

* 问题：一个工单可能分多次质检（如每班检测一次），质检表中同一工单号出现多次。直接关联会导致工单表的行被"复制"多次（笛卡尔膨胀）。

* 处理策略：在关联之前，先对质检表按工单号分组聚合，把多次检测的数量和不良数求和，得到每个工单的累计检测数和累计不良数，再计算不良率。这样质检表就变成了一对一的关系。

* 本例处理：在

"QualityAggregator" 中执行

"groupby("order_id").agg({"inspected": "sum", "defects": "sum"})"，然后再关联。

四、OOP 代码实现

4.1 项目结构

order_quality_energy/

├── data/

│ ├── orders.csv # 工单表

│ ├── quality.csv # 质检表

│ └── energy.csv # 能耗表

├── results/ # 输出结果

│ ├── merged_data.csv # 关联后的大宽表

│ ├── scatter_energy_defect.png # 能耗-不良率散点图

│ ├── bar_energy_by_model.png # 各型号单位能耗

│ ├── bar_defect_by_model.png # 各型号不良率

│ ├── box_energy_by_model.png # 各型号能耗箱线图

│ └── analysis_report.txt # 分析报告

├── order_quality_energy.py # 核心代码

├── test_order_quality_energy.py # 单元测试

├── README.md

└── requirements.txt

4.2 核心源码

<details>

<summary></summary>

"""

工单-质检-能耗三表关联分析：统计不同工单的单位产品能耗与不良率

=============================================================================

课程映射（长安大学《智能制造导论》）：

概述：生产管理、制造执行

技术基础：制造过程数据采集

支撑技术：工业大数据（多源数据融合）

智能工厂：能效管理、质量管理

演进范式：手工台账 → 单表统计 → 多表关联分析 → 实时综合看板

技术栈（严格）：

numpy # 数值计算

pandas # 多表关联、聚合

matplotlib # 可视化

networkx # 无

scikit-learn # 无

scipy # 无

torch # 无

"""

from __future__ import annotations

import os

from dataclasses import dataclass, field

from pathlib import Path

from typing import List, Optional, Dict

import numpy as np

import pandas as pd

import matplotlib.pyplot as plt

plt.rcParams["font.sans-serif"] = ["SimHei", "DejaVu Sans"]

plt.rcParams["axes.unicode_minus"] = False

# ----------------------------------------------------------------------

# 1. 配置

# ----------------------------------------------------------------------

@dataclass

class AnalysisConfig:

"""分析配置"""

data_dir: str = "data"

results_dir: str = "results"

# 文件名

orders_file: str = "orders.csv"

quality_file: str = "quality.csv"

energy_file: str = "energy.csv"

# 列名

order_id_col: str = "order_id"

model_col: str = "product_model"

planned_col: str = "planned_qty"

actual_col: str = "actual_output"

inspected_col: str = "inspected_qty"

defects_col: str = "defects"

energy_col: str = "energy_kwh"

# 异常阈值

energy_anomaly_threshold: float = 5.0 # 单位能耗超过5 kWh/件

defect_anomaly_threshold: float = 0.05 # 不良率超过5%

random_seed: int = 42

# ----------------------------------------------------------------------

# 2. 数据加载器

# ----------------------------------------------------------------------

class DataLoader:

"""数据加载器"""

def __init__(self, config: AnalysisConfig):

self.config = config

self.data_dir = Path(config.data_dir)

os.makedirs(self.data_dir, exist_ok=True)

def generate_synthetic_data(self, n_orders: int = 50):

"""生成模拟数据"""

print(f"[INFO] 生成模拟数据（{n_orders}个工单）...")

np.random.seed(self.config.random_seed)

# 产品型号

models = ["M-A100", "M-B200", "M-C300", "M-D400"]

# 工单表

orders = []

for i in range(1, n_orders + 1):

model = np.random.choice(models)

planned = np.random.randint(500, 2000)

# 实际产出略低于计划（有损耗）

actual = int(planned * np.random.uniform(0.92, 1.0))

orders.append({

"order_id": f"WO-{1000 + i}",

"product_model": model,

"planned_qty": planned,

"actual_output": actual

})

# 质检表（一个工单1-3次质检）

quality = []

for order in orders:

n_inspections = np.random.randint(1, 4)

remaining = order["actual_output"]

for j in range(n_inspections):

if j == n_inspections - 1:

inspected = remaining

else:

inspected = np.random.randint(remaining // 3, remaining // 2)

remaining -= inspected

# 不良率基础值 + 随机波动

base_defect = {"M-A100": 0.01, "M-B200": 0.02, "M-C300": 0.015, "M-D400": 0.03}

defect_rate = base_defect[order["product_model"]] * np.random.uniform(0.5, 2.0)

defects = int(inspected * defect_rate)

quality.append({

"order_id": order["order_id"],

"inspected_qty": inspected,

"defects": defects

})

# 能耗表（每个工单一条，偶尔缺失）

energy = []

for order in orders:

if np.random.random() < 0.05: # 5%概率缺失

continue

# 单位能耗基础值 + 随机波动

base_energy = {"M-A100": 2.0, "M-B200": 3.0, "M-C300": 2.5, "M-D400": 4.0}

energy_per_unit = base_energy[order["product_model"]] * np.random.uniform(0.8, 1.4)

total_energy = energy_per_unit * order["actual_output"]

energy.append({

"order_id": order["order_id"],

"energy_kwh": round(total_energy, 2)

})

# 保存

self.data_dir.mkdir(parents=True, exist_ok=True)

pd.DataFrame(orders).to_csv(self.data_dir / self.config.orders_file, index=False)

pd.DataFrame(quality).to_csv(self.data_dir / self.config.quality_file, index=False)

pd.DataFrame(energy).to_csv(self.data_dir / self.config.energy_file, index=False)

print(f" 工单: {len(orders)}")

print(f" 质检记录: {len(quality)}")

print(f" 能耗记录: {len(energy)}")

return pd.DataFrame(orders), pd.DataFrame(quality), pd.DataFrame(energy)

def load_data(self) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:

"""加载三张表"""

print(f"[INFO] 加载数据...")

orders_path = Path(self.config.data_dir) / self.config.orders_file

quality_path = Path(self.config.data_dir) / self.config.quality_file

energy_path = Path(self.config.data_dir) / self.config.energy_file

if not (orders_path.exists() and quality_path.exists() and energy_path.exists()):

self.generate_synthetic_data()

orders_df = pd.read_csv(orders_path)

quality_df = pd.read_csv(quality_path)

energy_df = pd.read_csv(energy_path)

print(f" 工单: {len(orders_df)} 条")

print(f" 质检: {len(quality_df)} 条")

print(f" 能耗: {len(energy_df)} 条")

return orders_df, quality_df, energy_df

# ----------------------------------------------------------------------

# 3. 质检数据聚合器

# ----------------------------------------------------------------------

class QualityAggregator:

"""聚合质检数据（一对多 → 一对一）"""

def __init__(self, config: AnalysisConfig):

self.config = config

def aggregate(self, quality_df: pd.DataFrame) -> pd.DataFrame:

"""按工单号聚合质检数据"""

print(f"[INFO] 聚合质检数据...")

q_agg = quality_df.groupby(self.config.order_id_col).agg(

total_inspected=(self.config.inspected_col, "sum"),

total_defects=(self.config.defects_col, "sum")

).reset_index()

# 计算不良率

q_agg["defect_rate"] = q_agg["total_defects"] / q_agg["total_inspected"]

print(f" 聚合后: {len(q_agg)} 个工单")

return q_agg

# ----------------------------------------------------------------------

# 4. 数据关联器

# ----------------------------------------------------------------------

class DataMerger:

"""三表关联"""

def __init__(self, config: AnalysisConfig):

self.config = config

def merge(self, orders_df: pd.DataFrame, q_agg: pd.DataFrame,

energy_df: pd.DataFrame) -> pd.DataFrame:

"""关联三张表"""

print(f"[INFO] 关联三张表...")

# 工单 + 质检（左连接，保留所有工单）

merged = orders_df.merge(

q_agg,

on=self.config.order_id_col,

how="left"

)

# + 能耗（内连接，只保留有能耗数据的工单）

merged = merged.merge(

energy_df[[self.config.order_id_col, self.config.energy_col]],

on=self.config.order_id_col,

how="inner"

)

# 计算单位产品能耗

merged["energy_per_unit"] = merged[self.config.energy_col] / merged[self.config.actual_col]

print(f" 关联后: {len(merged)} 条记录")

return merged

# ----------------------------------------------------------------------

# 5. 指标计算器

# ----------------------------------------------------------------------

class MetricsCalculator:

"""计算分析指标"""

def __init__(self, config: AnalysisConfig):

self.config = config

def calculate_by_model(self, merged: pd.DataFrame) -> pd.DataFrame:

"""按产品型号分组统计"""

print(f"[INFO] 按产品型号统计...")

model_stats = merged.groupby(self.config.model_col).agg(

order_count=(self.config.order_id_col, "count"),

avg_energy_per_unit=("energy_per_unit", "mean"),

avg_defect_rate=("defect_rate", "mean"),

std_energy=("energy_per_unit", "std"),

std_defect=("defect_rate", "std"),

total_energy=(self.config.energy_col, "sum"),

total_output=(self.config.actual_col, "sum")

).reset_index()

model_stats["overall_energy_per_unit"] = model_stats["total_energy"] / model_stats["total_output"]

print(f" 产品型号数: {len(model_stats)}")

return model_stats

def find_anomalies(self, merged: pd.DataFrame) -> pd.DataFrame:

"""标记异常工单"""

merged = merged.copy()

merged["energy_anomaly"] = merged["energy_per_unit"] > self.config.energy_anomaly_threshold

merged["defect_anomaly"] = merged["defect_rate"] > self.config.defect_anomaly_threshold

merged["is_anomaly"] = merged["energy_anomaly"] | merged["defect_anomaly"]

n_anomaly = merged["is_anomaly"].sum()

print(f" 异常工单: {n_anomaly} 个")

return merged

# ----------------------------------------------------------------------

# 6. 可视化器

# ----------------------------------------------------------------------

class Visualizer:

"""可视化分析结果"""

def __init__(self, config: AnalysisConfig):

self.config = config

self.results_dir = Path(config.results_dir)

os.makedirs(self.results_dir, exist_ok=True)

def plot_scatter(self, merged: pd.DataFrame):

"""能耗 vs 不良率散点图"""

print(f"[INFO] 绘制能耗-不良率散点图...")

fig, ax = plt.subplots(figsize=(10, 7))

colors = {"M-A100": "#E74C3C", "M-B200": "#3498DB",

"M-C300": "#F39C12", "M-D400": "#27AE60"}

for model in merged[self.config.model_col].unique():

subset = merged[merged[self.config.model_col] == model]

ax.scatter(

subset["energy_per_unit"],

subset["defect_rate"] * 100, # 转百分比

c=colors.get(model, "#999999"),

label=model,

s=80,

alpha=0.7,

edgecolors="white",

linewidth=0.5

)

# 异常阈值线

ax.axvline(self.config.energy_anomaly_threshold, color="red",

linestyle="--", alpha=0.5, label="能耗阈值")

ax.axhline(self.config.defect_anomaly_threshold * 100, color="red",

linestyle="--", alpha=0.5, label="不良率阈值")

ax.set_xlabel("单位产品能耗 (kWh/件)", fontsize=12)

ax.set_ylabel("不良率 (%)", fontsize=12)

ax.set_title("单位产品能耗 vs 不良率", fontsize=14, fontweight="bold")

ax.legend(fontsize=10)

ax.grid(True, alpha=0.3)

plt.tight_layout()

plt.savefig(self.results_dir / "scatter_energy_defect.png",

dpi=150, bbox_inches="tight")

plt.close()

print(f" 已保存: {self.results_dir / 'scatter_energy_defect.png'}")

def plot_bar_energy_by_model(self, model_stats: pd.DataFrame):

"""各型号平均单位能耗柱状图"""

print(f"[INFO] 绘制各型号单位能耗柱状图...")

fig, ax = plt.subplots(figsize=(10, 6))

models = model_stats[self.config.model_col]

energies = model_stats["avg_energy_per_unit"]

bars = ax.bar(models, energies, color=["#E74C3C", "#3498DB", "#F39C12", "#27AE60"],

edgecolor="white", alpha=0.8)

# 标注数值

for bar, val in zip(bars, energies):

ax.text(bar.get_x() + bar.get_width() / 2, bar.get_height() + 0.05,

f"{val:.2f}", ha="center", va="bottom", fontsize=10, fontweight="bold")

ax.set_ylabel("平均单位能耗 (kWh/件)", fontsize=12)

ax.set_title("各产品型号平均单位产品能耗", fontsize=14, fontweight="bold")

ax.grid(True, alpha=0.3, axis="y")

plt.tight_layout()

plt.savefig(self.results_dir / "bar_energy_by_model.png",

dpi=150, bbox_inches="tight")

plt.close()

print(f" 已保存: {self.results_dir / 'bar_energy_by_model.png'}")

def plot_bar_defect_by_model(self, model_stats: pd.DataFrame):

"""各型号平均不良率柱状图"""

print(f"[INFO] 绘制各型号不良率柱状图...")

fig, ax = plt.subplots(figsize=(10, 6))

models = model_stats[self.config.model_col]

defect_rates = model_stats["avg_defect_rate"] * 100 # 转百分比

bars = ax.bar(models, defect_rates, color=["#9B59B6", "#E67E22", "#1ABC9C", "#34495E"],

edgecolor="white", alpha=0.8)

for bar, val in zip(bars, defect_rates):

ax.text(bar.get_x() + bar.get_width() / 2, bar.get_height() + 0.1,

f"{val:.2f}%", ha="center", va="bottom", fontsize=10, fontweight="bold")

ax.set_ylabel("平均不良率 (%)", fontsize=12)

ax.set_title("各产品型号平均不良率", fontsize=14, fontweight="bold")

ax.grid(True, alpha=0.3, axis="y")

plt.tight_layout()

plt.savefig(self.results_dir / "bar_defect_by_model.png",

dpi=150, bbox_inches="tight")

plt.close()

print(f" 已保存: {self.results_dir / 'bar_defect_by_model.png'}")

def plot_box_energy_by_model(self, merged: pd.DataFrame):

"""各型号能耗分布箱线图"""

print(f"[INFO] 绘制各型号能耗箱线图...")

fig, ax = plt.subplots(figsize=(10, 6))

data = [merged[merged[self.config.model_col] == m]["energy_per_unit"]

for m in merged[self.config.model_col].unique()]

bp = ax.boxplot(data, labels=merged[self.config.model_col].unique(),

patch_artist=True)

colors = ["#E74C3C", "#3498DB", "#F39C12", "#27AE60"]

for patch, color in zip(bp["boxes"], colors):

patch.set_facecolor(color)

patch.set_alpha(0.8)

ax.set_ylabel("单位产品能耗 (kWh/件)", fontsize=12)

ax.set_title("各产品型号单位能耗分布", fontsize=14, fontweight="bold")

ax.grid(True, alpha=0.3, axis="y")

plt.tight_layout()

plt.savefig(self.results_dir / "box_energy_by_model.png",

dpi=150, bbox_inches="tight")

plt.close()

print(f" 已保存: {self.results_dir / 'box_energy_by_model.png'}")

# ----------------------------------------------------------------------

# 7. 报告生成器

# ----------------------------------------------------------------------

class ReportGenerator:

"""分析报告生成器"""

def __init__(self, config: AnalysisConfig):

self.config = config

self.results_dir = Path(config.results_dir)

os.makedirs(self.results_dir, exist_ok=True)

def generate(self, merged: pd.DataFrame, model_stats: pd.DataFrame) -> str:

"""生成报告"""

print(f"[INFO] 生成分析报告...")

report_lines = []

report_lines.append("=" * 80)

report_lines.append("工单-质检-能耗关联分析报告")

report_lines.append("=" * 80)

# 概况

report_lines.append(f"\n数据概况:")

利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！