基于LSTM的DDoS检测模型SHAP解释报错问题求助

问题概述 #

我使用LSTM模型进行DDoS攻击检测，模型已完成训练且预测结果准确，但无法生成SHAP解释，运行时出现输入形状不匹配的错误。

模型训练代码 #

#Change features with object or string value into numeric numbers
ord_feat = ['protocol_type', 'service', 'flag']
#Nom_feat = column value with 0 or 1
nom_feat = ['land', 'logged_in', 'is_host_login', 'is_guest_login']
num_feat = ['src_bytes','dst_bytes','wrong_fragment','urgent','hot','num_failed_logins','num_compromised','root_shell','su_attempted','num_root','num_file_creations','num_shells','num_access_files','num_outbound_cmds','count','srv_count','serror_rate','srv_serror_rate','rerror_rate','srv_rerror_rate','same_srv_rate','diff_srv_rate','srv_diff_host_rate','dst_host_count','dst_host_srv_count','dst_host_same_srv_rate','dst_host_diff_srv_rate','dst_host_same_src_port_rate','dst_host_srv_diff_host_rate','dst_host_serror_rate','dst_host_srv_serror_rate','dst_host_rerror_rate','dst_host_srv_rerror_rate']

X_train, y_train = df.drop(columns=['class'], axis=1, inplace=False), df['class'].values

ohe = OneHotEncoder(sparse=False)
oe = OrdinalEncoder()

ohe.fit(X_train[nom_feat].values)
oe.fit(X_train[ord_feat].values)

scalar = StandardScaler()
scalar.fit(X_train[num_feat].values)
X_train_nom = ohe.transform(X_train[nom_feat].values)
X_train_ord = oe.transform(X_train[ord_feat].values)
X_train_num = scalar.transform(X_train[num_feat].values)

X_train = np.concatenate([X_train_ord, X_train_num, X_train_nom], axis=1)
#SVM Approach-------------------------------------------------------------------
from sklearn import svm

classifier = svm.SVC(kernel = "linear")
classifier.fit(X_train, y_train)
y_predict = classifier.predict(X_test)

from sklearn import metrics
print("SVM ACCURACY : ",metrics.accuracy_score(y_test, y_predict))
#SVM Approach Done--------------------------------------------------------------
#reshape Train dataset into 3d array
X_train = X_train.reshape((X_train.shape[0],1,X_train.shape[1]))
y_train = y_train.reshape((y_train.shape[0],1,1))

#Scale the num, ord, nom datasets
X_test, y_test = df_val.drop(columns=['class'], axis = 1, inplace=False), df_val['class'].values

# 注意：此处不应重新fit编码器和缩放器，需复用训练集拟合结果
ohe.fit(X_test[nom_feat].values)
oe.fit(X_test[ord_feat].values)
#reshape Test dataset into 3d array

scalar.fit(X_test[num_feat].values)
X_test_nom = ohe.transform(X_test[nom_feat].values)
X_test_ord = oe.transform(X_test[ord_feat].values)
X_test_num = scalar.transform(X_test[num_feat].values)
X_test = np.concatenate([X_test_ord, X_test_num, X_test_nom], axis=1)
X_test.shape
X_test = X_test.reshape((X_test.shape[0],1,X_test.shape[1]))
y_test = y_test.reshape((y_test.shape[0],1,1))
model = Sequential()
#50 time steps, and 2 features
#LSTM INOUT (Batch size, Time steps, units)
model.add(LSTM(units = 44, input_shape=(1,44), return_sequences=True))
model.add(Dense(1))
model.add(Dense(1))
model.compile(loss="mean_absolute_error", optimizer = 'adam', metrics = ["accuracy"])
# model.summary()

history = model.fit(X_train, y_train, epochs=30, validation_data = (X_test, y_test))# validation_data=(X_test, y_test)

SHAP初始化代码 #

def initEncodersAndScaler(self):
    df = pd.read_csv("attack_test.csv")
    # Подготовка данных
    X, Y = df.drop(columns=['class'], axis = 1, inplace=False), df['class'].values

    # Обучение энкодеров и масштабировщика
    self.ohe.fit(X[nom_feat].values)
    self.oe.fit(X[ord_feat].values)
    self.scalar.fit(X[num_feat].values)

    X_test_nom = self.ohe.transform(X[nom_feat].values)
    X_test_ord = self.oe.transform(X[ord_feat].values)
    X_test_num = self.scalar.transform(X[num_feat].values)
    X = np.concatenate([X_test_ord, X_test_num, X_test_nom], axis=1)    
    # Создаем маскер с правильным размером
    print(X.shape)
    self.masker = shap.maskers.Independent(data = X)  
    self.explainer = shap.Explainer(self.model, self.masker)  # Создаем объект объяснителя

