python import numpy as np import pandas as pd import matplotlib.pyplot as plt from sklearn.preprocessing import StandardScaler from tensorflow.keras import layers, Model, backend as K

读取训练数据 (仅示例文件路径)

columns = ['unit', 'time', 'os1', 'os2', 'os3'] + [f'sm{i}' for i in range(21)] train_df = pd.read_csv('train_FD001.txt', sep='\s+', header=None, names=columns)

添加 RUL 标签：每个发动机的最大时间减去当前时间

max_cycle = train_df.groupby('unit')['time'].max().reset_index() max_cycle.columns = ['unit', 'max_cycle'] train_df = train_df.merge(max_cycle, on='unit') train_df['RUL'] = train_df['max_cycle'] - train_df['time'] train_df.drop('max_cycle', axis=1, inplace=True)

### 定义分段线性 RUL 标签
```python
def add_piecewise_rul(df, early_rul=130):
    df['RUL'] = df.groupby('unit')['time'].transform(lambda x: early_rul - x)
    df['RUL'] = df['RUL'].clip(upper=early_rul)  # 早期发动机RUL固定为130
    return df

train_df = add_piecewise_rul(train_df)

特征选择与标准化

sensor_cols = [f'sm{i}' for i in range(1,22)] + ['os1','os2','os3']
features = sensor_cols
# 按 unit 分组归一化，避免数据泄漏
unit_std = train_df.groupby('unit')[features].transform('std')
unit_mean = train_df.groupby('unit')[features].transform('mean')
train_df[features] = (train_df[features] - unit_mean) / unit_std

生成滑动窗口样本

def create_sequences(data, feature_cols, target_col, window_size=30):
    X, y = [], []
    for unit in data['unit'].unique():
        unit_data = data[data['unit']==unit].sort_values('time')
        feat = unit_data[feature_cols].values
        targ = unit_data[target_col].values
        for i in range(len(feat) - window_size):
            X.append(feat[i:i+window_size])
            y.append(targ[i+window_size])
    return np.array(X), np.array(y)

window_size = 30
X_train, y_train = create_sequences(train_df, features, 'RUL', window_size)

构建 LSTM 模型

def build_lstm_model(input_shape):
    inputs = layers.Input(shape=input_shape)
    x = layers.LSTM(64, return_sequences=True)(inputs)
    x = layers.Dropout(0.3)(x)
    x = layers.LSTM(32)(x)
    x = layers.Dropout(0.3)(x)
    outputs = layers.Dense(1)(x)
    model = Model(inputs, outputs)
    model.compile(optimizer='adam', loss='mse', metrics=['mae'])
    return model

model = build_lstm_model((window_size, X_train.shape[2]))
history = model.fit(X_train, y_train, epochs=50, batch_size=128, validation_split=0.1, verbose=0)

评估与预测

# 在测试集上重复相同预处理，并用模型预测 RUL
# 注意：测试集评估时应使用官方 Score 函数
def score_func(y_true, y_pred):
    d = y_true - y_pred
    return np.sum(np.where(d<0, np.exp(-d/13)-1, np.exp(d/10)-1))