複数時系列データを扱うLSTMの実装について

Question

イタリックテキスト### 前提
LSTMの勉強のため、pythonでLSTMを用いて株価の予測を行うシステムを作成しています。
100日分のデータの内,80日分を学習データとして20日分の予測を行うシステムは完成しました。
様々なLSTMの実装を参考にしましたが、「どれも一つの時系列データの前半部分を学習に使って、後半部分を予測する」というのが趣旨でした。
しかし、僕は「A社,B社,C社...と複数の時系列データを100日分学習させて、ではZ社はどういう株価の推移をするのか,説明変数を与えて1～100日目まで全部予想する」というシステムを作成したいです。
そもそもこのような,複数の時系列データを学習させて,説明変数のみで0から予測するシステムをLSTMで作るのは可能なのでしょうか。
可能であるならば,どのような点を変更すれば良いのか。
ご教授お願いします。
初めての質問なので至らぬ点があればお願いします。

また、今回使用した入力データは80(時間)*129(説明変数),
出力データの次元は20(時間)*1(目的変数)となっています。
LSTMはpytorchで実装しました。

実現したいこと

t=1→80までを学習データ,t=81→100テストデータとするのではなく,
複数の時系列データのt=1→100を学習データとし,別の時系列データをt=1→100まで予測するモデルの作成

###参考にしたサイト
PyTorchを使ってLSTMでコロナ陽性者数を予測してみる
https://qiita.com/tsubauaaa/items/8411a22465811ec2ee11

pytorch で LSTM に入門したい...したくない？
https://hilinker.hatenablog.com/entry/2018/06/23/204910

該当のソースコード

python
ソースコード
import torch
import torch.nn as nn
import torch.optim as optim
import matplotlib.pyplot as plt
import pandas as pd
import numpy as np
%matplotlib inline
device=torch.device('cpu')

#基本情報
seq_length = 30# 1データの予測に用いる過去フレームは40とする
test_data_size = 20# 直近20frameをテストデータにする
pred_frames = 20# 予測するフレーム
epochs_num = 1#バッチを何回回すか
batch_size = 10

path="D:\最強データ\喜び_13～20\Actor_1\合体13.csv"
df=pd.read_csv(path,index_col=0,header=0)
#要らない項目の削除
df.pop("AU01_r")
df.pop("AU04_r")
df.pop("AU05_r")
df.pop("AU07_r")
df.pop("AU12_r")
df.pop("AU15_r")
df.pop("AU23_r")

data = df.values.astype(float)
data_length=data.shape[0]
data.shape

#学習データとテストデータ
train_data = data[:-test_data_size]
test_data = data[-test_data_size:]

#学習データとテストデータを最小値0と最大値1の範囲で正規化してTensor型にする
from sklearn.preprocessing import MinMaxScaler
scaler = MinMaxScaler(feature_range=(0, 1))
train_data_normalized = scaler.fit_transform(train_data)
train_data_normalized = torch.FloatTensor(train_data_normalized)

test_data_normalized = scaler.fit_transform(test_data)
test_data_normalized = torch.FloatTensor(test_data_normalized)

#シーケンスに沿ったデータを作成する関数
def make_sequence_data(input_data, sequence_length):
data_length = len(input_data) # 全体のデータ数取得
seq_data=[]#説明変数が入った時系列データを入れるリスト
target_data=[]#seq_dataの目的変数を入れるリスト
# 全体からシーケンス分引いた数までループする
for i in range(data_length - sequence_length):
# 1個ずらして、シーケンス分のデータを取得していく
seq = input_data[i:i+sequence_length]
target = input_data[:,128][i+sequence_length:i+sequence_length+1]
seq_data.append(seq)
target_data.append(target)
return seq_data, target_data

seq,labels = make_sequence_data(train_data_normalized, seq_length)

#LSTMmodelを作成してインスタンス生成&損失関数と最適化関数を定義
class LSTM(nn.Module):
def init(self, input_size=129, hidden_layer_size=100, output_size=1):
super().init()
self.hidden_layer_size = hidden_layer_size
self.lstm = nn.LSTM(input_size=input_size, hidden_size=hidden_layer_size, batch_first=True)
self.linear = nn.Linear(in_features=hidden_layer_size, out_features=output_size)

def forward(self, x):
    # LSTMのinputは(batch_size, seq_len, input_size)にする
    # LSTMのoutputは(batch_size, seq_len, hidden_layer_size)となる
    # hidden stateとcell stateにはNoneを渡して0ベクトルを渡す
    lstm_out, (hn, cn) = self.lstm(x, None)
    # Linearのinputは(N,∗,in_features)にする
    prediction = self.linear(lstm_out[:, -1, :])
    #[:,-1,:]で各バッチの最終時刻の特徴量の行が取れる
    return prediction

