The same token rarely trades at exactly the same price on two different blockchains at the same time. Liquidity pools have different depths, bridges introduce latency, and arbitrageurs are not everywhere at once. That gap — sometimes a fraction of a cent, sometimes several basis points — is where profit lives. Cross-chain arbitrage is the practice of systematically exploiting those differences, and AI agents are uniquely suited to doing it at scale.
In this guide you will build a complete automated arbitrage system using the Purple Flea Wallet API. We cover the mechanics of the opportunity, why AI agents outperform humans here, a full Python implementation, and an honest assessment of the risks involved.
What Is Cross-Chain Arbitrage?
Arbitrage at its core is buying an asset cheap in one market and selling it at a higher price in another. Cross-chain arbitrage applies this to decentralized exchanges (DEXs) and liquidity pools spread across different blockchain networks. Because each chain has its own AMMs, order books, and liquidity providers, prices diverge constantly.
Consider USDC. It is nominally pegged to $1.00, but the peg is maintained by markets, not by magic. On a given day the price spread can look something like this:
The spread between Arbitrum ($1.0005) and Base ($0.9996) is 0.09%. Buy USDC on Base, bridge it to Arbitrum, sell it there. If your total cost in gas and bridge fees is less than 0.09%, you pocket the difference. Do that hundreds of times a day with different token pairs and the returns compound into something meaningful.
This same mechanic applies to ETH, WBTC, liquid staking tokens (stETH, rETH), and any other asset that exists on multiple chains simultaneously. The key insight is that price convergence always happens eventually — the question is whether you are positioned to capture it before the market does.
Cross-chain arbitrage profits are not free money — they are compensation for providing a market efficiency service. You are narrowing price gaps between chains so that end users get fairer prices. The market rewards this with the spread.
Why AI Agents Are the Ideal Executor
A human trader monitoring six chains simultaneously, calculating net profit after dynamic gas costs, and executing within a three-second window before the opportunity closes is not realistic. AI agents handle every part of this problem natively.
- 24/7 continuous monitoring. Markets do not sleep. Cross-chain price dislocations are most common during high-volatility periods — often at unusual hours when human attention lapses. An agent never logs off.
- Millisecond reaction time. By the time a human spots an opportunity, opens a bridge, and submits a transaction, the spread has often collapsed. An AI agent can detect, calculate, and submit in under a second.
-
Zero emotional bias. Humans hesitate. They second-guess small spreads
and over-commit to larger ones. An agent executes the math exactly: if
net_profit > threshold, execute. Otherwise, wait. - Simultaneous multi-pair scanning. A single agent process can monitor dozens of token-chain combinations at once, far beyond any human's cognitive capacity.
- Dynamic fee adjustment. Gas prices fluctuate. An agent recalculates profitability in real time with live gas estimates and only executes when the math works.
- Consistent position sizing. Risk parameters are enforced programmatically — no overrides, no "just this once" large bets driven by overconfidence.
The Purple Flea Wallet API for Cross-Chain Operations
Building cross-chain infrastructure from scratch requires integrating RPC nodes, bridge protocols, DEX routers, and gas estimation logic across every chain you want to support. The Purple Flea Wallet API abstracts all of that. Your agent calls two endpoints and the platform handles routing.
?chain=arbitrum or
?chain=polygon to query any supported network. Use this to monitor
liquidity and current holdings across chains before executing trades.
from_chain, to_chain,
from_token, to_token, and amount. The API
automatically selects the optimal bridge and DEX route, returning a transaction hash
and estimated completion time.
Querying Balances Across Chains
# Check USDC balance on Arbitrum
curl "https://api.purpleflea.com/v1/wallet/balance?chain=arbitrum&token=USDC" \
-H "Authorization: Bearer $PF_API_KEY"
# Response
{
"chain": "arbitrum",
"token": "USDC",
"balance": "5000.00",
"usd_value": "5000.25",
"price_usd": "1.0005"
}
# Check same token on Base
curl "https://api.purpleflea.com/v1/wallet/balance?chain=base&token=USDC" \
-H "Authorization: Bearer $PF_API_KEY"
# Response
{
"chain": "base",
"token": "USDC",
"balance": "3200.00",
"usd_value": "3198.72",
"price_usd": "0.9996"
}Executing a Cross-Chain Swap
curl -X POST "https://api.purpleflea.com/v1/wallet/swap" \
-H "Authorization: Bearer $PF_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"from_chain": "base",
"to_chain": "arbitrum",
"from_token": "USDC",
"to_token": "USDC",
"amount": "1000.00",
"slippage": 0.005
}'
# Response
{
"tx_hash": "0xabc123...",
"estimated_output": "1000.07",
"estimated_fees": "0.42",
"bridge_used": "stargate",
"eta_seconds": 340,
"status": "pending"
}The API selects the best bridge automatically — weighing cost, speed, and reliability. Your agent receives a clean response with estimated output, fees already factored in, and an ETA in seconds. No bridge SDK integration required on your side.
Identifying Profitable Opportunities
An arbitrage opportunity exists only when the spread exceeds total costs. The decision logic is straightforward, but getting the inputs right is everything.
Step 1 — Monitor stablecoin prices across chains
Stablecoins are the best starting point because their "true" value is known ($1.00),
making spread calculation trivial. Poll /v1/wallet/balance across all
supported chains every 10-30 seconds for tokens like USDC, USDT, and DAI.
Step 2 — Calculate net profit after all fees
The naive spread is not your profit. You must subtract:
- Gas to initiate the bridge transaction on the source chain
- Bridge protocol fee (usually 0.01–0.04% of the transfer amount)
- Gas to receive and finalize on the destination chain
- DEX swap fee if selling into a pool (usually 0.01–0.3%)
- Slippage cost on both sides of the trade
Step 3 — Execute only above your minimum threshold
Set a minimum net profit threshold — commonly 0.1% for stablecoins — and only trigger execution when that bar is cleared. This protects against edge cases where gas spikes or slippage consume the spread mid-execution.
A 0.1% minimum threshold on a $5,000 position equals $5 per trade. After 200 successful trades in a month, that is $1,000 gross profit. Tighter thresholds increase trade frequency but raise the probability of unprofitable executions from gas variance.
Complete Python Implementation
The following implementation is a production-ready arbitrage agent. It includes a price scanner, profit calculator with live fee estimation, auto-executor triggered by threshold breaches, and a position tracker to avoid over-exposure on any single chain.
import asyncio
import httpx
import logging
from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Optional, Tuple
from decimal import Decimal, ROUND_DOWN
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s"
)
log = logging.getLogger("arb_agent")
# ─── Configuration ────────────────────────────────────────────
API_BASE = "https://api.purpleflea.com"
API_KEY = "YOUR_PURPLE_FLEA_API_KEY"
CHAINS = ["arbitrum", "polygon", "base", "bnb", "optimism"]
TOKENS = ["USDC", "USDT", "DAI"]
MIN_PROFIT_PCT = Decimal("0.001") # 0.1% minimum net profit
MAX_POSITION = Decimal("5000") # max USD per single trade
MIN_POSITION = Decimal("500") # skip tiny trades (gas inefficient)
MAX_CHAIN_ALLOC = Decimal("15000") # max exposure per chain
POLL_INTERVAL = 15 # seconds between price scans
SLIPPAGE = Decimal("0.005") # 0.5% slippage tolerance
# Bridge fee estimates per chain pair (in USD flat + bps of notional)
BRIDGE_FEES: Dict[str, Tuple[Decimal, Decimal]] = {
"default": (Decimal("0.30"), Decimal("0.0003")), # $0.30 + 0.03%
"arbitrum-base": (Decimal("0.12"), Decimal("0.0002")),
"base-arbitrum": (Decimal("0.12"), Decimal("0.0002")),
}
# ─── Data Models ──────────────────────────────────────────────
@dataclass
class PriceSnapshot:
chain: str
token: str
price: Decimal
balance: Decimal
usd_value: Decimal
fetched_at: datetime = field(default_factory=datetime.utcnow)
@dataclass
class ArbOpportunity:
buy_chain: str
sell_chain: str
token: str
buy_price: Decimal
sell_price: Decimal
raw_spread: Decimal
est_fees: Decimal
net_profit: Decimal
size: Decimal
net_profit_pct: Decimal
@dataclass
class Position:
opportunity: ArbOpportunity
tx_hash: str
opened_at: datetime
status: str = "pending" # pending | completed | failed
closed_at: Optional[datetime] = None
realized_pnl: Optional[Decimal] = None
# ─── Position Tracker ─────────────────────────────────────────
class PositionTracker:
def __init__(self):
self.open_positions: List[Position] = []
self.closed_positions: List[Position] = []
def chain_exposure(self, chain: str) -> Decimal:
"""Total USD locked in open trades touching this chain."""
total = Decimal("0")
for p in self.open_positions:
opp = p.opportunity
if opp.buy_chain == chain or opp.sell_chain == chain:
total += opp.size
return total
def can_trade(self, opp: ArbOpportunity) -> bool:
buy_exp = self.chain_exposure(opp.buy_chain)
sell_exp = self.chain_exposure(opp.sell_chain)
return (buy_exp + opp.size <= MAX_CHAIN_ALLOC and
sell_exp + opp.size <= MAX_CHAIN_ALLOC)
def add(self, pos: Position):
self.open_positions.append(pos)
def close(self, tx_hash: str, realized_pnl: Decimal):
for pos in self.open_positions:
if pos.tx_hash == tx_hash:
pos.status = "completed"
pos.closed_at = datetime.utcnow()
pos.realized_pnl = realized_pnl
self.open_positions.remove(pos)
self.closed_positions.append(pos)
break
def total_realized_pnl(self) -> Decimal:
return sum(
p.realized_pnl for p in self.closed_positions
if p.realized_pnl is not None
)
# ─── Price Scanner ────────────────────────────────────────────
class PriceScanner:
def __init__(self, client: httpx.AsyncClient):
self.client = client
async def fetch_price(self, chain: str, token: str) -> Optional[PriceSnapshot]:
try:
r = await self.client.get(
f"{API_BASE}/v1/wallet/balance",
params={"chain": chain, "token": token},
timeout=8.0
)
r.raise_for_status()
data = r.json()
return PriceSnapshot(
chain=chain,
token=token,
price=Decimal(str(data["price_usd"])),
balance=Decimal(str(data["balance"])),
usd_value=Decimal(str(data["usd_value"])),
)
except Exception as e:
log.warning(f"Price fetch failed [{chain}/{token}]: {e}")
return None
async def scan_all(self) -> List[PriceSnapshot]:
tasks = [
self.fetch_price(chain, token)
for chain in CHAINS
for token in TOKENS
]
results = await asyncio.gather(*tasks)
return [r for r in results if r is not None]
# ─── Profit Calculator ────────────────────────────────────────
def estimate_fees(buy_chain: str, sell_chain: str, size: Decimal) -> Decimal:
key = f"{buy_chain}-{sell_chain}"
flat, bps = BRIDGE_FEES.get(key, BRIDGE_FEES["default"])
bridge_fee = flat + (size * bps)
dex_fee = size * Decimal("0.0001") # 0.01% pool fee
gas_est = Decimal("0.50") # conservative gas estimate
return (bridge_fee + dex_fee + gas_est).quantize(Decimal("0.0001"))
def find_opportunities(snapshots: List[PriceSnapshot]) -> List[ArbOpportunity]:
opps = []
# Group by token
by_token: Dict[str, List[PriceSnapshot]] = {}
for s in snapshots:
by_token.setdefault(s.token, []).append(s)
for token, snaps in by_token.items():
for i, buy in enumerate(snaps):
for sell in snaps[i + 1:]:
if buy.price >= sell.price:
continue
# buy cheap, sell expensive
raw_spread = sell.price - buy.price
size = min(MAX_POSITION, buy.usd_value, sell.usd_value)
if size < MIN_POSITION:
continue
est_fees = estimate_fees(buy.chain, sell.chain, size)
gross = raw_spread * size / buy.price
net_profit = gross - est_fees
net_pct = net_profit / size
if net_pct >= MIN_PROFIT_PCT:
opps.append(ArbOpportunity(
buy_chain=buy.chain,
sell_chain=sell.chain,
token=token,
buy_price=buy.price,
sell_price=sell.price,
raw_spread=raw_spread,
est_fees=est_fees,
net_profit=net_profit.quantize(Decimal("0.01"), ROUND_DOWN),
size=size.quantize(Decimal("0.01"), ROUND_DOWN),
net_profit_pct=net_pct,
))
return sorted(opps, key=lambda o: o.net_profit_pct, reverse=True)
# ─── Trade Executor ───────────────────────────────────────────
class TradeExecutor:
def __init__(self, client: httpx.AsyncClient, tracker: PositionTracker):
self.client = client
self.tracker = tracker
async def execute(self, opp: ArbOpportunity) -> Optional[Position]:
if not self.tracker.can_trade(opp):
log.info(f"Skipping: chain exposure limit reached")
return None
log.info(
f"Executing: {opp.token} {opp.buy_chain}→{opp.sell_chain} "
f"size=${opp.size} net={opp.net_profit_pct:.4%}"
)
try:
r = await self.client.post(
f"{API_BASE}/v1/wallet/swap",
json={
"from_chain": opp.buy_chain,
"to_chain": opp.sell_chain,
"from_token": opp.token,
"to_token": opp.token,
"amount": str(opp.size),
"slippage": float(SLIPPAGE),
},
timeout=15.0
)
r.raise_for_status()
data = r.json()
tx_hash = data["tx_hash"]
pos = Position(
opportunity=opp,
tx_hash=tx_hash,
opened_at=datetime.utcnow(),
)
self.tracker.add(pos)
log.info(f"Order submitted: {tx_hash} ETA {data['eta_seconds']}s")
return pos
except Exception as e:
log.error(f"Execution failed: {e}")
return None
# ─── Main Agent Loop ──────────────────────────────────────────
async def main():
headers = {"Authorization": f"Bearer {API_KEY}"}
tracker = PositionTracker()
async with httpx.AsyncClient(headers=headers) as client:
scanner = PriceScanner(client)
executor = TradeExecutor(client, tracker)
log.info("Cross-chain arbitrage agent started")
log.info(f"Monitoring {len(CHAINS)} chains x {len(TOKENS)} tokens")
while True:
snapshots = await scanner.scan_all()
opps = find_opportunities(snapshots)
if opps:
log.info(f"Found {len(opps)} opportunities this cycle")
for opp in opps[:3]: # take top 3 per cycle max
await executor.execute(opp)
await asyncio.sleep(0.5)
else:
log.debug("No opportunities above threshold")
pnl = tracker.total_realized_pnl()
log.info(
f"Session PnL: ${pnl:.2f} | "
f"Open: {len(tracker.open_positions)} | "
f"Closed: {len(tracker.closed_positions)}"
)
await asyncio.sleep(POLL_INTERVAL)
if __name__ == "__main__":
asyncio.run(main())The agent runs a continuous scan every 15 seconds, building a matrix of prices across all chain-token combinations. For each unique token it computes every possible buy/sell pair, subtracts fees, and ranks opportunities by net profit percentage. Only the top opportunities above the 0.1% threshold trigger execution. Position exposure per chain is capped at $15,000 to prevent over-concentration.
Risk Factors to Model Carefully
Cross-chain arbitrage carries real risks. A profitable trade on paper can become a loss in execution. Build awareness of these factors into your risk management before going live.
| Risk | Severity | Description & Mitigation |
|---|---|---|
| Bridge delays | High | Cross-chain bridges typically take 5–20 minutes to finalize. During that window the price on the sell side can move against you. Mitigate by targeting stable pairs and maintaining a wider spread buffer on volatile market days. |
| Price slippage | Medium | Large trades move the pool price. A $5,000 USDC buy on a shallow pool may shift the price by 0.05% just from your own trade. Always model slippage at realistic pool depths, and set a slippage tolerance in your API calls. |
| Gas cost spikes | Medium | Gas prices can surge 3–10x in seconds during network congestion. A trade that was profitable at 20 gwei becomes a loss at 80 gwei. Use live gas estimates from the API, not static values, and add a gas buffer to your fee model. |
| Smart contract risk | Low | Bridge and DEX contracts have been audited but are not immune to exploits. Diversify across multiple bridge providers (the Purple Flea API rotates automatically) and avoid keeping large idle balances on any single chain. |
| Liquidity withdrawal | Low | LPs can remove liquidity mid-trade, collapsing the pool depth you relied on for your profitability estimate. Check pool liquidity before execution and set a minimum pool depth threshold. |
Realistic Return Expectations
Cross-chain arbitrage is not a guaranteed income stream. Returns depend heavily on market conditions, competition from other bots, and the chains you operate on. Here is a realistic benchmark framework:
On a calm market day with narrow spreads, your agent might find 5–10 qualifying opportunities. During volatile periods — a major protocol launch, a sudden liquidity migration, or a large whale rebalancing — the frequency can spike to 30–40 trades per day with larger spreads. Market inefficiencies cluster around events.
$20,000 capital, average 12 trades/day at $1,500 per trade, 0.1% net profit per trade = $18/day = ~$540/month gross. Subtract any infrastructure costs (VPS, API subscription). This is a modest but consistent return, and it compounds as you expand to more chain-token pairs.
Advanced: Triangular Cross-Chain Arbitrage
Once your agent is stable on two-leg arbitrage, the next frontier is triangular cross-chain arbitrage: routing through three chains (A → B → C → A) to capture compounding inefficiencies that a simple two-leg trade cannot access.
The logic works as follows. Suppose:
- USDC on Arbitrum is $1.0004 (expensive)
- USDT on Optimism is $0.9997 (cheap)
- DAI on Polygon has a USDC/DAI rate that favors buying DAI with USDC then selling for USDT
No single two-leg route captures all three inefficiencies. A triangular route does.
You extend the find_opportunities function to enumerate three-leg paths:
from itertools import permutations
def find_triangular_opportunities(
snapshots: List[PriceSnapshot]
) -> List[dict]:
"""
Find three-leg paths A→B→C→A that yield net positive return.
Returns paths sorted by estimated net profit descending.
"""
by_key = {(s.chain, s.token): s for s in snapshots}
paths = []
for chain_a, chain_b, chain_c in permutations(CHAINS, 3):
for token_a in TOKENS:
for token_b in TOKENS:
if token_b == token_a:
continue
snap_a = by_key.get((chain_a, token_a))
snap_b = by_key.get((chain_b, token_b))
snap_c = by_key.get((chain_c, token_a)) # return to token_a
if not (snap_a and snap_b and snap_c):
continue
# Leg 1: buy token_b on chain_b using token_a value
size = min(MAX_POSITION, snap_a.usd_value)
if size < MIN_POSITION:
continue
fees_ab = estimate_fees(chain_a, chain_b, size)
fees_bc = estimate_fees(chain_b, chain_c, size)
fees_ca = Decimal("0") # optional 3rd bridge back
# Gross: price difference across the three legs
ratio_ab = snap_b.price / snap_a.price
ratio_bc = snap_c.price / snap_b.price
gross = size * (ratio_ab * ratio_bc - Decimal("1"))
total_fees = fees_ab + fees_bc + fees_ca
net_profit = gross - total_fees
net_pct = net_profit / size
if net_pct >= MIN_PROFIT_PCT:
paths.append({
"path": [
f"{chain_a}/{token_a}",
f"{chain_b}/{token_b}",
f"{chain_c}/{token_a}",
],
"size": size,
"net_profit": net_profit.quantize(Decimal("0.01")),
"net_pct": float(net_pct),
"total_fees": total_fees,
})
return sorted(paths, key=lambda p: p["net_pct"], reverse=True)Triangular paths are rarer but yield larger spreads when found. The tradeoff is complexity: three legs means three bridge windows, three sets of gas, and three points where price can move against you. Reserve this strategy for established agents with a solid two-leg track record.
When integrating triangular opportunities into your main loop, execute legs sequentially and check mid-route profitability before committing to the next leg. If leg one's output is below the threshold needed to make legs two and three profitable, abort and log the miss rather than completing a losing sequence.
Start Capturing Cross-Chain Spreads
The Purple Flea Wallet API gives your agent a single unified interface for balance queries and cross-chain swaps across all major EVM networks. No bridge integrations, no RPC juggling.
Get Your API Key Read the Docs