[PYTHON] Aktienkursprognose mit LSTM_1

"[PyTorch Neural Network Implementation Handbook](https://www.amazon.co.jp/PyTorch%E3%83%8B%E3%83%A5%E3%83%BC%E3%83%A9%E3%83" % AB% E3% 83% 8D% E3% 83% 83% E3% 83% 88% E3% 83% AF% E3% 83% BC% E3% 82% AF% E5% AE% 9F% E8% A3% 85 % E3% 83% 8F% E3% 83% B3% E3% 83% 89% E3% 83% 96% E3% 83% 83% E3% 82% AF-Python% E3% 83% A9% E3% 82% A4 % E3% 83% 96% E3% 83% A9% E3% 83% AA% E5% AE% 9A% E7% 95% AA% E3% 82% BB% E3% 83% AC% E3% 82% AF% E3 % 82% B7% E3% 83% A7% E3% 83% B3-% E5% AE% AE% E6% 9C% AC-% E5% 9C% AD% E4% B8% 80% E9% 83% 8E / dp / 4798055476) ”, Kapitel 5 RNN wurde gelesen, daher habe ich versucht, den Aktienkurs zu analysieren.

Es ist das Memorandum.

Einführung

Aus den Aktienkursen (Eröffnungskurs, Hochpreis, Niedrigpreis, Schlusskurs) der Toyota Motor Corporation (7203) für die letzten 20 Jahre haben wir vorausgesagt, ob die Rendite am nächsten Tag (Schlusskurs am nächsten Tag - Eröffnungskurs am nächsten Tag) 2-3,5% betragen wird ( Binäres Klassifizierungsproblem).

Der Grund für 2-3,5% ist, 1) die Mindestrendite zu sichern und 2) fundamentale Faktoren wie Aktienkursschwankungen aufgrund von Nachrichten zu ignorieren. Als ich zuvor die tägliche Rendite (Schlusskurs-Eröffnungskurs) von TOPIX500 analysierte, lag die Rendite von 2-3,5% bei etwa 5%, was genau richtig für die Vorhersage war.

Ergebnis Als ich die Daten der letzten 75 Tage als erklärende Variable vorhergesagt habe, habe ich eine korrekte Antwortrate von 97,42% markiert (← verdächtig, daher Überprüfung erforderlich).

Zur Implementierung habe ich auf das Buch am Anfang und diese Seite verwiesen. https://stackabuse.com/time-series-prediction-using-lstm-with-pytorch-in-python/

Vorbereitung

Importieren Sie zunächst die erforderlichen Bibliotheken.

import torch
import torch.nn as nn
import torch.optim as optim

%matplotlib inline
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

Da sich die Analysedaten in Google Drive befinden, stellen Sie sicher, dass Sie mit dem folgenden Code auf das Laufwerk zugreifen können.

from google.colab import drive
drive.mount('/content/drive')

Überprüfen Sie, ob cuda verwendet werden kann, und geben Sie das Gerät an.

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
device

Daten lesen

Dieses Mal werden wir die Aktienkursdaten der Toyota Motor Corporation (7203) verwenden (Daten für 27 Jahre seit 1983).

df_init = pd.read_csv('/content/drive/My Drive/XXXXXXXXXX/7203.csv', encoding='sjis')
df_init.head()

Selbst wenn zu viele Variablen vorhanden sind, 1) ist die Berechnungszeit lang und 2) besteht die Gefahr eines Überlernens. Daher konzentrieren wir uns dieses Mal nur auf die Variablen für den Eröffnungspreis, den hohen Preis, den niedrigen Preis und den Schlusskurs.

df = pd.DataFrame()
df['open'] = df_init['Offener Preis']
df['high'] = df_init['Hoher Preis']
df['low'] = df_init['Niedriger Preis']
df['close'] = df_init['Schlusskurs']
#Am nächsten Tag zurück(Schlusskurs-Offener Preis)Berechnen und 2-3.5%Setzen Sie das Flag auf 1, wenn.
df['return'] = (df_init['Schlusskurs'].shift() - df_init['Offener Preis'].shift())/df_init['Offener Preis'].shift()
df['return'] = ((df['return']>=0.02) & (df['return']<=0.035)).astype(int)
print(len(df))
print(sum(df['return']))
df.head()

