时间序列预测实战(二十四)PyTorch实现RNN进行多元和单元预测（附代码+数据集+完整解析）-编程知识

一、本文介绍

本篇文章给大家带来的是利用我个人编写的架构进行RNN时间序列卷积进行时间序列建模（专门为了时间序列领域新人编写的架构，简单且不同于市面上大家用GPT写的代码），包括结果可视化、支持单元预测、多元预测、模型拟合效果检测、预测未知数据、以及滚动长期预测功能。该结构是一个通用架构任何模型嵌入其中都可运行。下面来介绍一下RNN：循环神经网络（RNN）是深度学习中用于处理序列数据的一种基本网络结构。RNN的核心原理是它拥有一个循环结构，这使得网络能够保持一个内部的状态，从而捕捉到数据中随时间演变的信息。在RNN中，每个节点（或称为单元）在处理当前输入的同时，也会考虑前一时间步的隐藏状态。这种结构让RNN能够在时间序列分析、语言模型、文本生成等任务中建模数据中的时序依赖性，所以它可以用于时序预测，但是因为其存在导致梯度爆炸或者梯度消失的问题所以后来引出了LSTM，GRU等RNN模型。

专栏目录：时间序列预测目录：深度学习、机器学习、融合模型、创新模型实战案例

专栏：时间序列预测专栏：基础知识+数据分析+机器学习+深度学习+Transformer+创新模型

预测功能效果展示(不是测试集是预测未知数据，所以图中没有对比的数据)->

损失截图(损失这里我先展示一个训练过程中的后面会自动生成损失图像)->

根据损失来看模型的拟合效果还是很好的，但后面还是做了检验模型拟合效果的功能让大家真正的评估模型的效果。

测试集状况->

(从下面的图片可以看出模型在测试集的效果还可以，毕竟只是单一的RNN结构，为啥发单一因为我发现发高端或者融合的根本没人看)

一、本文介绍

二、RNN的框架原理

三、数据集介绍

四、参数讲解

五、完整代码

六、训练模型

七、预测结果

7.1 预测未知数据效果图

7.2 测试集效果图

7.3 CSV文件生成效果图

7.4 检验模型拟合效果图

八、全文总结

二、RNN的框架原理

循环神经网络（RNN）是一种专门用于处理序列数据的神经网络。它的主要思想是在处理序列的每个元素时，网络不仅考虑当前输入，还会考虑之前的信息。这使得RNN非常适合处理时间序列数据、语言模型等任务。

RNN的工作机制主要包括以下几点：

隐藏状态: RNN维护一个隐藏状态，它捕获到目前为止处理过的信息。
序列处理: 在处理序列的每个时间步时，RNN会更新其隐藏状态。
权重共享: 在不同时间步，RNN使用相同的权重，这减少了模型的复杂性并提高了训练效率。
输出: 根据任务的不同，RNN可以在每个时间步产生输出，或者仅在最后一个时间步产生输出。

总结来说，RNN之所以强大，是因为它能够利用序列数据的时间依赖性，通过维护一个在时间步之间传递的隐藏状态来实现这一点。这种结构使得RNN在处理诸如文本、语音等序列数据方面表现出色。然而，RNN也存在一定的局限性，比如难以处理长序列中的长期依赖问题，这通常通过引入LSTM或GRU等变体来解决。

下面的链接里包含了RNN系列的完整发展流程大家有兴趣的可以看看->

RNN发展流程：点击即可跳转

下面分享给大家两个LSTM和GRU的原理图吧，RNN实在没啥好说的，主要不发这些经典的文章还没人看。

LSTM结构图->

GRU结构图->

三、数据集介绍

本文是实战讲解文章，上面主要是简单讲解了一下网络结构比较具体的流程还是很复杂的涉及到很多的数学计算，下面我们来讲一讲模型的实战内容，第一部分是我利用的数据集。

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

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

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

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

四、参数讲解

