Proxies for Blockchain Applications: DeFi, NFT, and Crypto Trading

The blockchain ecosystem has exploded with innovative applications across decentralized finance (DeFi), non-fungible tokens (NFTs), and cryptocurrency trading. However, these applications often face challenges with rate limiting, IP blocking, geographical restrictions, and network reliability. Proxies provide essential infrastructure for blockchain applications, enabling reliable data access, geographic distribution, and enhanced security. This comprehensive guide explores how to leverage proxies for various blockchain use cases.

Understanding Blockchain Proxy Requirements

Unique Challenges in Blockchain Development

High-Frequency API Calls:

from typing import List, Dict, Optional
from datetime import datetime, timedelta

class BlockchainProxyManager:
    def __init__(self, proxy_pool: List[Dict], rpc_endpoints: List[str]):
        self.proxy_pool = proxy_pool
        self.rpc_endpoints = rpc_endpoints
        self.current_proxy_index = 0
        self.request_counts = {}
        self.rate_limits = {
            'ethereum': {'requests_per_second': 10, 'daily_limit': 100000},
            'polygon': {'requests_per_second': 20, 'daily_limit': 200000},
            'bsc': {'requests_per_second': 15, 'daily_limit': 150000}
        }
        
    async def get_block_data(self, 
                           chain: str, 
                           block_number: int, 
                           retries: int = 3) -> Optional[Dict]:
        """Fetch block data with proxy rotation and rate limiting"""
        
        for attempt in range(retries):
            proxy = self._get_next_proxy()
            
            # Check rate limits
            if not self._check_rate_limit(chain):
                await asyncio.sleep(1)
                continue
                
            try:
                rpc_payload = {
                    "jsonrpc": "2.0",
                    "method": "eth_getBlockByNumber",
                    "params": [hex(block_number), True],
                    "id": 1
                }
                
                proxy_config = {
                    'http': f"http://{proxy['username']}:{proxy['password']}@{proxy['host']}:{proxy['port']}",
                    'https': f"http://{proxy['username']}:{proxy['password']}@{proxy['host']}:{proxy['port']}"
                }
                
                async with aiohttp.ClientSession() as session:
                    async with session.post(
                        self._get_rpc_endpoint(chain),
                        json=rpc_payload,
                        proxy=proxy_config['http'],
                        timeout=aiohttp.ClientTimeout(total=30)
                    ) as response:
                        
                        if response.status == 200:
                            data = await response.json()
                            self._update_request_count(chain)
                            return data.get('result')
                        elif response.status == 429:
                            # Rate limited, try next proxy
                            await asyncio.sleep(2 ** attempt)
                            continue
                            
            except Exception as e:
                print(f"Request failed on attempt {attempt + 1}: {e}")
                if attempt < retries - 1:
                    await asyncio.sleep(2 ** attempt)
                    
        return None
        
    def _get_next_proxy(self) -> Dict:
        """Get next proxy using round-robin"""
        proxy = self.proxy_pool[self.current_proxy_index]
        self.current_proxy_index = (self.current_proxy_index + 1) % len(self.proxy_pool)
        return proxy
        
    def _check_rate_limit(self, chain: str) -> bool:
        """Check if we're within rate limits for the chain"""
        current_time = datetime.now()
        
        if chain not in self.request_counts:
            self.request_counts[chain] = {'requests': [], 'daily_count': 0, 'last_reset': current_time}
            
        # Clean old requests (older than 1 second)
        self.request_counts[chain]['requests'] = [
            req_time for req_time in self.request_counts[chain]['requests']
            if current_time - req_time < timedelta(seconds=1)
        ]
        
        # Check per-second limit
        if len(self.request_counts[chain]['requests']) >= self.rate_limits[chain]['requests_per_second']:
            return False
            
        # Check daily limit
        if current_time.date() != self.request_counts[chain]['last_reset'].date():
            self.request_counts[chain]['daily_count'] = 0
            self.request_counts[chain]['last_reset'] = current_time
            
        if self.request_counts[chain]['daily_count'] >= self.rate_limits[chain]['daily_limit']:
            return False
            
        return True
        
    def _update_request_count(self, chain: str):
        """Update request counters"""
        current_time = datetime.now()
        self.request_counts[chain]['requests'].append(current_time)
        self.request_counts[chain]['daily_count'] += 1