Erstellen Sie Daten für die Zeitreihenanalyse. Dieses Mal werden wir die Daten der letzten 75 Tage (≓ 3 Monate) als erklärende Variable verwenden.

window = 75

def create_inout_sequences(in_data, in_label, window):
    out_seq = []
    out_label = [] 
    length = len(in_data)
    for i in range(window, length):
        tmp_data = in_data[i-window:i+1] / in_data[i,3]
        tmp_label = [in_label[i]]
        out_seq.append(torch.Tensor(tmp_data))
        out_label.append(torch.Tensor(tmp_label).type(torch.long))
    return out_seq, out_label

out_seq, out_label = create_inout_sequences(df.iloc[:,:4].values, df.iloc[:,4].values, window)

Geben Sie die Daten aus und prüfen Sie, ob es sich um die gewünschten Daten handelt.

print(len(out_seq))
print(out_seq[0])
print(out_label[0])

'''Ausgabe
8660
tensor([[1.0577, 1.0673, 1.0481, 1.0481],
        [1.0577, 1.0577, 1.0192, 1.0385],
        [1.0192, 1.0288, 1.0000, 1.0096],
        [1.0288, 1.0385, 1.0096, 1.0288],
        [1.0288, 1.0288, 1.0192, 1.0288],
~~ weggelassen ~~
        [1.0288, 1.0385, 1.0288, 1.0385],
        [1.0288, 1.0385, 1.0192, 1.0192],
        [1.0192, 1.0288, 1.0000, 1.0000],
        [1.0096, 1.0192, 1.0000, 1.0192],
        [1.0192, 1.0288, 1.0000, 1.0000]])
tensor([0])
'''

Teilen Sie die Daten in Training, Bewertung und Inferenz ein. Die Anzahl der einzelnen Daten ist angemessen. Zwischen den einzelnen Daten liegt ein Intervall von 100 Tagen (> 75 Tagen), damit sich die Daten nicht überschneiden.

x_train = out_seq[:5000]
x_valid = out_seq[5100:6000]
x_test = out_seq[6100:]
y_train = out_label[:5000]
y_valid = out_label[5100:6000]
y_test = out_label[6100:]

Modellieren

Erstellen Sie ein Modell bestehend aus Eingabe → LSTM → Vollständig verbundene Ebene. Da es sich um eine binäre Klassifizierung handelt, ist die Ausgabedimension "2". Da es sich um einen Versuch handelt, hat die Größe des Stapels oder der verborgenen Schicht keine tiefe Bedeutung.

input_size=4
batch_size = 32
hidden_layer_size=50
output_size=2

class LstmClassifier(nn.Module):
    def __init__(self, input_size, hidden_layer_size, output_size, batch_size):
        super().__init__()
        self.batch_size = batch_size
        self.hidden_layer_size = hidden_layer_size
        #lstm ist standardmäßig Batch_first=Falsch, also Charge_first=Wahr
        self.lstm = nn.LSTM(input_size, hidden_layer_size, batch_first=True)
        self.fc = nn.Linear(hidden_layer_size, output_size)
        self.softmax = nn.Softmax(dim=1)
        #Legen Sie den anfänglichen verborgenen Zustand und den Zellenzustand fest
        self.hidden_cell = (torch.zeros(1, self.batch_size, self.hidden_layer_size).to(device),
                            torch.zeros(1, self.batch_size, self.hidden_layer_size).to(device))

    def forward(self, input_seq):
        x = input_seq
        #LSTM verbreiten
        lstm_out, self.hidden_cell = self.lstm(x, self.hidden_cell)
        out = self.fc(self.hidden_cell[0])
        out = out[-1]
        return out


model = LstmClassifier(input_size, hidden_layer_size, output_size, batch_size)
model = model.to(device)
model

