PyTorch自定义函数结合FFN训练时优化失效问题排查

• 已将实际遇到的问题转化为可复现的具象示例，核心目标是求解目标计算公式中的beta参数，该参数后续将用于其他业务场景
• 涉及的计算公式示意图：，公式中x为输入变量
• 已知在该简单示例场景下，可先估计y值再通过OLS方法计算beta，但该方法无法适配实际业务场景，因此未采用
• 实现方案：基于PyTorch追踪梯度，使用SGD优化器更新参数，采用简单FFN（前馈神经网络）作为函数f的近似结构
• 故障表现：训练过程中自定义函数环节优化失效，模型拟合效果未达预期，需要定位故障原因并获取可行的优化方案
• 完整复现代码如下：

import pandas as pd
import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
import copy
import torch.nn.functional as F
import matplotlib.pyplot as plt


# 构造数据集
df = pd.date_range('2020-12-01', '2022-05-31', freq='H')
df = pd.DataFrame(df[0:len(df)-1], columns={'DateTime'})
np.random.seed(1)
A = 10

X = pd.DataFrame(np.random.uniform(0, 1, len(df)), columns={'X1'})
X['X2'] = np.random.normal(0, 1, len(df))
X['X3'] = np.random.normal(0, 1, len(df))


Y = pd.DataFrame()


for i in range(0, len(df)):
    Y[str(i)] = X['X1'][i] ** 2 + X['X1'][i] * X['X2'][i] * range(0, A) + X['X1'][i] **2 * X['X2'][i] * np.array(range(0, A))**2



Y = Y.T.reset_index(drop=True)
df = pd.concat([df, Y], axis=1)


# 数据集划分配置
valid_start = '2021-07-01'
test_start = '2022-01-01'

Out = 2

X_array = X.copy()

Y_train = df[df['DateTime'].dt.date.astype(str) < valid_start].drop(columns='DateTime')
Y_valid = df[df['DateTime'].dt.date.astype(str) >= valid_start].drop(columns='DateTime')
Y_test = df[df['DateTime'].dt.date.astype(str) >= valid_start].drop(columns='DateTime')
Y_test_date = pd.DataFrame(df[df['DateTime'].dt.date.astype(str) >= valid_start]['DateTime'], columns={'DateTime'})


X_train = X_array[X_array.index.isin(Y_train.index)]
X_valid = X_array[X_array.index.isin(Y_valid.index)]
X_test = X_array[X_array.index.isin(Y_test.index)]

# 设备配置
torch.cuda.is_available()
torch.cuda.device_count()
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
print(device)

# 转换为张量
Y_train_tensor = torch.tensor(np.float32(Y_train)).to(device)
X_train_tensor = torch.tensor(np.float32(X_train)).to(device)

Y_valid_tensor = torch.tensor(np.float32(Y_valid)).to(device)
X_valid_tensor = torch.tensor(np.float32(X_valid)).to(device)

Y_test_tensor = torch.tensor(np.float32(Y_test)).to(device)
X_test_tensor = torch.tensor(np.float32(X_test)).to(device)

## 模型定义
class Model(nn.Module):

    def __init__(self):
        super().__init__()
        self.fc1 = nn.Linear(X_train_tensor.shape[1], 4)
        self.out = nn.Linear(4, int(Out) + 1)

    def forward(self, x):
        x = F.relu(self.fc1(x))
        x = self.out(x)

        return x

net = Model().to(device)
criterion = nn.MSELoss()
optimizer = optim.SGD(net.parameters(), lr=0.001)
Range_tensor = torch.tensor(np.float32(range(0, A))).to(device)

# 早停配置
Network_weight = [copy.deepcopy(net.out.weight)]

Epoch = 100
patience = 10
Valloss_Array = []
trigger_times = 0
last_loss = 1000000000000000
Valloss_Array.append(last_loss)


