CCXT Triangular Arbitrage: A Complete Python Example

Triangular arbitrage is one of the oldest tricks in financial trading — and in crypto, it's very much alive. The idea is simple: instead of waiting for price differences between exchanges, you exploit price imbalances between three trading pairs on the same exchange. CCXT, the universal crypto trading library, gives you everything you need to build this kind of scanner in Python with surprisingly little code. This guide walks you through the logic, the math, and working code you can actually run.

What Is Triangular Arbitrage and Why Does It Work

Imagine you have $1,000 USDT sitting on Binance. You notice that if you convert USDT → BTC → ETH → USDT in sequence, you end up with $1,012 USDT. That $12 profit came from a momentary misalignment in the prices of three trading pairs. That's triangular arbitrage.

Here's a real-world analogy: you're at an airport currency exchange. You convert USD to EUR, EUR to GBP, then GBP back to USD — and somehow end up with more dollars than you started with because each counter had slightly off rates. In crypto, market makers and bots constantly try to keep rates aligned, but during volatile periods or low liquidity windows, gaps appear.

On platforms like Binance and OKX, which list hundreds of trading pairs, triangular paths are everywhere. A common triangle looks like: USDT → BTC → ETH → USDT, but you can also find paths through altcoins like BNB, SOL, or MATIC. The key is that multiplying through all three exchange rates should equal exactly 1.0 if markets are efficient — any deviation above 1.0 (minus fees) is a potential trade.

Key Takeaway: Triangular arbitrage happens within a single exchange, which means no withdrawal delays, no cross-exchange transfer risk, and faster execution. The downside is that opportunities are smaller and disappear in milliseconds.

Setting Up CCXT for Arbitrage Scanning

CCXT (CryptoCurrency eXchange Trading Library) supports over 100 exchanges through a single unified API. For triangular arbitrage, you need real-time order book or ticker data across multiple pairs — and CCXT makes fetching that data consistent regardless of which exchange you target.

Start by installing CCXT and setting up your exchange connection. For scanning purposes, you don't need API keys — public market data is available without authentication. You only need keys when you're ready to execute real trades.

pip install ccxt

import ccxt

# Connect to Binance (no API key needed for market data)
exchange = ccxt.binance({
    'enableRateLimit': True,  # respect exchange rate limits
})

# Load all available markets
markets = exchange.load_markets()
print(f"Loaded {len(markets)} markets from Binance")

# Fetch all tickers at once (much faster than one-by-one)
tickers = exchange.fetch_tickers()
print(f"Fetched {len(tickers)} tickers")

The key efficiency trick here is `fetch_tickers()` — it pulls all prices in a single API call instead of requesting each pair individually. On Binance this returns 400+ tickers in one shot, which is essential for keeping your scanner fast enough to catch real opportunities.

Building the Triangular Arbitrage Scanner

The core logic involves three steps: find all valid triangles from available trading pairs, calculate the product of exchange rates through each triangle, and flag any path where the product exceeds 1.0 after fees. Here's a complete working scanner:

import ccxt
from itertools import permutations

def find_triangles(markets, base_currency='USDT'):
    """
    Find all triangular paths starting and ending with base_currency.
    Example: USDT -> BTC -> ETH -> USDT
    """
    triangles = []
    
    # Get all currencies that pair with base_currency
    base_pairs = [m for m in markets if m.endswith(f'/{base_currency}')]
    intermediate_currencies = [m.split('/')[0] for m in base_pairs]
    
    for mid_currency in intermediate_currencies:
        # Find pairs between mid_currency and other currencies
        mid_pairs = [
            m for m in markets
            if m.startswith(f'{mid_currency}/')
            and not m.endswith(f'/{base_currency}')
        ]
        
        for pair in mid_pairs:
            third_currency = pair.split('/')[1]
            # Check if third_currency pairs back to base
            return_pair = f'{third_currency}/{base_currency}'
            if return_pair in markets:
                triangles.append((
                    f'{mid_currency}/{base_currency}',  # leg 1: buy mid with base
                    f'{mid_currency}/{third_currency}',  # leg 2: sell mid for third
                    f'{third_currency}/{base_currency}'  # leg 3: sell third for base
                ))
    
    return triangles


def calculate_profit(triangle, tickers, fee_rate=0.001):
    """
    Calculate the profit multiplier for a given triangle.
    Returns a value > 1.0 if profitable (before considering slippage).
    """
    leg1, leg2, leg3 = triangle
    
    try:
        # Leg 1: buy base->mid (use ask price, we're buying)
        rate1 = 1 / tickers[leg1]['ask']  # how much mid we get per USDT
        
        # Leg 2: sell mid->third (use bid price, we're selling mid)
        rate2 = tickers[leg2]['bid']  # how much third we get per mid
        
        # Leg 3: sell third->base (use bid price, we're selling third)
        rate3 = tickers[leg3]['bid']  # how much USDT we get per third
        
        # Calculate profit multiplier after fees
        multiplier = rate1 * rate2 * rate3
        multiplier_after_fees = multiplier * ((1 - fee_rate) ** 3)
        
        return multiplier_after_fees
    except (KeyError, TypeError, ZeroDivisionError):
        return 0.0


