Find arbitrage opportunities across CEX/DEX with real-time detection
Scans multiple exchanges for arbitrage opportunities with real-time detection and risk analysis
/plugin marketplace add jeremylongshore/claude-code-plugins-plus-skills/plugin install authentication-validator@claude-code-plugins-plusAdvanced arbitrage opportunity scanner that detects profitable trading opportunities across centralized exchanges (CEX), decentralized exchanges (DEX), cross-chain bridges, and funding rate differentials. Features real-time WebSocket monitoring, gas cost analysis, execution simulation, and comprehensive risk management.
This command scans multiple markets simultaneously to identify price discrepancies that can be exploited for profit. It supports:
┌─────────────────────────────────────────────────────────────┐
│ Arbitrage Finder Engine │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ CEX Monitor │ │ DEX Monitor │ │ Bridge Monitor│ │
│ │ (WebSocket) │ │ (RPC Calls) │ │ (API Calls) │ │
│ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │
│ │ │ │ │
│ └──────────┬───────┴──────────────────┘ │
│ ▼ │
│ ┌──────────────────────┐ │
│ │ Opportunity Detector │ │
│ │ - Price Analysis │ │
│ │ - Spread Calculation│ │
│ │ - Gas Estimation │ │
│ └──────────┬───────────┘ │
│ ▼ │
│ ┌──────────────────────┐ │
│ │ Profitability Engine │ │
│ │ - Fee Calculation │ │
│ │ - Slippage Analysis │ │
│ │ - Risk Scoring │ │
│ └──────────┬───────────┘ │
│ ▼ │
│ ┌──────────────────────┐ │
│ │ Execution Simulator │ │
│ │ - Order Routing │ │
│ │ - Timing Analysis │ │
│ │ - Success Prediction│ │
│ └──────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
#!/usr/bin/env python3
"""
Advanced Arbitrage Opportunity Finder
Supports CEX, DEX, flash loans, cross-chain, and funding rate arbitrage
"""
import asyncio
import json
import time
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, field
from decimal import Decimal
from enum import Enum
import websockets
import aiohttp
from web3 import Web3
from web3.middleware import geth_poa_middleware
import ccxt.async_support as ccxt
# ============================================================================
# Data Models
# ============================================================================
class ArbitrageType(Enum):
"""Types of arbitrage opportunities"""
CEX_CEX = "cex_cex"
DEX_TRIANGULAR = "dex_triangular"
FLASH_LOAN = "flash_loan"
CROSS_CHAIN = "cross_chain"
FUNDING_RATE = "funding_rate"
STATISTICAL = "statistical"
class RiskLevel(Enum):
"""Risk assessment levels"""
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
EXTREME = "extreme"
@dataclass
class PriceData:
"""Price information from an exchange"""
exchange: str
symbol: str
bid: Decimal
ask: Decimal
timestamp: float
volume_24h: Decimal
liquidity: Decimal
@dataclass
class GasCost:
"""Gas cost estimation"""
gas_limit: int
gas_price_gwei: Decimal
cost_eth: Decimal
cost_usd: Decimal
@dataclass
class ArbitrageOpportunity:
"""Detected arbitrage opportunity"""
type: ArbitrageType
symbol: str
buy_exchange: str
sell_exchange: str
buy_price: Decimal
sell_price: Decimal
spread_percentage: Decimal
gross_profit_usd: Decimal
fees_usd: Decimal
gas_cost: Optional[GasCost]
net_profit_usd: Decimal
execution_time_seconds: float
risk_level: RiskLevel
capital_required: Decimal
confidence_score: float
metadata: Dict = field(default_factory=dict)
# ============================================================================
# Exchange Connectors
# ============================================================================
class CEXConnector:
"""Centralized exchange connector with WebSocket support"""
def __init__(self, exchanges: List[str]):
self.exchanges = {}
self.price_cache = {}
self.ws_connections = {}
# Initialize CCXT exchanges
for exchange_id in exchanges:
try:
exchange_class = getattr(ccxt, exchange_id)
self.exchanges[exchange_id] = exchange_class({
'enableRateLimit': True,
'options': {'defaultType': 'spot'}
})
except AttributeError:
print(f"Exchange {exchange_id} not supported")
async def connect_websockets(self, symbols: List[str]):
"""Connect to exchange WebSocket feeds"""
tasks = []
for exchange_id in self.exchanges:
for symbol in symbols:
tasks.append(self._subscribe_ticker(exchange_id, symbol))
await asyncio.gather(*tasks, return_exceptions=True)
async def _subscribe_ticker(self, exchange_id: str, symbol: str):
"""Subscribe to ticker updates via WebSocket"""
exchange = self.exchanges[exchange_id]
try:
while True:
ticker = await exchange.watch_ticker(symbol)
price_data = PriceData(
exchange=exchange_id,
symbol=symbol,
bid=Decimal(str(ticker['bid'])),
ask=Decimal(str(ticker['ask'])),
timestamp=ticker['timestamp'] / 1000,
volume_24h=Decimal(str(ticker.get('baseVolume', 0))),
liquidity=Decimal(str(ticker.get('quoteVolume', 0)))
)
cache_key = f"{exchange_id}:{symbol}"
self.price_cache[cache_key] = price_data
except Exception as e:
print(f"WebSocket error {exchange_id} {symbol}: {e}")
await asyncio.sleep(5)
await self._subscribe_ticker(exchange_id, symbol)
def get_price(self, exchange_id: str, symbol: str) -> Optional[PriceData]:
"""Get cached price data"""
cache_key = f"{exchange_id}:{symbol}"
return self.price_cache.get(cache_key)
async def get_trading_fees(self, exchange_id: str) -> Tuple[Decimal, Decimal]:
"""Get maker and taker fees"""
exchange = self.exchanges[exchange_id]
await exchange.load_markets()
fees = exchange.fees.get('trading', {})
maker = Decimal(str(fees.get('maker', 0.001)))
taker = Decimal(str(fees.get('taker', 0.001)))
return maker, taker
async def close(self):
"""Close all exchange connections"""
tasks = [exchange.close() for exchange in self.exchanges.values()]
await asyncio.gather(*tasks)
class DEXConnector:
"""Decentralized exchange connector"""
def __init__(self, rpc_url: str, chain_id: int):
self.w3 = Web3(Web3.HTTPProvider(rpc_url))
self.w3.middleware_onion.inject(geth_poa_middleware, layer=0)
self.chain_id = chain_id
# Uniswap V2 Router ABI (simplified)
self.router_abi = [
{
"inputs": [
{"internalType": "uint256", "name": "amountIn", "type": "uint256"},
{"internalType": "address[]", "name": "path", "type": "address[]"}
],
"name": "getAmountsOut",
"outputs": [{"internalType": "uint256[]", "name": "amounts", "type": "uint256[]"}],
"stateMutability": "view",
"type": "function"
}
]
# DEX router addresses
self.routers = {
'uniswap_v2': '0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D',
'sushiswap': '0xd9e1cE17f2641f24aE83637ab66a2cca9C378B9F',
'pancakeswap': '0x10ED43C718714eb63d5aA57B78B54704E256024E', # BSC
}
async def get_amounts_out(
self,
dex: str,
amount_in: int,
path: List[str]
) -> List[int]:
"""Get output amounts for a token swap path"""
router_address = self.routers.get(dex)
if not router_address:
return []
router = self.w3.eth.contract(
address=Web3.toChecksumAddress(router_address),
abi=self.router_abi
)
try:
amounts = router.functions.getAmountsOut(
amount_in,
[Web3.toChecksumAddress(addr) for addr in path]
).call()
return amounts
except Exception as e:
print(f"Error getting amounts for {dex}: {e}")
return []
async def find_triangular_arbitrage(
self,
dex: str,
base_token: str,
quote_tokens: List[str],
amount_in: int
) -> List[Dict]:
"""Find triangular arbitrage opportunities"""
opportunities = []
# Test all triangular paths: base -> token1 -> token2 -> base
for token1 in quote_tokens:
for token2 in quote_tokens:
if token1 == token2:
continue
path = [base_token, token1, token2, base_token]
amounts = await self.get_amounts_out(dex, amount_in, path)
if len(amounts) == 4:
amount_out = amounts[-1]
profit = amount_out - amount_in
if profit > 0:
profit_percentage = (profit / amount_in) * 100
opportunities.append({
'dex': dex,
'path': path,
'amount_in': amount_in,
'amount_out': amount_out,
'profit': profit,
'profit_percentage': profit_percentage
})
return opportunities
async def estimate_gas(self, dex: str, path: List[str]) -> int:
"""Estimate gas cost for swap"""
# Base gas costs (empirical estimates)
base_gas = {
'uniswap_v2': 150000,
'sushiswap': 150000,
'pancakeswap': 120000,
}
# Add gas per hop
gas_per_hop = 30000
total_gas = base_gas.get(dex, 150000) + (len(path) - 2) * gas_per_hop
return total_gas
# ============================================================================
# Arbitrage Detector
# ============================================================================
class ArbitrageDetector:
"""Main arbitrage opportunity detector"""
def __init__(
self,
cex_exchanges: List[str],
dex_rpc_url: str,
eth_usd_price: Decimal
):
self.cex = CEXConnector(cex_exchanges)
self.dex = DEXConnector(dex_rpc_url, chain_id=1)
self.eth_usd_price = eth_usd_price
self.opportunities = []
async def start_monitoring(self, symbols: List[str]):
"""Start monitoring for arbitrage opportunities"""
await self.cex.connect_websockets(symbols)
# Continuous detection loop
while True:
await self._detect_opportunities(symbols)
await asyncio.sleep(0.1) # 100ms scan interval
async def _detect_opportunities(self, symbols: List[str]):
"""Detect all types of arbitrage opportunities"""
tasks = [
self._detect_cex_arbitrage(symbols),
self._detect_dex_triangular(),
self._detect_flash_loan_opportunities(),
self._detect_funding_rate_arbitrage()
]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Aggregate opportunities
for result in results:
if isinstance(result, list):
self.opportunities.extend(result)
async def _detect_cex_arbitrage(self, symbols: List[str]) -> List[ArbitrageOpportunity]:
"""Detect CEX-to-CEX arbitrage opportunities"""
opportunities = []
for symbol in symbols:
prices = {}
# Collect prices from all exchanges
for exchange_id in self.cex.exchanges.keys():
price_data = self.cex.get_price(exchange_id, symbol)
if price_data:
prices[exchange_id] = price_data
# Find best buy and sell prices
if len(prices) < 2:
continue
best_buy = min(prices.items(), key=lambda x: x[1].ask)
best_sell = max(prices.items(), key=lambda x: x[1].bid)
buy_exchange, buy_data = best_buy
sell_exchange, sell_data = best_sell
# Calculate spread
buy_price = buy_data.ask
sell_price = sell_data.bid
spread = ((sell_price - buy_price) / buy_price) * 100
# Minimum 0.5% spread threshold
if spread < Decimal('0.5'):
continue
# Calculate profitability
capital = Decimal('10000') # $10k test capital
# Get trading fees
buy_fees = await self.cex.get_trading_fees(buy_exchange)
sell_fees = await self.cex.get_trading_fees(sell_exchange)
# Calculate gross profit
amount = capital / buy_price
gross_profit = amount * (sell_price - buy_price)
# Calculate fees
buy_fee = capital * buy_fees[1] # Taker fee
sell_fee = (amount * sell_price) * sell_fees[1]
total_fees = buy_fee + sell_fee
# Calculate net profit
net_profit = gross_profit - total_fees
if net_profit > 0:
# Risk assessment
risk = self._assess_risk(
spread,
buy_data.liquidity,
sell_data.liquidity,
net_profit / capital * 100
)
opportunity = ArbitrageOpportunity(
type=ArbitrageType.CEX_CEX,
symbol=symbol,
buy_exchange=buy_exchange,
sell_exchange=sell_exchange,
buy_price=buy_price,
sell_price=sell_price,
spread_percentage=spread,
gross_profit_usd=gross_profit,
fees_usd=total_fees,
gas_cost=None,
net_profit_usd=net_profit,
execution_time_seconds=30.0,
risk_level=risk,
capital_required=capital,
confidence_score=self._calculate_confidence(buy_data, sell_data),
metadata={
'buy_volume_24h': float(buy_data.volume_24h),
'sell_volume_24h': float(sell_data.volume_24h),
'timestamp': time.time()
}
)
opportunities.append(opportunity)
return opportunities
async def _detect_dex_triangular(self) -> List[ArbitrageOpportunity]:
"""Detect triangular arbitrage on DEXes"""
opportunities = []
# Token addresses (mainnet)
weth = '0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2'
usdc = '0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48'
usdt = '0xdAC17F958D2ee523a2206206994597C13D831ec7'
dai = '0x6B175474E89094C44Da98b954EedeAC495271d0F'
quote_tokens = [usdc, usdt, dai]
amount_in = int(1e18) # 1 ETH
for dex in ['uniswap_v2', 'sushiswap']:
tri_opps = await self.dex.find_triangular_arbitrage(
dex, weth, quote_tokens, amount_in
)
for opp in tri_opps:
# Estimate gas cost
gas_limit = await self.dex.estimate_gas(dex, opp['path'])
gas_price_gwei = Decimal('50') # Assume 50 gwei
gas_cost_eth = (gas_limit * gas_price_gwei) / Decimal('1e9')
gas_cost_usd = gas_cost_eth * self.eth_usd_price
# Calculate net profit
profit_eth = Decimal(opp['profit']) / Decimal('1e18')
profit_usd = profit_eth * self.eth_usd_price
net_profit_usd = profit_usd - gas_cost_usd
if net_profit_usd > Decimal('10'): # Min $10 profit
gas_cost = GasCost(
gas_limit=gas_limit,
gas_price_gwei=gas_price_gwei,
cost_eth=gas_cost_eth,
cost_usd=gas_cost_usd
)
opportunity = ArbitrageOpportunity(
type=ArbitrageType.DEX_TRIANGULAR,
symbol=f"{dex}_triangular",
buy_exchange=dex,
sell_exchange=dex,
buy_price=Decimal('0'),
sell_price=Decimal('0'),
spread_percentage=Decimal(opp['profit_percentage']),
gross_profit_usd=profit_usd,
fees_usd=Decimal('0'),
gas_cost=gas_cost,
net_profit_usd=net_profit_usd,
execution_time_seconds=15.0,
risk_level=RiskLevel.MEDIUM,
capital_required=Decimal('1') * self.eth_usd_price,
confidence_score=0.7,
metadata={
'path': opp['path'],
'dex': dex
}
)
opportunities.append(opportunity)
return opportunities
async def _detect_flash_loan_opportunities(self) -> List[ArbitrageOpportunity]:
"""Detect flash loan arbitrage opportunities"""
opportunities = []
# Flash loan arbitrage combines CEX and DEX price differences
# with zero capital (borrowed from Aave/dYdX)
# This is a simplified detection - production would need
# Aave/dYdX contract integration
# Look for large price differences that justify flash loan fees
# Flash loan fee: 0.09% on Aave
flash_loan_fee = Decimal('0.0009')
# Check if any CEX-DEX spread > flash loan fee + gas + profit margin
# Implementation would go here
return opportunities
async def _detect_funding_rate_arbitrage(self) -> List[ArbitrageOpportunity]:
"""Detect funding rate arbitrage on perpetual futures"""
opportunities = []
# Funding rate arbitrage:
# - Long spot, short perpetual (if funding rate is positive)
# - Short spot, long perpetual (if funding rate is negative)
# This would require futures exchange API integration
# Implementation would go here
return opportunities
def _assess_risk(
self,
spread: Decimal,
buy_liquidity: Decimal,
sell_liquidity: Decimal,
profit_percentage: Decimal
) -> RiskLevel:
"""Assess risk level of arbitrage opportunity"""
# Low liquidity = higher risk
if buy_liquidity < 10000 or sell_liquidity < 10000:
return RiskLevel.HIGH
# Large spread might indicate stale data or market manipulation
if spread > 5:
return RiskLevel.HIGH
# Low profit margin after fees
if profit_percentage < 0.5:
return RiskLevel.MEDIUM
return RiskLevel.LOW
def _calculate_confidence(
self,
buy_data: PriceData,
sell_data: PriceData
) -> float:
"""Calculate confidence score for opportunity"""
score = 1.0
# Penalize stale prices (older than 1 second)
current_time = time.time()
if current_time - buy_data.timestamp > 1:
score *= 0.8
if current_time - sell_data.timestamp > 1:
score *= 0.8
# Penalize low liquidity
if buy_data.liquidity < 50000:
score *= 0.9
if sell_data.liquidity < 50000:
score *= 0.9
# Bonus for high volume
if buy_data.volume_24h > 1000000:
score *= 1.1
if sell_data.volume_24h > 1000000:
score *= 1.1
return min(score, 1.0)
def get_top_opportunities(self, limit: int = 10) -> List[ArbitrageOpportunity]:
"""Get top arbitrage opportunities sorted by net profit"""
sorted_opps = sorted(
self.opportunities,
key=lambda x: x.net_profit_usd,
reverse=True
)
return sorted_opps[:limit]
# ============================================================================
# Execution Simulator
# ============================================================================
class ExecutionSimulator:
"""Simulate arbitrage execution to test strategies"""
def __init__(self):
self.execution_history = []
async def simulate_execution(
self,
opportunity: ArbitrageOpportunity,
slippage_percentage: Decimal = Decimal('0.1')
) -> Dict:
"""Simulate executing an arbitrage opportunity"""
# Calculate slippage impact
slippage_cost = opportunity.gross_profit_usd * (slippage_percentage / 100)
# Calculate final profit after slippage
final_profit = opportunity.net_profit_usd - slippage_cost
# Simulate execution timing
execution_time = opportunity.execution_time_seconds
# Random success probability based on risk level
success_probability = {
RiskLevel.LOW: 0.95,
RiskLevel.MEDIUM: 0.80,
RiskLevel.HIGH: 0.60,
RiskLevel.EXTREME: 0.30
}.get(opportunity.risk_level, 0.5)
import random
success = random.random() < success_probability
result = {
'opportunity': opportunity,
'executed': success,
'final_profit_usd': float(final_profit) if success else 0,
'slippage_cost': float(slippage_cost),
'execution_time': execution_time,
'timestamp': time.time()
}
self.execution_history.append(result)
return result
def get_statistics(self) -> Dict:
"""Get execution statistics"""
if not self.execution_history:
return {}
total_executions = len(self.execution_history)
successful = sum(1 for r in self.execution_history if r['executed'])
total_profit = sum(r['final_profit_usd'] for r in self.execution_history)
return {
'total_executions': total_executions,
'successful_executions': successful,
'success_rate': successful / total_executions,
'total_profit_usd': total_profit,
'average_profit_usd': total_profit / total_executions if total_executions > 0 else 0
}
# ============================================================================
# Backtesting Framework
# ============================================================================
class BacktestEngine:
"""Backtest arbitrage strategies on historical data"""
def __init__(self, initial_capital: Decimal):
self.initial_capital = initial_capital
self.capital = initial_capital
self.trades = []
self.equity_curve = []
def load_historical_data(self, filepath: str) -> List[Dict]:
"""Load historical opportunity data"""
with open(filepath, 'r') as f:
return json.load(f)
def run_backtest(
self,
opportunities: List[Dict],
strategy: str = 'all_positive'
) -> Dict:
"""Run backtest on historical opportunities"""
for opp_data in opportunities:
# Reconstruct opportunity object
opp = ArbitrageOpportunity(**opp_data)
# Strategy: execute all positive net profit opportunities
if strategy == 'all_positive' and opp.net_profit_usd > 0:
self._execute_trade(opp)
# Strategy: only low risk opportunities
elif strategy == 'low_risk_only' and opp.risk_level == RiskLevel.LOW:
self._execute_trade(opp)
# Track equity curve
self.equity_curve.append({
'timestamp': opp.metadata.get('timestamp'),
'capital': float(self.capital)
})
return self._calculate_metrics()
def _execute_trade(self, opportunity: ArbitrageOpportunity):
"""Execute a trade in backtest"""
# Check if we have enough capital
if self.capital < opportunity.capital_required:
return
# Apply profit/loss
self.capital += opportunity.net_profit_usd
self.trades.append({
'timestamp': opportunity.metadata.get('timestamp'),
'type': opportunity.type.value,
'symbol': opportunity.symbol,
'profit': float(opportunity.net_profit_usd)
})
def _calculate_metrics(self) -> Dict:
"""Calculate backtest performance metrics"""
if not self.trades:
return {'error': 'No trades executed'}
total_profit = self.capital - self.initial_capital
total_return = (total_profit / self.initial_capital) * 100
winning_trades = [t for t in self.trades if t['profit'] > 0]
losing_trades = [t for t in self.trades if t['profit'] < 0]
win_rate = len(winning_trades) / len(self.trades) * 100
avg_win = sum(t['profit'] for t in winning_trades) / len(winning_trades) if winning_trades else 0
avg_loss = sum(t['profit'] for t in losing_trades) / len(losing_trades) if losing_trades else 0
# Calculate max drawdown
peak = self.initial_capital
max_drawdown = 0
for point in self.equity_curve:
capital = point['capital']
if capital > peak:
peak = capital
drawdown = ((peak - capital) / peak) * 100
if drawdown > max_drawdown:
max_drawdown = drawdown
return {
'initial_capital': float(self.initial_capital),
'final_capital': float(self.capital),
'total_profit': float(total_profit),
'total_return_percentage': float(total_return),
'total_trades': len(self.trades),
'winning_trades': len(winning_trades),
'losing_trades': len(losing_trades),
'win_rate_percentage': float(win_rate),
'average_win': float(avg_win),
'average_loss': float(avg_loss),
'profit_factor': abs(avg_win / avg_loss) if avg_loss != 0 else float('inf'),
'max_drawdown_percentage': float(max_drawdown)
}
# ============================================================================
# Main CLI Interface
# ============================================================================
async def main():
"""Main entry point"""
print("🔍 Arbitrage Opportunity Finder")
print("=" * 60)
# Configuration
cex_exchanges = ['binance', 'coinbase', 'kraken']
dex_rpc_url = 'https://eth-mainnet.g.alchemy.com/v2/YOUR_API_KEY'
eth_usd_price = Decimal('2500') # Current ETH price
# Initialize detector
detector = ArbitrageDetector(cex_exchanges, dex_rpc_url, eth_usd_price)
# Symbols to monitor
symbols = ['BTC/USDT', 'ETH/USDT', 'BNB/USDT']
print(f"\n📊 Monitoring {len(symbols)} symbols on {len(cex_exchanges)} exchanges")
print(f"🔗 Connected to DEXes on Ethereum mainnet")
print(f"⏱️ Scan interval: 100ms\n")
# Start monitoring (would run indefinitely)
try:
# Run for 60 seconds as demo
await asyncio.wait_for(
detector.start_monitoring(symbols),
timeout=60.0
)
except asyncio.TimeoutError:
pass
# Get top opportunities
top_opps = detector.get_top_opportunities(limit=5)
print(f"\n🎯 Top {len(top_opps)} Opportunities Found:\n")
for i, opp in enumerate(top_opps, 1):
print(f"{i}. {opp.type.value.upper()} - {opp.symbol}")
print(f" Buy: {opp.buy_exchange} @ ${opp.buy_price:.2f}")
print(f" Sell: {opp.sell_exchange} @ ${opp.sell_price:.2f}")
print(f" Spread: {opp.spread_percentage:.2f}%")
print(f" Net Profit: ${opp.net_profit_usd:.2f}")
print(f" Risk: {opp.risk_level.value.upper()}")
print(f" Confidence: {opp.confidence_score:.1%}")
print()
# Simulate execution on best opportunity
if top_opps:
simulator = ExecutionSimulator()
result = await simulator.simulate_execution(top_opps[0])
print(f"🎮 Execution Simulation:")
print(f" Success: {'✅' if result['executed'] else '❌'}")
print(f" Final Profit: ${result['final_profit_usd']:.2f}")
print(f" Slippage Cost: ${result['slippage_cost']:.2f}")
# Cleanup
await detector.cex.close()
if __name__ == '__main__':
asyncio.run(main())
# Find all arbitrage opportunities
/find-arbitrage
# Filter by minimum profit
/find-arbitrage --min-profit 100
# Filter by risk level
/find-arbitrage --risk low
# Specific arbitrage type
/find-arbitrage --type cex_cex
# Start WebSocket monitoring
detector = ArbitrageDetector(['binance', 'coinbase', 'kraken'], rpc_url, eth_price)
await detector.start_monitoring(['BTC/USDT', 'ETH/USDT'])
# Get opportunities in real-time
opportunities = detector.get_top_opportunities(limit=10)
# Simulate execution with 0.2% slippage
simulator = ExecutionSimulator()
result = await simulator.simulate_execution(opportunity, slippage_percentage=Decimal('0.2'))
# Get statistics
stats = simulator.get_statistics()
print(f"Success Rate: {stats['success_rate']:.1%}")
print(f"Total Profit: ${stats['total_profit_usd']:.2f}")
# Load historical data
backtest = BacktestEngine(initial_capital=Decimal('10000'))
opportunities = backtest.load_historical_data('opportunities.json')
# Run backtest
metrics = backtest.run_backtest(opportunities, strategy='low_risk_only')
print(f"Total Return: {metrics['total_return_percentage']:.2f}%")
print(f"Win Rate: {metrics['win_rate_percentage']:.1f}%")
print(f"Max Drawdown: {metrics['max_drawdown_percentage']:.2f}%")
# Set maximum acceptable gas cost
max_gas_usd = Decimal('50')
# Filter opportunities
profitable_opps = [
opp for opp in opportunities
if opp.gas_cost is None or opp.gas_cost.cost_usd < max_gas_usd
]
# Calculate acceptable slippage
max_slippage = Decimal('0.5') # 0.5%
min_net_profit = opportunity.gross_profit_usd * (1 - max_slippage / 100)
# Only execute if profitable after slippage
if min_net_profit > opportunity.fees_usd:
execute_trade(opportunity)
# Send alert when opportunity found
if opportunity.net_profit_usd > 100:
send_notification(
f"💰 Arbitrage Alert: ${opportunity.net_profit_usd:.2f} profit on {opportunity.symbol}"
)
This plugin integrates with:
Find profitable arbitrage opportunities with real-time detection, risk management, and comprehensive simulation capabilities.