ここではサンプルのIoTセンサーからのデータを使ってTensorFlowで機械学習し、検知されたセンサーデータからアマチュア/普通/プロの判定をしてみます。

1. TensorFlowを準備します

BluemixのDockerコンテナ環境にTensorFlowをインストールするを参照し、Bluemix環境にTensorFlowを準備します
「デプロイ」ステージを構成するステップで、portとして、8888以外に、6006と9000も開けておくようにします

3. Python実行環境にflaskとpandasを入れておきます。flaskはpip istall flaskなどで、pandasはpip install pandasなどでインストール可能です
4. TenserFlowのDockerコンテナのデプロイによりBluemixから提供されるIPアドレスのうち、「パブリックIP」を使用してブラウザからアクセスします。flaskサーバーは「プライベートIP」を指定して9000ポートを開けます

2. サンプルを準備します

用意したTensorFlow環境に下記のソースを配置します。
model.pyでモデルを定義します
- 隠れ層2でそれぞれ32、16個のReLuで構成しています
- COLUMN_SIZEに対象にするデータの項目数を指定します。このサンプルでは、'accel', 'gyro', 'temp', 'humid'を対象にしています
- actual_classに判定に使用する目的変数の項目数を指定します。このサンプルでは amateur = 0, regular = 1, pros = 2の３個の値を使用しています
- ここではGradientDescentOptimizerを使用していますが、AdamOptimizerなど異なるアルゴリズムを使用する事も可能です
- 学習１００回毎に学習とテストの精度を出力します
- その他詳細はTensorFlowのTutorialをご参照ください

model.py

# -*- coding: utf-8 -*-

import tensorflow as tf
import numpy as np

class Model:
    HIDDEN_UNIT_SIZE  = 32
    HIDDEN_UNIT_SIZE2 = 16
    # 対象データの項目数
    COLUMN_SIZE = 4

    def __init__(self):
        self.__setup_placeholder()
        self.__setup_model()
        self.__setup_ops()

    def __enter__(self):
        self.session = tf.Session()
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.session.close()
        return False

    def save(self, path):
        saver = tf.train.Saver()
        saver.save(self.session, path)

    def restore(self, path):
        saver = tf.train.Saver()
        saver.restore(self.session, path)

    def __setup_placeholder(self):
        column_size = Model.COLUMN_SIZE
        self.iot_data   = tf.placeholder("float", [None, column_size])
        # 目的変数が３値
        self.actual_class = tf.placeholder("float", [None, 3])
        self.keep_prob    = tf.placeholder("float")
        self.label        = tf.placeholder("string")

    def __setup_model(self):
        column_size = Model.COLUMN_SIZE
        w1 = tf.Variable(tf.truncated_normal([column_size, Predictor.HIDDEN_UNIT_SIZE], stddev=0.1))
        b1 = tf.Variable(tf.constant(0.1, shape=[Predictor.HIDDEN_UNIT_SIZE]))
        h1 = tf.nn.relu(tf.matmul(self.iot_data, w1) + b1)

        w2 = tf.Variable(tf.truncated_normal([Predictor.HIDDEN_UNIT_SIZE, Predictor.HIDDEN_UNIT_SIZE2], stddev=0.1))
        b2 = tf.Variable(tf.constant(0.1, shape=[Predictor.HIDDEN_UNIT_SIZE2]))
        h2 = tf.nn.relu(tf.matmul(h1, w2) + b2)

        h2_drop = tf.nn.dropout(h2, self.keep_prob)
        # 目的変数が３値
        w2 = tf.Variable(tf.truncated_normal([Predictor.HIDDEN_UNIT_SIZE2, 3], stddev=0.1))
        # 目的変数が３値
        b2 = tf.Variable(tf.constant(0.1, shape=[3]))
        self.output = tf.nn.softmax(tf.matmul(h2_drop, w2) + b2)

    def __setup_ops(self):
        cross_entropy = -tf.reduce_sum(self.actual_class * tf.log(self.output))
        self.summary = tf.scalar_summary(self.label, cross_entropy)
        # 異なるアルゴリズムを使用可能
        # self.train_op = tf.train.AdamOptimizer(0.0001).minimize(cross_entropy)
        self.train_op = tf.train.GradientDescentOptimizer(0.0001).minimize(cross_entropy)
        self.merge_summaries = tf.merge_summary([self.summary])
        correct_prediction = tf.equal(tf.argmax(self.output,1), tf.argmax(self.actual_class,1))
        self.accuracy = tf.reduce_mean(tf.cast(correct_prediction, "float"))

