test.py

import pandas as pd
import numpy as np


import matplotlib.pyplot as plt
from sklearn.metrics import classification_report, roc_auc_score
from sklearn.preprocessing import OneHotEncoder
from sklearn.preprocessing import MinMaxScaler
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LogisticRegression
from sklearn.tree import DecisionTreeClassifier
from sklearn import tree
from sklearn.ensemble import RandomForestClassifier, \
    GradientBoostingClassifier
from xgboost import XGBClassifier
from imblearn.over_sampling import SMOTE, RandomOverSampler
from imblearn.under_sampling import RandomUnderSampler
import tensorflow as tf
import seaborn as sns
import warnings
from sklearn.svm import SVC
warnings.filterwarnings('ignore')

data = pd.read_csv('data/A08.csv')
label_col = 'RES'
feature_select_bool = True  # 是否做特征选择
imbalance_learning_NN_bool = True # 是否做神经网络的不平衡学习
imbalance_learning_bool = True  # 是否做不平衡学习

plt.rcParams['font.sans-serif'] = ['Times New Roman']
plt.rcParams['font.size'] = 15

def feature_select(train_x, train_y, test_x, method='corr'):
    if method == 'corr':
        # 1.相关系数法(Pearson)
        corr_ = train_x.join(train_y).corr()
        coefficient_ = corr_[train_y.name][:-1]
    if method == 'LR':
        # 2.基于惩罚项的特征选择法
        lr = LogisticRegression(penalty='l1', C=0.1).fit(train_x, train_y)
        coefficient_ = pd.Series(lr.coef_[0], index=train_x.columns)
    if method == 'DT':
        # 3.基于树模型的特征选择法
        dt = DecisionTreeClassifier().fit(train_x, train_y)
        coefficient_ = pd.Series(dt.feature_importances_, index=train_x.columns)
    if method == 'RF':
        # 3.基于树模型的特征选择法
        rf = RandomForestClassifier().fit(train_x, train_y)
        coefficient_ = pd.Series(rf.feature_importances_, index=train_x.columns)

    coefficient_sort = coefficient_.abs().sort_values(ascending=False)
    select_cols = coefficient_sort[coefficient_sort > coefficient_sort.quantile(0.83)].index.tolist()
    train_select_x = train_x[select_cols]
    test_select_x = test_x[select_cols]
    return train_select_x, test_select_x,select_cols

def imbalance_learning(train_x, train_y, method='SMOTE'):
    if method == 'SMOTE':
        sm = SMOTE(random_state=42)
        train_x, train_y = sm.fit_resample(train_x, train_y)
    if method == 'ROS':
        ros = RandomOverSampler(random_state=42)
        train_x, train_y = ros.fit_resample(train_x, train_y)
    if method == 'RUS':
        rus = RandomUnderSampler(random_state=42)
        train_x, train_y = rus.fit_resample(train_x, train_y)
    return train_x, train_y

def plot_kde(data, col, label_col):
    plt.figure(figsize=(10, 5))
    sns.kdeplot(x=col, hue=label_col, data=data, palette='Set2', fill=True)
    plt.tight_layout()
    plt.show()

x = data.drop(labels=[label_col] + [data.columns[0]], axis=1) # 删除第一列数据，axis=1表示按列操作
y = data[label_col]

train_x, test_x, train_y, test_y = train_test_split(x, y, stratify=y, test_size=0.3, random_state=40)

# 初始化SVM模型，并将probabilx`ity参数设置为True

model_names = ['XGBoost', 'SVM', 'RF']
model_list = [XGBClassifier(max_depth=6), SVC(probability=True), RandomForestClassifier(max_depth=6)]

result = pd.DataFrame(
    columns=['AUC', 'accuracy', 'precision0', 'recall0', 'f1-score0', 'precision1', 'recall1', 'f1-score1'],
    index=model_names)
for model_name, model in zip(model_names, model_list):
    if feature_select_bool:
        train_x, test_x, selected_features = feature_select(train_x, train_y, test_x, method='corr')
    if imbalance_learning_bool:
        train_x, train_y = imbalance_learning(train_x, train_y, method='SMOTE')
    model.fit(train_x, train_y)

# 对经过特征筛选后的特征量进行核密度估计图的可视化分析
    """selected_numerical_cols = train_x.select_dtypes(exclude='object').columns
    for col in selected_numerical_cols:
    plot_kde(data, col, label_col)"""

    y_pred = model.predict(test_x)
    y_proba = model.predict_proba(test_x)[:, 1]
    print(str(model))
    cr = classification_report(test_y, y_pred, output_dict=True)
    auc = roc_auc_score(test_y, y_proba)
    result.loc[model_name] = [auc, cr['accuracy'], cr['0']['precision'], cr['0']['recall'],
                              cr['0']['f1-score'], cr['1']['precision'], cr['1']['recall'],
                              cr['1']['f1-score']]


df = (data - data.min()) / (data.max() - data.min())
x = df.drop(labels=label_col, axis=1)
y = df[label_col]


#DNN
train_x, test_x, train_y, test_y = train_test_split(x, y, stratify = y, test_size=0.3, random_state=40)

model = tf.keras.Sequential([
    tf.keras.layers.Dense(64, activation='relu'),
    tf.keras.layers.Dense(32, activation='relu'),
    tf.keras.layers.Dense(1, activation='sigmoid')
])
model.compile(optimizer='adam',
              loss='binary_crossentropy',
              metrics=['accuracy', 'AUC'])
# early stopping，reduce learning rate
if imbalance_learning_bool:
    call_backs = [tf.keras.callbacks.EarlyStopping(patience=20, restore_best_weights=True, monitor='val_auc'),
                  tf.keras.callbacks.ReduceLROnPlateau(patience=10, monitor='val_auc', factor=0.1)]