DeFi Protocol Integration

Automated Liquidity Pool Monitoring:

from web3 import Web3
from eth_abi import decode_abi


class DeFiProxyConnector:
    def __init__(self, proxy_manager: BlockchainProxyManager):
        self.proxy_manager = proxy_manager
        self.web3_instances = {}
        self.contract_abis = self._load_contract_abis()
        
    def _load_contract_abis(self) -> Dict:
        """Load common DeFi contract ABIs"""
        return {
            'uniswap_v2_pair': [
                {
                    "inputs": [],
                    "name": "getReserves",
                    "outputs": [
                        {"internalType": "uint112", "name": "_reserve0", "type": "uint112"},
                        {"internalType": "uint112", "name": "_reserve1", "type": "uint112"},
                        {"internalType": "uint32", "name": "_blockTimestampLast", "type": "uint32"}
                    ],
                    "stateMutability": "view",
                    "type": "function"
                }
            ],
            'erc20': [
                {
                    "inputs": [{"internalType": "address", "name": "account", "type": "address"}],
                    "name": "balanceOf",
                    "outputs": [{"internalType": "uint256", "name": "", "type": "uint256"}],
                    "stateMutability": "view",
                    "type": "function"
                }
            ]
        }
        
    async def monitor_liquidity_pools(self, 
                                    pool_addresses: List[str], 
                                    chain: str = 'ethereum') -> Dict:
        """Monitor multiple liquidity pools for price changes"""
        
        pool_data = {}
        
        for pool_address in pool_addresses:
            try:
                # Get current reserves
                reserves = await self._get_pool_reserves(pool_address, chain)
                
                if reserves:
                    # Calculate price ratio
                    price_ratio = reserves['reserve1'] / reserves['reserve0'] if reserves['reserve0'] > 0 else 0
                    
                    pool_data[pool_address] = {
                        'reserves': reserves,
                        'price_ratio': price_ratio,
                        'timestamp': datetime.now().isoformat(),
                        'chain': chain
                    }
                    
            except Exception as e:
                print(f"Error monitoring pool {pool_address}: {e}")
                
        return pool_data
        
    async def _get_pool_reserves(self, pool_address: str, chain: str) -> Optional[Dict]:
        """Get liquidity pool reserves using proxy rotation"""
        
        # Prepare contract call data
        function_signature = Web3.keccak(text="getReserves()")[:4]
        
        call_data = {
            "to": pool_address,
            "data": function_signature.hex()
        }
        
        # Make eth_call through proxy
        result = await self.proxy_manager.make_rpc_call(
            chain, "eth_call", [call_data, "latest"]
        )
        
        if result and result != "0x":
            # Decode the result
            decoded = decode_abi(['uint112', 'uint112', 'uint32'], bytes.fromhex(result[2:]))
            
            return {
                'reserve0': decoded[0],
                'reserve1': decoded[1],
                'timestamp': decoded[2]
            }
            
        return None
        
    async def track_defi_transactions(self, 
                                    protocols: List[str], 
                                    transaction_types: List[str]) -> List[Dict]:
        """Track DeFi transactions across multiple protocols"""
        
        transactions = []
        
        for protocol in protocols:
            protocol_contracts = self._get_protocol_contracts(protocol)
            
            for contract_address in protocol_contracts:
                try:
                    # Get recent transactions
                    recent_txs = await self._get_contract_transactions(
                        contract_address, limit=100
                    )
                    
                    # Filter by transaction type
                    filtered_txs = [
                        tx for tx in recent_txs 
                        if self._classify_transaction(tx) in transaction_types
                    ]
                    
                    transactions.extend(filtered_txs)
                    
                except Exception as e:
                    print(f"Error tracking transactions for {protocol}: {e}")
                    
        return transactions
        
    def _classify_transaction(self, transaction: Dict) -> str:
        """Classify DeFi transaction type based on input data"""
        
        input_data = transaction.get('input', '')
        
        # Common DeFi function signatures
        function_signatures = {
            '0xa9059cbb': 'transfer',
            '0x23b872dd': 'transferFrom',
            '0x095ea7b3': 'approve',
            '0x38ed1739': 'swapExactTokensForTokens',
            '0x7ff36ab5': 'swapExactETHForTokens',
            '0x02751cec': 'remove_liquidity',
            '0xe8e33700': 'add_liquidity'
        }
        
        if len(input_data) >= 10:
            signature = input_data[:10]
            return function_signatures.get(signature, 'unknown')
            
        return 'unknown'

