all tests passing

Author: TexasCoding

pytest.ini

@@ -15,3 +15,11 @@ log_cli_date_format = %Y-%m-%d %H:%M:%S
 addopts =
     --verbose
     -xvs
+
+# Custom markers
+markers =
+    performance: marks tests as performance tests (deselect with '-m "not performance"')
+    slow: marks tests as slow (deselect with '-m "not slow"')
+    unit: marks tests as unit tests
+    integration: marks tests as integration tests
+    realtime: marks tests that require real-time connections

src/project_x_py/data/mmap_storage.py

@@ -378,11 +378,15 @@ def __init__(
         self.row_size = (len(self.columns) + 1) * self.dtype.itemsize
         self.current_size = 0
 
-        # Determine current size from existing file
+        # Load metadata to get actual data size if file exists
         if self.filename.exists():
             self.open()  # open() is idempotent and thread-safe
-            if self._data_file_size > 0 and self.row_size > 0:
-                self.current_size = self._data_file_size // self.row_size
+            self._load_metadata()
+            # Get current_size from metadata if available
+            if "_timeseries_meta" in self._metadata:
+                self.current_size = self._metadata["_timeseries_meta"].get(
+                    "current_size", 0
+                )
 
     def append_data(self, timestamp: float, values: dict[str, float]) -> bool:
         """
@@ -425,6 +429,10 @@ def append_data(self, timestamp: float, values: dict[str, float]) -> bool:
                     self.mmap.flush()
                 self.current_size += 1
 
+                # Update metadata with current size
+                self._metadata["_timeseries_meta"] = {"current_size": self.current_size}
+                self._save_metadata()
+
                 return True
 
             except Exception:

src/project_x_py/realtime/circuit_breaker.py

@@ -620,7 +620,7 @@ async def configure_circuit_breaker(
         max_recovery_time: float = 300.0,
         slow_call_threshold: float = 2.0,
         enable_global_circuit: bool = True,
-        _enable_per_event_circuits: bool = True,
+        enable_per_event_circuits: bool = True,
     ) -> None:
         """
         Configure circuit breaker settings.

src/project_x_py/realtime_data_manager/core.py

@@ -1083,19 +1083,20 @@ async def cleanup(self) -> None:
                 # EventBus handles all event cleanup
                 self.indicator_cache.clear()
 
-            # Backward-compatible attributes used in some tests/examples
-            # Use dynamic attribute access safely without type checker complaints
-            bars_attr = getattr(self, "bars", None)
-            if isinstance(bars_attr, dict):
-                for _tf in list(bars_attr.keys()):
-                    bars_attr[_tf] = []
-            ticks_attr = getattr(self, "ticks", None)
-            if isinstance(ticks_attr, list):
-                ticks_attr.clear()
-            dom_attr = getattr(self, "dom_data", None)
-            if isinstance(dom_attr, dict):
-                for _k in list(dom_attr.keys()):
-                    dom_attr[_k] = []
+        # Backward-compatible attributes used in some tests/examples
+        # Clear these regardless of lock type
+        # Use dynamic attribute access safely without type checker complaints
+        bars_attr = getattr(self, "bars", None)
+        if isinstance(bars_attr, dict):
+            for _tf in list(bars_attr.keys()):
+                bars_attr[_tf] = []
+        ticks_attr = getattr(self, "ticks", None)
+        if isinstance(ticks_attr, list):
+            ticks_attr.clear()
+        dom_attr = getattr(self, "dom_data", None)
+        if isinstance(dom_attr, dict):
+            for _k in list(dom_attr.keys()):
+                dom_attr[_k] = []
 
         self.logger.info("✅ RealtimeDataManager cleanup completed")
 

src/project_x_py/realtime_data_manager/memory_management.py

@@ -101,11 +101,11 @@
 if TYPE_CHECKING:
     from project_x_py.types.stats_types import RealtimeDataManagerStats
 
+import polars as pl
+
 if TYPE_CHECKING:
     from asyncio import Lock
 
-    import polars as pl
-
 logger = logging.getLogger(__name__)
 
 
@@ -392,58 +392,71 @@ async def _cleanup_old_data(self) -> None:
         if current_time - self.last_cleanup < self.cleanup_interval:
             return
 
-        async with self.data_lock:
-            total_bars_before = 0
-            total_bars_after = 0
-
-            # Cleanup each timeframe's data
-            for tf_key in self.timeframes:
-                if tf_key in self.data and not self.data[tf_key].is_empty():
-                    initial_count = len(self.data[tf_key])
-                    total_bars_before += initial_count
-
-                    # Check for buffer overflow first
-                    is_overflow, utilization = await self._check_buffer_overflow(tf_key)
-                    if is_overflow:
-                        await self._handle_buffer_overflow(tf_key, utilization)
-                        total_bars_after += len(self.data[tf_key])
-                        continue
-
-                    # Check if overflow is needed (if mixin is available)
-                    if hasattr(
-                        self, "_check_overflow_needed"
-                    ) and await self._check_overflow_needed(tf_key):
-                        await self._overflow_to_disk(tf_key)
-                        # Data has been overflowed, update count
-                        total_bars_after += len(self.data[tf_key])
-                        continue
-
-                    # Keep only the most recent bars (sliding window)
-                    if initial_count > self.max_bars_per_timeframe:
-                        self.data[tf_key] = self.data[tf_key].tail(
-                            self.max_bars_per_timeframe
-                        )
+        # Import here to avoid circular dependency
+        from project_x_py.utils.lock_optimization import AsyncRWLock
+
+        # Use appropriate lock method based on lock type
+        if isinstance(self.data_lock, AsyncRWLock):
+            async with self.data_lock.write_lock():
+                await self._perform_cleanup()
+        else:
+            async with self.data_lock:
+                await self._perform_cleanup()
 
+    async def _perform_cleanup(self) -> None:
+        """Perform the actual cleanup logic (extracted for lock handling)."""
+        total_bars_before = 0
+        total_bars_after = 0
+
+        # Cleanup each timeframe's data
+        for tf_key in self.timeframes:
+            if tf_key in self.data and not self.data[tf_key].is_empty():
+                initial_count = len(self.data[tf_key])
+                total_bars_before += initial_count
+
+                # Check for buffer overflow first
+                is_overflow, utilization = await self._check_buffer_overflow(tf_key)
+                if is_overflow:
+                    await self._handle_buffer_overflow(tf_key, utilization)
+                    total_bars_after += len(self.data[tf_key])
+                    continue
+
+                # Check if overflow is needed (if mixin is available)
+                if hasattr(
+                    self, "_check_overflow_needed"
+                ) and await self._check_overflow_needed(tf_key):
+                    await self._overflow_to_disk(tf_key)
+                    # Data has been overflowed, update count
                     total_bars_after += len(self.data[tf_key])
+                    continue
 
-            # Cleanup tick buffer - deque handles its own cleanup with maxlen
-            # No manual cleanup needed for deque with maxlen
+                # Keep only the most recent bars (sliding window)
+                if initial_count > self.max_bars_per_timeframe:
+                    self.data[tf_key] = self.data[tf_key].tail(
+                        self.max_bars_per_timeframe
+                    )
 
-            # Update stats
-            self.last_cleanup = current_time
-            self.memory_stats["bars_cleaned"] += total_bars_before - total_bars_after
-            self.memory_stats["total_bars"] = total_bars_after
-            self.memory_stats["last_cleanup"] = current_time
+                total_bars_after += len(self.data[tf_key])
 
-            # Log cleanup if significant
-            if total_bars_before != total_bars_after:
-                self.logger.debug(
-                    f"DataManager cleanup - Bars: {total_bars_before}→{total_bars_after}, "
-                    f"Ticks: {len(self.current_tick_data)}"
-                )
+        # Cleanup tick buffer - deque handles its own cleanup with maxlen
+        # No manual cleanup needed for deque with maxlen
 
-                # Force garbage collection after cleanup
-                gc.collect()
+        # Update stats
+        current_time = time.time()
+        self.last_cleanup = current_time
+        self.memory_stats["bars_cleaned"] += total_bars_before - total_bars_after
+        self.memory_stats["total_bars"] = total_bars_after
+        self.memory_stats["last_cleanup"] = current_time
+
+        # Log cleanup if significant
+        if total_bars_before != total_bars_after:
+            self.logger.debug(
+                f"DataManager cleanup - Bars: {total_bars_before}→{total_bars_after}, "
+                f"Ticks: {len(self.current_tick_data)}"
+            )
+
+            # Force garbage collection after cleanup
+            gc.collect()
 
     async def _periodic_cleanup(self) -> None:
         """Background task for periodic cleanup."""

src/project_x_py/realtime_data_manager/mmap_overflow.py

@@ -377,14 +377,28 @@ async def restore_from_overflow(self, timeframe: str, bars: int) -> bool:
                     # Take the requested number of bars from the end
                     restore_df = overflow_df.tail(bars)
 
-                    async with self.data_lock:
-                        # Prepend to current data
-                        if timeframe in self.data:
-                            self.data[timeframe] = pl.concat(
-                                [restore_df, self.data[timeframe]]
-                            )
-                        else:
-                            self.data[timeframe] = restore_df
+                    # Import here to avoid circular dependency
+                    from project_x_py.utils.lock_optimization import AsyncRWLock
+
+                    # Use appropriate lock method based on lock type
+                    if isinstance(self.data_lock, AsyncRWLock):
+                        async with self.data_lock.write_lock():
+                            # Prepend to current data
+                            if timeframe in self.data:
+                                self.data[timeframe] = pl.concat(
+                                    [restore_df, self.data[timeframe]]
+                                )
+                            else:
+                                self.data[timeframe] = restore_df
+                    else:
+                        async with self.data_lock:
+                            # Prepend to current data
+                            if timeframe in self.data:
+                                self.data[timeframe] = pl.concat(
+                                    [restore_df, self.data[timeframe]]
+                                )
+                            else:
+                                self.data[timeframe] = restore_df
 
                     logger.info(
                         f"Restored {len(restore_df)} bars for {timeframe} from overflow"

src/project_x_py/statistics/export.py

@@ -11,7 +11,7 @@
 import csv
 import json
 import re
-from datetime import datetime
+from datetime import UTC, datetime
 from io import StringIO
 from typing import Any, ClassVar, Union
 
@@ -64,7 +64,9 @@ async def to_json(
         data = self._stats_to_dict(stats)
 
         if include_timestamp:
-            data["export_timestamp"] = datetime.utcnow().isoformat() + "Z"
+            data["export_timestamp"] = (
+                datetime.now(UTC).isoformat().replace("+00:00", "Z")
+            )
 
         if self.sanitize_sensitive:
             data = self._sanitize_data(data)
@@ -88,7 +90,7 @@ async def to_prometheus(
             Prometheus format string
         """
         lines = []
-        timestamp = int(datetime.utcnow().timestamp() * 1000)
+        timestamp = int(datetime.now(UTC).timestamp() * 1000)
 
         # Health metrics
         health_stats = stats.get("health")
@@ -227,7 +229,11 @@ async def to_csv(
 
         writer.writerow(headers)
 
-        timestamp = datetime.utcnow().isoformat() + "Z" if include_timestamp else None
+        timestamp = (
+            datetime.now(UTC).isoformat().replace("+00:00", "Z")
+            if include_timestamp
+            else None
+        )
 
         # Flatten stats into rows
         rows = []
@@ -320,9 +326,41 @@ async def to_csv(
             ).items():
                 rows.append(["connections", "connection_status", status, conn_type])
 
+        # If no rows were generated from standard stats, handle custom structures
+        if not rows:
+            # Flatten any custom dictionary structure
+            def flatten_dict(d, parent_key="", sep="_"):
+                items = []
+                for k, v in d.items():
+                    new_key = f"{parent_key}{sep}{k}" if parent_key else k
+                    if isinstance(v, dict):
+                        items.extend(flatten_dict(v, new_key, sep=sep).items())
+                    else:
+                        items.append((new_key, v))
+                return dict(items)
+
+            flat_stats = flatten_dict(stats)
+            for key, value in flat_stats.items():
+                # Split the key to get category and name
+                parts = key.split("_", 1)
+                category = parts[0] if parts else "custom"
+                name = parts[1] if len(parts) > 1 else key
+                # Try to extract component from key
+                component = "system"
+                if "order_manager" in key:
+                    component = "order_manager"
+                elif "position_manager" in key:
+                    component = "position_manager"
+                elif "realtime" in key:
+                    component = "realtime"
+                row = [category, name, value, component]
+                if include_timestamp:
+                    row.append(timestamp)
+                rows.append(row)
+
         # Write rows
         for row in rows:
-            if include_timestamp:
+            if include_timestamp and len(row) == 4:
                 row.append(timestamp)
             writer.writerow(row)
 
@@ -342,7 +380,7 @@ async def to_datadog(
             Dictionary formatted for Datadog API
         """
         metrics = []
-        timestamp = int(datetime.utcnow().timestamp())
+        timestamp = int(datetime.now(UTC).timestamp())
 
         # Health metrics
         health_stats = stats.get("health")
@@ -517,6 +555,26 @@ async def export(
 
     def _stats_to_dict(self, stats: ComprehensiveStats) -> dict[str, Any]:
         """Convert ComprehensiveStats to dictionary."""
+        # Check if this appears to be a structured ComprehensiveStats
+        # by looking for standard stats keys
+        has_standard_keys = any(
+            key in stats
+            for key in [
+                "health",
+                "performance",
+                "memory",
+                "errors",
+                "connections",
+                "trading",
+            ]
+        )
+
+        # If no standard keys found, just return the dict as-is
+        # This handles test cases and non-standard data structures
+        if not has_standard_keys:
+            return dict(stats)
+
+        # Process standard ComprehensiveStats structure
         result = {}
 
         health_stats = stats.get("health")
@@ -583,6 +641,10 @@ def _stats_to_dict(self, stats: ComprehensiveStats) -> dict[str, Any]:
                 else None,
             }
 
+        # Also preserve any suite data if present
+        if "suite" in stats:
+            result["suite"] = stats["suite"]
+
         return result
 
     def _sanitize_data(self, data: Any) -> Any: