【RNN從入門到實戰】GRU入門到實戰——使用GRU預測股票。

摘要

GRU是LSTM網絡的一種效果很好的變體，它較LSTM網絡的結構更加簡單，而且效果也很好，因此也是當前非常流形的一種網絡。GRU既然是LSTM的變體，因此也是可以解決RNN網絡中的長依賴問題。

在LSTM中引入了三個門函數：輸入門、遺忘門和輸出門來控制輸入值、記憶值和輸出值。而在GRU模型中只有兩個門：分別是更新門和重置門。具體結構如下圖所示：

圖中的update gate和reset gate分別表示更新門和重置門。

更新門的作用類似于LSTM的遺忘和輸入門。它決定前一時刻的狀態信息哪些被丟棄和哪些被更新新信息，更新門的值越大說明前一時刻的狀態信息帶入越多。

重置門控制前一狀態有多少信息被寫入到當前的候選集

h

t

~

\tilde{h_{t}}

ht ~ 上，重置門越小，前一狀態的信息被寫入的越少。

詳見下圖：

GRU和LSTM的區別

GRU使用門控機制學習長期依賴關系的基本思想和 LSTM 一致，但還是有一些關鍵區別：

1、GRU 有兩個門（重置門與更新門），而 LSTM 有三個門（輸入門、遺忘門和輸出門）。

2、GRU 并不會控制并保留內部記憶（c_t），且沒有 LSTM 中的輸出門。

3、LSTM 中的輸入與遺忘門對應于 GRU 的更新門，重置門直接作用于前面的隱藏狀態。

4、在計算輸出時并不應用二階非線性。

如何選擇？

由于 GRU 參數更少，收斂速度更快，因此其實際花費時間要少很多，這可以大大加速了我們的迭代過程。 而從表現上講，二者之間孰優孰劣并沒有定論，這要依據具體的任務和數據集而定，而實際上，二者之間的 performance 差距往往并不大，遠沒有調參所帶來的效果明顯，我通常的做法，首先選擇GRU作為基本的單元，因為其收斂速度快，可以加速試驗進程，快速迭代，而我認為快速迭代這一特點很重要。如果實現沒其余優化技巧，才會嘗試將 GRU 換為 LSTM，看看有沒有變化。

實戰——使用GRU預測股票

制作數據集

數據的-：http://quotes.money.163.com/trade/lsjysj_zhishu_000001.html

首先選擇查詢數據的范圍，然后選擇下載數據：

點擊下載數據后，彈出個框，這個框里可以看到起止日期，數據的列，默認勾選全部，然后單擊下載。

將下載后的數據放到工程的根目錄下面，然后做一些處理。

第一步 讀入csv文件，編碼方式設置為gbk。

第二步 將日期列轉為“年-月-日”這樣的格式。

第三步 根據日期排序，數據默認是從降序，改為升序。

最后 保存到“sh.csv”中。

代碼如下：

import pandas as pd df = pd.read_csv("000001.csv", encoding='gbk') df["日期"] = pd.to_datetime(df["日期"], format="%Y-%m-%d") df.sort_values(by=["日期"], ascending=True) df = df.sort_values(by=["日期"], ascending=True) df.to_csv("sh.csv", index=False, sep=',')

到這里數據集制作完成了

讀入數據集，構建時序數據。

generate_df_affect_by_n_days函數，通過一個序列來生成一個矩陣（用于處理時序的數據）。就是把當天的前n天作為參數，當天的數據作為label。

readData中的文件名為：sh.csv ，參數column，代表選用的數據，本次預測只用了一列數據，列名就是column，參數n就是之前模型中所說的n，代表column前n天的數據。train_end表示的是后面多少個數據作為測試集。

import pandas as pd import matplotlib.pyplot as plt import datetime import torch import torch.nn as nn import numpy as np from torch.utils.data import Dataset, DataLoader def generate_df_affect_by_n_days(series, n, index=False): if len(series) <= n: raise Exception("The Length of series is %d, while affect by (n=%d)." % (len(series), n)) df = pd.DataFrame() for i in range(n): df['c%d' % i] = series.tolist()[i:-(n - i)] df['y'] = series.tolist()[n:] if index: df.index = series.index[n:] return df def readData(column='收盤價', n=30, all_too=True, index=False, train_end=-300): df = pd.read_csv("sh.csv", index_col=0, encoding='utf-8') df.fillna(0, inplace=True) df.replace(to_replace="None", value=0) del df["股票代碼"] del df["名稱"] df.index = list(map(lambda x: datetime.datetime.strptime(x, "%Y-%m-%d").date(), df.index)) df_column = df[column].copy() df_column_train, df_column_test = df_column[:train_end], df_column[train_end - n:] df_generate_from_df_column_train = generate_df_affect_by_n_days(df_column_train, n, index=index) print(df_generate_from_df_column_train) if all_too: return df_generate_from_df_column_train, df_column, df.index.tolist() return df_generate_from_df_column_train

