【Python】GMOコインでビットコイン等の暗号資産の自動売買を始める方法（実践編）

記事内に広告を含みます

この記事には広告を含む場合があります。

記事内で紹介する商品を購入することで、当サイトに売り上げの一部が還元されることがあります。

GMOコインでビットコイン等の暗号資産の自動売買を始める方法の準備編では、GMOコインの口座開設方法とAPIの使用方法について解説しました。準備がまだ整っていない方は、下記のリンクから口座の開設から進めてください。

プログラムの流れ

今回作成するプログラムは、下記のような流れになっています。

価格情報を取得
ゴールデンクロス・デッドクロスの検出
売買シグナルに基づいて注文

やっていることは非常にシンプルで、状態を監視して、シグナルが出たら注文を出します。これを繰り返します。

サンプルコード

移動平均線のゴールデンクロスとデッドクロスで自動売買するプログラムは以下のようになります。

Python

import time
from datetime import datetime
from datetime import timedelta
import requests
import hmac
import hashlib
import json
import pandas as pd


apiKey = 'YOUR_API_KEY'
secretKey = 'YOUR_SECRET'

# 設定項目
symbol = 'BTC_JPY'  # 使用する通貨ペア
interval = '1hour'  # 使用する時間軸
order_size = '0.01'  # 注文数量
MA_short_period = 5  # 移動平均線の短期の日数
MA_long_period = 25  # 移動平均線の長期の日数
get_price_period = 10  # 四本値を取得する日数

flag = {
    'order': {
        'side': '',
    },
    'position': {
        'exist': False,
        'side': None,
        'lot': 0.0,
    },
}


# 成行注文を出す関数
def market_order():
    global flag

    while True:
        try:
            timestamp = '{0}000'.format(int(time.mktime(datetime.now().timetuple())))
            method = 'POST'
            endPoint = 'https://api.coin.z.com/private'
            path = '/v1/order'
            reqBody = {
                'symbol': symbol,
                'side': flag['order']['side'],
                'size': order_size,
                'executionType': 'MARKET'
            }

            text = timestamp + method + path + json.dumps(reqBody)
            sign = hmac.new(bytes(secretKey.encode('ascii')), bytes(text.encode('ascii')), hashlib.sha256).hexdigest()

            headers = {
                'API-KEY': apiKey,
                'API-TIMESTAMP': timestamp,
                'API-SIGN': sign
            }

            res = requests.post(endPoint + path, headers=headers, data=json.dumps(reqBody))
            print(json.dumps(res.json(), indent=2))

            flag['order']['side'] = ''

            return

        except Exception as e:
            print(f'action=market_order error={e}')
            print('10秒後にリトライします')
            time.sleep(10)


# 決済注文を出す関数
def close_order():
    global flag

    while True:
        try:
            timestamp = '{0}000'.format(int(time.mktime(datetime.now().timetuple())))
            method = 'POST'
            endPoint = 'https://api.coin.z.com/private'
            path = '/v1/closeBulkOrder'
            reqBody = {
                'symbol': symbol,
                'side': flag['order']['side'],
                'size': order_size,
                'executionType': 'MARKET'
            }

            text = timestamp + method + path + json.dumps(reqBody)
            sign = hmac.new(bytes(secretKey.encode('ascii')), bytes(text.encode('ascii')), hashlib.sha256).hexdigest()

            headers = {
                'API-KEY': apiKey,
                'API-TIMESTAMP': timestamp,
                'API-SIGN': sign
            }

            res = requests.post(endPoint + path, headers=headers, data=json.dumps(reqBody))
            print(json.dumps(res.json(), indent=2))

            return

        except Exception as e:
            print(f'action=market_order error={e}')
            print('10秒後にリトライします')
            time.sleep(10)


