VAE代码，使用CICIoT2023数据集，已跑出准确率（accuracy）、召回率（recall）、精确率（precision）和F1值（F1-score），但无法绘制混淆矩阵，即类似下面这种图。求补充代码，画出混淆矩阵。
import numpy as np
import tensorflow as tf
from tensorflow.keras.layers import Input, Dense, Lambda, Conv1D, Flatten, SpatialDropout1D, Reshape
from tensorflow.keras.models import Model, Sequential
from tensorflow.keras import backend as K
from sklearn.preprocessing import MinMaxScaler, StandardScaler
import pandas as pd
import time
from tensorflow.keras import backend as K
# 读取数据
csv_path_train = 'CICIoT2023/CICIoT2023/benign.csv'
X_train = pd.read_csv(csv_path_train).values
X_train = np.nan_to_num(MinMaxScaler().fit_transform(StandardScaler().fit_transform(X_train)))
X_train = np.reshape(X_train, (-1, 100, 46))
csv_path_test = 'CICIoT2023/CICIoT2023/ceshi.csv'
Y_test = pd.read_csv(csv_path_test)
Y_test_normal = Y_test[Y_test.label == 'BenignTraffic'].drop(labels='label', axis=1).values
Y_test_normal = np.nan_to_num(MinMaxScaler().fit_transform(StandardScaler().fit_transform(Y_test_normal)))
Y_test_normal = np.reshape(Y_test_normal, (-1, 100, 46))
Y_test_abnormal = Y_test[Y_test.label != 'BenignTraffic'].drop(labels='label', axis=1).values
Y_test_abnormal = np.nan_to_num(MinMaxScaler().fit_transform(StandardScaler().fit_transform(Y_test_abnormal)))
Y_test_abnormal = np.reshape(Y_test_abnormal, (-1, 100, 46))
original_dim = 46 # 特征维度
latent_dim = 2 # 潜在空间维度
intermediate_dim = 256 #中间层维度
batch_size = 100 #训练时的批量大小
# 采样函数
def sampling(args):
z_mean, z_log_var = args
batch = K.shape(z_mean)[0]
dim = K.int_shape(z_mean)[1]
epsilon = K.random_normal(shape=(batch, dim))
return z_mean + K.exp(0.5 * z_log_var) * epsilon
# 编码器
inputs = Input(shape=(100, original_dim))
x = Conv1D(64, kernel_size=3, activation='relu', padding='same')(inputs)
x = Flatten()(x)
z_mean = Dense(latent_dim)(x)
z_log_var = Dense(latent_dim)(x)
z = Lambda(sampling, output_shape=(latent_dim,))([z_mean, z_log_var])
encoder = Model(inputs, [z_mean, z_log_var, z], name='encoder')
# 解码器
latent_inputs = Input(shape=(latent_dim,))
x = Dense(100 * 64, activation='relu')(latent_inputs)
x = Reshape((100, 64))(x) # 将展平的层重新调整为三维
x = Conv1D(64, kernel_size=3, activation='relu', padding='same')(x)
outputs = Conv1D(original_dim, kernel_size=3, activation='sigmoid', padding='same')(x)
decoder = Model(latent_inputs, outputs, name='decoder')
# VAE模型
outputs = decoder(encoder(inputs)[2]) # 连接编码器和解码器
vae = Model(inputs, outputs, name='vae')
# 定义损失函数
reconstruction_loss = tf.keras.losses.binary_crossentropy(K.flatten(inputs), K.flatten(outputs))
reconstruction_loss *= 100 # 对应维度
kl_loss = 1 + z_log_var - K.square(z_mean) - K.exp(z_log_var)
kl_loss = K.sum(kl_loss, axis=-1) * -0.5
vae_loss = K.mean(reconstruction_loss + kl_loss)
vae.add_loss(vae_loss)
vae.compile(optimizer='adam')
test_first_d = Y_test_abnormal.shape[0]
print(f"test_normal:{Y_test_normal.shape[0]}")
print(f"test_abnormal:{Y_test_abnormal.shape[0]}")
# 训练VAE
startTime = time.time() # 开始时间
vae.fit(X_train, epochs=100, batch_size=batch_size, validation_data=(Y_test_normal, None))
losssum1=0
losssum2=0
for i in range(1,batch_size):
loss11=vae.evaluate(Y_test_normal)
losssum1=+loss11
loss22 = vae.evaluate(Y_test_abnormal)
losssum2=+loss22
print(f"loss1:{loss11}")
print(f"loss2:{loss22}")
endTime = time.time() #结束时间
print(f"Took {round((endTime - startTime), 5)} seconds to calculate.")
vae.summary()
print("Testing data shape:", Y_test_normal.shape)
result = vae.evaluate(Y_test_normal)
print("Evaluate result:", result)
loss1 = vae.evaluate(Y_test_normal)
loss2 = vae.evaluate(Y_test_abnormal)
print(f"loss1:{loss1}")
print(f"loss2:{loss2}")
from sklearn.metrics import roc_curve, auc
# 预测部分
def evaluate_vae(X_test, model):
# 获取重构输出
reconstructions = model.predict(X_test)
# 计算重构误差
reconstruction_errors = np.mean(np.abs(X_test - reconstructions), axis=(1, 2))
return reconstruction_errors
# 计算阈值
def find_threshold(y_true, y_pred):
fpr, tpr, thresholds = roc_curve(y_true, y_pred)
# 选择使FPR和TPR之差最大的阈值
optimal_idx = np.argmax(tpr - fpr)
optimal_threshold = thresholds[optimal_idx]
return optimal_threshold
# 获取重构误差
recon_errors_normal = evaluate_vae(Y_test_normal, vae)
recon_errors_abnormal = evaluate_vae(Y_test_abnormal, vae)
# 合并数据
recon_errors = np.concatenate([recon_errors_normal, recon_errors_abnormal])
labels = np.array([0]*len(recon_errors_normal) + [1]*len(recon_errors_abnormal))
# 找到最佳阈值
optimal_threshold = find_threshold(labels, recon_errors)
# 使用阈值进行二分类
predictions = (recon_errors > optimal_threshold).astype(int)
from sklearn.metrics import confusion_matrix # 导入混淆矩阵计算函数
# 计算混淆矩阵
cm = confusion_matrix(labels, predictions)
TP = cm[1, 1]
TN = cm[0, 0]
FP = cm[0, 1]
FN = cm[1, 0]
# 计算评价指标
accuracy = (TP + TN) / (TP + TN + FP + FN)
precision = TP / (TP + FP)
recall = TP / (TP + FN)
f1 = 2 * (precision * recall) / (precision + recall)
# 绘制ROC曲线并计算AUC
fpr, tpr, _ = roc_curve(labels, recon_errors)
roc_auc = auc(fpr, tpr)
print(f"Accuracy: {accuracy}")
print(f"Precision: {precision}")
print(f"Recall: {recall}")
print(f"F1 Score: {f1}")
print(f"AUC: {roc_auc}")
# 绘制ROC曲线
import matplotlib.pyplot as plt
plt.figure()
plt.plot(fpr, tpr, label=f'AUC = {roc_auc:.2f}')
plt.plot([0, 1], [0, 1], 'k--')
plt.xlim([0.0, 1.0])
plt.ylim([0.0, 1.05])
plt.xlabel('False Positive Rate')
plt.ylabel('True Positive Rate')
plt.title('Receiver Operating Characteristic')
plt.legend(loc="lower right")
plt.show()
运行结果ROC曲线：