'''Ausgabe
LstmClassifier(
  (lstm): LSTM(4, 50, batch_first=True)
  (fc): Linear(in_features=50, out_features=2, bias=True)
  (softmax): Softmax(dim=1)
)
'''

Verwenden Sie die Kreuzentropie für die Verlustfunktion und Adam für die Optimierungsfunktion.

criterion = nn.CrossEntropyLoss()
optimiser = optim.Adam(model.parameters())

Lernen

Lassen Sie uns vorerst mit einer Anzahl von Epochen lernen, die auf 100 eingestellt sind.

Der Gradient wird für jede Epoche durch Trennen abgeschnitten. Da RNN jedoch über einen hohen Rechenaufwand verfügt, werden Zwischenergebnisse, die nicht mehr zur Reduzierung der Speichernutzung benötigt werden, durch Trennen gelöscht ([Referenz](https: /). /discuss.pytorch.org/t/runtimeerror-trying-to-backward-through-the-graph-a-second-time-but-the-buffers-have-already-been-freed-specify-retain-graph-true -wenn-beim-ersten-rückwärts-anrufen / 6795/3)).

num_epochs = 100
train_loss_list = []
train_acc_list = []
val_loss_list = []
val_acc_list = []

#Stoppen Sie die Backpropagation in der Mitte
def detach(states):
    return [state.detach() for state in states] 

#Kombinieren Sie Tensor
def cat_Tensor(data, i_batch, batch_size):
    for i, idx in enumerate(range(i_batch*batch_size, (i_batch+1)*batch_size)):
        #Erhöhen Sie die Abmessung
        tmp = torch.unsqueeze(data[idx], 0)
        if i==0:
            output = tmp
        else:
            output = torch.cat((output, tmp), 0)
    return output

for i_epoch in range(num_epochs):

    train_loss = 0
    train_acc = 0
    val_loss = 0
    val_acc = 0

    #train
    model.train()

    n_batch = len(x_train)//batch_size
    for i_batch in range(n_batch):
        seq = cat_Tensor(x_train, i_batch, batch_size)
        labels = cat_Tensor(y_train, i_batch, batch_size)
        labels = torch.squeeze(labels, 1)

        seq = seq.to(device)
        labels = labels.to(device)
        
        #Gradient zurücksetzen
        optimiser.zero_grad()
        #Stoppen Sie die Backpropagation in der Mitte. Fehler Gegenmaßnahmen
        model.hidden_cell = detach(model.hidden_cell)
        #Vorwärtsausbreitung
        outputs = model(seq)
        #Fehler bei der Weitergabe
        loss = criterion(outputs, labels)
        #Anhäufung von Fehlern
        train_loss += loss.item()
        train_acc += (outputs.max(1)[1] == labels).sum().item()
        #Backpropagation-Berechnung
        loss.backward()
        #Gewichtsaktualisierung
        optimiser.step()

    avg_train_loss = train_loss / n_batch
    avg_train_acc = train_acc / (n_batch*batch_size)

    #val
    model.eval()
    with torch.no_grad():
        n_batch = len(x_valid)//batch_size
        for i_batch in range(n_batch):
            seq = cat_Tensor(x_valid, i_batch, batch_size)
            labels = cat_Tensor(y_valid, i_batch, batch_size)
            labels = torch.squeeze(labels, 1)

            seq = seq.to(device)
            labels = labels.to(device)

            #Vorwärtsausbreitung
            outputs = model(seq)
            loss = criterion(outputs, labels)
            #Anhäufung von Fehlern
            val_loss += loss.item()
            val_acc += (outputs.max(1)[1] == labels).sum().item()

    avg_val_loss = val_loss / n_batch
    avg_val_acc = val_acc / (n_batch*batch_size)
    
    print ('Epoch [{}/{}], Loss: {loss:.4f}, val_loss: {val_loss:.4f}, Acc:{acc:.4f}, val_acc: {val_acc:.4f}' 
        .format(i_epoch+1, num_epochs, loss=avg_train_loss, val_loss=avg_val_loss, 
                acc=avg_train_acc, val_acc=avg_val_acc))
    
    train_loss_list.append(avg_train_loss)
    train_acc_list.append(avg_train_acc)
    val_loss_list.append(avg_val_loss)
    val_acc_list.append(avg_val_acc)


