fix: Address critical PR review issues

- Fixed JWT token exposure by moving tokens from URLs to headers
  - Updated SignalR connection to use Authorization header
  - Removed access_token from query parameters

- Replaced broad exception catching with specific types
  - Updated memory_management.py to handle specific exceptions
  - Updated core.py to catch ProjectX-specific exceptions
  - Added proper handling for asyncio.CancelledError

- Added comprehensive input validation for trading calculations
  - Added validation for all numeric parameters
  - Proper type checking for required types
  - Clear error messages for invalid inputs

These changes address the security vulnerabilities and improve error handling
as requested in the PR review.

Author: TexasCoding

src/project_x_py/realtime/connection_management.py

@@ -57,10 +57,15 @@ async def setup_connections(self: "ProjectXRealtimeClientProtocol") -> None:
                 raise ImportError("signalrcore is required for real-time functionality")
 
             async with self._connection_lock:
-                # Build user hub connection
+                # Build user hub connection with JWT in headers
                 self.user_connection = (
                     HubConnectionBuilder()
-                    .with_url(self.user_hub_url)
+                    .with_url(
+                        self.user_hub_url,
+                        options={
+                            "headers": {"Authorization": f"Bearer {self.jwt_token}"}
+                        },
+                    )
                     .configure_logging(
                         logging.INFO,
                         socket_trace=False,
@@ -76,10 +81,15 @@ async def setup_connections(self: "ProjectXRealtimeClientProtocol") -> None:
                     .build()
                 )
 
-                # Build market hub connection
+                # Build market hub connection with JWT in headers
                 self.market_connection = (
                     HubConnectionBuilder()
-                    .with_url(self.market_hub_url)
+                    .with_url(
+                        self.market_hub_url,
+                        options={
+                            "headers": {"Authorization": f"Bearer {self.jwt_token}"}
+                        },
+                    )
                     .configure_logging(
                         logging.INFO,
                         socket_trace=False,
@@ -425,13 +435,9 @@ async def update_jwt_token(
         # Disconnect existing connections
         await self.disconnect()
 
-        # Update token
+        # Update JWT token for header authentication
         self.jwt_token = new_jwt_token
 
-        # Update URLs with new token
-        self.user_hub_url = f"{self.base_user_url}?access_token={new_jwt_token}"
-        self.market_hub_url = f"{self.base_market_url}?access_token={new_jwt_token}"
-
         # Reset setup flag to force new connection setup
         self.setup_complete = False
 

src/project_x_py/realtime/core.py

@@ -133,7 +133,7 @@ def __init__(
             ... )
 
         Note:
-            - JWT token is appended as access_token query parameter
+            - JWT token is passed securely via Authorization header
             - Both hubs must connect successfully for full functionality
             - SignalR connections are established lazily on connect()
         """
@@ -152,9 +152,9 @@ def __init__(
         final_user_url = user_hub_url or default_user_url
         final_market_url = market_hub_url or default_market_url
 
-        # Build complete URLs with authentication
-        self.user_hub_url = f"{final_user_url}?access_token={jwt_token}"
-        self.market_hub_url = f"{final_market_url}?access_token={jwt_token}"
+        # Store URLs without tokens (tokens will be passed in headers)
+        self.user_hub_url = final_user_url
+        self.market_hub_url = final_market_url
 
         # Set up base URLs for token refresh
         if config:

src/project_x_py/realtime_data_manager/core.py

@@ -15,6 +15,7 @@
 import polars as pl
 import pytz
 
+from ..exceptions import ProjectXDataError, ProjectXError, ProjectXInstrumentError
 from .callbacks import CallbackMixin
 from .data_access import DataAccessMixin
 from .data_processing import DataProcessingMixin
@@ -350,8 +351,14 @@ async def initialize(self, initial_days: int = 1) -> bool:
             )
             return True
 
-        except Exception as e:
-            self.logger.error(f"❌ Failed to initialize: {e}")
+        except ProjectXInstrumentError as e:
+            self.logger.error(f"❌ Failed to initialize - instrument error: {e}")
+            return False
+        except ProjectXDataError as e:
+            self.logger.error(f"❌ Failed to initialize - data error: {e}")
+            return False
+        except ProjectXError as e:
+            self.logger.error(f"❌ Failed to initialize - ProjectX error: {e}")
             return False
 
     async def start_realtime_feed(self) -> bool:
@@ -447,8 +454,11 @@ async def on_new_bar(data):
             self.logger.info(f"✅ Real-time OHLCV feed started for {self.instrument}")
             return True
 
-        except Exception as e:
-            self.logger.error(f"❌ Failed to start real-time feed: {e}")
+        except RuntimeError as e:
+            self.logger.error(f"❌ Failed to start real-time feed - runtime error: {e}")
+            return False
+        except asyncio.TimeoutError as e:
+            self.logger.error(f"❌ Failed to start real-time feed - timeout: {e}")
             return False
 
     async def stop_realtime_feed(self) -> None:
@@ -474,7 +484,7 @@ async def stop_realtime_feed(self) -> None:
 
             self.logger.info(f"✅ Real-time feed stopped for {self.instrument}")
 
-        except Exception as e:
+        except RuntimeError as e:
             self.logger.error(f"❌ Error stopping real-time feed: {e}")
 
     async def cleanup(self) -> None:

src/project_x_py/realtime_data_manager/memory_management.py

@@ -72,8 +72,19 @@ async def _periodic_cleanup(self: "RealtimeDataManagerProtocol") -> None:
             try:
                 await asyncio.sleep(self.cleanup_interval)
                 await self._cleanup_old_data()
-            except Exception as e:
-                self.logger.error(f"Error in periodic cleanup: {e}")
+            except asyncio.CancelledError:
+                # Task cancellation is expected during shutdown
+                self.logger.debug("Periodic cleanup task cancelled")
+                raise
+            except MemoryError as e:
+                self.logger.error(f"Memory error during cleanup: {e}")
+                # Force immediate garbage collection
+                import gc
+
+                gc.collect()
+            except RuntimeError as e:
+                self.logger.error(f"Runtime error in periodic cleanup: {e}")
+                # Don't re-raise runtime errors to keep the cleanup task running
 
     def get_memory_stats(self: "RealtimeDataManagerProtocol") -> dict:
         """

src/project_x_py/utils/trading_calculations.py

@@ -22,8 +22,13 @@ def calculate_tick_value(
         >>> calculate_tick_value(0.5, 0.1, 1.0)
         5.0
     """
+    # Validate inputs
     if tick_size <= 0:
-        return 0.0
+        raise ValueError(f"tick_size must be positive, got {tick_size}")
+    if tick_value < 0:
+        raise ValueError(f"tick_value cannot be negative, got {tick_value}")
+    if not isinstance(price_change, (int, float)):
+        raise TypeError(f"price_change must be numeric, got {type(price_change)}")
 
     num_ticks = abs(price_change) / tick_size
     return num_ticks * tick_value
@@ -49,8 +54,15 @@ def calculate_position_value(
         >>> calculate_position_value(5, 2050.0, 1.0, 0.1)
         102500.0
     """
+    # Validate inputs
     if tick_size <= 0:
-        return 0.0
+        raise ValueError(f"tick_size must be positive, got {tick_size}")
+    if tick_value < 0:
+        raise ValueError(f"tick_value cannot be negative, got {tick_value}")
+    if price < 0:
+        raise ValueError(f"price cannot be negative, got {price}")
+    if not isinstance(size, int):
+        raise TypeError(f"size must be an integer, got {type(size)}")
 
     ticks_per_point = 1.0 / tick_size
     value_per_point = ticks_per_point * tick_value
@@ -68,12 +80,18 @@ def round_to_tick_size(price: float, tick_size: float) -> float:
     Returns:
         float: Price rounded to nearest tick
 
+    Raises:
+        ValueError: If tick_size is not positive or price is negative
+
     Example:
         >>> round_to_tick_size(2050.37, 0.1)
         2050.4
     """
+    # Validate inputs
     if tick_size <= 0:
-        return price
+        raise ValueError(f"tick_size must be positive, got {tick_size}")
+    if price < 0:
+        raise ValueError(f"price cannot be negative, got {price}")
 
     return round(price / tick_size) * tick_size
 
@@ -142,6 +160,20 @@ def calculate_position_sizing(
         >>> sizing = calculate_position_sizing(50000, 0.02, 2050, 2040, 1.0)
         >>> print(f"Position size: {sizing['position_size']} contracts")
     """
+    # Validate inputs
+    if account_balance <= 0:
+        raise ValueError(f"account_balance must be positive, got {account_balance}")
+    if not 0 < risk_per_trade <= 1:
+        raise ValueError(
+            f"risk_per_trade must be between 0 and 1, got {risk_per_trade}"
+        )
+    if entry_price <= 0:
+        raise ValueError(f"entry_price must be positive, got {entry_price}")
+    if stop_loss_price <= 0:
+        raise ValueError(f"stop_loss_price must be positive, got {stop_loss_price}")
+    if tick_value <= 0:
+        raise ValueError(f"tick_value must be positive, got {tick_value}")
+
     try:
         # Calculate risk per share/contract
         price_risk = abs(entry_price - stop_loss_price)