てけとーぶろぐ。

ソフトウェアの開発と、お絵かきと、雑記と。

Kindleの電子書籍をPDFファイルにする

電子書籍は便利なのですが
自分は電子ペーパーで読みたくて
持っている Kindle Paperwhite だと画面が小さい。

本によっては問題ないのですが
読みたい本の多くは大きな画面でないと読みづらく
それがために渋々紙の本をかうことが何度も。

比較的安価で画面の大きなQUADERNOとかで
Kindle電子書籍が読めたらもっと世界が広がるのになぁ。

ということで
Kindle電子書籍をPDFファイルにするツールをつくりました。

Reader Capture - Kindleの電子書籍をPDFファイルに変換!

Kindle電子書籍
Kindle for PC (Windows)のスクリーンショットを撮ることで
PDFファイルにします。

ちょっと調べると他にもやり方はあるようですが
簡単に使えること、使い続けられること、を重視してつくりました。

仕組みは汎用的なので他の電子書籍リーダーアプリでも使えるかもしれません。
保証はしませんけど…。

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()


3秒毎に特定の処理を実行するところは
Pythonでビットコインの価格を取得する(1) - てけとーぶろぐ。
のプログラムと同じです。

メール通知については
Pythonで株価アラートをつくる(1) - てけとーぶろぐ。
で扱っていますので参考にしていただければと思います。

プログラムの実行

プログラムを実行してみます。
しばらく放置して戻ってきてみると売買の取引が自動で行われていました。
GMOコインにログインして取引履歴を確認しても確かに取引を行っています。


ということで当初の目的である自動取引が実現できました。
暇を見てアルゴリズムの改善など行っていこうと思います。

Pythonでビットコインの自動取引をシミュレーションする(2)

前回は簡単な自動取引のアルゴリズムをつくって
シミュレーター上でそのアルゴリズムを動かしてみました。
取引の結果は大敗でしたがシミュレーション自体はできました。

ここからアルゴリズムを改良していくのもいいのですが
記事としては自動取引の実現を優先したいと思います。

今回は前回作成したプログラムに手を入れて
実際の取引所での自動取引に対応しやすくします。

実際の取引所での取引への対応とは?

実際の取引所での取引への対応とはどういうことでしょうか?
これはシミュレーションに使った自動取引のアルゴリズムをそのまま使って
実際の取引所で取引ができるようにすることと言えます。

前回作成したプログラムで言えば
自動取引のアルゴリズムはTraderクラスとして実装しました。

TraderクラスがSimulatorクラスとやり取りをして取引をしていました。
図にすると以下になります。

f:id:kurimayoshida:20200717214141p:plain
Simulatorクラスとのやり取り

これをTraderクラスはそのままに、やり取りの相手を取引所にして取引ができればいいわけです。
図にすると以下になります。

f:id:kurimayoshida:20200717214206p:plain
取引所とのやり取り

Apiクラスの追加

Traderクラスをそのままにやり取りの相手をシミュレーターにしたり取引所にしたりできればいいと言ったのですが、そのためには前回のプログラムから少し改修が必要です。
なぜならSimulatorクラスと取引所ではやり取りの口、すなわちインターフェースが違うからです。
例えばSimulatorクラスは get_cur_ticker() メソッドを呼ぶことで現在のビットコインの価格を返すようになっていましたが、取引所は、この連載で使っているGMOコインであればWeb API の GET /public/v1/ticker を呼ぶことで現在のビットコインの価格を返します。
また今のTraderクラスはシミュレーターとのやり取りに特化した作りになってしまっています。

これらに対してApiクラスを追加して対応します。
TraderクラスがSimulatorクラスや取引所とやり取りする際にApiクラスを介するようにして、Traderクラスからはどちらとやり取りする場合も同じメソッド群でApiクラスとやり取りすることでその後ろにあるSimulatorクラスや取引所とやり取りできるようにするのです。
図にすると以下の通りです。

f:id:kurimayoshida:20200717214220p:plain
Apiクラス

SimulatorApiクラス、GmoCoinApiクラス、どちらのクラスもTraderクラスから取引をするためのインターフェースが同じになるよう、共通の基底クラスとなるApiクラスを作ります。
Apiクラスには以下のようなメソッドを用意しています。

  • 余力情報の取得
  • 最新レートの取得
  • 成行の買い注文の発注
  • 成行の売り注文の発注
  • 注文の状態の取得
  • 資産情報の取得
from datetime import datetime
from dataclasses import dataclass
from enum import Enum, auto
from typing import List
import json
import requests
import os
import hmac
import time
from datetime_util import DatetimeUtil
import hashlib
from abc import ABCMeta, abstractmethod

class OrderStatus(Enum):
    WAITING = auto()
    ORDERED = auto()
    MODIFYING = auto()
    CANCELLING = auto()
    CANCELED = auto()
    EXECUTED = auto()
    EXPIRED = auto()

class Symbol(Enum):
    JPY = auto()
    BTC = auto()
    ETH = auto()
    BCH = auto()
    LTC = auto()
    XRP = auto()

@dataclass
class Ticker:
    ask_price: int
    bid_price: int
    high_price: int
    last_price: int
    low_price: int
    symbol: str
    timestamp: datetime
    volume: float

@dataclass
class Asset:
    amount: float
    symbol: Symbol


class Api(metaclass=ABCMeta):
    @abstractmethod
    def get_available_amount(self) -> int:
        pass

    @abstractmethod
    def get_ticker(self, symbol: Symbol) -> Ticker:
        pass

    @abstractmethod
    def place_market_buying_order(self, symbol: Symbol, size: float) -> int:
        pass

    @abstractmethod
    def place_market_selling_order(self, symbol: Symbol, size: float) -> int:
        pass

    @abstractmethod
    def get_order_status(self, order_id: int) -> List[OrderStatus]:
        pass

    @abstractmethod
    def get_assets(self) -> List[Asset]:
        pass

Traderクラスの改修