NFT Marketplace Integration

Real-Time NFT Data Collection

Multi-Marketplace NFT Monitor:

from datetime import datetime
from typing import List, Dict, Optional

class NFTMarketplaceMonitor:
    def __init__(self, proxy_manager: BlockchainProxyManager):
        self.proxy_manager = proxy_manager
        self.marketplace_apis = {
            'opensea': {
                'base_url': 'https://api.opensea.io/api/v1',
                'rate_limit': 4,  # requests per second
                'requires_key': True
            },
            'rarible': {
                'base_url': 'https://api.rarible.org/v0.1',
                'rate_limit': 10,
                'requires_key': False
            },
            'looksrare': {
                'base_url': 'https://api.looksrare.org/api/v1',
                'rate_limit': 5,
                'requires_key': True
            }
        }
        
    async def monitor_nft_collections(self, 
                                    collections: List[Dict], 
                                    marketplaces: List[str]) -> Dict:
        """Monitor NFT collections across multiple marketplaces"""
        
        monitoring_results = {}
        
        for marketplace in marketplaces:
            marketplace_data = {}
            
            for collection in collections:
                try:
                    collection_stats = await self._get_collection_stats(
                        collection, marketplace
                    )
                    
                    recent_sales = await self._get_recent_sales(
                        collection, marketplace, limit=50
                    )
                    
                    floor_price = await self._get_floor_price(
                        collection, marketplace
                    )
                    
                    marketplace_data[collection['contract_address']] = {
                        'stats': collection_stats,
                        'recent_sales': recent_sales,
                        'floor_price': floor_price,
                        'last_updated': datetime.now().isoformat()
                    }
                    
                except Exception as e:
                    print(f"Error monitoring {collection['name']} on {marketplace}: {e}")
                    
            monitoring_results[marketplace] = marketplace_data
            
        return monitoring_results
        
    async def _get_collection_stats(self, 
                                  collection: Dict, 
                                  marketplace: str) -> Optional[Dict]:
        """Get collection statistics from marketplace API"""
        
        api_config = self.marketplace_apis[marketplace]
        
        if marketplace == 'opensea':
            endpoint = f"{api_config['base_url']}/collection/{collection['slug']}/stats"
        elif marketplace == 'rarible':
            endpoint = f"{api_config['base_url']}/collections/{collection['contract_address']}"
        else:
            return None
            
        headers = self._get_marketplace_headers(marketplace)
        proxy = self.proxy_manager._get_next_proxy()
        
        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(
                    endpoint,
                    headers=headers,
                    proxy=f"http://{proxy['username']}:{proxy['password']}@{proxy['host']}:{proxy['port']}",
                    timeout=aiohttp.ClientTimeout(total=30)
                ) as response:
                    
                    if response.status == 200:
                        data = await response.json()
                        return self._normalize_collection_stats(data, marketplace)
                        
        except Exception as e:
            print(f"API request failed: {e}")
            
        return None
        
    async def track_nft_price_movements(self, 
                                      collection_addresses: List[str], 
                                      time_window: int = 24) -> Dict:
        """Track NFT price movements over specified time window"""
        
        price_movements = {}
        
        for address in collection_addresses:
            try:
                # Get sales data from blockchain
                sales_data = await self._get_blockchain_sales_data(
                    address, time_window
                )
                
                # Calculate price metrics
                if sales_data:
                    price_metrics = self._calculate_price_metrics(sales_data)
                    price_movements[address] = price_metrics
                    
            except Exception as e:
                print(f"Error tracking price movements for {address}: {e}")
                
        return price_movements
        
    async def _get_blockchain_sales_data(self, 
                                       contract_address: str, 
                                       hours: int) -> List[Dict]:
        """Get NFT sales data directly from blockchain"""
        
        # Calculate block range for time window
        latest_block = await self.proxy_manager.get_block_data('ethereum', 'latest')
        blocks_per_hour = 240  # ~15 second block time
        from_block = latest_block['number'] - (hours * blocks_per_hour)
        
        # Get Transfer events
        filter_params = {
            'fromBlock': hex(from_block),
            'toBlock': 'latest',
            'address': contract_address,
            'topics': [
                '0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef'  # Transfer event
            ]
        }
        
        logs = await self.proxy_manager.make_rpc_call(
            'ethereum', 'eth_getLogs', [filter_params]
        )
        
        sales_data = []
        
        for log in logs:
            try:
                # Decode transfer event
                transaction = await self.proxy_manager.make_rpc_call(
                    'ethereum', 'eth_getTransactionByHash', [log['transactionHash']]
                )
                
                if transaction and transaction['value'] != '0x0':
                    sales_data.append({
                        'token_id': int(log['topics'][3], 16),
                        'price': int(transaction['value'], 16) / 10**18,  # Convert from wei
                        'buyer': '0x' + log['topics'][2][-40:],
                        'seller': '0x' + log['topics'][1][-40:],
                        'timestamp': int(log['timeStamp'], 16),
                        'tx_hash': log['transactionHash']
                    })
                    
            except Exception as e:
                print(f"Error processing transfer log: {e}")
                
        return sales_data
        
    def _calculate_price_metrics(self, sales_data: List[Dict]) -> Dict:
        """Calculate price movement metrics"""
        
        if not sales_data:
            return {}
            
        prices = [sale['price'] for sale in sales_data]
        
        return {
            'total_sales': len(sales_data),
            'total_volume': sum(prices),
            'average_price': sum(prices) / len(prices),
            'min_price': min(prices),
            'max_price': max(prices),
            'price_change_24h': self._calculate_price_change(sales_data),
            'unique_buyers': len(set(sale['buyer'] for sale in sales_data)),
            'unique_sellers': len(set(sale['seller'] for sale in sales_data))
        }