# Main scanner
exchange = ccxt.binance({'enableRateLimit': True})
markets = exchange.load_markets()
tickers = exchange.fetch_tickers()

print("Finding triangular paths...")
triangles = find_triangles(markets, base_currency='USDT')
print(f"Found {len(triangles)} triangular paths")

print("\nScanning for opportunities...")
opportunities = []

for triangle in triangles:
    profit = calculate_profit(triangle, tickers)
    if profit > 1.001:  # at least 0.1% profit above fees
        opportunities.append((triangle, profit))

# Sort by most profitable first
opportunities.sort(key=lambda x: x[1], reverse=True)

if opportunities:
    print(f"\nFound {len(opportunities)} opportunities!")
    for triangle, profit in opportunities[:5]:
        print(f"  {triangle[0]} -> {triangle[1]} -> {triangle[2]}")
        print(f"  Profit multiplier: {profit:.6f} ({(profit-1)*100:.4f}%)")
else:
    print("No profitable opportunities at this moment (market is efficient)")

Key Takeaway: Notice we use `ask` price when buying and `bid` price when selling — not the mid price. This is critical. Using mid prices will give you fake opportunities that don't exist in reality. Always work with executable prices.

Executing Trades and Managing Real Positions

Scanning is the easy part. Execution is where most strategies fail. The biggest challenge in triangular arbitrage is that all three legs need to execute almost simultaneously — if leg 1 fills but leg 2 slips, you can end up in a worse position than if you'd done nothing. Here's a basic execution template with proper error handling:

import ccxt
import time

def execute_triangle(exchange, triangle, amount_usdt, dry_run=True):
    """
    Execute a triangular arbitrage trade.
    Always test with dry_run=True first!
    """
    leg1, leg2, leg3 = triangle
    orders = []
    
    try:
        print(f"Executing triangle: {leg1} -> {leg2} -> {leg3}")
        print(f"Starting with {amount_usdt} USDT")
        
        # Leg 1: Buy intermediate currency with USDT
        ticker1 = exchange.fetch_ticker(leg1)
        leg1_amount = amount_usdt / ticker1['ask']
        
        if not dry_run:
            order1 = exchange.create_market_buy_order(leg1, leg1_amount)
            orders.append(order1)
            print(f"Leg 1 filled: bought {leg1_amount:.6f} {leg1.split('/')[0]}")
            time.sleep(0.1)  # small pause to confirm fill
        else:
            print(f"[DRY RUN] Would buy {leg1_amount:.6f} {leg1.split('/')[0]}")
        
        # Leg 2: Sell intermediate for third currency
        ticker2 = exchange.fetch_ticker(leg2)
        leg2_amount = leg1_amount  # selling all of what we got
        
        if not dry_run:
            order2 = exchange.create_market_sell_order(leg2, leg2_amount)
            orders.append(order2)
            print(f"Leg 2 filled: sold for {leg2.split('/')[1]}")
            time.sleep(0.1)
        else:
            leg3_amount = leg1_amount * ticker2['bid']
            print(f"[DRY RUN] Would get {leg3_amount:.6f} {leg2.split('/')[1]}")
        
        # Leg 3: Sell third currency back to USDT
        if not dry_run:
            order3 = exchange.create_market_sell_order(leg3, leg3_amount)
            orders.append(order3)
            print(f"Leg 3 filled: back to USDT")
        else:
            final_usdt = leg3_amount * exchange.fetch_ticker(leg3)['bid']
            profit_pct = (final_usdt / amount_usdt - 1) * 100
            print(f"[DRY RUN] Would end with {final_usdt:.4f} USDT ({profit_pct:+.4f}%)")
        
        return orders
        
    except ccxt.InsufficientFunds as e:
        print(f"Insufficient funds: {e}")
        return None
    except ccxt.NetworkError as e:
        print(f"Network error on leg {len(orders)+1}: {e}")
        # IMPORTANT: if orders exist, you may need to unwind manually
        return orders


# Usage example - ALWAYS start with dry_run=True
exchange = ccxt.binance({
    'apiKey': 'YOUR_API_KEY',
    'secret': 'YOUR_SECRET',
    'enableRateLimit': True,
})

# Test with dry run first
execute_triangle(
    exchange,
    triangle=('BTC/USDT', 'ETH/BTC', 'ETH/USDT'),
    amount_usdt=100,
    dry_run=True
)

Platforms like Bybit and OKX also work with this same code — just swap `ccxt.binance()` for `ccxt.bybit()` or `ccxt.okx()`. The CCXT interface is unified, so your logic stays identical. However, each exchange has different fee structures and rate limits, so always check those before deploying live.

Risks, Slippage, and Why Most Opportunities Are Fake