Traderクラスを改修します。改修のポイントは以下です。

  • 直接SimulatorクラスとではなくApiクラスを介してSimulatorクラスとやりとりするため、コンストラクターでSimulatorクラスの代わりにApiクラスを渡し、それを使います。
  • 前回は資産の量(日本円やビットコインの量)をTraderクラスに持たせて、注文にあわせてその量をTraderクラス上で更新していましたが、実際の資産は取引所にあるということにあわせて、Simulatorクラスや取引所側に資産の量を持たせるようにします。そして、発注をAPIを介してSimulatorクラスや取引所に対して行うようにし、Simulatorクラスや取引所側で注文を処理して資産の量を更新します。
  • 前回は発注後即座に必ず約定する前提でした。実際の取引では必ずしもそうではなく、そして約定は取引所側で決まるので、発注をApiクラスを介してSimulatorクラスや取引所に対して行うことに加えて、Apiクラスを介して約定の確認を行うようにします。それに伴いトレードの状態に「買い注文約定待ち」「売り注文約定待ち」を追加します。

とりあえずTraderクラスで資産の量を持つ必要はなくなったためTraderクラスから資産の量を取り除いています。しかし例えば取引のアルゴリズムが今の資産の量を加味する場合は度々Apiクラスを通じて今の資産の量を取得するのも時間がかかるのでTraderクラスが資産の量をもつこともありえます。その場合であっても本当の資産の量はSimulatorクラスや取引所にあるものであり、Traderクラスが持っているものはその写しということになります。

from enum import Enum, auto
from api import Api, Symbol, OrderStatus
from price_data_manager import PriceDataManager
import datetime


class TradingStatus(Enum):
    # 買い発注待ち
    STANDING_BY_FOR_BUYING = auto()
    # 買い注文約定待ち
    ORDERED_BUYING_ORDER = auto()
    # 売り発注待ち
    STANDING_BY_FOR_SELLING = auto()
    # 売り注文約定待ち
    ORDERED_SELLING_ORDER = auto()


class SellingType(Enum):
    DEADLINE = auto()
    PROFIT_TAKING = auto()
    STOP_LOSS = auto()


class Trader:
    SHORT_MA_PERIOD_SEC = 1 * 60
    LONG_MA_PERIOD_SEC = 10 * 60

    MAX_HOLDING_SEC = 5 * 60 * 60
    SELLING_TH_PRICE_RATE = 1.01
    STOP_LOSS_SELLING_TH_PRICE_RATE = 0.995

    def __init__(self, api: Api) :
        self.state = TradingStatus.STANDING_BY_FOR_BUYING

        self.api = api
        self.order_id = None


        # 取引判断用データ ----

        self.price_data_manager = PriceDataManager()

        # 前回の移動平均
        self.last_long_ma_price = None
        self.last_short_ma_price = None

        self.bought_price = None
        self.bought_datetime = None


    def act(self):
        ticker = self.api.get_ticker(Symbol.BTC)
        self.price_data_manager.add_ticker(ticker)

        short_ma_price = self.price_data_manager.get_moving_average_ask_price(self.SHORT_MA_PERIOD_SEC)
        long_ma_price = self.price_data_manager.get_moving_average_ask_price(self.LONG_MA_PERIOD_SEC)

        if self.state == TradingStatus.STANDING_BY_FOR_BUYING:
            # ゴールデンクロスのチェック
            if (self.last_short_ma_price is not None and 
                self.last_short_ma_price < self.last_long_ma_price and 
                short_ma_price > long_ma_price):

                # 買い
                self.order_id = self.api.place_market_buying_order(Symbol.BTC, 1)
                self.state = TradingStatus.ORDERED_BUYING_ORDER
                self.bought_price = ticker.ask_price
                self.bought_datetime = ticker.timestamp
                print(f'{ticker.timestamp} 買い {ticker.ask_price}')

        elif self.state == TradingStatus.ORDERED_BUYING_ORDER:
            order_status_list = self.api.get_order_status(self.order_id)
            is_executed = False
            for status in order_status_list:
                if status == OrderStatus.EXECUTED:
                    is_executed = True
            if is_executed:
                self.state = TradingStatus.STANDING_BY_FOR_SELLING

        elif self.state == TradingStatus.STANDING_BY_FOR_SELLING:
            # 時間経過、指定の割合上昇、指定の割合下降で売る
            selling_type = None
            if ticker.timestamp >= self.bought_datetime + datetime.timedelta(seconds=self.MAX_HOLDING_SEC):
                selling_type = SellingType.DEADLINE
            if ticker.bid_price >= self.bought_price * self.SELLING_TH_PRICE_RATE:
                selling_type = SellingType.PROFIT_TAKING
            if ticker.bid_price <= self.bought_price * self.STOP_LOSS_SELLING_TH_PRICE_RATE:
                selling_type = SellingType.STOP_LOSS

            if selling_type:
                # 売り
                self.order_id = self.api.place_market_selling_order(Symbol.BTC, 1)
                self.state = TradingStatus.ORDERED_SELLING_ORDER

                if selling_type == SellingType.DEADLINE:
                    action = '売り(期限)'
                elif selling_type == SellingType.PROFIT_TAKING:
                    action = '売り(利確)'
                elif selling_type == SellingType.STOP_LOSS:
                    action = '売り(ロスカット)'
                print(f'{ticker.timestamp} {action} {ticker.bid_price}')

        elif self.state == TradingStatus.ORDERED_SELLING_ORDER:
            order_status_list = self.api.get_order_status(self.order_id)
            is_executed = False
            for status in order_status_list:
                if status == OrderStatus.EXECUTED:
                    is_executed = True
            if is_executed:
                self.state = TradingStatus.STANDING_BY_FOR_BUYING

        self.last_short_ma_price = short_ma_price
        self.last_long_ma_price = long_ma_price

Simulatorクラスの改修

Apiクラスの追加、Traderクラスの改修にあわせてSimulatorクラスも改修します。改修のポイントは以下です。

  • シミュレーター用のApiクラスであるSimulatorApiクラスを持つようにします。
  • 資産の量を持つようにします。
  • 注文を処理するようにします。

