「Raspberry Pi + Pythonでビットコイン自動取引」というところを目指して
何回かにわたって記事を書いていこうと思う。
最終的にはビットコインなのだけど
まずは手始めにより多くの人に馴染みがある日本の株式を扱う。
プログラムで好きな銘柄の株価を取得して
取得するだけではつまらないので
株価が設定した条件を満たしたらメールで通知するようにしてみよう。
証券会社のサイトにも似たような機能があるとは思うが
プログラムを書けばより高度な条件も設定できる。
環境
PythonはPython Releases for Windows | Python.orgから
「Python 3.7.3 - March 25, 2019」の「Windows x86-64 executable installer」をダウンロードしてインストールしている。
株価の取得
株探（かぶたん）からBeautifulSoup4を使ってスクレイピングで株価を取得する。
ある銘柄の株価は例えば銘柄コード「7860」であれば
https://kabutan.jp/stock/kabuka?code=7860
というURLから見られるようになっている。
そこでここから株価を得るクラスを「stock_price.py」としてつくる。
from bs4 import BeautifulSoup from urllib import request import datetime from dataclasses import dataclass @dataclass class Ticker: code: str price: int time: datetime.datetime opening_price: int high_price: int low_price: int closing_price: int from_the_day_before: int from_the_day_before_percentage: float volume: int class StockPrice: @staticmethod def get_ticker(code: str): url = 'https://kabutan.jp/stock/kabuka?code=' + code response = request.urlopen(url) soup = BeautifulSoup(response, 'html.parser') response.close() # 現在の株価と時刻は画面上部のstockinfo_i1から取得する div = soup.find('div', id='stockinfo_i1') span = div.find('span', class_='kabuka') time = div.find('time') # 時刻以外はstock_kabuka0クラスのテーブルから取得する table = soup.find('table', class_='stock_kabuka0') tbody = table.find('tbody') tds = tbody.find_all('td') if len(tds) != 7: raise RuntimeError('Unexpected HTML Structure.') price = int(span.text.replace(',', '').replace('円', '')) dt = datetime.datetime.fromisoformat(time.attrs['datetime']) opening_price = int(tds[0].text.replace(',', '')) high_price = int(tds[1].text.replace(',', '')) low_price = int(tds[2].text.replace(',', '')) closing_price = int(tds[3].text.replace(',', '')) from_the_day_before = int(tds[4].text.replace(',', '').replace('+', '')) from_the_day_before_percentage = float(tds[5].text.replace(',', '').replace('+', '')) volume = int(tds[6].text.replace(',', '')) daily_ticker = Ticker(code, price, dt, opening_price, high_price, low_price, closing_price, from_the_day_before, from_the_day_before_percentage, volume) return daily_ticker if __name__ == '__main__': print(StockPrice.get_ticker('1775'))
株価が取得できるか実行してみる。
> python .\stock_price.py Ticker(code='1775', price=1798, time=datetime.datetime(2020, 5, 8, 15, 0, tzinfo=datetime.timezone(datetime.timedelta(seconds=32400))), opening_price=1770, high_price=1800, low_price=1740, closing_price=1798, from_the_day_before=33, from_the_day_before_percentage=1.9, volume=11300)
取得できた。
メール通知
続けてメール通知をできるようにする。
メール送信にはGmailアカウントを使うことにする。
今回はGmailアカウントを使って自作プログラムからのメール送信を簡単に行うために
「安全性の低いアプリのアクセス」を許可することにする。
Webブラウザーで以下の操作を行う。
- メール送信に使うGmailアカウントでログイン
- Googleアカウントの「セキュリティ」→「安全性の低いアプリのアクセス」から安全性の低いアプリのアクセスをオンにする。
「mail_sender.py」としてメール送信を行うクラスをつくる。
import smtplib from email.mime.text import MIMEText from email.utils import formatdate import os class MailSender: def __init__(self, gmail_user, gmail_password, to_address): self._gmail_user = gmail_user self._gmail_password = gmail_password self._to_address = to_address def send_mail(self, subject, body): msg = MIMEText(body) msg['Subject'] = subject msg['From'] = self._gmail_user msg['To'] = self._to_address msg['Bcc'] = '' msg['Date'] = formatdate() smtpobj = smtplib.SMTP_SSL('smtp.gmail.com', 465, timeout=10) smtpobj.login(self._gmail_user, self._gmail_password) smtpobj.sendmail(self._gmail_user, self._to_address, msg.as_string()) smtpobj.close() if __name__ == '__main__': 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) mail_sender.send_mail('subject', 'body')
メール送信ができるか実行してみるのだがその前に
このプログラムはOSの環境変数からGmeilアカウントと送信先メールアドレスを取得するようになっている。
そこであらかじめ環境変数に値を設定しておく必要がある。
以下の環境変数を設定する。
| 環境変数名 | 説明 |
|---|---|
| GMAIL_USER | メール送信に使うGmailアカウントのメールアドレス |
| GMAIL_PASSWORD | メール送信に使うGmailアカウントのパスワード |
| TO_ADDRESS | 送信先メールアドレス |
コマンドプロンプトであれば以下のように設定できる。
> set GMAIL_USER=hoge@gmail.com
PowerShellであれば以下の通り。
> $env:GMAIL_USER="hoge@gmail.com"
準備ができたら実行する。
> python .\mail_sender.py
おそらく初回は送信できない。
その代わりに、メール送信に使うGmailアカウントのGmail受信トレイに
「不審なログインの試みをブロックしました」
といったメールが届いている。
これがまさに自作プログラムからのログインであるということであれば
「心当たりがある」とすることで以降ブロックされないようになる。
簡単な株価の監視
株価の取得とメール送信ができたので
それらを使って簡単な株価の監視をつくる。
一定時間ごとに指定の銘柄の株価の取得を行い
株価が指定の価格以上になったらメール通知するようにする。
一定時間ごとの実行は以下のような単純なものでもいいのだが
if __name__ == '__main__': while True: update() sleep(3)
後で拡張しやすいようにapschedulerのBackgroundSchedulerクラスを使う。
「main_single_code.py」として作成する。
import os from apscheduler.schedulers.background import BackgroundScheduler from mail_sender import MailSender from stock_price import StockPrice from time import sleep 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_condition_enabled global is_exception_raised if not is_condition_enabled: return try: ticker = StockPrice.get_ticker('1775') if ticker.price >= 100: mail_sender.send_mail('株価アラート', '株価が条件を満たしました') is_condition_enabled = False except Exception as exception: mail_sender.send_mail('StockPrice.get_ticker() Exception Alert', f'exception:\n{exception}') is_exception_raised = True if __name__ == '__main__': scheduler = BackgroundScheduler() scheduler.add_job(update, trigger='interval', seconds=3, max_instances=10) scheduler.start() try: while True: if is_exception_raised: break sleep(0.001) except KeyboardInterrupt: pass scheduler.shutdown()
実行すると銘柄コード「1775」の株価が100円以上になったら
指定のメールアドレスにメールが送られる。
次回はアラートの条件を外部ファイルで設定できるようにする。
