Build a Profitable
Arbitrage Bot

import requests, time, logging from dataclasses import dataclass from typing import Optional logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") log = logging.getLogger("funding_arb") TRADING_API = "https://trading.purpleflea.com" WALLET_API = "https://wallet.purpleflea.com" TRADING_KEY = "pf_trade_sk_..." WALLET_KEY = "pf_wallet_sk_..." OPEN_THRESHOLD = 0.0005 # 0.05% per 8h - enter position CLOSE_THRESHOLD = 0.0001 # 0.01% per 8h - exit position MAX_ALLOC_PCT = 0.20 # max 20% of portfolio in any single trade CHECK_INTERVAL = 28800 # 8 hours in seconds TOP_N_MARKETS = 10 # scan top N markets by funding rate @dataclass class Position: coin: str size_usd: float entry_funding_rate: float perp_position_id: str spot_amount: float total_funding_earned: float = 0.0 def trading_get(path: str, params: dict = None) -> dict: r = requests.get(f"{TRADING_API}{path}", params=params, headers={"Authorization": f"Bearer {TRADING_KEY}"}, timeout=10) r.raise_for_status() return r.json() def trading_post(path: str, body: dict) -> dict: r = requests.post(f"{TRADING_API}{path}", json=body, headers={"Authorization": f"Bearer {TRADING_KEY}"}, timeout=10) r.raise_for_status() return r.json() def wallet_post(path: str, body: dict) -> dict: r = requests.post(f"{WALLET_API}{path}", json=body, headers={"Authorization": f"Bearer {WALLET_KEY}"}, timeout=15) r.raise_for_status() return r.json() def get_portfolio_value() -> float: # GET /v1/wallet/balance - returns total_usd across all chains data = requests.get(f"{WALLET_API}/v1/wallet/balance", headers={"Authorization": f"Bearer {WALLET_KEY}"}).json() return data["total_usd"] def get_top_funding_markets() -> list[dict]: # GET /v1/markets - returns coins with funding_rate and mark_price markets = trading_get("/v1/markets") positive = [m for m in markets if m["funding_rate"] > OPEN_THRESHOLD] return sorted(positive, key=lambda x: x["funding_rate"], reverse=True)[:TOP_N_MARKETS] def open_funding_arb(coin: str, funding_rate: float, mark_price: float) -> Optional[Position]: portfolio = get_portfolio_value() size_usd = min(portfolio * MAX_ALLOC_PCT, 10000) coin_amt = size_usd / mark_price log.info(f"Opening arb: {coin} rate={funding_rate:.4%} size=${size_usd:.0f}") # Leg 1: buy spot via Wallet API - POST /v1/wallet/swap spot = wallet_post("/v1/wallet/swap", { "from_token": "USDC", "to_token": coin, "amount_usd": size_usd, "max_slippage_pct": 0.3 }) # Leg 2: short equal notional on perp - POST /v1/trade/open perp = trading_post("/v1/trade/open", { "coin": coin, "side": "sell", "size_usd": size_usd, "leverage": 1, "order_type": "market" }) log.info(f"Arb open: spot tx={spot['tx_hash'][:12]}... perp id={perp['position_id']}") return Position(coin=coin, size_usd=size_usd, entry_funding_rate=funding_rate, perp_position_id=perp["position_id"], spot_amount=coin_amt) def close_funding_arb(pos: Position): log.info(f"Closing arb: {pos.coin} total_earned=${pos.total_funding_earned:.2f}") # POST /v1/trade/close trading_post("/v1/trade/close", {"position_id": pos.perp_position_id}) # POST /v1/wallet/swap - sell spot back to USDC wallet_post("/v1/wallet/swap", { "from_token": pos.coin, "to_token": "USDC", "amount_tokens": pos.spot_amount, "max_slippage_pct": 0.3 }) def run_funding_bot(): active: dict[str, Position] = {} while True: try: markets = trading_get("/v1/markets") rates = {m["coin"]: m for m in markets} # Check existing positions - close if rate dropped below threshold for coin, pos in list(active.items()): rate = rates.get(coin, {}).get("funding_rate", 0) earned = pos.size_usd * rate pos.total_funding_earned += earned log.info(f"{coin} rate={rate:.4%} earned=${earned:.2f}") if rate < CLOSE_THRESHOLD: close_funding_arb(pos) del active[coin] # Open new positions for high-rate markets not already held for m in get_top_funding_markets(): if m["coin"] not in active: pos = open_funding_arb(m["coin"], m["funding_rate"], m["mark_price"]) if pos: active[m["coin"]] = pos except Exception as e: log.error(f"Bot error: {e}") time.sleep(CHECK_INTERVAL) if __name__ == "__main__": run_funding_bot()