model = LSTM()
model.to(device)
#損失関数と最適化関数を定義
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

#バッチを作成する関数
losses = []
training_size=data.shape[0]-seq_length-test_data_size#学習データの数
def mkRandomBatch(train_x,train_y, batch_size,training_size,seq_length):
batch_x=torch.zeros(training_size, seq_length, 129, dtype=torch.float)
batch_y=torch.zeros(training_size, 1, dtype=torch.float)
batch_x_list=[]
batch_y_list=[]
ramdom_list=[]#ランダムな数がどう発生したのか記録するリスト

for i in range(batch_size):
    idx = np.random.randint(0, len(train_x) - 1)
    batch_x_list.append(train_x[idx])
    batch_y_list.append(train_y[idx])
   
    
    batch_x[i]=batch_x_list[i]
    batch_y[i]=batch_y_list[i] 
    ramdom_list.append(idx)

return batch_x,batch_y

batch_seq, batch_labels = mkRandomBatch(seq,labels,batch_size,training_size,seq_length)

#エポックを回して学習させてる
for epoch in range(epochs_num):
for i in range(int(training_size / batch_size)):
optimizer.zero_grad()#勾配初期化
batch_seq, batch_labels = mkRandomBatch(seq,labels,batch_size,training_size,seq_length)
y_pred = model(batch_seq)
single_loss = criterion(y_pred, batch_labels)
single_loss.backward()
optimizer.step()
losses.append(single_loss.item())
print(f'epoch: {i}, loss : {single_loss.item()}')

test_inputs = train_data_normalized[-seq_length:].tolist()

#予測するフェーズ
model.eval()# モデルを評価モードとする
test_outputs = []# 予測値を入れるリスト
for i in range(pred_frames):
seq = torch.FloatTensor(test_inputs[-seq_length:])
seq = torch.unsqueeze(seq, 0)
seq = seq.to(device)
with torch.no_grad():
test_inputs.append(test_data_normalized.tolist()[i])#test_inputにどんどん追加
test_outputs.append(model(seq).item())

#予測結果の整形
#列方向に同じ値を追加して(20, 129)にする
np_test_outputs = np.array(test_outputs).reshape(-1,1)
np_test_outputs2 = np.hstack((np_test_outputs, np_test_outputs))
np_test_outputs3 = np.hstack((np_test_outputs2, np_test_outputs))
np_test_outputs4 = np.hstack((np_test_outputs3, np_test_outputs))
np_test_outputs5 = np.hstack((np_test_outputs4, np_test_outputs))
#....長いので省略
np_test_outputs128 = np.hstack((np_test_outputs127, np_test_outputs))
np_test_outputs129 = np.hstack((np_test_outputs128, np_test_outputs))

#予測値test_outputsを正規化したデータから元のデータに戻した配列
actual_predictions = scaler.inverse_transform(np_test_outputs129)

#予測結果グラフ表示のための準備-------------------------
#testdataの時間を1ずつ並べた配列
x = np.arange(data_length-pred_frames, data_length, 1)

#予測結果表示ーーーーーーーーーーーーーーーーーーーーー
fig_size = plt.rcParams['figure.figsize']#グラフのサイズを決める変数
fig_size[0] = 5
fig_size[1] = 5
plt.rcParams['figure.figsize'] = fig_size

plt.title('Number of AU6')#グラフタイトル
plt.ylabel('AU6')#y軸の値
plt.grid(True)#グラフにグリッド線を引く
plt.autoscale(axis='x', tight=True)#グラフの自動スケーリング

a=df['AU06_r']
b=a.values
plt.plot(b,label='jissokuti')#実測値の値を持ってくる
plt.plot(x, actual_predictions[:,-1], label='yosokuti')#xの範囲にactual_predictions[:,-1]の要素を配置する
plt.legend()#凡例を表示してくれる
plt.show()

試したこと

一つの時系列データを学習データとテストデータに分けて,予想を行うLSTMの実装

Answer 1

A社,B社,C社...と複数の時系列データを100日分学習させて、ではZ社はどういう株価の推移をするのか,説明変数を与えて1～100日目まで全部予想する

精度はさておき，実装は可能です．
これをやるためには，説明変数→株価推移というモデルを作成することになりますので，A社,B社,C社...と複数の会社の説明変数を用意できれば良いでしょう．

簡単のため，Tensorflow - Kerasで三角波のパラメータを説明変数，三角波をLSTMで出力するモデルを作成してみました．説明変数は「時刻，周波数x2，位相x2，バイアス」の計6つを予測したい波形長(今回は256)並べて(256, 6)とした入力を与え，長さ256の波形を得るものにしました．

import os
import numpy as np
from sklearn.model_selection import train_test_split

from matplotlib import pyplot as plt

