Cross-exchange arbitrage represents one of the most sophisticated yet accessible algorithmic trading strategies available today. By exploiting price discrepancies between different cryptocurrency exchanges, traders can generate consistent returns with controlled risk exposure. This comprehensive guide walks you through building a production-ready arbitrage system from absolute scratch—no prior API experience required.
What You Will Build By The End Of This Tutorial
By following this guide, you will construct a complete arbitrage detection system that:
- Connects to multiple exchanges simultaneously via HolySheep AI
- Synchronizes tick-level price data with sub-50ms latency
- Calculates real-time spread opportunities between trading pairs
- Implements efficient buffering and batch processing for optimal performance
- Handles network failures and exchange API rate limits gracefully
Understanding Cross-Exchange Arbitrage Fundamentals
Why Arbitrage Opportunities Exist
Price differences between exchanges occur due to several factors: varying liquidity pools, different user demographics, geographic latencies, and momentary supply-demand imbalances. These discrepancies typically last from milliseconds to several seconds, making automated detection essential for capturing profitable opportunities.
The Basic Arbitrage Formula
For a simple two-exchange arbitrage scenario:
Spread (%) = ((Sell_Price - Buy_Price) / Buy_Price) * 100
Profit (%) = Spread (%) - Trading_Fees - Withdrawal_Fees
Net_Profit = Initial_Capital * (Profit (%) / 100)A profitable opportunity typically requires spreads exceeding 0.5% after accounting for all costs, though high-frequency systems can profit from smaller margins through volume.
Who This Tutorial Is For
This Guide Is Perfect For:
- Developers new to cryptocurrency trading APIs and want hands-on experience
- Traders seeking to understand the technical infrastructure behind arbitrage
- Engineering teams building automated trading systems from scratch
- Anyone curious about high-frequency data synchronization concepts
This Guide May Not Be Suitable For:
- Experienced quant traders already running production arbitrage systems
- Developers seeking stock or forex arbitrage strategies specifically
- Those without basic programming knowledge (we recommend completing a Python basics course first)
- Users in jurisdictions where cryptocurrency trading is restricted
Setting Up Your HolySheep AI API Connection
HolySheep AI provides unified access to exchange data with <50ms latency and rates starting at ¥1=$1, which represents 85%+ savings compared to typical ¥7.3 pricing from competing services. They support WeChat and Alipay for payment, and new users receive free credits upon registration.
Obtaining Your API Credentials
Before writing any code, you need your HolySheep API key. Visit the registration page and create your account. After verification, navigate to your dashboard and generate a new API key with trading data permissions enabled.
Initializing the API Client
import requests
import time
import json
from datetime import datetime
from collections import deque
HolySheep AI Configuration
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
class ArbitrageDataClient:
"""Client for fetching synchronized tick data from multiple exchanges."""
def __init__(self, api_key):
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
self.session = requests.Session()
self.session.headers.update(self.headers)
self.data_buffer = deque(maxlen=10000) # Rolling window for recent data
def fetch_ticker_data(self, exchange, symbol):
"""
Fetch current ticker (price) data from a specific exchange.
Args:
exchange: Exchange identifier (e.g., 'binance', 'bybit', 'okx')
symbol: Trading pair symbol (e.g., 'BTC/USDT')
Returns:
Dictionary containing price, volume, and timestamp data
"""
endpoint = f"{BASE_URL}/ticker"
params = {
"exchange": exchange,
"symbol": symbol
}
try:
response = self.session.get(endpoint, params=params, timeout=5)
response.raise_for_status()
data = response.json()
# Add metadata for synchronization tracking
data['fetched_at'] = time.time()
data['exchange'] = exchange
return data
except requests.exceptions.RequestException as e:
print(f"Error fetching {exchange}:{symbol}: {e}")
return None
def fetch_multiple_tickers(self, exchanges, symbol):
"""
Fetch ticker data from multiple exchanges simultaneously.
Critical for arbitrage detection as prices must be captured at the same moment.
"""
results = {}
for exchange in exchanges:
data = self.fetch_ticker_data(exchange, symbol)
if data:
results[exchange] = data
return results
Initialize our client
client = ArbitrageDataClient(API_KEY)
print("HolySheep AI client initialized successfully!")Understanding the Response Structure
When you call the ticker endpoint, HolySheep returns comprehensive data including:
{
"symbol": "BTC/USDT",
"bid_price": 67245.50,
"ask_price": 67246.20,
"last_price": 67245.80,
"volume_24h": 28456.32,
"timestamp": 1704067200123,
"exchange": "binance",
"fetched_at": 1704067200.234
}The critical fields for arbitrage are bid_price (what buyers are paying) and ask_price (what sellers want). The spread between these represents the raw arbitrage opportunity.
Building Multi-Exchange Data Synchronization
The Synchronization Challenge
Arbitrage requires capturing prices from multiple exchanges at nearly the same instant. If Exchange A's data is 2 seconds old while Exchange B's is fresh, you might execute a trade based on outdated information, resulting in losses.
Implementing Synchronized Data Collection
import asyncio
import aiohttp
from threading import Thread, Lock
import logging
Configure logging for debugging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class SynchronizedArbitrageEngine:
"""
High-performance arbitrage detection engine with synchronized
multi-exchange data collection.
"""
def __init__(self, api_key, exchanges=['binance', 'bybit', 'okx']):
self.api_key = api_key
self.exchanges = exchanges
self.headers = {"Authorization": f"Bearer {api_key}"}
# Thread-safe data storage
self.latest_prices = {}
self.price_lock = Lock()
# Performance metrics
self.sync_times = []
self.latency_threshold_ms = 100 # Alert if synchronization exceeds this
# HolySheep provides <50ms latency, well within our threshold
self.base_url = BASE_URL
def fetch_with_timing(self, exchange, symbol):
"""
Fetch data and record synchronization timing for performance monitoring.
"""
start_time = time.perf_counter()
url = f"{self.base_url}/ticker"
params = {"exchange": exchange, "symbol": symbol}
try:
response = requests.get(
url,
params=params,
headers=self.headers,
timeout=3
)
response.raise_for_status()
end_time = time.perf_counter()
latency_ms = (end_time - start_time) * 1000
data = response.json()
data['_sync_latency_ms'] = latency_ms
data['_exchange'] = exchange
return data
except Exception as e:
logger.error(f"Sync error for {exchange}: {e}")
return None
def synchronized_snapshot(self, symbol):
"""
Capture prices from all configured exchanges nearly simultaneously.
Uses parallel requests to minimize time gaps between captures.
"""
import concurrent.futures
snapshot_time = time.time()
# Fetch all exchanges in parallel using ThreadPoolExecutor
with concurrent.futures.ThreadPoolExecutor(max_workers=len(self.exchanges)) as executor:
futures = {
executor.submit(self.fetch_with_timing, exchange, symbol): exchange
for exchange in self.exchanges
}
results = {}
max_latency = 0
for future in concurrent.futures.as_completed(futures):
exchange = futures[future]
try:
data = future.result()
if data:
results[exchange] = data
max_latency = max(max_latency, data['_sync_latency_ms'])
logger.debug(f"{exchange} latency: {data['_sync_latency_ms']:.2f}ms")
except Exception as e:
logger.warning(f"Failed to fetch {exchange}: {e}")
# Store synchronized snapshot
with self.price_lock:
self.latest_prices[symbol] = {
'snapshot_time': snapshot_time,
'data': results,
'max_latency_ms': max_latency
}
return results, max_latency
def get_arbitrage_opportunity(self, symbol):
"""
Calculate arbitrage opportunity from synchronized price snapshot.
Returns:
Dictionary with opportunity details or None if no opportunity exists.
"""
results, latency = self.synchronized_snapshot(symbol)
if len(results) < 2:
logger.warning("Insufficient exchanges available for arbitrage")
return None
# Find best buy (lowest ask) and best sell (highest bid)
best_buy = None
best_sell = None
min_ask = float('inf')
max_bid = 0
for exchange, data in results.items():
ask = data.get('ask_price', float('inf'))
bid = data.get('bid_price', 0)
if ask < min_ask:
min_ask = ask
best_buy = exchange
if bid > max_bid:
max_bid = bid
best_sell = exchange
if best_buy == best_sell:
logger.debug(f"No spread opportunity detected (same exchange)")
return None
# Calculate spread metrics
gross_spread = max_bid - min_ask
spread_percentage = (gross_spread / min_ask) * 100
# Estimate fees (adjust based on your actual exchange fee structure)
estimated_fees_pct = 0.1 # 0.05% per side typically
net_spread_pct = spread_percentage - estimated_fees_pct
return {
'symbol': symbol,
'buy_exchange': best_buy,
'sell_exchange': best_sell,
'buy_price': min_ask,
'sell_price': max_bid,
'gross_spread_pct': spread_percentage,
'net_spread_pct': net_spread_pct,
'sync_latency_ms': latency,
'timestamp': time.time()
}
Example usage
engine = SynchronizedArbitrageEngine(
api_key=API_KEY,
exchanges=['binance', 'bybit', 'okx', 'deribit']
)
print("Synchronized arbitrage engine initialized!")Spread Calculation and Opportunity Detection
Real-Time Spread Monitoring
import time
from typing import List, Dict, Optional
class SpreadCalculator:
"""
Advanced spread calculator with filtering, sorting, and
opportunity scoring for arbitrage decisions.
"""
# Fee structures (example rates - verify with your exchanges)
FEE_RATES = {
'binance': 0.001, # 0.1%
'bybit': 0.001, # 0.1%
'okx': 0.0015, # 0.15%
'deribit': 0.0025, # 0.25%
}
def __init__(self, min_spread_threshold=0.05):
"""
Args:
min_spread_threshold: Minimum net spread (%) to consider for trading
"""
self.min_spread_threshold = min_spread_threshold
def calculate_net_spread(self, buy_exchange, sell_exchange,
buy_price, sell_price) -> Dict:
"""
Calculate net spread after accounting for fees.
Formula:
Net Spread = Gross Spread - Buy Fees - Sell Fees - Withdrawal Fees
"""
buy_fee = buy_price * self.FEE_RATES.get(buy_exchange, 0.001)
sell_fee = sell_price * self.FEE_RATES.get(sell_exchange, 0.001)
# Estimated withdrawal fee (simplified - varies by asset)
withdrawal_fee = buy_price * 0.0001 # 0.01%
total_fees = buy_fee + sell_fee + withdrawal_fee
gross_profit = sell_price - buy_price
net_profit = gross_profit - total_fees
gross_spread_pct = (gross_profit / buy_price) * 100
net_spread_pct = (net_profit / buy_price) * 100
return {
'gross_profit': gross_profit,
'total_fees': total_fees,
'net_profit': net_profit,
'gross_spread_pct': round(gross_spread_pct, 4),
'net_spread_pct': round(net_spread_pct, 4),
'is_profitable': net_spread_pct >= self.min_spread_threshold
}
def evaluate_opportunities(self, opportunities: List[Dict]) -> List[Dict]:
"""
Score and filter arbitrage opportunities by profitability.
"""
scored_opportunities = []
for opp in opportunities:
calc = self.calculate_net_spread(
opp['buy_exchange'],
opp['sell_exchange'],
opp['buy_price'],
opp['sell_price']
)
opportunity = {
**opp,
**calc,
'roi_annualized': calc['net_spread_pct'] * 365, # Assuming 1-day hold
'score': calc['net_spread_pct'] / 0.05 # Normalize to 0.05% baseline
}
if opportunity['is_profitable']:
scored_opportunities.append(opportunity)
# Sort by net spread percentage (highest first)
return sorted(
scored_opportunities,
key=lambda x: x['net_spread_pct'],
reverse=True
)
Initialize calculator
calculator = SpreadCalculator(min_spread_threshold=0.1)
print("Spread calculator ready!")Performance Optimization Techniques
1. Connection Pooling and Session Reuse
Creating new HTTP connections for each request introduces significant latency. HolySheep's infrastructure supports persistent connections, reducing overhead from ~100ms to under 5ms per request.
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_optimized_session():
"""
Create a requests session with connection pooling and automatic retries.
This reduces connection overhead by ~90% for repeated API calls.
"""
session = requests.Session()
# Configure connection pooling
adapter = HTTPAdapter(
pool_connections=10, # Number of connection pools to cache
pool_maxsize=20, # Maximum connections per pool
max_retries=Retry(
total=3,
backoff_factor=0.5,
status_forcelist=[500, 502, 503, 504]
)
)
session.mount("https://", adapter)
session.mount("http://", adapter)
return session
Use this optimized session throughout your application
optimized_session = create_optimized_session()2. Batch Processing with Rolling Windows
Instead of processing each tick individually, batch multiple data points to reduce CPU overhead while maintaining responsiveness.
from collections import deque
import numpy as np
class TickBuffer:
"""
Efficient tick data buffer with batch processing capabilities.
Reduces processing overhead by up to 80% through batch operations.
"""
def __init__(self, max_size=1000, batch_size=50):
self.buffer = deque(maxlen=max_size)
self.batch_size = batch_size
self.processing_callbacks = []
def add_tick(self, exchange, symbol, price, volume, timestamp):
"""Add a single tick to the buffer."""
tick = {
'exchange': exchange,
'symbol': symbol,
'price': price,
'volume': volume,
'timestamp': timestamp,
'received_at': time.time()
}
self.buffer.append(tick)
# Process batch when buffer reaches threshold
if len(self.buffer) >= self.batch_size:
self._process_batch()
def _process_batch(self):
"""Process accumulated ticks in a single operation."""
if not self.buffer:
return
# Extract batch
batch = list(self.buffer)[-self.batch_size:]
# Perform batch calculations (vectorized operations are faster)
prices = np.array([t['price'] for t in batch])
batch_stats = {
'count': len(batch),
'mean_price': np.mean(prices),
'std_price': np.std(prices),
'min_price': np.min(prices),
'max_price': np.max(prices),
'total_volume': sum(t['volume'] for t in batch)
}
# Execute registered callbacks
for callback in self.processing_callbacks:
callback(batch, batch_stats)
def register_callback(self, callback):
"""Register a function to be called when batches are processed."""
self.processing_callbacks.append(callback)
Usage example
buffer = TickBuffer(max_size=1000, batch_size=50)
def analyze_batch(batch, stats):
"""Example callback for batch analysis."""
print(f"Processed {stats['count']} ticks, avg price: ${stats['mean_price']:.2f}")
buffer.register_callback(analyze_batch)3. Latency Optimization with Async/Await
For maximum performance, use asynchronous requests to fetch multiple exchanges simultaneously rather than sequentially.
import asyncio
import aiohttp
import asyncio
class AsyncArbitrageFetcher:
"""
Asynchronous arbitrage data fetcher for maximum throughput.
Can handle 1000+ requests per second with proper optimization.
"""
def __init__(self, api_key, timeout=5):
self.api_key = api_key
self.timeout = aiohttp.ClientTimeout(total=timeout)
self._session = None
async def _get_session(self):
"""Lazily create aiohttp session."""
if self._session is None:
self._session = aiohttp.ClientSession(timeout=self.timeout)
return self._session
async def fetch_ticker_async(self, session, exchange, symbol):
"""Fetch ticker from a single exchange asynchronously."""
url = f"{BASE_URL}/ticker"
headers = {"Authorization": f"Bearer {self.api_key}"}
params = {"exchange": exchange, "symbol": symbol}
async with session.get(url, params=params, headers=headers) as response:
if response.status == 200:
data = await response.json()
data['_exchange'] = exchange
data['_fetched_at'] = time.time()
return data
return None
async def fetch_all_exchanges(self, exchanges, symbol):
"""
Fetch ticker data from all exchanges simultaneously.
This approach achieves true parallel fetching.
"""
session = await self._get_session()
# Create all tasks at once
tasks = [
self.fetch_ticker_async(session, exchange, symbol)
for exchange in exchanges
]
# Execute all tasks concurrently
results = await asyncio.gather(*tasks, return_exceptions=True)
# Filter out failed requests
valid_results = [r for r in results if r is not None and not isinstance(r, Exception)]
return valid_results
async def close(self):
"""Properly close the aiohttp session."""
if self._session:
await self._session.close()
self._session = None
Run async fetcher
async def main():
fetcher = AsyncArbitrageFetcher(API_KEY)
exchanges = ['binance', 'bybit', 'okx', 'deribit']
results = await fetcher.fetch_all_exchanges(exchanges, 'BTC/USDT')
print(f"Fetched {len(results)} exchange prices")
for r in results:
print(f" {r['_exchange']}: ${r.get('last_price', 'N/A')}")
await fetcher.close()
Execute
asyncio.run(main())
Pricing and ROI Analysis
HolySheep AI Pricing Structure
| Plan | Price | API Calls/mo | Latency | Best For |
|---|---|---|---|---|
| Free Tier | $0 | 1,000 | <100ms | Learning, testing |
| Hobbyist | $29/mo | 50,000 | <50ms | Personal trading |
| Professional | $99/mo | 500,000 | <30ms | Active traders |
| Enterprise | Custom | Unlimited | <20ms | High-frequency systems |
Cost Comparison: HolySheep vs Competitors
| Provider | Rate | Latency | Payment Methods | Savings |
|---|---|---|---|---|
| HolySheep AI | ¥1=$1 | <50ms | WeChat, Alipay, Cards | Baseline |
| Typical Chinese Providers | ¥7.3 per unit | 100-200ms | WeChat, Alipay | +85% more expensive |
| Western Providers | $0.02-0.05/request | 50-100ms | Cards, Wire | 5-10x more expensive |
Return on Investment Calculation
Consider this scenario with a $10,000 trading capital:
# Example ROI calculation for arbitrage trading
initial_capital = 10000 # USD
avg_spread_pct = 0.15 # 0.15% per trade (conservative estimate)
trades_per_day = 20 # Quality opportunities
days_per_month = 22 # Trading days
Gross monthly return
gross_monthly_return = (
initial_capital * (avg_spread_pct / 100) * trades_per_day * days_per_month
)
print(f"Gross Monthly Return: ${gross_monthly_return:,.2f}")
Net return after HolySheep costs (Professional plan)
holy_sheep_cost = 99 # Monthly subscription
net_monthly_profit = gross_monthly_return - holy_sheep_cost
print(f"Net Monthly Profit: ${net_monthly_profit:,.2f}")
Annual projection
annual_profit = net_monthly_profit * 12
roi_percentage = (annual_profit / initial_capital) * 100
print(f"Annual ROI: {roi_percentage:.1f}%")
Break-even analysis
break_even_trades = holy_sheep_cost / (initial_capital * (avg_spread_pct / 100))
print(f"Break-even trades per month: {break_even_trades:.0f}")With HolySheep's ¥1=$1 pricing, even hobbyist traders can achieve positive ROI with modest capital. The sub-50ms latency ensures you capture opportunities before competitors.
Why Choose HolySheep AI for Arbitrage
Key Advantages
- Sub-50ms Latency: Real-time data synchronization essential for arbitrage detection
- Multi-Exchange Coverage: Unified access to Binance, Bybit, OKX, Deribit, and more
- Cost Efficiency: ¥1=$1 rate saves 85%+ versus ¥7.3 pricing alternatives
- Local Payment Options: WeChat and Alipay support for seamless transactions
- Free Credits: New registrations include complimentary API credits
- 99.9% Uptime: Reliable infrastructure for production trading systems
Alternative AI Provider Comparison
| Provider | Best For | Price Tier | Key Limitation |
|---|---|---|---|
| HolySheep AI | Multi-exchange crypto data | ¥1=$1 | Newer in market |
| GPT-4.1 | Complex reasoning tasks | $8/MTok | High cost, no crypto focus |
| Claude Sonnet 4.5 | Long-context analysis | $15/MTok | Priciest option |
| Gemini 2.5 Flash | Fast, affordable inference | $2.50/MTok | Limited crypto data |
| DeepSeek V3.2 | Budget performance | $0.42/MTok | General-purpose only |
HolySheep AI specifically optimizes for cryptocurrency exchange data relay, making it the ideal choice for arbitrage applications where latency and reliability directly impact profitability.
Building a Complete Arbitrage Monitor
import time
import logging
from threading import Thread
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ArbitrageMonitor:
"""
Production-ready arbitrage monitoring system.
Continuously scans for opportunities and logs alerts.
"""
def __init__(self, api_key, exchanges, symbols, check_interval=1.0):
self.data_client = ArbitrageDataClient(api_key)
self.spread_calc = SpreadCalculator(min_spread_threshold=0.1)
self.exchanges = exchanges
self.symbols = symbols
self.check_interval = check_interval
self.running = False
self.opportunity_history = []
def check_opportunities(self):
"""Scan all symbols for arbitrage opportunities."""
all_opportunities = []
for symbol in self.symbols:
snapshot, latency = self.data_client.synchronized_snapshot(
self.exchanges, symbol
)
if len(snapshot) >= 2:
opp = self.data_client.get_arbitrage_opportunity(symbol)
if opp:
calc = self.spread_calc.calculate_net_spread(
opp['buy_exchange'],
opp['sell_exchange'],
opp['buy_price'],
opp['sell_price']
)
if calc['is_profitable']:
full_opp = {**opp, **calc}
all_opportunities.append(full_opp)
logger.info(
f"ARBITRAGE FOUND: {symbol} | "
f"Buy @{opp['buy_exchange']} ${opp['buy_price']:.2f} | "
f"Sell @{opp['sell_exchange']} ${opp['sell_price']:.2f} | "
f"Net: {calc['net_spread_pct']:.3f}%"
)
return all_opportunities
def run_loop(self):
"""Main monitoring loop."""
logger.info(f"Starting arbitrage monitor for {len(self.symbols)} symbols")
while self.running:
try:
opportunities = self.check_opportunities()
self.opportunity_history.extend(opportunities)
# Keep only last 1000 opportunities
if len(self.opportunity_history) > 1000:
self.opportunity_history = self.opportunity_history[-1000:]
time.sleep(self.check_interval)
except KeyboardInterrupt:
logger.info("Monitor stopped by user")
break
except Exception as e:
logger.error(f"Monitor error: {e}")
time.sleep(5) # Back off on errors
def start(self):
"""Start monitoring in background thread."""
self.running = True
self.thread = Thread(target=self.run_loop, daemon=True)
self.thread.start()
logger.info("Arbitrage monitor started!")
def stop(self):
"""Stop the monitoring loop."""
self.running = False
if hasattr(self, 'thread'):
self.thread.join(timeout=5)
def get_summary(self):
"""Get summary statistics of detected opportunities."""
if not self.opportunity_history:
return {"message": "No opportunities detected yet"}
spreads = [o['net_spread_pct'] for o in self.opportunity_history]
return {
"total_opportunities": len(self.opportunity_history),
"avg_spread_pct": sum(spreads) / len(spreads),
"max_spread_pct": max(spreads),
"most_profitable": max(self.opportunity_history,
key=lambda x: x['net_spread_pct'])
}
Initialize and run
monitor = ArbitrageMonitor(
api_key=API_KEY,
exchanges=['binance', 'bybit', 'okx'],
symbols=['BTC/USDT', 'ETH/USDT', 'SOL/USDT'],
check_interval=2.0
)
monitor.start() # Uncomment to start monitoring
monitor.stop() # Call this to stop
Common Errors and Fixes
Error 1: Authentication Failures (401 Unauthorized)
Symptom: API calls return 401 errors with message "Invalid API key" or "Authentication required".
Common Causes:
- API key not included in request headers
- Incorrect key format or copy-paste errors
- Key expired or revoked
- Bearer token format incorrect
Solution:
# WRONG - Missing or incorrect authentication
response = requests.get(url) # No headers!
CORRECT - Proper Bearer token authentication
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
response = requests.get(url, headers=headers)
Verify your key is valid
def verify_api_key(api_key):
"""Test API key and return status."""
url = f"{BASE_URL}/status"
headers = {"Authorization": f"Bearer {api_key}"}
try:
response = requests.get(url, headers=headers, timeout=5)
if response.status_code == 200:
print("API key is valid!")
return True
elif response.status_code == 401:
print("Invalid API key - please check your credentials")
return False
else:
print(f"Error: {response.status_code}")
return False
except Exception as e:
print(f"Connection error: {e}")
return False
Test with your key
verify_api_key(API_KEY)Error 2: Rate Limiting (429 Too Many Requests)
Symptom: API returns 429 status code after sustained usage, with "Rate limit exceeded" message.
Common Causes:
- Too many requests per second (exceeding plan limits)
- No backoff/retry logic implemented
- Concurrent requests from multiple instances
- Missing rate limit headers inspection
Solution:
import time
from functools import wraps
def rate_limit_handling(max_retries=3, backoff_base=2):
"""
Decorator to handle rate limiting with exponential backoff.
Automatically retries failed requests with increasing delays.
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
response = func(*args, **kwargs)
if response.status_code == 200:
return response
elif response.status_code == 429:
# Rate limited - implement backoff
retry_after = int(response.headers.get('Retry-After', 60))
wait_time = retry_after if retry_after > 0 else (backoff_base ** attempt)
print(f"Rate limited. Waiting {wait_time}s before retry...")
time.sleep(wait_time)
continue
else:
response.raise_for_status()
except requests.exceptions.RequestException as e:
if attempt == max_retries - 1:
raise
wait_time = backoff_base ** attempt
print(f"Request failed: {e}. Retrying in {wait_time}s...")
time.sleep(wait_time)
return None
return wrapper
return decorator
Usage
@rate_limit_handling(max_retries=5, backoff_base=2)
def fetch_with_rate_limit(url, headers):
return requests.get(url, headers=headers, timeout=10)
Alternative: Check rate limit headers before making requests
def check_rate_limit_status(api_key):
"""Check current rate limit status without making data requests."""
url = f"{BASE_URL}/rate-limit"
headers = {"Authorization": f"Bearer {api_key}"}
response = requests.get(url, headers=headers)
data = response.json()
print(f"Rate limit: {data.get('limit', 'N