import asyncio, aiohttp, logging from itertools import combinations WALLET_API = "https://wallet.purpleflea.com" WALLET_KEY = "pf_wallet_sk_..." HEADERS = {"Authorization": f"Bearer {WALLET_KEY}"} MIN_SPREAD = 0.30 # minimum net spread % to execute TRADE_USD = 500 # USDC per trade leg CHAINS = ["ethereum", "arbitrum", "optimism", "polygon", "base"] TOKENS = ["ETH", "WBTC", "LINK", "UNI"] GAS_COST_USD = { "ethereum": 12.0, "arbitrum": 0.30, "optimism": 0.25, "polygon": 0.05, "base": 0.20 } async def get_quote(session: aiohttp.ClientSession, token: str, chain: str) -> float: # GET /v1/wallet/address with quote params - tokens received per $1 USDC async with session.get( f"{WALLET_API}/v1/wallet/address", params={"from": "USDC", "to": token, "amount_usd": TRADE_USD, "chain": chain}, headers=HEADERS, timeout=aiohttp.ClientTimeout(total=5) ) as r: d = await r.json() return d["out_amount"] / TRADE_USD async def scan_token(session: aiohttp.ClientSession, token: str) -> list: tasks = [get_quote(session, token, chain) for chain in CHAINS] prices = await asyncio.gather(*tasks, return_exceptions=True) return [ {"token": token, "chain": chain, "rate": price} for chain, price in zip(CHAINS, prices) if isinstance(price, float) ] async def find_opportunities() -> list[dict]: opportunities = [] async with aiohttp.ClientSession() as session: for token in TOKENS: quotes = await scan_token(session, token) for q1, q2 in combinations(quotes, 2): cheap, dear = sorted([q1, q2], key=lambda x: x["rate"], reverse=True) spread_pct = (cheap["rate"] - dear["rate"]) / dear["rate"] * 100 gas_total = GAS_COST_USD[cheap["chain"]] + GAS_COST_USD[dear["chain"]] gas_pct = gas_total / TRADE_USD * 100 net_spread = spread_pct - gas_pct - 0.1 if net_spread >= MIN_SPREAD: opportunities.append({ "token": token, "buy_chain": cheap["chain"], "sell_chain": dear["chain"], "gross_spread_pct": round(spread_pct, 3), "net_spread_pct": round(net_spread, 3), "expected_profit_usd":round(TRADE_USD * net_spread / 100, 2), }) return sorted(opportunities, key=lambda x: x["net_spread_pct"], reverse=True) async def execute_cross_chain_arb(opp: dict): async with aiohttp.ClientSession() as session: # Buy on cheap chain - POST /v1/wallet/swap buy = await session.post(f"{WALLET_API}/v1/wallet/swap", headers=HEADERS, json={ "from_token": "USDC", "to_token": opp["token"], "amount_usd": TRADE_USD, "chain": opp["buy_chain"], "max_slippage_pct": 0.5 }) buy_data = await buy.json() # Sell on expensive chain (pre-positioned token balance) sell = await session.post(f"{WALLET_API}/v1/wallet/swap", headers=HEADERS, json={ "from_token": opp["token"], "to_token": "USDC", "amount_usd": TRADE_USD, "chain": opp["sell_chain"], "max_slippage_pct": 0.5 }) logging.info(f"Executed {opp['token']}: " f"buy={opp['buy_chain']} sell={opp['sell_chain']} " f"net={opp['net_spread_pct']}%") if __name__ == "__main__": opps = asyncio.run(find_opportunities()) for o in opps: print(f"{o['token']:6} {o['buy_chain']:10} -> {o['sell_chain']:10} " f"net={o['net_spread_pct']:.2f}% profit=${o['expected_profit_usd']}")

