【AWS】Lambdaでbitcoinの価格通知
【きっかけ】
過去の記事参照
buffalokusojima.hatenablog.com
1. 使用するサービス
(AWS)
- Lambda
- IAM
- Cloudwatch Event
- SSM
- DynamoDB
(外部サービス)
- LINE API
2.概要
Cloudwatch Eventでbitcoinの価格を確認するLambdaを定期的に実行することで実現していきます。
一つのLambdaでbitflyerAPIとの通信とLINE通知、DynamoDBからレコードの取得を行っています。
価格は取引履歴の直近価格です。前回取得した直近価格と現在の直近価格を比較して、価格通知を行っています。
3. 実装
3-1.Lambda作成
以下コード入力
const AWS = require('aws-sdk'); const ssm = new (require('aws-sdk/clients/ssm'))(); const request = require('request'); const momentTimezone = require('moment-timezone'); AWS.config.update({region: 'ap-northeast-1'}); const ddb = new AWS.DynamoDB({apiVersion: '2012-08-10'}); exports.handler = (event, context, callback) => { // 前回取得した直近価格を保存 var PRICE_FX_BTC_JPY = process.env['FX_BTC_JPY']; console.log(PRICE_FX_BTC_JPY); var price_alert_data; // DynamoDBからレコードを取得 getDatafromDynamoDB(callback) .then(function(data){ if(data.Items.length == 0){ console.log("price check data not set"); callback(null, { statusCode: 200, body: JSON.stringify({message: "price check data not set"}), headers: {"Content-type": "application/json"} }); return; } price_alert_data = data.Items; const method = "GET" const path = "/v1/executions?product_code=FX_BTC_JPY" var option = { url: 'https://api.bitflyer.jp' + path, method: method, headers: { 'Content-Type': 'application/json' } } // bitflyerから直近取引履歴を取得 return sendRequest(option,callback); }) .then(function(data){ if(data.response.statusCode != 200){ console.error("Error:",data.response); callback(null, { statusCode: data.response.statusCode, body: JSON.stringify({message: data.response}), headers: {"Content-type": "application/json"} }); return; } var price_data = JSON.parse(data.body); if(price_data.length == 0){ console.log("price data not Found"); callback(null, { statusCode: 403, body: JSON.stringify({message: 'No Data Found'}), headers: {"Content-type": "application/json"} }); return; } console.log(price_data); if(typeof PRICE_FX_BTC_JPY == 'undefined'){ PRICE_FX_BTC_JPY = process.env['FX_BTC_JPY'] = price_data[0].price; console.log("Lambda Restarted"); callback(null,{ statusCode: 200, body: JSON.stringify({message: "Lambda Restarted"}), headers: {"Content-type": "application/json"} }); return; } // DynamoDBから取得したレコードを回し、直近価格に該当する場合LINE通知 for(const value of price_alert_data){ console.log(value); if(value.side.S == 'up' && Number(value.price.N) > PRICE_FX_BTC_JPY && Number(value.price.N) < Number(price_data[0].price)){ console.log("["+value.coin_pair.S+"] " + value.price.N + "over"); var message = '\n'; const dateTimeJst = momentTimezone(new Date()).tz('Asia/Tokyo').format('YYYY/MM/DD HH:mm:ss'); message += "Price Alart for " + value.side.S + "\n" + "price " + value.price.N + "\n" + dateTimeJst process.env['FX_BTC_JPY'] = price_data[0].price; console.log(message) sendLine(message, callback); return; }else if(value.side.S == 'down' && Number(value.price.N) < PRICE_FX_BTC_JPY && Number(value.price.N) > Number(price_data[0].price)){ console.log("["+value.coin_pair.S+"] " + value.price.N + "below"); var message = '\n'; const dateTimeJst = momentTimezone(new Date()).tz('Asia/Tokyo').format('YYYY/MM/DD HH:mm:ss'); message += "Price Alart for " + value.side.S + "\n" + "price " + value.price.N + "\n" + dateTimeJst process.env['FX_BTC_JPY'] = price_data[0].price; console.log(message) sendLine(message, callback); return; } } // 該当なしの場合は通知なし process.env['FX_BTC_JPY'] = price_data[0].price; console.log("No Alert"); callback(null,{ statusCode: 200, body: JSON.stringify({message: "No Alert"}), headers: {"Content-type": "application/json"} }); return; }); function getDatafromDynamoDB(callback){ return new Promise(function (resolve) { var params = { ExpressionAttributeValues: { ':c': {S: 'fx_btc_jpy'} }, KeyConditionExpression: 'coin_pair = :c', ProjectionExpression: 'coin_pair, price, side', TableName: 'price_check_bitflyer' }; // Call DynamoDB to add the item to the table ddb.query(params, function(err, data) { if (err) { console.log("Error", err); callback(null, { statusCode: 401, body: JSON.stringify({message: err.toString()}), headers: {"Content-type": "application/json"} }); resolve(null); return; } resolve(data); }); }); } function getParameterFromSystemManager(apikey_name, callback) { return new Promise(function (resolve) { var apikey = process.env[apikey_name]; if(!apikey || typeof apikey == undefined){ // Fetches a parameter called REPO_NAME from SSM parameter store. // Requires a policy for SSM:GetParameter on the parameter being read. var params = { Name: apikey_name, /* required */ WithDecryption:true }; ssm.getParameter(params, function(err, apikey) { if (err){ console.error(err.stack); callback(null,{ statusCode: 500, body: JSON.stringify({message: err.toString()}), headers: {"Content-type": "application/json"} }); resolve(null); return; } process.env[apikey_name] = apikey.Parameter.Value; resolve(apikey.Parameter.Value); }); }else resolve(apikey); }); } function sendLine(message, callback){ getParameterFromSystemManager('line-access-key', callback) .then(function(lineKey){ var option = { url: 'https://notify-api.line.me/api/notify', headers: { 'Content-Type': 'application/json', 'Authorization': 'Bearer ' + lineKey }, method: 'POST', form :{ message: message } }; return sendRequest(option,callback); }) .then(function(data){ if(data.response.statusCode != 200){ console.error(data) } callback(null, { statusCode: data.response.statusCode, body: data.body, headers: {"Content-type": "application/json"} }); }); } function sendRequest(option, callback){ return new Promise(function (resolve) { request(option, function(error, response, body){ if(error){ console.error(error); callback(null,{ statusCode: 500, body: JSON.stringify({message: error.toString()}), headers: {"Content-type": "application/json"} }); resolve(null); } var data = {response, body} resolve(data); }); }); } };
3-2. Lambdaにssm取得ロール付与とssm登録
過去記事参照
3-3. DynamoDBのテーブル登録
以下のように設定
・Primary partition key ・・・ coin_pair (コインペア)
・Primary sort key ・・・ price(価格)
これら二つ合わせて一意のレコードになります。
加えて、sideというカラムを加えてupとdown判定に使います。
注意点としては同じコインペア、価格に対し、sideがupとdown両立は出来ません。(必要性が無さそうだったから)
以下参考テーブル
3-4. LambdaにdynamoDB取得ロール付与
ssmで設定したのと同様に新たにポリシーを作成してアタッチします。(リソースのテーブル名は適宜変更してください)
{ "Version": "2012-10-17", "Statement": [ { "Sid": "VisualEditor0", "Effect": "Allow", "Action": [ "dynamodb:DescribeReservedCapacity*", "dynamodb:List*", "dynamodb:DescribeTimeToLive", "dynamodb:DescribeLimits" ], "Resource": "*" }, { "Sid": "VisualEditor1", "Effect": "Allow", "Action": [ "dynamodb:CreateTable", "dynamodb:BatchGet*", "dynamodb:PutItem", "dynamodb:DescribeTable", "dynamodb:Delete*", "dynamodb:Get*", "dynamodb:BatchWrite*", "dynamodb:Scan", "dynamodb:Query", "dynamodb:DescribeStream", "dynamodb:Update*" ], "Resource": "arn:aws:dynamodb:*:*:table/price_check_bitflyer" } ] }
3-5. Cloudwatch EventのRoleに作成したラムダを設定
過去記事参照
3-6. 最後に
以上で設定した時間おきにBTCの価格を確認し、設定した価格以上または以下であればLINEが飛ぶようになっています。
【AWS】Lambdaでbitflyer約定通知
【きっかけ】
元々AWSを勉強していたことと、３月１２日のコロナショックによりbitflyerにて、今年の利益を全て吹き飛ばし、損失街道まっしぐらになったことがきっかけです。(現在損失更新中)
というのも、暴落中にbitflyerのサイトへのアクセスが非常に悪くなり、取引がWebから全く出来なくなったところに敗因があります。(STOP入れてましたが現物で死にました)
しかし、その間でもAPIからならアクセスが可能とのことを風の噂で聞き、実際に試そうといったのが今回の経緯です。
備忘録的なものでコード等、お粗末なところがありますが、同じような考えの人の参考になればと思います。
1. 使用するサービス
(AWS)
- Lambda
- IAM
- Cloudwatch Event
- SSM
(外部サービス)
- LINE API
2.概要
Cloudwatch Eventで約定を確認するLambdaを定期的に実行することで実現していきます。
一つのLambdaでbitflyerAPIとの通信とLINE通知を行っています。
3. 実装
3-1.Lambda作成
以下コード入力
const ssm = new (require('aws-sdk/clients/ssm'))(); const request = require('request'); const crypto = require('crypto'); const momentTimezone = require('moment-timezone'); exports.handler = (event, context, callback) => { const id = process.env['id']; console.log('id: ' + id); //ssmからbitflyerのAPIキーを取得します getParameterFromSystemManager('bitflyer-keys',callback) .then(function(data){ //ここら辺はbitflyerAPIのリファレンスのテンプレ通りです const apikey = data.split(",")[0]; const sercretKey = data.split(",")[1]; var timestamp = Date.now().toString(); var method = 'GET'; var path = '/v1/me/getchildorders?product_code=FX_BTC_JPY &child_order_state=COMPLETED'; //約定ID指定(これで最新の約定を取得する) var afterID = '&after=' if(id){ path += afterID + id; } var text = timestamp + method + path; var sign = crypto.createHmac('sha256' , sercretKey).update(text).digest('hex'); var option = { url: 'https://api.bitflyer.jp' + path, method: method, headers: { 'ACCESS-KEY': apikey, 'ACCESS-TIMESTAMP': timestamp, 'ACCESS-SIGN': sign, 'Content-Type': 'application/json' } } //bitflyerAPIにリクエストを送ります return sendRequest(option, callback); }).then(function(data){ if(data.response.statusCode != 200){ console.error("Error:",data.response); callback(null, { statusCode: data.response.statusCode, body: JSON.stringify({message: data.response}), headers: {"Content-type": "application/json"} }); return; } data = JSON.parse(data.body); if(data.length == 0){ console.log('No data Found'); callback(null,{ statusCode: 200, body: JSON.stringify({message: 'No data Found'}), headers: {"Content-type": "application/json"} }); return; } console.log(data) const dateTimeUtc = momentTimezone.tz(data[0].child_order_date.split(" ")[0], 'UTC'); const dateTimeJst = momentTimezone(dateTimeUtc) .tz('Asia/Tokyo').format('YYYY/MM/DD HH:mm:ss'); const date = new Date(dateTimeJst); const toDay = new Date(momentTimezone(new Date()) .tz('Asia/Tokyo').format('YYYY/MM/DD HH:mm:ss')); //Lambdaの再起動判定です。インスタンスが新たに生成されると環境変数がリセットされてしまいます。 //現在時刻と約定時刻から判断してます if(toDay.getTime() - date.getTime() > 100000){ process.env['id'] = data[0].id; console.log('Lambda Restarted'); callback(null,{ statusCode: 500, body: JSON.stringify({message: 'Lambda Restarted'}), headers: {"Content-type": "application/json"} }); return; } process.env['id'] = data[0].id; var message = '\n'; //約定情報を回していきます data.forEach(function(value){ const dateTimeUtc = momentTimezone.tz(value.child_order_date.split(" ")[0], 'UTC'); const dateTimeJst = momentTimezone(dateTimeUtc) .tz('Asia/Tokyo').format('YYYY/MM/DD HH:mm:ss'); //Lineに送るメッセージ作成 message += "[" + value.side + ": " + value.child_order_type + "]\n" + "Date " + dateTimeJst + "\n" + "price " + value.price + "\n" + "size: " + value.size + "\n" + "average price: " + value.average_price + "\n" + "executed size: " + value.executed_size + "\n" + "----------------------\n"; }) console.log(message); sendLine(message,callback); }); function getParameterFromSystemManager(apikey_name, callback) { return new Promise(function (resolve) { var apikey = process.env[apikey_name]; if(!apikey || typeof apikey == undefined){ var params = { Name: apikey_name, /* required */ WithDecryption:true }; ssm.getParameter(params, function(err, apikey) { if (err){ console.error(err.stack); callback(null,{ statusCode: 500, body: JSON.stringify({message: err.toString()}), headers: {"Content-type": "application/json"} }); resolve(null); return; } process.env[apikey_name] = apikey.Parameter.Value; resolve(apikey.Parameter.Value); }); }else resolve(apikey); }); } function sendLine(message, callback){ getParameterFromSystemManager('line-access-key', callback) .then(function(lineKey){ var option = { url: 'https://notify-api.line.me/api/notify', headers: { 'Content-Type': 'application/json', 'Authorization': 'Bearer ' + lineKey }, method: 'POST', form :{ message: message } }; return sendRequest(option,callback); }) .then(function(data){ if(data.response.statusCode != 200){ console.error(data) } callback(null, { statusCode: data.response.statusCode, body: data.body, headers: {"Content-type": "application/json"} }); }); } function sendRequest(option, callback){ return new Promise(function (resolve) { request(option, function(error, response, body){ if(error){ console.error(error); callback(null,{ statusCode: 500, body: JSON.stringify({message: error.toString()}), headers: {"Content-type": "application/json"} }); resolve(null); } var data = {response, body} resolve(data); }); }); } };
3-2. Lambdaにssm取得ロール付与
LambdaのPermissionsタブから既存ロールを選択します
以下ポリシーを作成して付与(フルポリシーでもいいです)
{ "Version": "2012-10-17", "Statement": [ { "Sid": "VisualEditor0", "Effect": "Allow", "Action": [ "sts:AssumeRole", "ssm:GetParameter" ], "Resource": "*" } ] }
3-3. ssmのパラメータストアにキーを登録
bitflyerAPIとLINEAPIのキーをそれぞれ設定します。
(bitflyer-keyはapi,sercretの順になっています)
3-4. Cloudwatch EventのRoleに作成したラムダを設定
EventSourceはScheduleにし実行タイミングを設定し、TargetにLambdaを選択し、作成したLambda Functionを設定する
3-5. 最後に
以上で設定した時間おきに約定を確認し、約定してあればLINEが飛ぶようになっています。
利確時は朗報ですが、損失確定のお知らせの時はただムカつくだけです。
目的としては仕事中とかポジションを持っている時にわざわざサイトを見て確認する必要がないところです。
また、今後Lambdaを使って価格通知とかも載せていこうかと思います。
冒頭でいっていたRealTimeAPIも実装してみようかと思います。(こちらはnode or Pythonのオンプレの予定)
【AWS】CodePipelineを使ってcodeCommitのプルリクでLamdaをデプロイ
自分の備忘録的なものです。AWS公式をキャプチャ付きに簡単にまとめたものです。文章をだらだら読むより視覚的に理解したい人用です。
https://docs.aws.amazon.com/ja_jp/lambda/latest/dg/build-pipeline.html
1. AWS CloudFormation ロールを作成する
ここで予め、デプロイ時に必要となるロールを作成します。
コンソールからIAMへ行きます。
メニューから【Roles】を選択します。
trusted EntityでAWS Serviceを選択し、use Caseとして、下の方にある【Cloud formation】を選択し、【Next Permission】を選択
【Create policy】を選択し、ポリシーを作成します。
【JSON】タブを選択して以下をコピペします。
{
"Statement": [
{ "Action": [ "apigateway:*",
"codedeploy:*",
"lambda:*",
"cloudformation:CreateChangeSet",
"iam:GetRole",
"iam:CreateRole",
"iam:DeleteRole",
"iam:PutRolePolicy",
"iam:AttachRolePolicy",
"iam:DeleteRolePolicy",
"iam:DetachRolePolicy",
"iam:PassRole",
"s3:GetObject",
"s3:GetObjectVersion",
"s3:GetBucketVersioning"
],
"Resource": "*",
"Effect": "Allow"
}
],
"Version": "2012-10-17"
}
ポリシーに名前を付けます。
AWS公式の例だとcfn-lambda-pipelineとなっています。
2.リポジトリをセットアップする
codeCommitで例としてlambda-pipeline-repoというリポジトリを作成しているとします。
そこの直下に以下のファイルを入れます。index.js
var time = require('time'); exports.handler = (event, context, callback) => { var currentTime = new time.Date(); currentTime.setTimezone("America/Los_Angeles"); callback(null, { statusCode: '200', body: 'The time in Los Angeles is: ' + currentTime.toString(), }); };
ソースファイルになり、現在の時刻を返すLambda関数になります。
buildspec.yml
version: 0.2
phases:
install:
runtime-versions:
nodejs: 10
build:
commands:
- npm install time
- export BUCKET=ここは自身のS3バケットを指定する
- aws cloudformation package --template-file template.yml --s3-bucket $BUCKET --output-template-file outputtemplate.yml
artifacts:
type: zip
files:
- template.yml
- outputtemplate.ymlビルド時に必要なパッケージをインストールしたりするファイル ※BUCKETの部分は自身のS3のバケット名に置き換えてください。
template.yml
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: Outputs the time
Resources:
TimeFunction:
Type: AWS::Serverless::Function
Properties:
Handler: index.handler
Runtime: nodejs10.x
CodeUri: ./
Events:
MyTimeApi:
Type: Api
Properties:
Path: /TimeResource
Method: GETデプロイ時の環境を定義するファイル。APIゲートウェイにURLを指定して展開するとこまで書かれています。
故にこのファイルが無事実行されれば、URLを叩けば↑のLambdaが実行されることになります。
全て直下に置くので、ファイル構成をまとめると以下のようになります。
※ソースは別に直下であればどんな構成でも構いません。
3. CodepipelineでPipelineを作成する
コンソールから【CodePipeline】を検索して選択します。
【Create Pipeline】を選択します。
Pipelineの名前を入力します。例ではlambda-pipelineとなっています。
また、ロールは新しく作ることにしますので、【New service role】を選択します。既存のロールがある場合は【Existing service role】選択して下さい。
ソースの詳細を選択していきます。
まず、Source providerとして例としてAWS Code Commitを選択します。選択すると自動的にリポジトリ名入力などの入力欄が出てきます。
クリックするとドロップダウンで存在するレポジトリやブランチを選択できます。
今回は既に作成しているlambda-pipeline-repoとブランチとしてmasterを選択します。
Change detection optionsとしてCloud Watch Eventを選択することで、リポジトリの更新でCodePipelineが走るようになります。
ビルドの詳細を選択していきます。ソース同様、Build providerでAWS CodeBuildを選択するとリージョンの選択とプロジェクト名の入力が出ます。
プロジェクトはまだ作成していないので、【Create project】を選択します。
ビルドの設定を作成します。
Buildの名前を入力します。例ではlambda-pipeline-buildとなっています。
Environmentを以下のように設定
・Operating system ・・・ Ubuntu
・Runtimes ・・・ Standard
・image ・・・ aws/codebuild/standard:2.0
・image version ・・・ 最新
・Environment type ・・・ Linux
他はデフォルトで構いません。ロールも新しく作成する設定にしておきます。
ビルドファイルの名前を設定する。git上のフォルダ直下に存在するファイル名にします。例だとbuildspec.ymlです。
CodePipelineに戻ります。先ほどのBuild provider選択に作成したlambda-pipeline-buildが選択出来るようになるので選択します。
※表示されない場合は更新デプロイの設定をします。
それぞれ以下のように設定します。
・Deploy provider ・・・ CloudFormation
・Region ・・・ お好きなリージョン
・Action mode ・・・ Create or replace a changeset
・Stack name ・・・ lambda-pipeline-stack
・Change set name ・・・ lambda-pipeline-changeset
Stack nameとChange set nameは入力した名前で作成されます。
Templateの項目を埋めます。指定した名前のファイルがS3上に作成されます。
・Artifact name ・・・ BuildArtifact
・File name ・・・ outputtemplate.yml
Capabilities - optionalは【CAPABILITY_IAM】を選択します。
Roleは一番最初にIAMで作成したcfn-lambda-pipelineを選択します。
4.ビルドステージロールを更新する
IAMコンソールのRoleからビルドプロジェクト作成時に作成したロールcode-build-lambda-pipeline-service-roleを選択します。
ポリシーをアタッチします。S3にファイルを添付するのでS3のフルアクセスをアタッチします。
5.デプロイステージの完了
CodePipelineに戻り、作成済のパイプラインであるlambda-pipelineを選択します。
メニューにある【edit】を選択し、編集していきます。
デプロイにある【edit deploy】を選択してデプロイを編集していきます。
【Add action group】を選択します。
以下のように設定します。
・Action name ・・・ excute-changeset
・Action provider ・・・ AWS Cloudformation
・Region ・・・ お好きなリージョン
・Input artifacts ・・・ BuildArtifact
・Action mode ・・・ Excute a changeset
・Stack name ・・・ lambda-pipeline-stack
・Change set name ・・・ lambda-pipeline-changeset
6.アプリケーションのテスト
ここまでで、CodePipelineの設定は終了です。ここまでで、エラー等あれば、指定しているproviderや操作しているIAMユーザーのロールなどを確認してください。基本的にエラー内容に従えば解決するはずです。
アプリケーションのテストの前にCodePipelineの実行を行います。【Release Change】を選択して実行します。
※初回だと変更セット実行で変更セットがないというエラーが出ることがあるが、再実行で解決します。
アプリケーションのテストを行う為、コンソールからLamdaへ移ります。
サイドメニューから【Application】を選択すると、CodePipelineで作成したlambda-pipeline-stackが表示されます。
下の方へ行き、Resourcesにある【Serverless RestApi】を展開し、【Prod API endpoint】を選択します。
エラーメッセージが出ますので、URLのパスにTimeResourceを追加します。
すると、時間が表示されるのが確認できます。
以上で、CodePipelineでのLambdaのデプロイは終わりです。複数ラムダを展開したい場合は、template.ymlをいじればできます。
その際、ラムダ毎にインストールするパッケージを変えたい場合は、buildspec.ymlを上手くいじる必要あります。
何かアプリを作る時など機会があれば備忘録として書こうと思います。
※2020/3/22現在、デプロイステージで変更セットの実行がたまに失敗します。原因が分かれば追記します。再実行で解決するので今のとこ個人での使用であればそんなに影響はないと思います。
※2020/4/21追記
今更ながらしっかりエラー見て気づきましたが、Deployの同ステージ内で変更セットの作成と実行を同時にやってしまっているようでした。
正しくは【Add action group】で、変更セット実行は後にすると失敗はしませんでした。