parser.add_argument('-model', type=str, default='RNN', help="模型持续更新")parser.add_argument('-window_size', type=int, default=126, help="时间窗口大小, window_size > pre_len")parser.add_argument('-pre_len', type=int, default=24, help="预测未来数据长度")# dataparser.add_argument('-shuffle', action='store_true', default=True, help="是否打乱数据加载器中的数据顺序")parser.add_argument('-data_path', type=str, default='ETTh1-Test.csv', help="你的数据数据地址")parser.add_argument('-target', type=str, default='OT', help='你需要预测的特征列，这个值会最后保存在csv文件里')parser.add_argument('-input_size', type=int, default=7, help='你的特征个数不算时间那一列')parser.add_argument('-feature', type=str, default='M', help='[M, S, MS],多元预测多元,单元预测单元,多元预测单元')# learningparser.add_argument('-lr', type=float, default=0.001, help="学习率")parser.add_argument('-drop_out', type=float, default=0.05, help="随机丢弃概率,防止过拟合")parser.add_argument('-epochs', type=int, default=20, help="训练轮次")parser.add_argument('-batch_size', type=int, default=16, help="批次大小")parser.add_argument('-save_path', type=str, default='models')# modelparser.add_argument('-hidden_size', type=int, default=64, help="隐藏层单元数")parser.add_argument('-kernel_sizes', type=int, default=3)parser.add_argument('-laryer_num', type=int, default=2)# deviceparser.add_argument('-use_gpu', type=bool, default=True)parser.add_argument('-device', type=int, default=0, help="只设置最多支持单个gpu训练")# optionparser.add_argument('-train', type=bool, default=True)parser.add_argument('-test', type=bool, default=True)parser.add_argument('-predict', type=bool, default=True)parser.add_argument('-inspect_fit', type=bool, default=True)parser.add_argument('-lr-scheduler', type=bool, default=True)

为了大家方便理解，文章中的参数设置我都用的中文，所以大家应该能够更好的理解。下面我在进行一遍讲解。

	参数名称	参数类型	参数讲解
1	model	str	模型名称
2	window_size	int	时间窗口大小，用多少条数据去预测未来的数据
3	pre_len	int	预测多少条未来的数据
4	shuffle	store_true	是否打乱输入dataloader中的数据，不是数据的顺序
5	data_path	str	你输入数据的地址
6	target	str	你想要预测的特征列
7	input_size	int	输入的特征数不包含时间那一列！！！
8	feature	str	[M, S, MS],多元预测多元,单元预测单元,多元预测单元
9	lr	float	学习率大小
10	drop_out	float	丢弃概率
11	epochs	int	训练轮次
12	batch_size	int	批次大小
13	svae_path	str	模型的保存路径
14	hidden_size	int	隐藏层大小
15	kernel_size	int	卷积核大小
16	layer_num	int	lstm层数
17	use_gpu	bool	是否使用GPU
18	device	int	GPU编号
19	train	bool	是否进行训练
20	predict	bool	是否进行预测
21	inspect_fit	bool	是否进行检验模型
22	lr_schduler	bool	是否使用学习率计划

五、完整代码

复制粘贴到一个文件下并且按照上面的从参数讲解配置好参数即可运行~(极其适合新手和刚入门的读者)