class Trainer(Model):

    def train(self, steps, data):
        self.__prepare_train(self.session)
        for i in range(steps):
            self.__do_train(self.session, i, data)
            if i %100 == 0:
                self.__add_summary(self.session, i, data)
                self.__print_status(self.session, i, data)

    def __prepare_train(self, session):
        self.summary_writer = tf.train.SummaryWriter('logs', graph_def=session.graph_def)
        session.run(tf.initialize_all_variables())

    def __do_train(self, session, i, data):
        session.run(self.train_op, feed_dict=self.train_feed_dict(data))

    def __add_summary(self, session, i, data):
        summary_str = session.run(self.merge_summaries, feed_dict=self.train_feed_dict(data))
        summary_str += session.run(self.merge_summaries, feed_dict=self.test_feed_dict(data))
        self.summary_writer.add_summary(summary_str, i)

    def __print_status(self, session, i, data):
        train_accuracy = session.run(self.accuracy, feed_dict=self.train_feed_dict(data))
        test_accuracy  = session.run(self.accuracy, feed_dict=self.test_feed_dict(data))
        print 'step {} ,train_accuracy={} ,test_accuracy={} '.format(i, train_accuracy, test_accuracy)

    def train_feed_dict(self, data):
        return {
            self.iot_data: data.train_data(),
            self.actual_class: data.train_up_down(),
            self.keep_prob: 0.8,
            self.label: "train"
        }

    def test_feed_dict(self, data):
        return {
            self.iot_data: data.test_data(),
            self.actual_class: data.test_up_down(),
            self.keep_prob: 1,
            self.label: "test"
        }


class Predictor(Model):

    def predict( self, data ):
        return self.session.run(tf.argmax(self.output,1), feed_dict=self.predict_feed_dict(data))

    def predict_feed_dict(self, data):
        return {
            self.iot_data: data,
            self.keep_prob: 1,
            self.label: "predict"
        }

iot_results_loader.pyで学習用のデータをロードします
- ここでは、サンプル用データを使用しています
- outcome項で結果を定義しています
- 全体の２/３を学習データに、１/３をテストデータに使用しています

iot_results_loader.py

# -*- coding: utf-8 -*-

import pandas as pd

class IoTResults:

    def __float__(data):
        return 1.0

    def __init__(self, data):
        self.raw  = data.copy()
        self.data = data

    def predictor_data(self):
        return self.data

    def train_data(self):
        return self.__drop_outcome(self.__train_data())

    def test_data(self):
        return self.__drop_outcome(self.__test_data())

    def train_up_down(self):
        return self.__good_bad(self.__train_data()["outcome"])

    def test_up_down(self):
        return self.__good_bad(self.__test_data()["outcome"])

    def __train_data(self):
        # 全データの 2/3 を訓練データとして使用
        return self.data.loc[lambda df: df.index % 3 != 0, :]

    def __test_data(self):
        # 全データの 1/3 をテストデータとして使用
        return self.data.loc[lambda df: df.index % 3 == 0, :]

    def __drop_outcome(self, data):
        return data.drop("outcome", axis=1)

    def __good_bad(self, outcome):
        return outcome.apply(
            lambda p: pd.Series([
                1 if p <=  0 else 0,
                1 if p ==1  else 0,
                1 if p > 1  else 0
            ], index=['amateur', 'regular', 'pros']))


