PyCaret自定义LSTM/MLP模型效果差，求代码问题排查

我是PyCaret新手，做时间序列预测时用最新版PyCaret对比模型。因为PyCaret没有原生LSTM支持，我按照官方文档实现了scikit-learn API风格的自定义LSTM类，后来也用同样框架做了MLP，但两者的RMSE等指标远不如PyCaret内置的简单模型。我已经调过网络参数，这是自动调参能得到的最优结果，也排除了数据问题（输入是(101872,28)，包含27个特征，因为PyCaret的bug，索引是Int格式而非时间格式），怀疑是代码有问题，以下是我在Jupyter Notebook里的实现代码：

class LSTMModelClass(nn.Module):
    def __init__(self, input_size, hidden_size, num_layers, output_size):
        super(LSTMModelClass, self).__init__()
        self.input_size = input_size
        self.hidden_size = hidden_size
        self.num_layers = num_layers
        self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True)
        self.fc = nn.Linear(hidden_size, output_size)

    def forward(self, x):
        out, _ = self.lstm(x)
        out = out[-1,:]
        out = self.fc(out)
        return out
        
class LSTM(BaseForecaster):
    def __init__(self, input_size, hidden_size, num_layers, output_size,learning_rate=0.001,epochs=10):
        super(LSTM, self).__init__()
        self.input_size = input_size
        self.hidden_size = hidden_size
        self.num_layers = num_layers
        self.output_size = output_size
        self.learning_rate = learning_rate
        self.epochs = epochs
        self.model = LSTMModelClass(self.input_size, self.hidden_size, self.num_layers, self.output_size)
        
    def fit(self, y, fh=None, X=None):
        if X is not None:
            X = X.loc[y.index]
        loss_function = nn.MSELoss()
        optimizer = optim.Adam(self.model.parameters(), lr=self.learning_rate)
        X = torch.from_numpy(np.array(X)).float()
        y = torch.from_numpy(np.array(y)).float()
        # 训练循环
        self.model = self.model.train()
        for i in range(self.epochs):
            outputs = self.model(X)
            optimizer.zero_grad()
            loss = loss_function(outputs, y)
            loss.backward()
            optimizer.step()
            print(f'Epoch [{i+1}/{epochs}], Loss: {loss.item()}')
        return self

    def predict(self, fh, X=None):
        # Make forecasts.

        # Parameters
        # ----------
        # fh : array-like
        #     The forecasters horizon with the steps ahead to to predict.
        #     Default is
        #     one-step ahead forecast, i.e. np.array([1]).
        # X : pd.DataFrame, optional (default=None)
        #     Exogenous variables are ignored

        # Returns
        # -------
        # y_pred : pandas.Series
        #     Returns series of predicted values.
        # Make forecasts.
        self.model = self.model.eval()
        X_tensor = torch.from_numpy(np.array(X)).float()
        y_tensor = self.model(X_tensor)
        if y_tensor.is_cuda:
            y_tensor = y_tensor.cpu()
        numpy_array = y_tensor.detach().numpy()
        y_pred = pd.DataFrame(numpy_array,index = X.index)
        print(y_pred)
        return y_pred
    def check_is_fitted(estimator, attributes=None, *, msg=None, all_or_any=all):
        return

代码核心问题解析 #

1. LSTM输入维度错误
   LSTM要求输入是3维张量：(batch_size, sequence_length, input_size)，但你直接把X转成2维张量(样本数, 特征数)喂给模型，相当于每个样本的序列长度为1，完全没用到LSTM的时序建模能力，效果自然不如内置模型。

2. 无批量训练逻辑
   直接把全量数据集一次性喂给模型，会导致梯度不稳定、模型难以收敛，还可能因内存占用过高出现异常。

3. 序列输出处理错误
   out = out[-1,:]的写法只取了LSTM最后一个时间步的输出，但在输入为单时间步的场景下毫无意义；若为多时间步输入，也需根据任务需求确定输出逻辑。

4. check_is_fitted方法失效
   空实现的check_is_fitted会导致PyCaret无法判断模型是否完成训练，可能引发预测阶段的未拟合问题。

5. 未使用fh预测步长参数
   PyCaret的fh参数定义了预测步长，但你的predict方法完全忽略该参数，无法处理多步预测需求。

修正后的代码示例 #

import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
import pandas as pd
from pycaret.time_series import BaseForecaster