構建模型

模型見RNN類，采用GRU+全連接。隱藏層設置64，GRU的輸出是64維的，所以設置全連接也是64。

TrainSet是數據讀取類，將輸出數據的最后一列做標簽，前面的列做輸入，類的寫法是Pytorch的固定模式。

class RNN(nn.Module): def __init__(self, input_size): super(RNN, self).__init__() self.rnn = nn.GRU( input_size=input_size, hidden_size=64, num_layers=1, batch_first=True, ) self.out = nn.Sequential( nn.Linear(64, 1), ) self.hidden = None def forward(self, x): r_out, self.hidden = self.rnn(x) # None 表示 hidden state 會用全0的 state out = self.out(r_out) return out class TrainSet(Dataset): def __init__(self, data): # 定義好 image 的路徑 self.data, self.label = data[:, :-1].float(), data[:, -1].float() def __getitem__(self, index): return self.data[index], self.label[index] def __len__(self): return len(self.data)

訓練與測試

n為模型中的n

LR是模型的學習率

EPOCH是多次循環

train_end這個在之前的數據集中有提到。（注意是負數）

n = 30

LR = 0.0001

EPOCH = 100

train_end = -500

loss選用mse

預測的數據選擇“收盤價”

n = 10 LR = 0.0001 EPOCH = 100 train_end = -500 # 數據集建立 df, df_all, df_index = readData('收盤價', n=n, train_end=train_end) df_all = np.array(df_all.tolist()) plt.plot(df_index, df_all, label='real-data') df_numpy = np.array(df) df_numpy_mean = np.mean(df_numpy) df_numpy_std = np.std(df_numpy) df_numpy = (df_numpy - df_numpy_mean) / df_numpy_std df_tensor = torch.Tensor(df_numpy) trainset = TrainSet(df_tensor) trainloader = DataLoader(trainset, batch_size=16, shuffle=True) rnn = RNN(n) optimizer = torch.optim.Adam(rnn.parameters(), lr=LR) # optimize all cnn parameters loss_func = nn.MSELoss() for step in range(EPOCH): for tx, ty in trainloader: output = rnn(torch.unsqueeze(tx, dim=0)) loss = loss_func(torch.squeeze(output), ty) optimizer.zero_grad() # clear gradients for this training step loss.backward() # back propagation, compute gradients optimizer.step() print(step, loss) if step % 10: torch.save(rnn, 'rnn.pkl') torch.save(rnn, 'rnn.pkl') generate_data_train = [] generate_data_test = [] test_index = len(df_all) + train_end df_all_normal = (df_all - df_numpy_mean) / df_numpy_std df_all_normal_tensor = torch.Tensor(df_all_normal) for i in range(n, len(df_all)): x = df_all_normal_tensor[i - n:i] x = torch.unsqueeze(torch.unsqueeze(x, dim=0), dim=0) y = rnn(x) if i < test_index: generate_data_train.append(torch.squeeze(y).detach().numpy() * df_numpy_std + df_numpy_mean) else: generate_data_test.append(torch.squeeze(y).detach().numpy() * df_numpy_std + df_numpy_mean) plt.plot(df_index[n:train_end], generate_data_train, label='generate_train') plt.plot(df_index[train_end:], generate_data_test, label='generate_test') plt.legend() plt.show() plt.cla() plt.plot(df_index[train_end:-400], df_all[train_end:-400], label='real-data') plt.plot(df_index[train_end:-400], generate_data_test[:-400], label='generate_test') plt.legend() plt.show()

總結

本實例數據集使用上證的收盤價，構建時序數據集，使用GRU對數據預測，從結果上來看，走勢有一定的滯后性，由于只是用了價格預測價格，還是太簡單了，可以考慮加入更多的特征去優化這一算法。

