基于C-MAPSS数据集的航空发动机剩余寿命预测实战指南

1次阅读

共计 4799 个字符，预计需要花费 12 分钟才能阅读完成。

航空发动机的预测性维护是工业领域的重要课题，而 C -MAPSS 数据集作为该领域的标准基准，在实际应用中面临多重挑战：

传感器噪声干扰：发动机运行环境复杂，原始传感器数据包含大量噪声，直接影响特征提取效果
工况变化影响：不同飞行阶段（起飞、巡航、降落）导致传感器读数存在显著差异
数据不平衡问题：接近失效时的样本数量远少于正常状态样本
时间序列特性：传统统计方法难以捕捉长期依赖关系

对比传统方法（如指数退化模型）与深度学习方法：

传统方法依赖人工定义退化曲线，难以适应不同发动机的个体差异
简单神经网络容易过拟合短期波动而忽略整体趋势
LSTM 等时序模型能自动学习退化模式，但需要精心设计输入窗口

# Python 3.8+
import numpy as np
import pandas as pd
from sklearn.preprocessing import MinMaxScaler

def sliding_window(data, window_size=30):
    """构建时间滑动窗口样本"""
    X, y = [], []
    for unit_id in data['unit'].unique():
        unit_data = data[data['unit']==unit_id].sort_values('cycle')
        # 剩余寿命线性递减假设
        rul = np.arange(len(unit_data)-1, -1, -1)  
        for i in range(len(unit_data)-window_size):
            X.append(unit_data.iloc[i:i+window_size, 2:-1].values)  # 排除非传感器列
            y.append(rul[i+window_size])
    return np.array(X), np.array(y)

# 示例用法
data = pd.read_csv('train_FD001.txt', sep=' ', header=None)
scaler = MinMaxScaler()
data.iloc[:, 2:-1] = scaler.fit_transform(data.iloc[:, 2:-1])  # 归一化传感器数据
X_train, y_train = sliding_window(data)

关键处理步骤：

传感器数据归一化：使用 MinMaxScaler 将各传感器值缩放到 [0,1] 区间
滑动窗口构建：设置 30 个周期的时间窗口，每个样本包含多变量时序数据
健康指标计算：采用线性 RUL 标注（实际项目可结合物理模型调整）

import torch
import torch.nn as nn

class LSTMModel(nn.Module):
    def __init__(self, input_size, hidden_size=64):
        super().__init__()
        self.lstm = nn.LSTM(
            input_size=input_size,
            hidden_size=hidden_size,
            batch_first=True,
            num_layers=2
        )
        self.regressor = nn.Sequential(nn.Linear(hidden_size, 32),
            nn.ReLU(),
            nn.Linear(32, 1)
        )

    def forward(self, x):
        out, _ = self.lstm(x)  # [batch, seq_len, hidden]
        return self.regressor(out[:, -1, :])  # 取最后时间步

class TransformerModel(nn.Module):
    def __init__(self, input_size, d_model=64):
        super().__init__()
        self.embed = nn.Linear(input_size, d_model)
        encoder_layer = nn.TransformerEncoderLayer(
            d_model=d_model, 
            nhead=4,
            dim_feedforward=128
        )
        self.encoder = nn.TransformerEncoder(encoder_layer, num_layers=2)
        self.regressor = nn.Linear(d_model, 1)

    def forward(self, x):
        # x: [batch, seq_len, features]
        x = self.embed(x)  # [batch, seq_len, d_model]
        x = x.transpose(0, 1)  # Transformer 需要[seq_len, batch, features]
        out = self.encoder(x)
        return self.regressor(out[-1])  # 取最后时间步

使用 SHAP 值识别关键传感器：

import shap

# 选择 100 个样本作为背景数据
train_background = X_train[:100].reshape(-1, X_train.shape[2])
explainer = shap.DeepExplainer(model, torch.FloatTensor(train_background))

# 计算单个样本的 SHAP 值
sample_idx = 42
shap_values = explainer.shap_values(X_train[sample_idx:sample_idx+1])
shap.plots.waterfall(shap_values[0][0])

典型发现：

传感器 9（高压涡轮出口温度）对 RUL 预测贡献最大
部分传感器在特定工况下才显现重要性
传感器间存在协同效应（需设计交叉特征）

完整训练循环示例：

from torch.utils.data import DataLoader, TensorDataset

dataset = TensorDataset(torch.FloatTensor(X_train), 
    torch.FloatTensor(y_train)
)
dataloader = DataLoader(dataset, batch_size=64, shuffle=True)

model = LSTMModel(input_size=X_train.shape[2])
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
criterion = nn.MSELoss()

for epoch in range(50):
    model.train()
    for batch_x, batch_y in dataloader:
        pred = model(batch_x).flatten()
        loss = criterion(pred, batch_y)

        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

    # 验证集评估
    model.eval()
    with torch.no_grad():
        val_pred = model(torch.FloatTensor(X_val)).flatten()
        rmse = torch.sqrt(criterion(val_pred, torch.FloatTensor(y_val)))
    print(f"Epoch {epoch}: Train Loss {loss.item():.2f}, Val RMSE {rmse:.2f}")

除 RMSE 外，使用 NASA 提供的评分函数：

$$ score = \sum_{i=1}^N \begin{cases}
e^{-d_i/13}-1, & \text{if} d_i < 0 \
e^{d_i/10}-1, & \text{otherwise}
\end{cases} $$

其中 $d_i = \text{pred}_i – \text{true}_i$

实现代码：

def nasa_score(y_true, y_pred):
    d = y_pred - y_true
    return np.sum(np.where(d < 0, np.exp(-d/13)-1, np.exp(d/10)-1))

训练曲线：监控训练 / 验证损失变化，检测过拟合
预测对比图：选取测试样本展示真实值与预测曲线
残差分析：检查误差分布是否均匀

量化：使用 PyTorch 的量化工具

model_quant = torch.quantization.quantize_dynamic(model, {nn.LSTM, nn.Linear}, dtype=torch.qint8
)

剪枝：移除不重要的神经元连接

from torch.nn.utils import prune

parameters_to_prune = [(module, 'weight') 
    for module in filter(lambda m: isinstance(m, nn.Linear), model.modules())]
prune.global_unstructured(parameters_to_prune, pruning_method=prune.L1Unstructured, amount=0.3)

设置数据漂移检测模块（如 KS 检验）
保留 10% 的验证集用于模型微调

实现增量学习接口：

def partial_fit(model, new_data, epochs=5):
    optimizer = torch.optim.SGD(model.parameters(), lr=1e-4)
    for _ in range(epochs):
        loss = criterion(model(new_data[0]), new_data[1])
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

典型错误：

在划分训练测试集前进行全局归一化（应分别计算统计量）
使用未来数据构建特征（如滑动窗口包含未来周期）

正确做法：

# 正确的数据流
scaler = MinMaxScaler()
train_data = scaler.fit_transform(train_raw)  # 只在训练集上 fit
test_data = scaler.transform(test_raw)  # 测试集用相同 scaler

推荐搜索空间：

学习率：对数空间[1e-5, 1e-3]
隐藏层大小：[32, 64, 128]
窗口长度：[20, 30, 50]

使用 Optuna 框架示例：

import optuna

def objective(trial):
    params = {'lr': trial.suggest_float('lr', 1e-5, 1e-3, log=True),
        'hidden_size': trial.suggest_categorical('hidden_size', [32, 64, 128]),
        'window_size': trial.suggest_int('window_size', 20, 50)
    }
    # 构建模型并训练
    return validation_rmse

study = optuna.create_study(direction='minimize')
study.optimize(objective, n_trials=50)