class LSTMModelClass(nn.Module):
    def __init__(self, input_size, hidden_size, num_layers, output_size):
        super(LSTMModelClass, self).__init__()
        self.input_size = input_size
        self.hidden_size = hidden_size
        self.num_layers = num_layers
        self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True)
        self.fc = nn.Linear(hidden_size, output_size)

    def forward(self, x):
        # x shape: (batch_size, seq_len, input_size)
        out, _ = self.lstm(x)
        # 取每个样本序列最后一个时间步的输出
        out = out[:, -1, :]
        out = self.fc(out)
        return out
        
class LSTM(BaseForecaster):
    def __init__(self, input_size, hidden_size, num_layers, output_size, sequence_length=7, 
                 learning_rate=0.001, epochs=50, batch_size=32):
        super(LSTM, self).__init__()
        self.input_size = input_size
        self.hidden_size = hidden_size
        self.num_layers = num_layers
        self.output_size = output_size
        self.sequence_length = sequence_length  # 时序窗口长度
        self.learning_rate = learning_rate
        self.epochs = epochs
        self.batch_size = batch_size
        self.model = LSTMModelClass(self.input_size, self.hidden_size, self.num_layers, self.output_size)
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        self.model.to(self.device)

    # 生成时序样本
    def create_sequences(self, X, y):
        X_seq, y_seq = [], []
        for i in range(len(X) - self.sequence_length):
            X_seq.append(X[i:i+self.sequence_length])
            y_seq.append(y[i+self.sequence_length])
        return np.array(X_seq), np.array(y_seq)

    def fit(self, y, fh=None, X=None):
        if X is not None:
            X = X.loc[y.index].values
        else:
            # 无外生特征时用目标变量自身构建时序
            X = y.values.reshape(-1, 1)
        y = y.values

        # 生成时序训练样本
        X_seq, y_seq = self.create_sequences(X, y)
        
        # 转张量并移动到计算设备
        X_tensor = torch.from_numpy(X_seq).float().to(self.device)
        y_tensor = torch.from_numpy(y_seq).float().unsqueeze(1).to(self.device)

        loss_function = nn.MSELoss()
        optimizer = optim.Adam(self.model.parameters(), lr=self.learning_rate)

        self.model.train()
        num_batches = len(X_tensor) // self.batch_size

        for epoch in range(self.epochs):
            epoch_loss = 0.0
            # 打乱数据顺序
            permutation = torch.randperm(X_tensor.size()[0])
            
            for i in range(num_batches):
                indices = permutation[i*self.batch_size : (i+1)*self.batch_size]
                batch_x, batch_y = X_tensor[indices], y_tensor[indices]
                
                optimizer.zero_grad()
                outputs = self.model(batch_x)
                loss = loss_function(outputs, batch_y)
                loss.backward()
                optimizer.step()
                
                epoch_loss += loss.item()
            
            print(f'Epoch [{epoch+1}/{self.epochs}], Average Loss: {epoch_loss/num_batches:.6f}')
        
        return self

    def predict(self, fh, X=None):
        self.model.eval()
        
        if X is None:
            raise ValueError("预测需要提供包含最新时序特征的X数据")
        
        # 构建初始预测输入序列
        X_seq = X.values[-self.sequence_length:].reshape(1, self.sequence_length, self.input_size)
        X_tensor = torch.from_numpy(X_seq).float().to(self.device)
        
        predictions = []
        # 滚动生成多步预测
        for _ in range(max(fh)):
            with torch.no_grad():
                pred = self.model(X_tensor)
                predictions.append(pred.item())
                
                # 更新输入序列
                new_step = pred.detach().cpu().numpy().reshape(1, 1, self.input_size)
                X_seq = np.concatenate([X_seq[:, 1:, :], new_step], axis=1)
                X_tensor = torch.from_numpy(X_seq).float().to(self.device)
        
        # 生成预测结果索引
        last_idx = X.index[-1]
        pred_indices = [last_idx + step for step in fh]
        y_pred = pd.Series(predictions, index=pred_indices)
        
        return y_pred

    def check_is_fitted(self):
        if not hasattr(self, 'model') or self.model is None:
            raise ValueError("模型尚未完成训练")
        return True

额外优化建议 #

• 数据归一化：神经网络对数据尺度敏感，建议在PyCaret的setup阶段添加标准化/归一化预处理，或在模型内部实现数据缩放逻辑。
• 调整时序窗口长度：根据数据的周期特性设置sequence_length（如日数据设为7，月数据设为12）。
• 添加早停机制：训练时加入Early Stopping，避免过拟合同时节省训练时间。
• 优化多步预测逻辑：若需多步预测，可直接构建输出多步的模型结构，替代滚动预测提升效率。

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