transforemer

import os
import pandas as pd
import numpy as np
import tensorflow as tf
from tensorflow.keras.models import Model
from tensorflow.keras.layers import Input, Dense, Flatten, Reshape, MultiHeadAttention, LayerNormalization, Dropout, Add, GaussianNoise
from tensorflow.keras.optimizers import Adam
from tensorflow.keras.regularizers import l1_l2
from tensorflow.keras.callbacks import EarlyStopping
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import classification_report
import matplotlib.pyplot as plt

# 禁用GPU
os.environ["CUDA_VISIBLE_DEVICES"] = "-1"
tf.config.set_visible_devices([], 'GPU')

# 加载本地数据并查看前几行
data = pd.read_csv('/home/xjy/pythonProject/US_Accidents_March23.csv')
print("DataFrame 中的列名：", data.columns)
print(data.head())

# 假设找到实际的标签列 'Severity'
actual_label_column = 'Severity'

# 数据预处理
data = data.dropna()
features = data.select_dtypes(include=[np.number])
labels = data[actual_label_column] - 1  # 使用正确的列名，假设标签从1开始

scaler = StandardScaler()
features = scaler.fit_transform(features)

# 按照7:1:2的比例划分训练集、验证集和测试集
X_train, X_temp, y_train, y_temp = train_test_split(features, labels, test_size=0.3, random_state=42)
X_val, X_test, y_val, y_test = train_test_split(X_temp, y_temp, test_size=0.6667, random_state=42)

# 输出划分结果
print(f"训练集大小: {X_train.shape[0]}")
print(f"验证集大小: {X_val.shape[0]}")
print(f"测试集大小: {X_test.shape[0]}")

# 构建Transformer模型
def transformer_encoder(inputs, head_size, num_heads, ff_dim, dropout=0):
    # Multi-Head Attention layer
    attention = MultiHeadAttention(key_dim=head_size, num_heads=num_heads, dropout=dropout)(inputs, inputs)
    attention = Add()([attention, inputs])
    attention = LayerNormalization(epsilon=1e-6)(attention)

    # Feed Forward network
    ff = Dense(ff_dim, activation='relu', kernel_regularizer=l1_l2(l1=0.01, l2=0.01))(attention)
    ff = Dense(inputs.shape[-1])(ff)
    ff = Add()([ff, attention])
    ff = LayerNormalization(epsilon=1e-6)(ff)
    return ff

input_layer = Input(shape=(X_train.shape[1], 1))
reshape = Reshape((X_train.shape[1], 1))(input_layer)

# 增加Transformer编码器层的数量和复杂度
transformer_block = transformer_encoder(reshape, head_size=64, num_heads=4, ff_dim=64, dropout=0.2)
transformer_block = transformer_encoder(transformer_block, head_size=64, num_heads=4, ff_dim=64, dropout=0.2)
transformer_block = transformer_encoder(transformer_block, head_size=64, num_heads=4, ff_dim=64, dropout=0.2)

flatten = Flatten()(transformer_block)
noise = GaussianNoise(0.1)(flatten)
dense = Dense(64, activation='relu', kernel_regularizer=l1_l2(l1=0.01, l2=0.01))(noise)
dropout = Dropout(0.6)(dense)
output_layer = Dense(4, activation='softmax')(dropout)

model = Model(inputs=input_layer, outputs=output_layer)

# 编译模型
model.compile(optimizer=Adam(learning_rate=0.001), loss='sparse_categorical_crossentropy', metrics=['accuracy'])

# 训练模型
X_train = np.expand_dims(X_train, axis=-1)
X_val = np.expand_dims(X_val, axis=-1)
X_test = np.expand_dims(X_test, axis=-1)

# 设置 EarlyStopping 回调
early_stopping = EarlyStopping(monitor='val_loss', patience=5, restore_best_weights=True)

history = model.fit(X_train, y_train, epochs=50, batch_size=32, validation_data=(X_val, y_val), callbacks=[early_stopping])

# 评估模型
train_acc = history.history['accuracy'][-1]
val_acc = history.history['val_accuracy'][-1]
test_loss, test_acc = model.evaluate(X_test, y_test)

print(f'Training Accuracy: {train_acc}')
print(f'Validation Accuracy: {val_acc}')
print(f'Test Accuracy: {test_acc}')

# 画出 loss 和 accuracy 图
train_loss = history.history['loss']
val_loss = history.history['val_loss']
train_accuracy = history.history['accuracy']
val_accuracy = history.history['val_accuracy']

plt.figure(figsize=(12, 5))

plt.subplot(1, 2, 1)
plt.plot(train_loss, label='Training Loss')
plt.plot(val_loss, label='Validation Loss')
plt.title('Training and Validation Loss Over Epochs')
plt.xlabel('Epochs')
plt.ylabel('Loss')
plt.legend()
plt.grid(True)

plt.subplot(1, 2, 2)
plt.plot(train_accuracy, label='Training Accuracy')
plt.plot(val_accuracy, label='Validation Accuracy')
plt.title('Training and Validation Accuracy Over Epochs')
plt.xlabel('Epochs')
plt.ylabel('Accuracy')
plt.legend()
plt.grid(True)

plt.tight_layout()
plt.show()

# 判断是否过拟合
if train_loss[-1] < val_loss[-1] and train_accuracy[-1] > val_accuracy[-1]:
    print("可能存在过拟合，因为训练损失明显低于验证损失，且训练准确率高于验证准确率。")
else:
    print("训练和验证损失接近，训练和验证准确率也接近，过拟合可能性较小。")

# 保存预测结果
y_pred = model.predict(X_test)
y_pred_classes = np.argmax(y_pred, axis=1)
y_pred_classes_adjusted = y_pred_classes + 1
predictions_adjusted = pd.DataFrame({'True': y_test + 1, 'Predicted': y_pred_classes_adjusted})
predictions_adjusted.to_csv('/home/xjy/pythonProject/predictions_solo.csv', index=False)

# 打印评价指标
print(classification_report(y_test + 1, y_pred_classes_adjusted))