时间序列预测实战(二十六)PyTorch实现Seq2Seq(LSTM)（结果可视化+自动保存结果）
2023-12-15 04:45:37
一、本文介绍
本文给大家带来的时间序列模型是Seq2Seq，这个概念相信大家都不陌生了，网上的讲解已经满天飞了，但是本文给大家带来的是我在Seq2Seq思想上开发的一个模型和新的架构，其是专门为新手开发的，而且为了方便大家使用只定义了一个文件方便大家复制粘贴，架构功能包括：结果可视化、支持单元预测、多元预测、模型拟合效果检测、预测未知数据、以及滚动长期预测功能。Seq2Seq模型是一种处理序列数据的深度学习思想，广泛用于机器翻译、语音识别和文本摘要等任务(也能用于时间序列)。其核心思想是编码器-解码器。本文就用Seq2Seq实现LSTM进行时间序列预测，预测效果吊打普通的LSTM进行预测，同时滚动预测于本文已经更新。
需要注意得是本文的模型和结构均为自研，在Seq2Seq和LSTM的思想上进行了一定的扩展。?
预测功能效果展示(不是测试集是预测未知数据，所以图中没有对比的数据)->
同时本文的预测结果会自动生成CSV文件方便大家进行二次评估(此功能已经实现)->
以下CSV文件和图片模型均可自动生成。?
测试集状况->
PS：测试集是未输入到模型里进行训练的数据表现情况~
目录
二、Seq2Seq思想原理
2.1 Seq2Seq的基本原理
Seq2Seq（Sequence-to-Sequence）模型是一种处理序列数据的深度学习模型，广泛用于机器翻译、语音识别和文本摘要等任务。其核心思想是将一个序列（如一句话）转换成另一个序列，这两个序列的长度可以不同。
Seq2Seq模型主要包括以下几个机制和原理：
1. 编码器-解码器架构：Seq2Seq模型通常由两部分组成，编码器和解码器。编码器负责读取并理解输入序列，将其转换成一个固定长度的上下文向量（context vector）。解码器则利用这个上下文向量生成目标序列（其实最核心的就是这个，下面都是具体的应用了）。
2. 循环神经网络（RNN）：在传统的Seq2Seq模型中，编码器和解码器通常是循环神经网络（如LSTM或GRU）。RNN可以处理不同长度的输入序列，并在其隐藏层保持序列的状态信息。
3.注意力机制（Attention）：注意力机制是后来引入Seq2Seq模型的一项重要改进。它允许模型在生成每个目标词时“关注”输入序列的不同部分，从而提高了模型处理长句子时的效果和准确性。
4. 长短期记忆网络（LSTM）/门控循环单元（GRU）：为了解决RNN中的长期依赖问题，LSTM和GRU这样的网络结构被引入。它们能更好地捕捉序列中的长期依赖关系。
个人总结：我个人觉得Seq2Seq模型就像是学习语言的人。首先通过“编码器”理解输入的句子，然后用“解码器”来表达新的句子。就像我们学外语时先理解一句话的意思，再用自己的语言表达出来。加上“注意力机制”，模型还能更聪明地关注输入句子中最重要的部分，就好比我们在听别人说话时会注意对方重点强调的内容。
2.1.1 编码器-解码器介绍
通过上面我们知道Seq2Seq的主要核心思想是编码器和解码器，Seq2Seq一开始被发明出来是用于一些本文处理的，但是本文是时间序列领域的文章，所以我主要讲解一下编码器-解码器在时间序列领域的应用（大家需要注意的是这里的编码器和解码器和Transformer当中的还不一样是有着根本的区别的）。Seq2Seq模型的编码器和解码器如下工作：
编码器在时间序列领域的作用是理解和编码输入序列的历史数据。
1. 处理时间序列输入：编码器接收时间序列数据，例如过去几天的股票价格或气温记录。这些数据通常是连续的数值。
2. 特征提取：通过RNN(本文是用的GRU，类似于LSTM和RNN以后也会单独出文章)网络，编码器可以捕捉时间序列的特征和内在模式。RNN通过其时间递归结构，有效地处理时间序列数据中的时序依赖关系。
3. 上下文向量：编码器输出一个上下文向量，该向量是输入时间序列的压缩表示，包含了对过去数据的理解和总结。
解码器在时间序列领域的作用是基于编码器提供的信息来预测未来的时间序列数据。
1. 初始状态和输入：解码器的初始状态通常由编码器的最终状态设置。解码器的第一个输入可能是序列的最后一个观测值或特殊标记（我设置为解码器的输入是编码器的最后一个输出，因为我觉得这个状态的过去信息是最足的）。
2. 逐步预测：在每个时间步，解码器基于当前状态和前一步的预测输出（或初始输入）来生成下一个时间点的预测值。
3. 迭代更新：解码器的输出用于更新其状态，并作为下一个时间步的输入。这个过程在生成整个预测序列期间重复进行。
4. 序列生成：这个过程持续进行，直到生成了预定长度的预测序列。
总结：在时间序列领域，Seq2Seq模型的编码器-解码器结构使其能够有效处理具有复杂时间依赖性的序列数据。编码器学习并压缩历史数据的关键信息，而解码器则利用这些信息来预测未来的趋势和模式。
2.1.2 编码器-解码器结构图
下面的图是我编码器-解码器的结构图，其中包含了编码器（Encoder）和解码器（Decoder）的具体实现细节（需要注意的是这个图和本文模型无关，这里举出来只是为了让大家对编码器和解码器的流程有一个更清晰的认知）。下面是对这个图示的详细解释：
编码器部分（左侧）
-
嵌入层（Embed）：这个层将输入的一热编码（one-hot vector）转换为嵌入向量。在此图中，嵌入层将每个输入token转换为一个256维的向量。
-
GRU层：门控循环单元（Gated Recurrent Unit，GRU）是RNN的一种变体，可以捕捉长期依赖关系，同时缓解了传统RNN的梯度消失问题。在此模型中，每个时间步的GRU层接收前一时间步的隐藏状态和当前时间步的嵌入向量，然后输出新的隐藏状态。
-
时间步（t）：时间步是序列中的位置指示，从t=?2开始，直至t=0，表示输入序列的处理进程。
-
位置标记（Pos. = T）：这可能表示当前处理的token位于输入序列的最后一个位置，意味着编码器即将完成对输入序列的处理。
解码器部分（右侧）
-
嵌入层：解码器的嵌入层将一热编码的输出token转换为嵌入向量。
-
GRU层：解码器的GRU层接收来自上一时间步的隐藏状态和当前时间步的嵌入向量（或初始状态从编码器传来的上下文向量），然后输出新的隐藏状态。
-
全连接层：在每个时间步，全连接层将GRU的输出转换为词汇表大小的向量，这个向量包含了输出序列中下一个token的概率分布。
-
输出概率：这是解码器产生的，表示下一个可能输出token的概率分布。
-
SOS：这是“Start of Sequence”的缩写，表示解码器开始生成序列的信号。
-
时间步（t）：在解码器部分，时间步从t=1开始，直至t=T+1，其中T可能表示目标序列的长度。
整个过程是这样的：编码器读取输入序列的token，并逐个更新其隐藏状态。当编码器读取完所有的输入token后，最后的隐藏状态（?0?）被传递到解码器作为其初始状态。解码器从一个特殊的SOS token开始，逐步生成输出序列的token。在每一步，解码器基于当前的隐藏状态和上一步产生的token，预测下一个token，直到生成EOS token，表示序列生成结束。
2.2 滚动预测介绍
本文新增加了滚动预测的功能，这里来简单说一下这个功能，这个功能就是指我们可以第一次预测未来24个时间段的值然后我们像模型中填补 24个值再次去预测未来24个时间段的值(填补功能我设置成自动的了无需大家手动填补)，这个功能可以说是很实用的，这样我们可以准确的评估固定时间段的值，当我们实际使用时可以设置自动爬取数据从而产生实际效用。本文修改内容完全为本人个人开发，创作不易所以如果能够帮助到大家希望大家给我的文章点点赞，同时可以关注本专栏(免费阅读)，本专栏持续复现各种的顶会内容，无论你想发顶会还是其它水平的论文都能够对你有所帮助。
概念部分的就讲这么多，网上有很多概念理解的好博客，大家有兴趣都可以自己查找看看，本文是实战博客内容不多讲啦~?
三、数据集介绍
本文是实战讲解文章，上面主要是简单讲解了一下网络结构比较具体的流程还是很复杂的涉及到很多的数学计算，下面我们来讲一讲模型的实战内容，第一部分是我利用的数据集。
本文我们用到的数据集是ETTh1.csv，该数据集是一个用于时间序列预测的电力负荷数据集，它是 ETTh 数据集系列中的一个。ETTh 数据集系列通常用于测试和评估时间序列预测模型。以下是 ETTh1.csv 数据集的一些内容：
数据内容：该数据集通常包含有关电力系统的多种变量，如电力负荷、价格、天气情况等。这些变量可以用于预测未来的电力需求或价格。
时间范围和分辨率：数据通常按小时或天记录，涵盖了数月或数年的时间跨度。具体的时间范围和分辨率可能会根据数据集的版本而异。?
以下是该数据集的部分截图->
四、参数讲解?
parser = argparse.ArgumentParser(description='Time Series forecast')
parser.add_argument('-model', type=str, default='LSTM2LSTM', help="模型持续更新")
parser.add_argument('-window_size', type=int, default=24, help="时间窗口大小, window_size > pre_len")
parser.add_argument('-pre_len', type=int, default=1, help="预测未来数据长度")
# data
parser.add_argument('-shuffle', action='store_true', default=True, help="是否打乱数据加载器中的数据顺序")
parser.add_argument('-data_path', type=str, default='ETTh1.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],多元预测多元,单元预测单元,多元预测单元')
# learning
parser.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')
# model
parser.add_argument('-hidden_size', type=int, default=128, help="隐藏层单元数")
parser.add_argument('-laryer_num', type=int, default=2)
# device
parser.add_argument('-use_gpu', type=bool, default=True)
parser.add_argument('-device', type=int, default=0, help="只设置最多支持单个gpu训练")
# option
parser.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)
# 可选部分，滚动预测如果想要进行这个需要你有一个额外的文件和你的训练数据集完全相同但是数据时间点不同。
parser.add_argument('-rolling_predict', type=bool, default=True)
parser.add_argument('-roolling_data_path', type=str, default='ETTh1Test.csv', help="你滚动数据集的地址，此部分属于进阶功能")为了大家方便理解，文章中的参数设置我都用的中文，所以大家应该能够更好的理解。下面我在进行一遍讲解。?
| 参数名称 | 参数类型 | 参数讲解 | |
|---|---|---|---|
| 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 | 是否使用学习率计划 |
| 23 | rolling_predict | bool | 是否进行滚动预测功能， 此功能为本文新加 |
| 24 | rolling_data_path | bool | 你滚动数据集的地址，此部分属于进阶功能 |
五、完整代码
复制粘贴到一个文件下并且按照上面的从参数讲解配置好参数即可运行~(极其适合新手和刚入门的读者)
import argparse
import time
import numpy as np
import pandas as pd
import torch
import torch.nn as nn
from matplotlib import pyplot as plt
from torch.utils.data import DataLoader
from torch.utils.data import Dataset
from tqdm import tqdm
import torch.nn.functional as F
# 随机数种子
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.mean
std = torch.from_numpy(self.std).type_as(data).to(data.device) if torch.is_tensor(data) else self.std
return (data - mean) / std
def inverse_transform(self, data):
mean = torch.from_numpy(self.mean).type_as(data).to(data.device) if torch.is_tensor(data) else self.mean
std = torch.from_numpy(self.std).type_as(data).to(data.device) if torch.is_tensor(data) else self.std
if data.shape[-1] != mean.shape[-1]:
mean = mean[-1:]
std = std[-1:]
return (data * std) + mean
def 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 = sequences
def __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):
break
if 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_seq
def calculate_mae(y_true, y_pred):
# 平均绝对误差
mae = np.mean(np.abs(y_true - y_pred))
return mae
def 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.series
true_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)
# 转化为深度学习模型需要的类型Tensor
train_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)
# 创建 DataLoader
train_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, scaler
class LSTMEncoder(nn.Module):
def __init__(self, rnn_num_layers=1, input_feature_len=1, sequence_len=168, hidden_size=100, bidirectional=False):
super().__init__()
self.sequence_len = sequence_len
self.hidden_size = hidden_size
self.input_feature_len = input_feature_len
self.num_layers = rnn_num_layers
self.rnn_directions = 2 if bidirectional else 1
self.lstm = nn.LSTM(
num_layers=rnn_num_layers,
input_size=input_feature_len,
hidden_size=hidden_size,
batch_first=True,
bidirectional=bidirectional
)
def forward(self, input_seq):
ht = torch.zeros(self.num_layers * self.rnn_directions, input_seq.size(0), self.hidden_size, device='cuda')
ct = ht.clone()
if input_seq.ndim < 3:
input_seq.unsqueeze_(2)
lstm_out, (ht, ct) = self.lstm(input_seq, (ht,ct))
if self.rnn_directions > 1:
lstm_out = lstm_out.view(input_seq.size(0), self.sequence_len, self.rnn_directions, self.hidden_size)
lstm_out = torch.sum(lstm_out, axis=2)
return lstm_out, ht.squeeze(0)
class AttentionDecoderCell(nn.Module):
def __init__(self, input_feature_len, out_put, sequence_len, hidden_size):
super().__init__()
# attention - inputs - (decoder_inputs, prev_hidden)
self.attention_linear = nn.Linear(hidden_size + input_feature_len, sequence_len)
# attention_combine - inputs - (decoder_inputs, attention * encoder_outputs)
self.decoder_rnn_cell = nn.LSTMCell(
input_size=hidden_size,
hidden_size=hidden_size,
)
self.out = nn.Linear(hidden_size, input_feature_len)
def forward(self, encoder_output, prev_hidden, y):
if prev_hidden.ndimension() == 3:
prev_hidden = prev_hidden[-1] # 保留最后一层的信息
attention_input = torch.cat((prev_hidden, y), axis=1)
attention_weights = F.softmax(self.attention_linear(attention_input), dim=-1).unsqueeze(1)
attention_combine = torch.bmm(attention_weights, encoder_output).squeeze(1)
rnn_hidden, rnn_hidden = self.decoder_rnn_cell(attention_combine, (prev_hidden, prev_hidden))
output = self.out(rnn_hidden)
return output, rnn_hidden
class EncoderDecoderWrapper(nn.Module):
def __init__(self, input_size, output_size, hidden_size, num_layers, pred_len, window_size, teacher_forcing=0.3):
super().__init__()
self.encoder = LSTMEncoder(num_layers, input_size, window_size, hidden_size)
self.decoder_cell = AttentionDecoderCell(input_size, output_size, window_size, hidden_size)
self.output_size = output_size
self.input_size = input_size
self.pred_len = pred_len
self.teacher_forcing = teacher_forcing
self.linear = nn.Linear(input_size,output_size)
def __call__(self, xb, yb=None):
input_seq = xb
encoder_output, encoder_hidden = self.encoder(input_seq)
prev_hidden = encoder_hidden
if torch.cuda.is_available():
outputs = torch.zeros(self.pred_len, input_seq.size(0), self.input_size, device='cuda')
else:
outputs = torch.zeros(input_seq.size(0), self.output_size)
y_prev = input_seq[:, -1, :]
for i in range(self.pred_len):
if (yb is not None) and (i > 0) and (torch.rand(1) < self.teacher_forcing):
y_prev = yb[:, i].unsqueeze(1)
rnn_output, prev_hidden = self.decoder_cell(encoder_output, prev_hidden, y_prev)
y_prev = rnn_output
outputs[i, :, :] = rnn_output
outputs = outputs.permute(1, 0, 2)
if self.output_size == 1:
outputs = self.linear(outputs)
return outputs
def train(model, args, scaler, device):
start_time = time.time() # 计算起始时间
model = model
loss_function = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.005)
epochs = args.epochs
model.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 = model
model.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 = model
model.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=None, args=None, device=None, scaler=None, rolling_data=None, show=False):
# 预测未知数据的功能
df = pd.read_csv(args.data_path)
df = pd.concat((df,rolling_data), axis=0).reset_index(drop=True)
df = df.iloc[:, 1:][-args.window_size:].values # 转换为nadarry
pre_data = scaler.transform(df)
tensor_pred = torch.FloatTensor(pre_data).to(device)
tensor_pred = tensor_pred.unsqueeze(0) # 单次预测 , 滚动预测功能暂未开发后期补上
model = model
model.load_state_dict(torch.load('save_model.pth'))
model.eval() # 评估模式
pred = model(tensor_pred)[0]
pred = scaler.inverse_transform(pred.detach().cpu().numpy())
if show:
# 计算历史数据的长度
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()
return pred
def rolling_predict(model=None, args=None, device=None, scaler=None):
# 滚动预测
history_data = pd.read_csv(args.data_path)[args.target][-args.window_size * 4:].reset_index(drop=True)
pre_data = pd.read_csv(args.roolling_data_path)
columns = pre_data.columns[1:]
columns = ['forecast' + column for column in columns]
dict_of_lists = {column: [] for column in columns}
results = []
for i in range(int(len(pre_data)/args.pre_len)):
rolling_data = pre_data.iloc[:args.pre_len * i] # 转换为nadarry
pred = predict(model, args, device, scaler, rolling_data)
if args.feature == 'MS' or args.feature == 'S':
for i in range(args.pred_len):
results.append(pred[i][0].detach().cpu().numpy())
else:
for j in range(args.output_size):
for i in range(args.pre_len):
dict_of_lists[columns[j]].append(pred[i][j])
print(pred)
if args.feature == 'MS' or args.feature == 'S':
df = pd.DataFrame({'date':pre_data['date'], '{}'.format(args.target): pre_data[args.target],
'forecast{}'.format(args.target): pre_data[args.target]})
df.to_csv('Interval-{}'.format(args.data_path), index=False)
else:
df = pd.DataFrame(dict_of_lists)
new_df = pd.concat((pre_data,df), axis=1)
new_df.to_csv('Interval-{}'.format(args.data_path), index=False)
pre_len = len(dict_of_lists['forecast' + args.target])
# 绘图
plt.figure()
if args.feature == 'MS' or args.feature == 'S':
plt.plot(range(len(history_data)), history_data,
label='Past Actual Values')
plt.plot(range(len(history_data), len(history_data) + pre_len), pre_data[args.target][:pre_len].tolist(), label='Predicted Actual Values')
plt.plot(range(len(history_data), len(history_data) + pre_len), results, label='Predicted Future Values')
else:
plt.plot(range(len(history_data)), history_data,
label='Past Actual Values')
plt.plot(range(len(history_data), len(history_data) + pre_len), pre_data[args.target][:pre_len].tolist(), label='Predicted Actual Values')
plt.plot(range(len(history_data), len(history_data) + pre_len), dict_of_lists['forecast' + args.target], label='Predicted Future Values')
# 添加图例
plt.legend()
plt.style.use('ggplot')
# 添加标题和轴标签
plt.title('Past vs Predicted Future Values')
plt.xlabel('Time Point')
plt.ylabel('Value')
# 在特定索引位置画一条直线
plt.axvline(x=len(history_data), color='blue', linestyle='--', linewidth=2)
# 显示图表
plt.savefig('forcast.png')
plt.show()
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='Time Series forecast')
parser.add_argument('-model', type=str, default='LSTM2LSTM', help="模型持续更新")
parser.add_argument('-window_size', type=int, default=64, help="时间窗口大小, window_size > pre_len")
parser.add_argument('-pre_len', type=int, default=24, help="预测未来数据长度")
# data
parser.add_argument('-shuffle', action='store_true', default=True, help="是否打乱数据加载器中的数据顺序")
parser.add_argument('-data_path', type=str, default='ETTh1.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],多元预测多元,单元预测单元,多元预测单元')
# learning
parser.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')
# model
parser.add_argument('-hidden_size', type=int, default=128, help="隐藏层单元数")
parser.add_argument('-laryer_num', type=int, default=2)
# device
parser.add_argument('-use_gpu', type=bool, default=True)
parser.add_argument('-device', type=int, default=0, help="只设置最多支持单个gpu训练")
# option
parser.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)
# 可选部分，滚动预测如果想要进行这个需要你有一个额外的文件和你的训练数据集完全相同但是数据时间点不同。
parser.add_argument('-rolling_predict', type=bool, default=True)
parser.add_argument('-roolling_data_path', type=str, default='ETTh1Test.csv', help="你滚动数据集的地址，此部分属于进阶功能")
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 = 1
else:
args.output_size = args.input_size
# 实例化模型
try:
print(f">>>>>>>>>>>>>>>>>>>>>>>>>开始初始化{args.model}模型<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<")
model = EncoderDecoderWrapper(args.input_size, args.output_size, args.hidden_size, args.laryer_num, args.pre_len, args.window_size).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,show=True)
if args.predict:
print(f">>>>>>>>>>>>>>>>>>>>>>>>>滚动预测未来{args.pre_len}条数据<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<")
rolling_predict(model, args, device, scaler)
plt.show()
六、训练模型?
6.1 模型训练
我们配置好所有参数之后就可以开始训练模型了，根据我前面讲解的参数部分进行配置，不懂得可以评论区留言。
6.2 损失图像
损失的图像绘图如下->
?
?
?
七、预测结果
7.1 预测未知数据效果图
Seq2Seq(LSTM)的预测效果图(这里我只预测了未来24个时间段的值为未来一天的预测值)->
7.2 测试集效果图?
测试集上的表现->
?
7.3 CSV文件生成效果图?
此功能于本文最新已经更新啦~
将滚动预测结果生成了csv文件方便大家对比和评估，以下是我生成的csv文件可以说是非常的直观。
?我们可以利用其进行画图从而评估结果->?
?
7.4 检验模型拟合效果图
检验模型拟合情况->
(从下面的图片可以看出模型拟合的情况还行，上一篇RNN的有一点过拟合了其实，则会个表现还是很正常的)?
八、全文总结
到此本文的正式分享内容就结束了，在这里给大家推荐我的时间序列专栏，本专栏目前为新开的平均质量分98分，后期我会根据各种最新的前沿顶会进行论文复现，也会对一些老的模型进行补充，目前本专栏免费阅读(暂时，大家尽早关注不迷路~)，如果大家觉得本文帮助到你了，订阅本专栏，关注后续更多的更新~
?专栏回顾：?时间序列预测专栏——持续复习各种顶会内容——科研必备
如果大家有不懂的也可以评论区留言一些报错什么的大家可以讨论讨论看到我也会给大家解答如何解决！最后希望大家工作顺利学业有成！
?
文章来源:https://blog.csdn.net/java1314777/article/details/134864319
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。 如若内容造成侵权/违法违规/事实不符，请联系我的编程经验分享网邮箱：veading@qq.com进行投诉反馈，一经查实，立即删除！
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。 如若内容造成侵权/违法违规/事实不符，请联系我的编程经验分享网邮箱：veading@qq.com进行投诉反馈，一经查实，立即删除！