fix(realtime): Add proper type annotations for SignalR hub connections

- Created HubConnection type alias in types.base for BaseHubConnection
- Updated ProjectXRealtimeClient to use HubConnection instead of Any
- Updated ProjectXRealtimeClientProtocol to match implementation types
- Improved subscription handling with proper event waiting
- Added circuit breaker pattern to BatchedWebSocketHandler
- Fixed asyncio deprecation warnings (get_event_loop -> get_running_loop)

This ensures IDEs properly recognize connection object methods (send, on, start, stop)
and provides better type safety for real-time connection management.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <[email protected]>

Author: TexasCoding

src/project_x_py/realtime/batched_handler.py

@@ -62,6 +62,11 @@ def __init__(
         self.total_processing_time = 0.0
         self.last_batch_time = time.time()
 
+        # Circuit breaker state
+        self.failed_batches = 0
+        self.circuit_breaker_tripped_at: float | None = None
+        self.circuit_breaker_timeout = 60.0  # 60 seconds
+
         # Lock for thread safety
         self._lock = asyncio.Lock()
 
@@ -89,6 +94,18 @@ async def _process_batch(self) -> None:
         async with self._lock:
             if self.processing:
                 return
+
+            # Check circuit breaker
+            if self.circuit_breaker_tripped_at:
+                if (
+                    time.time() - self.circuit_breaker_tripped_at
+                ) > self.circuit_breaker_timeout:
+                    logger.warning("Resetting circuit breaker.")
+                    self.circuit_breaker_tripped_at = None
+                    self.failed_batches = 0
+                else:
+                    return  # Circuit breaker is tripped, do not process
+
             self.processing = True
 
         try:
@@ -136,6 +153,8 @@ async def _process_batch(self) -> None:
                 if self.process_callback:
                     try:
                         await self.process_callback(batch)
+                        # Reset failure count on success
+                        self.failed_batches = 0
                     except asyncio.CancelledError:
                         # Re-raise cancellation for proper shutdown
                         raise
@@ -145,12 +164,14 @@ async def _process_batch(self) -> None:
                             exc_info=True,
                         )
                         # Track failures for circuit breaker
-                        self.failed_batches = getattr(self, "failed_batches", 0) + 1
+                        self.failed_batches += 1
                         if self.failed_batches > 10:
                             logger.critical(
-                                "Batch processing circuit breaker triggered"
+                                f"Batch processing circuit breaker triggered for {self.circuit_breaker_timeout}s."
                             )
+                            self.circuit_breaker_tripped_at = time.time()
                             self.processing = False
+                            return  # Stop processing
 
                 # Update metrics
                 processing_time = time.time() - start_time

src/project_x_py/realtime/connection_management.py

@@ -269,7 +269,11 @@ async def connect(self: "ProjectXRealtimeClientProtocol") -> bool:
                 await self.setup_connections()
 
             # Store the event loop for cross-thread task scheduling
-            self._loop = asyncio.get_event_loop()
+            try:
+                self._loop = asyncio.get_running_loop()
+            except RuntimeError:
+                logger.error("No running event loop found.")
+                return False
 
             logger.debug(LogMessages.WS_CONNECT)
 
@@ -293,8 +297,22 @@ async def connect(self: "ProjectXRealtimeClientProtocol") -> bool:
                     )
                     return False
 
-                # Wait for connections to establish and stabilize
-                await asyncio.sleep(2.0)  # Increased wait time for connection stability
+                # Wait for connections to establish
+                try:
+                    await asyncio.wait_for(
+                        asyncio.gather(
+                            self.user_hub_ready.wait(), self.market_hub_ready.wait()
+                        ),
+                        timeout=10.0,
+                    )
+                except TimeoutError:
+                    logger.error(
+                        LogMessages.WS_ERROR,
+                        extra={
+                            "error": "Connection attempt timed out after 10 seconds."
+                        },
+                    )
+                    return False
 
                 if self.user_connected and self.market_connected:
                     self.stats["connected_time"] = datetime.now()
