源碼或數(shù)據(jù)集請點(diǎn)贊關(guān)注收藏后評論區(qū)留言或者私信博主要

由于獨(dú)特的設(shè)計(jì)結(jié)構(gòu) LSTM適合于處理和預(yù)測時(shí)間序列中間隔和延遲非常長的重要事件。

LSTM是一種含有LSTM區(qū)塊（blocks）或其他的一種類神經(jīng)網(wǎng)絡(luò)，文獻(xiàn)或其他資料中LSTM區(qū)塊可能被描述成智能網(wǎng)絡(luò)單元，因?yàn)樗梢杂洃洸欢〞r(shí)間長度的數(shù)值，區(qū)塊中有一個(gè)gate能夠決定input是否重要到能被記住及能不能被輸出output

LSTM有很多個(gè)版本，其中一個(gè)重要的版本是GRU（Gated Recurrent Unit），根據(jù)谷歌的測試表明，LSTM中最重要的是Forget gate，其次是Input gate，最次是Output gate。

介紹完LSTM的基本內(nèi)容 接下來實(shí)戰(zhàn)通過LSTM來預(yù)測股市收盤價(jià)格

先上結(jié)果?

1：隨著訓(xùn)練次數(shù)增加損失函數(shù)的圖像如下 可以看出基本符合肘部方法 但是局部會(huì)產(chǎn)生突變

?

?

?2：預(yù)測結(jié)果如下 紅色的是預(yù)測值 藍(lán)色的是真實(shí)值 可以看出除了某幾個(gè)極值點(diǎn)正確率較高

?代碼如下文章來源地址http://www.zghlxwxcb.cn/news/detail-511500.html

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import torch
import torch.nn as nn
from torch.utils.data import DataLoader, Dataset
import os
os.environ["KMP_DUPLICATE_LIB_OK"]="TRUE"