import argparse
import timeimport numpy as np
import pandas as pd
import torch
import torch.nn as nn
from matplotlib import pyplot as plt
from torch.nn.utils import weight_norm
from torch.utils.data import DataLoader
from torch.utils.data import Dataset
from tqdm import tqdm# 随机数种子
np.random.seed(0)class StandardScaler():def __init__(self):self.mean = 0.self.std = 1.def fit(self, data):self.mean = data.mean(0)self.std = data.std(0)def transform(self, data):mean = torch.from_numpy(self.mean).type_as(data).to(data.device) if torch.is_tensor(data) else self.meanstd = torch.from_numpy(self.std).type_as(data).to(data.device) if torch.is_tensor(data) else self.stdreturn (data - mean) / stddef inverse_transform(self, data):mean = torch.from_numpy(self.mean).type_as(data).to(data.device) if torch.is_tensor(data) else self.meanstd = torch.from_numpy(self.std).type_as(data).to(data.device) if torch.is_tensor(data) else self.stdif data.shape[-1] != mean.shape[-1]:mean = mean[-1:]std = std[-1:]return (data * std) + meandef plot_loss_data(data):# 使用Matplotlib绘制线图plt.figure()plt.figure(figsize=(10, 5))plt.plot(data, marker='o')# 添加标题plt.title("loss results Plot")# 显示图例plt.legend(["Loss"])plt.show()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 create_inout_sequences(input_data, tw, pre_len, config):# 创建时间序列数据专用的数据分割器inout_seq = []L = len(input_data)for i in range(L - tw):train_seq = input_data[i:i + tw]if (i + tw + pre_len) > len(input_data):breakif config.feature == 'MS':train_label = input_data[:, -1:][i + tw:i + tw + pre_len]else:train_label = input_data[i + tw:i + tw + pre_len]inout_seq.append((train_seq, train_label))return inout_seqdef calculate_mae(y_true, y_pred):# 平均绝对误差mae = np.mean(np.abs(y_true - y_pred))return maedef create_dataloader(config, device):print(">>>>>>>>>>>>>>>>>>>>>>>>>>>>创建数据加载器<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<")df = pd.read_csv(config.data_path)  # 填你自己的数据地址,自动选取你最后一列数据为特征列 # 添加你想要预测的特征列pre_len = config.pre_len  # 预测未来数据的长度train_window = config.window_size  # 观测窗口# 将特征列移到末尾target_data = df[[config.target]]df = df.drop(config.target, axis=1)df = pd.concat((df, target_data), axis=1)cols_data = df.columns[1:]df_data = df[cols_data]# 这里加一些数据的预处理, 最后需要的格式是pd.seriestrue_data = df_data.values# 定义标准化优化器# 定义标准化优化器scaler = StandardScaler()scaler.fit(true_data)train_data = true_data[int(0.3 * len(true_data)):]valid_data = true_data[int(0.15 * len(true_data)):int(0.30 * len(true_data))]test_data = true_data[:int(0.15 * len(true_data))]print("训练集尺寸:", len(train_data), "测试集尺寸:", len(test_data), "验证集尺寸:", len(valid_data))# 进行标准化处理train_data_normalized = scaler.transform(train_data)test_data_normalized = scaler.transform(test_data)valid_data_normalized = scaler.transform(valid_data)# 转化为深度学习模型需要的类型Tensortrain_data_normalized = torch.FloatTensor(train_data_normalized).to(device)test_data_normalized = torch.FloatTensor(test_data_normalized).to(device)valid_data_normalized = torch.FloatTensor(valid_data_normalized).to(device)# 定义训练器的的输入train_inout_seq = create_inout_sequences(train_data_normalized, train_window, pre_len, config)test_inout_seq = create_inout_sequences(test_data_normalized, train_window, pre_len, config)valid_inout_seq = create_inout_sequences(valid_data_normalized, train_window, pre_len, config)# 创建数据集train_dataset = TimeSeriesDataset(train_inout_seq)test_dataset = TimeSeriesDataset(test_inout_seq)valid_dataset = TimeSeriesDataset(valid_inout_seq)# 创建 DataLoadertrain_loader = DataLoader(train_dataset, batch_size=args.batch_size, shuffle=True, drop_last=True)test_loader = DataLoader(test_dataset, batch_size=args.batch_size, shuffle=False, drop_last=True)valid_loader = DataLoader(valid_dataset, batch_size=args.batch_size, shuffle=False, drop_last=True)print("通过滑动窗口共有训练集数据：", len(train_inout_seq), "转化为批次数据:", len(train_loader))print("通过滑动窗口共有测试集数据：", len(test_inout_seq), "转化为批次数据:", len(test_loader))print("通过滑动窗口共有验证集数据：", len(valid_inout_seq), "转化为批次数据:", len(valid_loader))print(">>>>>>>>>>>>>>>>>>>>>>>>>>>>创建数据加载器完成<<<<<<<<<<<<<<<<<<<<<<<<<<<")return train_loader, test_loader, valid_loader, scalerclass RNNs(nn.Module):"""Model class to declare an rnn and define a forward pass of the model."""def __init__(self, input_size, output_size, hidden_size, num_layers, pred_len):"""'rnn'"""# inherit the nn.Module class via 'super'super(RNNs, self).__init__()# store stuff in the classself.pre_len = pred_lenself.n_layers = num_layersself.hidden_size = hidden_sizeself.hidden = nn.Linear(input_size, self.hidden_size)self.relu = nn.ReLU()self.rnn = nn.RNN(self.hidden_size, self.hidden_size, num_layers, bias=True, batch_first=True)  # output (batch_size, obs_len, hidden_size)self.linear = nn.Linear(self.hidden_size, output_size)def forward(self, x):# rnn module expects data of shape [seq, batch_size, input_size]batch_size, obs_len, features_size = x.shape  # (batch_size, obs_len, features_size)xconcat = self.hidden(x)  # (batch_size, obs_len, hidden_size)H = torch.zeros(batch_size, obs_len - 1, self.hidden_size).to(device)  # (batch_size, obs_len-1, hidden_size)ht = torch.zeros(self.n_layers, batch_size, self.hidden_size).to(device)  # (num_layers, batch_size, hidden_size)for t in range(obs_len):xt = xconcat[:, t, :].view(batch_size, 1, -1)  # (batch_size, 1, hidden_size)out, ht = self.rnn(xt, ht)  # ht size (num_layers, batch_size, hidden_size)htt = ht[-1, :, :]  # (batch_size, hidden_size)if t != obs_len - 1:H[:, t, :] = httH = self.relu(H)  # (batch_size, obs_len-1, hidden_size)x = self.linear(H)return x[:, -self.pre_len:, :]def train(model, args, scaler, device):start_time = time.time()  # 计算起始时间model = modelloss_function = nn.MSELoss()optimizer = torch.optim.Adam(model.parameters(), lr=0.005)epochs = args.epochsmodel.train()  # 训练模式results_loss = []for i in tqdm(range(epochs)):losss = []for seq, labels in train_loader:optimizer.zero_grad()y_pred = model(seq)single_loss = loss_function(y_pred, labels)single_loss.backward()optimizer.step()losss.append(single_loss.detach().cpu().numpy())tqdm.write(f"\t Epoch {i + 1} / {epochs}, Loss: {sum(losss) / len(losss)}")results_loss.append(sum(losss) / len(losss))torch.save(model.state_dict(), 'save_model.pth')time.sleep(0.1)# valid_loss = valid(model, args, scaler, valid_loader)# 尚未引入学习率计划后期补上# 保存模型print(f">>>>>>>>>>>>>>>>>>>>>>模型已保存,用时:{(time.time() - start_time) / 60:.4f} min<<<<<<<<<<<<<<<<<<")plot_loss_data(results_loss)def valid(model, args, scaler, valid_loader):lstm_model = model# 加载模型进行预测lstm_model.load_state_dict(torch.load('save_model.pth'))lstm_model.eval()  # 评估模式losss = []for seq, labels in valid_loader:pred = lstm_model(seq)mae = calculate_mae(pred.detach().numpy().cpu(), np.array(labels.detach().cpu()))  # MAE误差计算绝对值(预测值  - 真实值)losss.append(mae)print("验证集误差MAE:", losss)return sum(losss) / len(losss)def test(model, args, test_loader, scaler):# 加载模型进行预测losss = []model = modelmodel.load_state_dict(torch.load('save_model.pth'))model.eval()  # 评估模式results = []labels = []for seq, label in test_loader:pred = model(seq)mae = calculate_mae(pred.detach().cpu().numpy(),np.array(label.detach().cpu()))  # MAE误差计算绝对值(预测值  - 真实值)losss.append(mae)pred = pred[:, 0, :]label = label[:, 0, :]pred = scaler.inverse_transform(pred.detach().cpu().numpy())label = scaler.inverse_transform(label.detach().cpu().numpy())for i in range(len(pred)):results.append(pred[i][-1])labels.append(label[i][-1])plt.figure(figsize=(10, 5))print("测试集误差MAE:", losss)# 绘制历史数据plt.plot(labels, label='TrueValue')# 绘制预测数据# 注意这里预测数据的起始x坐标是历史数据的最后一个点的x坐标plt.plot(results, label='Prediction')# 添加标题和图例plt.title("test state")plt.legend()plt.show()# 检验模型拟合情况
def inspect_model_fit(model, args, train_loader, scaler):model = modelmodel.load_state_dict(torch.load('save_model.pth'))model.eval()  # 评估模式results = []labels = []for seq, label in train_loader:pred = model(seq)[:, 0, :]label = label[:, 0, :]pred = scaler.inverse_transform(pred.detach().cpu().numpy())label = scaler.inverse_transform(label.detach().cpu().numpy())for i in range(len(pred)):results.append(pred[i][-1])labels.append(label[i][-1])plt.figure(figsize=(10, 5))# 绘制历史数据plt.plot(labels, label='History')# 绘制预测数据# 注意这里预测数据的起始x坐标是历史数据的最后一个点的x坐标plt.plot(results, label='Prediction')# 添加标题和图例plt.title("inspect model fit state")plt.legend()plt.show()def predict(model, args, device, scaler):# 预测未知数据的功能df = pd.read_csv(args.data_path)df = df.iloc[:, 1:][-args.window_size:].values  # 转换为nadarrypre_data = scaler.transform(df)tensor_pred = torch.FloatTensor(pre_data).to(device)tensor_pred = tensor_pred.unsqueeze(0)  # 单次预测 , 滚动预测功能暂未开发后期补上model = modelmodel.load_state_dict(torch.load('save_model.pth'))model.eval()  # 评估模式pred = model(tensor_pred)[0]pred = scaler.inverse_transform(pred.detach().cpu().numpy())# 假设 df 和 pred 是你的历史和预测数据# 计算历史数据的长度history_length = len(df[:, -1])# 为历史数据生成x轴坐标history_x = range(history_length)plt.figure(figsize=(10, 5))# 为预测数据生成x轴坐标# 开始于历史数据的最后一个点的x坐标prediction_x = range(history_length - 1, history_length + len(pred[:, -1]) - 1)# 绘制历史数据plt.plot(history_x, df[:, -1], label='History')# 绘制预测数据# 注意这里预测数据的起始x坐标是历史数据的最后一个点的x坐标plt.plot(prediction_x, pred[:, -1], marker='o', label='Prediction')plt.axvline(history_length - 1, color='red')  # 在图像的x位置处画一条红色竖线# 添加标题和图例plt.title("History and Prediction")plt.legend()if __name__ == '__main__':parser = argparse.ArgumentParser(description='Time Series forecast')parser.add_argument('-model', type=str, default='RNN', help="模型持续更新")parser.add_argument('-window_size', type=int, default=126, help="时间窗口大小, window_size > pre_len")parser.add_argument('-pre_len', type=int, default=24, help="预测未来数据长度")# dataparser.add_argument('-shuffle', action='store_true', default=True, help="是否打乱数据加载器中的数据顺序")parser.add_argument('-data_path', type=str, default='ETTh1-Test.csv', help="你的数据数据地址")parser.add_argument('-target', type=str, default='OT', help='你需要预测的特征列，这个值会最后保存在csv文件里')parser.add_argument('-input_size', type=int, default=7, help='你的特征个数不算时间那一列')parser.add_argument('-feature', type=str, default='MS', help='[M, S, MS],多元预测多元,单元预测单元,多元预测单元')parser.add_argument('-model_dim', type=list, default=[64, 128, 256], help='这个地方是这个TCN卷积的关键部分,它代表了TCN的层数我这里输''入list中包含三个元素那么我的TCN就是三层，这个根据你的数据复杂度来设置''层数越多对应数据越复杂但是不要超过5层')# learningparser.add_argument('-lr', type=float, default=0.001, help="学习率")parser.add_argument('-drop_out', type=float, default=0.05, help="随机丢弃概率,防止过拟合")parser.add_argument('-epochs', type=int, default=20, help="训练轮次")parser.add_argument('-batch_size', type=int, default=16, help="批次大小")parser.add_argument('-save_path', type=str, default='models')# modelparser.add_argument('-hidden_size', type=int, default=128, help="隐藏层单元数")parser.add_argument('-kernel_sizes', type=int, default=3)parser.add_argument('-laryer_num', type=int, default=2)# deviceparser.add_argument('-use_gpu', type=bool, default=True)parser.add_argument('-device', type=int, default=0, help="只设置最多支持单个gpu训练")# optionparser.add_argument('-train', type=bool, default=True)parser.add_argument('-test', type=bool, default=True)parser.add_argument('-predict', type=bool, default=True)parser.add_argument('-inspect_fit', type=bool, default=True)parser.add_argument('-lr-scheduler', type=bool, default=True)args = parser.parse_args()if isinstance(args.device, int) and args.use_gpu:device = torch.device("cuda:" + f'{args.device}')else:device = torch.device("cpu")print("使用设备:", device)train_loader, test_loader, valid_loader, scaler = create_dataloader(args, device)if args.feature == 'MS' or args.feature == 'S':args.output_size = 1else:args.output_size = args.input_size# 实例化模型try:print(f">>>>>>>>>>>>>>>>>>>>>>>>>开始初始化{args.model}模型<<<<<<<<<<<<<<<<<<<<<<<<<<<")model = RNNs(args.input_size, args.output_size, args.hidden_size, args.laryer_num, args.pre_len).to(device)print(f">>>>>>>>>>>>>>>>>>>>>>>>>开始初始化{args.model}模型成功<<<<<<<<<<<<<<<<<<<<<<<<<<<")except:print(f">>>>>>>>>>>>>>>>>>>>>>>>>开始初始化{args.model}模型失败<<<<<<<<<<<<<<<<<<<<<<<<<<<")# 训练模型if args.train:print(f">>>>>>>>>>>>>>>>>>>>>>>>>开始{args.model}模型训练<<<<<<<<<<<<<<<<<<<<<<<<<<<")train(model, args, scaler, device)if args.test:print(f">>>>>>>>>>>>>>>>>>>>>>>>>开始{args.model}模型测试<<<<<<<<<<<<<<<<<<<<<<<<<<<")test(model, args, test_loader, scaler)if args.inspect_fit:print(f">>>>>>>>>>>>>>>>>>>>>>>>>开始检验{args.model}模型拟合情况<<<<<<<<<<<<<<<<<<<<<<<<<<<")inspect_model_fit(model, args, train_loader, scaler)if args.predict:print(f">>>>>>>>>>>>>>>>>>>>>>>>>预测未来{args.pre_len}条数据<<<<<<<<<<<<<<<<<<<<<<<<<<<")predict(model, args, device, scaler)plt.show()