@@ -325,7 +343,7 @@ async def _start_connection_async(
             This is an internal method that bridges sync SignalR with async code.
         """
         # SignalR connections are synchronous, so we run them in executor
-        loop = asyncio.get_event_loop()
+        loop = asyncio.get_running_loop()
         await loop.run_in_executor(None, connection.start)
         logger.debug(LogMessages.WS_CONNECTED, extra={"hub": name})
 
@@ -365,13 +383,12 @@ async def disconnect(self: "ProjectXRealtimeClientProtocol") -> None:
             logger.debug(LogMessages.WS_DISCONNECT)
 
             async with self._connection_lock:
+                loop = asyncio.get_running_loop()
                 if self.user_connection:
-                    loop = asyncio.get_event_loop()
                     await loop.run_in_executor(None, self.user_connection.stop)
                     self.user_connected = False
 
                 if self.market_connection:
-                    loop = asyncio.get_event_loop()
                     await loop.run_in_executor(None, self.market_connection.stop)
                     self.market_connected = False
 
@@ -390,6 +407,7 @@ def _on_user_hub_open(self: "ProjectXRealtimeClientProtocol") -> None:
             - Logs connection success
         """
         self.user_connected = True
+        self.user_hub_ready.set()
         self.logger.info("✅ User hub connected")
 
     def _on_user_hub_close(self: "ProjectXRealtimeClientProtocol") -> None:
@@ -407,6 +425,7 @@ def _on_user_hub_close(self: "ProjectXRealtimeClientProtocol") -> None:
             Automatic reconnection will attempt based on configuration.
         """
         self.user_connected = False
+        self.user_hub_ready.clear()
         self.logger.warning("❌ User hub disconnected")
 
     def _on_market_hub_open(self: "ProjectXRealtimeClientProtocol") -> None:
@@ -421,6 +440,7 @@ def _on_market_hub_open(self: "ProjectXRealtimeClientProtocol") -> None:
             - Logs connection success
         """
         self.market_connected = True
+        self.market_hub_ready.set()
         self.logger.info("✅ Market hub connected")
 
     def _on_market_hub_close(self: "ProjectXRealtimeClientProtocol") -> None:
@@ -438,6 +458,7 @@ def _on_market_hub_close(self: "ProjectXRealtimeClientProtocol") -> None:
             Automatic reconnection will attempt based on configuration.
         """
         self.market_connected = False
+        self.market_hub_ready.clear()
         self.logger.warning("❌ Market hub disconnected")
 
     def _on_connection_error(

src/project_x_py/realtime/core.py

@@ -100,6 +100,7 @@ async def main():
 from project_x_py.realtime.connection_management import ConnectionManagementMixin
 from project_x_py.realtime.event_handling import EventHandlingMixin
 from project_x_py.realtime.subscriptions import SubscriptionsMixin
+from project_x_py.types.base import HubConnection
 
 if TYPE_CHECKING:
     from project_x_py.models import ProjectXConfig
@@ -285,17 +286,16 @@ def __init__(
             self.base_market_url = "https://rtc.topstepx.com/hubs/market"
 
         # SignalR connection objects
-        self.user_connection: Any | None = None
-        self.market_connection: Any | None = None
+        self.user_connection: HubConnection | None = None
+        self.market_connection: HubConnection | None = None
 
         # Connection state tracking
         self.user_connected = False
         self.market_connected = False
         self.setup_complete = False
 
-        # Event callbacks (pure forwarding, no caching) - already initialized in mixin
-        if not hasattr(self, "callbacks"):
-            self.callbacks: defaultdict[str, list[Any]] = defaultdict(list)
+        # Event callbacks (pure forwarding, no caching)
+        self.callbacks: defaultdict[str, list[Any]] = defaultdict(list)
 
         # Basic statistics (no business logic)
         self.stats = {
@@ -315,8 +315,10 @@ def __init__(
         self.logger.info(f"User Hub: {final_user_url}")
         self.logger.info(f"Market Hub: {final_market_url}")
 
-        # Async locks for thread-safe operations - check if not already initialized
-        if not hasattr(self, "_callback_lock"):
-            self._callback_lock = asyncio.Lock()
-        if not hasattr(self, "_connection_lock"):
-            self._connection_lock = asyncio.Lock()
+        # Async locks for thread-safe operations
+        self._callback_lock = asyncio.Lock()
+        self._connection_lock = asyncio.Lock()
+
+        # Async events for connection readiness
+        self.user_hub_ready = asyncio.Event()
+        self.market_hub_ready = asyncio.Event()

src/project_x_py/realtime/event_handling.py

@@ -430,7 +430,7 @@ def _schedule_async_task(
                 )
             except Exception as e:
                 # Fallback for logging - avoid recursion
-                print(f"Error scheduling async task: {e}")
+                self.logger.error(f"Error scheduling async task: {e}")
         else:
             # Fallback - try to create task in current loop context
             try:
@@ -439,7 +439,7 @@ def _schedule_async_task(
                 task.add_done_callback(lambda t: None)
             except RuntimeError:
                 # No event loop available, log and continue
-                print(f"No event loop available for {event_type} event")
+                self.logger.error(f"No event loop available for {event_type} event")
 
     async def _forward_event_async(
         self: "ProjectXRealtimeClientProtocol", event_type: str, args: Any