前回はTraderクラス上で資産の量を更新するだけで済ませていた注文の処理をSimulatorクラスが仮想の取引所としての役割を担って行うようにします。
place_market_buying_order()メソッドとplace_market_selling_order()メソッドで、それぞれ買い注文と売り注文を受けます。注文を受けたらそれらの注文は即座に約定したこととし、資産の量を更新し、order_status_list_dict として持つ各注文の状態も全量約定とします。
get_order_status()メソッドで注文の状態を問われたらorder_status_list_dictから返します。

from api import Ticker, OrderStatus
import pandas as pd
from typing import List


class Simulator:
    def __init__(self):
        self.cur_ticker = None

        self.yen_amount = 2000000
        self.btc_amount = 0

        self.next_order_id = 0
        self.order_status_list_dict = {}

        from simulator_api import SimulatorApi
        self.simulator_api = SimulatorApi(self)

    def simulate(self):
        from trader import Trader
        self.trader = Trader(self.simulator_api)

        print(f'開始 資産: {self.yen_amount}')

        # CSVファイルを読み込む
        df = pd.read_csv('ticker_20200514.csv')
        df['timestamp'] = pd.to_datetime(df['timestamp'])

        # 一行ずつ処理
        for _, row in df.iterrows():
            self.cur_ticker = Ticker(row['ask'], row['bid'], row['high'], row['last'],
                    row['low'], row['symbol'], row['timestamp'], row['volume'])
            self.trader.act()

        # 全行処理が終わったら結果を表示する
        print(f'終了 資産: {self.yen_amount + self.btc_amount * self.cur_ticker.bid_price}')

    def get_cur_ticker(self) -> Ticker:
        return self.cur_ticker

    def place_market_buying_order(self, symbol: str, size: float) -> int:
        # 簡易的に即約定とする
        order_id = self.next_order_id

        self.yen_amount -= self.cur_ticker.ask_price
        self.btc_amount += 1

        self.order_status_list_dict[order_id] = [OrderStatus.EXECUTED]
        self.next_order_id += 1
        return order_id

    def place_market_selling_order(self, symbol: str, size: float) -> int:
        # 簡易的に即約定とする
        order_id = self.next_order_id

        self.yen_amount += self.cur_ticker.bid_price
        self.btc_amount -= 1

        self.order_status_list_dict[order_id] = [OrderStatus.EXECUTED]
        self.next_order_id += 1
        return order_id

    def get_order_status(self, order_id: int) -> List[OrderStatus]:
        if order_id not in self.order_status_list_dict:
            return []
        return self.order_status_list_dict[order_id]

SimulatorApiクラスの作成

Apiクラスの各メソッドでSimulatorクラスに対して取引を行うSimulatorApiクラスを作成します。

from api import Api, Ticker, OrderStatus, Symbol, Asset
from simulator import Simulator
from typing import List


class SimulatorApi(Api):
    def __init__(self, simulator: Simulator) :
        self.simulator = simulator

    def get_available_amount(self) -> int:
        return self.simulator.yen_amount
    
    def get_ticker(self, symbol) -> Ticker:
        return self.simulator.get_cur_ticker()
    
    def place_market_buying_order(self, symbol: str, size: float) -> int:
        return self.simulator.place_market_buying_order(symbol, size)

    def place_market_selling_order(self, symbol: str, size: float) -> int:
        return self.simulator.place_market_selling_order(symbol, size)

    def get_order_status(self, order_id: int) -> List[OrderStatus]:
        return self.simulator.get_order_status(order_id)

    def get_assets(self) -> List[Asset]:
        assets = []
        assets.append(self.simulator.yen_amount, Symbol.JPY)
        assets.append(self.simulator.btc_amount, Symbol.BTC)
        return assets

シミュレーションの実行

シミュレーションの実行を行う main.py は変更なしです。シミュレーションを実行してみましょう。
Traderクラスがトレードの状態として「買い注文約定待ち」「売り注文約定待ち」という状態を持つようになり、3秒間約定の確認に取られるため若干結果に違いがでましたが特に大きな違いはありません。
相変わらずの大敗です…。

from simulator import Simulator


if __name__ == '__main__':

    simulator = Simulator()
    simulator.simulate()

次回はTraderクラスはそのままに、GMOコインのAPIを介して取引を行うGmoCoinApiクラスを作成して、いよいよ自動取引を実現させましょう。

Pythonでビットコインの自動取引をシミュレーションする(1)

前回までで、実際のビットコインの価格データを収集したり、その価格データをグラフ化したりしました。

今回はゴールデンクロスを使った簡単な取引のアルゴリズムをつくり取引をシミュレーションしてみます。

シミュレーションの概要

シミュレーションはシミュレーターという仮想の取引所の中である期間仮想のトレーダーに取引させて行うこととします。
そして開始時と終了時でトレーダーがもつ資産がどう変わったか等を見てアルゴリズムがうまく働いているかを確認します。

今回行うシミュレーションでは実装を簡単にするために実際の取引からいくらか単純化します。

シミュレーションの概要を以下に示します。

  • トレーダーはファイルに記録された価格データのタイムスタンプのタイミングで売買の判断と売買を行う。例えば以前作成した ticker_20200514.csv であれば 2020-05-06T10:15:39Z から 2020-05-13T06:00:00Z までの間で約3秒おきに行う。
  • シミュレーション開始時のトレーダーの資産は200万円。
  • シミュレーション終了時にトレーダーがビットコインを持っていたらbidの価格で円に換算して終了時の資産を計算する。
  • トレーダーのビットコインの売買単位は1単位のみ。
  • トレーダーが同時に持てるビットコインは1単位のみ。
  • 売買は即座にそのときの価格で約定することとする。

設計

以下のようなクラス構成を考えました。

Simulatorクラス

  • 仮想の取引所。この中でTraderが特定の時間間隔で行動する。
  • Traderに現在のビットコインの価格を提示する。
  • シミュレーションの結果を表示する。
  • トレーダーからの売買注文の受付はより詳細なシミュレーションには必要だが今回は省略。

Traderクラス