import tensorflow as tf
from tensorflow.python.keras.layers import Conv1D, Reshape, LSTM, Dense, Input
from tensorflow.python.keras.models import Model

# 擬似データ作成
def wave(t, f1, p1, f2, p2, bias):
  y = np.sin(f1 * t + p1) + np.cos(f2 * t + p2) + bias + np.random.randn(*t.shape) * 0.01
  return (y - y.min()) / (y.max() - y.min())

def make_data():
  x, y = list(), list()
  for i in range(2 ** 14):
    f1, f2 = np.random.randn(2) + np.pi
    p1, p2 = np.random.randn(2) * np.pi
    bias = (np.random.randn(1) + 0.1)[0]
    t = np.linspace(0, 2 * np.pi, 256)
    x.append([[_t, f1, p1, f2, p2, bias] for _t in t])
    y.append(wave(t, f1, p1, f2, p2, bias))
  return np.array(x), np.array(y)

# データ読み込み
x, y = make_data()
x_train, x_valid, y_train, y_valid = train_test_split(x, y, test_size = 0.25)

# モデル作成
inputs = Input(shape = (256, 6))
x = LSTM(32, return_sequences = True)(inputs)
x = Conv1D(1, 3, padding = "same", activation = "linear")(x)
model = Model(inputs, x)

# モデル実行
model.compile(
    optimizer = "adam",
    loss = "mse",
    metrics = ["mae"]
)
model.fit(
    x_train, y_train,
    epochs = 64,
    batch_size = 32,
    validation_data = (x_valid, y_valid)
)

# 結果の確認
pred = model.predict(x_valid)
for i in range(10):
  t = x_valid[i][:, 0]
  plt.figure(figsize = (8, 4))
  plt.plot(t, pred[i], label = "predict")
  plt.plot(t, wave(t, *x_valid[i][0, 1:]), label = "grandtruth")
  plt.ylim([0, 1])
  plt.grid()
  plt.legend()
  plt.show()

うまくフィットしない波形はあるものの，概ね説明変数から波形を予測できていることがわかります．実験した感じ，LSTMには予測したい時刻を必ず与える必要があるように感じました．

Comments

[aiue_minami]

回答ありがとうございます。
基礎的な部分の質問なのですが、「A社,B社,C社...と複数の会社の説明変数を用意できれば良いでしょう．」ということは
A,Bの会社の株価の推移からC社の株価を予測するモデルはそれぞれの資本金,従業員数,株価を説明変数とすると
カラム
A資本金,A従業員数,A株価,B資本金,B従業員数,B株価,C資本金,C従業員数,C株価
インデックス
1日目
…
100日目
みたいな行列がLSTMへ入力するデータになるのでしょうか(表が使えなかったためこの様な分かりにくい伝え方で大変申し訳ないです...)

[ps_aux_grep]

そうですね，正しくは会社ごとにデータを分けて
銘柄A
1日目: 時刻,A資本金,A従業員数,A株価
2日目: 時刻,A資本金,A従業員数,A株価
...
99日目: 時刻,A資本金,A従業員数,A株価
100日目: 時刻,A資本金,A従業員数,A株価

銘柄B
1日目: 時刻,B資本金,B従業員数,B株価
2日目: 時刻,B資本金,B従業員数,B株価
...
99日目: 時刻,B資本金,B従業員数,B株価
100日目: 時刻,B資本金,B従業員数,B株価

みたいにして，1つの教師データにつき，入力が2次元配列(時刻, 説明変数)，出力(株価推移,)が1次元配列となるデータセットを形成できれば良いと思います．

[aiue_minami]

ありがとうございます。
おかげさまでで入力データの外形がつかめました
何度も申し訳ありません
ソースコードの上にあった図の見方を教えていただけると助かります。
回答者様のプログラムについての質問なのですが、ランダムに生成した周波数や位相を元にsin波とcos波を生成して2の14乗個の三角波をつくる。
それらのうち3/4個を学習に回して1/4個を0から完全に予想しているということでしょうか？

[ps_aux_grep]

> ランダムに生成した周波数や位相を元にsin波とcos波を生成して2の14乗個の三角波をつくる。それらのうち3/4個を学習に回して1/4個を0から完全に予想しているということ

はい．それで間違いありません．ただ，0番目のLSTMユニットにパラメータを与えただけ，というわけではなく，それ以降未来のすべてのLSTMユニットにこれらパラメータ(今回は同一値ですが実際は時間で変化する値)を与えているので，実際に未知の波形を生成するために必要な同時刻のパラメータを集めたころには，その時刻以前の株価も出てしまっていると思います．したがって旧来の時刻tまでの波形→時刻t以降の波形予測のタスクの方が高精度になりそうな気がします．言ってしまえば今回のタスクは時刻0からtまでの波形→時刻0からtまでの波形みたいなタスクではあります．続きを読む ∨