fix: guard zero-target data sampling

CHANGELOG.md

@@ -14,6 +14,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Migration guides will be provided for all breaking changes
 - Semantic versioning (MAJOR.MINOR.PATCH) is strictly followed
 
+## [Unreleased]
+
+### 🐛 Fixed
+
+**Realtime Data Manager**:
+- Guarded data sampling for single-bar buffer limits so overflow cleanup
+  preserves the latest bar instead of raising a division-by-zero error.
+
 ## [3.5.8] - 2025-09-02
 
 ### 🐛 Fixed

src/project_x_py/realtime_data_manager/memory_management.py

@@ -286,7 +286,9 @@ async def _apply_data_sampling(self, timeframe: str) -> None:
 
         current_data = self.data[timeframe]
         current_size = len(current_data)
-        target_size = int(self.max_bars_per_timeframe * 0.7)  # Reduce to 70% of max
+        target_size = max(
+            1, int(self.max_bars_per_timeframe * 0.7)
+        )  # Reduce to 70% of max
 
         if current_size <= target_size:
             return
@@ -296,15 +298,20 @@ async def _apply_data_sampling(self, timeframe: str) -> None:
         self._sampling_ratios[timeframe] = sampling_ratio
 
         # Apply intelligent sampling - keep recent data and sample older data
-        recent_data_size = int(target_size * 0.3)  # Keep 30% as recent data
+        recent_data_size = min(
+            current_size,
+            max(1, int(target_size * 0.3)),
+        )  # Keep 30% as recent data
         sampled_older_size = target_size - recent_data_size
 
         # Keep all recent data
         recent_data = current_data.tail(recent_data_size)
 
         # Sample older data intelligently
         older_data = current_data.head(current_size - recent_data_size)
-        if len(older_data) > sampled_older_size:
+        if sampled_older_size <= 0 or older_data.is_empty():
+            sampled_older = current_data.head(0)
+        elif len(older_data) > sampled_older_size:
             # Sample every nth bar to maintain temporal distribution
             sample_step = max(1, len(older_data) // sampled_older_size)
             # Use gather to sample every nth row

tests/realtime_data_manager/test_memory_management.py

@@ -328,6 +328,35 @@ async def test_apply_data_sampling_large_dataset(self, memory_manager):
         assert final_size == target_size
         assert final_size < 200
 
+    @pytest.mark.asyncio
+    async def test_apply_data_sampling_single_bar_limit_preserves_latest(self):
+        """Test data sampling handles single-bar limits without division by zero."""
+        memory_manager = MockRealtimeDataManager(max_bars=1)
+        memory_manager.configure_dynamic_buffer_sizing(enabled=True)
+        timestamps = [
+            datetime(2025, 1, 1, 10, minute, tzinfo=timezone.utc)
+            for minute in range(10)
+        ]
+        sample_data = pl.DataFrame(
+            {
+                "timestamp": timestamps,
+                "open": [100.0 + i for i in range(10)],
+                "high": [101.0 + i for i in range(10)],
+                "low": [99.0 + i for i in range(10)],
+                "close": [100.5 + i for i in range(10)],
+                "volume": [1000] * 10,
+            }
+        )
+        memory_manager.data["1min"] = sample_data
+        memory_manager.last_bar_times["1min"] = timestamps[-1]
+
+        await memory_manager._apply_data_sampling("1min")
+
+        final_data = memory_manager.data["1min"]
+        assert len(final_data) == 1
+        assert final_data.select(pl.col("timestamp")).item() == timestamps[-1]
+        assert memory_manager.last_bar_times["1min"] == timestamps[-1]
+
     @pytest.mark.asyncio
     async def test_apply_data_sampling_preserves_recent_data(self, memory_manager):
         """Test data sampling preserves most recent data."""
@@ -705,7 +734,9 @@ async def test_get_memory_stats_with_overflow_stats(self, memory_manager):
         """Test memory stats include overflow statistics."""
         # Mock overflow stats method
         mock_overflow_stats = {"disk_overflow_count": 5, "disk_usage_mb": 100.0}
-        memory_manager.get_overflow_stats_summary = AsyncMock(return_value=mock_overflow_stats)
+        memory_manager.get_overflow_stats_summary = AsyncMock(
+            return_value=mock_overflow_stats
+        )
 
         stats = await memory_manager.get_memory_stats()