Build a Crypto Arbitrage Bot with Purple Flea

Arbitrage is one of the oldest trading strategies — and one of the few where you can earn without directional market risk. Purple Flea's 275-market perpetual trading API creates three distinct arbitrage opportunities accessible to AI agents. This guide covers each with full Python implementations and realistic profit calculations.

The Three Arbitrage Types

Strategy 1: Funding Rate Arbitrage

Perpetual futures maintain price parity with spot via a funding mechanism: when the perp trades above spot, longs pay shorts every 8 hours. When funding is positive and above your borrowing cost, you can earn passively by holding a short perp position.

The math: BTC-USD funding rate is 0.01% every 8 hours = 0.03% daily = ~10.95% APY. On $100 of short exposure, that's $10.95/year in pure funding income, market-neutral.

import requests
import time

PF_KEY = "pf_live_your_key"
PF_BASE = "https://purpleflea.com/api"

pf = requests.Session()
pf.headers["Authorization"] = f"Bearer {PF_KEY}"

MIN_FUNDING_APY = 20.0  # Only enter if annualized rate > 20%
POSITION_SIZE = 10.0   # $10 per position

def get_funding_rates() -> list:
    """Fetch funding rates for all perp markets."""
    r = pf.get(f"{PF_BASE}/trading/funding-rates")
    return r.json()

def funding_rate_to_apy(rate_8h: float) -> float:
    """Convert 8-hour funding rate to annualized percentage."""
    return rate_8h * 3 * 365 * 100

def find_funding_opportunities() -> list:
    """Find markets with attractive positive funding rates."""
    rates = get_funding_rates()
    opportunities = []
    for market in rates:
        apy = funding_rate_to_apy(market["funding_rate"])
        if apy >= MIN_FUNDING_APY:
            opportunities.append({
                "symbol": market["symbol"],
                "funding_rate_8h": market["funding_rate"],
                "apy": apy,
                "next_funding_time": market.get("next_funding_time")
            })
    return sorted(opportunities, key=lambda x: -x["apy"])

def enter_funding_arb(symbol: str, size_usd: float):
    """Enter a funding rate arbitrage: short perp (collects funding)."""
    # Note: In a full implementation, you'd also buy spot to hedge
    # Here we show the perp short (the funding-collecting leg)
    r = pf.post(f"{PF_BASE}/trading/perp/order", json={
        "symbol": symbol,
        "side": "short",
        "size_usd": size_usd,
        "leverage": 1,  # No leverage — this is a carry trade
        "reduce_only": False
    })
    return r.json()

# Funding rate arb scanner loop
def run_funding_arb_bot():
    positions = {}  # Track active arb positions

    while True:
        opps = find_funding_opportunities()
        print(f"\nFunding opportunities: {len(opps)}")

        for opp in opps[:3]:  # Max 3 simultaneous positions
            sym = opp["symbol"]
            print(f"  {sym}: {opp['apy']:.1f}% APY (8h rate: {opp['funding_rate_8h']*100:.4f}%)")

            if sym not in positions:
                result = enter_funding_arb(sym, POSITION_SIZE)
                positions[sym] = result
                print(f"  → Entered funding arb: {sym} ${POSITION_SIZE} short")

        # Close positions where funding has dropped below threshold
        active_syms = {o["symbol"] for o in opps}
        for sym in list(positions.keys()):
            if sym not in active_syms:
                pf.post(f"{PF_BASE}/trading/perp/close", json={"symbol": sym})
                del positions[sym]
                print(f"  ← Closed {sym} (funding dropped below threshold)")

        time.sleep(3600)  # Check every hour

Strategy 2: Cross-Market Price Arbitrage

With 275 markets, you can occasionally find price discrepancies between related pairs. The most reliable version involves triangular arbitrage: if BTC-USD × ETH-BTC ≠ ETH-USD (after fees), there's a risk-free profit.

import asyncio
import aiohttp

FEE_RATE = 0.001  # 0.1% taker fee per leg
MIN_PROFIT_PCT = 0.3  # Minimum 0.3% profit after fees

async def get_prices_concurrent(session, symbols: list) -> dict:
    """Fetch multiple prices concurrently for speed."""
    tasks = [
        session.get(
            f"{PF_BASE}/trading/price/{sym}",
            headers={"Authorization": f"Bearer {PF_KEY}"}
        )
        for sym in symbols
    ]
    responses = await asyncio.gather(*tasks)
    prices = {}
    for sym, resp in zip(symbols, responses):
        data = await resp.json()
        prices[sym] = data["price"]
    return prices

