generated from LufsX/Experiment-LogReg
This commit is contained in:
+113
-99
@@ -1,120 +1,134 @@
|
||||
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 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
|
||||
|
||||
注意事项:
|
||||
- 在 function.py 中实现对率回归算法。
|
||||
- 可以自行导入其他库，需要在 requirements.txt 添加依赖，若未在 requirements.txt 中添加使用到的依赖，则无法自动评测，成绩为零。
|
||||
- 自动评测脚本仅会加载 train_logreg、predict_logreg 和 load_data 函数，请勿修改函数名。
|
||||
# 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()
|
||||
|
||||
测评环境:
|
||||
- CPU: Intel Xeon(R) CPU E3-1231 v3 @ 3.40GHz
|
||||
- OS: Ubuntu 22.04.4 LTS
|
||||
- Python: 3.10.12
|
||||
- numpy: 2.1.1
|
||||
"""
|
||||
# plot_feature_distributions(data) # 可视化特征分布
|
||||
|
||||
"""
|
||||
请在这里填写学号和姓名，不填写即成绩无效。
|
||||
"""
|
||||
id = "202210423118" # 请填写你的学号
|
||||
name = "刘国庆" # 请填写你的姓名
|
||||
# 3. 数据预处理 - 标准化
|
||||
scaler = StandardScaler()
|
||||
X = scaler.fit_transform(X)
|
||||
|
||||
# 4. 数据集拆分
|
||||
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42, stratify=y)
|
||||
|
||||
# Sigmoid函数
|
||||
def sigmoid(z):
|
||||
return 1 / (1 + np.exp(-z))
|
||||
# 5. 数据集平衡 - 使用 SMOTE 过采样
|
||||
smote = SMOTE(random_state=42)
|
||||
X_train_resampled, y_train_resampled = smote.fit_resample(X_train, y_train)
|
||||
|
||||
# 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)
|
||||
|
||||
# 对率回归训练函数
|
||||
def train_logreg(X, y, learning_rate=0.05, iterations=500):
|
||||
"""
|
||||
对率回归训练函数。
|
||||
# 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)
|
||||
|
||||
Args:
|
||||
- X (numpy array): 特征数据。
|
||||
- y (numpy array): 标签数据。
|
||||
- learning_rate (float): 学习率, 默认值为0.05。
|
||||
- iterations (int): 迭代次数, 默认值为500。
|
||||
# 使用 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)
|
||||
|
||||
Returns:
|
||||
- numpy array: 训练得到的权重和偏置（合并为一个数组）。
|
||||
"""
|
||||
n_samples, n_features = X.shape
|
||||
weights = np.zeros(n_features + 1) # 增加一位用于偏置
|
||||
# 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
|
||||
|
||||
# 增加一列1表示偏置
|
||||
X = np.c_[X, np.ones((n_samples, 1))]
|
||||
input_size = X_train.shape[1]
|
||||
num_classes = len(np.unique(y))
|
||||
model = SimpleNN(input_size, num_classes)
|
||||
|
||||
# 梯度下降法
|
||||
for _ in range(iterations):
|
||||
linear_model = np.dot(X, weights)
|
||||
y_pred = sigmoid(linear_model)
|
||||
# 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)
|
||||
|
||||
# 计算梯度
|
||||
dw = (1 / n_samples) * np.dot(X.T, (y_pred - y))
|
||||
# 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)
|
||||
|
||||
# 更新权重
|
||||
weights -= learning_rate * dw
|
||||
# 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
|
||||
|
||||
return weights
|
||||
# 训练循环
|
||||
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()
|
||||
|
||||
# 对率回归预测函数
|
||||
def predict_logreg(inX, weights):
|
||||
"""
|
||||
对率回归预测函数。
|
||||
|
||||
Args:
|
||||
- inX (numpy array): 输入待分类的样本。
|
||||
- weights (numpy array): 训练得到的权重和偏置。
|
||||
|
||||
Returns:
|
||||
- int: 预测的分类标签 (0 或 1)。
|
||||
"""
|
||||
# 增加偏置项
|
||||
inX = np.append(inX, 1)
|
||||
linear_model = np.dot(inX, weights)
|
||||
y_pred = sigmoid(linear_model)
|
||||
return 1 if y_pred >= 0.5 else 0
|
||||
|
||||
|
||||
# 数据加载函数
|
||||
def load_data(filepath):
|
||||
"""
|
||||
从文件中加载数据，处理缺失值，将 ? 替换为 NaN 并填充缺失值为 0。
|
||||
|
||||
Args:
|
||||
- filepath (str): 数据文件路径。
|
||||
|
||||
Returns:
|
||||
- np.array: 特征数据。
|
||||
- np.array: 标签数据。
|
||||
"""
|
||||
# 使用 genfromtxt 读取文件并将缺失值标记为 NaN
|
||||
data = np.genfromtxt(filepath, delimiter=' ', dtype=float, missing_values='?', filling_values=0.0)
|
||||
|
||||
# 分离特征和标签
|
||||
X = data[:, :-1] # 所有列（除了最后一列）是特征
|
||||
y = data[:, -1] # 最后一列是标签
|
||||
return X, y
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# 加载训练集
|
||||
X_train, y_train = load_data("./data/horse-colic.data")
|
||||
|
||||
# 加载测试集
|
||||
X_test, y_test = load_data("./data/horse-colic.test")
|
||||
|
||||
# 训练对率回归模型
|
||||
weights = train_logreg(X_train, y_train, learning_rate=0.05, iterations=500)
|
||||
|
||||
# 测试单个数据
|
||||
y_pred = predict_logreg(X_test[11], weights)
|
||||
print(f"预测结果: {y_pred}, 实际结果: {y_test[11]}")
|
||||
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user