CMAPSS数据集实战指南：从数据预处理到RUL预测的完整流程

1次阅读

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

CMAPSS（Commercial Modular Aero-Propulsion System Simulation）数据集是 NASA 提供的航空发动机退化仿真数据集，包含多台发动机在不同工况下的传感器时序数据。数据集分为四个子集（FD001-FD004），复杂度依次递增，其中：

数据组成 ：每行记录包含发动机编号、周期编号（运行时间）、3 个工况参数（海拔、温度等）和 21 个传感器读数（温度、压力、转速等）
任务目标 ：根据传感器历史数据预测发动机的剩余使用寿命（RUL），即从当前时刻到失效的剩余周期数

import pandas as pd
import numpy as np

# 加载训练数据（以 FD001 为例）train_data = pd.read_csv('CMAPSSData/train_FD001.txt', sep=' ', header=None)
train_data.dropna(axis=1, how='all', inplace=True)  # 删除全空列
cols = ['engine_id', 'cycle'] + [f'op_{i}' for i in range(1,4)] + [f'sensor_{i}' for i in range(1,22)]
train_data.columns = cols

采用线性退化假设：每台发动机的最大周期数减去当前周期即为 RUL

# 计算每台发动机的最大运行周期
max_cycles = train_data.groupby('engine_id')['cycle'].max().reset_index()
max_cycles.columns = ['engine_id', 'max_cycle']

# 合并并计算 RUL
train_data = train_data.merge(max_cycles, on='engine_id', how='left')
train_data['RUL'] = train_data['max_cycle'] - train_data['cycle']

对传感器数据执行 Z -score 标准化（注意仅使用训练集统计量）：

from sklearn.preprocessing import StandardScaler

sensor_cols = [f'sensor_{i}' for i in range(1,22)]
scaler = StandardScaler()
train_data[sensor_cols] = scaler.fit_transform(train_data[sensor_cols])

# 计算滑动窗口统计特征（窗口大小 =5）window_size = 5
train_data['sensor_ma'] = train_data.groupby('engine_id')['sensor_1'].transform(lambda x: x.rolling(window_size).mean())

通过相关性分析筛选重要传感器（示例）：

# 计算传感器与 RUL 的相关系数
corr_matrix = train_data[sensor_cols + ['RUL']].corr()
important_sensors = corr_matrix['RUL'].abs().sort_values(ascending=False)[:10].index

import torch
from torch.utils.data import Dataset

class EngineDataset(Dataset):
    def __init__(self, df, seq_length=30):
        self.sequences = []
        for engine_id in df['engine_id'].unique():
            engine_data = df[df['engine_id'] == engine_id]
            for i in range(len(engine_data) - seq_length):
                seq = engine_data.iloc[i:i+seq_length][important_sensors].values
                label = engine_data.iloc[i+seq_length]['RUL']
                self.sequences.append((seq, label))

    def __len__(self):
        return len(self.sequences)

    def __getitem__(self, idx):
        seq, label = self.sequences[idx]
        return torch.FloatTensor(seq), torch.FloatTensor([label])

import torch.nn as nn

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

    def forward(self, x):
        out, _ = self.lstm(x)  # out.shape = (batch, seq_len, hidden_size)
        out = self.linear(out[:, -1, :])  # 只取最后一个时间步
        return out

model = LSTMModel(input_size=len(important_sensors))
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)

for epoch in range(10):
    for seq, label in train_loader:
        optimizer.zero_grad()
        output = model(seq)
        loss = criterion(output, label)
        loss.backward()
        optimizer.step()

from sklearn.metrics import mean_squared_error

# 计算 RMSE
rmse = np.sqrt(mean_squared_error(true_ruls, pred_ruls))

# 早期预测误差分析（前 50 周期）early_pred_error = pred_ruls[true_ruls > 50] - true_ruls[true_ruls > 50]