PriceDataManagerクラス

  • 価格データを蓄積する。
  • 蓄積した価格データのうち直近の指定秒間の価格データを使って移動平均値を計算する。


クラス間の所有関係を図にすると以下のようになります。

f:id:kurimayoshida:20200627080004p:plain
クラス間の関係

メインの処理の作成

Simulatorクラスのインスタンスのsimulate()を実行するだけです。
simulate()がシミュレーションを行って結果を表示してくれるはずです。

from simulator import Simulator


if __name__ == '__main__':

    simulator = Simulator()
    simulator.simulate()

Simulatorクラスの作成

Pythonでビットコインの価格を取得する(2) - てけとーぶろぐ。で作成した価格データのCSVファイルを読み込み、先頭のデータから1行よんではその価格をシミュレーション上の今の価格としてトレーダーを行動させます。
トレーダーは行動する中でSimulatorクラスのget_cur_ticker()メソッドを呼んで価格を取得するのでシミュレーション上の今の価格を取得することになります。

from api import Ticker
import pandas as pd


class Simulator:
    def __init__(self):
        self.cur_ticker = None

    def simulate(self):
        from trader import Trader
        self.trader = Trader(self)
        self.trader.yen_amount = 2000000
        self.trader.btc_amount = 0

        print(f'開始 資産: {self.trader.yen_amount}')

        # CSVファイルを読み込む
        df = pd.read_csv('ticker_20200514.csv')
        df['timestamp'] = pd.to_datetime(df['timestamp'])

        # 一行ずつ処理
        for _, row in df.iterrows():
            self.cur_ticker = Ticker(row['ask'], row['bid'], row['high'], row['last'],
                    row['low'], row['symbol'], row['timestamp'], row['volume'])
            self.trader.act()

        # 全行処理が終わったら結果を表示する
        print(f'終了 資産: {self.trader.yen_amount + self.trader.btc_amount * self.cur_ticker.bid_price}')

    def get_cur_ticker(self) -> Ticker:
        return self.cur_ticker

Traderクラスの作成

act()メソッドがシミュレーター内で一定仮想時間ごとに実行されます。
現在の状態(ビットコインを持っているか否か)に応じて売買の判断をします。
具体的には、ビットコインを持っていないときは1分の移動平均線と10分の移動平均線ゴールデンクロスで買いの判断をします。ビットコインを持っているときは買いから5時間経過するか、買いから1%価格が上がるか、0.5%価格が下がるかしたら売りの判断をします。

from enum import Enum, auto
from simulator import Simulator
from price_data_manager import PriceDataManager
import datetime


class TradingStatus(Enum):
    # 買い発注待ち
    STANDING_BY_FOR_BUYING = auto()
    # 売り発注待ち
    STANDING_BY_FOR_SELLING = auto()


class SellingType(Enum):
    DEADLINE = auto()
    PROFIT_TAKING = auto()
    STOP_LOSS = auto()


class Trader:
    SHORT_MA_PERIOD_SEC = 1 * 60
    LONG_MA_PERIOD_SEC = 10 * 60

    MAX_HOLDING_SEC = 5 * 60 * 60
    SELLING_TH_PRICE_RATE = 1.01
    STOP_LOSS_SELLING_TH_PRICE_RATE = 0.995

    def __init__(self, simulator: Simulator) :
        # TODO Simulator ではなくAPIを渡してAPIの通信先が
        # Simulator でも GMOコインのAPI でも同じように扱えるようにする

        self.state = TradingStatus.STANDING_BY_FOR_BUYING
        self.simulator = simulator

        # 資産 ----

        self.yen_amount = 0
        self.btc_amount = 0

        # 取引判断用データ ----

        self.price_data_manager = PriceDataManager()

        # 前回の移動平均
        self.last_long_ma_price = None
        self.last_short_ma_price = None

        self.bought_price = None
        self.bought_datetime = None


    def act(self):
        ticker = self.simulator.get_cur_ticker()
        self.price_data_manager.add_ticker(ticker)

        short_ma_price = self.price_data_manager.get_moving_average_ask_price(self.SHORT_MA_PERIOD_SEC)
        long_ma_price = self.price_data_manager.get_moving_average_ask_price(self.LONG_MA_PERIOD_SEC)

        if self.state == TradingStatus.STANDING_BY_FOR_BUYING:
            # ゴールデンクロスのチェック
            if (self.last_short_ma_price is not None and 
                self.last_short_ma_price < self.last_long_ma_price and 
                short_ma_price > long_ma_price):

                # 買い
                # 簡易的に今の価格で買えたこととする
                self.yen_amount -= ticker.ask_price
                self.btc_amount += 1
                self.state = TradingStatus.STANDING_BY_FOR_SELLING
                self.bought_price = ticker.ask_price
                self.bought_datetime = ticker.timestamp
                print(f'{ticker.timestamp} 買い {ticker.ask_price}')

        elif self.state == TradingStatus.STANDING_BY_FOR_SELLING:
            # 時間経過、指定の割合上昇、指定の割合下降で売る
            selling_type = None
            if ticker.timestamp >= self.bought_datetime + datetime.timedelta(seconds=self.MAX_HOLDING_SEC):
                selling_type = SellingType.DEADLINE
            if ticker.bid_price >= self.bought_price * self.SELLING_TH_PRICE_RATE:
                selling_type = SellingType.PROFIT_TAKING
            if ticker.bid_price <= self.bought_price * self.STOP_LOSS_SELLING_TH_PRICE_RATE:
                selling_type = SellingType.STOP_LOSS

            if selling_type:
                # 売り
                # 簡易的に今の価格で売れたこととする
                self.yen_amount += ticker.bid_price
                self.btc_amount -= 1
                self.state = TradingStatus.STANDING_BY_FOR_BUYING

                if selling_type == SellingType.DEADLINE:
                    action = '売り(期限)'
                elif selling_type == SellingType.PROFIT_TAKING:
                    action = '売り(利確)'
                elif selling_type == SellingType.STOP_LOSS:
                    action = '売り(ロスカット)'
                print(f'{ticker.timestamp} {action} {ticker.bid_price}')

        self.last_short_ma_price = short_ma_price
        self.last_long_ma_price = long_ma_price