# ポジションを確認する関数
def check_positions():
    global flag

    timestamp = '{0}000'.format(int(time.mktime(datetime.now().timetuple())))
    method = 'GET'
    endPoint = 'https://api.coin.z.com/private'
    path = '/v1/positionSummary'

    text = timestamp + method + path
    sign = hmac.new(bytes(secretKey.encode('ascii')), bytes(text.encode('ascii')), hashlib.sha256).hexdigest()
    parameters = {
        'symbol': symbol
    }

    headers = {
        'API-KEY': apiKey,
        'API-TIMESTAMP': timestamp,
        'API-SIGN': sign
    }

    res = requests.get(endPoint + path, headers=headers, params=parameters)
    positions = eval(json.dumps(res.json(), indent=2))
    if not positions['data']['list']:
        flag['position']['lot'] = 0
        flag['position']['side'] = None
        flag['position']['exist'] = False

        print('現在ポジションは存在しません')

        return

    flag['position']['lot'] = float(positions['data']['list'][0]['sumPositionQuantity'])
    flag['position']['side'] = positions['data']['list'][0]['side']
    flag['position']['exist'] = True

    print('ポジション数量：' + str(flag['position']['lot']))
    print('売買区分：' + str(flag['position']['side']))

    return


# 四本値を取得する関数
def get_price_data():
    while True:
        try:
            price_data = []
            response_data = []

            if 0 <= datetime.today().hour <= 5:
                today = datetime.today().date()

            else:
                today = datetime.today().date() + timedelta(days=1)

            for i in range(get_price_period):
                period = i - get_price_period
                date = (today + timedelta(days=period)).strftime('%Y%m%d')
                endPoint = 'https://api.coin.z.com/public'
                path = '/v1/klines?symbol=' + symbol + '&interval=' + interval + '&date=' + date

                response = requests.get(endPoint + path)
                response_data.append(response.json())

            for j in range(len(response_data)):
                for i in response_data[j]['data']:
                    price_data.append({'openTime': i['openTime'],
                                       'openTime_dt': datetime.fromtimestamp(int(i['openTime'][:-3]))
                                      .strftime('%Y/%m/%d %H:%M'),
                                       'open': i['open'],
                                       'high': i['high'],
                                       'low': i['low'],
                                       'close': i['close']})

            print('TIME： ' + str(price_data[-1]['openTime_dt']) + ' H： ' + str(price_data[-1]['high']) + ' L： ' +
                  str(price_data[-1]['low']) + ' C： ' + str(price_data[-1]['close']))

            df_price_data = pd.DataFrame(price_data)

            return df_price_data

        except Exception as e:
            print(f'action=get_price_data error={e}')
            print('10秒後にリトライします')
            time.sleep(10)

# 移動平均線のGC、DCを判定してエントリー注文を出す関数
def entry_signal(df):
    global flag

    df['MA-short'] = df['open'].rolling(MA_short_period).mean()
    df['MA-long'] = df['open'].rolling(MA_long_period).mean()

    if df.iat[-2, 6] < df.iat[-5, 7] and df.iat[-1, 6] > df.iat[-4, 7]:
        flag['order']['side'] = 'BUY'
        print('ゴールデンクロスが発生しました。')

    elif df.iat[-2, 6] > df.iat[-2, 7] and df.iat[-1, 6] < df.iat[-1, 7]:
        flag['order']['side'] = 'SELL'
        print('デッドクロスが発生しました。')

    else:
        flag['order']['side'] = ''

    return flag


# メイン処理
check_positions()

while True:

    price = get_price_data()
    entry_signal(price)

    if flag['position']['exist'] is False:
        if flag['order']['side'] == 'BUY':
            print('買いの成行注文を出します。')
            market_order()
            time.sleep(10)

            check_positions()
            time.sleep(60)

        if flag['order']['side'] == 'SELL':
            print('売りの成行注文を出します。')
            market_order()
            time.sleep(10)

            check_positions()
            time.sleep(60)

        elif flag['position']['exist'] is True:
        if flag['position']['side'] == 'BUY' and flag['order']['side'] == 'SELL':
            print('決済をして、売りの成行注文を出します。')
            close_order()
            time.sleep(10)

            market_order()
            time.sleep(10)
            check_positions()
            
            time.sleep(60)

        elif flag['position']['side'] == 'SELL' and flag['order']['side'] == 'BUY':
            print('決済をして、買いの成行注文を出します。')
            close_order()
            time.sleep(10)

            market_order()
            time.sleep(10)
            check_positions()
            
            time.sleep(60)

    time.sleep(60)

