Five arbitrage strategies. Complete Python implementation. Low-latency order execution via Purple Flea's Trading API. Your bot, your alpha, your profits.
Purple Flea's Trading API gives you the execution layer for every major arb strategy. Pick one or combine multiple for a diversified alpha portfolio.
Exploit price discrepancies between centralized exchanges (CEX) and decentralized venues (DEX). Monitor price feeds from multiple sources, detect spread > threshold, execute on the cheaper side and sell on the premium side simultaneously. Purple Flea provides the CEX-side execution; connect to a DEX via your agent's on-chain wallet.
Trade three correlated pairs in sequence to exploit pricing inconsistencies within a single venue. Example: BTC → ETH → USDC → BTC, where the round-trip yields more than you started with. Purple Flea's Trading API supports rapid sequential order placement for all three legs.
Identify mean-reverting pairs (e.g., BTC-PERP vs ETH-PERP) using cointegration tests. When the spread diverges beyond 2 standard deviations, open opposing positions. Close when the spread reverts to mean. Purple Flea provides real-time orderbook data for spread calculation.
When perpetual futures funding rates are extremely positive, short the perp and long spot. Collect the funding payment every 8 hours while staying delta-neutral. Purple Flea's perp trading API provides real-time funding rate data and one-click position management.
The same token often trades at different prices across Ethereum, Arbitrum, Base, and Solana. Use Purple Flea's wallet API to hold balances on multiple chains. Bridge assets via programmatic bridging APIs and capture the inter-chain spread before it closes.
Every endpoint your arb bot needs: real-time orderbooks, fast market orders, portfolio tracking, and position management — all REST-based, no WebSocket required to start.
Drop this into your project. The class handles opportunity detection, multi-leg execution, P&L tracking, and graceful shutdown. Extend the detect_opportunity method with your own signal logic.
"""
arbitrage_bot.py — Purple Flea ArbitrageBot
Complete implementation: opportunity detection, execution, P&L tracking.
Author: your-agent-id
Purple Flea Trading API: https://trading.purpleflea.com
"""
import asyncio, httpx, time, logging
from dataclasses import dataclass, field
from typing import Optional, List
from datetime import datetime
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
log = logging.getLogger("arb_bot")
PF_BASE = "https://trading.purpleflea.com/api/v1"
@dataclass
class ArbOpportunity:
strategy: str # "cex_dex" | "triangular" | "stat_arb" | "funding"
symbol_a: str
symbol_b: str
spread_pct: float # e.g. 0.45 means 0.45%
side_a: str # "buy" or "sell"
side_b: str
size_usdc: float
confidence: float # 0.0 – 1.0
detected_at: float = field(default_factory=time.time)
@dataclass
class TradeRecord:
opportunity: ArbOpportunity
order_id_a: Optional[str] = None
order_id_b: Optional[str] = None
fill_price_a: float = 0.0
fill_price_b: float = 0.0
realized_pnl: float = 0.0
fees_paid: float = 0.0
status: str = "pending" # pending | partial | complete | failed
executed_at: Optional[float] = None
class ArbitrageBot:
"""
Purple Flea ArbitrageBot — multi-strategy, async, production-ready.
Usage:
bot = ArbitrageBot(api_key="pf_live_xxxx", max_position_usdc=500)
await bot.run()
"""
MAKER_FEE = 0.0005 # 0.05%
TAKER_FEE = 0.0010 # 0.10%
MIN_SPREAD = 0.0025 # 0.25% minimum net spread after fees
POLL_INTERVAL = 1.5 # seconds between opportunity scans
def __init__(
self,
api_key: str,
max_position_usdc: float = 200,
max_drawdown_pct: float = 0.10,
pairs: List[str] = None
):
self.api_key = api_key
self.max_position_usdc = max_position_usdc
self.max_drawdown_pct = max_drawdown_pct
self.pairs = pairs or ["BTC-PERP", "ETH-PERP", "SOL-PERP"]
self.headers = {"X-Api-Key": api_key}
# P&L tracking
self.trades: List[TradeRecord] = []
self.total_pnl: float = 0.0
self.total_fees: float = 0.0
self.starting_balance: float = 0.0
self.running: bool = False
async def _get(self, path: str, **params) -> dict:
async with httpx.AsyncClient(timeout=8) as client:
r = await client.get(f"{PF_BASE}{path}", headers=self.headers, params=params)
r.raise_for_status()
return r.json()
async def _post(self, path: str, payload: dict) -> dict:
async with httpx.AsyncClient(timeout=8) as client:
r = await client.post(f"{PF_BASE}{path}", headers=self.headers, json=payload)
r.raise_for_status()
return r.json()
async def get_portfolio(self) -> dict:
"""Fetch current USDC balance and margin state."""
return await self._get("/portfolio")
async def get_orderbook(self, symbol: str) -> dict:
"""Full L2 orderbook snapshot for a symbol."""
return await self._get(f"/orderbook/{symbol}")
async def get_ticker(self, symbol: str) -> dict:
"""Best bid/ask + last price for a symbol."""
return await self._get(f"/ticker/{symbol}")
async def get_funding_rate(self, symbol: str) -> float:
"""Current 8-hour funding rate for a perpetual."""
data = await self._get(f"/funding/{symbol}")
return data.get("current_rate", 0.0)
async def place_order(
self,
symbol: str,
side: str,
size_usdc: float,
order_type: str = "market",
leverage: float = 1.0,
limit_price: Optional[float] = None
) -> dict:
"""Place a market or limit order. Returns order dict with order_id."""
payload = {
"symbol": symbol,
"side": side,
"size_usdc": size_usdc,
"leverage": leverage,
"order_type": order_type
}
if limit_price:
payload["limit_price"] = limit_price
return await self._post("/orders", payload)
async def detect_opportunity(self) -> Optional[ArbOpportunity]:
"""
Scan configured pairs for arbitrage opportunities.
Strategy: compare funding rates across pairs; if one pair has extreme
positive funding (> 0.10%), short it and long the correlated pair.
Extend this method with your own signal logic.
"""
tasks = [self.get_ticker(pair) for pair in self.pairs]
tickers = await asyncio.gather(*tasks, return_exceptions=True)
best_spread = 0.0
opportunity = None
for i, t_a in enumerate(tickers):
if isinstance(t_a, Exception): continue
for j, t_b in enumerate(tickers):
if i == j or isinstance(t_b, Exception): continue
# Funding rate spread between two pairs
rate_a = t_a.get("funding_rate", 0.0)
rate_b = t_b.get("funding_rate", 0.0)
spread = abs(rate_a - rate_b)
if spread > self.MIN_SPREAD and spread > best_spread:
best_spread = spread
opportunity = ArbOpportunity(
strategy="funding",
symbol_a=self.pairs[i],
symbol_b=self.pairs[j],
spread_pct=spread * 100,
side_a="sell" if rate_a > rate_b else "buy",
side_b="buy" if rate_a > rate_b else "sell",
size_usdc=min(100, self.max_position_usdc / 2),
confidence=min(1.0, spread / 0.005)
)
return opportunity
async def execute_opportunity(self, opp: ArbOpportunity) -> TradeRecord:
"""
Execute a two-leg arbitrage opportunity.
Places both legs as close to simultaneously as possible using asyncio.gather.
"""
record = TradeRecord(opportunity=opp)
log.info(f"Executing {opp.strategy} arb: {opp.symbol_a}/{opp.symbol_b} spread={opp.spread_pct:.3f}%")
try:
order_a, order_b = await asyncio.gather(
self.place_order(opp.symbol_a, opp.side_a, opp.size_usdc),
self.place_order(opp.symbol_b, opp.side_b, opp.size_usdc),
)
record.order_id_a = order_a.get("order_id")
record.order_id_b = order_b.get("order_id")
record.fill_price_a = order_a.get("fill_price", 0)
record.fill_price_b = order_b.get("fill_price", 0)
record.executed_at = time.time()
# Estimate net P&L after fees
gross_pnl = opp.size_usdc * (opp.spread_pct / 100)
fees = opp.size_usdc * 2 * self.TAKER_FEE # both legs, taker
record.realized_pnl = gross_pnl - fees
record.fees_paid = fees
record.status = "complete"
self.total_pnl += record.realized_pnl
self.total_fees += fees
log.info(f"Arb complete: net PnL=${record.realized_pnl:.4f} fees=${fees:.4f}")
except Exception as e:
record.status = "failed"
log.error(f"Arb execution failed: {e}")
self.trades.append(record)
return record
async def check_drawdown(self) -> bool:
"""Return True if within drawdown limits, False if bot should pause."""
if self.starting_balance == 0:
return True
portfolio = await self.get_portfolio()
current = portfolio.get("usdc_balance", self.starting_balance)
drawdown = (self.starting_balance - current) / self.starting_balance
if drawdown > self.max_drawdown_pct:
log.warning(f"Drawdown limit hit: {drawdown:.1%} > {self.max_drawdown_pct:.1%}. Pausing.")
return False
return True
def print_summary(self):
"""Print a P&L summary to stdout."""
wins = [t for t in self.trades if t.realized_pnl > 0]
losses = [t for t in self.trades if t.realized_pnl <= 0]
print(f"\n=== Arb Bot Summary ===")
print(f"Total trades : {len(self.trades)}")
print(f"Wins / Losses: {len(wins)} / {len(losses)}")
print(f"Total PnL : ${self.total_pnl:+.4f} USDC")
print(f"Total fees : ${self.total_fees:.4f} USDC")
async def run(self):
"""Main event loop. Runs until KeyboardInterrupt or drawdown limit."""
portfolio = await self.get_portfolio()
self.starting_balance = portfolio.get("usdc_balance", 0)
log.info(f"ArbitrageBot started. Balance: ${self.starting_balance:.2f} USDC")
self.running = True
try:
while self.running:
ok = await self.check_drawdown()
if not ok:
break
opp = await self.detect_opportunity()
if opp and opp.confidence > 0.6:
await self.execute_opportunity(opp)
else:
log.debug("No opportunity found. Waiting.")
await asyncio.sleep(self.POLL_INTERVAL)
except KeyboardInterrupt:
log.info("Shutdown requested.")
finally:
self.running = False
self.print_summary()
# Entry point
if __name__ == "__main__":
import os
bot = ArbitrageBot(
api_key=os.environ["PF_API_KEY"],
max_position_usdc=250,
max_drawdown_pct=0.08,
pairs=["BTC-PERP", "ETH-PERP", "SOL-PERP"]
)
asyncio.run(bot.run())Successful arbitrage is a latency and cost game. Here's how to tune each parameter for maximum net yield on Purple Flea's infrastructure.
Arbitrage feels safe until it isn't. The biggest risks are execution failure, slippage, and correlated drawdowns. Build these controls in from day one.
Cap total open notional across all pairs. If the sum of all open positions exceeds your configured limit, skip new opportunities until positions are closed. Prevents over-leveraging during volatile markets.
Monitor starting equity vs current equity after every cycle.
If the drawdown exceeds your threshold, halt the bot and alert.
The ArbitrageBot class implements this in check_drawdown().
An arb opportunity detected 500ms ago may already be gone.
Always check the detected_at timestamp before placing orders.
If the opportunity is older than your staleness threshold, skip it.
BTC-PERP and ETH-PERP are highly correlated. Holding opposing positions in correlated pairs does not always produce the expected hedge. Run a rolling 24h correlation check before opening stat-arb positions.
If leg A fills but leg B fails, you now have an unintended directional position.
Immediately attempt to cancel or close leg A.
The bot logs status="partial" so you can review and close manually.
Before every order, verify the gross spread exceeds total fees. Purple Flea taker fee is 0.10% per side — a two-leg arb costs 0.20% minimum. Your spread must exceed 0.25% for a safe margin of profit.
From zero to running arb bot in under 10 minutes. All steps use only the Purple Flea REST API — no SDK needed.
POST to https://trading.purpleflea.com/api/v1/register. Provide agent_id (any unique string) and receive your api_key instantly. Store it as an environment variable.
New agents can claim $1 USDC at https://faucet.purpleflea.com/api/claim. It's enough to run a few small arb cycles and verify your strategy logic works before depositing real capital.
Run pip install httpx asyncio — no special SDK required. Set PF_API_KEY as an environment variable. Adjust max_position_usdc and pairs to match your risk tolerance and target markets.
Deploy on a server geographically close to Purple Flea's infrastructure (EU-West). Use python arbitrage_bot.py or wrap it with pm2 / systemd for automatic restarts and log management.
The bot prints a full P&L summary on shutdown. Review win rate, average spread captured, and total fees paid. Tune MIN_SPREAD and POLL_INTERVAL based on real performance data.
# 1. Register agent, get API key
curl -s -X POST https://trading.purpleflea.com/api/v1/register \
-H 'Content-Type: application/json' \
-d '{"agent_id": "my-arb-bot-001"}' | jq .api_key
# 2. Claim faucet USDC (new agents only)
curl -s -X POST https://faucet.purpleflea.com/api/claim \
-H 'Content-Type: application/json' \
-d '{"agent_id": "my-arb-bot-001"}' | jq .
# 3. Check starting balance
curl -s https://trading.purpleflea.com/api/v1/portfolio \
-H "X-Api-Key: $PF_API_KEY" | jq .usdc_balance
# 4. Scan current ticker (verify API connection)
curl -s https://trading.purpleflea.com/api/v1/ticker/BTC-PERP \
-H "X-Api-Key: $PF_API_KEY" | jq .
# 5. Deploy bot with pm2 (persistent, auto-restart)
pip install httpx
export PF_API_KEY="your-api-key-here"
pm2 start "python arbitrage_bot.py" --name arb-bot
pm2 logs arb-botRegister in 30 seconds. No KYC. Claim $1 free USDC. Start capturing spreads today.