for epoch in range(Epoch):  # 训练循环
    x = X_train_tensor
    y = Y_train_tensor

    # 梯度清零
    optimizer.zero_grad()

    # 前向传播 + 反向传播 + 参数更新
    outputs = net(x)

    y_hat = []
    for e in range(0, len(y)):
        y_hat_temp = outputs[e, 0] + outputs[e, 1] * Range_tensor + outputs[e, 2] * torch.pow(Range_tensor, 2)

        y_hat.append(y_hat_temp)


    y_hat = torch.stack(y_hat)

    loss = criterion(y_hat, y)

    loss.backward()
    optimizer.step()
    print('Running training loss is; ' + str(loss.item()))
    Network_weight.append([copy.deepcopy(net.fc1.weight), copy.deepcopy(net.out.weight)])

    ################## 验证环节 ##################
    x = X_valid_tensor
    y = Y_valid_tensor

    outputs = net(x)

    y_hat = []
    for e in range(0, len(y)):

        y_hat_temp = outputs[e, 0] + outputs[e, 1] * Range_tensor + outputs[e, 2] * torch.pow(Range_tensor, 2)

        y_hat.append(y_hat_temp)

    y_hat = torch.stack(y_hat)


    Valloss = criterion(y_hat, y)


    if Valloss.item() >= min(Valloss_Array):
        trigger_times += 1

        if trigger_times >= patience:
            print('Early stopping!')
            break

    else:
        trigger_times = 0
        Best = [copy.deepcopy(net.fc1.weight), copy.deepcopy(net.out.weight)]

    Valloss_Array.append(Valloss.item())


# 加载最优权重
net.fc1.weight = Best[0]
net.out.weight = Best[1]


outputs = net(X_test_tensor)

y_hat = []
for e in range(0, len(Y_test_tensor)):

    y_hat_temp = outputs[e, 0] + outputs[e, 1] * Range_tensor + outputs[e, 2] * torch.pow(Range_tensor, 2)

    y_hat.append(y_hat_temp)

y_hat = torch.stack(y_hat)

BestModelLoss = criterion(y_hat, Y_test_tensor)


Y_Prediction = pd.DataFrame(y_hat.cpu().detach().numpy(), index=Y_test.index, columns={0, 1, 2, 3, 4, 5, 6, 7, 8, 9})

Y_test_date['SE'] = ((Y_Prediction - Y_test) ** 2).sum(axis=1)

Y_Prediction['DateTime'] = Y_test_date['DateTime']
Y_test['DateTime'] = Y_test_date['DateTime']

### 结果可视化
MinDate = Y_test_date.loc[Y_test_date['SE'] == Y_test_date['SE'].min(), 'DateTime']
YMinForecast = np.array(Y_Prediction[(Y_Prediction['DateTime'] == MinDate.values[0])].drop(columns='DateTime'))
YMin = np.array(Y_test[(Y_test['DateTime'] == MinDate.values[0])].drop(columns='DateTime'))

MaxDate = Y_test_date.loc[Y_test_date['SE'] == Y_test_date['SE'].max(), 'DateTime']
YMaxForecast = np.array(Y_Prediction[(Y_Prediction['DateTime'] == MaxDate.values[0])].drop(columns='DateTime'))
YMax = np.array(Y_test[(Y_test['DateTime'] == MaxDate.values[0])].drop(columns='DateTime'))

MedianValue = Y_test_date['SE'].iloc[(Y_test_date['SE']-Y_test_date['SE'].median()).abs().argsort()[:2]].values[0]

MedDate = Y_test_date.loc[Y_test_date['SE'] == MedianValue, 'DateTime']
YMedForecast = np.array(Y_Prediction[(Y_Prediction['DateTime'] == MedDate.values[0])].drop(columns='DateTime'))
YMed = np.array(Y_test[(Y_test['DateTime'] == MedDate.values[0])].drop(columns='DateTime'))


# 绘图
plt.plot(range(0, y_hat.shape[1]), YMinForecast.reshape(-1), label='Forecast')
plt.plot(range(0, y_hat.shape[1]), YMin.reshape(-1), label='Actual')
plt.grid()
plt.title("Best performance")
plt.legend()
plt.show()

plt.plot(range(0, y_hat.shape[1]), YMaxForecast.reshape(-1), label='Forecast')
plt.plot(range(0, y_hat.shape[1]), YMax.reshape(-1), label='Actual')
plt.grid()
plt.title("Poor performance")
plt.legend()
plt.show()

plt.plot(range(0, y_hat.shape[1]), YMedForecast.reshape(-1), label='Forecast')
plt.plot(range(0, y_hat.shape[1]), YMed.reshape(-1), label='Actual')
plt.grid()
plt.title("Median performance")
plt.legend()
plt.show()

内容的提问来源于stack exchange，提问作者Chri571p