async def scan_triangular_arb():
    """Scan for triangular arbitrage opportunities across BTC/ETH/SOL."""
    async with aiohttp.ClientSession() as session:
        # Fetch all relevant prices simultaneously
        prices = await get_prices_concurrent(session, [
            "BTC-USD", "ETH-USD", "SOL-USD",
            "ETH-BTC", "SOL-BTC", "SOL-ETH"
        ])

        # Check: BTC/USD → ETH/BTC → ETH/USD triangle
        btc_usd = prices["BTC-USD"]
        eth_btc = prices["ETH-BTC"]
        eth_usd_implied = btc_usd * eth_btc
        eth_usd_direct = prices["ETH-USD"]

        # Account for 3-leg fees: (1-fee)^3
        fee_adj = (1 - FEE_RATE) ** 3
        profit_pct = (eth_usd_direct / eth_usd_implied - 1) * fee_adj * 100

        if abs(profit_pct) >= MIN_PROFIT_PCT:
            direction = "BTC→ETH→USD" if profit_pct > 0 else "USD→ETH→BTC"
            print(f"ARB SIGNAL: {direction} | {profit_pct:+.3f}% | Implied ETH/USD: ${eth_usd_implied:.2f} vs ${eth_usd_direct:.2f}")
            return {"direction": direction, "profit_pct": profit_pct, "size": 5.0}

        return None

async def arb_scanner_loop():
    while True:
        opp = await scan_triangular_arb()
        if opp:
            print(f"Executing arb: {opp['profit_pct']:+.3f}% expected profit")
            # Execute the arbitrage legs
        await asyncio.sleep(2)  # Scan every 2 seconds

Strategy 3: Statistical Arbitrage — BTC/ETH Pair Trade

BTC and ETH have a historical correlation of ~0.85. When the spread between their returns deviates significantly, it tends to revert. This is the basis of statistical arbitrage.

import numpy as np
from collections import deque

# Rolling window for spread statistics
LOOKBACK = 50          # 50 price observations
ENTRY_ZSCORE = 2.0    # Enter when z-score exceeds ±2.0
EXIT_ZSCORE = 0.5     # Exit when z-score reverts to ±0.5
POSITION_USD = 5.0    # $5 per leg

btc_prices = deque(maxlen=LOOKBACK)
eth_prices = deque(maxlen=LOOKBACK)
current_position = None  # "long_eth_short_btc" or "short_eth_long_btc"

def compute_spread_zscore() -> float:
    """Calculate z-score of current ETH/BTC spread vs historical."""
    if len(btc_prices) < LOOKBACK:
        return 0.0

    btc_arr = np.array(btc_prices)
    eth_arr = np.array(eth_prices)

    # Log returns spread
    btc_returns = np.log(btc_arr[1:] / btc_arr[:-1])
    eth_returns = np.log(eth_arr[1:] / eth_arr[:-1])

    spread = eth_returns - btc_returns
    zscore = (spread[-1] - spread.mean()) / (spread.std() + 1e-10)
    return zscore

def stat_arb_signal(btc_price: float, eth_price: float) -> str:
    """Returns trade signal based on current spread z-score."""
    global current_position

    btc_prices.append(btc_price)
    eth_prices.append(eth_price)

    zscore = compute_spread_zscore()
    print(f"BTC=${btc_price:,.0f} ETH=${eth_price:,.0f} Spread z-score: {zscore:+.2f}")

    # Entry signals
    if current_position is None:
        if zscore > ENTRY_ZSCORE:
            # ETH is overperforming → short ETH, long BTC
            return "short_eth_long_btc"
        elif zscore < -ENTRY_ZSCORE:
            # BTC is overperforming → short BTC, long ETH
            return "long_eth_short_btc"

    # Exit signals
    elif abs(zscore) < EXIT_ZSCORE:
        prev = current_position
        current_position = None
        return f"close_{prev}"

    return "hold"

def execute_pair_trade(signal: str):
    """Execute a pair trade based on signal."""
    global current_position

    if signal == "long_eth_short_btc":
        pf.post(f"{PF_BASE}/trading/perp/order", json={"symbol": "ETH-USD", "side": "long", "size_usd": POSITION_USD, "leverage": 1})
        pf.post(f"{PF_BASE}/trading/perp/order", json={"symbol": "BTC-USD", "side": "short", "size_usd": POSITION_USD, "leverage": 1})
        current_position = "long_eth_short_btc"
        print("  Entered: LONG ETH / SHORT BTC")

    elif signal == "short_eth_long_btc":
        pf.post(f"{PF_BASE}/trading/perp/order", json={"symbol": "ETH-USD", "side": "short", "size_usd": POSITION_USD, "leverage": 1})
        pf.post(f"{PF_BASE}/trading/perp/order", json={"symbol": "BTC-USD", "side": "long", "size_usd": POSITION_USD, "leverage": 1})
        current_position = "short_eth_long_btc"
        print("  Entered: SHORT ETH / LONG BTC")

    elif signal.startswith("close_"):
        pf.post(f"{PF_BASE}/trading/perp/close-all")
        print("  Closed pair trade position")

Profit Calculations and Realistic Expectations