Cryptocurrency Trading Applications

Multi-Exchange Arbitrage

Cross-Exchange Price Monitor:

from typing import Dict, List, Optional
from datetime import datetime

class CryptoArbitrageMonitor:
    def __init__(self, proxy_manager: BlockchainProxyManager):
        self.proxy_manager = proxy_manager
        self.exchanges = {}
        self.price_cache = {}
        self.arbitrage_opportunities = []
        
    async def initialize_exchanges(self, exchange_configs: List[Dict]):
        """Initialize exchange connections with proxy support"""
        
        for config in exchange_configs:
            try:
                proxy = self.proxy_manager._get_next_proxy()
                proxy_url = f"http://{proxy['username']}:{proxy['password']}@{proxy['host']}:{proxy['port']}"
                
                exchange_class = getattr(ccxt, config['exchange'])
                
                exchange_instance = exchange_class({
                    'apiKey': config.get('api_key'),
                    'secret': config.get('secret'),
                    'password': config.get('passphrase'),
                    'sandbox': config.get('sandbox', False),
                    'enableRateLimit': True,
                    'proxies': {
                        'http': proxy_url,
                        'https': proxy_url,
                    },
                    'timeout': 30000,
                })
                
                await exchange_instance.load_markets()
                self.exchanges[config['exchange']] = exchange_instance
                
            except Exception as e:
                print(f"Failed to initialize {config['exchange']}: {e}")
                
    async def monitor_arbitrage_opportunities(self, 
                                            trading_pairs: List[str], 
                                            min_profit_threshold: float = 0.005) -> List[Dict]:
        """Monitor arbitrage opportunities across exchanges"""
        
        opportunities = []
        
        for pair in trading_pairs:
            try:
                # Get prices from all exchanges
                exchange_prices = await self._get_prices_across_exchanges(pair)
                
                if len(exchange_prices) < 2:
                    continue
                    
                # Find arbitrage opportunities
                pair_opportunities = self._find_arbitrage_opportunities(
                    pair, exchange_prices, min_profit_threshold
                )
                
                opportunities.extend(pair_opportunities)
                
            except Exception as e:
                print(f"Error monitoring arbitrage for {pair}: {e}")
                
        return opportunities
        
    async def _get_prices_across_exchanges(self, pair: str) -> Dict:
        """Get current prices for a trading pair across all exchanges"""
        
        prices = {}
        tasks = []
        
        for exchange_name, exchange in self.exchanges.items():
            if pair in exchange.markets:
                task = self._fetch_ticker_with_retry(exchange, pair, exchange_name)
                tasks.append(task)
                
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        for result in results:
            if isinstance(result, dict) and 'exchange' in result:
                prices[result['exchange']] = result
                
        return prices
        
    async def _fetch_ticker_with_retry(self, 
                                     exchange, 
                                     pair: str, 
                                     exchange_name: str, 
                                     retries: int = 3) -> Optional[Dict]:
        """Fetch ticker with retry mechanism"""
        
        for attempt in range(retries):
            try:
                ticker = await exchange.fetch_ticker(pair)
                
                return {
                    'exchange': exchange_name,
                    'symbol': pair,
                    'bid': ticker['bid'],
                    'ask': ticker['ask'],
                    'last': ticker['last'],
                    'volume': ticker['baseVolume'],
                    'timestamp': ticker['timestamp']
                }
                
            except Exception as e:
                if attempt < retries - 1:
                    await asyncio.sleep(2 ** attempt)
                else:
                    print(f"Failed to fetch {pair} from {exchange_name}: {e}")
                    
        return None
        
    def _find_arbitrage_opportunities(self, 
                                    pair: str, 
                                    exchange_prices: Dict, 
                                    min_profit: float) -> List[Dict]:
        """Find arbitrage opportunities for a trading pair"""
        
        opportunities = []
        exchanges = list(exchange_prices.keys())
        
        for i, buy_exchange in enumerate(exchanges):
            for sell_exchange in exchanges[i+1:]:
                
                buy_price = exchange_prices[buy_exchange]['ask']
                sell_price = exchange_prices[sell_exchange]['bid']
                
                # Calculate potential profit
                if sell_price > buy_price:
                    profit_ratio = (sell_price - buy_price) / buy_price
                    
                    if profit_ratio >= min_profit:
                        opportunities.append({
                            'pair': pair,
                            'buy_exchange': buy_exchange,
                            'sell_exchange': sell_exchange,
                            'buy_price': buy_price,
                            'sell_price': sell_price,
                            'profit_ratio': profit_ratio,
                            'profit_percentage': profit_ratio * 100,
                            'timestamp': datetime.now().isoformat()
                        })
                        
                # Check reverse direction
                if buy_price > sell_price:
                    reverse_profit_ratio = (buy_price - sell_price) / sell_price
                    
                    if reverse_profit_ratio >= min_profit:
                        opportunities.append({
                            'pair': pair,
                            'buy_exchange': sell_exchange,
                            'sell_exchange': buy_exchange,
                            'buy_price': sell_price,
                            'sell_price': buy_price,
                            'profit_ratio': reverse_profit_ratio,
                            'profit_percentage': reverse_profit_ratio * 100,
                            'timestamp': datetime.now().isoformat()
                        })
                        
        return opportunities

