""" Example demonstrating multi-exchange connectivity with Binance, Coinbase, and Kraken. Shows how to connect to multiple exchanges simultaneously and handle their data. """ import asyncio import logging from datetime import datetime from ..connectors.binance_connector import BinanceConnector from ..connectors.coinbase_connector import CoinbaseConnector from ..connectors.kraken_connector import KrakenConnector from ..models.core import OrderBookSnapshot, TradeEvent # Set up logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class MultiExchangeManager: """Manages connections to multiple exchanges.""" def __init__(self): """Initialize multi-exchange manager.""" # Initialize connectors self.connectors = { 'binance': BinanceConnector(), 'coinbase': CoinbaseConnector(use_sandbox=True), # Use sandbox for testing 'kraken': KrakenConnector() } # Data tracking self.data_received = { 'binance': {'orderbooks': 0, 'trades': 0}, 'coinbase': {'orderbooks': 0, 'trades': 0}, 'kraken': {'orderbooks': 0, 'trades': 0} } # Set up data callbacks for name, connector in self.connectors.items(): connector.add_data_callback(lambda data, exchange=name: self._handle_data(exchange, data)) def _handle_data(self, exchange: str, data): """Handle data from any exchange.""" try: if isinstance(data, OrderBookSnapshot): self.data_received[exchange]['orderbooks'] += 1 logger.info(f"📊 {exchange.upper()}: Order book for {data.symbol} - " f"Bids: {len(data.bids)}, Asks: {len(data.asks)}") # Show best bid/ask if available if data.bids and data.asks: best_bid = max(data.bids, key=lambda x: x.price) best_ask = min(data.asks, key=lambda x: x.price) spread = best_ask.price - best_bid.price logger.info(f" Best: {best_bid.price} / {best_ask.price} (spread: {spread:.2f})") elif isinstance(data, TradeEvent): self.data_received[exchange]['trades'] += 1 logger.info(f"💰 {exchange.upper()}: Trade {data.symbol} - " f"{data.side} {data.size} @ {data.price}") except Exception as e: logger.error(f"Error handling data from {exchange}: {e}") async def connect_all(self): """Connect to all exchanges.""" logger.info("Connecting to all exchanges...") connection_tasks = [] for name, connector in self.connectors.items(): task = asyncio.create_task(self._connect_exchange(name, connector)) connection_tasks.append(task) # Wait for all connections results = await asyncio.gather(*connection_tasks, return_exceptions=True) # Report results for i, (name, result) in enumerate(zip(self.connectors.keys(), results)): if isinstance(result, Exception): logger.error(f"❌ Failed to connect to {name}: {result}") elif result: logger.info(f"✅ Connected to {name}") else: logger.warning(f"⚠️ Connection to {name} returned False") async def _connect_exchange(self, name: str, connector) -> bool: """Connect to a single exchange.""" try: return await connector.connect() except Exception as e: logger.error(f"Error connecting to {name}: {e}") return False async def subscribe_to_symbols(self, symbols: list): """Subscribe to order book and trade data for given symbols.""" logger.info(f"Subscribing to symbols: {symbols}") for symbol in symbols: for name, connector in self.connectors.items(): try: if connector.is_connected: # Subscribe to order book await connector.subscribe_orderbook(symbol) logger.info(f"📈 Subscribed to {symbol} order book on {name}") # Subscribe to trades await connector.subscribe_trades(symbol) logger.info(f"💱 Subscribed to {symbol} trades on {name}") # Small delay between subscriptions await asyncio.sleep(0.5) else: logger.warning(f"⚠️ {name} not connected, skipping {symbol}") except Exception as e: logger.error(f"Error subscribing to {symbol} on {name}: {e}") async def run_for_duration(self, duration_seconds: int): """Run data collection for specified duration.""" logger.info(f"Running data collection for {duration_seconds} seconds...") start_time = datetime.now() # Print statistics periodically while (datetime.now() - start_time).seconds < duration_seconds: await asyncio.sleep(10) # Print stats every 10 seconds self._print_statistics() logger.info("Data collection period completed") def _print_statistics(self): """Print current data statistics.""" logger.info("📊 Current Statistics:") total_orderbooks = 0 total_trades = 0 for exchange, stats in self.data_received.items(): orderbooks = stats['orderbooks'] trades = stats['trades'] total_orderbooks += orderbooks total_trades += trades logger.info(f" {exchange.upper()}: {orderbooks} order books, {trades} trades") logger.info(f" TOTAL: {total_orderbooks} order books, {total_trades} trades") async def disconnect_all(self): """Disconnect from all exchanges.""" logger.info("Disconnecting from all exchanges...") for name, connector in self.connectors.items(): try: await connector.disconnect() logger.info(f"✅ Disconnected from {name}") except Exception as e: logger.error(f"Error disconnecting from {name}: {e}") def get_connector_stats(self): """Get statistics from all connectors.""" stats = {} for name, connector in self.connectors.items(): try: if hasattr(connector, 'get_stats'): stats[name] = connector.get_stats() else: stats[name] = { 'connected': connector.is_connected, 'exchange': connector.exchange_name } except Exception as e: stats[name] = {'error': str(e)} return stats async def demonstrate_multi_exchange(): """Demonstrate multi-exchange connectivity.""" logger.info("=== Multi-Exchange Connectivity Demo ===") # Create manager manager = MultiExchangeManager() try: # Connect to all exchanges await manager.connect_all() # Wait a moment for connections to stabilize await asyncio.sleep(2) # Subscribe to some popular symbols symbols = ['BTCUSDT', 'ETHUSDT'] await manager.subscribe_to_symbols(symbols) # Run data collection for 30 seconds await manager.run_for_duration(30) # Print final statistics logger.info("=== Final Statistics ===") manager._print_statistics() # Print connector statistics logger.info("=== Connector Statistics ===") connector_stats = manager.get_connector_stats() for exchange, stats in connector_stats.items(): logger.info(f"{exchange.upper()}: {stats}") except Exception as e: logger.error(f"Error in multi-exchange demo: {e}") finally: # Clean up await manager.disconnect_all() async def test_individual_connectors(): """Test each connector individually.""" logger.info("=== Individual Connector Tests ===") # Test Binance logger.info("Testing Binance connector...") binance = BinanceConnector() try: symbols = await binance.get_symbols() logger.info(f"Binance symbols available: {len(symbols)}") # Test order book snapshot orderbook = await binance.get_orderbook_snapshot('BTCUSDT') if orderbook: logger.info(f"Binance order book: {len(orderbook.bids)} bids, {len(orderbook.asks)} asks") except Exception as e: logger.error(f"Binance test error: {e}") # Test Coinbase logger.info("Testing Coinbase connector...") coinbase = CoinbaseConnector(use_sandbox=True) try: symbols = await coinbase.get_symbols() logger.info(f"Coinbase symbols available: {len(symbols)}") # Test order book snapshot orderbook = await coinbase.get_orderbook_snapshot('BTCUSDT') if orderbook: logger.info(f"Coinbase order book: {len(orderbook.bids)} bids, {len(orderbook.asks)} asks") except Exception as e: logger.error(f"Coinbase test error: {e}") # Test Kraken logger.info("Testing Kraken connector...") kraken = KrakenConnector() try: symbols = await kraken.get_symbols() logger.info(f"Kraken symbols available: {len(symbols)}") # Test order book snapshot orderbook = await kraken.get_orderbook_snapshot('BTCUSDT') if orderbook: logger.info(f"Kraken order book: {len(orderbook.bids)} bids, {len(orderbook.asks)} asks") except Exception as e: logger.error(f"Kraken test error: {e}") async def main(): """Run all demonstrations.""" logger.info("Starting Multi-Exchange Examples...") try: # Test individual connectors first await test_individual_connectors() await asyncio.sleep(2) # Then test multi-exchange connectivity await demonstrate_multi_exchange() logger.info("All multi-exchange examples completed successfully!") except Exception as e: logger.error(f"Error running examples: {e}") if __name__ == "__main__": # Run the examples asyncio.run(main())