GRU for Time Series in Pytorch–Pytorch实现时间序列分析

  • A+
所属分类:深度学习
摘要这一篇文章会介绍使用RNN来完成时间序列的分析,更准确的说是时间序列的预测。本篇内容会介绍使用GRU来预测AR(5)的数据。

简介

这一篇文章,会完成使用RNN, 更具体的说是使用GRU来实现时间序列的分析, 用来做预测. 最终的效果如下, 后面会有每一步的具体步骤。

GRU for Time Series in Pytorch--Pytorch实现时间序列分析

主要参考的文章为下面两篇文章 :

数据输入

在处理时间序列数据的时候,对于输入我们有两种处理的办法。我们会将数据分为chunk。

如图一, 比如我们使用前6天预测第七天, 再使用前七天预测第八天, 以此类推.

如图二, 比如我们使用0-6天预测第七天, 使用1-7天预测第八天, 以此类推.

GRU for Time Series in Pytorch--Pytorch实现时间序列分析

在我们这篇文章,我们会采取使用图2的形式. 我们会以7天作为一个阶段, 使用七天的数据预测第八天.

In our case study, I will use samples consisting of 7-days sliding windows with step size equal to 1, used to predict the next value (1-step ahead forecast).

下面我们看一下具体的实现。这里一段内容参考自上面的链接二。

具体的实现

导入所需要的库

  1. import torch
  2. import torch.nn as nn
  3. import torch.optim as optim
  4. import numpy as np
  5. from sklearn.model_selection import train_test_split
  6. import matplotlib.pyplot as plt

Generating Autoregressive data for experiments(生成所需要的数据)

这里的代码来自上面的链接一, 我在代码注释中也写了来源, 这个数据我没仔细看, 后面就是直接使用了。

  1. # 代码来源 : http://www.jessicayung.com/generating-autoregressive-data-for-experiments/
  2. # 代码来源 : https://github.com/jessicayung/blog-code-snippets
  3. class TimeSeriesData:
  4.     def __init__(self, num_datapoints, test_size=0.2, max_t=20, num_prev=1,
  5.                  noise_var=1):
  6.         """
  7.         Template class for generating time series data.
  8.         :param test_size: in (0,1), data to be used in test set as a fraction of all data generated.
  9.         """
  10.         self.num_datapoints = num_datapoints
  11.         self.test_size = test_size
  12.         self.num_prev = num_prev
  13.         self.max_t = max_t
  14.         self.data = None
  15.         self.noise_var = noise_var
  16.         self.y = np.zeros(num_datapoints + num_prev*4) # TODO: check this
  17.         self.bayes_preds = np.copycopy(self.y)
  18.         # Generate data and reshape data
  19.         self.create_data()
  20.         # Split into training and test sets
  21.         self.train_test_split()
  22.     def create_data(self):
  23.         self.generate_data()
  24.         self.reshape_data()
  25.     def generate_data(self):
  26.         """Generates data in self.y, may take as implicit input timesteps self.t.
  27.         May also generate Bayes predictions."""
  28.         raise NotImplementedError("Generate data method not implemented.")
  29.     def reshape_data(self):
  30.         self.x = np.reshape([self.y[i:i + self.num_prev] for i in range(
  31.             self.num_datapoints)], (-1, self.num_prev))
  32.         self.y = np.copycopy(self.y[self.num_prev:])
  33.         self.bayes_preds = np.copycopy(self.bayes_preds[self.num_prev:])
  34.     def train_test_split(self):
  35.         test_size = int(len(self.y) * self.test_size)
  36.         self.data = [self.X_train, self.X_test, self.y_train,
  37.                      self.y_test] = \
  38.                     self.x[:-test_size], self.x[-test_size:], \
  39.                     self.y[:-test_size], self.y[-test_size:]
  40.         self.bayes_preds = [self.bayes_train_preds, self.bayes_test_preds] = self.bayes_preds[:-test_size], self.bayes_preds[-test_size:]
  41.     def return_data(self):
  42.         return self.data
  43.     def return_train_test(self):
  44.         return self.X_train, self.y_train, self.X_test, self.y_test
  45. class ARData(TimeSeriesData):
  46.     """Class to generate autoregressive data."""
  47.     def __init__(self, *args, coeffs=None, **kwargs):
  48.         self.given_coeffs = coeffs
  49.         super(ARData, self).__init__(*args, **kwargs)
  50.         if coeffs is not None:
  51.             self.num_prev = len(coeffs) - 1
  52.     def generate_data(self):
  53.         self.generate_coefficients()
  54.         self.generate_initial_points()
  55.         # + 3*self.num_prev because we want to cut first (3*self.num_prev) datapoints later
  56.         # so dist is more stationary (else initial num_prev datapoints will stand out as diff dist)
  57.         for i in range(self.num_datapoints+3*self.num_prev):
  58.             # Generate y value if there was no noise
  59.             # (equivalent to Bayes predictions: predictions from oracle that knows true parameters (coefficients))
  60.             self.bayes_preds[i + self.num_prev] = np.dot(self.y[i:self.num_prev+i][::-1], self.coeffs)
  61.             # Add noise
  62.             self.y[i + self.num_prev] = self.bayes_preds[i + self.num_prev] + self.noise()
  63.         # Cut first 20 points so dist is roughly stationary
  64.         self.bayes_preds = self.bayes_preds[3*self.num_prev:]
  65.         self.y = self.y[3*self.num_prev:]
  66.     def generate_coefficients(self):
  67.         if self.given_coeffs is not None:
  68.             self.coeffs = self.given_coeffs
  69.         else:
  70.             filter_stable = False
  71.             # Keep generating coefficients until we come across a set of coefficients
  72.             # that correspond to stable poles
  73.             while not filter_stable:
  74.                 true_theta = np.random.random(self.num_prev) - 0.5
  75.                 coefficients = np.append(1, -true_theta)
  76.                 # check if magnitude of all poles is less than one
  77.                 if np.max(np.abs(np.roots(coefficients))) < 1:
  78.                     filter_stable = True
  79.             self.coeffs = true_theta
  80.     def generate_initial_points(self):
  81.         # Initial datapoints distributed as N(0,1)
  82.         self.y[:self.num_prev] = np.random.randn(self.num_prev)
  83.     def noise(self):
  84.         # Noise distributed as N(0, self.noise_var)
  85.         return self.noise_var * np.random.randn()
  86. # A set of coefficients that are stable (to produce replicable plots, experiments)
  87. fixed_ar_coefficients = {2: [ 0.46152873, -0.29890739],
  88.     5: [ 0.02519834, -0.24396899,  0.2785921,   0.14682383,  0.39390468],
  89.                         10: [-0.10958935, -0.34564819,  0.3682048,   0.3134046,  -0.21553732,  0.34613629,
  90.   0.41916508,  0.0165352,   0.14163503, -0.38844378],
  91.                          20: [ 0.1937815,   0.01201026,  0.00464018, -0.21887467, -0.20113385, -0.02322278,
  92.   0.34285319, -0.21069086,  0.06604683, -0.22377364,  0.11714593, -0.07122126,
  93.  -0.16346554,  0.03174824,  0.308584,    0.06881604,  0.24840789, -0.32735569,
  94.   0.21939492, 0.3996207 ]}

