クラウドとディープラーニングと価格予測の話

AlpacaJapanさん主催のMarketTech Meetup #02でLTをさせていただいたので、その時の内容をここにもまとめておきます。

スライドはこちらです。

クラウドとディープラーニングと価格予測の話 from KazushiKitaya

www.slideshare.net

ディープラーニングで価格予測モデルを作る

「ディープラーニングで金融商品の価格予測モデルを作る」という場合に、

• いろいろなアーキテクチャ、ハイパーパラメータ、データで学習したい
• いろいろな学習済みモデル、データ、ハイパーパラメータでバックテストをしたい
• 複数のモデルを組み合わせて一つのモデルを作る、すなわちアンサンブルをしたい

といったことがしたくなります。

しかし、当たり前ながらディープラーニングの学習にはには時間がかかります。 アーキテクチャ5種類、ハイパーパラメータ5種類、データ5種類だけでも、組み合わせによって125通りのモデルができます。 それぞれに1時間かかっていたら5日間かかります。

バックテストにおいても、組み合わせが増えるとバカにならない時間がかかります。

クラウドを使っていいい感じにディープラーニングしよう！というのが、今回のトピックです。

クラウドでディープラーニングをすると

クラウドの特徴として「独立した処理については、一つずつ動かしても、一気に動かしても、かかるコストは同じ」ということがあげられます。 この特徴は、今考えている問題と非常に相性がいいです。

さきほどの例、つまり1時間かかる処理が125個ある例、では、5日間ではなく、1時間ですべての処理が終わらせることができます。

また、「様々な種類のマシンを利用できる」というのもクラウドの大きな利点です。GPUが載っているのも多数あります。

また、今回のような処理は、ユーザーを待たせているわけではないため「すぐに結果が必要」でないこともあります。

そのため、「需要の低い時間に安い価格でインスタンスを借りて計算する」ということも可能になります。AWSでいうスポットインスタンス、GCPでいうプリエンプティブインスタンスです。このおかげで思ったよりコストがかさまなかったりします。

ちなみに自分たちはAWSのスポットインスタンスを用いることで、p2.xlargeというインスタンスを70%オフくらいで使うことができています。

考えなくてはいけないこと

もちろんいいことだけではなく、考えなくてはならないこともあります。

まず１つ目が「処理実行の管理をしないといけない」ということです。いくつかのインスタンスを立ち上げ、いい感じに分散させて処理を実行する必要があります。また、スポットインスタンスを用いる場合には考えることが増えます。ローカルで実験するのに比べると、とても手間がかかります。

そして２つ目が「実行結果が分散されてしまう」ということです。それぞれの処理の結果は複数のインスタンスに散らばっている状態になるので、それをどうにかして一箇所に集めることが必要になります。やはりローカルで実験するよりは手間がかかります。

その上で、実際どうやっているか

上にあげたようなことを考え「実際にこんな感じでクラウドで価格予測モデルを作っているよ」という話をします。

実際どうやっているか①

まず、学習のためのDockerコンテナを作成します。 実行コマンドのコマンドライン引数によって、アーキテクチャ、データ、ハイパーパラメータを指定できるようにします。

また、学習が終わったら、生成されたモデルをS3などのクラウドストレージに、パラメータなどをデータベースに保存するようにしています。こうしていろいろなマシンで実行された学習の結果が一箇所に集まるようにします。

余談ですが、Nvidia Dockerを用いることでDockerでもGPUを活用できます。

実際どうやっているか②

そしてあとは学習の実行ですが、自分たちはAWS Batchというサービスを用いています。 このサービスは、ジョブのキューを用意してくれます。ジョブというのは、Dockerイメージと実行コマンドの組み合わせです。

自分たちがキューにジョブを追加すると、AWS Batchがうまいことインスタンスを立ち上げ、そのジョブを実行してくれるというわけです。 スポットインスタンスも簡単に利用できるようになっています。

実際どうやっているか③

