「Raspberry Pi + Pythonでビットコイン自動取引」を目指して
まずは株価アラート、の2回目。
前回はアラート条件をソースコードに埋め込んでしまっていたので
それを外部のアラート条件定義ファイルで定義できるようにする。
複数銘柄の監視にも対応させる。
アラート条件定義ファイル
アラート条件定義ファイルは、手での編集のしやすさ、プログラムからの扱いやすさ、表現力を考えて以下のようなjsonファイルとする。
{ "alertConditions": [ { "alertConditionId": "f870215c-944a-11ea-8747-185e0fdcc982", "stockCode": "4755", "conditionType": "isEqualToOrHigher", "values": [ 1000 ], "isEnabled": true }, { "alertConditionId": "f870215d-944a-11ea-80ac-185e0fdcc982", "stockCode": "6758", "conditionType": "isEqualToOrHigher", "values": [ 5000 ], "isEnabled": false } ] }
各プロパティの説明は以下の通り。
これで複数銘柄のアラート条件を定義できる。
条件の種類としてはとりあえず基本的な「X円以上になったら」「X円以下になったら」の2種類に対応させることにする。
| プロパティ | 説明 |
|---|---|
| alertConditions | アラート条件の配列 |
| alertConditionId | アラート条件のID |
| stockCode | 銘柄コード |
| conditionType | アラート条件の種類 "isEqualToOrHigher":X円以上になったら "isEqualToOrLower":X円以下になったら |
| values | アラート条件に使う値の配列(現状1つの値しか使わない) |
| isEnabled | 有効か否か(このアラート条件について監視を行うか) |
アラート条件のクラス化
ファイルのフォーマットが決まったので
まずはファイルを扱うクラスを「alert_condition.py」としてつくる。
dataclass, dataclasses_jsonを使うと
そのクラスのインスタンスをjsonにしたり
jsonからそのクラスのインスタンスをつくったり
といった機能を持つクラスが簡単につくれる。
そこでdataclass, dataclasses_jsonを使ってアラート条件定義ファイルのjsonに対応したAlertConditionクラス、AlertConditionListクラスを作成する。
これだけの定義でもう alert_condition_list.to_json() でクラスのインスタンスをjsonにしたり、AlertConditionList.from_json() でjsonからクラスのインスタンスをつくったりできてしまう。
from typing import List, Dict from dataclasses import dataclass from dataclasses_json import LetterCase, dataclass_json import uuid from collections import OrderedDict class AlertConditionType: IS_EQUAL_TO_OR_HIGHER = 'isEqualToOrHigher' IS_EQUAL_TO_OR_LOWER = 'isEqualToOrLower' @dataclass_json(letter_case=LetterCase.CAMEL) @dataclass class AlertCondition: alert_condition_id: str stock_code: str condition_type: str values: List[int] is_enabled: bool @dataclass_json(letter_case=LetterCase.CAMEL) @dataclass class AlertConditionList: alert_conditions: List[AlertCondition] class AlertConditionUtil: JSON_FILE_NAME = 'conditions.json' @classmethod def save(cls, alert_condition_list: AlertConditionList): with open(cls.JSON_FILE_NAME, 'w', encoding='utf-8') as f: f.write(alert_condition_list.to_json(indent=4, ensure_ascii=False)) @classmethod def load(cls) -> AlertConditionList: with open(cls.JSON_FILE_NAME, 'r', encoding='utf-8') as f: return AlertConditionList.from_json(f.read()) @staticmethod def generate_id() -> str: return uuid.uuid1() if __name__ == '__main__': ac1 = AlertCondition(AlertConditionUtil.generate_id(), '4755', AlertConditionType.IS_EQUAL_TO_OR_HIGHER, [1000], True) ac2 = AlertCondition(AlertConditionUtil.generate_id(), '6758', AlertConditionType.IS_EQUAL_TO_OR_HIGHER, [5000], True) acl = AlertConditionList([ac1, ac2]) AlertConditionUtil.save(acl) new_acl = AlertConditionUtil.load() print(new_acl.to_json(indent=4, ensure_ascii=False))
あとはjsonファイルへの保存、jsonファイルからの読み込み、アラート条件IDの生成といったユーティリティメソッドを持ったAlertConditionUtilクラスもつくっておく。
実行してみるとアラート条件定義ファイル「conditions.json」が作成される。
アラート条件に従った監視
アラート条件定義ファイルからアラート条件を読み込むことができるようになったので
このアラート条件を使って株価の監視をしてみる。
import os from apscheduler.schedulers.background import BackgroundScheduler from mail_sender import MailSender from stock_price import StockPrice from time import sleep from alert_condition import AlertConditionUtil, AlertConditionType import datetime GMAIL_USER = os.environ.get("GMAIL_USER") GMAIL_PASSWORD = os.environ.get("GMAIL_PASSWORD") TO_ADDRESS = os.environ.get("TO_ADDRESS") mail_sender = MailSender(GMAIL_USER, GMAIL_PASSWORD, TO_ADDRESS) is_condition_enabled = True is_exception_raised = False def update(): global is_exception_raised acl = AlertConditionUtil.load() # 有効なアラート条件の銘柄コードのユニークなリストを取得 stock_codes = [ac.stock_code for ac in acl.alert_conditions if ac.is_enabled] stock_codes = list(set(stock_codes)) # 株価を取得 ticker_dict = {} try: for stock_code in stock_codes: ticker = StockPrice.get_ticker(stock_code) ticker_dict[stock_code] = ticker except Exception as exception: mail_sender.send_mail('update() Exception Alert', f'exception:\n{exception}') is_exception_raised = True return # アラート条件の確認 for alert_condition in acl.alert_conditions: if not alert_condition.is_enabled: continue stock_code = alert_condition.stock_code values = alert_condition.values ticker = ticker_dict[stock_code] if alert_condition.condition_type == AlertConditionType.IS_EQUAL_TO_OR_HIGHER: if ticker.price >= values[0]: mail_sender.send_mail(f'[株価アラート] 銘柄コード:{stock_code}', f'銘柄コード:{stock_code}の株価が{values[0]}円以上になりました。') alert_condition.is_enabled = False elif alert_condition.condition_type == AlertConditionType.IS_EQUAL_TO_OR_LOWER: value = alert_condition.values[0] if ticker.price <= value: mail_sender.send_mail(f'[株価アラート] 銘柄コード:{stock_code}', f'銘柄コード:{stock_code}の株価が{values[0]}円以下になりました。') alert_condition.is_enabled = False AlertConditionUtil.save(acl) if __name__ == '__main__': scheduler = BackgroundScheduler() # 毎分0秒に実行する scheduler.add_job(update, 'cron', second=0, max_instances=10) scheduler.start() try: while True: if is_exception_raised: break sleep(0.001) except KeyboardInterrupt: pass scheduler.shutdown()
前回作成した監視と基本は同じ。主な違いは以下の点。
- アラート条件をアラート条件定義ファイルから読み込んだものを使っていること
- 複数銘柄対応のためにループしていること
それから一定時間ごとの処理は毎分0秒のタイミングで実行するようにした。
apschedulerだとこの変更がパラメーターの変更だけでできてしまう。
実行すると、プログラムは毎分0秒のタイミングでアラート条件に従った監視を行い、条件を満たすとその旨を知らせるメールを送信する。
一度条件を満たしたアラート条件については無効になる。
だいぶ実用的になってきたのではないだろうか。
しかし実際に使うとなると、プログラムを実行しつづけなくてはならない。
PCをずっと起動しておくの…？
電気代大丈夫…？
ということで、次回はどう動かし続けるかについて書きたい。