import requests, math, time, logging from collections import deque from statistics import mean, stdev TRADING_API = "https://trading.purpleflea.com" TRADING_KEY = "pf_trade_sk_..." HEADERS = {"Authorization": f"Bearer {TRADING_KEY}"} WINDOW = 720 # 30 days of hourly candles ENTRY_Z = 2.0 # z-score threshold to open position EXIT_Z = 0.5 # z-score threshold to close position STOP_Z = 3.5 # emergency stop if spread keeps widening SIZE_USD = 5000 # notional per leg ($5k BTC + $5k ETH) spread_history: deque = deque(maxlen=WINDOW) current_position: dict = {} def get_mark_prices() -> tuple[float, float]: # GET /v1/markets - retrieve BTC and ETH mark_price markets = requests.get(f"{TRADING_API}/v1/markets", headers=HEADERS).json() prices = {m["coin"]: m["mark_price"] for m in markets} return prices["BTC"], prices["ETH"] def compute_z_score(btc: float, eth: float) -> float: spread = math.log(btc) - math.log(eth) spread_history.append(spread) if len(spread_history) < 60: return 0.0 # not enough history yet mu = mean(spread_history) sig = stdev(spread_history) return (spread - mu) / sig if sig > 0 else 0.0 def open_position(side: str) -> dict: # side = 'long_btc' means buy BTC perp + sell ETH perp btc_side = "buy" if side == "long_btc" else "sell" eth_side = "sell" if side == "long_btc" else "buy" btc_pos = requests.post(f"{TRADING_API}/v1/trade/open", headers=HEADERS, json={"coin": "BTC", "side": btc_side, "size_usd": SIZE_USD, "leverage": 1, "order_type": "market"}).json() eth_pos = requests.post(f"{TRADING_API}/v1/trade/open", headers=HEADERS, json={"coin": "ETH", "side": eth_side, "size_usd": SIZE_USD, "leverage": 1, "order_type": "market"}).json() logging.info(f"Opened {side}: BTC-{btc_side} ETH-{eth_side} ${SIZE_USD} each") return {"side": side, "btc_id": btc_pos["position_id"], "eth_id": eth_pos["position_id"]} def close_position(pos: dict): for pid in [pos["btc_id"], pos["eth_id"]]: requests.post(f"{TRADING_API}/v1/trade/close", headers=HEADERS, json={"position_id": pid}) logging.info(f"Closed position: {pos['side']}") def run_stat_arb(): global current_position while True: try: btc, eth = get_mark_prices() z = compute_z_score(btc, eth) logging.info(f"BTC={btc:.0f} ETH={eth:.2f} z={z:.3f}") if current_position: side = current_position["side"] reverted = (abs(z) < EXIT_Z) blown_out = (abs(z) > STOP_Z) crossed = ((side == "long_btc" and z > 0) or (side == "long_eth" and z < 0)) if reverted or blown_out or crossed: close_position(current_position) current_position = {} else: if z < -ENTRY_Z: current_position = open_position("long_btc") elif z > ENTRY_Z: current_position = open_position("long_eth") except Exception as e: logging.error(f"Stat arb error: {e}") time.sleep(3600) if __name__ == "__main__": run_stat_arb()

