Webhookでのティック取得から東証「arrowhead」への流し込み -後編-

本投稿は TECOTEC Advent Calendar 2020 の7日目の記事です。

投資戦略システム事業部サーバーエンジニアの伊奈です。 今回の記事は Webhookでのティック取得から東証「arrowhead」への流し込み の後編です。

国内株の自動売買や独自の分析チャートを描画できたトレードステーションがサービスを終了したため、一から自動売買のシステムを作ることにしました。

自動売買のシステムは大きく分けると次の、一. 歩み値情報の収集、二. 歩み値情報の加工とシグナルの算出、三. ブローカーへの発注の3つに分かれると思います。

前編ではその一の歩み値を収集する工程において使用した、AWS DynamoDB、Lambda、API Gatewayの設定について書きました。

後編では、

その一の「歩み値を収集」から前編の続きに当たるDynamoDBのデータの読み込み、

その二の「歩み値情報の加工とシグナルの算出」からLambdaでPandasを使う、

その三の「ブローカーへの発注」からLambdaでブラウザの自動操作

をやっていきます。

目次

はじめに

前編ではDynamoDBに歩み値データを格納するところまでやっているので、今回はDynamoDBから歩み値データを取り出し、加工してシグナルを算出し、ブラウザを自動操作してブローカーへ発注するところをやっていきます。

すべてLambda 関数で実行することになりますが、前編で関数の作るところなどは記載しているのでその辺の説明はしません。

前編からの変更点

前編でDynamoDBのテーブルを作成したときは雑にパーティションキーを設定していましたが、Lambdaから読み込みやすいようにパーティションキーをYYYY-MM-DDの形式で格納されるようにしました。

f:id:teco_inas:20201204131350p:plain

'date'をキーにして取り出していくことになります。

Lambda 関数でDynamoDBの読み込み

ScanとQuery

DynamoDBのデータを読み込む方法はScanとQueryの2種類の方法があります Scanは常にテーブル全体をスキャンします。Queryはキーを指定して条件一致で取得します。

Scan：

Scanはテーブル全体を読み込むことになるので読み込みキャパシティを多く消費することになります。 読み込みキャパシティ(/書き込みキャパシティ)を詳しくは説明しませんが、できるだけDynamoDBを無料枠内で利用したいためにScanは使いません。

Scanではフィルタを使って結果を絞り込むことができますが、テーブル全体をスキャンした後に適用されるので読み込みキャパシティを消費することに変わりはありません。

# Scanの例
table = dynamodb.Table('table_name')
scanData = table.scan(
        FilterExpression=Attr('Date').eq('2020-12-01')
    )
return scanData['Items']

公式でも

フィルタによって多数の結果が除外されるようなサイズの大きいテーブルまたはインデックスでは、可能な限り Scan オペレーションを使用しないことをお勧めします。また、テーブルやインデックスが大きくなるに従って、Scan オペレーションは低速になります。 https://docs.aws.amazon.com/ja_jp/amazondynamodb/latest/developerguide/bp-query-scan.html

とあるのでQueryを使ってデータを読み込みます。

Query：

# Queryの例
import boto3
from boto3.dynamodb.conditions import Key

def lambda_handler(event, context):
    dynamoDB = boto3.resource("dynamodb")
    table = dynamoDB.Table("table_name")

    # DynamoDBへのquery処理実行
    queryData = table.query(
      KeyConditionExpression = Key("Date").eq("2020-11-07"),
      ScanIndexForward = False,
      Limit = 50
    )

    return queryData['Items']

テスト実行すると、

f:id:teco_inas:20201204131449p:plain

取得できていることが確認できました。このデータを使ってシグナルを算出していきます。

Lambda 関数でデータ分析

DynamoDBからデータを取得できたので、この歩み値データを加工して好きな売買シグナルを作っていきます。