使用上面的代码生成我们需要的数据, 这样我们就有了训练和测试的时候需要的数据了。

  1. #####################
  2. # Generate data
  3. #####################
  4. num_datapoints = 100
  5. test_size = 0.2
  6. input_size = 20
  7. noise_var = 0
  8. data = ARData(num_datapoints, num_prev=input_size, test_size=test_size, noise_var=noise_var, coeffs=fixed_ar_coefficients[input_size])
  9. # --------------------
  10. # Device configuration
  11. # --------------------
  12. device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
  13. # data for train and test
  14. X_train = torch.from_numpy(data.X_train).float().to(device)
  15. X_test = torch.from_numpy(data.X_test).float().to(device)
  16. y_train = torch.from_numpy(data.y_train).float().to(device)
  17. y_test = torch.from_numpy(data.y_test).float().to(device)

 定义生成器(用于逐次生成训练的数据)

根据上面我们介绍的,我们需要每七天的数据作为一组,进入模型进行训练. 如我们需要逐次返回 :

  • 3/1-3/7的数据为训练集, 3/8为需要预测的;
  • 3/2-3/8的数据为训练集, 3/9为需要预测的
  • 3/3-3/9的数据为训练集, 3/10为需要预测的
  • ...

于是我们使用yield, 最后可以返回一个生成器.

  1. # 使用slide window, window size=7
  2. # 即返回 data=1-7 label=8; data=2-8 label=9
  3. def create_dataset(X_data, y_data, look_back=1):
  4.     length = X_data.size(0)
  5.     for i in range(0,length-look_back):
  6.         result = (X_data[i:(i+look_back),:],y_data[i+look_back])
  7.         yield result

简单测试一下,就是可以逐次返回训练的样本和target.

  1. test_data = create_dataset(X_train,y_train, look_back=7)
  2. for (i,j) in test_data:
  3.     print(i.shape,j)
GRU for Time Series in Pytorch--Pytorch实现时间序列分析

定义网络

