簡述

網上看到有人用Tensorflow寫了的但是沒看到有用pytorch寫的。
所以我就寫了一份。寫的過程中沒有參照任何TensorFlow版本的（因為我對TensorFlow目前理解有限），所以寫得比較簡單，看來來似乎也比較容易實現（歡迎各位大佬改進之后，發家致富，帶帶小弟hhh）。

效果

先簡單的看看效果（會有點夸張hhh）：

• 注意，我沒有用全部數據！！而是真的用的訓練集合來做的，下面的都是真實的…
• 不過，也沒那么夸張，后面有講解這幅圖。

基于：

我以前寫的幾篇文章。

• RNN代碼解釋pytorch
• 所用的數據比較多，一般網上沒辦法直接獲取。可以參照我以前寫一個方法中的方法二 如何下載滬深300歷史數據 當然也可以用tushare來實現（獲取稍微少一點的數據）~
• 主要依賴于模型的建立 【時序數據處理】pandas某些列由于n個數據導致的，通過Series生成Dataframe
• 【解決辦法】pandas畫出時序數據（股票數據）橫軸不是時間

項目描述

模型假設

我這里認為每天的滬深300的最高價格，是依賴于當天的前n天的滬深300的最高價。
然后用RNN的LSTM模型來估計（捕捉到時序信息）。
讓模型學會用前n天的最高價，來判斷當天的最高價。

思路很簡單，所以模型也很簡單~

導入包

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

讀取數據的函數

• 其中generate_df_affect_by_n_days函數，通過一個序列來生成一個矩陣（用于處理時序的數據）。就是把當天的前n天作為參數，當天的數據作為label。
• readData中的文件名為：sh.csv 也可以是其他的文件大致類似的。自己可以修改。參數n就是之前模型中所說的n。train_end表示的是后面多少個數據作為測試集。

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 dfdef readData(column='high', n=30, all_too=True, index=False, train_end=-300):df = pd.read_csv("sh.csv", index_col=0)df.index = list(map(lambda x: datetime.datetime.strptime(x, "%Y-%m-%d"), 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)if all_too:return df_generate_from_df_column_train, df_column, df.index.tolist()return df_generate_from_df_column_train

類（RNN和數據讀取類）

• RNN的類很簡單，看模型也知道其實就兩部分
• 數據讀取類，這個主要為了滿足pytorch習慣而設定的（建議大家也要養成這樣的習慣）。

class RNN(nn.Module):def __init__(self, input_size):super(RNN, self).__init__()self.rnn = nn.LSTM(input_size=input_size,hidden_size=64,num_layers=1,batch_first=True)self.out = nn.Sequential(nn.Linear(64, 1))def forward(self, x):r_out, (h_n, h_c) = self.rnn(x, None) # None 表示 hidden state 會用全0的 stateout = self.out(r_out)return outclass 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

模型訓練

其實前面都是準備工作
這里就正式開始了。

獲取數據

• 訓練模型仍然使用minibatch的思路
• 注意，模型必須要先把數據標準化，不然損失會很難降低下來。

# 數據集建立 df, df_all, df_index = readData('high', 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=10, shuffle=True)

訓練模型部分

• 這個部分如果是不想要訓練的話（比如已經訓練好了模型），替換為rnn = torch.load('rnn.pkl') （記得把原來的注釋掉）

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 steploss.backward() # back propagation, compute gradientsoptimizer.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_enddf_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()

畫圖的結果就是：

• 這里，我在測試集合上，也是使用真實的當天的前n天的真實數據來作為輸入。表面上看起來有點奇怪，但是這是可以理解的。現實生活中是可以做到的。
• 但是這幅圖的話，太密了，下面我會畫一張其中的片段的圖片。
• 這是放大之后的：
  

會發現出現有一定的滯后效應。不過也很正常啊，輸入就只有價格…來預測價格… 這個模型的解釋力真的不太行…

完整代碼

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, DataLoaderdef 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 dfdef readData(column='high', n=30, all_too=True, index=False, train_end=-300):df = pd.read_csv("sh.csv", index_col=0)df.index = list(map(lambda x: datetime.datetime.strptime(x, "%Y-%m-%d"), 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)if all_too:return df_generate_from_df_column_train, df_column, df.index.tolist()return df_generate_from_df_column_trainclass RNN(nn.Module):def __init__(self, input_size):super(RNN, self).__init__()self.rnn = nn.LSTM(input_size=input_size,hidden_size=64,num_layers=1,batch_first=True)self.out = nn.Sequential(nn.Linear(64, 1))def forward(self, x):r_out, (h_n, h_c) = self.rnn(x, None) # None 表示 hidden state 會用全0的 stateout = self.out(r_out)return outclass 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 = 30 LR = 0.0001 EPOCH = 100 train_end = -500 # 數據集建立 df, df_all, df_index = readData('high', 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=10, shuffle=True)# rnn = torch.load('rnn.pkl')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 steploss.backward() # back propagation, compute gradientsoptimizer.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_enddf_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()

總結

以上是生活随笔為你收集整理的LSTM实现股票预测--pytorch版本【120+行代码】的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。