class IoTResultsLoader:

    def retrieve_iot_data(self):
        df = pd.DataFrame(
            {'accel': [5,5,23,8,7,8,16,0,7,5,6,18,7,0,8,13,6,7,8,6,8,16,8,1,8,7,18,10,8,7,9,0,7,6,14,12,8,5,7,6,8,1,8,7,10,6,7,7,8,18,7,6,8,0,8,6,5,0,8,16,7,8,6,7,9,9,8,7,8,0,8,7,6,7,9,8,9,15,8,7,6,1,8,7,9,7,6,9,8,14,7,6,7,9,8,0,7,8,8,9],
             'gyro': [4,8,20,6,2,4,12,2,12,6,8,14,6,2,8,22,0,10,6,4,8,18,2,4,4,8,20,8,4,6,4,2,4,6,12,12,2,6,4,2,10,2,8,4,6,2,0,4,0,14,2,4,0,4,10,8,4,2,4,12,8,4,2,0,4,0,8,4,6,2,2,0,4,8,6,4,0,12,0,4,8,1,10,4,2,0,2,4,6,12,4,2,2,0,4,3,4,0,2,4],
             'temp': [20,18,25,21,23,25,26,8,19,18,20,48,22,6,24,42,23,19,20,18,23,35,22,6,24,22,40,19,20,30,22,7,28,24,33,28,25,30,28,29,27,6,12,20,18,20,22,21,22,34,28,30,26,7,25,20,19,6,21,27,18,20,18,20,24,26,22,28,24,8,24,25,21,22,28,26,19,32,23,21,24,6,21,20,19,22,25,27,26,34,24,20,23,19,22,7,27,23,24,18],
             'humid': [50,44,58,56,48,42,60,40,58,54,56,60,44,40,58,62,50,52,54,56,48,46,50,42,56,60,64,52,56,60,58,44,44,48,54,52,56,60,62,64,58,48,58,48,48,50,54,60,62,52,48,44,50,54,52,58,46,48,50,56,62,60,54,48,44,52,54,58,50,48,48,52,58,62,60,52,60,58,56,48,44,52,56,54,58,48,52,54,50,58,54,56,62,60,60,52,48,46,58,48],
             'outcome': [1,1,0,1,1,1,0,2,1,1,1,0,1,2,1,0,1,1,1,1,1,0,1,2,1,1,0,1,1,1,1,2,1,1,0,0,1,1,1,1,1,2,1,1,1,1,1,1,1,0,1,1,1,2,1,1,1,2,1,0,1,1,1,1,1,1,1,1,1,2,1,1,1,1,1,1,1,0,1,1,1,2,1,1,1,1,1,1,1,0,1,1,1,1,1,2,1,1,1,1]})
        return df

train.pyで学習させます
- 必要に応じ学習回数を変更して試してみます
- 学習結果が./model.ckptに保存されます

train.py


# -*- coding: utf-8 -*-

from iot_results_loader import *
from model import *

loader = IoTResultsLoader()
data = IoTResults(loader.retrieve_iot_data())

with Trainer() as trainer:
    # 学習回数
    trainer.train(3000, data)
    trainer.save("./model.ckpt")

server.pyでサーバーを実行します
- flaskで9000ポートでアクセスします。TenserFlowのDockerコンテナのデプロイによりBluemixから提供されるIPアドレスのうち、パブリックIPを使用してブラウザからアクセスします。flaskサーバーはプライベートIPへ向け9000ポートを開けます

server.py


# -*- coding: utf-8 -*-

import pandas as pd
from flask import Flask, jsonify, request
from iot_results_loader import *
from model import *

loader = IoTResultsLoader()

predictor = Predictor()
predictor.__enter__()
predictor.restore("./model.ckpt")

iot_data  = loader.retrieve_iot_data()

# webapp
app = Flask(__name__)

