株価予測の基礎 RNN/LSTMによる時系列データ

https://aiacademy.jp/texts/show/?context=subject-nn_deep&id=48

①sin波の計算及びsin波にノイズをのせる

Sin波＝正弦波

https://www.google.com/url?sa=t&rct=j&q=&esrc=s&source=web&cd=&cad=rja&uact=8&ved=2ahUKEwjFgZjClJbvAhXYc3AKHXmeD1kQFjABegQIAhAD&url=http%3A%2F%2Fwww.wakariyasui.sakura.ne.jp%2Fp%2Fwave%2Fhadou%2Fseigennha.html&usg=AOvVaw0HgKEciI6DI1KFMg5mKm9u　　　　など

# パラメータの定義
random.seed(0)
# 乱数の係数
random_factor = 0.05
# サイクルあたりのステップ数
steps_per_cycle = 80
# 生成するサイクル数
number_of_cycles = 50
# 入力の長さ
length_of_sequences = 100
# 入力値・出力値の次元の大きさ
in_out_neurons = 1
# 隠れ要素のニューロン数
hidden_neurons = 300

乱数セットの設定

# 乱数のシード値固定
random.seed(0)

0であれば0、100であれば100の乱数が生成

0から4000までのsinの各値を入れる箱(セル)を作成

np_ary = np.arange(steps_per_cycle＜サイクルあたりのステップ数＞ * number_of_cycles ＜サイクル数＞+ 1)　　＜＝4000個のデータ＞
df = pd.DataFrame(np_ary, columns=["x"]＜カラム名（列名）＞)

 

作ったデータフレームの箱(セル)の中にsinのステップ数に応じたデータを入力

pi_t ＜円をsteps_per_cycleで分割した値＞

= 2 * math.pi / steps_per_cycle

df["sin(x)"] 　＜←pandasのカラムを事前に作らずとも指定＞

= df.x.apply　＜←データフレームのカラム「x」の各セルを()の中に入力する＞(lambda t:＜←入力されるtを:から定義している＞ 

＜sinθのθの中にこれらの値＋ノイズ＞

math.sin(t * pi_t +

random.uniform(-1.0, +1.0) ＜小数点を含むA（-1.0)以上B(+1.0)以下の乱数を生成＞

* random_factor<ノイズの大きさを調整>))

作ったデータフレームをグラフに出力

df"sin(x)".head(steps_per_cycle * 2<sinの2周期分のsin波のデータを取得> ).plot()<その値をグラフに>
plt.show()

②入力データ及び正解データそれぞれのリストを作成する。

データを学習モデルに入れられる形にする

正解データの作成

def Input_Ans_Extract(data, input_num = 100):

# 空リスト作成
InputList, AnsList = ,

# 0から(len(data)- input_num)だけの整数が配列としてある
for i in range(len(data) - input_num):#3501個のデータ分

#doc配列にi個目からi+100番目のデータを行列の形にして入れていく
# InputData = data.iloc[i:i+input_num].as_matrix()#3501~3601
InputData = data.iloc[i:i+input_num].values#3501~3601

# AnsData = data.iloc[i+input_num].as_matrix()
AnsData = data.iloc[i+input_num].values
InputList.append(InputData)
AnsList.append(AnsData)

# numpyで扱えるように行列をその形にして再定義
InputList_np = np.array(InputList)
AnsList_np = np.array(AnsList)
return InputList_np, AnsList_np

用意した空のリスト

に

numpy形式の各行列データを入力

values

を使うことでそのデータの形式をnumpyの行列に

③ データを学習用とテスト用に分ける。

def Data_Split(df, test_size=0.1＜１割＞, input_num = 100):
train_size = round＜()中の値の小数点以下を四捨五入する関数＞(len(df) * (1 - test_size＜9割＞))

#3601個

train_size = int(train_size)

#学習データは0-3601

Input_train, Ans_train = Input_Ans_Extract(df.iloc[0:train_size], input_num)
#テストデータは3601〜

Input_test, Ans_test = Input_Ans_Extract(df.iloc[train_size:], input_num)
return (Input_train, Ans_train), (Input_test, Ans_test)

(Input_train, Ans_train), (Input_test, Ans_test) = Data_Split(df"sin(x)", input_num = length_of_sequences)

④学習モデルの構築及び学習を実行する

model = Sequential()
model.add(LSTM(hidden_neurons, batch_input_shape=(None＜Noneとすることで値を指定せず、任意の数でバッチサイズをきめることができる＞　, length_of_sequences, in_out_neurons＜出力の数＞), return_sequences=False))
# 一つの入力で100のリスト

# 入力値の個数に対して１つの出力
model.add(Dense(in_out_neurons))

# 活性化関数linear
model.add(Activation("linear"))

# コンパイル
model.compile＜書き上げたモデルを構築＞(loss="mean_squared_error＜誤差逆順伝番法＞", optimizer="rmsprop"＜勾配法の一種＞)

#fitの中にも学習と検証データが必要