MEV (Maximal Extractable Value) Monitoring

MEV Opportunity Detection:

from typing import Dict, List, Optional
from dataclasses import dataclass

@dataclass
class MEVOpportunity:
    type: str
    profit_estimate: float
    gas_cost: float
    net_profit: float
    block_number: int
    transaction_hashes: List[str]
    timestamp: datetime

class MEVMonitor:
    def __init__(self, proxy_manager: BlockchainProxyManager):
        self.proxy_manager = proxy_manager
        self.mempool_monitor = MempoolMonitor(proxy_manager)
        self.dex_pools = {}
        
    async def scan_mev_opportunities(self, 
                                   scan_types: List[str] = ['arbitrage', 'sandwich', 'liquidation']) -> List[MEVOpportunity]:
        """Scan for MEV opportunities in real-time"""
        
        opportunities = []
        
        # Get pending transactions from mempool
        pending_txs = await self.mempool_monitor.get_pending_transactions()
        
        for scan_type in scan_types:
            if scan_type == 'arbitrage':
                arb_opportunities = await self._scan_arbitrage_opportunities(pending_txs)
                opportunities.extend(arb_opportunities)
                
            elif scan_type == 'sandwich':
                sandwich_opportunities = await self._scan_sandwich_opportunities(pending_txs)
                opportunities.extend(sandwich_opportunities)
                
            elif scan_type == 'liquidation':
                liquidation_opportunities = await self._scan_liquidation_opportunities()
                opportunities.extend(liquidation_opportunities)
                
        return opportunities
        
    async def _scan_sandwich_opportunities(self, pending_txs: List[Dict]) -> List[MEVOpportunity]:
        """Scan for sandwich attack opportunities"""
        
        opportunities = []
        
        # Filter for large DEX swaps
        large_swaps = [
            tx for tx in pending_txs
            if self._is_large_dex_swap(tx) and float(tx.get('value', '0')) > 1e18  # > 1 ETH
        ]
        
        for swap_tx in large_swaps:
            try:
                # Analyze potential sandwich profit
                sandwich_analysis = await self._analyze_sandwich_potential(swap_tx)
                
                if sandwich_analysis['profitable']:
                    opportunity = MEVOpportunity(
                        type='sandwich',
                        profit_estimate=sandwich_analysis['profit_estimate'],
                        gas_cost=sandwich_analysis['gas_cost'],
                        net_profit=sandwich_analysis['net_profit'],
                        block_number=0,  # Pending
                        transaction_hashes=[swap_tx['hash']],
                        timestamp=datetime.now()
                    )
                    
                    opportunities.append(opportunity)
                    
            except Exception as e:
                print(f"Error analyzing sandwich opportunity: {e}")
                
        return opportunities
        
    async def _analyze_sandwich_potential(self, target_tx: Dict) -> Dict:
        """Analyze potential profit from sandwich attack"""
        
        # Decode transaction to understand the swap
        decoded_tx = await self._decode_dex_transaction(target_tx)
        
        if not decoded_tx:
            return {'profitable': False}
            
        # Get current pool state
        pool_address = decoded_tx['pool_address']
        current_reserves = await self.proxy_manager._get_pool_reserves(pool_address, 'ethereum')
        
        if not current_reserves:
            return {'profitable': False}
            
        # Simulate price impact of target transaction
        token_in = decoded_tx['token_in']
        token_out = decoded_tx['token_out']
        amount_in = decoded_tx['amount_in']
        
        # Calculate optimal sandwich amounts
        front_run_amount = self._calculate_optimal_frontrun_amount(
            current_reserves, amount_in, decoded_tx['slippage_tolerance']
        )
        
        # Estimate profits
        estimated_profit = self._calculate_sandwich_profit(
            current_reserves, amount_in, front_run_amount
        )
        
        # Estimate gas costs (front-run + back-run transactions)
        gas_cost = await self._estimate_sandwich_gas_cost()
        
        net_profit = estimated_profit - gas_cost
        
        return {
            'profitable': net_profit > 0.01,  # Minimum 0.01 ETH profit
            'profit_estimate': estimated_profit,
            'gas_cost': gas_cost,
            'net_profit': net_profit,
            'front_run_amount': front_run_amount
        }

