📚 课程目录第4章制造业设备故障预测

4. 制造业设备故障预测

任务说明

本案例使用制造业设备传感器时序数据，训练一个二分类模型，预测设备在未来一段时间内是否会发生故障，并给出提前预警时间。输入为设备当前及历史传感器读数，输出为故障概率及预警剩余时间。

数据集与依赖

使用公开的AI4I 2020 Predictive Maintenance Dataset（UCI），包含10,000条设备记录，每行代表一台设备在某时间点的状态。特征包括：Air temperature [K], Process temperature [K], Rotational speed [rpm], Torque [Nm], Tool wear [min]。目标列Machine failure表示该设备是否发生故障（0/1）。

依赖：pandas, scikit-learn, imbalanced-learn（可选，用于处理类别不平衡）。

实现思路

核心任务分为两步：

故障分类：训练随机森林分类器，根据传感器特征预测设备是否故障。
提前预警时间：利用数据集中Tool wear [min]特征作为退化指标，当模型预测故障概率超过阈值时，根据当前工具磨损量估算剩余可用时间（预警时间）。

完整代码

import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import classification_report, confusion_matrix
import numpy as np

# 1. 加载数据（请将文件路径替换为实际路径）
df = pd.read_csv('ai4i2020.csv')

# 2. 特征与标签
feature_cols = ['Air temperature [K]', 'Process temperature [K]',
                'Rotational speed [rpm]', 'Torque [Nm]', 'Tool wear [min]']
X = df[feature_cols]
y = df['Machine failure']

# 3. 划分训练/测试集
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y
)

# 4. 训练随机森林分类器
clf = RandomForestClassifier(n_estimators=200, max_depth=10,
                             random_state=42, class_weight='balanced')
clf.fit(X_train, y_train)

# 5. 评估
y_pred = clf.predict(X_test)
print("=== 分类评估 ===")
print(classification_report(y_test, y_pred, zero_division=0))

# 6. 提前预警时间计算
def predict_failure_with_warning(model, sensor_data, threshold=0.5):
    """
    输入: model - 训练好的分类器
          sensor_data - DataFrame，包含与训练时相同的特征列
          threshold - 故障概率阈值
    返回: 故障预测(0/1), 故障概率, 预警剩余时间(分钟)
    """
    prob = model.predict_proba(sensor_data)[:, 1]  # 故障概率
    pred = (prob >= threshold).astype(int)

    # 预警时间估算：基于工具磨损量
    # 假设工具磨损达到200分钟时必然故障（数据集中的最大磨损阈值）
    max_wear = 200.0
    current_wear = sensor_data['Tool wear [min]'].values
    remaining_time = np.maximum(max_wear - current_wear, 0)

    # 若预测为故障，返回剩余时间；否则返回None表示无预警
    warning_time = []
    for p, rt in zip(pred, remaining_time):
        if p == 1:
            warning_time.append(rt)
        else:
            warning_time.append(None)
    return pred, prob, warning_time

# 7. 示例：对测试集前5条记录进行预测
sample = X_test.head(5).copy()
pred, prob, warn = predict_failure_with_warning(clf, sample)
print("\n=== 示例预测（前5条测试样本）===")
for i in range(len(sample)):
    status = "故障" if pred[i] == 1 else "正常"
    warn_str = f"{warn[i]:.1f} 分钟" if warn[i] is not None else "无预警"
    print(f"样本{i+1}: 预测={status}, 故障概率={prob[i]:.3f}, 预警时间={warn_str}")

运行方式

从UCI下载AI4I 2020数据集，保存为ai4i2020.csv，与脚本置于同一目录。
安装依赖：pip install pandas scikit-learn
运行脚本：python predict_failure.py

验证方法

通过以下标准判断模型是否有效：

分类性能：观察输出的classification_report，重点关注故障类（1）的召回率（recall）。由于故障样本较少，召回率应不低于0.7，表示能捕获大部分真实故障。
预警合理性：检查示例预测输出。对于预测为故障的样本，预警时间应为正值（工具磨损未达上限）；对于正常样本，预警时间为无预警。
阈值调整：若故障类召回率偏低，可降低threshold（如0.3）以增加故障检出率，但会提高误报率。

预期输出（示例）

=== 分类评估 ===
              precision    recall  f1-score   support

           0       0.99      0.99      0.99      1935
           1       0.86      0.78      0.82        65

    accuracy                           0.98      2000
   macro avg       0.92      0.89      0.90      2000
weighted avg       0.98      0.98      0.98      2000

=== 示例预测（前5条测试样本）===
样本1: 预测=正常, 故障概率=0.012, 预警时间=无预警
样本2: 预测=正常, 故障概率=0.008, 预警时间=无预警
样本3: 预测=故障, 故障概率=0.934, 预警时间=37.0 分钟
样本4: 预测=正常, 故障概率=0.021, 预警时间=无预警
样本5: 预测=正常, 故障概率=0.045, 预警时间=无预警

以上输出为基于代码逻辑的预期结果示例，实际数值会因随机种子和数据划分略有不同。判定通过的标准是：故障类召回率≥0.7，且预警时间逻辑正确（故障样本有正数预警时间，正常样本无预警）。

关键说明

预警时间基于Tool wear的线性退化假设，实际工程中应使用更复杂的剩余寿命（RUL）模型。
本案例聚焦于分类+简单预警的闭环，未包含特征工程优化和超参数调优，如需提升精度可自行扩展。
类别不平衡已通过class_weight='balanced'处理，若效果仍不理想，可尝试SMOTE过采样。