Arbitrage Bot Development

Arbitrage Trading Bot Development

Arbitrage is using price differences between markets for profit without (theoretically) market risk. In practice, risk exists: execution risk, latency risk, funding risk. Let's cover real arbitrage strategies and their technical implementation.

Arbitrage Strategy Types

Simple Arbitrage (Cross-Exchange)

Same asset trades at different prices on two exchanges. BTC on Binance — $42,100, on OKX — $42,150. Buy on Binance, sell on OKX, $50 profit is ours.

Main problem: by execution time, prices equalize. Need maximum low latency and pre-positioned balances on both exchanges.

import asyncio
import aiohttp
from decimal import Decimal

class SimpleArbitrageBot:
    def __init__(self):
        self.binance = ccxt.binance({'apiKey': BINANCE_KEY, 'secret': BINANCE_SECRET})
        self.okx = ccxt.okx({'apiKey': OKX_KEY, 'secret': OKX_SECRET})
        self.min_profit_pct = Decimal('0.15')  # minimum 0.15% after fees
    
    async def check_opportunity(self, symbol: str) -> ArbitrageOpportunity | None:
        # Parallel price request from both exchanges
        binance_ticker, okx_ticker = await asyncio.gather(
            self.binance.fetch_ticker(symbol),
            self.okx.fetch_ticker(symbol),
        )
        
        binance_bid = Decimal(str(binance_ticker['bid']))
        binance_ask = Decimal(str(binance_ticker['ask']))
        okx_bid = Decimal(str(okx_ticker['bid']))
        okx_ask = Decimal(str(okx_ticker['ask']))
        
        # Option 1: buy on Binance, sell on OKX
        if okx_bid > binance_ask:
            spread = (okx_bid - binance_ask) / binance_ask * 100
            net_spread = spread - BINANCE_TAKER_FEE - OKX_TAKER_FEE
            if net_spread > self.min_profit_pct:
                return ArbitrageOpportunity(
                    buy_exchange='binance', buy_price=binance_ask,
                    sell_exchange='okx', sell_price=okx_bid,
                    net_profit_pct=net_spread
                )
        
        # Option 2: buy on OKX, sell on Binance
        if binance_bid > okx_ask:
            spread = (binance_bid - okx_ask) / okx_ask * 100
            net_spread = spread - OKX_TAKER_FEE - BINANCE_TAKER_FEE
            if net_spread > self.min_profit_pct:
                return ArbitrageOpportunity(
                    buy_exchange='okx', buy_price=okx_ask,
                    sell_exchange='binance', sell_price=binance_bid,
                    net_profit_pct=net_spread
                )
        
        return None
    
    async def execute_arbitrage(self, opp: ArbitrageOpportunity, quantity: Decimal):
        # Execute both legs simultaneously
        buy_task = self.place_order(opp.buy_exchange, 'buy', quantity, opp.buy_price)
        sell_task = self.place_order(opp.sell_exchange, 'sell', quantity, opp.sell_price)
        
        buy_result, sell_result = await asyncio.gather(buy_task, sell_task, 
                                                        return_exceptions=True)
        
        # Handle partial execution
        if isinstance(buy_result, Exception) or isinstance(sell_result, Exception):
            await self.handle_partial_execution(buy_result, sell_result, opp)

Triangular Arbitrage (Intra-Exchange)

On one exchange: BTC → ETH → USDT → BTC. If product of rates > 1 + fees — opportunity exists.