バックテストも学習と同様です。 Dockeイメージを用意し、コマンドライン引数によってテストしたいモデルや使いたいデータなどを指定するようにします。 そして、実行が終了した際に結果をクラウドストレージやデータベースに保存するようにしています。 また、AWS Batchを用いているのも同様です。

実際どうやっているか④

AWSのコンソールやCLIのみでこれらの操作すべてをやるのは厳しく、また機能的にも足りないところが出てくるので、機能を拡張し、そしてGUIも用意しています。

具体的には、

• 学習ジョブ・バックテストジョブの追加
• 実行中ジョブのログの表示
• 学習済みモデルの詳細を表示
• 実行済みバックテストの詳細を表示

といった機能をもたせています。

これのおかげで、スマホからでも学習・バックテストを回したり、それらの結果を確認できたりもします。

おわりに

クラウドを使うとディープラーニングが快適にできます。そして、クラウドプロバイダはディープラーニングに役立つサービスをたくさん提供しているので利用すべきです。

おまけ

本当に相場を学習して勝てているのかの見定めは難しいです。「4回の大きい価格変動を当てる」モデルは16個に一個くらいはできてしまいます。

また、探索空間が膨大に増えますが、複数モデルのアンサンブルもしています。

winstonでのエラーログにエラーメッセージを含める

winstonでエラーをロギングするときに発生する問題

以下のようにwinstonのloggerを用意したとします。

const winston = require('winston');

const logger = winston.createLogger({
  level: 'info',
  format: winston.format.json(),
  transports: [
    new winston.transports.Console(),
  ],
});

そして以下のような文を実行します。

logger.info('something has happened');

結果は想像の通り、以下のようになります。

{"message":"something has happened","level":"info"}

しかし、以下の場合はどうでしょう。

logger.error(new Error('something has happened'));

このような出力になり、スタックの内容やエラーメッセージは表示されません。

{"level":"error"}

解決策

そこで、loggerの定義を拡張して以下のようにします。

loggerに渡されたオブジェクト(iとしています)がErrorオブジェクトのインスタンスだった場合に、i.messageの内容をログ出力のmessageとして設定するようにしています。

const logger = winston.createLogger({
  level: 'info',
  format: winston.format.combine(
    winston.format((i) => {
      if (i instanceof Error) {
        return Object.assign({}, i, {
          message: i.message,
        });
      }

      return i;
    })(),
    winston.format.json(),
  ),
  transports: [
    new winston.transports.Console(),
  ]
});

さきほどと同じ文を実行すると以下のような出力となり、うまくいっていることがわかります。

{"level":"error","message":"something has happened"}

stackなど、Errorオブジェクトの他のプロパティに関しても、同じようにして追加することができるはずです。

おまけ

winstonのloggerは、オブジェクトを渡されたとき、列挙可能なプロパティ（とそれに対応する値）のみをログに含めるようです。Errorオブジェクトのmessageプロパティややstackプロパティは列挙可能ではないので、このようなことになっているみたいです。

※ 執筆時のwinstonのバージョンは3.2.1です。

winstonでexpressのリクエストログを扱う

expressにおけるリクエストのロギングはmorganを使うのが鉄板ですが、winstonなどの他の高機能なロガーで、まとめてログを扱いたいことが多いと思います。

この記事では、どのようにしてそれを実現するのかを簡単な例で説明します。

準備

空のディレクトリを作成し、以下のようにしてexpress, morgan, winstonをインストールします。

$ npm install express morgan winston

コード

ルートが/のみの簡単なサーバーを例にとることにします。

具体的には以下のようなことをしています。

• loggerという名前のwinstonのロガーインスタンスを作成
• morganでリクエストログを（文字列の状態で）winstonに流し込むように
• ステータスコード200を返すだけのルートを作成
• ポート3000でリクエストを受け付けるように

const express = require('express');
const morgan = require('morgan');
const winston = require('winston');

