AI×医療事務(カナダ)

今は医療事務しているので医療事務関連、IT関連をのんびりと備忘録も含め記載していきます カナダ・歯科助手のことも復習がてら記載中 

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

 

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

 

 

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

Sin波=正弦波

 

f:id:chocolate22554:20210304165705p:plain

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

f:id:chocolate22554:20210304180115p:plain

 

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

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

正解データの作成

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

 

用意した空のリスト

f:id:chocolate22554:20210304180437p:plain

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

f:id:chocolate22554:20210304180611p:plain

values

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

 

 

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

def Data_Split(df, test_size=0.1<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のリスト

 

# 入力値の個数に対して1つの出力
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) 

 

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

f:id:chocolate22554:20210304183626p:plain

↓完了

f:id:chocolate22554:20210304183811p:plain

モデルの保存

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

 

モデルの読み込み

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波の 動き をうまく学習

f:id:chocolate22554:20210304184045p:plain

 

 

 

すべてのコード↓

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は出力の数

# 入力値の個数に対して1つの出力
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()