def read_dataset(dataset_type):
    assert dataset_type == "train" or dataset_type == "test"
    df = pd.read_csv(stock_market_price_{}.csv'.format(dataset_type))  # 讀入股票數(shù)據(jù)
    data = np.array(df['close'])  # 獲取收盤價(jià)序列
    data = data[::-1]  # 反轉(zhuǎn)，使數(shù)據(jù)按照日期先后順序排列

    normalize_data = (data - np.mean(data)) / np.std(data)  # 標(biāo)準(zhǔn)化
    normalize_data = normalize_data[:, np.newaxis]  # 增加維度
    X, y = [], []
    for i in range(len(normalize_data) - time_step):
        _x = normalize_data[i:i + time_step]
        _y = normalize_data[i + time_step]
        X.append(_x.tolist())
        y.append(_y.tolist())
    # plt.figure()
    # plt.plot(data)
    # plt.show() # 以折線圖展示data
    return X, y


# 實(shí)驗(yàn)參數(shù)設(shè)置
time_step = 7    # 用前七天的數(shù)據(jù)預(yù)測第八天
hidden_size = 4  # 隱藏層維度
lstm_layers = 1  # 網(wǎng)絡(luò)層數(shù)
batch_size = 64  # 每一批次訓(xùn)練多少個(gè)樣例
input_size = 1   # 輸入層維度
output_size = 1  # 輸出層維度
lr = 0.05        # 學(xué)習(xí)率


class myDataset(Dataset):
    def __init__(self, x, y):
        self.x = x
        self.y = y

    def __getitem__(self, index):
        return torch.Tensor(self.x[index]), torch.Tensor(self.y[index])

    def __len__(self):
        return len(self.x)


class LSTM(nn.Module):
    def __init__(self, input_size, output_size, hidden_size, device):
        super(LSTM, self).__init__()
        self.input_size=input_size
        self.output_size=output_size
        self.hidden_size=hidden_size
        self.device=device
        def _one(a,b):
            return nn.Parameter(torch.FloatTensor(a,b).to(self.device))
        def _three():
            return(_one(input_size,hidden_size),
                   _one(hidden_size,hidden_size),
                   nn.Parameter(torch.zeros(hidden_size).to(self.device)))
        self.W_xi,self.W_hi,self.b_i=_three()
        self.W_xf, self.W_hf, self.b_f = _three()
        self.W_xo, self.W_ho, self.b_o = _three()
        self.W_xc, self.W_hc, self.b_c = _three()
        self.W_hq=_one(hidden_size,output_size)
        self.b_q=nn.Parameter(torch.zeros(output_size).to(self.device))
        self.params=[self.W_xi,self.W_hi,self.b_i,self.W_xf, self.W_hf, self.b_f, self.W_xo, self.W_ho, self.b_o,self.W_xc, self.W_hc, self.b_c,
                     self.W_hq,self.b_q]
        for param in self.params:
            if param.dim()==2:
                nn.init.xavier_normal_(param)





        

    def init_lstm_state(self, batch_size):
        return (torch.zeros((batch_size, self.hidden_size), device=self.device),
                torch.zeros((batch_size, self.hidden_size), device=self.device))
    

    def forward(self, seq):
        (H,C)=self.init_lstm_state(seq.shape[0])
        for step in range(seq.shape[1]):
            X=seq[:,step,:]
            I=torch.sigmoid((X@self.W_xi)+(H@self.W_hi)+self.b_i)
            F = torch.sigmoid((X @ self.W_xf) + (H @ self.W_hf) + self.b_f)
            O = torch.sigmoid((X @ self.W_xo) + (H @ self.W_ho) + self.b_o)
            C_tilda=torch.tanh(torch.matmul(X.float(),self.W_xc)+torch.matmul(H.float(),self.W_hc)+self.b_c)
            C=F*C+I*C_tilda
            H=O*torch.tanh(C)
        Y=(H@self.W_hq)+self.b_q
        return Y,(H,C)




X_train, y_train = read_dataset('train')
X_test, y_test = read_dataset('test')
train_dataset = myDataset(X_train, y_train)
test_dataset = myDataset(X_test, y_test)
train_loader = DataLoader(train_dataset, batch_size, shuffle=True)
test_loader = DataLoader(test_dataset, 1)

# 設(shè)定訓(xùn)練輪數(shù)
num_epochs = 50
device = torch.device('cuda') if torch.cuda.is_available() else torch.device('cpu')
hist = np.zeros(num_epochs)
model = LSTM(input_size, output_size, hidden_size, device)
# 定義優(yōu)化器和損失函數(shù)
optimiser = torch.optim.Adam(model.parameters(), lr=lr)  # 使用Adam優(yōu)化算法
loss_func = torch.nn.MSELoss(reduction='mean')  # 使用均方差作為損失函數(shù)
for epoch in range(num_epochs):
    epoch_loss = 0
    for i, data in enumerate(train_loader):
        X, y = data
        pred_y, _ = model(X.to(device))
        loss = loss_func(pred_y, y.to(device))
        optimiser.zero_grad()
        loss.backward()
        optimiser.step()
        epoch_loss += loss.item()
    print("Epoch ", epoch, "MSE: ", epoch_loss)
    hist[epoch] = epoch_loss
plt.plot(hist)
plt.show()

# 測試
model.eval()
result = []
for i, data in enumerate(test_loader):
    X, y = data
    pred_y, _ = model(X.to(device))
    result.append(pred_y.item())

plt.plot(range(len(y_test)), y_test, label="true_y", color="blue")
plt.plot(range(len(result)), result, label="pred_y", color="red")
plt.legend(loc='best')
plt.show()

到了這里，關(guān)于LSTM神經(jīng)網(wǎng)絡(luò)實(shí)現(xiàn)對股市收盤價(jià)格的預(yù)測實(shí)戰(zhàn)（python實(shí)現(xiàn) 附源碼 超詳細(xì))的文章就介紹完了。如果您還想了解更多內(nèi)容，請?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！