const logger = winston.createLogger({
  transports: [
    new winston.transports.Console(),
  ],
});

const app = express();

app.use(morgan('tiny', {
  stream: {
    write: message => logger.info(message.trim()),
  },
}));

app.get('/', (req, res) => {
  res.sendStatus(200);
});

app.listen(3000);

確認

上のコードをindex.jsとして保存し、以下のコマンドでサーバーを立ち上げます。

$ node index.js

curlでリクエストを送ってみます。

$ curl http://localhost:3000/
OK

サーバーのアプリケーションの標準出力に以下のように出力されているはずです。うまくいっていることがわかります。

{"message":"GET / 200 2 - 3.712 ms","level":"info"}

DynamoDBから特定のテーブルの全項目データを取得する(TypeScript/Node.js)

AWS SDKのDynamoDB.DocumentClientのscanメソッドを用いるのですが、そのままではテーブルのサイズが大きい場合に一部のデータしか取得できません。

バッチ処理などで、テーブル内の全項目を取得したいシーンもあると思います。（頻繁にそういった操作が必要になるのなら、アプリケーションの設計ミスのような気がしますが...）

そういった際はLastEvaluatedKeyを用いれば簡単に実現可能です。以下のコードはTypeScriptですが、JavaScriptの場合もほぼ同じです（型定義を消せばいいだけ）。

import awsSdk from 'aws-sdk';
import { DocumentClient } from 'aws-sdk/lib/dynamodb/document_client';

const documentClient = new awsSdk.DynamoDB.DocumentClient({
  region: 'ap-northeast-1',
});

const scanAll = async (params: DocumentClient.ScanInput) => {
  const items: DocumentClient.AttributeMap[] = [];

  let lastEvaluatedKey: undefined | DocumentClient.Key;

  while (true) {
    const res = await documentClient.scan({
      ...params,
      ExclusiveStartKey: lastEvaluatedKey,
    }).promise();

    if (res.Items) {
      res.Items.forEach(item => items.push(item));
    }

    if (!res.LastEvaluatedKey) {
      break;
    }

    lastEvaluatedKey = res.LastEvaluatedKey;
  }

  return items;
};

以下のように用いることができます。

scanAll({ TableName: 'テーブル名' })
  .then((items) => {
    // ...
  });

express.js + passport.jsで1分でbasic認証を実装する

passport.jsを用いると、expressでbasic認証が簡単にできます。

準備

$ npm i express passport passport-http

コード

以下のファイルをindex.jsとして保存します。

const express = require('express');
const passport = require('passport');
const passportHttp = require('passport-http');

passport.use(new passportHttp.BasicStrategy(
  function(username, password, done) {
    if (username === 'correct-username' && password == 'correct-password') {
      return done(null, true);
    } else {
      return done(null, false);
    }
  }
));

const app = express();

app.get('/', passport.authenticate('basic', { session: false }), (req, res) => {
  res.sendStatus(200);
});

app.listen(3000);

確認

サーバーを立ち上げます。

$ node index.js

curlで動作を確かめてみます。

$ curl -u correct-username:correct-password http://localhost:3000
OK
$ curl http://localhost:3000
Unauthorized

正しく動いていることがわかります。

ウェブ開発でNode.jsを採用するメリット・デメリット

ウェブ開発においてある程度の地位を築いた感のあるNode.jsですが、自分も2年ほど使っています。 この記事では、Node.jsの良い点・悪い点をまとめてみます。

メリット

フロントエンドと同じ言語(JavaScript)でバックエンドを書くことができる

これは素晴らしいことだと思います。「一人がフロントエンドとバックエンドの両方に関わる」ということが容易になり、バックエンド開発者とフロントエンド開発者の間で共有できる知見の幅も広がります。

プロトタイピングに向いている

JavaScriptがダイナミックな言語だということや、フレームワークが充実していることにより、高速なプロトタイプ開発が可能です。

AltJSを用いてJavaScriptに足りない点を補完できる