PriceDataManagerクラスの作成

add_ticker()メソッドで価格データを蓄積しget_moving_average_ask_price()メソッドやget_moving_average_bid_price()メソッドで蓄積した価格データを使ってaskやbidの価格の移動平均値を求めます。

from api import Ticker
from collections import deque
import datetime

class PriceDataManager:
    MAX_MOVING_AVERAGE_PERIOD_SEC = 10 * 60

    _PRICE_TYPE_ASK = 'ask'
    _PRICE_TYPE_BID = 'bid'

    def __init__(self) :
        self.ticker_queue = deque()

    def add_ticker(self, ticker: Ticker) :
        self.ticker_queue.append(ticker)
        # Tickerが増えすぎないように指定期間より過去のTickerは削除する
        th_datetime = ticker.timestamp - datetime.timedelta(seconds=self.MAX_MOVING_AVERAGE_PERIOD_SEC)
        while self.ticker_queue[0].timestamp < th_datetime:
            self.ticker_queue.popleft()

    def get_moving_average_ask_price(self, period_sec) -> float:
        return self._get_moving_average_price(period_sec, self._PRICE_TYPE_ASK)

    def get_moving_average_bid_price(self, period_sec) -> float:
        return self._get_moving_average_price(period_sec, self._PRICE_TYPE_BID)

    def _get_moving_average_price(self, period_sec, price_type) -> float:
        th_datetime = self.ticker_queue[-1].timestamp - datetime.timedelta(seconds=period_sec)
        num = 0
        sum = 0.0
        for ticker in reversed(self.ticker_queue):
            if th_datetime < ticker.timestamp:
                num += 1
                if price_type == self._PRICE_TYPE_ASK:
                    sum += ticker.ask_price
                elif price_type == self._PRICE_TYPE_BID:
                    sum += ticker.bid_price
        return sum / num

プログラムの実行

プログラムを実行してみます。

> python .\main.py
開始 資産: 2000000
2020-05-06 10:44:33+00:00 買い 981000
2020-05-06 10:49:15+00:00 売り(ロスカット) 978350
2020-05-06 10:53:48+00:00 買い 979990
2020-05-06 13:23:30+00:00 売り(ロスカット) 975000
2020-05-06 13:31:15+00:00 買い 977130
2020-05-06 18:31:15+00:00 売り(期限) 978670
2020-05-06 18:31:33+00:00 買い 980520
2020-05-06 18:41:56+00:00 売り(ロスカット) 978020
2020-05-06 18:48:15+00:00 買い 979990
2020-05-06 23:48:18+00:00 売り(期限) 985000
2020-05-07 00:07:57+00:00 買い 971000
2020-05-07 02:39:57+00:00 売り(利確) 990969
…
2020-05-13 05:44:57+00:00 買い 955187
2020-05-13 05:53:02+00:00 売り(ロスカット) 950501
終了 資産: 1616646

買いと売りを繰り返して…
買いのタイミングは前回のチャートと見比べるとゴールデンクロスのタイミングになっていそうです。
うーん、なんかロスカットがおおいかな…。
結果…200万円が161万円に!

…改良が必要ですね。シミュレーションしてよかった!と前向きにとらえましょう…。

Pythonでビットコインの価格をグラフ化する

前回収集した価格データをグラフ化して、自動売買のアルゴリズムを考えてみましょう。

価格データのグラフ化

前回収集した価格データのCSVファイルが以下です。

ticker_20200514.zip

こちらをPythonプログラムでグラフ化してみます。
グラフの描画にはnumpyパッケージ, matplotlibパッケージを使います。
またCSVファイルの読み込み等にpandasパッケージを使います。

それぞれをインストールします。

> pip install matplotlib
> pip install pandas

matplotlibを使ったグラフの描画については以下のサイトを参考にしました。

note-tech.com
note-tech.com

全体のコードは以下になります。

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt


if __name__ == '__main__':

    df = pd.read_csv('ticker_20200514.csv')

    df['timestamp'] = pd.to_datetime(df['timestamp'])

    START_TIMESTAMP = '2020-05-06 10:00'
    END_TIMESTAMP = '2020-05-06 18:00'
    SAMPLING_PERIOD_SEC = 3
    SHORT_MA_PERIOD_SEC = 1 * 60
    LONG_MA_PERIOD_SEC = 10 * 60

    # データの期間を絞る
    df = df[(df['timestamp'] >= START_TIMESTAMP) & (df['timestamp'] < END_TIMESTAMP)]

    # 移動平均の列を作成
    df['askShortMa'] = df['ask'].rolling(SHORT_MA_PERIOD_SEC // SAMPLING_PERIOD_SEC, min_periods=1).mean()
    df['askLongMa'] = df['ask'].rolling(LONG_MA_PERIOD_SEC // SAMPLING_PERIOD_SEC, min_periods=1).mean()

    # グラフ描画
    fig, ax = plt.subplots()
    ax.plot(df['timestamp'], df['ask'], label='ask')
    ax.plot(df['timestamp'], df['askShortMa'], label='ask ma 1 min')
    ax.plot(df['timestamp'], df['askLongMa'], label='ask ma 10 min')
    ax.legend(fontsize=16)
    ax.set_xlabel('timestamp', fontsize=16)
    ax.set_ylabel('¥', fontsize=16)
    ax.set_title('Ticker', fontsize=16)
    plt.show()

上記サイトから幾つか変更点があります。

まず全データを描画すると時間がかかるためデータを絞りました。
以下のコードで指定の期間分に絞ったDataFrameを作成しています。

    # データの期間を絞る
    df = df[(df['timestamp'] >= START_TIMESTAMP) & (df['timestamp'] < END_TIMESTAMP)]

また移動平均の求め方について
上記サイトではnumpyのconvolve()を使っていますが
pandasでできる範囲はpandasでということで以下のサイトを参考にpandasのrolling()で移動平均を求めています。

note.nkmk.me

以下のコードになります。
「min_periods=1」として先頭の方でデータが足りなくても1つ以上データがあればある分だけで平均値を求めるようにしています。

    # 移動平均の列を作成
    df['askShortMa'] = df['ask'].rolling(SHORT_MA_PERIOD_SEC // SAMPLING_PERIOD_SEC, min_periods=1).mean()
    df['askLongMa'] = df['ask'].rolling(LONG_MA_PERIOD_SEC // SAMPLING_PERIOD_SEC, min_periods=1).mean()

実行すると以下のグラフが描画されます。

f:id:kurimayoshida:20200607160330p:plain

自動取引のアルゴリズムを考える

グラフを眺めてみるとどうでしょうか?
移動平均の期間を変えたり、表示する期間を変えたりしてもいいかと思います。
1分の移動平均線が10分の移動平均線を下から突き抜けるとある程度価格が上昇しているように見えます。
いわゆるゴールデンクロスですね。

参考: 「ゴールデンクロス」「デッドクロス」とは? | お金のキャンパス

このゴールデンクロスを買いシグナルとする自動取引のアルゴリズムでいくことにしましょう。
分析と言うには程遠い感じですしゆくゆくは機械学習の学習結果を使った自動取引などにも挑戦したいとは思うのですが
まずは実現しやすいものでどの程度の成績がでるか試してみます。


次回はこのゴールデンクロスを使った売買のアルゴリズムをつくり
売買をシミュレーションしてみます。

Pythonでビットコインの価格を取得する(2)

前回でビットコインの現在の価格を取得してコンソールに出力することができました。

自動取引の判断材料に使ったり、自動取引のシミュレーションをしたりするためには
現在の価格だけでなく、ある程度の期間の価格が必要になります。
これは現在の価格を取得してそれを記録することを一定期間行えばいいわけです。

記録の方法としてはファイルに書き込んでファイルに記録するのが簡単かと思います。
ですがここでは記録した価格データへのアクセスのしやすさを考えてデータベースに記録してみましょう。

データベースにも色々種類がありますが扱いが簡単で記録するデータのフォーマットも柔軟に変えられるMongoDBを使うことにします。

MongoDBのインストール(Windows PC)

Download Center: Community Server | MongoDB
から

  • Version: 4.2.7 (current release)
  • OS: Windows x64
  • Package: ZIP

を選択してDownloadをクリック。
ダウンロードした「mongodb-win32-x86_64-2012plus-4.2.7.zip」を
適当なディレクトリーに展開します。
(例: C:\MongoDB\mongodb-win32-x86_64-2012plus-4.2.6)

mongodの起動と終了

MongoDBデーモンであるmongodを起動してみます。
DBへデータを読み書きする場合は
あらかじめこのmongodを起動しておく必要があります。

コマンドプロンプトを起動して
カレントディレクトリーをMongoDBインストール先のbinディレクトリーにします。

> cd C:\MongoDB\mongodb-win32-x86_64-2012plus-4.2.7\bin

binディレクトリー内のmongod.exeを実行します。
実行の際にはデータディレクトリーの指定ができます。
ここでは「mongodb-win32-x86_64-2012plus-4.2.7」内の「data」ディレクトリーをデータディレクトリーとします。
データディレクトリーとして指定するディレクトリーはあらかじめ作成しておく必要があるので作成してから実行します。

> mongod --dbpath ../data

これでmongodが起動しました。

終了する場合は起動したコマンドプロンプト上でCtrl+Cキーを押します。

pymongoのインストール

PythonからMongoDBにアクセスするために
pymongoパッケージをインストールします。

Raspbianで簡単にインストールできるMongoDBのバージョンが2.4で
MongoDB 2.4 に対応した一番新しいpymongoのバージョンは3.5.1のようです。

PCで動かしていたプログラムがそのままRaspberry Pi(Raspbian OS)でも使えるよう
pymongoのバージョンを合わせるようにします。
つまりpymongo 3.5.1をバージョン指定してインストールします。

> pip install pymongo==3.5.1
MongoDBへの記録

ビットコインの現在の価格を取得してMongoDBに書き込むプログラムを作成します。
前回作成したビットコインの価格をコンソールに出力するプログラムをベースに
コンソールに出力する部分をMongoDBへの書き込みに書き換えて作成します。

前回作成したプログラム「main_print.py」を「main_mongo.py」という名前でコピーします。
主にこれのupdate()関数を書き換えていきます。
「main_mongo.py」の全体は以下のコードになります。

from api import GmoCoinApi
from apscheduler.schedulers.background import BackgroundScheduler
from time import sleep
import json
from pymongo import MongoClient


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

client = MongoClient('mongodb://localhost:27017/')
db = client['coin']


def update():
    global is_exception_raised
    try:
        ticker = gmo_coin_api.get_ticker('BTC')
        print(ticker)
        post = {
            'ask': ticker.ask_price,
            'bid': ticker.bid_price,
            'high': ticker.high_price,
            'last': ticker.last_price,
            'low': ticker.low_price,
            'symbol': ticker.symbol,
            'timestamp': ticker.timestamp,
            'volume': ticker.volume,
        }
        db['ticker'].insert_one(post)
    except Exception as exception:
        print(exception)
        is_exception_raised = True


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

    is_exception_raised = False

    try:
        while True:          
            if is_exception_raised:
                break

            sleep(0.001)

    except KeyboardInterrupt:
        pass

    scheduler.shutdown()

以下のコードでMongoDBにアクセスするためのMongoClientクラスのインスタンスを作成しています。

client = MongoClient('mongodb://localhost:27017/')

引数は接続先のホスト名です。
このプログラムを実行しているのと同じマシン上で
mongodを起動するので「localhost」になります。

続く以下のコードで指定の名前のデータベースを表す
Databaseクラスのインスタンスを取得します。

db = client['coin']

「coin」という名前のデータベースを使うことにします。
データベースはあらかじめ作成していなくても
存在しない場合は作成されるので問題ありません。

データベースへのデータの書き込みはDatabaseクラスのインスタンスを使って行います。
以下のコードで「post」を「ticker」というコレクションに書き込んでいます。

        db['ticker'].insert_one(post)

DB内のデータの構成を簡単に説明すると
coinデータベース内に、tickerコレクションがあり、tickerコレクション内に
「_id」「ask」「bid」「high」「last」「low」「symbol」「timestamp」「volume」
といったフィールドを持つ価格データのドキュメント(オブジェクト)が複数格納されているという構成になります。
「_id」フィールドは自動的に作られる、オブジェクトのIDを持つフィールドです。

実行してみましょう。
実行にあたってはmongodを起動しておくことを忘れないでください。

mongo shellでデータの確認

プログラムの実行の結果DBにデータが書き込まれたか確認してみましょう。

Pythonプログラムからpymongoを使ってDBのデータを取得して確認してもいいのですが
簡単な確認であればmongo shellを使うのが楽です。

まずmongodを起動しておきます。

新しいコマンドプロンプトを起動しmongo shellを起動します。

> mongo

起動するとMongoDBに対するコマンドを受け付ける状態となります。
ここにデータ取得のコマンド等を入力してデータを確認したりします。

mongo shellを終了する場合はCtrl+Cキーを押します。

データの確認のために必要なコマンドを紹介します。

データベース一覧の表示

> show dbs

使用するデータベースの指定
データベースの削除や、データベース内のコレクションからドキュメントを検索したりするためにはあらかじめこのコマンドで使用するデータベースを指定しておく必要があります。

> use coin

※「coin」データベースの場合

データベースの削除

> use coin
> db.dropDatabase() 

※「coin」データベースの場合

データベース内のコレクション一覧の表示

> use coin
> show collections

※「coin」データベースの場合

データベース内のコレクションの削除

> use coin
> db.ticker.drop()

※「coin」データベース、「ticker」コレクションの場合

コレクション内のドキュメント一覧の表示

> use coin
> db.ticker.find()

※「coin」データベース、「ticker」コレクションの場合

コレクション内の最新X件のドキュメントの表示

> use coin
> db.ticker.find().sort({_id: -1}).limit(10);

※「coin」データベース、「ticker」コレクション、最新10件の場合

参照: https://docs.mongodb.com/manual/mongo/

さて、DBにデータが書き込まれたか確認できたでしょうか?

MongoDBのインストール(Raspberry Pi)

作成したプログラムをRaspberry Piでも動かしてみましょう。
そのためにはRaspberry PiにもMongoDBをインストールする必要があります。

Raspberry Piの場合はコマンド1つでMongoDBをインストールできます。

$ sudo apt-get install mongodb-server

またRaspberry Piの場合はデーモンとしてインストールされるので
インストールしたら以降Raspberry Piを起動すると自動的にWindowsでいうところの「mongod」が起動した状態となり「mongod」起動の操作は不要です。
その他はPCと特に変わりません。

無事Raspberry Piで実行できたらしばらく起動しっぱなしにして
価格データを収集してみましょう。

起動しっぱなしにする際は
Pythonで株価アラートをつくる(4) - てけとーぶろぐ。
で紹介したバックグラウンドでの実行を使ってください。

また長時間起動しっぱなしにしてみると
例えば取引所のシステムメンテナンスが入って
エラーが起きて停止してしまう
といったことも起こることが分かるかと思います。

エラーが起きて停止したことを知るために
Pythonで株価アラートをつくる(1) - てけとーぶろぐ。
でつくったメール通知も組み込むと簡単に停止を知ることができます。

CSVファイルへのエクスポート

価格データの書き込みや検索を行わず
単にある期間の価格データを読み込みたいだけといった用途には
価格データはDBに入っているよりもCSVファイルなどになっていたほうが扱いやすいということもありますね。
例えば価格の推移のグラフを描画するプログラムで価格データを使う場合などです。

MongoDBにはDB内のデータをCSVファイル等にエクスポートするツールが付属しています。

例としてcoinデータベースのtickerコレクションの各フィールドをticker.csvとしてMongoDBのインストール先のディレクトリー(binディレクトリーの親ディレクトリー)に出力してみましょう。

> cd C:\MongoDB\mongodb-win32-x86_64-2012plus-4.2.7\bin
> mongoexport --db coin --collection ticker --type=csv --fields _id,ask,bid,high,last,low,symbol,timestamp,volume --out ../ticker.csv

Raspberry Piの場合はMongoDB 2.4でオプションが少し異なり「--type=csv」ではなく「--csv」になります。

$ mongoexport --db coin --collection ticker --csv --fields _id,ask,bid,high,last,low,symbol,timestamp,volume --out ticker.csv

サンプルとして何日間か分のビットコインの価格をエクスポートしたCSVを以下にアップロードしました。
ticker_20200514.zip

次回は収集した価格データをグラフ化して分析し、自動売買のアルゴリズムを考えてみたいと思います。

Pythonでビットコインの価格を取得する(1)

Raspberry Pi + Pythonビットコイン自動取引」と銘打って書きはじめ
要素技術を身につけるための株価アラートの作成を終え
いよいよビットコインを扱うのですがそれにあたりまえがきを少し。

やりたいことは自動売買

Raspberry PiPythonビットコインも実は結果的にそれらを選ぶことになったもので
やりたいことは自作プログラムでの自動売買なんです。

プログラミングをしたことがあって、何らか投資をしたことがあるという人だったら
誰もが一度は夢見ることではないでしょうか?

自分は10年以上前からできたらいいなと思っていました。
それが最近になってようやく敷居をさげてくれるものが出てきていよいよ実現できるのではないかと思っています。

あたらめてコンセプト

Raspberry Pi + Pythonビットコイン自動取引」のコンセプトをあらためて載せておきます。

  • 高実現性

実現しなくては意味がありません。なんとかして実現します。実現させるために方法を選びます。

  • 低コスト

趣味で、あわよくば利益をというものなのに、コストがかかりすぎても仕方ありませんので…。

この2つをコンセプトとします。

なぜRaspberry Pi?なぜPython? なぜビットコイン

なぜRaspberry Piなのかについては前回まででお分かりいただけたかと思います。
なぜPythonなのかは実装のしやすさからです。
そしてなぜビットコインなのか。
ビットコインはその生まれからか、新興の投資対象であるからか、国内の取引所に自作プログラムから実行可能なAPIを用意しているところがいくつかあります。そこが理由です。
例えば国内株式についてはそのような仕組みを用意している証券会社は見当たりませんでした。
これがないと価格取得や発注のプログラムから用意しないといけないですし、その運用・保守もしなければいけなくなります。
そしてそれが今まで自分が自動取引を断念してきた主な理由でした。

要は自作プログラムでの自動売買を低コストで実現するためのRaspberry PiでありPythonでありビットコインです。

取引所を選ぶ

早速取引所を選びます。

  • 自作プログラムから実行可能なAPI
  • 国内の取引所
  • 安定性が期待できる有名所
  • 安い手数料

あたりからGMOコインを使うことにします。

coin.z.com

そうときまったら口座開設しましょう。

GMOコインのAPIを使う

早速GMOコインのAPIを使ってみましょう。
GMOコインには親切なAPIドキュメントがあり
Pythonでのサンプルコードも用意されています。

APIドキュメント| GMOコイン


GMOコインのAPIにはPublic APIとPrivate APIがあり
Public APIであれば口座開設せずとも実行することができます。
逆にPrivate APIは口座開設を行ってから得られるAPIキーとAPIシークレットがなければ実行することはできません。

幸いビットコインの価格を得るAPI「GET /public/v1/orderbooks」はPublic APIなので口座開設せずとも実行することができます。

API「GET /public/v1/orderbooks」のドキュメントは以下です。
APIドキュメント| GMOコイン

ドキュメントにあるサンプルの通りで十分ではありますが
今後を考えてAPIを実行するクラスを作っておきます。
結果もdataclassで返すようにします。
APIを実行し、APIが返してきたJSONに含まれる値をTickerクラスに詰め込んで返しているだけです。

from dataclasses import dataclass
from datetime import datetime
import requests
from datetime_util import DatetimeUtil

@dataclass
class Ticker:
    ask_price: int
    bid_price: int
    high_price: int
    last_price: int
    low_price: int
    symbol: str
    timestamp: datetime
    volume: float

class GmoCoinApi:
    def __init__(self, end_point) :
        self.end_point = end_point
    
    def get_ticker(self, symbol) -> Ticker:
        path = '/public/v1/ticker?symbol=' + symbol
        response = self._send_get_request(self.end_point + path)
        response_json = response.json()

        if response_json['status'] != 0:
            raise RuntimeError('API error.')

        ask_price = int(response_json['data'][0]['ask'])
        bid_price = int(response_json['data'][0]['bid'])

        # highにnullが入ってくることがあったのでその場合はNoneにする        
        response_json_high = response_json['data'][0]['high']
        if response_json_high is None:
            high_price = None
        else:
            high_price = int(response_json_high)

        last_price = int(response_json['data'][0]['last'])

        # lowにnullが入ってくることがあったのでその場合はNoneにする        
        response_json_low = response_json['data'][0]['low']
        if response_json_low is None:
            low_price = None
        else:
            low_price = int(response_json_low)

        symbol = response_json['data'][0]['symbol']
        timestamp = DatetimeUtil.iso_to_datetime(response_json['data'][0]['timestamp'])
        volume = float(response_json['data'][0]['volume'])

        ticker = Ticker(ask_price, bid_price, high_price, last_price, low_price, symbol, timestamp, volume)
        return ticker


    def _send_get_request(self, url, params=None, **kwargs):
        response = requests.get(url, params=params, **kwargs)
        return response


if __name__ == '__main__':

    gmo_coin_api = GmoCoinApi('https://api.coin.z.com')
    ticker = gmo_coin_api.get_ticker('BTC')
    print(ticker)


この中で使っている DatetimeUtil.iso_to_datetime() は
APIが返してくるISO 8601形式の時刻をdatetimeに変換するユーティリティメソッドで以下がソースコードです。

import datetime

class DatetimeUtil(object):

    @staticmethod
    def iso_to_datetime(iso: str) -> datetime:
        if iso is None:
            return None
            
        dt = datetime.datetime.fromisoformat(iso.replace('Z', '+00:00'))
        if (dt.tzinfo is not None) and (dt.tzinfo.utcoffset(dt) is not None):
            return dt
        else:
            return (dt.astimezone())

    @staticmethod
    def datetime_to_iso(dt: datetime) -> str:
        if dt is None:
            return None

        if (dt.tzinfo is not None) and (dt.tzinfo.utcoffset(dt) is not None):
            if dt.tzinfo == datetime.timezone.utc:
                return dt.isoformat().replace('+00:00', 'Z')
            else:
                return (dt.astimezone(datetime.timezone.utc)).isoformat().replace('+00:00', 'Z')
        else:
            return (dt.astimezone(datetime.timezone.utc)).isoformat().replace('+00:00', 'Z')

実行してみましょう。
実行にあたっては「requests」のパッケージが必要なので
未インストールの場合はインストールします。

> pip install requests

価格が取得できましたね。

時々刻々と変化する価格を取得するためには定期実行することになります。
株価アラートと同様にapschedulerを使って3秒おきに取得してみます。

from api import GmoCoinApi
from apscheduler.schedulers.background import BackgroundScheduler
from time import sleep
import json


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


def update():
    global is_exception_raised
    try:
        ticker = gmo_coin_api.get_ticker('BTC')
        print(ticker)
    except Exception as exception:
        print(exception)
        is_exception_raised = True


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

    is_exception_raised = False

    try:
        while True:          
            if is_exception_raised:
                break

            sleep(0.001)

    except KeyboardInterrupt:
        pass

    scheduler.shutdown()

実行してみます。
実行にあたっては「apscheduler」のパッケージが必要なので
未インストールの場合はインストールします。

> pip install apscheduler

こんなふうにしてビットコインの価格を取得して、あとは、蓄積して、分析して、売買アルゴリズムを作っていこうと思います。

次回は取得した価格データを分析のためにデータベースに保存してみましょう。