SHAP调用代码 #

def run(self):
    for index, row in self.data.iterrows():
        if self.stop_flag:
            break  # Если установлен флаг, прекращаем выполнение
        
        # Создаем объект с признаками пакета
        packet_features = NetworkPacketFeatures(row.to_dict())
        
        # Подготовка данных для предсказания
        features = packet_features.to_array()
        print(features.shape)
        explanation = packet_features.explainer(features)
        shap_values = explanation.values  # Получаем значения SHAP
        print(f"SHAP Values для пакета: {shap_values}")  # Выводим значения SHAP
        features = features.reshape((features.shape[0], 1, features.shape[1]))  # Подготовка данных
        prediction = packet_features.model.predict(features)  # Предсказание
        scale_pred = prediction[0][0][0]
        
        self.update_signal.emit(packet_features, scale_pred)  # Отправка сигнала

报错信息 #

ValueError: Input 0 of layer "sequential_5" is incompatible with the layer: expected shape=(None, 1, 44), found shape=(3324, 44)
Aborted (core dumped)

解决方案 #

报错核心原因是SHAP传入模型的数据维度与模型要求的3D输入不匹配，模型期望输入形状为(None, 1, 44)，但实际传入的是2D数组。需从以下两处修改：

1. 修改SHAP初始化代码：调整masker数据维度 #

将准备好的2D数据X转换为模型要求的3D格式后，再传入masker：

def initEncodersAndScaler(self):
    df = pd.read_csv("attack_test.csv")
    # Подготовка данных
    X, Y = df.drop(columns=['class'], axis = 1, inplace=False), df['class'].values

    # Обучение энкодеров и масштабировщика
    self.ohe.fit(X[nom_feat].values)
    self.oe.fit(X[ord_feat].values)
    self.scalar.fit(X[num_feat].values)

    X_test_nom = self.ohe.transform(X[nom_feat].values)
    X_test_ord = self.oe.transform(X[ord_feat].values)
    X_test_num = self.scalar.transform(X[num_feat].values)
    X = np.concatenate([X_test_ord, X_test_num, X_test_nom], axis=1)    
    # 转换为3D维度：(样本数, 时间步, 特征数)
    X = X.reshape((X.shape[0], 1, X.shape[1]))
    print(X.shape)
    self.masker = shap.maskers.Independent(data = X)  
    self.explainer = shap.Explainer(self.model, self.masker)  # Создаем объект объяснителя

2. 修改SHAP调用代码：调整单个样本维度 #

单个数据包的features是2D数组，需先转换为3D格式再传入explainer：

def run(self):
    for index, row in self.data.iterrows():
        if self.stop_flag:
            break  # Если установлен флаг, прекращаем выполнение
        
        # Создаем объект с признаками пакета
        packet_features = NetworkPacketFeatures(row.to_dict())
        
        # Подготовка данных для предсказания
        features = packet_features.to_array()
        # 转换为3D维度：(1, 1, 44)，匹配模型输入要求
        features_3d = features.reshape((features.shape[0], 1, features.shape[1]))
        print(features_3d.shape)
        explanation = packet_features.explainer(features_3d)
        shap_values = explanation.values  # Получаем значения SHAP
        print(f"SHAP Values для пакета: {shap_values}")  # Выводим значения SHAP
        
        prediction = packet_features.model.predict(features_3d)  # Предсказание
        scale_pred = prediction[0][0][0]
        
        self.update_signal.emit(packet_features, scale_pred)  # Отправка сигнала

额外优化：修正数据预处理错误 #

在模型训练代码中，测试集的编码器和缩放器不应重新fit，需复用训练集的拟合结果，否则会导致数据分布不一致，影响模型性能和解释效果：

# 替换测试集预处理代码：
# 删除以下三行fit操作
# ohe.fit(X_test[nom_feat].values)
# oe.fit(X_test[ord_feat].values)
# scalar.fit(X_test[num_feat].values)

# 直接用训练好的编码器和缩放器转换测试集
X_test_nom = ohe.transform(X_test[nom_feat].values)
X_test_ord = oe.transform(X_test[ord_feat].values)
X_test_num = scalar.transform(X_test[num_feat].values)

内容的提问来源于stack exchange，提问作者VolkKiiroi