接下来我们定义预测所需要的网络结构。

  1. class TimeSeries(nn.Module):
  2.     def __init__(self, input_dim, hidden_dim, batch_size, output_dim=1, num_layers=2):
  3.         super(TimeSeries,self).__init__()
  4.         self.input_dim = input_dim # 输入的特征的个数
  5.         self.hidden_dim = hidden_dim
  6.         self.batch_size = batch_size
  7.         self.num_layers = num_layers
  8.         self.output_dim = output_dim
  9.         # Defeine the GRU layer
  10.         self.gru = nn.GRU(self.input_dim, self.hidden_dim, self.num_layers, dropout=0.1)
  11.         # Define the output layer
  12.         self.fc = nn.Linear(self.hidden_dim, self.output_dim)
  13.     def init_hidden(self):
  14.         # initialise hidden state
  15.         return (torch.zeros(self.num_layers, self.batch_size, self.hidden_dim).to(device))
  16.         # return (torch.zeros(self.num_layers, self.batch_size, self.hidden_dim))
  17.     def forward(self,x):
  18.         self.batch_size = x.size(1)
  19.         self.hidden = self.init_hidden() # 初始化hidden state
  20.         gru_out, self.hidden = self.gru(x, self.hidden)
  21.         y_pred = self.fc(gru_out[-1])
  22.         return y_pred

简单测试一下网络的编写是否是正确的。

  1. # 模型的测试
  2. input_data = torch.from_numpy(np.random.randn(7,1,20)).float().to(device)
  3. model = TimeSeries(input_dim=20, hidden_dim=20, batch_size=1, output_dim=1, num_layers=2).to(device)
  4. model(input_data)
  5. """
  6. tensor([[-0.0021]], device='cuda:0', grad_fn=<AddmmBackward>)
  7. """

模型的训练

准备好模型和数据之后,我们就可以开始训练了。训练的代码如下所示:

  1. # 模型的初始化
  2. model = TimeSeries(input_dim=20, hidden_dim=28, batch_size=1, output_dim=1, num_layers=3).to(device)
  3. # 定义损失函数和优化器
  4. criterion = nn.MSELoss()
  5. optimizer = optim.Adam(model.parameters(),lr=0.001)
  6. # 模型的训练
  7. pre_output = y_train.clone()
  8. num_epochs = 1000
  9. hist = np.zeros(num_epochs) # 用来记录每一个epoch的误差
  10. for t in range(num_epochs):
  11.     test_data = create_dataset(X_train,y_train, look_back=7) # 这个要每次刷新
  12.     loss = torch.tensor([0.0]).float().to(device)
  13.     for num, (i,j) in enumerate(test_data):
  14.         optimizer.zero_grad()
  15.         # reset hidden states
  16.         model.hidden = model.init_hidden()
  17.         # forward
  18.         output = model(i.unsqueeze(1))
  19.         pre_output[num+7]=output
  20.         # backward
  21.         loss = loss + criterion(output, j)
  22.     loss.backward()
  23.     # optimize
  24.     optimizer.step()
  25.     # 打印与记录误差
  26.     if t%10 == 0:
  27.         print("Epoch:{:0>4d}, MSE:{:<10.9f}".format(t,loss.item()))
  28.         print('============')
  29.     hist[t] = loss.item()

基本都是常规的操作,有两个需要注意一下:

  • 不是每次都进行反向传播和系数的优化, 我们会把一个epoch所有的loss求和, 一起进行反向传播.
  • 还有一点要注意的是, create_dataset需要写在epoch的循环里面, 可以理解一下为什么.
GRU for Time Series in Pytorch--Pytorch实现时间序列分析

结果的可视化

在这里我没在测试集上做测试, 就直接看一下训练集的效果。首先看一下预测的值与实际值画的图 :

  1. # 打印预测和实际
  2. fig = plt.figure(figsize=(12,7))
  3. ax = fig.add_subplot(1,1,1)
  4. ax.plot(pre_output.cpu().detach().numpy(), label="Preds")
  5. ax.plot(y_train.cpu().detach().numpy(), label="Data")
  6. ax.legend()
GRU for Time Series in Pytorch--Pytorch实现时间序列分析

接着看一下训练过程中Loss的变化的过程 :

  1. # 打印Loss
  2. fig = plt.figure(figsize=(12,7))
  3. ax = fig.add_subplot(1,1,1)
  4. ax.plot(hist, label="Training loss")
  5. ax.legend()

大概在300个epoch的时候,就已经到较好的效果了。

GRU for Time Series in Pytorch--Pytorch实现时间序列分析

结语

上面是关于使用Pytorch来实现时间序列的预测,完整的notebook可以参考下面的链接。

RNN for Time Series in Pytorch

  • 微信公众号
  • 关注微信公众号
  • weinxin
  • QQ群
  • 我们的QQ群号
  • weinxin
王 茂南

发表评论

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen: