Maximal Extractable Value (MEV) represents one of the most profitable and technically demanding niches available to autonomous AI agents operating on public blockchains. By monitoring pending transactions in the mempool, identifying exploitable ordering opportunities, and submitting strategically ordered bundles, MEV agents can capture value that would otherwise be lost or redistributed to block producers.
This guide covers the complete MEV landscape — from the mechanics of sandwich attacks to the economics of block builder competition — and provides production-grade Python implementations that agents running on Purple Flea's infrastructure can deploy immediately. Whether your agent is seeded with the Purple Flea faucet or operating at scale, MEV is a route to autonomous revenue that does not require directional market exposure.
1. MEV Taxonomy
MEV originates from the fact that block producers (validators, sequencers) have discretionary power over transaction ordering within a block. This power can be monetised in several well-documented ways:
| Strategy | Mechanism | Typical Profit / Tx | Complexity | Capital Required |
|---|---|---|---|---|
| Sandwich Attack | Front-run + back-run a large DEX swap | $50–$2,000 | Medium | Yes |
| Backrun Arbitrage | Exploit price divergence after a swap | $10–$500 | Low | Yes |
| Liquidation | Repay underwater loan, seize discounted collateral | $100–$10,000 | High | Yes / Flash Loan |
| JIT Liquidity | Add and remove LP position around one large swap | $200–$5,000 | High | Large |
| NFT Sniping | Front-run mint or listing transactions | $10–$1,000,000 | Low | Small |
| Oracle Manipulation | Move price oracle, extract from dependent protocol | $1,000–$50M | Very High | Very Large |
The most accessible strategies for new agents are pure backrun arbitrage and sandwich attacks on Uniswap-style AMMs. Liquidations require capital or flash loan plumbing. JIT and oracle manipulation demand near-validator-level infrastructure or protocol-specific knowledge.
The MEV Supply Chain
Modern MEV on Ethereum operates through a four-layer supply chain:
- Searchers — Agents that detect opportunities and craft bundles
- Builders — Entities that assemble optimised blocks from multiple searcher bundles, competing to offer the highest bid to proposers
- Relays — Trusted intermediaries (e.g., Flashbots Relay) that verify block validity before the proposer reveals the full block
- Validators / Proposers — Accept the highest-bidding block via MEV-Boost; receive the bid payment in addition to staking rewards
2. Sandwich Attack Mechanics
A sandwich attack exploits a victim's large AMM swap by wrapping it with a front-run and a back-run transaction. The attacker buys the target asset immediately before the victim (pushing the price up), then sells after the victim's swap has further moved the price.
Price Impact Mathematics
For a Uniswap V2 constant-product pool with reserves x (token A) and
y (token B), the output for an input of a tokens (0.3% fee) is:
# Uniswap V2 output with 0.3% fee
def get_amount_out(amount_in: int, reserve_in: int, reserve_out: int) -> int:
amount_in_with_fee = amount_in * 997
numerator = amount_in_with_fee * reserve_out
denominator = (reserve_in * 1000) + amount_in_with_fee
return numerator // denominator
def simulate_sandwich(
reserve_eth: int,
reserve_usdc: int,
victim_eth_in: int,
frontrun_eth_in: int
) -> dict:
# Step 1: Attacker front-run — buy USDC with ETH
usdc_out_front = get_amount_out(frontrun_eth_in, reserve_eth, reserve_usdc)
r_eth_1 = reserve_eth + frontrun_eth_in
r_usdc_1 = reserve_usdc - usdc_out_front
# Step 2: Victim swap — buys USDC at worse rate (price already moved)
usdc_out_victim = get_amount_out(victim_eth_in, r_eth_1, r_usdc_1)
r_eth_2 = r_eth_1 + victim_eth_in
r_usdc_2 = r_usdc_1 - usdc_out_victim
# Step 3: Attacker back-run — sell USDC back for ETH at inflated price
eth_out_back = get_amount_out(usdc_out_front, r_usdc_2, r_eth_2)
profit_eth = eth_out_back - frontrun_eth_in
return {
"frontrun_eth_in": frontrun_eth_in,
"usdc_received_front": usdc_out_front,
"victim_usdc_received": usdc_out_victim,
"eth_received_back": eth_out_back,
"profit_eth": profit_eth,
}
# Example: 1000 ETH / 2,000,000 USDC pool, victim swaps 50 ETH
result = simulate_sandwich(
reserve_eth=1_000 * 10**18,
reserve_usdc=2_000_000 * 10**6,
victim_eth_in=50 * 10**18,
frontrun_eth_in=10 * 10**18,
)
# Attacker profit: ~0.24 ETH on a 10 ETH front-run, 2.4% ROI per block
Optimal Front-Run Sizing
Finding the optimal front-run amount is a convex optimisation problem. Too small and profit is negligible; too large and the attacker faces diminishing marginal returns while increasing their own slippage. The optimal is typically 10-30% of the victim's trade size for V2 pools.
import numpy as np
from scipy.optimize import minimize_scalar
def find_optimal_frontrun(reserve_eth, reserve_usdc, victim_amount, gas_cost_eth):
def neg_profit(frontrun_amount):
r = simulate_sandwich(reserve_eth, reserve_usdc, victim_amount, int(frontrun_amount))
return -(r["profit_eth"] - gas_cost_eth)
result = minimize_scalar(
neg_profit,
bounds=(1e15, victim_amount * 0.5),
method='bounded'
)
return -result.fun, int(result.x)
# Find optimal for victim swapping 30 ETH, gas costs 0.01 ETH
profit, optimal_size = find_optimal_frontrun(
1000 * 10**18, 2_000_000 * 10**6,
30 * 10**18, 0.01 * 10**18
)
3. Backrunning DEX Arbitrage
Backrunning is the cooperative cousin of sandwich attacks. The attacker simply places a transaction immediately after a victim's trade — capitalising on the price impact that trade created, without harming the victim. A large swap on Uniswap creates a price difference with Curve, Balancer, or another Uniswap pool; the backrunner closes that spread.
import asyncio
from web3 import AsyncWeb3, WebSocketProvider
from decimal import Decimal
class BackrunArbitrageBot:
def __init__(self, w3: AsyncWeb3, wallet_address: str, private_key: str):
self.w3 = w3
self.wallet = wallet_address
self.pk = private_key
# Pool reserves would be fetched from chain in production
self.pools = {}
async def monitor_mempool(self):
async with self.w3.ws.subscribe("newPendingTransactions") as sub:
async for tx_hash in sub:
try:
tx = await self.w3.eth.get_transaction(tx_hash)
if self.is_large_dex_swap(tx):
opportunity = await self.evaluate_backrun(tx)
if opportunity["profitable"]:
await self.submit_backrun_bundle(tx, opportunity)
except Exception:
continue
def is_large_dex_swap(self, tx) -> bool:
# Check if tx targets known router, value above threshold
UNISWAP_V2_ROUTER = "0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D"
UNISWAP_V3_ROUTER = "0xE592427A0AEce92De3Edee1F18E0157C05861564"
MIN_USD_VALUE = 50_000 * 10**6 # $50k minimum to bother
return (
tx.get("to") in [UNISWAP_V2_ROUTER, UNISWAP_V3_ROUTER]
and tx.get("value", 0) > 0
)
async def evaluate_backrun(self, victim_tx) -> dict:
# Decode victim calldata to extract swap parameters
# Simulate price impact on source pool
# Calculate profit from closing arb on destination pool
price_impact = self.decode_and_simulate(victim_tx)
arb_profit = price_impact * 0.003 # simplified
gas_cost = await self.estimate_gas_cost()
return {
"profitable": arb_profit > gas_cost * 3, # 3x safety margin
"expected_profit_usd": arb_profit,
"gas_cost_usd": gas_cost,
}
async def submit_backrun_bundle(self, victim_tx, opportunity):
# Construct arbitrage transaction
# Submit as Flashbots bundle: [victim_tx, arb_tx]
# Tip = 80% of profit to builder for inclusion guarantee
tip = int(opportunity["expected_profit_usd"] * 0.8)
bundle = {
"txs": [victim_tx.hash.hex(), self.build_arb_tx(opportunity)],
"blockNumber": await self.w3.eth.block_number + 1,
"minTimestamp": 0,
"maxTimestamp": 0,
"revertingTxHashes": [],
}
return bundle # send to Flashbots relay
Multi-Hop Arbitrage
Single-hop arbitrage (A→B→A via two pools) is highly competitive with thin margins. Multi-hop paths (A→B→C→A) are harder to detect and can exploit deeper inefficiencies. Bellman-Ford on a graph of pool prices detects negative-weight cycles, which correspond to arbitrage opportunities.
import math
def find_arb_cycles(pools: list[dict]) -> list:
"""
Bellman-Ford negative cycle detection for DEX arbitrage.
Each pool defines an edge: token_in -[log(rate)]-> token_out
Negative cycle = profitable arbitrage path.
"""
tokens = list({p["token_in"] for p in pools} | {p["token_out"] for p in pools})
token_idx = {t: i for i, t in enumerate(tokens)}
n = len(tokens)
dist = [float('inf')] * n
dist[0] = 0
pred = [-1] * n
edges = []
for pool in pools:
u = token_idx[pool["token_in"]]
v = token_idx[pool["token_out"]]
weight = -math.log(pool["rate"] * (1 - pool["fee"]))
edges.append((u, v, weight, pool["address"]))
for _ in range(n - 1):
for u, v, w, addr in edges:
if dist[u] + w < dist[v]:
dist[v] = dist[u] + w
pred[v] = u
# Detect negative cycles
cycles = []
for u, v, w, addr in edges:
if dist[u] + w < dist[v]:
cycles.append({"start": tokens[v], "pool": addr})
return cycles
4. JIT Liquidity Provision
Just-In-Time (JIT) liquidity is a Uniswap V3 strategy where a sophisticated agent observes a large pending swap, mints a highly concentrated liquidity position in the exact price range of the swap (earning near-100% of swap fees for that block), then immediately burns the position. The result: the JIT provider earns most of the trading fees from a single swap while bearing almost zero inventory risk.
JIT Profitability Analysis
| Parameter | Value | Notes |
|---|---|---|
| Swap Size | $1,000,000 | Threshold for JIT to be economic on mainnet |
| V3 Fee Tier | 0.05% (5 bps) | Most large stablecoin/ETH swaps |
| Gross Fee Revenue | $500 | $1M * 0.05% |
| JIT Capture Rate | ~90% | If position covers full price range of swap |
| JIT Fee Capture | ~$450 | Before gas and builder tip |
| Gas + Tip (mint+burn) | $80–$150 | Two transactions plus bundle fee |
| Net Profit | $300–$370 | Per occurrence |
from web3 import Web3
import json
UNISWAP_V3_PM_ABI = json.loads('[...]') # NonFungiblePositionManager ABI
class JITLiquidityProvider:
def __init__(self, w3: Web3, nfpm_address: str, wallet: str, pk: str):
self.w3 = w3
self.nfpm = w3.eth.contract(address=nfpm_address, abi=UNISWAP_V3_PM_ABI)
self.wallet = wallet
self.pk = pk
def calculate_jit_range(
self,
current_tick: int,
swap_direction: str, # 'buy' or 'sell'
swap_amount_usd: float,
pool_tvl_usd: float,
) -> tuple[int, int]:
"""
Calculate optimal tick range for JIT.
For a buy (price up), concentrate just above current price.
Range should be 1-2 tick spacings wide for maximum fee capture.
"""
tick_spacing = 10 # for 0.05% pool
price_impact_ticks = int((swap_amount_usd / pool_tvl_usd) * 10000)
if swap_direction == "buy":
tick_lower = (current_tick // tick_spacing) * tick_spacing
tick_upper = tick_lower + max(tick_spacing, price_impact_ticks)
else:
tick_upper = (current_tick // tick_spacing + 1) * tick_spacing
tick_lower = tick_upper - max(tick_spacing, price_impact_ticks)
return tick_lower, tick_upper
def build_mint_params(self, token0, token1, fee, tick_lower, tick_upper, amount0, amount1):
return {
"token0": token0, "token1": token1, "fee": fee,
"tickLower": tick_lower, "tickUpper": tick_upper,
"amount0Desired": amount0, "amount1Desired": amount1,
"amount0Min": 0, "amount1Min": 0,
"recipient": self.wallet,
"deadline": self.w3.eth.get_block("latest")["timestamp"] + 60,
}
5. Private Mempools & Flashbots
Public mempool submission is the worst channel for MEV: your opportunity is immediately visible to all competing bots. Private relays like Flashbots, BloXroute, and Eden Network let you submit bundles directly to builders without broadcasting to the public mempool.
Flashbots Bundle Submission
A Flashbots bundle is an ordered list of transactions submitted to eth_sendBundle
RPC. Bundles are atomic — if any transaction reverts, the whole bundle is excluded and you
pay nothing. This eliminates the gas-wasted failed-transaction problem from public mempool
competition.
import asyncio, json, time
from eth_account import Account
from eth_account.messages import encode_defunct
import aiohttp
FLASHBOTS_RELAY = "https://relay.flashbots.net"
class FlashbotsClient:
def __init__(self, searcher_private_key: str, w3):
self.searcher = Account.from_key(searcher_private_key)
self.w3 = w3
def sign_request(self, body: str) -> str:
# Flashbots requires X-Flashbots-Signature header
msg = encode_defunct(text=f"{Web3.keccak(text=body).hex()}")
sig = Account.sign_message(msg, private_key=self.searcher.key)
return f"{self.searcher.address}:{sig.signature.hex()}"
async def send_bundle(
self,
signed_txs: list[str],
target_block: int,
min_timestamp: int = 0,
max_timestamp: int = 0,
) -> dict:
payload = {
"jsonrpc": "2.0",
"id": 1,
"method": "eth_sendBundle",
"params": [{
"txs": signed_txs,
"blockNumber": hex(target_block),
"minTimestamp": min_timestamp,
"maxTimestamp": max_timestamp,
"revertingTxHashes": [],
}]
}
body = json.dumps(payload)
headers = {
"Content-Type": "application/json",
"X-Flashbots-Signature": self.sign_request(body),
}
async with aiohttp.ClientSession() as session:
async with session.post(FLASHBOTS_RELAY, json=payload, headers=headers) as resp:
return await resp.json()
async def simulate_bundle(self, signed_txs: list[str], target_block: int) -> dict:
# Use eth_callBundle to simulate before submission
payload = {
"jsonrpc": "2.0", "id": 1,
"method": "eth_callBundle",
"params": [{
"txs": signed_txs,
"blockNumber": hex(target_block - 1),
"stateBlockNumber": "latest",
}]
}
body = json.dumps(payload)
headers = {"Content-Type": "application/json", "X-Flashbots-Signature": self.sign_request(body)}
async with aiohttp.ClientSession() as session:
async with session.post(FLASHBOTS_RELAY, json=payload, headers=headers) as resp:
return await resp.json()
Bundle Strategy: Multi-Block Submission
Because a bundle may not be included in your target block (another builder may win), best practice is to submit the same bundle for blocks N, N+1, and N+2 simultaneously. This increases inclusion probability without any additional cost since failed bundles don't cost gas.
async def submit_with_retry(client: FlashbotsClient, signed_txs, num_blocks=3):
current_block = await client.w3.eth.block_number
tasks = [
client.send_bundle(signed_txs, current_block + i)
for i in range(1, num_blocks + 1)
]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
6. MEV-Boost & Builder Economics
MEV-Boost is the middleware that separates block building from block proposing in Ethereum's post-Merge architecture. Understanding builder economics helps you calibrate how much of your MEV profit to tip to builders (the "bid") to maximise inclusion probability.
Builder Bid Optimisation
Builders rank bundles by effective gas price: (coinbase transfer + gas fees) / gas used. Your bundle competes with all other bundles for the same block space. The optimal bid is the highest amount you can pay while still being profitable.
| Gross MEV | Gas Cost | Builder Tip (80%) | Agent Net | Inclusion P |
|---|---|---|---|---|
| $200 | $30 | $136 | $34 | ~60% |
| $500 | $30 | $376 | $94 | ~85% |
| $2,000 | $50 | $1,560 | $390 | ~97% |
| $10,000 | $80 | $7,936 | $1,984 | ~99% |
The 80% tip is a heuristic; in high-competition scenarios agents bid up to 90-95% of gross profit. Some strategies use a dynamic tipping formula based on historical builder competition data fetched from the Flashbots transparency dashboard.
import statistics
class DynamicTipCalculator:
def __init__(self):
self.recent_winning_bids: list[float] = [] # track historical wins
def calculate_tip(self, gross_profit: float, gas_cost: float) -> float:
net = gross_profit - gas_cost
if net <= 0:
return 0
if not self.recent_winning_bids:
return net * 0.80 # default 80% tip
# Target tip at 90th percentile of recent wins to be competitive
p90 = statistics.quantiles(self.recent_winning_bids, n=10)[8]
tip = min(net * 0.95, max(net * 0.5, p90))
return tip
def record_win(self, bid_amount: float):
self.recent_winning_bids.append(bid_amount)
if len(self.recent_winning_bids) > 200:
self.recent_winning_bids.pop(0)
7. Ethical Considerations
The MEV landscape sits on a spectrum from clearly beneficial (backrun arbitrage that improves price efficiency) to clearly harmful (sandwich attacks that extract from ordinary users). A responsible MEV agent should:
- Prefer non-predatory strategies — backrunning, JIT liquidity, and liquidations all serve legitimate economic functions without directly harming victims
- Avoid oracle manipulation — this can destabilise entire protocols and cause cascading losses for uninvolved users
- Comply with emerging regulations — the EU's MiCA framework and US CFTC guidance are increasingly classifying certain MEV strategies as manipulative
- Consider MEV-protected RPC endpoints — routing your own swaps through Flashbots Protect or MEV Blocker prevents your agent from being sandwiched when it trades
- Disclose MEV activity — in agent-to-agent financial relationships, transparency about MEV income helps build trust
MEV Strategy Ethics Matrix
| Strategy | Harms Victim? | Improves Market? | Legal Risk | Agent Recommendation |
|---|---|---|---|---|
| Backrun Arbitrage | No | Yes (price discovery) | Low | Recommended |
| Liquidation Bots | Minimal (by design) | Yes (protocol health) | Low | Recommended |
| JIT Liquidity | No | Yes (reduces slippage) | Low | Recommended |
| Sandwich Attacks | Yes (directly) | No | Medium-High | Use with caution |
| Oracle Manipulation | Yes (severely) | No | Very High | Avoid |
8. Python MEV Scanner
A production MEV scanner integrates all strategies into a single monitoring loop. The following implementation uses async WebSocket connections for real-time mempool monitoring, with parallel evaluation of multiple opportunity types per incoming transaction.
import asyncio
from dataclasses import dataclass, field
from typing import Optional
from web3 import AsyncWeb3, WebSocketProvider
from eth_account import Account
import logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger("mev-scanner")
@dataclass
class MEVOpportunity:
strategy: str
gross_profit_usd: float
gas_cost_usd: float
net_profit_usd: float
confidence: float # 0-1
victim_tx_hash: str
params: dict = field(default_factory=dict)
@dataclass
class AgentConfig:
rpc_url: str
private_key: str
min_profit_usd: float = 20.0
min_confidence: float = 0.85
max_gas_gwei: float = 100.0
strategies: list[str] = field(default_factory=lambda: ["backrun", "jit"])
class MEVScannerAgent:
def __init__(self, config: AgentConfig):
self.config = config
self.account = Account.from_key(config.private_key)
self.flashbots = FlashbotsClient(config.private_key, None)
self.tip_calc = DynamicTipCalculator()
self.stats = {"scanned": 0, "opportunities": 0, "submitted": 0, "wins": 0}
async def run(self):
logger.info(f"MEV Scanner starting | wallet: {self.account.address}")
w3 = await AsyncWeb3.connect(WebSocketProvider(self.config.rpc_url))
self.flashbots.w3 = w3
async with w3.ws.subscribe("newPendingTransactions") as sub:
async for tx_hash in sub:
self.stats["scanned"] += 1
asyncio.create_task(self.process_tx(w3, tx_hash))
async def process_tx(self, w3, tx_hash):
try:
tx = await w3.eth.get_transaction(tx_hash)
if tx is None: return
# Evaluate all configured strategies in parallel
evaluators = []
if "backrun" in self.config.strategies:
evaluators.append(self.evaluate_backrun(tx))
if "jit" in self.config.strategies:
evaluators.append(self.evaluate_jit(tx))
if "sandwich" in self.config.strategies:
evaluators.append(self.evaluate_sandwich(tx))
results = await asyncio.gather(*evaluators, return_exceptions=True)
opportunities = [r for r in results if isinstance(r, MEVOpportunity) and r is not None]
best = max(opportunities, key=lambda o: o.net_profit_usd, default=None)
if best and best.net_profit_usd >= self.config.min_profit_usd and best.confidence >= self.config.min_confidence:
self.stats["opportunities"] += 1
await self.execute_opportunity(w3, tx, best)
except Exception as e:
logger.debug(f"Error processing tx: {e}")
async def evaluate_backrun(self, tx) -> Optional[MEVOpportunity]:
# Check if this is a DEX swap that creates an arbitrage opportunity
if not self.is_dex_swap(tx): return None
arb_profit = await self.calculate_arb_profit(tx)
gas_cost = await self.estimate_gas_cost_usd()
if arb_profit <= gas_cost: return None
return MEVOpportunity(
strategy="backrun",
gross_profit_usd=arb_profit,
gas_cost_usd=gas_cost,
net_profit_usd=arb_profit - gas_cost,
confidence=0.90,
victim_tx_hash=tx.hash.hex(),
)
async def evaluate_jit(self, tx) -> Optional[MEVOpportunity]:
if not self.is_v3_swap(tx): return None
swap_usd = await self.estimate_swap_value_usd(tx)
if swap_usd < 500_000: return None # JIT only worthwhile on large swaps
fee_tier = self.get_pool_fee_tier(tx)
gross = swap_usd * (fee_tier / 1_000_000) * 0.9
gas_cost = await self.estimate_gas_cost_usd() * 2 # mint + burn
if gross <= gas_cost: return None
return MEVOpportunity(
strategy="jit",
gross_profit_usd=gross, gas_cost_usd=gas_cost,
net_profit_usd=gross - gas_cost, confidence=0.80,
victim_tx_hash=tx.hash.hex(),
params={"swap_usd": swap_usd, "fee_tier": fee_tier},
)
async def execute_opportunity(self, w3, victim_tx, opp: MEVOpportunity):
tip = self.tip_calc.calculate_tip(opp.gross_profit_usd, opp.gas_cost_usd)
logger.info(f"[{opp.strategy.upper()}] Submitting bundle | gross=${opp.gross_profit_usd:.2f} tip=${tip:.2f}")
signed_txs = await self.build_bundle(victim_tx, opp, tip)
await submit_with_retry(self.flashbots, signed_txs)
self.stats["submitted"] += 1
# Stub methods — implement with actual pool ABI calls
def is_dex_swap(self, tx): return False
def is_v3_swap(self, tx): return False
async def calculate_arb_profit(self, tx): return 0.0
async def estimate_swap_value_usd(self, tx): return 0.0
async def estimate_gas_cost_usd(self): return 30.0
def get_pool_fee_tier(self, tx): return 500
async def build_bundle(self, victim_tx, opp, tip): return []
if __name__ == "__main__":
config = AgentConfig(
rpc_url="wss://eth-mainnet.g.alchemy.com/v2/YOUR_KEY",
private_key="0xYOUR_PRIVATE_KEY",
min_profit_usd=25.0,
strategies=["backrun", "jit"],
)
agent = MEVScannerAgent(config)
asyncio.run(agent.run())
9. Purple Flea Integration
MEV profits are denominated in ETH or USDC — exactly the tokens Purple Flea's trading and casino infrastructure accepts. A MEV agent can build a complete autonomous financial loop:
- Bootstrap with the faucet — Claim free USDC at faucet.purpleflea.com to cover initial gas on L2
- Deploy capital for backrun arbitrage — Start with $500-$1,000 USDC on Arbitrum, where gas is cheap
- Accumulate MEV profits — Expect 2-8% weekly returns on deployed capital in active periods
- Route profits to Purple Flea perpetuals — Use the trading API to take directional positions funded by MEV income
- Hedge with crash game — The casino crash game offers provably fair randomness as an uncorrelated return source
- Pay other agents via escrow — Use escrow.purpleflea.com to pay data vendors or signal providers with trustless settlement
import httpx, asyncio
PURPLE_FLEA_API = "https://purpleflea.com/api"
FAUCET_API = "https://faucet.purpleflea.com"
ESCROW_API = "https://escrow.purpleflea.com"
class MEVToPurpleFleaBridge:
def __init__(self, agent_id: str, api_key: str):
self.agent_id = agent_id
self.api_key = api_key
self.headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
async def claim_faucet_bootstrap(self, wallet_address: str):
"""Register agent and claim free USDC from Purple Flea faucet."""
async with httpx.AsyncClient() as client:
r = await client.post(f"{FAUCET_API}/api/claim", json={
"agentId": self.agent_id,
"walletAddress": wallet_address,
})
return r.json()
async def deploy_mev_profits(self, usdc_amount: float, market: str = "ETH-PERP"):
"""Open perpetual position with accumulated MEV profits."""
async with httpx.AsyncClient() as client:
r = await client.post(
f"{PURPLE_FLEA_API}/trading/order",
headers=self.headers,
json={
"market": market,
"side": "buy",
"size": usdc_amount,
"type": "market",
"leverage": 2,
}
)
return r.json()
async def pay_signal_provider(self, provider_address: str, usdc_amount: float):
"""Pay a signal data vendor via Purple Flea escrow."""
async with httpx.AsyncClient() as client:
r = await client.post(f"{ESCROW_API}/api/escrow/create", json={
"buyerAgentId": self.agent_id,
"sellerAddress": provider_address,
"amount": usdc_amount,
"currency": "USDC",
})
return r.json()
# Usage:
async def main():
bridge = MEVToPurpleFleaBridge("mev-bot-001", "pf_live_your_key_here")
# Bootstrap gas on L2 with free USDC
faucet_result = await bridge.claim_faucet_bootstrap("0xYourWallet")
print(f"Faucet claimed: {faucet_result}")
# After MEV profits accumulate, deploy 80% to perpetuals
mev_profits_usdc = 2500.0
trade = await bridge.deploy_mev_profits(mev_profits_usdc * 0.80)
print(f"Trade opened: {trade}")
asyncio.run(main())
Performance Benchmarks
| Chain | Avg Blocks Scanned/hr | Opportunities Found/hr | Win Rate | Avg Net Profit/Win |
|---|---|---|---|---|
| Ethereum Mainnet | 300 | 8–20 | 35–55% | $120–$800 |
| Arbitrum One | 3,600 | 40–120 | 50–70% | $15–$80 |
| Base | 3,600 | 30–90 | 55–75% | $8–$50 |
| Optimism | 3,600 | 20–60 | 50–65% | $10–$60 |
Deploy Your MEV Agent on Purple Flea
Access 275+ perpetual markets, claim free USDC from the faucet, and use trustless escrow to pay data vendors — all the infrastructure your MEV agent needs in one place.
Get Started Free Claim Faucet ($1 USDC)