""" Test Multi-Exchange Consolidated Order Book (COB) Provider This script demonstrates the functionality of the new multi-exchange COB data provider: 1. Real-time order book aggregation from multiple exchanges 2. Fine-grain price bucket generation 3. CNN/DQN feature generation 4. Dashboard integration 5. Market analysis and signal generation Run this to test the COB provider with live data streams. """ import asyncio import logging import time from datetime import datetime from core.multi_exchange_cob_provider import MultiExchangeCOBProvider from core.cob_integration import COBIntegration from core.data_provider import DataProvider # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) class COBTester: """Test harness for Multi-Exchange COB Provider""" def __init__(self): self.symbols = ['BTC/USDT', 'ETH/USDT'] self.data_provider = None self.cob_integration = None self.test_duration = 300 # 5 minutes # Statistics tracking self.stats = { 'cob_updates_received': 0, 'bucket_updates_received': 0, 'cnn_features_generated': 0, 'dqn_features_generated': 0, 'signals_generated': 0, 'start_time': None } async def run_test(self): """Run comprehensive COB provider test""" logger.info("Starting Multi-Exchange COB Provider Test") logger.info(f"Testing symbols: {self.symbols}") logger.info(f"Test duration: {self.test_duration} seconds") try: # Initialize components await self._initialize_components() # Run test scenarios await self._run_basic_functionality_test() await self._run_feature_generation_test() await self._run_dashboard_integration_test() await self._run_signal_analysis_test() # Monitor for specified duration await self._monitor_live_data() # Generate final report self._generate_test_report() except Exception as e: logger.error(f"Test failed: {e}") finally: await self._cleanup() async def _initialize_components(self): """Initialize COB provider and integration components""" logger.info("Initializing COB components...") # Create data provider (optional - for integration testing) self.data_provider = DataProvider(symbols=self.symbols) # Create COB integration self.cob_integration = COBIntegration( data_provider=self.data_provider, symbols=self.symbols ) # Register test callbacks self.cob_integration.add_cnn_callback(self._cnn_callback) self.cob_integration.add_dqn_callback(self._dqn_callback) self.cob_integration.add_dashboard_callback(self._dashboard_callback) # Start COB integration await self.cob_integration.start() # Allow time for connections await asyncio.sleep(5) self.stats['start_time'] = datetime.now() logger.info("COB components initialized successfully") async def _run_basic_functionality_test(self): """Test basic COB provider functionality""" logger.info("Testing basic COB functionality...") # Wait for order book data await asyncio.sleep(10) for symbol in self.symbols: # Test consolidated order book retrieval cob_snapshot = self.cob_integration.get_cob_snapshot(symbol) if cob_snapshot: logger.info(f"{symbol} COB Status:") logger.info(f" Exchanges active: {cob_snapshot.exchanges_active}") logger.info(f" Volume weighted mid: ${cob_snapshot.volume_weighted_mid:.2f}") logger.info(f" Spread: {cob_snapshot.spread_bps:.2f} bps") logger.info(f" Bid liquidity: ${cob_snapshot.total_bid_liquidity:,.0f}") logger.info(f" Ask liquidity: ${cob_snapshot.total_ask_liquidity:,.0f}") logger.info(f" Liquidity imbalance: {cob_snapshot.liquidity_imbalance:.3f}") # Test price buckets price_buckets = self.cob_integration.get_price_buckets(symbol) if price_buckets: bid_buckets = len(price_buckets.get('bids', {})) ask_buckets = len(price_buckets.get('asks', {})) logger.info(f" Price buckets: {bid_buckets} bids, {ask_buckets} asks") # Test exchange breakdown exchange_breakdown = self.cob_integration.get_exchange_breakdown(symbol) if exchange_breakdown: logger.info(f" Exchange breakdown:") for exchange, data in exchange_breakdown.items(): market_share = data.get('market_share', 0) * 100 logger.info(f" {exchange}: {market_share:.1f}% market share") else: logger.warning(f"No COB data available for {symbol}") logger.info("Basic functionality test completed") async def _run_feature_generation_test(self): """Test CNN and DQN feature generation""" logger.info("Testing feature generation...") for symbol in self.symbols: # Test CNN features cnn_features = self.cob_integration.get_cob_features(symbol) if cnn_features is not None: logger.info(f"{symbol} CNN features: shape={cnn_features.shape}, " f"min={cnn_features.min():.4f}, max={cnn_features.max():.4f}") else: logger.warning(f"No CNN features available for {symbol}") # Test market depth analysis depth_analysis = self.cob_integration.get_market_depth_analysis(symbol) if depth_analysis: logger.info(f"{symbol} Market Depth Analysis:") logger.info(f" Depth levels: {depth_analysis['depth_analysis']['bid_levels']} bids, " f"{depth_analysis['depth_analysis']['ask_levels']} asks") dominant_exchanges = depth_analysis['depth_analysis'].get('dominant_exchanges', {}) logger.info(f" Dominant exchanges: {dominant_exchanges}") logger.info("Feature generation test completed") async def _run_dashboard_integration_test(self): """Test dashboard data generation""" logger.info("Testing dashboard integration...") # Dashboard integration is tested via callbacks # Statistics are tracked in the callback functions await asyncio.sleep(5) logger.info("Dashboard integration test completed") async def _run_signal_analysis_test(self): """Test signal generation and analysis""" logger.info("Testing signal analysis...") for symbol in self.symbols: # Get recent signals recent_signals = self.cob_integration.get_recent_signals(symbol, count=10) logger.info(f"{symbol} recent signals: {len(recent_signals)} generated") for signal in recent_signals[-3:]: # Show last 3 signals logger.info(f" Signal: {signal.get('type')} - {signal.get('side')} - " f"Confidence: {signal.get('confidence', 0):.3f}") logger.info("Signal analysis test completed") async def _monitor_live_data(self): """Monitor live data for the specified duration""" logger.info(f"Monitoring live data for {self.test_duration} seconds...") start_time = time.time() last_stats_time = start_time while time.time() - start_time < self.test_duration: # Print periodic statistics current_time = time.time() if current_time - last_stats_time >= 30: # Every 30 seconds self._print_periodic_stats() last_stats_time = current_time await asyncio.sleep(1) logger.info("Live data monitoring completed") def _print_periodic_stats(self): """Print periodic statistics during monitoring""" elapsed = (datetime.now() - self.stats['start_time']).total_seconds() logger.info("Periodic Statistics:") logger.info(f" Elapsed time: {elapsed:.0f} seconds") logger.info(f" COB updates: {self.stats['cob_updates_received']}") logger.info(f" Bucket updates: {self.stats['bucket_updates_received']}") logger.info(f" CNN features: {self.stats['cnn_features_generated']}") logger.info(f" DQN features: {self.stats['dqn_features_generated']}") logger.info(f" Signals: {self.stats['signals_generated']}") # Calculate rates if elapsed > 0: cob_rate = self.stats['cob_updates_received'] / elapsed logger.info(f" COB update rate: {cob_rate:.2f}/sec") def _generate_test_report(self): """Generate final test report""" elapsed = (datetime.now() - self.stats['start_time']).total_seconds() logger.info("=" * 60) logger.info("MULTI-EXCHANGE COB PROVIDER TEST REPORT") logger.info("=" * 60) logger.info(f"Test Duration: {elapsed:.0f} seconds") logger.info(f"Symbols Tested: {', '.join(self.symbols)}") logger.info("") # Data Reception Statistics logger.info("Data Reception:") logger.info(f" COB Updates Received: {self.stats['cob_updates_received']}") logger.info(f" Bucket Updates Received: {self.stats['bucket_updates_received']}") logger.info(f" Average COB Rate: {self.stats['cob_updates_received'] / elapsed:.2f}/sec") logger.info("") # Feature Generation Statistics logger.info("Feature Generation:") logger.info(f" CNN Features Generated: {self.stats['cnn_features_generated']}") logger.info(f" DQN Features Generated: {self.stats['dqn_features_generated']}") logger.info("") # Signal Generation Statistics logger.info("Signal Analysis:") logger.info(f" Signals Generated: {self.stats['signals_generated']}") logger.info("") # Component Statistics cob_stats = self.cob_integration.get_statistics() logger.info("Component Statistics:") logger.info(f" Active Exchanges: {', '.join(cob_stats.get('active_exchanges', []))}") logger.info(f" Streaming Status: {cob_stats.get('is_streaming', False)}") logger.info(f" Bucket Size: {cob_stats.get('bucket_size_bps', 0)} bps") logger.info(f" Average Processing Time: {cob_stats.get('avg_processing_time_ms', 0):.2f} ms") logger.info("") # Per-Symbol Analysis logger.info("Per-Symbol Analysis:") for symbol in self.symbols: cob_snapshot = self.cob_integration.get_cob_snapshot(symbol) if cob_snapshot: logger.info(f" {symbol}:") logger.info(f" Active Exchanges: {len(cob_snapshot.exchanges_active)}") logger.info(f" Spread: {cob_snapshot.spread_bps:.2f} bps") logger.info(f" Total Liquidity: ${(cob_snapshot.total_bid_liquidity + cob_snapshot.total_ask_liquidity):,.0f}") recent_signals = self.cob_integration.get_recent_signals(symbol) logger.info(f" Signals Generated: {len(recent_signals)}") logger.info("=" * 60) logger.info("Test completed successfully!") async def _cleanup(self): """Cleanup resources""" logger.info("Cleaning up resources...") if self.cob_integration: await self.cob_integration.stop() if self.data_provider and hasattr(self.data_provider, 'stop_real_time_streaming'): await self.data_provider.stop_real_time_streaming() logger.info("Cleanup completed") # Callback functions for testing def _cnn_callback(self, symbol: str, data: dict): """CNN feature callback for testing""" self.stats['cnn_features_generated'] += 1 if self.stats['cnn_features_generated'] % 100 == 0: logger.debug(f"CNN features generated: {self.stats['cnn_features_generated']}") def _dqn_callback(self, symbol: str, data: dict): """DQN feature callback for testing""" self.stats['dqn_features_generated'] += 1 if self.stats['dqn_features_generated'] % 100 == 0: logger.debug(f"DQN features generated: {self.stats['dqn_features_generated']}") def _dashboard_callback(self, symbol: str, data: dict): """Dashboard data callback for testing""" self.stats['cob_updates_received'] += 1 # Check for signals in dashboard data signals = data.get('recent_signals', []) self.stats['signals_generated'] += len(signals) async def main(): """Main test function""" logger.info("Multi-Exchange COB Provider Test Starting...") try: tester = COBTester() await tester.run_test() except KeyboardInterrupt: logger.info("Test interrupted by user") except Exception as e: logger.error(f"Test failed with error: {e}") raise if __name__ == "__main__": asyncio.run(main())