Here's the uncomfortable truth: most opportunities your scanner finds won't be profitable when you actually try to trade them. There are several reasons for this that aren't obvious until you've lost money learning them the hard way.

• Slippage kills thin opportunities: Market orders eat into the order book. If you're trading $10,000 and the opportunity is 0.2%, but your order moves the price 0.15%, you're left with almost nothing.
• Ticker data is stale: By the time fetch_tickers() returns and you process the data, prices have already moved. On Binance especially, milliseconds matter.
• Taker fees stack fast: Three legs at 0.1% each means 0.3% in fees alone. You need an opportunity above 0.35% just to break even — those are genuinely rare.
• Partial fills create stranded positions: If leg 2 only partially fills, you hold an asset you didn't plan to hold. Now you have execution risk plus market risk.
• Competition is fierce: Dedicated HFT bots with co-located servers are scanning the same data. If you found it via REST API, they found it 100ms earlier via WebSocket.

A more realistic approach for retail traders is to use triangular arbitrage logic as a signal layer rather than a full execution strategy. Tools like VoiceOfChain provide real-time market signals that help you understand when specific pairs are mispriced — letting you time entries better rather than trying to beat bots to microsecond opportunities.

The practical sweet spot for retail traders using CCXT is running this kind of scanner on Gate.io or KuCoin spot markets, focusing on lower-liquidity altcoin triangles where HFT competition is lighter and spreads are wider. You won't get rich, but you can learn execution mechanics without competing directly against professional infrastructure.

Key Takeaway: Start by paper-trading your scanner for at least a week. Log every 'opportunity' it finds, then check what the prices were 500ms later. You'll quickly see how fast they vanish — and calibrate your minimum threshold accordingly.

Upgrading Your Scanner: WebSockets and Async

REST API polling is the slowest way to do this. The real upgrade is switching to WebSocket streams so prices update in real time. CCXT Pro (the async/websocket version of CCXT) makes this relatively painless:

import asyncio
import ccxt.pro as ccxtpro

async def watch_tickers_and_scan():
    exchange = ccxtpro.binance({'enableRateLimit': True})
    
    # Symbols we care about for a BTC/ETH/USDT triangle
    symbols = ['BTC/USDT', 'ETH/BTC', 'ETH/USDT']
    
    # Cache for latest prices
    price_cache = {}
    
    async def watch_symbol(symbol):
        while True:
            ticker = await exchange.watch_ticker(symbol)
            price_cache[symbol] = ticker
            
            # Check triangle whenever any price updates
            if len(price_cache) == 3:
                check_triangle(price_cache)
    
    def check_triangle(cache):
        try:
            # USDT -> BTC -> ETH -> USDT
            rate1 = 1 / cache['BTC/USDT']['ask']   # get BTC per USDT
            rate2 = cache['ETH/BTC']['bid']          # get ETH per BTC  
            rate3 = cache['ETH/USDT']['bid']         # get USDT per ETH
            
            multiplier = rate1 * rate2 * rate3 * (0.999 ** 3)  # 0.1% fee x3
            
            if multiplier > 1.002:  # 0.2% minimum threshold
                profit_pct = (multiplier - 1) * 100
                print(f"OPPORTUNITY: {profit_pct:.4f}% profit")
        except KeyError:
            pass
    
    # Watch all symbols concurrently
    await asyncio.gather(*[watch_symbol(s) for s in symbols])
    await exchange.close()

asyncio.run(watch_tickers_and_scan())

This WebSocket version reacts to price changes in under 10ms on a local connection, compared to 200-500ms with REST polling. It's still not competitive with co-located HFT bots, but it's dramatically more realistic for testing whether a triangle actually produces consistent signals.

Frequently Asked Questions

Where to Go From Here

Building a triangular arbitrage scanner with CCXT is one of the best practical exercises in algo trading — even if you never trade it live. You'll learn how order books work, how fees compound, how to think about execution latency, and how to work with real market data at scale. Those skills transfer directly to building any kind of trading bot.

If you want to move beyond pure mechanics, consider layering in market signals from platforms like VoiceOfChain alongside your scanner. When the broader market is in a trending or volatile state, pricing inefficiencies appear more frequently across pairs — knowing that context can help you decide when your scanner is worth running versus when the market is too efficient to bother.

The natural next steps after mastering this scanner are: switching to WebSocket feeds for lower latency, adding order book depth analysis to filter out opportunities where your trade size would cause slippage, and testing on testnet environments that Binance and OKX both provide. Build incrementally, log everything, and don't deploy real capital until you've run paper tests for at least a month.

• Start with REST API scanning to learn the logic, then upgrade to CCXT Pro WebSockets
• Always use bid/ask prices — never mid prices — in your profit calculations
• Three legs at 0.1% fee each means you need >0.3% gross profit just to break even
• Gate.io and KuCoin have less HFT competition on altcoin triangles than Binance
• Paper trade for at least a month before risking real capital on execution