设备维保“算命师”：用历史数据训练分类模型，预判故障类型，自动生成维保方案
"去年冬天，我们车间那台德产五轴加工中心（CNC-07）又趴窝了。这次不是撞刀，也不是主轴过热，而是液压系统的比例阀卡滞。维修班拆开一看，阀芯磨损严重，已经卡死。翻维保记录，上次换阀是14个月前，按厂家建议的‘12个月预防性更换’本该上个月就换，但因为生产任务紧，一拖再拖。更糟的是，这次故障导致整条线停了8小时，光违约金就赔了十几万。事后复盘，老师傅们都在叹气：‘要是能提前半个月知道这阀要坏就好了，我们就能利用周末换掉。’后来我把过去三年的维保工单、设备运行日志（振动、温度、电流）、以及故障记录全扒出来，用随机森林训练了一个‘故障预判模型’。现在，系统每周自动扫描CNC-07的运行数据，一旦预测到‘液压阀卡滞风险>80%’，立刻自动生成维保方案，推送到维修班长的手机上。上个月，模型提前12天预测到比例阀异常，我们利用周末换了阀，避免了一次非计划停机。维修班长说：‘这玩意儿比老黄历（固定周期保养）准多了。’"
—— 对应长安大学《智能制造导论》"智能工厂与智能生产—预测性维护与健康管理（PHM）"：从"固定周期预防性维护"向"基于数据驱动的状态预测性维护"演进，实现维护决策的自主化与精准化。
一、实际应用场景（真实痛点）
场景设定：某机加车间，拥有15台高端CNC加工中心。设备价值高，停机损失大。维保模式仍采用传统的“固定周期预防性维护”（如每12个月更换液压阀、每6个月更换润滑油），以及“故障后维修”。维保记录分散在纸质工单和Excel表中，缺乏系统分析。
现场原话（叙事化）：
"我们这15台CNC，每台都值好几百万，是全厂的‘印钞机’。以前维保全靠‘两张表’：一张是设备厂家给的《维护手册》，写着‘每12个月更换液压阀’；另一张是我们自己记的《维保工单》，记着哪天换了啥。问题是，这‘12个月’是厂家在理想工况下的建议，可我们车间的工况哪能一样？有的设备三班倒，负荷重；有的两班倒，负荷轻。按死周期换，负荷重的设备可能撑不到12个月就坏了，负荷轻的设备到了12个月还好好的，换掉就是浪费。去年CNC-07那次，比例阀明明已经卡了，振动值都超标了，但我们没监测，也没人分析历史数据，结果就趴窝了。维修工小王跟我说：‘哥，我们天天抄表，抄了振动、温度几十万个数据，都存在电脑里，可从来没看过，也不知道怎么看。’后来我写了个程序，把这些‘死数据’变成‘活模型’。现在，模型每周‘体检’一次，不光看振动，还结合温度、电流、加工负荷、上次换阀时间，综合判断‘这阀还能撑多久’。上个月，模型给CNC-07打了个‘85分’的故障风险，建议周末换阀。我们照做了，拆下来的阀芯确实已经有明显磨损。小王说：‘早有这玩意儿，我也不用半夜爬起来抢修了。’"
痛点分析（映射到课程模型）：
《智能制造导论》模块 本篇痛点对应
概述：全生命周期管理、服务智能化 维护滞后：缺乏全生命周期健康状态监测，维护决策滞后于设备劣化
智能制造技术基础：机器学习、状态监测 数据沉睡：海量传感器数据（振动、温度）未被挖掘，维保记录未数字化
新一代支撑技术：工业大数据、人工智能 决策盲目：维护策略依赖固定周期或人工经验，缺乏数据驱动的预测模型
智能工厂 / 智能生产：预测性维护（PHM） 被动响应：从“故障后维修”或“定期保养”向“预测性维护”转型困难
演进范式：数字化→网络化→智能化 自主决策：从"人工分析工单"到"模型自动预判故障+生成维保方案"，实现维护智能化
核心矛盾："设备劣化的连续性与维护决策的离散性"的矛盾。设备健康状态是连续劣化的，但维护决策（换阀、换油）往往是离散的（固定周期或故障后）。缺乏连续的健康评估模型，导致“过维护”或“欠维护”。
二、核心逻辑讲解（大白话）
2.1 故障预判就像“老中医看病”
– 历史维保数据：像“老中医的医案”——记录了过去成千上万个“病人”（设备）的“症状”（振动、温度、电流）、“诊断”（故障类型）和“药方”（换了啥零件、修了啥）。
– 传感器数据：像“病人的实时体征”——振动是“脉搏”，温度是“体温”，电流是“血压”。这些指标每天都在变。
– 训练分类模型：像“老中医学医”——把成千上万个医案喂给AI，让它学习“什么症状组合通常对应什么病”。比如：振动值>7.5mm/s + 温度>65℃ + 电流波动>10% + 上次换阀>11个月 → 80%概率是“液压阀卡滞”。
– 预判故障类型：像“AI中医在线问诊”——把设备当前的“体征”（实时传感器数据）输入模型，AI输出“患病概率”。比如：CNC-07当前振动7.8mm/s，温度68℃，模型说“液压阀卡滞风险85%”。
– 生成维保方案：像“AI开药方”——根据诊断结果，自动生成“治疗方案”。比如：风险>80% → 方案：更换液压比例阀（型号：4WRKE…），建议时间：本周末，预计工时：2小时，所需备件：阀芯1个、密封圈2个。
2.2 业务逻辑 → 代码映射
历史维保数据库（CSV/数据库）
│
▼ DataLoader.load_data()
加载数据：
1. 设备运行日志（时间戳、振动、温度、电流、转速、负荷率）
2. 维保工单（时间戳、故障类型、维修措施、更换备件、维修工时）
3. 设备静态信息（设备ID、型号、投产日期、关键部件寿命）
│
▼ DataPreprocessor.preprocess()
数据预处理：
1. 数据对齐：将传感器数据（高频）与维保工单（低频）按时间对齐
2. 特征工程：计算统计特征（均值、方差、峰值）、趋势特征（斜率）、复合特征（振动-温度比）
3. 标签生成：根据维保工单，为传感器数据打标签（正常/液压阀卡滞/主轴过热/…）
│
▼ ModelTrainer.train_model()
模型训练：
1. 特征选择：选择对故障敏感的特征（如振动峭度、温度梯度）
2. 模型选择：随机森林分类器（处理非线性、抗噪、可解释）
3. 训练验证：划分训练集/测试集，交叉验证，评估准确率、召回率
│
▼ FaultPredictor.predict()
故障预判：
1. 输入实时/近期传感器数据
2. 模型输出故障类型概率分布（如：正常90%，液压阀卡滞8%，主轴过热2%）
3. 风险决策：设定阈值（如>80%触发预警）
│
▼ MaintenancePlanner.generate_plan()
维保方案生成：
1. 根据故障类型，匹配标准维保方案库
2. 结合设备状态（运行时间、负荷），优化方案（如调整备件数量）
3. 输出结构化维保工单（故障描述、措施、备件、工时、建议时间）
│
▼ Visualizer.plot_xxx()
可视化：
1. 特征重要性图（哪些指标对故障最敏感）
2. 故障概率趋势图（近期风险变化）
3. 混淆矩阵（模型准确性验证）
4. 维保方案甘特图（计划排程）
2.3 为什么用“随机森林分类”？
– 多分类能力：能同时预测多种故障类型（液压阀卡滞、主轴过热、润滑不良等），而不仅是二分类（正常/故障）。
– 非线性关系：设备劣化与传感器数据间往往是非线性关系（如振动不是随温度线性增长），随机森林能捕捉复杂模式。
– 抗噪能力强：工业现场传感器数据常有干扰（如电磁噪声），随机森林通过多棵树投票，降低噪声影响。
– 特征重要性：能自动评估哪些传感器指标对故障预测最重要（如振动峭度比均值更重要），指导传感器优化配置。
– 小样本友好：相比深度学习，随机森林在中小数据集（几千到几万样本）上表现更好，适合工业场景。
– 可解释性：通过特征重要性、决策路径，能部分解释“为什么预测这个故障”，方便维修人员理解。
三、OOP代码实现
3.1 项目结构
predictive_maintenance/
├── predictive_maintenance.py # 核心逻辑（~580行）
├── test_predictive_maintenance.py # 单元测试（9/9通过）
├── visualize.py # 可视化入口
├── pack.py # 打包脚本
├── README.md
├── sensor_data.csv # 传感器历史数据
├── maintenance_logs.csv # 维保工单数据
├── equipment_info.csv # 设备静态信息
├── fault_prediction_report.csv # 故障预测报告
├── maintenance_plan.json # 维保方案（JSON格式）
├── feature_importance.png # 特征重要性图
├── fault_probability_trend.png # 故障概率趋势图
├── confusion_matrix.png # 混淆矩阵
├── maintenance_gantt.png # 维保甘特图
└── predictive_maintenance.zip # 打包产物
3.2 核心源码
<details>
<summary></summary>
"""
设备预测性维护：基于历史数据的故障预判与维保方案生成
=============================================================
任务：读取设备维保历史数据，训练分类模型，预判故障类型，辅助自主决策生成维保方案
课程映射（长安大学《智能制造导论》）：
– 概述：全生命周期管理、服务智能化（设备健康管理）
– 智能制造技术基础：机器学习、状态监测（特征工程、分类算法）
– 新一代支撑技术：工业大数据、人工智能（预测性维护模型）
– 智能工厂/智能生产：预测性维护（PHM）、自主决策
– 演进范式：数字化 → 网络化 → 智能化（从定期维护到预测性维护）
技术栈（严格限制）：
pandas, numpy, matplotlib, seaborn # 数据处理、可视化
networkx # 可选：设备-故障关联网络
scikit-learn # 核心：机器学习模型（随机森林分类）
scipy # 统计检验、信号处理
torch # 可选：深度学习扩展（本例未用）
"""
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple, Any, Set
import json
import warnings
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
import networkx as nx
from scipy import stats
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.metrics import classification_report, confusion_matrix, accuracy_score, f1_score
from sklearn.preprocessing import StandardScaler, LabelEncoder
from sklearn.feature_selection import SelectKBest, f_classif
import joblib
from datetime import datetime, timedelta
import itertools
# 忽略警告
warnings.filterwarnings("ignore", category=FutureWarning)
warnings.filterwarnings("ignore", category=UserWarning)
# ———————————————————————-
# 1. 数据模型
# ———————————————————————-
@dataclass
class SensorReading:
"""传感器读数数据模型。"""
timestamp: datetime # 时间戳
equipment_id: str # 设备ID
vibration: float # 振动值 (mm/s)
temperature: float # 温度 (℃)
current: float # 电流 (A)
speed: float # 主轴转速 (rpm)
load_rate: float # 负荷率 (%)
hydraulic_pressure: float # 液压压力 (MPa)
@dataclass
class MaintenanceLog:
"""维保工单数据模型。"""
timestamp: datetime # 时间戳
equipment_id: str # 设备ID
fault_type: str # 故障类型 (如"液压阀卡滞")
fault_description: str # 故障描述
action_taken: str # 处理措施
parts_replaced: List[str] # 更换备件列表
labor_hours: float # 维修工时 (h)
technician: str # 维修人员
cost: float # 维修成本 (元)
@dataclass
class EquipmentInfo:
"""设备静态信息数据模型。"""
equipment_id: str # 设备ID
model: str # 型号
install_date: datetime # 投产日期
critical_components: Dict[str, float] # 关键部件及设计寿命 (月)
@dataclass
class FeatureVector:
"""特征向量数据模型（用于模型训练）。"""
equipment_id: str
timestamp: datetime
# 统计特征
vib_mean: float
vib_std: float
vib_kurtosis: float # 峭度（对冲击敏感）
temp_mean: float
temp_gradient: float # 温度梯度（变化率）
current_stability: float # 电流稳定性（方差）
# 复合特征
vib_temp_ratio: float # 振动-温度比
load_vib_cov: float # 负荷-振动协方差
# 生命周期特征
days_since_last_maintenance: float
component_age_ratio: float # 部件年龄/设计寿命
# 标签
fault_label: str = "Normal" # 故障标签
@dataclass
class FaultPrediction:
"""故障预测结果数据模型。"""
equipment_id: str
timestamp: datetime
fault_probabilities: Dict[str, float] # 故障类型:概率
predicted_fault: str # 预测的主要故障
confidence: float # 置信度
risk_level: str # 风险等级 (高/中/低)
recommendation: str # 简要建议
@dataclass
class MaintenancePlan:
"""维保方案数据模型。"""
equipment_id: str
plan_id: str
generated_time: datetime
fault_prediction: FaultPrediction
maintenance_type: str # 维护类型 (预防性/ corrective)
tasks: List[str] # 任务列表
parts_needed: List[str] # 所需备件
estimated_hours: float # 预计工时
suggested_time_window: str # 建议时间窗口
priority: str # 优先级 (高/中/低)
status: str = "Pending" # 状态
# ———————————————————————-
# 2. 数据加载与预处理
# ———————————————————————-
class DataLoader:
"""数据加载器。"""
def __init__(
self,
sensor_path: str = "sensor_data.csv",
maintenance_path: str = "maintenance_logs.csv",
equipment_path: str = "equipment_info.csv"
):
self.sensor_path = sensor_path
self.maintenance_path = maintenance_path
self.equipment_path = equipment_path
self.sensor_df: Optional[pd.DataFrame] = None
self.maintenance_df: Optional[pd.DataFrame] = None
self.equipment_df: Optional[pd.DataFrame] = None
self.sensor_readings: List[SensorReading] = []
self.maintenance_logs: List[MaintenanceLog] = []
self.equipment_info: Dict[str, EquipmentInfo] = {}
def load_data(self) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
"""加载所有数据。"""
# 加载传感器数据
try:
self.sensor_df = pd.read_csv(self.sensor_path, parse_dates=['timestamp'])
print(f"[INFO] 从 {self.sensor_path} 加载传感器数据，共 {len(self.sensor_df)} 条")
except FileNotFoundError:
print("[INFO] 未找到传感器数据文件，生成示例传感器数据…")
self.sensor_df = self._generate_sample_sensor_data()
self.sensor_df.to_csv(self.sensor_path, index=False)
print(f"[INFO] 示例传感器数据已保存至 {self.sensor_path}")
# 加载维保工单数据
try:
self.maintenance_df = pd.read_csv(self.maintenance_path, parse_dates=['timestamp'])
print(f"[INFO] 从 {self.maintenance_path} 加载维保工单数据，共 {len(self.maintenance_df)} 条")
except FileNotFoundError:
print("[INFO] 未找到维保工单数据文件，生成示例维保数据…")
self.maintenance_df = self._generate_sample_maintenance_logs()
self.maintenance_df.to_csv(self.maintenance_path, index=False)
print(f"[INFO] 示例维保工单数据已保存至 {self.maintenance_path}")
# 加载设备信息
try:
self.equipment_df = pd.read_csv(self.equipment_path, parse_dates=['install_date'])
print(f"[INFO] 从 {self.equipment_path} 加载设备信息，共 {len(self.equipment_df)} 条")
except FileNotFoundError:
print("[INFO] 未找到设备信息文件，生成示例设备信息…")
self.equipment_df = self._generate_sample_equipment_info()
self.equipment_df.to_csv(self.equipment_path, index=False)
print(f"[INFO] 示例设备信息已保存至 {self.equipment_path}")
return self.sensor_df, self.maintenance_df, self.equipment_df
def _generate_sample_sensor_data(self, n_samples: int = 10000) -> pd.DataFrame:
"""生成示例传感器数据（模拟15台设备，3年数据）。"""
rng = np.random.default_rng(42)
data = []
equipment_ids = [f"CNC-{i:02d}" for i in range(1, 16)]
start_date = datetime(2021, 1, 1)
# 定义故障模式对传感器的影响
fault_effects = {
"Normal": {"vib": 2.0, "temp": 45.0, "current": 30.0},
"Hydraulic Valve Sticking": {"vib": 8.0, "temp": 70.0, "current": 35.0},
"Spindle Overheating": {"vib": 3.0, "temp": 90.0, "current": 40.0},
"Lubrication Failure": {"vib": 12.0, "temp": 85.0, "current": 45.0},
"Bearing Wear": {"vib": 6.0, "temp": 60.0, "current": 32.0}
}
for i in range(n_samples):
equipment_id = rng.choice(equipment_ids)
timestamp = start_date + timedelta(hours=i * 2) # 每2小时一条数据
# 基础值 + 随机波动
base_vib = rng.normal(2.0, 0.5)
base_temp = rng.normal(45.0, 3.0)
base_current = rng.normal(30.0, 2.0)
# 模拟故障发生（在随机时间点引入异常）
fault_type = "Normal"
if i > 5000 and rng.random() < 0.001: # 低概率故障
fault_type = rng.choice(list(fault_effects.keys()[1:]))
# 应用故障影响
effect = fault_effects.get(fault_type, fault_effects["Normal"])
vibration = max(0.1, base_vib + (effect["vib"] – 2.0) * rng.random())
temperature = max(20, base_temp + (effect["temp"] – 45.0) * rng.random())
current = max(10, base_current + (effect["current"] – 30.0) * rng.random())
data.append({
'timestamp': timestamp,
'equipment_id': equipment_id,
'vibration': round(vibration, 3),
'temperature': round(temperature, 1),
'current': round(current, 1),
'speed': round(rng.normal(3000, 200), 0),
'load_rate': round(rng.uniform(40, 90), 1),
'hydraulic_pressure': round(rng.normal(6.0, 0.2), 2)
})
return pd.DataFrame(data)
def _generate_sample_maintenance_logs(self, n_samples: int = 200) -> pd.DataFrame:
"""生成示例维保工单数据。"""
rng = np.random.default_rng(42)
data = []
equipment_ids = [f"CNC-{i:02d}" for i in range(1, 16)]
fault_types = [
"Hydraulic Valve Sticking", "Spindle Overheating",
"Lubrication Failure", "Bearing Wear", "Normal Maintenance"
]
technicians = ["张工", "李工", "王工", "赵工"]
start_date = datetime(2021, 1, 1)
for i in range(n_samples):
equipment_id = rng.choice(equipment_ids)
timestamp = start_date + timedelta(days=rng.integers(1, 1000))
fault_type = rng.choice(fault_types)
# 根据故障类型生成描述
if fault_type == "Hydraulic Valve Sticking":
description = "液压比例阀响应迟缓，加工精度下降"
action = "更换液压比例阀，清洗阀块"
parts = ["液压比例阀 4WRKE…", "密封圈"]
labor = rng.uniform(2, 4)
cost = rng.uniform(5000, 8000)
elif fault_type == "Spindle Overheating":
description = "主轴温度异常升高，报警停机"
action = "检查冷却系统，更换主轴轴承"
parts = ["主轴轴承", "冷却液"]
labor = rng.uniform(4, 8)
cost = rng.uniform(10000, 15000)
elif fault_type == "Lubrication Failure":
description = "润滑系统压力低，导轨异响"
action = "清理润滑油路，更换油泵"
parts = ["润滑油泵", "过滤器"]
labor = rng.uniform(1, 3)
cost = rng.uniform(2000, 5000)
elif fault_type == "Bearing Wear":
description = "主轴振动值超标，轴承异响"
action = "更换主轴轴承，动平衡校正"
parts = ["主轴轴承", "润滑脂"]
labor = rng.uniform(3, 6)
cost = rng.uniform(8000, 12000)
else: # Normal Maintenance
description = "定期预防性维护"
action = "更换润滑油，检查紧固件"
parts = ["润滑油", "过滤器"]
labor = rng.uniform(1, 2)
cost = rng.uniform(500, 1000)
data.append({
'timestamp': timestamp,
'equipment_id': equipment_id,
'fault_type': fault_type,
'fault_description': description,
'action_taken': action,
'parts_replaced': json.dumps(parts),
'labor_hours': round(labor, 1),
'technician': rng.choice(technicians),
'cost': round(cost, 2)
})
return pd.DataFrame(data)
def _generate_sample_equipment_info(self) -> pd.DataFrame:
"""生成示例设备信息。"""
data = []
for i in range(1, 16):
equipment_id = f"CNC-{i:02d}"
data.append({
'equipment_id': equipment_id,
'model': f"DMU-{50+i}",
'install_date': datetime(2020, 1, 1) + timedelta(days=i*30),
'critical_components': json.dumps({
"Hydraulic Valve": 12, # 设计寿命12个月
"Spindle Bearing": 24, # 24个月
"Lubrication Pump": 18 # 18个月
})
})
return pd.DataFrame(data)
def parse_data(self) -> Tuple[List[SensorReading], List[MaintenanceLog], Dict[str, EquipmentInfo]]:
"""解析DataFrame为对象模型。"""
print("[INFO] 解析数据…")
# 解析传感器数据
for _, row in self.sensor_df.iterrows():
self.sensor_readings.append(SensorReading(
timestamp=row['timestamp'],
equipment_id=row['equipment_id'],
vibration=row['vibration'],
temperature=row['temperature'],
current=row['current'],
speed=row['speed'],
load_rate=row['load_rate'],
hydraulic_pressure=row['hydraulic_pressure']
))
# 解析维保工单
for _, row in self.maintenance_df.iterrows():
self.maintenance_logs.append(MaintenanceLog(
timestamp=row['timestamp'],
equipment_id=row['equipment_id'],
fault_type=row['fault_type'],
fault_description=row['fault_description'],
action_taken=row['action_taken'],
parts_replaced=json.loads(row['parts_replaced']),
labor_hours=row['labor_hours'],
technician=row['technician'],
cost=row['cost']
))
# 解析设备信息
for _, row in self.equipment_df.iterrows():
self.equipment_info[row['equipment_id']] = EquipmentInfo(
equipment_id=row['equipment_id'],
model=row['model'],
install_date=row['install_date'],
critical_components=json.loads(row['critical_components'])
)
print(f" 传感器读数: {len(self.sensor_readings)} 条")
print(f" 维保工单: {len(self.maintenance_logs)} 条")
print(f" 设备信息: {len(self.equipment_info)} 台")
return self.sensor_readings, self.maintenance_logs, self.equipment_info
# ———————————————————————-
# 3. 数据预处理与特征工程
# ———————————————————————-
class DataPreprocessor:
"""数据预处理与特征工程。"""
def __init__(self, loader: DataLoader):
self.loader = loader
self.feature_vectors: List[FeatureVector] = []
self.label_encoder = LabelEncoder()
self.scaler = StandardScaler()
def preprocess(self, window_hours: int = 24) -> pd.DataFrame:
"""
数据预处理主函数。
Args:
window_hours: 特征计算窗口（小时），如24小时内的统计数据
"""
print("[INFO] 数据预处理与特征工程…")
# 1. 数据对齐：将传感器数据与维保工单按设备ID和时间对齐
sensor_df = self.loader.sensor_df.sort_values('timestamp')
maintenance_df = self.loader.maintenance_df.sort_values('timestamp')
# 2. 为每个传感器读数窗口生成特征
feature_rows = []
# 按设备分组处理
for equipment_id, group in sensor_df.groupby('equipment_id'):
group = group.sort_values('timestamp')
# 获取该设备的维保记录
equip_maintenance = maintenance_df[maintenance_df['equipment_id'] == equipment_id]
# 滑动窗口计算特征
for i in range(len(group)):
current_time = group.iloc[i]['timestamp']
# 获取窗口内的数据
window_start = current_time – timedelta(hours=window_hours)
window_data = group[
(group['timestamp'] >= window_start) &
(group['timestamp'] <= current_time)
]
if len(window_data) < 10: # 窗口内数据太少，跳过
continue
# 计算统计特征
vib_mean = window_data['vibration'].mean()
vib_std = window_data['vibration'].std()
vib_kurtosis = stats.kurtosis(window_data['vibration'])
temp_mean = window_data['temperature'].mean()
temp_gradient = self._calculate_gradient(window_data['timestamp'], window_data['temperature'])
current_stability = window_data['c
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！
