fix: address PR review feedback for v3.1.0

## Security & Stability Improvements
- Add path validation to prevent directory traversal attacks
- Set secure file permissions (0700) for mmap storage directories
- Improve error handling with circuit breaker for batch processing
- Add proper resource cleanup with __del__ for mmap files

## Configuration Enhancements
- Make overflow threshold configurable (default 80%)
- Make compression threshold configurable (default 1KB)
- Add mmap_storage_path configuration option
- Add mmap_cleanup_days for automatic old file removal

## Error Handling Improvements
- Better exception handling in batch processing
- Add circuit breaker after 10 consecutive failures
- Properly re-raise asyncio.CancelledError for clean shutdown
- Add detailed logging with exc_info for debugging

## Resource Management
- Implement proper cleanup in __del__ for mmap storage
- Add automatic cleanup of old overflow files (>7 days)
- Close all storage instances properly on cleanup
- Use contextlib.suppress for cleaner exception handling

🤖 Generated with Claude Code

Co-Authored-By: Claude <noreply@anthropic.com>

Author: TexasCoding

src/project_x_py/client/cache.py

@@ -57,9 +57,13 @@ def __init__(self) -> None:
         )
         self._opt_market_data_cache_time: dict[str, float] = {}
 
-        # Compression settings
-        self.compression_threshold = 1024  # Compress data > 1KB
-        self.compression_level = 3  # lz4 compression level (0-16)
+        # Compression settings (configurable)
+        self.compression_threshold = getattr(self, "config", {}).get(
+            "compression_threshold", 1024
+        )  # Compress data > 1KB
+        self.compression_level = getattr(self, "config", {}).get(
+            "compression_level", 3
+        )  # lz4 compression level (0-16)
 
     def _serialize_dataframe(self, df: pl.DataFrame) -> bytes:
         """

src/project_x_py/realtime/batched_handler.py

@@ -109,10 +109,21 @@ async def _process_batch(self) -> None:
                 if self.process_callback:
                     try:
                         await self.process_callback(batch)
+                    except asyncio.CancelledError:
+                        # Re-raise cancellation for proper shutdown
+                        raise
                     except Exception as e:
                         logger.error(
-                            f"Error processing batch of {len(batch)} messages: {e}"
+                            f"Error processing batch of {len(batch)} messages: {e}",
+                            exc_info=True,
                         )
+                        # Track failures for circuit breaker
+                        self.failed_batches = getattr(self, "failed_batches", 0) + 1
+                        if self.failed_batches > 10:
+                            logger.critical(
+                                "Batch processing circuit breaker triggered"
+                            )
+                            self.processing = False
 
                 # Update metrics
                 processing_time = time.time() - start_time

src/project_x_py/realtime_data_manager/mmap_overflow.py

@@ -6,7 +6,8 @@
 maintaining fast access to recent data.
 """
 
-from datetime import datetime
+from contextlib import suppress
+from datetime import datetime, timedelta
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
@@ -41,11 +42,31 @@ def __init__(self) -> None:
         """Initialize memory-mapped overflow storage."""
         super().__init__()
 
-        # Storage configuration
-        self.enable_mmap_overflow = True
-        self.overflow_threshold = 0.8  # Start overflow at 80% of max bars
-        self.mmap_storage_path = Path.home() / ".projectx" / "data_overflow"
-        self.mmap_storage_path.mkdir(parents=True, exist_ok=True)
+        # Storage configuration (can be overridden via config)
+        self.enable_mmap_overflow = getattr(self, "config", {}).get(
+            "enable_mmap_overflow", True
+        )
+        self.overflow_threshold = getattr(self, "config", {}).get(
+            "overflow_threshold", 0.8
+        )  # Start overflow at 80% of max bars
+
+        # Validate and create storage path
+        base_path = getattr(self, "config", {}).get(
+            "mmap_storage_path", Path.home() / ".projectx" / "data_overflow"
+        )
+        self.mmap_storage_path = Path(base_path)
+
+        # Validate path to prevent directory traversal
+        try:
+            self.mmap_storage_path = self.mmap_storage_path.resolve()
+            if not str(self.mmap_storage_path).startswith(str(Path.home())):
+                raise ValueError(f"Invalid storage path: {self.mmap_storage_path}")
+            self.mmap_storage_path.mkdir(
+                parents=True, exist_ok=True, mode=0o700
+            )  # Secure permissions
+        except Exception as e:
+            logger.error(f"Failed to create overflow storage directory: {e}")
+            self.enable_mmap_overflow = False
 
         # Storage instances per timeframe
         self._mmap_storages: dict[str, MemoryMappedStorage] = {}
@@ -260,19 +281,43 @@ async def get_historical_data(
     async def cleanup_overflow_storage(self) -> None:
         """Clean up old overflow files and close storage instances."""
         try:
-            # Close all storage instances
-            for storage in self._mmap_storages.values():
-                storage.close()
+            # Close all storage instances properly
+            for timeframe, storage in list(self._mmap_storages.items()):
+                try:
+                    storage.close()
+                    logger.debug(f"Closed mmap storage for {timeframe}")
+                except Exception as e:
+                    logger.warning(f"Error closing storage for {timeframe}: {e}")
             self._mmap_storages.clear()
 
-            # Clean up old files (optional - keep last N days)
-            # This could be configured based on requirements
+            # Clean up old files based on config
+            cleanup_days = getattr(self, "config", {}).get("mmap_cleanup_days", 7)
+            if cleanup_days > 0:
+                cutoff_time = datetime.now() - timedelta(days=cleanup_days)
+                try:
+                    for file_path in self.mmap_storage_path.glob("*.mmap"):
+                        if file_path.stat().st_mtime < cutoff_time.timestamp():
+                            file_path.unlink()
+                            # Also remove metadata file
+                            meta_path = file_path.with_suffix(".meta")
+                            if meta_path.exists():
+                                meta_path.unlink()
+                            logger.info(f"Removed old overflow file: {file_path.name}")
+                except Exception as e:
+                    logger.warning(f"Error cleaning old files: {e}")
 
             logger.info("Cleaned up overflow storage")
 
         except Exception as e:
             logger.error(f"Error cleaning up overflow storage: {e}")
 
+    def __del__(self) -> None:
+        """Ensure cleanup on deletion."""
+        # Synchronous cleanup for destructor
+        for storage in self._mmap_storages.values():
+            with suppress(Exception):
+                storage.close()
+
     def get_overflow_stats(self) -> dict[str, Any]:
         """
         Get statistics about overflow storage.