'''Ausgabe
Epoch [1/100], Loss: 0.1198, val_loss: 0.0632, Acc:0.9439, val_acc: 0.9743
Epoch [2/100], Loss: 0.1147, val_loss: 0.0609, Acc:0.9397, val_acc: 0.9743
Epoch [3/100], Loss: 0.1119, val_loss: 0.0590, Acc:0.9403, val_acc: 0.9743
Epoch [4/100], Loss: 0.1096, val_loss: 0.0569, Acc:0.9407, val_acc: 0.9743
Epoch [5/100], Loss: 0.1069, val_loss: 0.0557, Acc:0.9417, val_acc: 0.9754
Epoch [6/100], Loss: 0.1046, val_loss: 0.0544, Acc:0.9437, val_acc: 0.9754
Epoch [7/100], Loss: 0.1032, val_loss: 0.0525, Acc:0.9455, val_acc: 0.9799
Epoch [8/100], Loss: 0.1023, val_loss: 0.0507, Acc:0.9459, val_acc: 0.9799
Epoch [9/100], Loss: 0.1012, val_loss: 0.0500, Acc:0.9457, val_acc: 0.9788
Epoch [10/100], Loss: 0.0998, val_loss: 0.0486, Acc:0.9469, val_acc: 0.9799
~~ weggelassen ~~
Epoch [95/100], Loss: 0.0669, val_loss: 0.0420, Acc:0.9688, val_acc: 0.9888
Epoch [96/100], Loss: 0.0665, val_loss: 0.0419, Acc:0.9692, val_acc: 0.9888
Epoch [97/100], Loss: 0.0662, val_loss: 0.0419, Acc:0.9698, val_acc: 0.9888
Epoch [98/100], Loss: 0.0659, val_loss: 0.0419, Acc:0.9702, val_acc: 0.9888
Epoch [99/100], Loss: 0.0656, val_loss: 0.0419, Acc:0.9704, val_acc: 0.9888
Epoch [100/100], Loss: 0.0652, val_loss: 0.0417, Acc:0.9708, val_acc: 0.9888
'''

Lassen Sie uns visualisieren, ob Sie richtig lernen.

import matplotlib.pyplot as plt
%matplotlib inline

plt.figure()
plt.plot(range(num_epochs), train_loss_list, color='blue', linestyle='-', label='train_loss')
plt.plot(range(num_epochs), val_loss_list, color='green', linestyle='--', label='val_loss')
plt.legend()
plt.xlabel('epoch')
plt.ylabel('loss')
plt.title('Training and validation loss')
plt.grid()

plt.figure()
plt.plot(range(num_epochs), train_acc_list, color='blue', linestyle='-', label='train_acc')
plt.plot(range(num_epochs), val_acc_list, color='green', linestyle='--', label='val_acc')
plt.legend()
plt.xlabel('epoch')
plt.ylabel('acc')
plt.title('Training and validation accuracy')
plt.grid()

Inferenz

Ich werde versuchen, anhand von Daten, die ich nicht für Training und Bewertung verwendet habe, eine Vorhersage zu treffen.

model.eval()
with torch.no_grad():
    total = 0
    test_acc = 0
    
    n_batch = len(x_test)//batch_size
    for i_batch in range(n_batch):
        seq = cat_Tensor(x_test, i_batch, batch_size)
        labels = cat_Tensor(y_test, i_batch, batch_size)
        labels = torch.squeeze(labels, 1)
        
        seq = seq.to(device)
        labels = labels.to(device)

        outputs = model(seq)
        test_acc += (outputs.max(1)[1] == labels).sum().item()
        total += labels.size(0)
    print('Richtigkeit: {} %'.format(100 * test_acc / total)) 

'''Ausgabe
Richtigkeit: 97.421875 %
'''

Die Genauigkeit betrug 97,42%, und wir konnten sehr genaue Vorhersagen treffen. Ich halte es jedoch für zu teuer, daher möchte ich es später überprüfen.