三角裁定取引・Triangular arbitrage（三角アービトラージ）の仮想通貨（暗号資産）アルゴリズム取引をPython×CCXTで製作

三角裁定取引、Triangular arbitrage（三角アービトラージ）とは、３つ以上の通貨間で売り買いを行い、最初に売った通貨が最終的に最初に売った通貨を増やす形で買い戻すことを目的とする取引のこと。通貨を増やせるタイミングが発生してから取引を開始できるので損失のリスクが低く、収益を上げることができる手法です。
但し、低リスクで収益を得ることが出来るメリットの反面、大勝ができるような取引手法ではありません。

Python × CCXTによるアルゴリズム取引

三角裁定取引、Triangular arbitrage（三角アービトラージ）は３つの通貨間の価格不一致により通貨を増やして買い戻せる機会が発生しますが、そのような機会は頻繁に発生せず、また発生してもすぐに通貨間の価格が正常に戻ってしてしまいます。よって人間がモニターで複数の通貨を監視し、マウスとキーボードで取引を行うことは不可能です。
必然的にコンピューターによるアルゴリズム取引を行う必要があります。仮想通貨（暗号資産）取引所が提供しているAPIを利用してトレーディングアプリを製作するにあたり、適しているコンピューター言語がPythonになります。
PythonはCCXTを利用することで、価格の取得などのコマンドを簡易的なコードで記述することができます。

仮想通貨（暗号資産）取引所 BINANCE

三角裁定取引、Triangular arbitrage（三角アービトラージ）は、3つ以上の通貨を売り買いする必要があるため3つ以上の基軸通貨を用意している仮想通貨（暗号資産）取引所で取引しなくてはなりません。
残念ながらbitFryer（ビットフライヤー）やcoincheck（コインチェック）のような日本の仮想通貨（暗号資産）取引所ではBitcoin（ビットコイン）やEthereum（イーサリアム）の2つの通貨までになっております。
一方、世界最大の仮想通貨（暗号資産）を取り扱っているBINANCEでは、
1.ビットコイン(BTC)
2.バイナンスコイン(BNB)
3.イーサリアム(ETH)
4.テザー(USDT)
を基軸通貨として取り扱っています。

Python

ビットコイン(BTC)、イーサリアム(ETH)、テザー(USDT)の3つの仮想通貨（暗号資産）で三角裁定取引、Triangular arbitrage（三角アービトラージ）を行う価格を監視するPythonのコードを紹介します。

TriangularArbitrage.py

# -*- coding: utf-8 -*-
"""
Created on Wed Apr 27 15:16:46 2022

@author: Noriyuki
"""

#import(モジュールを宣言)
import ccxt #複数の仮想通貨取引所のAPIライブラリ
import time #sleep関数を使用する為
from pprint import pprint #処理内容をコンソールに出力する為
import csv #ビットコインの価格をCSVファイルへ書込む為
import datetime #日時取得の為
import sys #終了処理の為

#定数定義
CYCLEWAIT = 5 #待機秒
CYCLECOUNT = 60 #繰返す回数 5秒×60回 = 300秒 (5分)
FILEPATH = 'TriangularArbitrageData.csv' #ファイル名
BTC = 0.1 

#CCXTの情報取得
binanceCCXT = ccxt.binance() #バイナス
bitflyerCCXT = ccxt.bitflyer()  #ビットフライヤー

def main(argv):
    # 引数の処理
    Cyclecount = 0
    if len(argv) >= 2:
        Cyclecount = argv[1]
    else:
        Cyclecount = CYCLECOUNT
        
    Cyclewait = 0
    if len(argv) >= 3:
        Cyclewait = argv[2]
    else:
        Cyclewait = CYCLEWAIT
    
    btc = 0
    if len(argv) >= 4:
        btc = argv[3]
    else:
        btc = BTC
    

    print('Roop:' + str(Cyclecount))
    print('Wait:' + str(Cyclewait))

    #csvモジュールで書込み
    with open(FILEPATH, 'a', newline='') as f: #ファイルオープン
        writer = csv.writer(f)
        #CSVヘッダー
        writer.writerow(['Count', 'Time', 'ETH/BTC', 'ETH/USDT', 'BTC/USDT', 'Profit', 'Rate', 'Result'])
        #カウント初期化
        i = 1
        while i <= int(Cyclecount): #定数定義された回数までループ

            dt = datetime.datetime.now().strftime('%H:%M:%S') #時刻を変数に代入
            tickerETHBTC = binanceCCXT.fetch_ticker(symbol='ETH/BTC')
            tickerETHUSDT = binanceCCXT.fetch_ticker(symbol='ETH/USDT')
            tickerBTCUSDT = binanceCCXT.fetch_ticker(symbol='BTC/USDT')
            bitflyer = bitflyerCCXT.fetch_ticker('BTC/JPY', params = { "product_code" : "BTC_JPY" }) #CCXTでビットフライヤーの価格取得
            
            lastETHBTC = tickerETHBTC["last"]
            lastETHUSDT = tickerETHUSDT["last"]
            lastBTCUSDT = tickerBTCUSDT["last"]
            lastbitflyer = bitflyer["last"]
            
            #結果 = ETH_BTC ÷ ETH_USDT × BTC_USDT 
            result = 0
            result = float(btc) * lastETHBTC / lastETHUSDT * lastBTCUSDT
            
            if result > float(btc):
                pl = 'Profit'
            elif result == btc:
                pl = '-'
            else:
                pl = 'Loss'
            
            #損益を日本円に換算
            profit = 0
            profit = (result - float(btc)) * lastbitflyer
            
            #上昇率/下降率
            rate = (float(btc) / profit) * 100
            
            #CSV
            writer.writerow([i, dt, lastETHBTC, lastETHUSDT, lastBTCUSDT, '{:.3f}'.format(profit), '{:.3f}'.format(rate), pl])

            #コンソール
            pprint(str(i) + " [" + str(dt) + "] [" + str(lastETHBTC) + "] [" + str(lastETHUSDT) + "] [" + str(lastBTCUSDT) + "] [￥" + '{:.3f}'.format(profit) + "] [" + '{:.3f}'.format(rate) + "%] [" + str(pl) + "]")
            
            #回数をカウントアップ
            i += 1
            #待機
            time.sleep(int(Cyclewait))
    #終了
    sys.exit()
    
if __name__ == '__main__':
    choice = input("実行します。よろしいですか？ [y/n]: ").lower()
    if choice in ['y', 'Y']:
        main(sys.argv)
    elif choice in ['n', 'N']:
        sys.exit()