import requests, time, math, logging from collections import deque from statistics import mean, stdev from dataclasses import dataclass, field from typing import Optional logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(name)s] %(levelname)s %(message)s") TRADING_API = "https://trading.purpleflea.com" WALLET_API = "https://wallet.purpleflea.com" TRADING_KEY = "pf_trade_sk_..." WALLET_KEY = "pf_wallet_sk_..." MAX_ALLOC = 0.20 # max 20% of portfolio per trade FUNDING_OPEN = 0.0005 # 0.05%/8h threshold to open FUNDING_CLOSE= 0.0001 # 0.01%/8h threshold to close STAT_ENTRY_Z = 2.0 STAT_EXIT_Z = 0.5 STAT_STOP_Z = 3.5 @dataclass class BotState: funding_positions: dict = field(default_factory=dict) stat_position: Optional[dict] = None spread_history: deque = field(default_factory=lambda: deque(maxlen=720)) total_earned: float = 0.0 trade_count: int = 0 class PFClient: def __init__(self): self.t = {"Authorization": f"Bearer {TRADING_KEY}"} self.w = {"Authorization": f"Bearer {WALLET_KEY}"} def portfolio_value(self) -> float: # GET /v1/wallet/balance return requests.get(f"{WALLET_API}/v1/wallet/balance", headers=self.w, timeout=8).json()["total_usd"] def markets(self) -> list: # GET /v1/markets - returns coins with funding_rate, mark_price return requests.get(f"{TRADING_API}/v1/markets", headers=self.t, timeout=8).json() def open_trade(self, coin, side, size_usd, leverage=1) -> dict: # POST /v1/trade/open return requests.post(f"{TRADING_API}/v1/trade/open", headers=self.t, json={"coin": coin, "side": side, "size_usd": size_usd, "leverage": leverage, "order_type": "market"}).json() def close_trade(self, position_id) -> dict: # POST /v1/trade/close return requests.post(f"{TRADING_API}/v1/trade/close", headers=self.t, json={"position_id": position_id}).json() def swap(self, from_t, to_t, amount_usd, chain) -> dict: # POST /v1/wallet/swap return requests.post(f"{WALLET_API}/v1/wallet/swap", headers=self.w, json={"from_token": from_t, "to_token": to_t, "amount_usd": amount_usd, "chain": chain, "max_slippage_pct": 0.3}).json() api = PFClient() state = BotState() log = logging.getLogger("orchestrator") def run_funding_cycle(): markets = api.markets() portfolio = api.portfolio_value() rates = {m["coin"]: m for m in markets} for coin, pos in list(state.funding_positions.items()): rate = rates.get(coin, {}).get("funding_rate", 0) earned = pos["size_usd"] * rate state.total_earned += earned log.info(f"[funding] {coin} rate={rate:.4%} earned=${earned:.2f} total=${state.total_earned:.2f}") if rate < FUNDING_CLOSE: api.close_trade(pos["perp_id"]) api.swap(coin, "USDC", pos["size_usd"], "ethereum") del state.funding_positions[coin] log.info(f"[funding] Closed {coin} - rate below threshold") top = sorted([m for m in markets if m["funding_rate"] > FUNDING_OPEN], key=lambda x: x["funding_rate"], reverse=True)[:10] for m in top: if m["coin"] in state.funding_positions: continue size = min(portfolio * MAX_ALLOC, 10000) api.swap("USDC", m["coin"], size, "ethereum") perp = api.open_trade(m["coin"], "sell", size) state.funding_positions[m["coin"]] = { "size_usd": size, "perp_id": perp["position_id"]} state.trade_count += 1 log.info(f"[funding] Opened {m['coin']} rate={m['funding_rate']:.4%} size=${size:.0f}") def run_stat_cycle(): markets = api.markets() prices = {m["coin"]: m["mark_price"] for m in markets} btc, eth = prices["BTC"], prices["ETH"] spread = math.log(btc) - math.log(eth) state.spread_history.append(spread) if len(state.spread_history) < 60: return z = ((spread - mean(state.spread_history)) / max(stdev(state.spread_history), 1e-9)) log.info(f"[stat] BTC={btc:.0f} ETH={eth:.2f} z={z:.3f}") if state.stat_position: if abs(z) < STAT_EXIT_Z or abs(z) > STAT_STOP_Z: for pid in [state.stat_position["btc_id"], state.stat_position["eth_id"]]: api.close_trade(pid) log.info(f"[stat] Closed at z={z:.3f}") state.stat_position = None elif abs(z) > STAT_ENTRY_Z: portfolio = api.portfolio_value() size = min(portfolio * MAX_ALLOC, 5000) btc_side = "buy" if z < 0 else "sell" eth_side = "sell" if z < 0 else "buy" bp = api.open_trade("BTC", btc_side, size) ep = api.open_trade("ETH", eth_side, size) state.stat_position = {"btc_id": bp["position_id"], "eth_id": ep["position_id"]} state.trade_count += 1 log.info(f"[stat] Opened z={z:.3f} BTC-{btc_side} ETH-{eth_side} ${size:.0f}") def main(): log.info("Arbitrage orchestrator started") cycle = 0 while True: try: run_stat_cycle() # runs every cycle (hourly) if cycle % 8 == 0: # funding check every 8 hours run_funding_cycle() except Exception as e: log.error(f"Cycle error: {e}") cycle += 1 time.sleep(3600) if __name__ == "__main__": main()