完整代碼

import pandas as pd import matplotlib.pyplot as plt import datetime import torch import torch.nn as nn import numpy as np from torch.utils.data import Dataset, DataLoader import os os.environ['KMP_DUPLICATE_LIB_OK'] = 'TRUE' def generate_df_affect_by_n_days(series, n, index=False): if len(series) <= n: raise Exception("The Length of series is %d, while affect by (n=%d)." % (len(series), n)) df = pd.DataFrame() for i in range(n): df['c%d' % i] = series.tolist()[i:-(n - i)] df['y'] = series.tolist()[n:] if index: df.index = series.index[n:] return df def readData(column='收盤價', n=30, all_too=True, index=False, train_end=-300): df = pd.read_csv("sh.csv", index_col=0, encoding='utf-8') df.fillna(0, inplace=True) df.replace(to_replace="None", value=0) del df["股票代碼"] del df["名稱"] df.index = list(map(lambda x: datetime.datetime.strptime(x, "%Y-%m-%d").date(), df.index)) df_column = df[column].copy() df_column_train, df_column_test = df_column[:train_end], df_column[train_end - n:] df_generate_from_df_column_train = generate_df_affect_by_n_days(df_column_train, n, index=index) print(df_generate_from_df_column_train) if all_too: return df_generate_from_df_column_train, df_column, df.index.tolist() return df_generate_from_df_column_train class RNN(nn.Module): def __init__(self, input_size): super(RNN, self).__init__() self.rnn = nn.GRU( input_size=input_size, hidden_size=64, num_layers=1, batch_first=True, ) self.out = nn.Sequential( nn.Linear(64, 1), ) self.hidden = None def forward(self, x): r_out, self.hidden = self.rnn(x) # None 表示 hidden state 會用全0的 state out = self.out(r_out) return out class TrainSet(Dataset): def __init__(self, data): # 定義好 image 的路徑 self.data, self.label = data[:, :-1].float(), data[:, -1].float() def __getitem__(self, index): return self.data[index], self.label[index] def __len__(self): return len(self.data) n = 10 LR = 0.0001 EPOCH = 100 train_end = -500 # 數據集建立 df, df_all, df_index = readData('收盤價', n=n, train_end=train_end) df_all = np.array(df_all.tolist()) plt.plot(df_index, df_all, label='real-data') df_numpy = np.array(df) df_numpy_mean = np.mean(df_numpy) df_numpy_std = np.std(df_numpy) df_numpy = (df_numpy - df_numpy_mean) / df_numpy_std df_tensor = torch.Tensor(df_numpy) trainset = TrainSet(df_tensor) trainloader = DataLoader(trainset, batch_size=64, shuffle=True) rnn = RNN(n) optimizer = torch.optim.Adam(rnn.parameters(), lr=LR) # optimize all cnn parameters loss_func = nn.MSELoss() for step in range(EPOCH): for tx, ty in trainloader: output = rnn(torch.unsqueeze(tx, dim=0)) loss = loss_func(torch.squeeze(output), ty) optimizer.zero_grad() # clear gradients for this training step loss.backward() # back propagation, compute gradients optimizer.step() print(step, loss) if step % 10: torch.save(rnn, 'rnn.pkl') torch.save(rnn, 'rnn.pkl') generate_data_train = [] generate_data_test = [] test_index = len(df_all) + train_end df_all_normal = (df_all - df_numpy_mean) / df_numpy_std df_all_normal_tensor = torch.Tensor(df_all_normal) for i in range(n, len(df_all)): x = df_all_normal_tensor[i - n:i] x = torch.unsqueeze(torch.unsqueeze(x, dim=0), dim=0) y = rnn(x) if i < test_index: generate_data_train.append(torch.squeeze(y).detach().numpy() * df_numpy_std + df_numpy_mean) else: generate_data_test.append(torch.squeeze(y).detach().numpy() * df_numpy_std + df_numpy_mean) plt.plot(df_index[n:train_end], generate_data_train, label='generate_train') plt.plot(df_index[train_end:], generate_data_test, label='generate_test') plt.legend() plt.show() plt.cla() plt.plot(df_index[train_end:-400], df_all[train_end:-400], label='real-data') plt.plot(df_index[train_end:-400], generate_data_test[:-400], label='generate_test') plt.legend() plt.show()

AI 神經網絡

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。