深入解析C-MAPSS数据集：从数据预处理到故障预测建模实战

1次阅读

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

C-MAPSS 数据集作为航空发动机退化预测的标杆数据，包含 21 个传感器采集的时序数据（如温度、压力、转速等），但实际使用时会遇到三类典型问题：

传感器噪声干扰 ：原始数据中存在高频抖动（如 FD004 子集的振动传感器），直接建模会导致模型捕捉噪声而非真实退化趋势
工况多样性 ：同一发动机在不同飞行阶段（爬升 / 巡航）的传感器数值差异可达 300%，简单归一化会丢失工况信息
样本不平衡 ：正常样本占比超 80%，早期故障样本不足（如 FD003 中 RUL>150 的样本仅占 6%）

使用滑动窗口滤波处理噪声，配合 Isolation Forest 剔除异常点：

from sklearn.ensemble import IsolationForest
import numpy as np

def smooth_sensor(data, window_size=5):
    return np.convolve(data, np.ones(window_size)/window_size, mode='same')

# 示例：处理 FD001 的传感器 2 数据
train_fd001['s2'] = smooth_sensor(train_fd001['s2'].values)

# 异常检测（contamination 参数根据实际调整）clf = IsolationForest(contamination=0.05)
outliers = clf.fit_predict(train_fd001[['s2','s3']])
clean_data = train_fd001[outliers == 1]

时域特征（均值、方差）结合频域特征（FFT 能量）提升信息密度：

from scipy.fft import fft

def extract_features(window):
    features = {'mean': window.mean(),
        'std': window.std(),
        # 频域能量（取前 5 个主频）'fft_energy': np.sum(np.abs(fft(window)[:5]))
    }
    return features

模型	优点	缺点	适用场景
LSTM	自动学习时序依赖	训练耗时	长序列预测
XGBoost	特征重要性清晰	需手动构造时序特征	快速基线
Transformer	并行计算效率高	需要大量数据	多工况联合建模

from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.feature_selection import VarianceThreshold

preprocessor = Pipeline([('scaler', StandardScaler()),
    # 剔除方差小于 0.1 的特征
    ('selector', VarianceThreshold(threshold=0.1))
])

import torch
import torch.nn as nn

class AttentionLSTM(nn.Module):
    def __init__(self, input_dim):
        super().__init__()
        self.lstm = nn.LSTM(input_dim, 64, batch_first=True)
        self.attention = nn.Sequential(nn.Linear(64, 32),
            nn.Tanh(),
            nn.Linear(32, 1)
        )

    def forward(self, x):
        lstm_out, _ = self.lstm(x)
        # 计算注意力权重
        attn_weights = torch.softmax(self.attention(lstm_out), dim=1)
        return (lstm_out * attn_weights).sum(dim=1)

量化和剪枝 ：将 FP32 模型转为 INT8 可减少 70% 推理延迟
滑动窗口缓存 ：维护最近 50 个时间点的数据队列，避免重复计算

使用 KL 散度检测特征分布变化：

from scipy.stats import entropy

def detect_drift(new_data, baseline_dist):
    current_dist = np.histogram(new_data, bins=20)[0]
    return entropy(baseline_dist, current_dist) > 0.5  # 阈值根据业务调整