import csv, math from dataclasses import dataclass # Load historical funding data: each row = {coin, timestamp, rate, mark_price} # Source: GET /v1/markets?history=true&days=90 (export to CSV) @dataclass class BacktestResult: total_pnl: float num_trades: int win_rate: float max_drawdown: float sharpe_ratio: float def backtest_funding_arb( data_path: str, open_threshold: float = 0.0005, close_threshold: float = 0.0001, size_usd: float = 10000, fee_pct: float = 0.0002 # 0.02% per leg ) -> BacktestResult: pnl_series, positions, trades = [], {}, [] with open(data_path) as f: rows = list(csv.DictReader(f)) for row in rows: coin = row["coin"] rate = float(row["funding_rate"]) price = float(row["mark_price"]) if coin in positions: period_pnl = size_usd * rate pnl_series.append(period_pnl) positions[coin]["cumulative"] += period_pnl if rate < close_threshold: entry_cost = size_usd * fee_pct * 2 net = positions[coin]["cumulative"] - entry_cost trades.append(net) del positions[coin] elif rate > open_threshold: positions[coin] = {"size_usd": size_usd, "cumulative": 0.0} total_pnl = sum(trades) win_rate = sum(1 for t in trades if t > 0) / max(len(trades), 1) # Max drawdown calculation cum = 0; peak = 0; max_dd = 0 for p in pnl_series: cum += p; peak = max(peak, cum) max_dd = max(max_dd, peak - cum) # Annualised Sharpe ratio (3 periods/day) n = len(pnl_series) mu = sum(pnl_series) / max(n, 1) var = sum((p - mu) ** 2 for p in pnl_series) / max(n, 1) sharpe = (mu / math.sqrt(var)) * math.sqrt(3 * 365) if var > 0 else 0 return BacktestResult( total_pnl=round(total_pnl, 2), num_trades=len(trades), win_rate=round(win_rate, 3), max_drawdown=round(max_dd, 2), sharpe_ratio=round(sharpe, 2) ) result = backtest_funding_arb("funding_history.csv") print(f"90-day P&L: ${result.total_pnl:,.2f}") print(f"Trades: {result.num_trades}") print(f"Win rate: {result.win_rate:.1%}") print(f"Max drawdown: ${result.max_drawdown:.2f}") print(f"Sharpe ratio: {result.sharpe_ratio:.2f}")