cleanup, cob ladder still broken

core/multi_exchange_cob_provider.py

@@ -33,7 +33,7 @@ except ImportError:
 import numpy as np
 import pandas as pd
 from datetime import datetime, timedelta
-from typing import Dict, List, Optional, Tuple, Any, Callable, Union
+from typing import Dict, List, Optional, Tuple, Any, Callable, Union, Awaitable
 from collections import deque, defaultdict
 from dataclasses import dataclass, field
 from threading import Thread, Lock
@@ -194,6 +194,11 @@ class MultiExchangeCOBProvider:
         # Thread safety
         self.data_lock = asyncio.Lock()
         
+        # Initialize aiohttp session and connector to None, will be set up in start_streaming
+        self.session: Optional[aiohttp.ClientSession] = None
+        self.connector: Optional[aiohttp.TCPConnector] = None
+        self.rest_session: Optional[aiohttp.ClientSession] = None # Added for explicit None initialization
+        
         # Create REST API session
         # Fix for Windows aiodns issue - use ThreadedResolver instead
         connector = aiohttp.TCPConnector(
@@ -286,64 +291,62 @@ class MultiExchangeCOBProvider:
         return configs
     
     async def start_streaming(self):
-        """Start streaming from all configured exchanges"""
-        if self.is_streaming:
-            logger.warning("COB streaming already active")
-            return
-        
-        logger.info("Starting Multi-Exchange COB streaming")
+        """Start real-time order book streaming from all configured exchanges"""
+        logger.info(f"Starting COB streaming for symbols: {self.symbols}")
         self.is_streaming = True
         
-        # Start streaming tasks for each exchange and symbol
+        # Setup aiohttp session here, within the async context
+        await self._setup_http_session()
+
+        # Start WebSocket connections for each active exchange and symbol
         tasks = []
-        
-        for exchange_name in self.active_exchanges:
-            for symbol in self.symbols:
-                # WebSocket task for real-time top 20 levels
-                task = asyncio.create_task(
-                    self._stream_exchange_orderbook(exchange_name, symbol)
-                )
-                tasks.append(task)
-                
-                # REST API task for deep order book snapshots
-                deep_task = asyncio.create_task(
-                    self._stream_deep_orderbook(exchange_name, symbol)
-                )
-                tasks.append(deep_task)
-                
-                # Trade stream task for SVP
-                if exchange_name == 'binance':
-                    trade_task = asyncio.create_task(
-                        self._stream_binance_trades(symbol)
-                    )
-                    tasks.append(trade_task)
-        
-        # Start consolidation and analysis tasks
-        tasks.extend([
-            asyncio.create_task(self._continuous_consolidation()),
-            asyncio.create_task(self._continuous_bucket_updates())
-        ])
-        
-        # Wait for all tasks
-        try:
-            await asyncio.gather(*tasks)
-        except Exception as e:
-            logger.error(f"Error in streaming tasks: {e}")
-        finally:
-            self.is_streaming = False
+        for symbol in self.symbols:
+            for exchange_name, config in self.exchange_configs.items():
+                if config.enabled and exchange_name in self.active_exchanges:
+                    # Start WebSocket stream
+                    tasks.append(self._stream_exchange_orderbook(exchange_name, symbol))
+                    
+                    # Start deep order book (REST API) stream
+                    tasks.append(self._stream_deep_orderbook(exchange_name, symbol))
+                    
+                    # Start trade stream (for SVP)
+                    if exchange_name == 'binance': # Only Binance for now
+                        tasks.append(self._stream_binance_trades(symbol))
+
+        # Start continuous consolidation and bucket updates
+        tasks.append(self._continuous_consolidation())
+        tasks.append(self._continuous_bucket_updates())
+
+        logger.info(f"Starting {len(tasks)} COB streaming tasks")
+        await asyncio.gather(*tasks)
+
+    async def _setup_http_session(self):
+        """Setup aiohttp session and connector"""
+        self.connector = aiohttp.TCPConnector(
+            resolver=aiohttp.ThreadedResolver() # This is now created inside async function
+        )
+        self.session = aiohttp.ClientSession(connector=self.connector)
+        self.rest_session = aiohttp.ClientSession(connector=self.connector) # Moved here from __init__
+        logger.info("aiohttp session and connector setup completed")
 
     async def stop_streaming(self):
-        """Stop streaming from all exchanges"""
-        logger.info("Stopping Multi-Exchange COB streaming")
+        """Stop real-time order book streaming and close sessions"""
+        logger.info("Stopping COB Integration")
         self.is_streaming = False
-        
-        # Close REST API session
-        if self.rest_session:
+
+        if self.session and not self.session.closed:
+            await self.session.close()
+            logger.info("aiohttp session closed")
+
+        if self.rest_session and not self.rest_session.closed:
             await self.rest_session.close()
-            self.rest_session = None
-        
-        # Wait a bit for tasks to stop gracefully
-        await asyncio.sleep(1)
+            logger.info("aiohttp REST session closed")
+
+        if self.connector and not self.connector.closed:
+            await self.connector.close()
+            logger.info("aiohttp connector closed")
+
+        logger.info("COB Integration stopped")
 
     async def _stream_deep_orderbook(self, exchange_name: str, symbol: str):
         """Fetch deep order book data via REST API periodically"""
@@ -1086,12 +1089,12 @@ class MultiExchangeCOBProvider:
     
     # Public interface methods
     
-    def subscribe_to_cob_updates(self, callback: Callable[[str, COBSnapshot], None]):
+    def subscribe_to_cob_updates(self, callback: Callable[[str, COBSnapshot], Awaitable[None]]):
         """Subscribe to consolidated order book updates"""
         self.cob_update_callbacks.append(callback)
         logger.info(f"Added COB update callback: {len(self.cob_update_callbacks)} total")
     
-    def subscribe_to_bucket_updates(self, callback: Callable[[str, Dict], None]):
+    def subscribe_to_bucket_updates(self, callback: Callable[[str, Dict], Awaitable[None]]):
         """Subscribe to price bucket updates"""
         self.bucket_update_callbacks.append(callback)
         logger.info(f"Added bucket update callback: {len(self.bucket_update_callbacks)} total")