仮想通貨現物アビトラ機会検知機（CCXTからの引用）

趣旨

CCXTにおいてあったアビトラ検知用のコードを少しだけ改造してみやすくしてみました。
コード自体は元ページのものを見た方がいいかも知れません。

取引所間のシンボルの違いによってあまり使い物にならないと思いますが、botいじってる感万歳の結果が表示されるのでぜひやってみてください。

python arbitrage.py

import ccxt
import sys


def style(s, style):
    return style + s + '\033[0m'

def green(s):
    return style(s, '\033[92m')

def yellow(s):
    return style(s, '\033[93m')

def red(s):
    return style(s, '\033[91m')

def dump(*args):
    print(' '.join([str(arg) for arg in args]))

def fetch_prices(exchange, symbols):
    prices = {}
    for symbol in symbols:
        try:
            ticker = exchange.fetch_ticker(symbol)
            prices[symbol] = ticker['last']
        except Exception as e:
            print(f"Could not fetch price for {symbol} on {exchange.id}: {str(e)}")
    return prices

def calculate_arbitrage(exchanges, arbitrableSymbols):
    investment_amount_usd = 10000  # Assume an investment amount of $10,000 USD
    for symbol in arbitrableSymbols:
        prices = [(exchange.id, fetch_prices(exchange, [symbol]).get(symbol, None)) for exchange in exchanges.values() if symbol in exchange.symbols]
        prices = [price for price in prices if price[1] is not None]
        if len(prices) > 1:
            max_price = max(prices, key=lambda x: x[1])
            min_price = min(prices, key=lambda x: x[1])
            if max_price[1] > min_price[1]:
                amount_to_buy = investment_amount_usd / min_price[1]
                potential_revenue = amount_to_buy * max_price[1]
                potential_profit = potential_revenue - investment_amount_usd
                dump(green(f"Arbitrage Opportunity for {symbol}: Buy at {min_price[0]} for ${min_price[1]}, Sell at {max_price[0]} for ${max_price[1]}, Potential Profit: ${potential_profit:.2f}"))

def main():
    if len(sys.argv) > 1:
        ids = sys.argv[1:]
        exchanges = {}

        for id in ids:
            exchange = getattr(ccxt, id)()
            exchanges[id] = exchange
            try:
                exchange.load_markets()
                dump(green(f"{id} loaded {len(exchange.symbols)} markets"))
            except Exception as e:
                dump(red(f"Failed to load markets for {id}: {str(e)}"))
                continue

        allSymbols = [symbol for id in ids for symbol in exchanges[id].symbols if hasattr(exchanges[id], 'symbols')]
        uniqueSymbols = list(set(allSymbols))
        arbitrableSymbols = sorted([symbol for symbol in uniqueSymbols if allSymbols.count(symbol) > 1])

        calculate_arbitrage(exchanges, arbitrableSymbols)
    else:
        dump("Usage: python script.py exchange1 exchange2 exchange3 ...")

if __name__ == '__main__':
    main()

以下のコマンドで実行します。

python arbitrage_pairs.py binance bybit bitget gate bingx okx