4/22kerasで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

パラメータの定義

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番の乱数が生成されます。

pandasの解説

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

np_ary = np.arange(steps_per_cycle * number_of_cycles + 1)
df = pd.DataFrame(np_ary, columns=["x"])

カラム名（列名）が「x」のデータフレームを作ります。
ここでステップ数分のセルを作りたいので、サイクルあたりのステップ数にサイクル数をかけることで4000個のデータを作ることが出来ます。
データフレームのロー(行)は0から始まるので最後に+1してあげることでセル数は4001個となりますが最後の行名は4000になります。
np.arange()はnumpy形式で0から()の中身の数だけ整数を生成する関数です。

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

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

pi_t は円をsteps_per_cycleで分割した値です。
pandasのカラムはdf[""]としてあげることで、事前に作らずとも指定することができます。
カラムに先ほど作ったステップ数の行に対応する、sinの値を入れていきます。

・「df.x.apply()」は「データフレームのカラム「x」の各セルを()の中に入力する」と言う意味です。
・「lambda t:」は「入力されたがるtを:から定義している」と言う意味になります。

つまり、これら二つを合わせると『カラム「t」の各セルをxに入力する』と言う意味になります。
sinθのθの中にこれらの値と他にノイズを入れてあります。random.uniform(A,B)は小数点を含むA以上B以下の乱数を生成します。random_factorを掛けることでその影響力、つまりノイズの大きさを調整することができます。

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

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

df.head()は()中の値だけをセル値を取得する関数です。ここでは「サイクルあたりのステップ数」×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
        AnsData = data.iloc[i+input_num].as_matrix()
        InputList.append(InputData)
        AnsList.append(AnsData)
    #numpyで扱えるように行列をその形にして再定義
    InputList_np = np.array(InputList)
    AnsList_np = np.array(AnsList)
    return InputList_np, AnsList_np

用意した空のリストにnumpy形式の各行列データを入力します。
ここで、as_matrix()を使うことでそのデータの形式をnumpyの行列にしてくれます。

学習データとテストデータの解説

学習データとテストデータを分ける

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)

round()は()中の値の小数点以下を四捨五入する関数です。
len()でdfの行数を取得します。
「test_size=0.1」はテストデータを１割取って置くこと意味するため、「1 - test_size」で9割りと言う意味になります。
4001個の9割なので訓練データは四捨五入して3601個になります。
対してテストデータは3601個から最後までと言う意味になります。

(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, nb_epoch=3, validation_split=0.05) 
predicted = model.predict(Input_test)

batch_input_shape=(バッチサイズ＝入力データのセット件数, 中間層に入力するデータの数, 中間層に投入する次元数)
mean_squared_errorは誤差逆順伝番法です。rmspropは勾配法の一種です。
X_train, y_trainは先ほど作ったデータになります。
batch_size=60は１セットあたりのサンプル数で、一回の学習に60個のデータセットという意味です。nb_epoch=3は学習データを使い切る回数を表しています。
fitの中にも学習と検証データが必要でvalidation_split=0.05は全データの内0.05(5%)が検証用として扱われます。
```

保存部分

json_string = model.to_json()
open('model.json', 'w').write(json_string)
model.save_weights('weights.hdf5')

読み込み部分

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

全体コード

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
#% matplotlib inline
import matplotlib.pyplot as plt

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

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

################################################################

def Input_Ans_Extract(data, input_num = 100):
    InputList, AnsList = [], []
    for i in range(len(data) - input_num):
        InputData = data.iloc[i:i+input_num].as_matrix()
        AnsData = data.iloc[i+input_num].as_matrix()
        InputList.append(InputData)
        AnsList.append(AnsData)
    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))
    train_size = int(train_size)
    Input_train, Ans_train = Input_Ans_Extract(df.iloc[0:train_size], input_num)
    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))  
model.add(Dense(in_out_neurons))  
model.add(Activation("linear"))  
model.compile(loss="mean_squared_error", optimizer="rmsprop")
model.fit(Input_train, Ans_train, batch_size=60, nb_epoch=3, validation_split=0.05) 

################################################################

predicted = model.predict(Input_test) 

################################################################

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