【Python】Coincheck APIからビットコイン1分ローソク足データを自動取得

Coincheckは１時間のローソク足データを無償で取得できますが、仮想通貨のボラティリティを考えると１分足の方が何かと便利です。しかし１分足は公式に提供されていません。

それなら自分で１分ローソク足チャートを作成に必要なデータを自動取得するプログラムをAWS上で運用しよう！というのが今回の記事になります。

ちなみにbitFlyerなどの一部の取引所はCryptowatcherから取得できますが、少額自動売買ではCoincheckの取引手数料無料が魅力的なため、『Coincheckの１分ローソク足を取得する』という点に拘っていきます。

ローソク足チャートに必要なデータ
Coincheck APIからpythonを使ってOHLCVを取得
最後にAWS上で1分毎にプログラムを実行する

ローソク足チャートに必要なデータ

ローソク足は時間単位における始値、高値、安値、終値を箱ひげ図で描画したグラフです。時間単位で価格が上昇した場合は青or緑、下落した場合は赤色の箱ひげ図で表されます。

描画に必要な値の４つの値の意味は以下の通り。

始値（open）：時間単位開始時の価格
高値（high）：時間単位内の成立した取引の最高値
安値（low）：時間単位内に成立した取引の最安値
終値（close）：時間単位に最後に成立した取引価格

以上を略してOHLCとか、出来高（volume）も含めてOHLCVとも呼ばれます。ということで1分間隔のOHLCVデータを取得できれば１分ローソク足が描画できます。

Coincheck APIからpythonを使ってOHLCVを取得

残念ながらCoincheck APIには1分足のOHLCVを取得する機能がありません。そこで直近の全取引データを取得して、その取引内容をOHLCVを自作する必要があります。

Coincheck APIから全取引データを取得

全取引データの取得APIは公式に記載されているURL（https://coincheck.com/api/trades）にrequestsのgetメソッドでアクセスします。

import requests

url = 'https://coincheck.com/api/trades'
params = {
    'pair': 'btc_jpy', # 通貨ペアを指定
    'limit': 100       # デフォルトでは10件しか取得しないので、最大の100を指定
         }
all_trades = requests.get(url, params=params).json()

取得した全取引データ（辞書型）の”data” keyにリスト形式で取引データが格納されており、さらにリスト中に辞書型で各取引のデータが以下の形式で格納されています。

all_trades['data'] = [
                      {
                        'id': '取引ID', 
                        'amount': '0.15',
                        'rate': '4000000',
                        'pair': 'btc_jpy',
                        'order_type': 'sell or buy',
                        'created_at': '2023-01-10T05:55:38.000Z' 
                      },...
                     ]

全取引データを直近１分間のOHLCVに変換

取得した取引データのうち、OHLCVに必要になるのは以下３つ。

”amount”: 成立した取引の出来高。volumeの算出に必要。
”rate”: 成立した取引価格。OHLCの算出に必要。
“created_at”：成立した時刻。1分ごとのデータに集約するために必要。ただし米国太平洋時間なので注意。

ここで注意すべきは時間データが米国太平洋時間であるという点です。こちらを日本時間に修正しつつ、取得した現在時刻の１分前のデータだけを集約します。表形式にしてしまえば時刻データはpandasで簡単に扱えます！

import pandas as pd

df = pd.DataFrame(all_trades['data'])                  # 取得した取引データをデータフレームに変換
df['created_at'] = pd.to_datetime(df['created_at'])    # 時刻データをdatetime形式に変換
df = df.set_index('created_at')                        # 時刻をindexに設定
df.index = df.index.tz_convert('Asia/Tokyo')           # 時刻データを日本時間に変換

now = pd.Timestamp.now(tz='Asia/Tokyo')                # 現在時刻を取得
now = now.replace(second=0)                            # 現在時刻の秒を切り捨てる
one_minute_ago = now - pd.Timedelta(minutes=1)         # 1分前の時刻を取得

last_1min = df.between_time(one_minute_ago.time(), now.time()) # 直前１分間のデータにスライス
last_1min = last_1min[['amount', 'rate']].astype(float)

last_1minのデータフレームからOHLCVデータを抽出してCSVファイルに格納します。

import csv

ohlcv = [
    str(one_minute_ago),
    last_1min['rate'][-1],     # open
    last_1min['rate'].max(),   # high
    last_1min['rate'].min(),   # low
    last_1min['rate'][0],      # close
    last_1min['amount'].sum()  # volume
]

with open('./ohlcv.csv', 'a') as f:
    csv.writer(f).writerow(ohlcv)

出来高が多い時間帯の対策

ここまでのコードを纏めてcronを使って１分毎に実行すればよいのですが、ビットコイン価格が急騰/急落したタイミングなど取引量が多い場合は取引件数が100では足りない可能性があります。

その対策として、プログラムを実行して20秒後、40秒後に当該分のデータ、60秒後に直前１分間のデータをそれぞれ取得することで最大300件データを取得するようにします。まとめたコードは以下になります。

import csv
from time import sleep

import pandas as pd
import requests

url = 'https://coincheck.com/api/trades'
params = {
    'pair': 'btc_jpy',
    'limit': 100
    }

missing_data = False
dfs = []

# 時刻取得
now = pd.Timestamp.now(tz='Asia/Tokyo').replace(second=0) # 現在
in_one_minute = now + pd.Timedelta(minutes=1)             # １分後

for i in range(3): # 20, 40, 60秒後にそれぞれデータを取得
    sleep(20)

    all_trades = requests.get(url, params=params).json()
    df = pd.DataFrame(all_trades['data'])
    df['created_at'] = pd.to_datetime(df['created_at'])
    df = df.set_index('created_at')
    df.index = df.index.tz_convert('Asia/Tokyo')
    df = df.between_time(now.time(), in_one_minute.time())
    dfs.append(df)

    if len(df) == 100:                      # データ数が100のときはフラグを立てる
        missing_data = True

df = pd.concat(dfs, axis=0)                 # データフレーム集約
df = df.drop_duplicates()                   # 重複削除
df = df[['amount', 'rate']].astype(float)

# get ohlcv
ohlcv = [
    str(now),
    df['rate'][-1],
    df['rate'].max(),
    df['rate'].min(),
    df['rate'][0],
    df['amount'].sum()
]

ohlcv.append('data missing') if missing_data else ohlcv.append('-')

with open('./ohlcv.csv', 'a') as f:
    csv.writer(f).writerow(ohlcv)

1分間に300件以上の取引がある場合は一応アラートを残しつつ諦めます笑