时间序列预测实战(十六)PyTorch实现GRU-FCN模型长期预测并可视化结果

往期回顾：时间序列预测专栏——包含上百种时间序列模型带你从入门到精通时间序列预测

一、本文介绍

本文讲解的实战内容是GRU-FCN(门控循环单元-全卷积网络)，这是一种结合了GRU（用于处理时间序列数据）和FCN（全卷积网络）的深度学习模型，这种融合模型在处理时间序列预测的时候是十分经典且效果很好的组合方式。它的优势在于能够同时捕捉时间序列数据中的时间动态和空间特征，从而提高模型的准确度和效率，本文给大家讲解的就是这个模型，本文的实战内容通过时间序列领域最经典的数据集——电力负荷数据集为例，带大家深入的了解GRU和FCN的基本原理和融合思想。

预测类型->单元预测、长期预测，(多元预测的功能我目前没有做这个模型大家有兴趣可以自己尝试有不会的可以在评论区问我)

代码地址->文末提供复制粘贴即可运行的代码块

一、本文介绍

二、框架原理介绍

1.GRU的基本原理

1.1GRU的基本框架

2.FCN的基本原理

3.模型融合思想

三、数据集介绍

四、项目的全部代码

五、模型代码的详细讲解

六、模型的训练和预测

6.1模型的训练

6.2模型的评估

6.2.1结果展示

6.2.2结果分析

全文总结

二、框架原理介绍

1.GRU的基本原理

GRU（门控循环单元）是一种循环神经网络（RNN）的变体，主要用于处理序列数据，它的基本原理可以概括如下：

门控机制：GRU的核心是门控机制，包括更新门（update gate）和重置门（reset gate）。这些门控制着信息的流动，即决定哪些信息应该被保留，哪些应该被遗忘。

更新门：更新门帮助模型决定过去的信息有多少需要保留到当前状态。它是通过当前输入和前一个隐状态计算得出的，用于调节隐状态的更新程度。

重置门：重置门决定了多少过去的信息需要被忘记。它同样依赖于当前输入和前一个隐状态的信息。当重置门接近0时，模型会“忘记”过去的隐状态，只依赖于当前输入。

当前隐状态的计算：利用更新门和重置门的输出，结合前一隐状态和当前输入，GRU计算出当前的隐状态。这个隐状态包含了序列到目前为止的重要信息。

输出：GRU的最终输出通常是在序列的每个时间步上产生的，或者在序列的最后一个时间步产生，取决于具体的应用场景。

总结：GRU相较于传统的RNN，其优势在于能够更有效地处理长序列数据，减轻了梯度消失的问题。同时，它通常比LSTM（长短期记忆网络）更简单，因为它有更少的参数。

1.1GRU的基本框架

上面的图片为一个GRU的基本结构图，解释如下->

更新门（z） 在决定是否用新的隐藏状态更新当前隐藏状态时扮演重要角色。
重置门（r） 决定是否忽略之前的隐藏状态。

这些部分是GRU的核心组成，它们共同决定了网络如何在序列数据中传递和更新信息，这对于时间序列分析至关重要。

2.FCN的基本原理

全卷积网络（FCN, Fully Convolutional Network）FCN 完全由卷积层构成CN 通过在时间序列数据上应用一系列卷积和池化操作，有效地提取时间序列中的局部特征和模式。这对于理解数据中的复杂模式和趋势非常有用。卷积层可以捕捉时间序列数据中的局部依赖性和时间动态，不同的卷积核可以学习到数据的不同方面，例如趋势、周期性和异常(这一点在我定义的模型中也有体现)。

3.模型融合思想

将门控循环单元（GRU）和全卷积网络（FCN）结合起来的思想是为了利用这两种神经网络结构的优势，以更有效地处理时间序列数据->

特征提取与时间依赖性的结合：FCN 强于从时间序列数据中提取局部特征和模式，而 GRU 擅长捕捉时间序列的长期依赖性。将这两者结合，可以同时利用这两种能力。

FCN 部分的作用：在组合模型中，FCN 通常作为前端，用于从原始时间序列数据中提取重要的特征。FCN 的卷积层可以有效地识别出时间序列中的关键局部模式和结构特点。