Best Practices for Blockchain Applications

Error Handling and Resilience

Robust Error Handling:

from functools import wraps
from typing import Callable, Any

def blockchain_retry(max_retries: int = 3, backoff_factor: float = 2.0):
    """Decorator for retrying blockchain operations with exponential backoff"""
    
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        async def wrapper(*args, **kwargs) -> Any:
            last_exception = None
            
            for attempt in range(max_retries):
                try:
                    return await func(*args, **kwargs)
                    
                except Exception as e:
                    last_exception = e
                    
                    # Don't retry on certain errors
                    if 'insufficient funds' in str(e).lower():
                        raise e
                        
                    if attempt < max_retries - 1:
                        wait_time = backoff_factor ** attempt
                        await asyncio.sleep(wait_time)
                        print(f"Retrying {func.__name__} in {wait_time}s (attempt {attempt + 1})")
                        
            raise last_exception
            
        return wrapper
    return decorator

class BlockchainErrorHandler:
    def __init__(self):
        self.error_counts = {}
        self.circuit_breakers = {}
        
    def handle_rpc_error(self, error: Exception, endpoint: str) -> bool:
        """Handle RPC errors and implement circuit breaker pattern"""
        
        error_type = type(error).__name__
        
        # Track error frequency
        if endpoint not in self.error_counts:
            self.error_counts[endpoint] = {}
            
        if error_type not in self.error_counts[endpoint]:
            self.error_counts[endpoint][error_type] = 0
            
        self.error_counts[endpoint][error_type] += 1
        
        # Implement circuit breaker
        if self.error_counts[endpoint][error_type] > 5:
            self.circuit_breakers[endpoint] = time.time() + 300  # 5 minute cooldown
            return False
            
        return True
        
    def is_endpoint_available(self, endpoint: str) -> bool:
        """Check if endpoint is available (not circuit broken)"""
        
        if endpoint in self.circuit_breakers:
            if time.time() > self.circuit_breakers[endpoint]:
                del self.circuit_breakers[endpoint]
                return True
            return False
            
        return True