ある程度大規模なものを作るとなると、JavaScriptにはない機能（型など）が欲しくなってきます。 そういう場合にはAltJSを用いることで解決することができます。 特にTypeScriptは開発も進んでおり、エコシステムとしても成熟が進んでいるのでよく用いられています。

シンプルな設計思想

豊富なパッケージ

NPMにはかなり多くのパッケージが公開されています。ここをみると、他の言語・レジストリよりも多くのパッケージが公開されていることがわかります（量だけではなく質も大切ですがそれを定量化するのは難しいので...）。

効率的なIO処理

シングルスレッド上でのイベントループにより、少ないオーバーヘッドで効率的にIOを並列に処理できます。大量のスレッドでメモリが逼迫されるということもなく、よくスケールします。

デメリット

計算量の多いタスクに向いていない

計算に時間のかかるタスクを実行するとき、（workerを使って別スレッドで実行するなどしない限り）イベントループをブロックします。イベントループをブロックしている間は他の処理が基本的にストップしてしまいます。

イベンドリブンなコードには慣れを要する

Node.jsでは、コールバックを利用するなどしてイベントドリブンなコードを書くことになります。1スレッド/1リクエストのモデルではシンプルに書けることも、少し複雑になってしまいます。これには慣れを要します。

コールバック地獄

Node.jsのコードでは、複数のコールバックがネストされてコードが意味不明になってしまう「コールバック地獄」を生み出してしまいやすいです。現在はPromiseや async-awaitなどの普及によって改善されつつあります。

まとめ

パフォーマンスのボトルネックが、計算よりもIOになりそうな場合（ウェブアプリケーションの多くはそうだと思います）においてはNode.jsは採用する価値があると思います。実際、Paypal, Netflix, LinkedInなども利用しているようです。

Nvidia Dockerを用いてDockerコンテナからGPUを利用する

ディープラーニング系の処理でもDockerを使いたい！ということで、Nvidia Dockerを利用してDockerコンテナでGPUを使ってみました。 マシンはAWSのp2.xlargeインスタンス（NVIDIA K80を積んでいます）を、OSはUbuntu 18.04を利用しました。

Dockerのインストール・設定

まずはDockerをインストールします。

インストール

バージョンは1.12以降である必要があります。

Dockerの公式ドキュメント を参考にしています。

$ sudo apt-get update
$ sudo apt-get install \
    apt-transport-https \
    ca-certificates \
    curl \
    gnupg-agent \
    software-properties-common
$ curl -fsSL https://download.docker.com/linux/ubuntu/gpg | sudo apt-key add -
$ sudo add-apt-repository \
   "deb [arch=amd64] https://download.docker.com/linux/ubuntu \
   $(lsb_release -cs) \
   stable"
$ sudo apt-get update
$ sudo apt-get install docker-ce docker-ce-cli containerd.io

確認

実際に入ったか確かめます。Hello from Docker! みたいなのが表示されれば成功です。

$ sudo docker run hello-world

権限周りの設定

rootでdockerコマンドを毎回実行するのもあれなので、設定します。 dockerグループに現在のユーザーを追加します。

$ sudo usermod -aG docker $USER

再ログインすると、sudoなしでdockerコマンドを実行できるようになります。

Nvidia CUDA Toolkitのインストール

Nvidia Dockerをインストールするために必要なCUDA Toolkitをインストールします。 Nvidiaの公式ドキュメントが参考になります。

準備

gccやmakeなどもろもろが必要になるので、インストールします。

$ sudo apt-get install build-essential

インストール

インストーラをダウンロードします。

$ wget https://developer.nvidia.com/compute/cuda/10.1/Prod/local_installers/cuda_10.1.105_418.39_linux.run

こちらを参考にハッシュ値を確認します。

$ md5sum cuda_10.1.105_418.39_linux.run

インストーラを実行します。

$ sudo sh cuda_10.1.105_418.39_linux.run

