Zenn
🐍

PythonとGCP Cloud Functionsで指定した企業の株価データとチャート図を定期的にSlack通知する

2021/05/01に公開
Docker
GCP
Python
Slack
pandas
tech

（※6095の銘柄でSlack通知させたイメージ）

正月休みでノリで作ったスクリプトについて書いていきます。

やりたいこと

  • 指定した企業の株価を定期的にSlackに通知したい
  • 推移を比較したいので、一定期間の株価の推移のグラフが欲しい
  • できれば無料でやりたい

背景

  • 自社の株価の推移や時価総額をなんとなく知っておきたかった
    • 投資目的ではなかったので、だいたいデイリーくらいで追えればよい
  • これまでは気が向いたら株価をググっていたが、めんどくさいのでSlackに通知してほしいと思った

https://twitter.com/terry_i_/status/1340165963409358856?s=20

使用技術

  • Python(3.9)
    • pandas-datareader
    • mplfinance(matplotlib)
  • GCP
    • Cloud Functions
    • Cloud Schedulder
    • Cloud Pub/Sub
  • Docker(開発環境)
  • Slack bot API

リポジトリ

  • 本リポジトリをCloneしてCloud Functionsにデプロイ→環境変数を設定すれば、Slackに通知されるようになります

https://github.com/f-teruhisa/stooqifier

実装

概要

  • 大まかな構成は上記の通りで、Cloud FunctionsでPythonスクリプトを動かしてデータ取得→Slack通知という流れ
    • Cloud Schedulerでcronを定期実行し、Cloud Pub/Subにメッセージを送信
    • Cloud Pub/Subのキューイングでメッセージを受信をトリガーにCloud Functions実行
    • Cloud Functionsに登録したPythonスクリプトを実行する

API選定

  • 色々と調べた結果、無料でAPIを叩けてかつデイリーで株価を取得できるStooqというサイトのAPIを使用した
  • Stooqはpandas関連のpipライブラリがあり、API利用が比較的容易にできそうだった
    • これにより、使用する言語はPythonに自動的に決定した

https://pandas-datareader.readthedocs.io/en/latest/readers/stooq.html

スクリプト

ということで、作ったスクリプトのメインコードがこちら。

https://github.com/f-teruhisa/stooqifier/blob/master/main.py

以下でポイントを解説していきます。

株価データ取得

  • 以下のメソッドでStooqのAPIを用いて株価を取得しています
  • 後ほどチャート図を生成するために、取得したデータを一度CSVファイルにしています
    • Cloud FunctionsではCSVファイルの書き込みはtmpディレクトリでないと不可なので注意
import pandas_datareader.stooq as stooq

def generate_csv_with_datareader():
    """
    Generate a csv file of OHLCV with date with stooq API
    """
    # 株価推移の開始日を指定(3ヶ月を指定)
    start_date = today - relativedelta(months=3)
    # Stooqのライブラリ経由でAPIを叩く(stock_codeは環境変数で株コードを指定)
    stooq_reader = stooq.StooqDailyReader(stock_code, start=start_date, end=today)
    # APIで取得したデータを一旦CSVファイルにする
    stooq_reader.read().to_csv(f"/tmp/{str(today)}.csv")

チャート図の生成

import pandas as pd
import mplfinance as mpf

def generate_stock_chart_image():
    """
    Generate a six-month stock chart image with mplfinance
    """
    dataframe = pd.read_csv(f"/tmp/{str(today)}.csv", index_col=0, parse_dates=True)
    # 推移チャートを作るために日付でデータをソートする
    dataframe = dataframe.sort_values('Date')
    mpf.plot(dataframe, type='candle', figratio=(12,4),
         volume=True, mav=(5, 25), style='yahoo',
         savefig=f"/tmp/{str(today)}.png")

Slack通知

  • 取得したCSVの(ヘッダーを覗いた)一番上の行が最新の株価情報なので、その内容をSlack通知用のSlackクラスに渡す
    • Cloud Functionsではmain関数を呼び出しているが、引数にeventとcontextを指定しないと動かないので指定しています
import csv

def main(event, context):
    """
    The main function that will be executed when this Python file is executed
    """
    generate_csv_with_datareader()
    generate_stock_chart_image()
    with open(f"/tmp/{str(today)}.csv", 'r', encoding="utf-8") as file:
        # Skip header row
        reader = csv.reader(file)
        header = next(reader)
        for i, row in enumerate(csv.DictReader(file, header)):
            # Send only the most recent data to Slack notification
            if i == 0:
                slack.Slack(today, row).post()

