一、文章介绍

利用PyTorch实现最简单的一个LSTM-GRU模型；适合完全的小白

二、模型融合

目的

长期依赖 vs 短期波动

• ​LSTM优势​
  通过细胞状态（Cell State）和三个门控机制（输入/遗忘/输出门），擅长捕捉超过10个时间步的长期依赖关系。
  生鲜场景应用：准确建模价格季节性（如节假日周期）、库存衰减趋势（如蔬菜保质期规律）。

• ​GRU优势​
  合并门控结构（更新门和重置门），参数减少33%，对短期波动（1-3天）响应更灵敏。
  生鲜场景应用：快速适应突发天气变化导致的日销量突变、竞争对手临时调价。

思路

最简单的把LSTM的输出当作GRN的输入,最后全连接层实现维度变化

• LSTM层处理原始输入序列，捕获长期趋势
• GRU层对LSTM输出进行二次处理，捕捉短期波动
• 全连接层输出最终预测结果

三、代码实现

import torch
import torch.nn as nn
import numpy as np
import matplotlib.pyplot as plt
from sklearn.preprocessing import MinMaxScaler


# 配置参数
class Config:
    seq_length = 7  # 输入序列长度(过去7天)
    pred_length = 1  # 预测未来1天
    hidden_size = 64  # 隐藏层维度
    num_layers = 2  # LSTM层数
    batch_size = 16
    learning_rate = 0.001
    epochs = 200


# 生成模拟数据（带趋势和季节性的时间序列）
def generate_synthetic_data(samples=500):
    t = np.arange(samples)
    # 基础趋势：线性趋势+周期性波动
    trend = 0.1 * t
    seasonal = 2 * np.sin(2 * np.pi * t / 30)  # 30天周期
    noise = np.random.normal(0, 0.5, samples)
    data = trend + seasonal + noise
    return data.reshape(-1, 1)


# 数据预处理
def create_dataset(data, look_back=7):
    scaler = MinMaxScaler(feature_range=(-1, 1))
    data_norm = scaler.fit_transform(data)

    X, y = [], []
    for i in range(len(data_norm) - look_back - Config.pred_length):
        X.append(data_norm[i:(i + look_back), 0])
        y.append(data_norm[i + look_back:i + look_back + Config.pred_length, 0])
    return torch.FloatTensor(np.array(X)), torch.FloatTensor(np.array(y)), scaler


# 定义LSTM-GRU混合模型
class HybridModel(nn.Module):
    def __init__(self, input_size=1):
        super().__init__()
        self.lstm = nn.LSTM(
            input_size=input_size,
            hidden_size=Config.hidden_size,
            num_layers=Config.num_layers,
            batch_first=True,
            dropout=0.2
        )
        self.gru = nn.GRU(
            input_size=Config.hidden_size,
            hidden_size=Config.hidden_size // 2,
            batch_first=True
        )
        self.fc = nn.Sequential(
            nn.Linear(Config.hidden_size // 2, 32),
            nn.ReLU(),
            nn.Dropout(0.1),
            nn.Linear(32, Config.pred_length)
        )

    def forward(self, x):
        # 输入x形状: (batch_size, seq_len, input_size)
        lstm_out, _ = self.lstm(x)  # (batch, seq_len, hidden_size)

        # 取LSTM最后一层输出作为GRU输入
        gru_out, _ = self.gru(lstm_out)  # (batch, seq_len, hidden_size//2)

        # 取最后一个时间步输出
        last_out = gru_out[:, -1, :]  # (batch, hidden_size//2)

        return self.fc(last_out)  # (batch, pred_length)


# 训练流程
def train_model():
    # 准备数据
    data = generate_synthetic_data()
    X, y, scaler = create_dataset(data, Config.seq_length)
    dataset = torch.utils.data.TensorDataset(X.unsqueeze(-1), y)
    train_loader = torch.utils.data.DataLoader(dataset, batch_size=Config.batch_size, shuffle=True)

    # 初始化模型
    model = HybridModel()
    criterion = nn.MSELoss()
    optimizer = torch.optim.Adam(model.parameters(), lr=Config.learning_rate)

    # 训练循环
    losses = []
    for epoch in range(Config.epochs):
        model.train()
        epoch_loss = 0
        for batch_x, batch_y in train_loader:
            optimizer.zero_grad()
            outputs = model(batch_x)
            loss = criterion(outputs, batch_y)
            loss.backward()
            nn.utils.clip_grad_norm_(model.parameters(), 0.5)  # 梯度裁剪
            optimizer.step()
            epoch_loss += loss.item()

        avg_loss = epoch_loss / len(train_loader)
        losses.append(avg_loss)
        if (epoch + 1) % 50 == 0:
            print(f'Epoch [{epoch + 1}/{Config.epochs}], Loss: {avg_loss:.4f}')

    # 绘制损失曲线
    plt.plot(losses)
    plt.title('Training Loss')
    plt.show()

    return model, scaler


# 预测与可视化
def predict_and_plot(model, scaler):
    # 生成测试数据
    test_data = generate_synthetic_data(samples=100)
    test_X, test_y, _ = create_dataset(test_data, Config.seq_length)
    test_X = test_X.unsqueeze(-1)  # (samples, seq_len, 1)

    # 预测
    model.eval()
    with torch.no_grad():
        predictions = model(test_X).numpy()

    # 反归一化
    pred_vals = scaler.inverse_transform(predictions)
    true_vals = scaler.inverse_transform(test_y.numpy())

    # 可视化最后50个样本
    plt.figure(figsize=(12, 6))
    plt.plot(true_vals[-50:], label='True')
    plt.plot(pred_vals[-50:], '--', label='Predicted')
    plt.title('LSTM-GRU Time Series Prediction')
    plt.legend()
    plt.show()


# 运行主程序
if __name__ == "__main__":
    trained_model, data_scaler = train_model()
    predict_and_plot(trained_model, data_scaler)

四、优化思路

思路一：模型融合优化;

思路二：添加注意力机制