def find_triangular_opportunity(tickers: dict) -> TriangularPath | None:
    """
    Find path A → B → C → A where final sum > initial
    """
    currencies = ['BTC', 'ETH', 'BNB', 'XRP', 'SOL']
    
    for a, b, c in permutations(currencies, 3):
        pair_ab = f"{a}/{b}"
        pair_bc = f"{b}/{c}"
        pair_ca = f"{c}/{a}"
        
        if not all(p in tickers for p in [pair_ab, pair_bc, pair_ca]):
            continue
        
        # Calculate cycle efficiency
        # Buy A->B: pay ask A/B
        rate_ab = Decimal(str(tickers[pair_ab]['ask']))
        # Buy B->C: pay ask B/C
        rate_bc = Decimal(str(tickers[pair_bc]['ask']))
        # Sell C->A: receive bid C/A
        rate_ca = Decimal(str(tickers[pair_ca]['bid']))
        
        # From 1 unit A we get:
        result = (1 / rate_ab) * (1 / rate_bc) * rate_ca
        
        # Subtract 3 fees (taker each step)
        after_fees = result * ((1 - TAKER_FEE) ** 3)
        
        profit_pct = (after_fees - 1) * 100
        
        if profit_pct > 0.05:  # minimum 0.05% profit
            return TriangularPath(
                a=a, b=b, c=c,
                rates=(rate_ab, rate_bc, rate_ca),
                profit_pct=profit_pct,
            )
    
    return None

Statistical Arbitrage (Pairs Trading)

More complex: find statistically cointegrated pairs (BTC/ETH historically move together). On spread divergence beyond threshold — long laggard, short leader.

Execution Risks and Mitigation

Execution Risk

Between spotting opportunity and execution, price can move. Binance WebSocket latency — 10–50 ms. Another bot can "eat" the opportunity.

Mitigation:

• Colocation: server in same datacenter as exchange (AWS Tokyo for Binance, AWS Frankfurt for OKX)
• WebSocket instead of REST: subscribe to orderbook updates, not polling
• Pre-placed orders: limit orders close to market

class LowLatencyArbitrageBot:
    def __init__(self):
        # Subscribe to WebSocket, update cache
        self.price_cache = {}  # current prices without request latency
    
    async def subscribe_prices(self, symbol: str):
        """WebSocket subscription — update cache on each tick"""
        async with websockets.connect(BINANCE_WS_URL) as ws:
            await ws.send(json.dumps({
                'method': 'SUBSCRIBE',
                'params': [f'{symbol.lower()}@bookTicker'],
                'id': 1
            }))
            async for msg in ws:
                data = json.loads(msg)
                # bookTicker gives best bid/ask without delay
                self.price_cache[symbol] = {
                    'bid': Decimal(data['b']),
                    'ask': Decimal(data['a']),
                    'ts': time.time_ns(),
                }
                await self.check_opportunity_fast(symbol)

Inventory Risk

If one leg executed but other didn't — open position with market risk. Called "leg risk".

Transfer Risk (Cross-Exchange)

Cross-exchange arbitrage requires maintaining balances on both exchanges. Transfer between takes 10–60 minutes (multiple blockchain confirmations). Price can shift.

Solution: capital-intensive model — hold sufficient balance on each exchange ahead of time. Rebalance every few hours when imbalance exceeds threshold.

async def check_balance_rebalance(self):
    """If imbalance > threshold — auto-transfer"""
    for exchange, balance in self.get_all_balances().items():
        for currency, amount in balance.items():
            target = self.target_balances[exchange][currency]
            deviation = abs(amount - target) / target * 100
            
            if deviation > 20:  # imbalance more than 20%
                await self.initiate_transfer(exchange, currency, target - amount)

P&L and Monitoring

class ArbitragePnL:
    def record_trade(self, trade: ArbitrageTrade):
        gross_profit = trade.sell_amount - trade.buy_amount
        fees = trade.buy_fee + trade.sell_fee + trade.transfer_fee
        net_profit = gross_profit - fees
        
        self.daily_pnl += net_profit
        self.total_volume += trade.quantity
        
        if net_profit < 0:
            self.losing_trades += 1
            log.warning(f"Losing arbitrage: net {net_profit:.4f} USDT")
    
    def get_stats(self) -> dict:
        return {
            'daily_pnl': self.daily_pnl,
            'win_rate': self.winning_trades / max(self.total_trades, 1),
            'avg_profit_per_trade': self.daily_pnl / max(self.total_trades, 1),
            'volume': self.total_volume,
        }

Development Timeline

• Simple cross-exchange arbitrage with 2 exchanges: 4–6 weeks
• Triangular arbitrage: 3–4 weeks
• Statistical arbitrage: 6–10 weeks (includes cointegration research)
• Production-ready system with monitoring and auto-rebalancing: 3–4 months