GRU 部分的作用：GRU 作为模型的后端，用于处理 FCN 提取的特征。由于 GRU 的门控机制，它能够有效地捕捉和利用时间序列数据中的长期依赖关系。

最后将结果进行组合拼接从而融合了两种模型的各自优势和特点

三、数据集介绍

我们本文用到的数据集是官方的ETTh1.csv ，该数据集是一个用于时间序列预测的电力负荷数据集，它是 ETTh 数据集系列中的一个。ETTh 数据集系列通常用于测试和评估时间序列预测模型。以下是 ETTh1.csv 数据集的一些内容：

数据内容：该数据集通常包含有关电力系统的多种变量，如电力负荷、价格、天气情况等。这些变量可以用于预测未来的电力需求或价格。

时间范围和分辨率：数据通常按小时或天记录，涵盖了数月或数年的时间跨度。具体的时间范围和分辨率可能会根据数据集的版本而异。

以下是该数据集的部分截图->

四、项目的全部代码

"""
作者：CSDN Snu77 时间序列预测专栏
"""
import time
import numpy as np
import pandas as pd
import torch.nn as nn
from matplotlib import pyplot as plt
from sklearn.preprocessing import MinMaxScaler
from torch.utils.data import DataLoader
import torch
from torch.utils.data import Dataset
import torch.nn.functional as F# 随机数种子
np.random.seed(0)class TimeSeriesDataset(Dataset):def __init__(self, sequences):self.sequences = sequencesdef __len__(self):return len(self.sequences)def __getitem__(self, index):sequence, label = self.sequences[index]return torch.Tensor(sequence), torch.Tensor(label)def calculate_mae(y_true, y_pred):# 平均绝对误差mae = np.mean(np.abs(y_true - y_pred))return mae"""
数据定义部分
"""
true_data = pd.read_csv('ETTh1-Test.csv')  # 填你自己的数据地址,自动选取你最后一列数据为特征列target = 'OT'  # 添加你想要预测的特征列
test_size = 0.15  # 训练集和测试集的尺寸划分
train_size = 0.85  # 训练集和测试集的尺寸划分
pre_len = 4  # 预测未来数据的长度
train_window = 32  # 观测窗口# 这里加一些数据的预处理, 最后需要的格式是pd.series
true_data = np.array(true_data[target])# 定义标准化优化器
scaler_train = MinMaxScaler(feature_range=(0, 1))
scaler_test = MinMaxScaler(feature_range=(0, 1))# 训练集和测试集划分
train_data = true_data[:int(train_size * len(true_data))]
test_data = true_data[-int(test_size * len(true_data)):]
print("训练集尺寸:", len(train_data))
print("测试集尺寸:", len(test_data))# 进行标准化处理
train_data_normalized = scaler_train.fit_transform(train_data.reshape(-1, 1))
test_data_normalized = scaler_test.fit_transform(test_data.reshape(-1, 1))# 转化为深度学习模型需要的类型Tensor
train_data_normalized = torch.FloatTensor(train_data_normalized)
test_data_normalized = torch.FloatTensor(test_data_normalized)def create_inout_sequences(input_data, tw, pre_len):# 创建时间序列数据专用的数据分割器inout_seq = []L = len(input_data)for i in range(L - tw):train_seq = input_data[i:i + tw]if (i + tw + 4) > len(input_data):breaktrain_label = input_data[i + tw:i + tw + pre_len]inout_seq.append((train_seq, train_label))return inout_seq# 定义训练器的的输入
train_inout_seq = create_inout_sequences(train_data_normalized, train_window, pre_len)
test_inout_seq = create_inout_sequences(test_data_normalized, train_window, pre_len)# 创建数据集
train_dataset = TimeSeriesDataset(train_inout_seq)
test_dataset = TimeSeriesDataset(test_inout_seq)# 创建 DataLoader
batch_size = 64  # 你可以根据需要调整批量大小
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True, drop_last=True)
test_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False, drop_last=True)class GRUConvNet(nn.Module):def __init__(self, input_size, hidden_size, num_layers, pre_len, output_size=1):super(GRUConvNet, self).__init__()# 可以外部定义的参数self.pre_len = pre_lenself.input_size = input_sizeself.hidden_size = hidden_sizeself.num_layers = num_layersself.output_size = output_size# GRU层self.gru = nn.GRU(input_size=self.input_size, hidden_size=hidden_size, num_layers=self.num_layers,batch_first=True)self.gru_dropout = nn.Dropout(0.05)# 卷积层self.conv1 = nn.Conv1d(in_channels=self.input_size, out_channels=hidden_size, kernel_size=9, padding='same')self.bn1 = nn.BatchNorm1d(hidden_size)self.conv2 = nn.Conv1d(in_channels=hidden_size, out_channels=hidden_size * 2, kernel_size=6, padding='same')self.bn2 = nn.BatchNorm1d(hidden_size * 2)self.conv3 = nn.Conv1d(in_channels=hidden_size * 2, out_channels=hidden_size, kernel_size=3, padding='same')self.bn3 = nn.BatchNorm1d(hidden_size)self.global_avg_pool = nn.AdaptiveAvgPool1d(1)# 输出层self.fc1 = nn.Linear(self.hidden_size, self.output_size)self.fc1 = nn.Linear(self.hidden_size, self.output_size)def forward(self, x):# GRU层x_, _ = self.gru(x)x_ = self.gru_dropout(x_[:, :, :])  # 使用序列的最后一个输出# 卷积层y = x.permute(0, 2, 1)y = F.relu(self.bn1(self.conv1(y)))y = F.relu(self.bn2(self.conv2(y)))y = F.relu(self.bn3(self.conv3(y)))y = self.global_avg_pool(y)y = y.permute(0, 2, 1)# 合并GRU和卷积层的输出combined = torch.cat([x_, y], dim=1)# 输出层output = self.fc1(combined)return output[:, -self.pre_len:, :]lstm_model = GRUConvNet(input_size=1, output_size=1, num_layers=2, hidden_size=32, pre_len=pre_len)
loss_function = nn.MSELoss()
optimizer = torch.optim.Adam(lstm_model.parameters(), lr=0.005)
epochs = 20
Train = True  # 训练还是预测if Train:losss = []lstm_model.train()  # 训练模式for i in range(epochs):start_time = time.time()  # 计算起始时间for seq, labels in train_loader:lstm_model.train()optimizer.zero_grad()y_pred = lstm_model(seq)single_loss = loss_function(y_pred, labels)single_loss.backward()optimizer.step()print(f'epoch: {i:3} loss: {single_loss.item():10.8f}')losss.append(single_loss.detach().numpy())torch.save(lstm_model.state_dict(), 'save_model.pth')print(f"模型已保存,用时:{(time.time() - start_time) / 60:.4f} min")else:# 加载模型进行预测lstm_model.load_state_dict(torch.load('save_model.pth'))lstm_model.eval()  # 评估模式results = []reals = []losss = []for seq, labels in test_loader:pred = lstm_model(seq)mae = calculate_mae(pred.detach().numpy(), np.array(labels))  # MAE误差计算绝对值(预测值  - 真实值)losss.append(mae)for j in range(batch_size):for i in range(pre_len):reals.append(labels[j][i][0].detach().numpy())results.append(pred[j][i][0].detach().numpy())reals = scaler_test.inverse_transform(np.array(reals).reshape(1, -1))[0]results = scaler_test.inverse_transform(np.array(results).reshape(1, -1))[0]print("模型预测结果：", results)print("预测误差MAE:", losss)plt.figure()plt.style.use('ggplot')# 创建折线图plt.plot(reals, label='real', color='blue')  # 实际值plt.plot(results, label='forecast', color='red', linestyle='--')  # 预测值# 增强视觉效果plt.grid(True)plt.title('real vs forecast')plt.xlabel('time')plt.ylabel('value')plt.legend()plt.savefig('test——results.png')

五、模型代码的详细讲解

下面我来仔细的讲解其中的每一行代码的具体含义，我们首先来看程序的入口再看调用方法内的代码->这段代码就是我们的模型内部，

整个代码的流程我会从模型的入口参数定义开始进行讲解，然后顺序讲解在直到模型的结束。

true_data = pd.read_csv('ETTh1.csv')  # 填你自己的数据地址,自动选取你最后一列数据为特征列

这一步就是读取你的数据了~不给大家讲了主要是csv的格式数据。

target = 'OT'  # 添加你想要预测的特征列
test_size = 0.15  # 训练集和测试集的尺寸划分
train_size = 0.85  # 训练集和测试集的尺寸划分
pre_len = 4  # 预测未来数据的长度
train_window = 32  # 观测窗口

这一步就是参数定义的部分，讲解我已经再代码里标注了出来，需要说说的就是，pre_len和train_window这两个参数，

其中pre_len就是你预测未来数据的长度，假设你有一百条数据你想知道未来多少条数据的信息就填多少。

train_window是数据的观测窗口，就是你利用多少条数据去预测你定义的pre_len长度。

# 这里加一些数据的预处理, 最后需要的格式是pd.series
true_data = np.array(true_data[target])

这是提取出特征列，根据前面你定义的target。

# 定义标准化优化器
scaler_train = MinMaxScaler(feature_range=(0, 1))
scaler_test = MinMaxScaler(feature_range=(0, 1))# 训练集和测试集划分
train_data = true_data[:int(train_size * len(true_data))]
test_data = true_data[-int(test_size * len(true_data)):]
print("训练集尺寸:", len(train_data))
print("测试集尺寸:", len(test_data))# 进行标准化处理
train_data_normalized = scaler_train.fit_transform(train_data.reshape(-1, 1))
test_data_normalized = scaler_test.fit_transform(test_data.reshape(-1, 1))# 转化为深度学习模型需要的类型Tensor
train_data_normalized = torch.FloatTensor(train_data_normalized)
test_data_normalized = torch.FloatTensor(test_data_normalized)

这部分是定义优化器，我们的深度学习模型输入一般都是-1到1(虽然这不是必须的，但是如果你不进行标准化处理效果真是天差地别)，然后是测试集和训练集的划分，和根据数据进行标准化处理的操作，并且将数据转化为tensor的格式(tensor是我们深度学习特有的数据格式)。

# 定义训练器的的输入
train_inout_seq = create_inout_sequences(train_data_normalized, train_window, pre_len)
test_inout_seq = create_inout_sequences(test_data_normalized, train_window, pre_len)

这一部分是重点！！！！！

时间序列的数据和其他领域的不一样他需要滑窗的数据形式，假设我有100条数据，前面定义的滑窗大小是32预测未来数据的长度是4那么他就会用32和4去滑动数据，

所以我们的到数据是多少呢就是100 - 32 - 4 =54条数据(每条数据包含32条观测数据和4个标签数据)，这里必须理解大家这是时间序列的基础，他是不能够直接用Dataloader进行数据加载的。

# 创建数据集
train_dataset = TimeSeriesDataset(train_inout_seq)
test_dataset = TimeSeriesDataset(test_inout_seq)# 创建 DataLoader
batch_size = 32  # 你可以根据需要调整批量大小
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True, drop_last=True)
test_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False, drop_last=True)

这部分是创建数据集和Dataloader数据加载器，利用Dataloader的好处是可以避免内存爆炸，但是我们时间序列的数据一般都不大不会有这种情况。

class GRUConvNet(nn.Module):def __init__(self, input_size, hidden_size, num_layers, pre_len, output_size=1):super(GRUConvNet, self).__init__()# 可以外部定义的参数self.pre_len = pre_lenself.input_size = input_sizeself.hidden_size = hidden_sizeself.num_layers = num_layersself.output_size = output_size# GRU层self.gru = nn.GRU(input_size=self.input_size, hidden_size=hidden_size, num_layers=self.num_layers,batch_first=True)self.gru_dropout = nn.Dropout(0.05)# 卷积层self.conv1 = nn.Conv1d(in_channels=self.input_size, out_channels=hidden_size, kernel_size=9, padding='same')self.bn1 = nn.BatchNorm1d(hidden_size)self.conv2 = nn.Conv1d(in_channels=hidden_size, out_channels=hidden_size * 2, kernel_size=6, padding='same')self.bn2 = nn.BatchNorm1d(hidden_size * 2)self.conv3 = nn.Conv1d(in_channels=hidden_size * 2, out_channels=hidden_size, kernel_size=3, padding='same')self.bn3 = nn.BatchNorm1d(hidden_size)self.global_avg_pool = nn.AdaptiveAvgPool1d(1)# 输出层self.fc1 = nn.Linear(self.hidden_size, self.output_size)self.fc1 = nn.Linear(self.hidden_size, self.output_size)def forward(self, x):# GRU层x_, _ = self.gru(x)x_ = self.gru_dropout(x_[:, :, :])  # 使用序列的最后一个输出# 卷积层y = x.permute(0, 2, 1)y = F.relu(self.bn1(self.conv1(y)))y = F.relu(self.bn2(self.conv2(y)))y = F.relu(self.bn3(self.conv3(y)))y = self.global_avg_pool(y)y = y.permute(0, 2, 1)# 合并GRU和卷积层的输出combined = torch.cat([x_, y], dim=1)# 输出层output = self.fc1(combined)return output[:, -self.pre_len:, :]

讲完了初始化操作->我们再来看看正向传播里面的代码，里面的代码就是我们将输入先经过gru经过一个处理，然后取出它最后的一个时间步为预测值，然后我们将输入进去的值在输如给FCN里面里面进行处理经过三个不同大小的卷积处理以后输入到一个全局池化的操作(这个是现在常用的操作用全局池化代替全连接层进行输出)，最后我们将GRU和FCN的输入经过拼接在输入给一个全连接层进行输出得到我们的预测值，大家有兴趣建议还是debug一下我这么讲你是不能理解的，最好还是实际动手debug看一下其中的通道数变化情况，我相信debug一遍你会收获非常多。

lstm_model = GRU(input_dim=1, output_dim=1, num_layers=2, hidden_dim=train_window, pre_len=pre_len)
loss_function = nn.MSELoss()
optimizer = torch.optim.Adam(lstm_model.parameters(), lr=0.005)
epochs = 20
Train = True  # 训练还是预测

这里实例化了我们的模型，定义了MSE损失函数，和优化器Adam和训练轮次，其中的Train是来判断是否进行训练。

if Train:losss = []lstm_model.train()  # 训练模式for i in range(epochs):start_time = time.time()  # 计算起始时间for seq, labels in train_loader:lstm_model.train()optimizer.zero_grad()y_pred = lstm_model(seq)single_loss = loss_function(y_pred, labels)single_loss.backward()optimizer.step()print(f'epoch: {i:3} loss: {single_loss.item():10.8f}')losss.append(single_loss.detach().numpy())torch.save(lstm_model.state_dict(), 'save_model.pth')print(f"模型已保存,用时:{(time.time() - start_time) / 60:.4f} min")

如果Train为True则开始训练执行上面的代码，这是一个标准pytorch框架下的训练过程就不给大家说了，如果不能理解的话大家可以去补补基础，或者评论区问我我在给大家讲讲。

else:# 加载模型进行预测lstm_model.load_state_dict(torch.load('save_model.pth'))lstm_model.eval()  # 评估模式results = []reals = []losss = []for seq, labels in test_loader:pred = lstm_model(seq)mae = calculate_mae(pred.detach().numpy(), np.array(labels))  # MAE误差计算绝对值(预测值  - 真实值)losss.append(mae)for j in range(batch_size):for i in range(pre_len):reals.append(labels[j][i][0].detach().numpy())results.append(pred[j][i][0].detach().numpy())

如果Train为False时候则开始进行评估模式我们利用test的数据集进行测试评估训练模型，

 reals = scaler_test.inverse_transform(np.array(reals).reshape(1, -1))[0]results = scaler_test.inverse_transform(np.array(results).reshape(1, -1))[0]print("模型预测结果：", results)print("预测误差MAE:", losss)plt.figure()plt.style.use('ggplot')# 创建折线图plt.plot(reals, label='real', color='blue')  # 实际值plt.plot(results, label='forecast', color='red', linestyle='--')  # 预测值# 增强视觉效果plt.grid(True)plt.title('real vs forecast')plt.xlabel('time')plt.ylabel('value')plt.legend()plt.savefig('test——results.png')

这一部分是我们预测值和真实值之间的对比，来确定我们预测的好坏，后面的结果分析会有展示。