途中で、規約への同意が求められます。また、その後何をインストールしたいのか聞かれますが全部選択しても大丈夫だと思います。

確認

インストールが終了したら、以下のコマンドを実行してみます。現在のGPUの状態などが表示されたら成功です。

$ nvidia-smi

Nvidia Dockerのインストール

以下のようなコマンドでNvidia Dockerをインストールします。これで終わりです。

$ curl -s -L https://nvidia.github.io/nvidia-docker/gpgkey | \
  sudo apt-key add -
$ distribution=$(. /etc/os-release;echo $ID$VERSION_ID)
$ curl -s -L https://nvidia.github.io/nvidia-docker/$distribution/nvidia-docker.list | \
  sudo tee /etc/apt/sources.list.d/nvidia-docker.list
$ sudo apt-get update
$ sudo apt-get install -y nvidia-docker2
$ sudo pkill -SIGHUP dockerd

実際に使ってみる

TensorFlowの公式ドキュメントにあった、GPUを用いた計算の例をを実行してみます。2次元のテンソルの要素の和を求めています。

$ docker run --runtime=nvidia -it --rm tensorflow/tensorflow:latest-gpu \
   python -c "import tensorflow as tf; tf.enable_eager_execution(); print(tf.reduce_sum(tf.random_normal([1000, 1000])))"
2019-02-28 00:15:38.920366: I tensorflow/core/platform/cpu_feature_guard.cc:141] Your CPU supports instructions that this TensorFlow binary was not compiled to use: AVX2 FMA
2019-02-28 00:15:41.032922: I tensorflow/stream_executor/cuda/cuda_gpu_executor.cc:998] successful NUMA node read from SysFS had negative value (-1), but there must be at least one NUMA node, so returning NUMA node zero
2019-02-28 00:15:41.033785: I tensorflow/compiler/xla/service/service.cc:150] XLA service 0x3f815f0 executing computations on platform CUDA. Devices:
2019-02-28 00:15:41.033815: I tensorflow/compiler/xla/service/service.cc:158]   StreamExecutor device (0): Tesla K80, Compute Capability 3.7
2019-02-28 00:15:41.055212: I tensorflow/core/platform/profile_utils/cpu_utils.cc:94] CPU Frequency: 2300025000 Hz
2019-02-28 00:15:41.055484: I tensorflow/compiler/xla/service/service.cc:150] XLA service 0x3fe90b0 executing computations on platform Host. Devices:
2019-02-28 00:15:41.055521: I tensorflow/compiler/xla/service/service.cc:158]   StreamExecutor device (0): <undefined>, <undefined>
2019-02-28 00:15:41.056190: I tensorflow/core/common_runtime/gpu/gpu_device.cc:1433] Found device 0 with properties:
name: Tesla K80 major: 3 minor: 7 memoryClockRate(GHz): 0.8235
pciBusID: 0000:00:1e.0
totalMemory: 11.17GiB freeMemory: 11.11GiB
2019-02-28 00:15:41.056223: I tensorflow/core/common_runtime/gpu/gpu_device.cc:1512] Adding visible gpu devices: 0
2019-02-28 00:15:41.057198: I tensorflow/core/common_runtime/gpu/gpu_device.cc:984] Device interconnect StreamExecutor with strength 1 edge matrix:
2019-02-28 00:15:41.057230: I tensorflow/core/common_runtime/gpu/gpu_device.cc:990]      0
2019-02-28 00:15:41.057255: I tensorflow/core/common_runtime/gpu/gpu_device.cc:1003] 0:   N
2019-02-28 00:15:41.057853: I tensorflow/core/common_runtime/gpu/gpu_device.cc:1115] Created TensorFlow device (/job:localhost/replica:0/task:0/device:GPU:0 with 10805 MB memory) -> physical GPU (device: 0, name: Tesla K80, pci bus id: 0000:00:1e.0, compute capability: 3.7)
tf.Tensor(54.97377, shape=(), dtype=float32)

うまくいったみたいです。