else:
    call_backs = [tf.keras.callbacks.EarlyStopping(patience=20, restore_best_weights=True, monitor='val_loss')]

if feature_select_bool:
    train_x, test_x, selected_features = feature_select(train_x, train_y, test_x, method='corr')
    for feature in selected_features:
        print(feature)
    print("Total number of selected features:", len(selected_features))

hist = model.fit(train_x, train_y, epochs=100, batch_size=40, validation_data=(test_x, test_y), callbacks=call_backs)

y_proba = model.predict(test_x)
y_pred = np.where(y_proba > 0.5, 1, 0)#将预测的概率值转换为二元分类结果
print(str(model))
cr = classification_report(test_y, y_pred, output_dict=True)
auc = roc_auc_score(test_y, y_proba)
result.loc['DNN'] = [auc, cr['accuracy'], cr['0.0']['precision'], cr['0.0']['recall'], cr['0.0']['f1-score'],
                     cr['1.0']['precision'], cr['1.0']['recall'], cr['1.0']['f1-score']]


#RNN
# Split data into training and testing sets
train_x, test_x, train_y, test_y = train_test_split(x, y, stratify=y, test_size=0.3, random_state=40)

# Define RNN model using Sequential API
model = tf.keras.Sequential([
    tf.keras.layers.Input(shape=(train_x.shape[1],)),
    tf.keras.layers.Reshape((train_x.shape[1], 1)),
    tf.keras.layers.SimpleRNN(64, activation='relu'),
    tf.keras.layers.Dense(1, activation='sigmoid')
])

# Compile model
model.compile(optimizer='adam',
              loss='binary_crossentropy',
              metrics=['accuracy', 'AUC'])

# Train model
model.fit(train_x, train_y, epochs=100, batch_size=60, validation_data=(test_x, test_y), callbacks=call_backs)

y_proba = model.predict(test_x)
y_pred = np.where(y_proba > 0.5, 1, 0)#将预测的概率值转换为二元分类结果
print(str(model))
cr = classification_report(test_y, y_pred, output_dict=True)
result.loc['BiLSTM'] = [auc, cr['accuracy'], cr['0.0']['precision'], cr['0.0']['recall'], cr['0.0']['f1-score'],
                     cr['1.0']['precision'], cr['1.0']['recall'], cr['1.0']['f1-score']]



#CNN

# Split data into training and testing sets
train_x, test_x, train_y, test_y = train_test_split(x, y, stratify=y, test_size=0.3, random_state=40)

# Define CNN model using Sequential API
model = tf.keras.Sequential([
    tf.keras.layers.Conv1D(filters=64, kernel_size=3, activation='relu', input_shape=(train_x.shape[1], 1)),
    tf.keras.layers.MaxPooling1D(pool_size=2),
    tf.keras.layers.Conv1D(filters=128, kernel_size=3, activation='relu'),
    tf.keras.layers.MaxPooling1D(pool_size=2),
    tf.keras.layers.Dropout(0.5),  # Add dropout layer for regularization

    tf.keras.layers.Flatten(),
    tf.keras.layers.Dense(128, activation='relu'),
    tf.keras.layers.Dropout(0.5),  # Add dropout layer for regularization
    tf.keras.layers.Dense(1, activation='sigmoid')
])


# Compile model
model.compile(optimizer='adadelta',
              loss='binary_crossentropy',
              metrics=['accuracy', 'AUC'])
# Train model
model.fit(train_x, train_y, epochs=80, batch_size=50, validation_data=(test_x, test_y), callbacks=call_backs)

y_proba = model.predict(test_x)
y_pred = np.where(y_proba > 0.5, 1, 0)#将预测的概率值转换为二元分类结果
print(str(model))
cr = classification_report(test_y, y_pred, output_dict=True)
result.loc['CNN'] = [auc, cr['accuracy'], cr['0.0']['precision'], cr['0.0']['recall'], cr['0.0']['f1-score'],
                     cr['1.0']['precision'], cr['1.0']['recall'], cr['1.0']['f1-score']]




from transformers import TFDistilBertModel, DistilBertTokenizer
from tensorflow.keras.layers import Input, Dense, GlobalAveragePooling1D
from tensorflow.keras.models import Model
from tensorflow.keras.optimizers import Adam
from tensorflow.keras.losses import BinaryCrossentropy


# 神经网络可视化loss\acc\auc
def plot_learning_curves(history, indicator='loss'):
    plt.figure(figsize=(8, 4))
    plt.plot(history.epoch, history.history[indicator], "r--", label="train")
    plt.plot(history.epoch, history.history["val_" + indicator], "b-", label="val")
    plt.xlabel("Epochs")
    plt.ylabel(indicator)
    plt.legend(loc="upper right")
    plt.grid(True)
    plt.tight_layout()
    plt.show()

plot_learning_curves(hist, indicator='loss')
plot_learning_curves(hist, indicator='auc')
plot_learning_curves(hist, indicator='accuracy')

# 模型结果可视化对比
def indicator_plot(result,indicator):
    plt.rcParams['font.sans-serif'] = ['Times New Roman']
    plt.rcParams['font.size'] = 10
    plt.figure(figsize=(5,3))
    plt.bar(x = np.arange(result.shape[0]),height=result[indicator],color=sns.color_palette('Set2'))
    plt.xticks(np.arange(result.shape[0]),result.index)
    plt.ylim(result[indicator].min()-0.01,result[indicator].max()+0.01)
    plt.title(indicator)
    plt.grid(axis='y',linestyle='--')
    plt.show()

for indicator in result.columns:
     indicator_plot(result,indicator)
#if __name__ == '__main__':
    # print_hi('PyCharm')
    #print(device_lib.list_local_devices())