@app.route('/api/predictor', methods=['POST'])
def predict():
    data = pd.DataFrame(eval(request.data))
    results = predictor.predict(IoTResults(data).predictor_data().iloc[[0]])
    return jsonify(result=("pros!" if results[0] == 2 else "regular" if results[0] == 1 else "amateur!"))

if __name__ == '__main__':
    app.run(host='172.30.0.3', port=9000)

Node-REDからアクセスします

TensorFlowへアクセスするフロー

[{"id":"e1d1df28.30cdb","type":"inject","z":"23ccc58c.26b582","name":"","topic":"","payload":"","payloadType":"date","repeat":"","crontab":"","once":false,"x":120,"y":100,"wires":[["c3e01383.ee175"]]},{"id":"81d9e766.6e82","type":"http request","z":"23ccc58c.26b582","name":"","method":"POST","ret":"txt","url":"http://xx.xx.xx.xx:9000/api/predictor","tls":"","x":450,"y":160,"wires":[["40a2e5d8.976e6c"]]},{"id":"11916de2.fb6c0a","type":"debug","z":"23ccc58c.26b582","name":"","active":true,"console":"false","complete":"false","x":690,"y":240,"wires":[]},{"id":"c3e01383.ee175","type":"template","z":"23ccc58c.26b582","name":"amateur","field":"payload","fieldType":"msg","format":"handlebars","syntax":"mustache","template":"{'accel': [13], 'gyro': [20], 'temp': [25], 'humid': [58]}\t","x":280,"y":100,"wires":[["81d9e766.6e82"]]},{"id":"c00d807a.6e942","type":"template","z":"23ccc58c.26b582","name":"pros","field":"payload","fieldType":"msg","format":"handlebars","syntax":"mustache","template":"{'accel': [0], 'gyro': [2], 'temp': [8], 'humid': [40]}\t","x":270,"y":220,"wires":[["81d9e766.6e82"]]},{"id":"ba871882.501c9","type":"template","z":"23ccc58c.26b582","name":"regular","field":"payload","fieldType":"msg","format":"handlebars","syntax":"mustache","template":"{'accel': [5], 'gyro': [4], 'temp': [20], 'humid': [50]}\t","x":270,"y":160,"wires":[["81d9e766.6e82"]]},{"id":"473e7329.e15e1c","type":"inject","z":"23ccc58c.26b582","name":"","topic":"","payload":"","payloadType":"date","repeat":"","crontab":"","once":false,"x":120,"y":220,"wires":[["c00d807a.6e942"]]},{"id":"ca48901b.3e3678","type":"inject","z":"23ccc58c.26b582","name":"","topic":"","payload":"","payloadType":"date","repeat":"","crontab":"","once":false,"x":120,"y":160,"wires":[["ba871882.501c9"]]},{"id":"f662f04b.eb2808","type":"template","z":"23ccc58c.26b582","name":"出力","field":"payload","fieldType":"msg","format":"handlebars","syntax":"mustache","template":"あなたは {{payload}} ですね！","x":530,"y":240,"wires":[["11916de2.fb6c0a"]]},{"id":"40a2e5d8.976e6c","type":"function","z":"23ccc58c.26b582","name":"取り出し","func":"var data = JSON.parse(msg.payload);\nmsg.payload = data.result;\nreturn msg;\n","outputs":1,"noerr":0,"x":620,"y":160,"wires":[["f662f04b.eb2808"]]}]

3. 実行します

python train.pyで学習させます
python server.pyでサーバーを実行します
上記のNode-REDのフローからアクセスし、amateur/regular/prosの判定を確認します
python -m tensorflow.tensorboard --logdir=./logsなどでtensorboardを稼働しブラウザで6006ポートへアクセスして学習の状況などを確認します。IPアドレスはTenserFlowのDockerコンテナのデプロイによりBluemixから提供されるIPアドレスのうち、パブリックIPを使用してブラウザからアクセスします。

参考

TensorFlowを使った為替(FX)のトレードシステムを作るチュートリアルを大変参考にさせていただきました。どうもありがとうございます。