[PYTHON] Prévision du cours de l'action avec LSTM_1

"[Manuel d'implémentation du réseau neuronal PyTorch](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) », chapitre 5 RNN, j'ai donc essayé d'analyser le cours de l'action.

C'est le mémorandum.

introduction

À partir des cours de l'action (cours d'ouverture, prix élevé, prix bas, cours de clôture) de Toyota Motor Corporation (7203) pour les 20 dernières années, nous avons prédit si le rendement du lendemain (cours de clôture du lendemain-cours d'ouverture du lendemain) serait de 2-3,5% ( Problème de classification binaire).

La raison de 2-3,5% est que 1) pour garantir le rendement minimum, et 2) pour ignorer les facteurs fondamentaux tels que les fluctuations du cours des actions dues aux nouvelles. De plus, lorsque j'ai analysé le rendement quotidien (cours de clôture-ouverture) de TOPIX500 auparavant, le rendement de 2-3,5% était d'environ 5%, ce qui était juste pour la prédiction.

résultat Lorsque j'ai prédit les données des 75 derniers jours comme variable explicative, j'ai marqué un taux de réponse correcte de 97,42% (← suspect, donc vérification requise).

Pour la mise en œuvre, je me suis référé au livre au début et à ce site. https://stackabuse.com/time-series-prediction-using-lstm-with-pytorch-in-python/

Préparation

Tout d'abord, importez les bibliothèques requises.

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

Puisque les données d'analyse sont sur Google Drive, assurez-vous que vous pouvez accéder au lecteur avec le code suivant.

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

Vérifiez si cuda peut être utilisé et spécifiez le périphérique.

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

Lire les données

Cette fois, nous utiliserons les données sur le cours de l'action de Toyota Motor Corporation (7203) (données sur 27 ans depuis 1983).

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

Même s'il y a trop de variables, 1) le temps de calcul sera long, et 2) il y a un risque de surapprentissage, donc cette fois nous nous concentrerons sur les variables uniquement pour le prix d'ouverture, le prix élevé, le prix bas et le cours de clôture.

df = pd.DataFrame()
df['open'] = df_init['Prix ouvert']
df['high'] = df_init['Prix élevé']
df['low'] = df_init['Bas prix']
df['close'] = df_init['le dernier prix']
#Retour le lendemain(le dernier prix-Prix ouvert)Calculer et 2-3.5%Définissez l'indicateur sur 1 quand.
df['return'] = (df_init['le dernier prix'].shift() - df_init['Prix ouvert'].shift())/df_init['Prix ouvert'].shift()
df['return'] = ((df['return']>=0.02) & (df['return']<=0.035)).astype(int)
print(len(df))
print(sum(df['return']))
df.head()

Créez des données pour l'analyse de séries chronologiques. Cette fois, nous utiliserons les données des 75 derniers jours (≓ 3 mois) comme variable explicative.

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)

Sortez les données et vérifiez si ce sont les données souhaitées.

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

'''production
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],
~~ Omis ~~
        [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])
'''

Divisez les données en formation, évaluation et inférence. Le nombre de chaque donnée est approprié. Il y a un intervalle de 100 jours (> 75 jours) entre chaque donnée afin que les données ne se chevauchent pas.

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:]

La modélisation

Construisez un modèle composé de Input → LSTM → Fully Connected Layer. Puisqu'il s'agit d'une classification binaire, la dimension de sortie est "2". Puisqu'il s'agit d'un essai, il n'y a pas de signification profonde à la taille du lot ou de la couche cachée.

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 utilise par défaut le batch_first=Faux, donc lot_first=Vrai
        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)
        #Définir l'état masqué initial et l'état de la cellule
        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
        #Propager LSTM
        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

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

Utilisez l'entropie croisée pour la fonction de perte et Adam pour la fonction d'optimisation.

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

Apprentissage

Pour le moment, effectuons l'apprentissage avec le nombre d'époques fixé à 100.

Le gradient est coupé avec detach pour chaque époque, mais comme RNN a une grande quantité de calcul, il semble que les résultats intermédiaires qui ne sont plus nécessaires pour réduire l'utilisation de la mémoire soient supprimés avec detach ([Reference](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 -quand-appel-en-arrière-la-première fois / 6795/3)).

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

#Arrêtez la rétropropagation au milieu
def detach(states):
    return [state.detach() for state in states] 

#Combiner Tensor
def cat_Tensor(data, i_batch, batch_size):
    for i, idx in enumerate(range(i_batch*batch_size, (i_batch+1)*batch_size)):
        #Augmenter la dimension
        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)
        
        #Réinitialiser le dégradé
        optimiser.zero_grad()
        #Arrêtez la rétropropagation au milieu. Contre-mesures d'erreur
        model.hidden_cell = detach(model.hidden_cell)
        #Propagation vers l'avant
        outputs = model(seq)
        #Erreur de propagation de retour
        loss = criterion(outputs, labels)
        #Accumulation d'erreur
        train_loss += loss.item()
        train_acc += (outputs.max(1)[1] == labels).sum().item()
        #Calcul de rétropropagation
        loss.backward()
        #Mise à jour du poids
        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)

            #Propagation vers l'avant
            outputs = model(seq)
            loss = criterion(outputs, labels)
            #Accumulation d'erreur
            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)


'''production
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
~~ Omis ~~
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
'''

Visualisons si vous apprenez correctement.

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()

inférence

Je vais essayer de faire une prédiction en utilisant des données que je n'ai pas utilisées pour la formation et l'évaluation.

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('précision: {} %'.format(100 * test_acc / total)) 

'''production
précision: 97.421875 %
'''

La précision était de 97,42% et nous avons pu faire des prédictions très précises. Cependant, j'estime que c'est trop cher, je voudrais donc le vérifier plus tard.