model.fit＜学習アルゴリズムを指定＞(Input_train＜入力リスト＞, Ans_train＜正解リスト＞＜＝学習を実行＞, batch_size=60＜一回の学習に60個のデータセット＞, epochs=3, validation_split=0.05＜全データの内0.05(5%)が検証用＞)

# model.fit(Input_train, Ans_train, batch_size=60, nb_epoch=3＜学習データを使い切る回数＞, validation_split=0.05)

predicted = model.predict(Input_test)　

手順②からのコードで実行　下記出現↓

↓完了

モデルの保存

学習した後にモデルを保存しておく事で、読み込んで使う事が出来る

 

 

 

 

 

1

json_string = model.to_json()

2

open('model.json', 'w').write(json_string)

3

model.save_weights('weights.hdf5')

モデルの読み込み

kerasで作成したモデルを読み込む

from keras.models import model_from_json
modelname = input("モデルファイル = 「model.json」")
paramname = input("学習ファイル = 「weights.hdf5」")
json_string = open(modelname).read()
model = model_from_json(json_string)
model.load_weights(paramname)

# predicted = model.predict(Input_test)

⑤学習を終えたモデルをテストし、そのテスト結果を表示する

↓実行！

dataf = pd.DataFrame(predicted[:200])
dataf.columns = ["predict"]
dataf.plot()
dataf["answer"] = Ans_test[:200]
dataf.plot()
plt.show()

ノイズが消え、ほぼsin波と同じ形状のグラフ

sin波の　動き　をうまく学習

すべてのコード↓

import pandas as pd

import numpy as np

import math

import random

from keras.models import Sequential

from keras.layers.core import Dense, Activation

from keras.layers.recurrent import LSTM

import matplotlib.pyplot as plt

# 警告文の非表示

import warnings

warnings.simplefilter('ignore')

# パラメータの定義

random.seed(0)

# 乱数の係数

random_factor = 0.05

# サイクルあたりのステップ数

steps_per_cycle = 80

# 生成するサイクル数

number_of_cycles = 50

# 入力の長さ

length_of_sequences = 100

# 入力値・出力値の次元の大きさ

in_out_neurons = 1

# 隠れ要素のニューロン数

hidden_neurons = 300

# 乱数のシード値固定

random.seed(0)

np_ary = np.arange(steps_per_cycle * number_of_cycles + 1)

df = pd.DataFrame(np_ary, columns=["x"])

pi_t = 2 * math.pi / steps_per_cycle

df["sin(x)"] = df.x.apply(lambda t: math.sin(t * pi_t + random.uniform(-1.0, +1.0) * random_factor))

df[["sin(x)"]].head(steps_per_cycle * 2).plot()

plt.show()

def Input_Ans_Extract(data, input_num = 100):

# 空リスト作成

InputList, AnsList = ,

# 0から(len(data)- input_num)だけの整数が配列としてある

for i in range(len(data) - input_num):#3501個のデータ分

#doc配列にi個目からi+100番目のデータを行列の形にして入れていく

# InputData = data.iloc[i:i+input_num].as_matrix()#3501~3601

InputData = data.iloc[i:i+input_num].values#3501~3601

# AnsData = data.iloc[i+input_num].as_matrix()

AnsData = data.iloc[i+input_num].values

InputList.append(InputData)

AnsList.append(AnsData)

# numpyで扱えるように行列をその形にして再定義

InputList_np = np.array(InputList)

AnsList_np = np.array(AnsList)

return InputList_np, AnsList_np

def Data_Split(df, test_size=0.1, input_num = 100):

train_size = round(len(df) * (1 - test_size))#3601個

train_size = int(train_size)

#学習データは0-3601

Input_train, Ans_train = Input_Ans_Extract(df.iloc[0:train_size], input_num)

#テストデータは3601〜

Input_test, Ans_test = Input_Ans_Extract(df.iloc[train_size:], input_num)

return (Input_train, Ans_train), (Input_test, Ans_test)

(Input_train, Ans_train), (Input_test, Ans_test) = Data_Split(df[["sin(x)"]], input_num = length_of_sequences)

model = Sequential()

model.add(LSTM(hidden_neurons, batch_input_shape=(None, length_of_sequences, in_out_neurons), return_sequences=False))

# Noneとすることで値を指定せず、任意の数でバッチサイズをきめることができる

# 一つの入力で100のリスト

#in_out_neuronsは出力の数

# 入力値の個数に対して１つの出力

model.add(Dense(in_out_neurons))

# 活性化関数linear

model.add(Activation("linear"))

# コンパイル

model.compile(loss="mean_squared_error", optimizer="rmsprop")

model.fit(Input_train, Ans_train, batch_size=60, epochs=3, validation_split=0.05)

# model.fit(Input_train, Ans_train, batch_size=60, nb_epoch=3, validation_split=0.05)

predicted = model.predict(Input_test)

json_string = model.to_json()

open('model.json', 'w').write(json_string)

model.save_weights('weights.hdf5')

dataf = pd.DataFrame(predicted[:200])

dataf.columns = ["predict"]

dataf.plot()

dataf["answer"] = Ans_test[:200]

dataf.plot()

plt.show()