Rate Limiting and Quota Management

Smart Rate Limiting:

from collections import defaultdict, deque


class BlockchainRateLimiter:
    def __init__(self):
        self.request_history = defaultdict(deque)
        self.quotas = {
            'ethereum': {'per_second': 10, 'per_minute': 300, 'per_hour': 3600},
            'polygon': {'per_second': 20, 'per_minute': 600, 'per_hour': 7200},
            'bsc': {'per_second': 15, 'per_minute': 450, 'per_hour': 5400}
        }
        
    async def wait_for_rate_limit(self, chain: str, priority: int = 1):
        """Wait if necessary to respect rate limits"""
        
        current_time = time.time()
        
        # Clean old requests
        self._clean_old_requests(chain, current_time)
        
        # Check rate limits
        while not self._check_rate_limits(chain, current_time):
            wait_time = self._calculate_wait_time(chain, priority)
            await asyncio.sleep(wait_time)
            current_time = time.time()
            
        # Record this request
        self.request_history[chain].append(current_time)
        
    def _clean_old_requests(self, chain: str, current_time: float):
        """Remove old requests from history"""
        
        while (self.request_history[chain] and 
               current_time - self.request_history[chain][0] > 3600):  # 1 hour
            self.request_history[chain].popleft()
            
    def _check_rate_limits(self, chain: str, current_time: float) -> bool:
        """Check if we're within rate limits"""
        
        requests = self.request_history[chain]
        quotas = self.quotas[chain]
        
        # Check per-second limit
        recent_requests = sum(1 for req_time in requests if current_time - req_time < 1)
        if recent_requests >= quotas['per_second']:
            return False
            
        # Check per-minute limit
        minute_requests = sum(1 for req_time in requests if current_time - req_time < 60)
        if minute_requests >= quotas['per_minute']:
            return False
            
        # Check per-hour limit
        hour_requests = len(requests)
        if hour_requests >= quotas['per_hour']:
            return False
            
        return True

Conclusion

Proxies are essential infrastructure for modern blockchain applications, enabling reliable access to blockchain data, multi-chain interactions, and sophisticated trading strategies. Whether you're building DeFi protocols, NFT marketplaces, or cryptocurrency trading systems, implementing robust proxy infrastructure ensures your applications can scale effectively while maintaining security and compliance.

The key to success lies in understanding the unique requirements of blockchain applications, implementing proper error handling and rate limiting, and choosing the right proxy architecture for your specific use case. As the blockchain ecosystem continues to evolve, proxy infrastructure will become increasingly critical for applications that need to interact reliably with multiple chains and protocols.

Ready to build scalable blockchain applications with enterprise-grade proxy infrastructure? Contact our blockchain specialists for customized guidance on implementing proxy solutions for your DeFi, NFT, or crypto trading applications.