Python训练营打卡Day36
仔细回顾一下神经网络到目前的内容,没跟上进度的同学补一下进度。
- 作业:对之前的信贷项目,利用神经网络训练下,尝试用到目前的知识点让代码更加规范和美观。
- 探索性作业(随意完成):尝试进入nn.Module中,查看他的方法
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, roc_auc_score
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from typing import Tuple, Dict, Any, List# 设置随机种子确保结果可复现
torch.manual_seed(42)
np.random.seed(42)class CreditRiskNN(nn.Module):"""信贷风险评估神经网络模型"""def __init__(self, input_dim: int, hidden_dims: List[int], output_dim: int = 1):"""初始化神经网络模型参数:input_dim: 输入特征维度hidden_dims: 隐藏层维度列表output_dim: 输出维度,默认为1(二分类问题)"""super(CreditRiskNN, self).__init__()# 构建网络层layers = []prev_dim = input_dim# 添加隐藏层for hidden_dim in hidden_dims:layers.append(nn.Linear(prev_dim, hidden_dim))layers.append(nn.ReLU())layers.append(nn.BatchNorm1d(hidden_dim))layers.append(nn.Dropout(0.3)) # 防止过拟合prev_dim = hidden_dim# 添加输出层layers.append(nn.Linear(prev_dim, output_dim))layers.append(nn.Sigmoid()) # 二分类问题使用Sigmoid激活函数self.network = nn.Sequential(*layers)def forward(self, x: torch.Tensor) -> torch.Tensor:"""前向传播"""return self.network(x)def load_and_preprocess_data(file_path: str) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor]:"""加载并预处理信贷数据参数:file_path: 数据文件路径返回:X_train, X_test, y_train, y_test: 训练集和测试集特征与标签"""# 加载数据(示例数据,实际使用时替换为真实数据路径)try:data = pd.read_csv(file_path)except FileNotFoundError:# 生成示例数据用于演示print("使用示例数据进行演示,实际使用时请替换为真实数据")n_samples = 1000n_features = 20# 生成随机特征X = np.random.randn(n_samples, n_features)# 生成标签(确保类别不平衡,模拟真实信贷数据)y = np.zeros(n_samples)y[:int(n_samples * 0.2)] = 1 # 20%的正样本(违约)np.random.shuffle(y)# 添加一些与标签相关的特征,使模型有学习的基础X[:, 0] += y * 2X[:, 1] += y * 1.5X[:, 2] -= y * 1.2data = pd.DataFrame(np.hstack([X, y.reshape(-1, 1)]))data.columns = [f'feature_{i}' for i in range(n_features)] + ['target']# 分离特征和标签X = data.drop('target', axis=1).valuesy = data['target'].values# 划分训练集和测试集X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42, stratify=y)# 特征标准化scaler = StandardScaler()X_train = scaler.fit_transform(X_train)X_test = scaler.transform(X_test)# 转换为PyTorch张量X_train = torch.FloatTensor(X_train)X_test = torch.FloatTensor(X_test)y_train = torch.FloatTensor(y_train).view(-1, 1)y_test = torch.FloatTensor(y_test).view(-1, 1)return X_train, X_test, y_train, y_testdef create_data_loader(X: torch.Tensor, y: torch.Tensor, batch_size: int = 64) -> DataLoader:"""创建数据加载器"""dataset = TensorDataset(X, y)return DataLoader(dataset, batch_size=batch_size, shuffle=True)def train_model(model: nn.Module, train_loader: DataLoader, test_loader: DataLoader,epochs: int = 50, lr: float = 0.001
) -> Tuple[nn.Module, Dict[str, List[float]]]:"""训练神经网络模型参数:model: 神经网络模型train_loader: 训练数据加载器test_loader: 测试数据加载器epochs: 训练轮数lr: 学习率返回:model: 训练好的模型history: 训练历史记录"""# 定义损失函数和优化器criterion = nn.BCELoss()optimizer = optim.Adam(model.parameters(), lr=lr, weight_decay=1e-5)# 记录训练历史history = {'train_loss': [],'train_acc': [],'val_loss': [],'val_acc': []}# 训练循环for epoch in range(epochs):# 训练模式model.train()train_loss = 0.0correct = 0total = 0for inputs, targets in train_loader:# 前向传播outputs = model(inputs)loss = criterion(outputs, targets)# 反向传播和优化optimizer.zero_grad()loss.backward()optimizer.step()# 记录训练损失和准确率train_loss += loss.item()predicted = (outputs > 0.5).float()total += targets.size(0)correct += (predicted == targets).sum().item()# 计算平均损失和准确率train_loss /= len(train_loader)train_acc = 100.0 * correct / total# 验证模式model.eval()val_loss = 0.0correct = 0total = 0with torch.no_grad():for inputs, targets in test_loader:outputs = model(inputs)loss = criterion(outputs, targets)val_loss += loss.item()predicted = (outputs > 0.5).float()total += targets.size(0)correct += (predicted == targets).sum().item()# 计算平均验证损失和准确率val_loss /= len(test_loader)val_acc = 100.0 * correct / total# 记录历史history['train_loss'].append(train_loss)history['train_acc'].append(train_acc)history['val_loss'].append(val_loss)history['val_acc'].append(val_acc)if (epoch + 1) % 10 == 0:print(f'Epoch [{epoch+1}/{epochs}], 'f'Train Loss: {train_loss:.4f}, Train Acc: {train_acc:.2f}%, 'f'Val Loss: {val_loss:.4f}, Val Acc: {val_acc:.2f}%')return model, historydef evaluate_model(model: nn.Module, X_test: torch.Tensor, y_test: torch.Tensor) -> None:"""评估模型性能参数:model: 训练好的模型X_test: 测试特征y_test: 测试标签"""model.eval()with torch.no_grad():y_pred_probs = model(X_test)y_pred = (y_pred_probs > 0.5).float().numpy()y_true = y_test.numpy()# 打印分类报告print("\n分类报告:")print(classification_report(y_true, y_pred))# 计算AUC分数auc_score = roc_auc_score(y_true, y_pred_probs.numpy())print(f"AUC分数: {auc_score:.4f}")# 打印混淆矩阵cm = confusion_matrix(y_true, y_pred)print("\n混淆矩阵:")print(cm)def plot_training_history(history: Dict[str, List[float]]) -> None:"""绘制训练历史曲线参数:history: 训练历史记录"""plt.figure(figsize=(12, 5))# 绘制损失曲线plt.subplot(1, 2, 1)plt.plot(history['train_loss'], label='训练损失')plt.plot(history['val_loss'], label='验证损失')plt.xlabel('Epoch')plt.ylabel('Loss')plt.title('训练和验证损失')plt.legend()plt.grid(True)# 绘制准确率曲线plt.subplot(1, 2, 2)plt.plot(history['train_acc'], label='训练准确率')plt.plot(history['val_acc'], label='验证准确率')plt.xlabel('Epoch')plt.ylabel('Accuracy (%)')plt.title('训练和验证准确率')plt.legend()plt.grid(True)plt.tight_layout()plt.show()def main():"""主函数,整合整个流程"""# 加载和预处理数据X_train, X_test, y_train, y_test = load_and_preprocess_data("credit_data.csv")# 创建数据加载器train_loader = create_data_loader(X_train, y_train)test_loader = create_data_loader(X_test, y_test, shuffle=False)# 创建模型input_dim = X_train.shape[1]model = CreditRiskNN(input_dim=input_dim, hidden_dims=[64, 32, 16])# 训练模型print("开始训练模型...")model, history = train_model(model, train_loader, test_loader, epochs=50)# 评估模型print("\n模型评估结果:")evaluate_model(model, X_test, y_test)# 绘制训练历史plot_training_history(history)# 保存模型torch.save(model.state_dict(), 'credit_risk_model.pth')print("\n模型已保存为: credit_risk_model.pth")if __name__ == "__main__":main()
nn.Module
是 PyTorch 中所有神经网络模块的基类,它提供了很多有用的方法,如:
parameters()
: 返回模型的可训练参数named_parameters()
: 返回带名称的可训练参数children()
: 返回模型的子模块modules()
: 返回模型的所有模块train()
: 将模型设置为训练模式eval()
: 将模型设置为评估模式to(device)
: 将模型移动到指定设备(CPU/GPU)state_dict()
: 返回模型的状态字典,可用于保存和加载模型
可以通过继承nn.Module
来创建自定义的神经网络,就像示例中的CreditRiskNN
类一样。
@浙大疏锦行