人工神经网络回归___PyTorch_
一、深度学习
PyTorch 这框架挺好用的，灵活又高效，用它来做 ANN 回归很方便。下面用 ANN 进行两种情形的回归预测，一个是单个因变量，另外是多个因变量。
1. 单个因变量情形
1.1 导入必要的库
import torch
import torch.nn as nn
import torch.optim as optim
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
import numpy as np
import matplotlib.pyplot as plt
# 设置中文显示
plt.rcParams["font.family"] = ["SimHei"]
plt.rcParams["axes.unicode_minus"] = False # 解决负号显示问题
plt.rcParams["font.size"] = 15
1.2 生成仿真数据
# 1. 生成仿真数据
def generate_data(n_samples=100, n_features=3, random_seed=42):
"""生成回归任务的仿真数据"""
np.random.seed(random_seed)
# 生成特征
X = np.random.randn(n_samples, n_features) # 3个特征
# 生成目标变量（非线性关系，增加任务难度）
# 真实函数：y = x1^2 + 2*x2 - 0.5*x3 + 0.8*x1*x2 + 噪声
y = (X[:, 0]**2 +
2 * X[:, 1] -
0.5 * X[:, 2] +
0.8 * X[:, 0] * X[:, 1] +
np.random.randn(n_samples) * 0.5) # 添加噪声
return X, y.reshape(-1, 1) # y转为列向量
# 生成1000个样本，3个特征
X, y = generate_data(n_samples=1000, n_features=3)
1.3 数据预处理
# 2. 数据预处理
# 划分训练集和测试集（7:3）
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42)
# 特征标准化
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)
# 转换为PyTorch张量
X_train_tensor = torch.tensor(X_train_scaled, dtype=torch.float32)
y_train_tensor = torch.tensor(y_train, dtype=torch.float32)
X_test_tensor = torch.tensor(X_test_scaled, dtype=torch.float32)
y_test_tensor = torch.tensor(y_test, dtype=torch.float32)
1.4 模型定义
# 参数设置
n_in, n_h1, n_h2, n_out, batch_size, dropout_rate = X.shape[1], 3, 3, 1, 50, 0
# 保证模型初始权重、偏置固定，后期复现
torch.manual_seed(42)
# 创建顺序模型，包含线性层与 ReLU 激活函数
model = nn.Sequential(
nn.Linear(n_in, n_h1), # 输入层到隐藏层1的线性变换
nn.ReLU(), # 隐藏层1的ReLU激活函数
nn.Dropout(p=dropout_rate), # dropout_rate=0，此处不启用随机丢弃
nn.Linear(n_h1, n_h2), # 隐藏层1到隐藏层2的线性变换
nn.ReLU(), # 隐藏层2的ReLU激活函数
nn.Linear(n_h2, n_out), # 隐藏层到输出层的线性变换
)
1.5 训练模型
# 损失函数和随机梯度下降优化器
criterion = torch.nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.01, weight_decay=0.0001) # 学习率为0.01, L2正则化系数
# 执行梯度下降算法进行模型训练
epochs = 300
train_losses, test_losses = [], []
for epoch in range(epochs):
# 训练模式
model.train()
optimizer.zero_grad() # 清零梯度
train_output = model(X_train_tensor) # 前向传播，计算预测值
loss = criterion(train_output, y_train_tensor) # 计算损失
train_losses.append(loss.item())
loss.backward() # 反向传播，计算梯度
optimizer.step() # 更新模型参数
# 测试模式：自动关闭Dropout（所有神经元参与计算）
with torch.no_grad():
model.eval()
test_outputs = model(X_test_tensor)
test_loss = criterion(test_outputs, y_test_tensor)
test_losses.append(test_loss.item())
if (epoch + 1) % 10 == 0:
print(f'Epoch [{epoch+1}/{epochs}], 训练损失: {loss.item():.4f}, 测试损失: {test_loss.item():.4f}')
1.6 模型评估
# 6. 模型评估
model.eval()
with torch.no_grad():
y_pred_train = model(X_train_tensor).numpy()
y_pred_test = model(X_test_tensor).numpy()
1.7 可视化结果
# 7. 可视化结果
plt.figure(figsize=(18, 15))
# 绘制损失曲线
plt.subplot(3, 1, 1)
plt.plot(range(1, len(train_losses)+1), train_losses, label='训练损失')
plt.plot(range(1, len(test_losses)+1), test_losses, label='测试损失')
plt.xlabel('Epoch')
plt.ylabel('MSE损失')
plt.title('训练与测试损失曲线')
plt.xticks()
plt.grid()
plt.legend()
# 绘制预测值与真实值对比
plt.subplot(3, 1, 2)
plt.scatter(y_test, y_pred_test, color='blue', label='测试集')
plt.scatter(y_train, y_pred_train, color='orange', alpha=0.5, label='训练集')
plt.plot([y.min(), y.max()], [y.min(), y.max()], 'r--') # 理想线
plt.xlabel('真实值')
plt.ylabel('预测值')
plt.title('预测值 vs 真实值')
plt.legend()
plt.grid()
plt.tight_layout()
plt.show()
图 1：单个因变量情形——训练/测试损失曲线与预测值 vs 真实值散点图
1.8 计算 R² 分数
# 计算R²分数（决定系数）
def r2_score(y_true, y_pred):
ss_total = np.sum((y_true - np.mean(y_true)) ** 2)
ss_residual = np.sum((y_true - y_pred) ** 2)
return 1 - (ss_residual / ss_total)
train_r2 = r2_score(y_train, y_pred_train)
test_r2 = r2_score(y_test, y_pred_test)
print(f'\n训练集R²分数: {train_r2:.4f}')
print(f'测试集R²分数: {test_r2:.4f}')
2. 多个因变量情形
这里以两个因变量为例。
2.1 生成仿真数据
# 1. 生成仿真数据
def generate_data(n_samples=1000, n_features=4, random_seed=42):
"""生成回归任务的仿真数据，包含4个特征和2个因变量"""
np.random.seed(random_seed) # 固定随机种子，保证可复现性
# 生成4个特征变量
X = np.random.randn(n_samples, n_features) # 形状为(n_samples, 4)
# 生成第一个因变量 y1（非线性关系）
# y1 = x1² + 2*x2 - 0.5*x3 + 0.8*x1*x2 + 噪声
y1 = (X[:, 0]**2 +
2 * X[:, 1] -
0.5 * X[:, 2] +
0.8 * X[:, 0] * X[:, 1] +
np.random.randn(n_samples) * 0.5) # 添加噪声
# 生成第二个因变量 y2（不同的非线性关系）
# y2 = 0.3*x2*x3 + x3*x4 - x1 + sqrt(|x4|) + 噪声
y2 = (0.3 * X[:, 1] * X[:, 2] +
X[:, 2] * X[:, 3] -
X[:, 0] +
np.sqrt(np.abs(X[:, 3])) +
np.random.randn(n_samples) * 0.2) # 添加噪声
# 组合两个因变量，形状为(n_samples, 2)
y = np.column_stack((y1, y2))
return X, y
X, y = generate_data(n_samples=10000)
2.2 数据预处理
# 2. 数据预处理
# 划分训练集和测试集（8:2）
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=1996)
# 特征标准化
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)
# 转换为PyTorch张量
X_train_tensor = torch.tensor(X_train_scaled, dtype=torch.float32)
y_train_tensor = torch.tensor(y_train, dtype=torch.float32)
X_test_tensor = torch.tensor(X_test_scaled, dtype=torch.float32)
y_test_tensor = torch.tensor(y_test, dtype=torch.float32)
2.3 模型定义
# 参数设置
n_in, n_h1, n_h2, n_out, batch_size, dropout_rate = X.shape[1], 10, 10, 2, 256, 0.05
# 保证模型初始权重、偏置固定，后期复现
torch.manual_seed(42)
# 创建顺序模型，包含线性层与 Sigmoid/ReLU 激活函数
model = nn.Sequential(
nn.Linear(n_in, n_h1), # 输入层到隐藏层1的线性变换
nn.Sigmoid(), # 隐藏层1的Sigmoid激活函数
nn.Dropout(p=dropout_rate), # 训练时随机丢弃5%的神经元
nn.Linear(n_h1, n_h2), # 隐藏层1到隐藏层2的线性变换
nn.ReLU(), # 隐藏层2的ReLU激活函数
nn.Dropout(p=dropout_rate), # 训练时随机丢弃5%的神经元
nn.Linear(n_h2, n_out), # 隐藏层到输出层的线性变换
)
2.4 训练模型
# 损失函数和随机梯度下降优化器
criterion = torch.nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.01, weight_decay=0) # 学习率为0.01
# 计算R²分数（决定系数）
def r2_score(y_true, y_pred):
ss_total = np.sum((y_true - np.mean(y_true)) ** 2)
ss_residual = np.sum((y_true - y_pred) ** 2)
return 1 - (ss_residual / ss_total)
# 执行梯度下降算法进行模型训练
epochs = 2000
train_losses, test_losses = [], []
for epoch in range(epochs):
# 训练模式
model.train()
optimizer.zero_grad() # 清零梯度
train_output = model(X_train_tensor) # 前向传播，计算预测值
loss = criterion(train_output, y_train_tensor) # 计算损失
train_losses.append(loss.item())
loss.backward() # 反向传播，计算梯度
optimizer.step() # 更新模型参数
# 测试模式：自动关闭Dropout（所有神经元参与计算）
with torch.no_grad():
model.eval()
test_outputs = model(X_test_tensor)
test_loss = criterion(test_outputs, y_test_tensor)
test_losses.append(test_loss.item())
test_y1_r2 = r2_score(y_test_tensor[:,0].numpy(), test_outputs[:,0].numpy())
test_y2_r2 = r2_score(y_test_tensor[:,1].numpy(), test_outputs[:,1].numpy())
if (epoch + 1) % 100 == 0:
print(f'Epoch [{epoch+1}/{epochs}], 训练损失: {loss.item():.4f}, 测试损失: {test_loss.item():.4f},测试集y1/y2 R2为({test_y1_r2:.4f},{test_y2_r2:.4f})')
图 2：多个因变量情形——训练过程日志（每 100 轮输出损失与测试集 y1/y2 的 R²）
2.5 模型评估
# 6. 模型评估
model.eval()
with torch.no_grad():
y_pred_train = model(X_train_tensor).numpy()
y_pred_test = model(X_test_tensor).numpy()
2.6 可视化结果
# 7. 可视化结果
plt.figure(figsize=(18, 15))
# 绘制损失曲线
plt.subplot(3, 1, 1)
plt.plot(range(1, len(train_losses)+1), train_losses, label='训练损失')
plt.plot(range(1, len(test_losses)+1), test_losses, label='测试损失')
plt.xlabel('Epoch')
plt.ylabel('MSE损失')
plt.title('训练与测试损失曲线')
plt.xticks()
plt.grid()
plt.legend()
# 绘制y1预测值与真实值对比
plt.subplot(3, 1, 2)
plt.plot(range(1, len(y_test[:,0])+1),y_test[:,0],label = "y1实际值")
plt.plot(range(1, len(y_pred_test[:,0])+1),y_pred_test[:,0],label = "y1预测值")
plt.xlabel('预测样本')
plt.ylabel('预测值')
plt.title('y1预测值 vs y1真实值')
plt.legend()
plt.grid()
plt.tight_layout()
# 绘制y2预测值与真实值对比
plt.subplot(3, 1, 3)
plt.plot(range(1, len(y_test[:,1])+1),y_test[:,1],label = "y2实际值")
plt.plot(range(1, len(y_pred_test[:,1])+1),y_pred_test[:,1],label = "y2预测值")
plt.xlabel('预测样本')
plt.ylabel('预测值')
plt.title('y2预测值 vs y2真实值')
plt.legend()
plt.grid()
plt.tight_layout()
图 3：多个因变量情形——损失曲线与 y1、y2 实际值/预测值对比
2.7 计算 R² 分数
# y1的R²分数
train_r2 = r2_score(y_train[:,0], y_pred_train[:,0])
test_r2 = r2_score(y_test[:,0], y_pred_test[:,0])
print(f'\n训练集R²分数: {train_r2:.4f}')
print(f'测试集R²分数: {test_r2:.4f}')
# y2的R²分数
train_r2 = r2_score(y_train[:,1], y_pred_train[:,1])
test_r2 = r2_score(y_test[:,1], y_pred_test[:,1])
print(f'\n训练集R²分数: {train_r2:.4f}')
print(f'测试集R²分数: {test_r2:.4f}')