ChatGPTを使ってpytorchでRNNを実装して株価の予測してみた。コード実装編

import torch
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset, Dataset
import numpy as np
import pandas as pd
from sklearn.preprocessing import MinMaxScaler
import matplotlib.pyplot as plt

data = pd.read_csv(".csv")
X = data["DATE"]
Y = data[""]
data["DATE"] = data["DATE"].str.replace("-", "/")
data = data.to_string(header=False, index=False)

# データの整形
data = [line.split() for line in data.strip().split('\n')]
dates, values = zip(*data)
values = list(map(float, values))

# データの正規化（0から1の範囲に変換）
scaler = MinMaxScaler()
values_normalized = scaler.fit_transform(np.array(values).reshape(-1, 1))
values_normalized = values_normalized.flatten()

# RNNモデルの定義
class SimpleRNN(nn.Module):
    def __init__(self, input_size, hidden_size, output_size):
        super(SimpleRNN, self).__init__()
        self.hidden_size = hidden_size
        self.rnn = nn.RNN(input_size, hidden_size, batch_first=True)
        self.fc = nn.Linear(hidden_size, output_size)

    def forward(self, x):
        out, _ = self.rnn(x)
        out = self.fc(out[:, -1, :])
        return out

train_size = int(len(values_normalized) * 0.7)
test_size = len(values_normalized) - train_size
train = values_normalized[:train_size]
test = values_normalized[train_size:]

sequence_length = 10
n_sample = train_size - sequence_length - 1 
input_sequences = np.zeros((n_sample, sequence_length, 1))
target_values = np.zeros((n_sample, 1))

for i in range(n_sample):
    input_sequences[i] = values_normalized[i:i+sequence_length].reshape(-1, 1)
    target_values[i] = values_normalized[i+sequence_length:i+sequence_length+1]

input_sequences = torch.tensor(input_sequences, dtype=torch.float).to(device)
target_values2 = torch.tensor(target_values, dtype=torch.float).to(device)
dataset = TensorDataset(input_sequences, target_values2)
train_loader = DataLoader(dataset, batch_size=4, shuffle=True)

n_sample_test = test_size - sequence_length - 1
test_data = np.zeros((n_sample_test, sequence_length, 1))
correct_test_data = np.zeros((n_sample_test, 1))

for i in range(n_sample_test):
    test_data[i] = test[i : i+sequence_length].reshape(-1, 1)
    correct_test_data[i] = test[i+sequence_length : i+sequence_length+1]

test_sequences = torch.tensor(test_data, dtype=torch.float).to(device)
target_values3 = torch.tensor(correct_test_data, dtype=torch.float).to(device)
dataset_test = TensorDataset(test_sequences, target_values3)
test_loader = DataLoader(dataset_test, batch_size=4, shuffle=False)


input_size = 1
hidden_size = 64
output_size = 1
model = SimpleRNN(input_size, hidden_size, output_size).to(device)

loss_fnc = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
loss_list = []
loss_list_test = []
epochs = 2000

for i in range(epochs+1):
    model.train()
    running_loss =0.0
    running_loss_test =0.0
    for j, (x, t) in enumerate(train_loader):
        optimizer.zero_grad()
        y = model(x)
        loss = loss_fnc(y, t)
        loss.backward()  #逆伝番        
        optimizer.step()   
        running_loss += loss.item() 
    running_loss /= j+1
    loss_list.append(running_loss)

    for l, (x_test, t_test) in enumerate(test_loader):
        y_test = model(x_test)
        loss_test = loss_fnc(y_test, t_test)
        running_loss_test += loss_test.item()
    running_loss_test /= l+1
    loss_list_test.append(running_loss_test)

    if i%10 == 0:
        print('Epoch:', i, 'Loss_Train:', running_loss, 'Loss_test:', running_loss_test)
        input_train = list(input_sequences[0].reshape(-1))
        predicted_train_plot = []
        model.eval()
        for k in range(n_sample):
            x = torch.tensor(input_train[-sequence_length:]).to(device)
            x = x.reshape(1, sequence_length, 1)
            x = x.float()
            y = model(x)
            if k <= n_sample-2: 
                input_train.append(input_sequences[k+1][sequence_length-1].item())
            predicted_train_plot.append(y[0].item())

        plt.plot(range(len(values_normalized)), values_normalized, label='Correct')
        plt.plot(range(sequence_length, sequence_length+len(predicted_train_plot)), predicted_train_plot, label='Predicted')
        plt.legend()
        plt.show()

plt.plot(range(len(loss_list)), loss_list, label='train')
plt.plot(range(len(loss_list_test)), loss_list_test, label='test')
plt.legend()
plt.xlabel("epochs")
plt.ylabel("loss")
plt.show()

input_test = list(test_data[0].reshape(-1))
predicted_test_plot = []
model.eval()
for k in range(n_sample_test):
    x = torch.tensor(input_test[-sequence_length:])
    x = x.reshape(1, sequence_length, 1)
    x = x.float()
    y = model(x)
    if k <= n_sample_test-2: 
         input_test.append(test_data[k+1][sequence_length-1].item())
    predicted_test_plot.append(y[0].item())

plt.plot(range(len(values_normalized)), values_normalized, label='Correct')
plt.plot(range(sequence_length+train_size, sequence_length+len(predicted_test_plot)+train_size), predicted_test_plot , label='Predicted')
plt.legend()
plt.show()

# 未来の値を予測
future_sequence = torch.FloatTensor(values_normalized[-sequence_length:]).view(1, -1, 1)
predicted_value_normalized = model(future_sequence).item()

# 予測値を元のスケールに戻す
predicted_value = scaler.inverse_transform(np.array([[predicted_value_normalized]]))

print(f'Predicted Value: {predicted_value[0][0]:.2f}')

fig, ax = plt.subplots()
ax.plot(range(len(values_normalized)),Y)
ax.scatter(len(values_normalized) + 1, predicted_value, color="red")