Pythonでデータ分析をする際によく使われるライブラリとしてNumPyやPandasがありますが、Lambda関数ではNumPyやPandasなどの外部のライブラリをいきなりimportすることはできません。

外部ライブラリを実行したい場合は、インストール済みのライブラリと実行ファイルをzipファイルにまとめてLambdaにアップロードするか、Lambda Layerの形式で取り込む必要があります。

Lambda Layerを使うと複数のLambda 関数でライブラリを共有できたり、いろいろメリットがあるみたいです。

Lambda LayersにPandasを設定

NumPyとSciPyは公式から出されていますが、なぜかPandasはないので作っていきます。

気をつける点としてLambda 関数内で使用されるネイティブバイナリはLambda 関数の実行環境と同じ環境でコンパイルする必要があります。

今回のlambda 関数を作成する際にランタイムはPython3.8を選択したので、Amazon Linux 2と同じ環境を作る必要があります。

各ランタイムに対応する実行環境はこちらで確認できます。↓ docs.aws.amazon.com

Amazon Linux 2と同じ環境を用意する方法はいくつかありますが、そのひとつdockerを使った方法を記載します。

AWSから公式にビルド用のイメージが提供されているのでこちらを利用します。

$ pip install -r pandas -t python/lib/python3.8/site-packages/
$ zip -r pandas.zip ./python

以上です。あとはAWSのコンソールからポチポチしていけばLayerを作成できます。

Lambda > レイヤー > レイヤーの作成 から適当な名前を付けて.zip ファイルをアップロードで作成したPandasのzipをアップして「作成」

f:id:teco_inas:20201204131605p:plain

Lambda > 関数 > 自分の関数 でLayersからレイヤーの追加をクリックし、カスタムレイヤーから作成したレイヤーを選択して追加

f:id:teco_inas:20201204131651p:plain

これでPandasが使えるようになったので、DynamoDBからとってきたデータをpandas.DataFrameにします。DynamoDBから取得したデータは辞書のリストの形式になっているのでそのままDataFrameにすることができます。

import boto3
import numpy as np
import pandas
from boto3.dynamodb.conditions import Key

def lambda_handler(event, context):
    dynamoDB = boto3.resource("dynamodb")
    table = dynamoDB.Table("table_name")

    # DynamoDBへのquery処理実行
    queryData = table.query(
      KeyConditionExpression = Key("Date").eq("2020-11-07"),
      ScanIndexForward = False,
      Limit = 50
    )

    df = pandas.DataFrame(queryData['Items'])
    print(df)

    return

テスト実行すると、

f:id:teco_inas:20201204131726p:plain

DataFrameになっていることが確認できました。あとは煮るなり焼くなり好きなように加工するだけなので割愛します。

Lambda 関数でブラウザ操作

DataFrameを加工してシグナルを算出出来たら次はブローカーへの発注です。

いろいろやり方はありますが、今回はLambdaからブラウザの自動操作をすることにします。 理由は、ある程度どの証券会社でもできることと、自動売買以外にも使い道があるためです。

例によって Lambda LayersにHeadless ChromeとChromeDriverを設定していきますが、Pandasの時と違いランタイムはPython3.7を選択しています。

最初にランタイムPython3.8でAmazon Linux 2の環境で作ったのですがうまくいかず、結局諦めてランタイムPython3.7でAmazon Linuxの環境を使ったので正確な原因はわかりませんでした。

Lambda LayersにHeadless ChromeとChromeDriverを設定

使用するブラウザはGoogle Chromeですが、そのままLayerに含めるとサイズがでかくなりすぎるため軽量化されているserverless-chromeというものを使用します。 (デプロイパッケージサイズ上限は250 MB (解凍、レイヤーを含む)AWS Lambda のクォータ - AWS Lambda)

serverless-chrome ↓

github.com

Layerを作っていきますが、dockerではなくAWS Cloud9を使用しました。

