How It Works

Project Overview

This arbitrage detection engine is a high-frequency trading research tool built to identify price inefficiencies across cryptocurrency markets. It demonstrates two types of arbitrage strategies:

• Triangular Arbitrage — Exploiting price discrepancies between three trading pairs on a single exchange
• Cross-Exchange Arbitrage — Profiting from price differences of the same asset across multiple exchanges

The engine connects to real exchange WebSocket APIs to receive live market data, processes thousands of price updates per second, and identifies profitable opportunities in real-time.

Triangular Arbitrage

Triangular arbitrage exploits pricing inefficiencies between three related trading pairs on a single exchange. The idea is to start with one currency, trade through two others, and end up with more of the original currency than you started with.

How It Works

The Mathematics

For a triangle path A → B → C → A, we calculate the gross return based on whether we're buying or selling at each leg:

The formula accounts for:

• Ask prices (what you pay when buying)
• Bid prices (what you receive when selling)
• Trading fees applied at each leg (typically 0.1% per trade on Binance)

Implementation in Code

# From strategy/calculator.py
def calculate(self, path: TrianglePath, orderbook: OrderbookManager) -> CalcResult:
    amount = 1.0  # Start with 1 unit

    for leg in path.legs:
        bbo = orderbook.get(leg.symbol)  # Get best bid/offer

        if leg.side == OrderSide.BUY:
            price = bbo.ask_price  # Pay the ask when buying
            amount = (amount / price) * (1 - self.fee_rate)
        else:
            price = bbo.bid_price  # Receive the bid when selling
            amount = (amount * price) * (1 - self.fee_rate)

    return CalcResult(net_return=amount)

Cross-Exchange Arbitrage

Cross-exchange arbitrage exploits price differences for the same asset across different exchanges. If Bitcoin is trading at $97,500 on Binance but $97,600 on Kraken, you could theoretically buy on Binance and sell on Kraken for a $100 profit.

Connected Exchanges

How It Works

Code Architecture

The project follows a modular, layered architecture designed for maintainability and performance. Each module has a single responsibility and communicates through well-defined interfaces.

WebSocket Connections

The engine maintains persistent WebSocket connections to exchange APIs for real-time price updates. Each exchange has its own message format, so we normalize the data into a common structure.

Exchange WebSocket URLs

# Binance - Combined stream for multiple symbols
wss://stream.binance.com:9443/stream?streams=btcusdt@bookTicker/ethusdt@bookTicker

# Kraken - Ticker subscription
wss://ws.kraken.com

# Coinbase - Product ticker channel
wss://ws-feed.exchange.coinbase.com

# OKX - Public ticker channel
wss://ws.okx.com:8443/ws/v5/public

# Bybit - Spot tickers
wss://stream.bybit.com/v5/public/spot

Message Handling

Each exchange sends data in a different format. Here's how we handle Binance's bookTicker messages:

# Binance bookTicker message format
{
    "s": "BTCUSDT",     # Symbol
    "b": "97485.20",   # Best bid price
    "B": "2.5",        # Best bid quantity
    "a": "97485.50",   # Best ask price
    "A": "1.2"         # Best ask quantity
}

# We normalize to our BBO (Best Bid/Offer) structure
@dataclass
class BBO:
    symbol: str
    bid_price: float
    bid_qty: float
    ask_price: float
    ask_qty: float
    timestamp_us: int

Reconnection Logic

WebSocket connections can drop due to network issues or exchange maintenance. The engine implements automatic reconnection with exponential backoff:

async def _run_binance(self):
    while self._state.running:
        try:
            async with session.ws_connect(url, heartbeat=30) as ws:
                async for msg in ws:
                    await self._handle_message(msg)
        except Exception:
            if self._state.running:
                await asyncio.sleep(3)  # Reconnect delay

Opportunity Detection

The core of the engine is the opportunity detection loop. Every time a price updates, we check if any arbitrage opportunities have emerged.

Triangular Detection Flow

1. Receive price update for symbol (e.g., BTCUSDT)
2. Find all triangle paths that include this symbol
3. For each path, check if we have prices for all 3 legs
4. Calculate potential profit using the formula
5. If profit > threshold, emit opportunity event

async def _check_opportunities(self, updated_symbol: str):
    for triangle in self._triangles:
        if updated_symbol not in triangle.symbols:
            continue

        # Get all prices for this triangle
        prices = {}
        for leg in triangle.legs:
            bbo = self._orderbook.get(leg.symbol)
            if not bbo:
                break
            prices[leg.symbol] = bbo.ask_price if leg.side == BUY else bbo.bid_price

        # Calculate and emit if profitable
        result = self._calculator.calculate(triangle, self._orderbook)
        profit_pct = (result.net_return - 1) * 100

        if profit_pct > self._min_profit_threshold:
            await self._emit("opportunity", {...})

Cross-Exchange Detection

For cross-exchange arbitrage, we compare the best bid across all exchanges with the best ask across all exchanges:

async def _check_opportunities(self):
    for symbol, exchanges in self._state.prices.items():
        # Find best bid (highest) and best ask (lowest)
        best_bid, best_bid_ex = max(
            ((p["bid"], ex) for ex, p in exchanges.items()),
            key=lambda x: x[0]
        )
        best_ask, best_ask_ex = min(
            ((p["ask"], ex) for ex, p in exchanges.items()),
            key=lambda x: x[0]
        )

        # Calculate profit: buy at best_ask, sell at best_bid
        profit_pct = ((best_bid - best_ask) / best_ask) * 100

Real-Time Dashboard

The dashboard uses WebSockets for bidirectional communication between the server and browser. This enables instant updates without polling.

Server → Client Events

• price — New price update from an exchange
• opportunity — Detected arbitrage opportunity
• connection — Exchange connection status change
• status — Bot running state change

Client → Server Commands

• start — Start the detection engine
• stop — Stop the detection engine
• setMode — Switch between arbitrage modes

# FastAPI WebSocket endpoint
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    connected_clients.append(websocket)

    try:
        while True:
            data = await websocket.receive_text()
            msg = orjson.loads(data)

            if msg["action"] == "start":
                await start_bot()
            elif msg["action"] == "stop":
                await stop_bot()
    finally:
        connected_clients.remove(websocket)

Tech Stack

Performance Optimizations

• O(1) orderbook lookups — Prices stored in a flat dictionary for instant access
• Pre-computed triangle paths — Paths discovered at startup, not during detection
• Slots dataclasses — 30% memory reduction with @dataclass(slots=True)
• Connection reuse — Single aiohttp session per exchange
• Float arithmetic — Using floats over Decimal for 10x calculation speed (precision validated)