また、Slack通知クラスは別ファイルに切り出している（後々LINEなどの他の通知先を追加しやすいようにするため）。Slackクラスのpostメソッドを実行時にrequestsでSlackのAPIを叩く。

class Slack():
    # 略
    def post(self):
        """
        POST request to Slack file upload API
        API docs: https://api.slack.com/methods/files.upload
        """
        file = {'file': open(f"/tmp/{str(self.date)}.png", 'rb')}
        requests.post(url="https://slack.com/api/files.upload",params=self.params, files=file)

使用したSlackのAPIはファイルアップロードAPI。画像ファイルをSlackにPOSTする必要があるため。requestsのpostメソッドのfilesに添付したいファイルを指定すればOK。

https://api.slack.com/methods/files.upload

株価のデータはOHLC(+Volume)というフォーマットで共通化されているので、これをゴニョゴニョして送信時のテキストを作っている。

  def __format_text(self, ohlcv):
        """
        Create params data for sending Slack notification with API.
        :param dict[str, str, str, str, str, str] ohlcv:
        :type ohlcv: {
            'Date': '2020-12-29',
            'Open': '7620',
            'High': '8070',
			'Low': '7610',
			'Close': '8060',
			'Volume': '823700'
        }
        :return: String
        """
        text = f"本日は{self.date.strftime('%Y年%m月%d日')}です。\n" \
               f"取得可能な最新日付の株価情報をお知らせします。 \n\n"\
               f"*始値* {int(ohlcv['Open']):,d}\n" \
			   f"*高値* {int(ohlcv['High']):,d}\n" \
			   f"*安値* {int(ohlcv['Low']):,d}\n" \
			   f"*終値* {int(ohlcv['Close']):,d}\n" \
			   f"*出来高* {int(ohlcv['Volume']):,d}"
        return text

インフラ

gcloudコマンドで関数をCloud Functionsにデプロイ

  • Cloud Functionsのプロジェクトを作成し、その環境にPythonの関数スクリプトをデプロイする
    • 先んじてCloud Pub/Subのトピックを作成し、デプロイ時に指定できるようにしておく
  • gcloudコマンドで以下のように指定すればCLIからデプロイ可能
$ gcloud functions deploy #{FUNCTION_NAME} --entry-point main --project #{PROJECT_ID} --region #{REGION} --runtime python39 --trigger-topic #{PUBUSB_TOPIC_NAME}

https://cloud.google.com/sdk/gcloud/reference/functions/deploy

環境変数の設定

  • Cloud Functions上でファイルを編集し、.sample.env ファイルの名前を.envに変更して環境変数を設定(.envはgitignoreしているため)
  • STOCK_CODEには取得したい株価の銘柄コードを指定
  • SLACK_API_TOKENにはSlack botを作成して取得できるトークンを指定
  • SLACK_CHANNEL_ID はSlackのチャンネルのIDを指定 
# .sample.env => .env
STOCK_CODE=
SLACK_API_TOKEN=
SLACK_CHANNEL_ID=

Cloud Schedulerでジョブの設定

  • Cloud Schedulerのジョブを作成する(画像は平日の正午にジョブを実行する設定例)
  • 作成したPub/Subに対してメッセージを送信することで、Cloud Functionsの実行をトリガーする

https://cloud.google.com/scheduler/docs/quickstart#create_a_job

備考

  • Python 3.9のランタイムをCloud Functionsがbetaで使用できるようにしてくれたのでいち早く使ってみた
    • 結局3.9特有の機能は使いませんでした
    • 2021/05/01現在、AWS Lambdaでは残念ながらPython 3.9をサポートしていないので意図せずCloud Functions専用スクリプトになってしまった(AWS Lambda LayersでECRにプッシュしたイメージを起動するようにすれば動かせるはず)
    • ローカル環境によってはPythonのバージョンを汚したくない等あると思い、Dockerを使った
  • Cloud Functionsがサポートしている関係上、Pythonの依存関係をPipenvではなくrequirements.txtで管理している

参考記事

  • 作る途中でお世話になった記事を置いてきます

https://dev.classmethod.jp/articles/try-cloud-functions-scheduler-pubsub/

https://kiseno-log.com/2020/06/29/pythonで日本株の株価を取得してローソク足の描画/

https://qiita.com/kai_kou/items/dca21cdfd8375a247c2f

Discussion