AWS Cloud9 は、クラウドでコードを記述、実行、デバッグできる統合開発環境でデフォルトの設定でAmazon Linuxの環境で動いているのでこちらを使います。

aws.amazon.com

AWS Cloud9は最小構成 + デフォルトの設定で導入しました。(使用するまでの手順は省きます。)

AWS Cloud9のターミナルで以下を実行して終わりです。

気を付ける点としては、serverless-chromeが対応しているChromeのバージョンとchromedriverが対応しているChromeのバージョンを合わせる必要があることです。２つの組み合わせに問題があるとうまくChromeを起動できません。

$ mkdir -p headless/python/bin
$ cd headless/python
$ wget https://github.com/adieuadieu/serverless-chrome/releases/download/v1.0.0-55/stable-headless-chromium-amazonlinux-2017-03.zip
$ unzip stable-headless-chromium-amazonlinux-2017-03.zip -d bin/
$ wget https://chromedriver.storage.googleapis.com/2.44/chromedriver_linux64.zip
$ unzip chromedriver_linux64.zip -d bin/
$ rm stable-headless-chromium-amazonlinux-2017-03.zip
$ rm chromedriver_linux64.zip
$ cd ../../
$ zip -r headless-chrome.zip ./headless

あとは作成したzipをコンソールからポチポチしてLayerに設定します。

Lambda 関数でChromeDriverが使えるようになったので、動作確認でgoogleにアクセスしてソースコードを取得してみます。

import json
from selenium import webdriver

def headless_chrome():
   options = webdriver.ChromeOptions()

   options.binary_location = "/opt/headless/python/bin/headless-chromium"

   options.add_argument("--headless")
   options.add_argument("--no-sandbox")
   options.add_argument("--single-process")
   options.add_argument("--disable-gpu")
   options.add_argument("--window-size=1280x1696")
   options.add_argument("--disable-application-cache")
   options.add_argument("--disable-infobars")
   options.add_argument("--hide-scrollbars")
   options.add_argument("--enable-logging")
   options.add_argument("--log-level=0")
   options.add_argument("--ignore-certificate-errors")
   options.add_argument("--homedir=/tmp")
   options.add_argument('--disable-dev-shm-usage')

   driver = webdriver.Chrome(
       executable_path="/opt/headless/python/bin/chromedriver",
       chrome_options=options
   )

   return driver

def lambda_handler(event, context):
   driver = headless_chrome()
   driver.get("https://www.google.co.jp")

   result = {
       'body': driver.page_source
   }

   driver.quit()

   return result

テスト実行すると、

f:id:teco_inas:20201204131826p:plain

動作確認できました。

実際は動くまでに10回以上試しているのですが(Amazon Linux 2を試したり、dockerを試したり)、最終的にはコード内にChromeオプションのdisable-dev-shm-usageを追加することで解決できました。

Chromeのデフォルト設定ではメモリスペースに/dev/shmが使われてしまうのですが(/dev/shmに十分な容量がないとエラーになってしまう)、disable-dev-shm-usageを指定することで、代わりに/tmpディレクトリを使うようになるオプションのようです。

おわりに

全2回にわたってAWS API Gateway、DynamoDB、Lambdaを使って自動売買システムの構築をしてきました。歩み値の加工の部分やブローカーへの発注のためのSeleniumの部分は具体的には書きませんでしたが、これで自動売買ができる環境は整ったと思います。

AWSのサービスをいくつか使ってみて、無料枠の中で結構使えるということが分かったので、いろいろ挑戦してみたいと思います。

まだブローカーへの発注部分は出来上がっていないのですが、とりあえずは AmazonでPS5が定価で入荷していたら自動で購入するシステムを作ろうと思います。

Amazon Web Servicesおよびかかる資料で使用されるその他の AWS 商標 は、米国および/またはその他の諸国における、Amazon.com, Inc. またはその関連会社の商標です。

www.tecotec.co.jp