Pythonでビットコインの自動取引をする

前回はシミュレーターでの自動取引を実際の取引所での自動取引に対応しやすくするためにプログラムの改修を行いました。
今回はそれを更に進めていよいよ実際の取引所での自動取引を実現します。

実現しますがアルゴリズムがしょぼくて勝てないことはシミュレーションで確認済みなので実行は動作確認にとどめたいと思います…。

前回準備を行ったのでやるべきことはそう多くはありません。

実際の取引所であるGMOコインとやり取りするApiクラスであるGmoCoinApiクラスを作成する。
TraderクラスにGmoCoinApiクラスを渡して3秒毎にTraderクラスのact()メソッドを実行するメインプログラムを作成する。

順番に進めていきます。

GmoCoinApiクラスの作成

Apiクラスを継承し、GMOコインのAPIを呼んで、実際の取引所であるGMOコインとやり取りするGmoCoinApiクラスを作成します。

長くなるのでソースコードは割愛します。
GMOコインのAPIの呼び出しについては
Pythonでビットコインの価格を取得する(1) - てけとーぶろぐ。
を参考にしていただければと思います。

2点アドバイスがあります。

GMOコインのAPI呼び出しの際のリクエストとレスポンスをログファイルにロギングすると問題が起きたときの調査に役立ちます。
GMOコインのAPI呼び出しが正常終了しなかった場合は例外を投げるといいかと思います。例外が投げられたらプログラムを停止していいかと思いますが、GMOコインのメンテナンスで失敗するケースについては特別に扱ってプログラムが停止しないようにしてもいいかもしれません。プログラムを長く動かしていると割とメンテナンスに遭遇すると思いますので。

メインプログラムの作成

メインプログラムを作成します。概要としては以下の通りです。

SimulatorApiクラスではなくGmoCoinApiクラスを渡してTraderクラスのインスタンスを作成します。
apschedulerのBackgroundSchedulerクラスを使って3秒毎にTraderクラスのact()メソッドを実行します。
シミュレーターと違って長時間の実行になり終始見張っておけないでしょうから、ログファイルにログを残すようにします。またエラー時にメール通知も行うようにします。

from gmo_coin_api import GmoCoinApi
from trader import Trader
from apscheduler.schedulers.background import BackgroundScheduler
import logging.config
from app_const import AppConst
import os
from email.mime.text import MIMEText
from email.utils import formatdate
import smtplib
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")

logging.config.fileConfig(AppConst.LOGGING_CONFIG_FILE_PATH)
logger = logging.getLogger('fileLogger')

gmo_coin_api = GmoCoinApi('https://api.coin.z.com')
trader = Trader(gmo_coin_api)

def send_mail(subject, body):
    msg = MIMEText(body)
    msg['Subject'] = subject
    msg['From'] = GMAIL_USER
    msg['To'] = TO_ADDRESS
    msg['Bcc'] = ''
    msg['Date'] = formatdate()

    smtpobj = smtplib.SMTP_SSL('smtp.gmail.com', 465, timeout=10)
    smtpobj.login(GMAIL_USER, GMAIL_PASSWORD)
    smtpobj.sendmail(GMAIL_USER, TO_ADDRESS, msg.as_string())
    smtpobj.close()

def update():
    global is_exception_raised
    try:
        trader.act()
    except Exception as exception:
        logger.error('exception:\n{}'.format(exception))
        send_mail('Trade Exception Alert', 'exception:\n{}'.format(exception))
        is_exception_raised = True

if __name__ == '__main__':

    # apscheduler のロギングを抑制
    logging.getLogger('apscheduler.executors.default').setLevel(logging.WARNING)
    logging.getLogger('apscheduler.scheduler').setLevel(logging.WARNING)

    scheduler = BackgroundScheduler()
    scheduler.add_job(update, trigger='interval', seconds=3, max_instances=10)
    scheduler.start()

    is_exception_raised = False

    try:
        while True:
            # TODO 監視システムに生きていることを知らせる
           
            if is_exception_raised:
                break

            sleep(0.001)

    except KeyboardInterrupt:
        logger.info('KeyboardInterrupt')

    scheduler.shutdown()