From 87c80d1bb74ad812515ac223ea5fc4b4e7574b8b Mon Sep 17 00:00:00 2001 From: 43606070 <43606070@qq.com> Date: Tue, 29 Oct 2024 19:29:49 +0800 Subject: [PATCH] =?UTF-8?q?=E6=9B=B4=E6=96=B0=20function.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- function.py | 158 +++++++++++++--------------------------------------- 1 file changed, 40 insertions(+), 118 deletions(-) diff --git a/function.py b/function.py index 729e398..a985fcc 100644 --- a/function.py +++ b/function.py @@ -1,133 +1,55 @@ +import csv import numpy as np -import pandas as pd -import torch -import torch.nn as nn -import torch.optim as optim -from torch.utils.data import DataLoader, TensorDataset from sklearn.model_selection import train_test_split -from sklearn.preprocessing import StandardScaler -from sklearn.metrics import classification_report, confusion_matrix, ConfusionMatrixDisplay +from sklearn.linear_model import LogisticRegression +from sklearn.metrics import classification_report, confusion_matrix from imblearn.over_sampling import SMOTE -from sklearn.utils.class_weight import compute_class_weight -import matplotlib.pyplot as plt -import seaborn as sns -# 1. 加载数据 -# 假设数据在 Pandas DataFrame 中 -data = pd.read_csv("path_to_data.csv") -X = data.drop(columns=['label']).values -y = data['label'].values +# 读取 CSV 数据的函数 +def load_data(file_path): + features = [] + labels = [] + with open(file_path, 'r') as csvfile: + reader = csv.reader(csvfile) + next(reader) # 跳过标题行 + for row in reader: + features.append([float(x) for x in row[:-1]]) # 假设最后一列为标签 + labels.append(int(row[-1])) + return np.array(features), np.array(labels) -# 2. 数据分析与特征分布 -def plot_feature_distributions(data, label_column='label'): - for column in data.columns: - if column != label_column: - sns.kdeplot(data[data[label_column] == 1][column], label='Label 1') - sns.kdeplot(data[data[label_column] == 2][column], label='Label 2') - plt.title(f'Feature Distribution for {column}') - plt.xlabel(column) - plt.ylabel("Density") - plt.legend() - plt.show() +# 评估模型的函数 +def evaluate_model(y_true, y_pred): + print("Confusion Matrix:") + print(confusion_matrix(y_true, y_pred)) + print("\nClassification Report:") + print(classification_report(y_true, y_pred)) -# plot_feature_distributions(data) # 可视化特征分布 +# 主函数 +def main(): + # 加载数据 + file_path = 'data.csv' # 修改为你的 CSV 文件路径 + X, y = load_data(file_path) -# 3. 数据预处理 - 标准化 -scaler = StandardScaler() -X = scaler.fit_transform(X) + # 拆分数据集 + X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42) -# 4. 数据集拆分 -X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42, stratify=y) + # 处理不平衡数据 + smote = SMOTE(random_state=42) + X_train_res, y_train_res = smote.fit_resample(X_train, y_train) -# 5. 数据集平衡 - 使用 SMOTE 过采样 -smote = SMOTE(random_state=42) -X_train_resampled, y_train_resampled = smote.fit_resample(X_train, y_train) + # 训练逻辑回归模型 + model = LogisticRegression(max_iter=1000) + model.fit(X_train_res, y_train_res) -# 6. 类别权重计算 -class_weights = compute_class_weight('balanced', classes=np.unique(y_train_resampled), y=y_train_resampled) -class_weights_tensor = torch.tensor(class_weights, dtype=torch.float) + # 进行预测 + y_pred = model.predict(X_test) -# 7. 数据转换为 Tensor -X_train_tensor = torch.tensor(X_train_resampled, dtype=torch.float32) -y_train_tensor = torch.tensor(y_train_resampled, dtype=torch.long) -X_test_tensor = torch.tensor(X_test, dtype=torch.float32) -y_test_tensor = torch.tensor(y_test, dtype=torch.long) + # 评估模型 + evaluate_model(y_test, y_pred) + +if __name__ == "__main__": + main() -# 使用 DataLoader -batch_size = 32 -train_dataset = TensorDataset(X_train_tensor, y_train_tensor) -train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True) -test_dataset = TensorDataset(X_test_tensor, y_test_tensor) -test_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False) - -# 8. 定义模型 -class SimpleNN(nn.Module): - def __init__(self, input_size, num_classes): - super(SimpleNN, self).__init__() - self.fc1 = nn.Linear(input_size, 128) - self.fc2 = nn.Linear(128, 64) - self.fc3 = nn.Linear(64, num_classes) - self.relu = nn.ReLU() - - def forward(self, x): - x = self.relu(self.fc1(x)) - x = self.relu(self.fc2(x)) - x = self.fc3(x) - return x - -input_size = X_train.shape[1] -num_classes = len(np.unique(y)) -model = SimpleNN(input_size, num_classes) - -# 9. 定义损失函数和优化器 -criterion = nn.CrossEntropyLoss(weight=class_weights_tensor) -optimizer = optim.Adam(model.parameters(), lr=0.001) -scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer, 'min', patience=2, factor=0.5) - -# 10. 训练模型 -def train(model, train_loader, criterion, optimizer): - model.train() - running_loss = 0.0 - for X_batch, y_batch in train_loader: - optimizer.zero_grad() - outputs = model(X_batch) - loss = criterion(outputs, y_batch) - loss.backward() - optimizer.step() - running_loss += loss.item() * X_batch.size(0) - return running_loss / len(train_loader.dataset) - -# 11. 验证模型 -def validate(model, test_loader, criterion): - model.eval() - running_loss = 0.0 - y_true, y_pred = [], [] - with torch.no_grad(): - for X_batch, y_batch in test_loader: - outputs = model(X_batch) - loss = criterion(outputs, y_batch) - running_loss += loss.item() * X_batch.size(0) - _, predicted = torch.max(outputs, 1) - y_true.extend(y_batch.cpu().numpy()) - y_pred.extend(predicted.cpu().numpy()) - return running_loss / len(test_loader.dataset), y_true, y_pred - -# 训练循环 -num_epochs = 20 -for epoch in range(num_epochs): - train_loss = train(model, train_loader, criterion, optimizer) - val_loss, y_true, y_pred = validate(model, test_loader, criterion) - scheduler.step(val_loss) - print(f"Epoch [{epoch+1}/{num_epochs}], Train Loss: {train_loss:.4f}, Validation Loss: {val_loss:.4f}") - -# 12. 评估模型性能 -print("Classification Report:\n", classification_report(y_true, y_pred)) - -# 混淆矩阵 -cm = confusion_matrix(y_true, y_pred) -disp = ConfusionMatrixDisplay(confusion_matrix=cm) -disp.plot() -plt.show()