投資戦略システム事業部サーバーエンジニアの伊奈です。
2020年5月7日、個人的に衝撃なニュースがありました。それは、「日本株取引ツール トレードステーションサービス終了のお知らせ！！」です。
トレードステーションはトレード用プログラム言語「EasyLanguage」を使用した取引戦略に基づく注文の自動執行機能や、独自の分析チャートの描画など、アクティブトレーダーを支援する強力なツールで、国内株式の自動売買を行うトレーダーのユーザーは多かったと思います。
トレードステーションサービス終了後の国内株式自動売買の(個人的)最適なツールを考えた結果と、AWSの勉強のため、一から自動売買のシステムを作ることにしました。
自動売買のシステムは大きく分けると次の、一. 歩み値情報の収集、二. 歩み値情報の加工とシグナルの算出、三. ブローカーへの発注の3つに分かれると思いますが、 今回はその一の歩み値を収集する工程において使用した、AWS DynamoDB、Lambda、API Gatewayを設定したときの手順の備忘録的エントリーです。
目次
はじめに
歩み値とは取引が成立するごとに生成される何時何分何秒に、何円で何株約定したかという情報です。 東証の注文処理時間は0.2ミリ秒ということもあり、個人がリアルタイムで正確な歩み値情報を取得するのは難しいです。
そのため今回は疑似歩み値として秒足30秒足の時刻、価格、出来高を使用します。
歩み値情報を1秒間の最初の値、最高の値、最低の値、最後の値、出来高の合計を算出すれば1秒足となり、1分足や1時間足も同様の考え方で作成できます。(なんか怒られそうだったので、1秒足ではなく30秒足に、価格データは国内の証券取引所以外のデータを使いました。)
動作の流れはこんな感じです。 データ提供元がWebhookで秒足の情報を送信します。API Gatewayが受信すると、Lambdaを起動します。Lambdaは受け取ったデータをDynamoDBに保存します。
DynamoDBの設定
テーブルの作成
１. AWS マネジメントコンソールからDynamoDB>テーブル>テーブルの作成に進みます。
２. 適当な'テーブル名'と'プライマリーキー'を入力して「作成」をクリックします。
３. テーブル一覧に作成したテーブル名が表示されていることを確認します。
IAMロールの作成
４. AWS マネジメントコンソールからIAM>アクセス管理>ロール>「ロールの作成」をクリックします。
５-１. '信頼されたエンティティの種類を選択'で'AWSサービス'を選択、
５-２. 'ユースケースの選択'で'Lambda'を選択し、「次のステップ：アクセス権限」をクリックします。
６. ポリシーの選択で'AmazonDynamoDBFullAccess'と'AWSLambdaDynamoDBExecutionRole'を選択し、「次のステップ：タグ」をクリックします。
７. 'タグの追加(オプション)'では何もせず、「次のステップ：確認」をクリックします。
８. 確認画面で'ロール名'を適当に入力して「ロールの作成」をクリックします。
Lambdaの設定
Lambda関数の作成
１. AWS マネジメントコンソールからLambda>関数>「関数の作成」をクリックします。
２-１. 関数の作成画面のオプションで'一から作成'を選択、
２-２. '関数名'を適当に入力、
２-３. 'ランタイム'でPython3.8を選択、
２-４. '実行ロール'で'既存のロールを使用する'を選択、
２-５. '既存のロール'に作成しておいたロール名を入力、
２-６. 「関数の作成」をクリックします。
Lambda関数のコード内容
作成されたLambda関数のlambda_function.pyの中身を修正します。
AWS SDK for Python (Boto3)のput_item()メソッドでDynamoDBにデータを書き込みます。
import json import boto3 import datetime from decimal import Decimal from boto3.dynamodb.conditions import Key dynamodb = boto3.resource('dynamodb') table = dynamodb.Table("SecondBar_nk225m1") def modify_item(event): fix_dict = {} __datetime = datetime.datetime.strptime(event['datetime'], '%Y-%m-%dT%H:%M:%SZ') _datetime = __datetime + datetime.timedelta(hours=9) fix_dict['datetime'] = _datetime.strftime('%Y-%m-%d %H:%M:%S') fix_dict['date'] = _datetime.strftime('%Y-%m-%d') fix_dict['hour'] = _datetime.strftime('%H') fix_dict['minute'] = _datetime.strftime('%M') fix_dict['second'] = _datetime.strftime('%S') fix_dict['close'] = Decimal(event['close']) fix_dict['volume'] = Decimal(event['volume']) return fix_dict def put_dynamodb(dict): putResponse = table.put_item( Item={ 'Datetime': dict['datetime'], 'Date': dict['date'], 'Hour': dict['hour'], 'Minute': dict['minute'], 'Second': dict['second'], 'Close': dict['close'], 'Volume': dict['volume'], } ) if putResponse['ResponseMetadata']['HTTPStatusCode'] != 200: print(putResponse) else: print('PUT Successed.') return putResponse def lambda_handler(event, context): fix_dict = modify_item(event) put_dynamodb(fix_dict)
API Gatewayの設定
APIの作成
１. AWS マネジメントコンソールからAPI Gateway>「APIを作成」をクリックします。
２. 'REST API'の「構築」をクリックします。
３-１. 'プロトコルを選択する'でRESTを選択、
３-２. '新しいAPIの作成'で'新しいAPI'を選択、
３-３. 適当な'API名'を入力して、 「APIを作成」をクリックします。
４. 作成されたAPIのリソースから'アクション'でメニューを開き、「リソースの作成」をクリックします。
５. 適当な'リソース名'を入力し、「リソースの作成」をクリックします。
６. 'アクション'でメニューを開き、「メソッドの作成」をクリックします。
７. 'POST'を選択し、チェックマークをクリックします。
８-１. '統合タイプ'で'Lambda関数'を選択、
８-２. 'Lambda関数'に作成したLambda関数名を入力して、「保存」をクリックします。
９. 「Lambda関数に権限を追加する」という確認画面が出てくるので「OK」をクリックします。
APIのデプロイ
１. 'アクション'でメニューを開き、「APIのデプロイ」をクリックします。
２-１. 'デプロイされるステージ'で'新しいステージ'を選択、
２-２. 適当な'ステージ名'を入力し、「デプロイ」をクリックします。
動作確認
APIのステージエディターに表示されるURLをWebhook送信元でWebhookURLとして設定します。
この時、"https://apiId.execute-api.region.amazonaws.com/stageName/resourceName"になっていることを注意します。resourceNameがないと403エラーになります。
こんなデータが送られてくる想定です。
{ "datetime": "2020-09-07T09:56:00Z", "close": "108", "volume": "3" }
DynamoDBを確認すると、
データが保存できていることが確認できました。
おわりに
今回はAWS API Gateway、Lambda、DynamoDBを使って、Webhookのデータを保存する最低限の手順を書きました。 今回は書ききれなかったLambdaのテストやCloudWatchでのログの出し方、無料枠のうちどれくらいの使用率なのかとかも機会がありましたら書きたいと思います。
Amazon Web Servicesおよびかかる資料で使用されるその他の AWS 商標 は、米国および/またはその他の諸国における、Amazon.com, Inc. またはその関連会社の商標です。