ポイントの解説

上記のサンプルコードのポイントについて解説します。

ライブラリ

Python

import time
from datetime import datetime
from datetime import timedelta
import requests
import hmac
import hashlib
import json
import pandas as pd

このプログラムでは、上記のようなライブラリを使用します。インストールされていないライブラリが場合は、pipでインストールしておいてください。

datetimeは現在時刻を取得するためのライブラリです。timestampを作成する際に使用します。

requestsは、APIリクエストを送るためのライブラリです。

hmacとhashlibは、APIシークレットを暗号化して送るために使うライブラリです。

jsonは、指値価格や注文量などのパラーメータをJSON形式に変換するためにのライブラリです。JSONとは「JavaScript Object Notation」の略で、軽量なテキストベースのデータ交換フォーマットです。

pandasは、データ分析ツールです。取得した価格データの格納や移動平均線を計算するときに使用します。

設定項目

Python

apiKey = 'YOUR_API_KEY'
secretKey = 'YOUR_SECRET'

# 設定項目
symbol = 'BTC_JPY'  # 使用する通貨ペア
interval = '1hour'  # 使用する時間軸
order_size = '0.01'  # 注文数量
MA_short_period = 5  # 移動平均線の短期の日数
MA_long_period = 25  # 移動平均線の長期の日数
get_price_period = 10  # 四本値を取得する日数

flag = {
    'order': {
        'side': '',
    },
    'position': {
        'exist': False,
        'side': None,
        'lot': 0.0,
    },
}

apiKeyとsecretKeyは、準備編で取得したapiKeyとsecretKeyを記入してください。

設定項目は、通貨ペアや使用する時間軸などを設定してください。今回はビットコインと円のレバレッジ取引を行います。使用する時間足は1時間で、1回の注文数量は0.01BTCです。

今回は移動平均線のゴールデンクロスとデッドクロスをシグナルにしているので、MA_short_periodとMA_long_periodについては、5と25を設定しています。

