「Raspberry Pi + Pythonでビットコイン自動取引」というところを目指して
何回かにわたって記事を書いていこうと思う。
最終的にはビットコインなのだけど
まずは手始めにより多くの人に馴染みがある日本の株式を扱う。
プログラムで好きな銘柄の株価を取得して
取得するだけではつまらないので
株価が設定した条件を満たしたらメールで通知するようにしてみよう。
証券会社のサイトにも似たような機能があるとは思うが
プログラムを書けばより高度な条件も設定できる。
株価の取得
株探（かぶたん）から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()
div = soup.find('div', id='stockinfo_i1')
span = div.find('span', class_='kabuka')
time = div.find('time')
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アカウントと送信先メールアドレスを取得するようになっている。
そこであらかじめ環境変数に値を設定しておく必要がある。
以下の環境変数を設定する。
コマンドプロンプトであれば以下のように設定できる。
> 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円以上になったら
指定のメールアドレスにメールが送られる。
次回はアラートの条件を外部ファイルで設定できるようにする。
