Chainerで楕円軌道を学習する

chainerを用いて、Feedforward Neural Networkにより楕円軌道を学習します。
テストはオープンループで行います。

chainerの使い方の解説をしながらこの問題を解いていきます。

問題とアプローチ

[問題]
x=0.8cos(θ),y=0.8sin(θ)の楕円軌道を学習する。

[アプローチ]
x(θn)＝(0.8cos(θn),0.8sin(θn)), (0<=θn<=2π)
についてx(θn)からx(θn+1)を予測するFNNを設計します。
そして、このFNNを用いて訓練データを用いて学習し、テストデータを用いて結果をテストして確かめます。

データの作成と変数設定

訓練データとテストデータ

楕円上の任意の点を指定します。

[訓練データ]
xtrain(θn)＝(0.8cos(θn),0.8sin(θn)), θn=2πn/20(0<=n<=20,nは自然数)
とします。以下のように書きます。

a=np.linspace(0,20,sample_no).reshape(sample_no,1) 
xtrain=np.zeros(input_no*sample_no).reshape(sample_no,input_no).astype(np.float32)
xtrain[:,0]=[0.8*np.cos(2.0*np.pi*i/20) for i in a]
xtrain[:,1]=[0.8*np.sin(2.0*np.pi*i/20) for i in a]

[テストデータ]
xtest(θn)＝(0.8cos(θn),0.8sin(θn)), θn=2πn/27(0<=n<=27,nは自然数)
とします。同様に、以下のように書きます。

a=np.linspace(0,27,sample_no).reshape(sample_no,1) 
xtest=np.zeros(input_no*sample_no).reshape(sample_no,input_no).astype(np.float32)
xtest[:,0]=[0.8*np.cos(2.0*np.pi*i/27) for i in a]
xtest[:,1]=[0.8*np.sin(2.0*np.pi*i/27) for i in a]

変数

以下のように設定します。
教師データの個数：sample_no
学習回数：epoch
層の数：
　入力層：input_no=2(固定)
　中間層：hidden_no
　出力層：output_no=2(固定)
バッチサイズ：bs

chainerによる学習モデルの構成

コネクションの準備(Link)

LinearというLinkをl1,l2という名前で登録します。

class FNN(Chain):
    def __init__(self): #コネクションを準備する
        super(FNN,self).__init__(
            l1=L.Linear(input_no,hidden_no),
            l2=L.Linear(hidden_no,output_no),

なお、
l1=L.Linear(input_no,hidden_no),
l2=L.Linear(hidden_no,output_no),
は、
self.add_link("l1",F.Linear(input_no,hidden_no)),
self.add_link("l2",F.Linear(hidden_no,output_no)),
と書くのと同じことです。

forward計算

登録したLinkを関数として呼びます。引数は原則variableクラスだそうです。
※chainer.variable(data):dataという数値のchainer用変数を作る。
活性化関数はchainer.functions(F)から呼び出します。
（今回はtanhです。）

#class FNN
    def fwd(self,x):
        h1=F.tanh(self.l1(x))
        h2=F.tanh(self.l2(h1))
        return h2

    def get_predata(self,x):
        return self.fwd(Variable(x.astype(np.float32).reshape(sample_no,input_no))).data

損失関数

chainer.functionsから呼び出される。今回は平均二乗誤差を用います。

#class FNN
    def __call__(self,x,y): #損失関数
        return F.mean_squared_error(self.fwd(x),y)

最適化

optimizerを作成し、これにFNNモデルをセットします。
SGD、Adam、RMSPropなど様々あるそうです。

model=FNN()
optimizer=optimizers.SGD()
optimizer.setup(model)

訓練

一つ先のデータセット(Xtrain_n+1)を現在のデータセット(Xtrain_n)の正解地として学習させます。
勾配の初期化と逆伝播、更新は
　　optimizer.zero_grads()
　　loss.backward()
　　optimizer.update()
を繰り返し行います。

for i in range(epoch):
    for j in range(sample_no): #一つ先を入れる
        if (j+1<sample_no):
            ytrain[j]=np.array(xtrain[j+1])
        else:
            ytrain[j]=np.array(xtrain[0])

    model.zerograds()
    loss=model(xtrain,ytrain)
    loss.backward()
    optimizer.update()

テスト

オープンループで行うので、毎回テストデータを読み込み、次の地点を予測します。
テストデータをFeedForwardで計算し、youに出力するだけなので、以下のように書くだけです。

yout=model.get_predata(xtest)

テスト結果

教師データ（目標の楕円）を青で、テスト結果を赤で描画しています。

学習回数100回：まだまだずれています

学習回数5000回:だいぶ楕円に近づきました

学習回数5000000回：ほぼ正解ですね

コード全文

ellipse.py

#-*- coding:utf-8 -*-
import numpy as np
import chainer
from chainer import cuda,Function,gradient_check,Variable,optimizers,serializers,utils
from chainer import Link,Chain,ChainList
import chainer.functions as F
import chainer.links as L
from sklearn import datasets
import matplotlib.pyplot as plt

#教師データの個数
sample_no=100 

#学習回数
epoch=500000

#層の数
input_no=2
hidden_no=2
output_no=2

#教師データ作成
a=np.linspace(0,20,sample_no).reshape(sample_no,1) 
xtrain=np.zeros(input_no*sample_no).reshape(sample_no,input_no).astype(np.float32)
xtrain[:,0]=[0.8*np.cos(2.0*np.pi*i/20) for i in a]
xtrain[:,1]=[0.8*np.sin(2.0*np.pi*i/20) for i in a]

#モデルの構築
class FNN(Chain):
    def __init__(self): #コネクションを準備する
        super(FNN,self).__init__(
            l1=L.Linear(input_no,hidden_no),
            l2=L.Linear(hidden_no,output_no),
        )
    def __call__(self,x,y): #損失関数
        return F.mean_squared_error(self.fwd(x),y)

    def fwd(self,x):
        h1=F.tanh(self.l1(x))
        h2=F.tanh(self.l2(h1))
        return h2

    def get_predata(self,x):
        return self.fwd(Variable(x.astype(np.float32).reshape(sample_no,input_no))).data

#最適化手法
model=FNN()
optimizer=optimizers.SGD()
optimizer.setup(model)

#訓練の正解値を格納
ytrain=np.zeros(input_no*sample_no).reshape(sample_no,input_no).astype(np.float32)

#バッチサイズ
bs=25

#訓練
for i in range(epoch):
    for j in range(sample_no): #一つ先を入れる
        if (j+1<sample_no):
            ytrain[j]=np.array(xtrain[j+1])
        else:
            ytrain[j]=np.array(xtrain[0])

    model.zerograds()
    loss=model(xtrain,ytrain)
    loss.backward()
    optimizer.update()

#テスト(openloop)
a=np.linspace(0,27,sample_no).reshape(sample_no,1) 
xtest=np.zeros(input_no*sample_no).reshape(sample_no,input_no).astype(np.float32)
xtest[:,0]=[0.8*np.cos(2.0*np.pi*i/27) for i in a]
xtest[:,1]=[0.8*np.sin(2.0*np.pi*i/27) for i in a]
yout=model.get_predata(xtest)
print yout

#描画
plt.plot(yout[:,0],yout[:,1],"r",label="training data")    #赤で学習結果を描く
plt.plot(xtrain[:,0],xtrain[:,1],"b",label="teaching data") #青で教師データを描く
plt.show()