これらの設定は自由に変更してみてください。詳細については、GMOコインのAPIのドキュメント(https://api.coin.z.com/docs/#)で確認できます。

flagは、注文のシグナルやポジションの状態といった、プログラムが動作しているときの状態を記録しておくものです。

成行注文を出す関数

Python

timestamp = '{0}000'.format(int(time.mktime(datetime.now().timetuple())))
            method = 'POST'
            endPoint = 'https://api.coin.z.com/private'
            path = '/v1/order'
            reqBody = {
                'symbol': symbol,
                'side': flag['order']['side'],
                'size': order_size,
                'executionType': 'MARKET'
            }

            text = timestamp + method + path + json.dumps(reqBody)
            sign = hmac.new(bytes(secretKey.encode('ascii')), bytes(text.encode('ascii')), hashlib.sha256).hexdigest()

            headers = {
                'API-KEY': apiKey,
                'API-TIMESTAMP': timestamp,
                'API-SIGN': sign
            }
            
            res = requests.post(endPoint + path, headers=headers, data=json.dumps(reqBody))

上記のコードは注文を出す定型文です。

こちらで変更するところはreqBodyの内容です。'executionType'の'MARKET'は成行注文を意味しています。指値注文の場合は'LIMIT'となり、注文価格などを設定する必要があります。詳細については、GMOコインのAPIのドキュメント(https://api.coin.z.com/docs/#)で確認してください。

四本値を取得する関数

Python

def get_price_data():
    while True:
        try:
            price_data = []
            response_data = []

            if 0 <= datetime.today().hour <= 5:
                today = datetime.today().date()

            else:
                today = datetime.today().date() + timedelta(days=1)

            for i in range(get_price_period):
                period = i - get_price_period
                date = (today + timedelta(days=period)).strftime('%Y%m%d')
                endPoint = 'https://api.coin.z.com/public'
                path = '/v1/klines?symbol=' + symbol + '&interval=' + interval + '&date=' + date

                response = requests.get(endPoint + path)
                response_data.append(response.json())

            for j in range(len(response_data)):
                for i in response_data[j]['data']:
                    price_data.append({'openTime': i['openTime'],
                                       'openTime_dt': datetime.fromtimestamp(int(i['openTime'][:-3]))
                                      .strftime('%Y/%m/%d %H:%M'),
                                       'open': i['open'],
                                       'high': i['high'],
                                       'low': i['low'],
                                       'close': i['close']})

            print('TIME： ' + str(price_data[-1]['openTime_dt']) + ' H： ' + str(price_data[-1]['high']) + ' L： ' +
                  str(price_data[-1]['low']) + ' C： ' + str(price_data[-1]['close']))

            df_price_data = pd.DataFrame(price_data)

            return df_price_data

        except Exception as e:
            print(f'action=get_price_data error={e}')
            print('10秒後にリトライします')
            time.sleep(10)

GMOコインの仕様で、0時から5時の間は最新の1時間足が取得できなかったので、if 0 <= datetime.today().hour <= 5:で取得する条件を変えています。

また、GMOコインのPublic APIでは1日単位でしか取得できなかったので、for i in range(get_price_period):で必要な日数を繰り返し取得して、ひとつにまとめています。そしてデータを扱いやすいようにpandas の DataFrameに変換しています。

移動平均線のGC、DCを判定してエントリー注文を出す関数

Python

# 移動平均線のGC、DCを判定してエントリー注文を出す関数
def entry_signal(df):
    global flag

    df['MA-short'] = df['open'].rolling(MA_short_period).mean()
    df['MA-long'] = df['open'].rolling(MA_long_period).mean()

    if df.iat[-2, 6] < df.iat[-2, 7] and df.iat[-1, 6] > df.iat[-1, 7]:
        flag['order']['side'] = 'BUY'
        print('ゴールデンクロスが発生しました。')

    elif df.iat[-2, 6] > df.iat[-2, 7] and df.iat[-1, 6] < df.iat[-1, 7]:
        flag['order']['side'] = 'SELL'
        print('デッドクロスが発生しました。')

    else:
        flag['order']['side'] = ''

    return flag

移動平均線は、Pandas の関数である rollingで簡単に計算することができます。

ゴールデンクロスやデッドクロスの判定方法は、短期と長期の価格の比較が入れ替わったタイミングがゴールデンクロス又はデッドクロスが発生したタイミングとなります。

また、TA-Libというライブラリを使用すれば、ボリンジャーバンドやMACDといったインジケーターを簡単に計算することができます。他のインジケーターも試してみたい場合は、下記の記事も参考にしてください。

2024年7月1日【Python】テクニカル分析ライブラリTA-Libの使い方を解説

メイン処理

Python

# メイン処理
check_positions()

while True:

    price = get_price_data()
    entry_signal(price)

    if flag['position']['exist'] is False:
        if flag['order']['side'] == 'BUY':
            print('買いの成行注文を出します。')
            market_order()
            time.sleep(10)

            check_positions()
            time.sleep(60)

        if flag['order']['side'] == 'SELL':
            print('売りの成行注文を出します。')
            market_order()
            time.sleep(10)

            check_positions()
            time.sleep(60)

        elif flag['position']['exist'] is True:
        if flag['position']['side'] == 'BUY' and flag['order']['side'] == 'SELL':
            print('決済をして、売りの成行注文を出します。')
            close_order()
            time.sleep(10)

            market_order()
            time.sleep(10)
            check_positions()
            
            time.sleep(60)

        elif flag['position']['side'] == 'SELL' and flag['order']['side'] == 'BUY':
            print('決済をして、買いの成行注文を出します。')
            close_order()
            time.sleep(10)

            market_order()
            time.sleep(10)
            check_positions()
            
            time.sleep(60)

    time.sleep(60)