Improve position snapshot processing and incremental caching

- Add incremental snapshot caching to process only new snapshots
- Add cache method `position_snapshot_ids()`
- Add cache method `position_snapshot_bytes()`
- Add housekeeping to prune stale snapshot entires
- Add portfolio and snapshot caching tests

Author: cjdsellers

examples/live/binance/binance_futures_testnet_exec_tester.py

@@ -58,6 +58,14 @@
         # snapshot_orders=True,
         # snapshot_positions=True,
         # snapshot_positions_interval_secs=5.0,
+        purge_closed_orders_interval_mins=1,  # Example of purging closed orders for HFT
+        purge_closed_orders_buffer_mins=0,  # Purged orders closed for at least an hour
+        purge_closed_positions_interval_mins=1,  # Example of purging closed positions for HFT
+        purge_closed_positions_buffer_mins=0,  # Purge positions closed for at least an hour
+        purge_account_events_interval_mins=1,  # Example of purging account events for HFT
+        purge_account_events_lookback_mins=0,  # Purge account events occurring more than an hour ago
+        purge_from_database=True,  # Set True with caution
+        graceful_shutdown_on_exception=True,
     ),
     cache=CacheConfig(
         # database=DatabaseConfig(),

nautilus_trader/cache/base.pxd

@@ -189,7 +189,9 @@ cdef class CacheFacade:
     cpdef Position position(self, PositionId position_id)
     cpdef Position position_for_order(self, ClientOrderId client_order_id)
     cpdef PositionId position_id(self, ClientOrderId client_order_id)
+    cpdef set[PositionId] position_snapshot_ids(self, InstrumentId instrument_id=*)
     cpdef list position_snapshots(self, PositionId position_id=*)
+    cpdef list position_snapshot_bytes(self, PositionId position_id)
     cpdef list positions(self, Venue venue=*, InstrumentId instrument_id=*, StrategyId strategy_id=*, PositionSide side=*)
     cpdef list positions_open(self, Venue venue=*, InstrumentId instrument_id=*, StrategyId strategy_id=*, PositionSide side=*)
     cpdef list positions_closed(self, Venue venue=*, InstrumentId instrument_id=*, StrategyId strategy_id=*)

nautilus_trader/cache/base.pyx

@@ -447,10 +447,18 @@ cdef class CacheFacade:
         """Abstract method (implement in subclass)."""
         raise NotImplementedError("method `position_id` must be implemented in the subclass")  # pragma: no cover
 
+    cpdef set[PositionId] position_snapshot_ids(self, InstrumentId instrument_id = None):
+        """Abstract method (implement in subclass)."""
+        raise NotImplementedError("method `position_snapshot_ids` must be implemented in the subclass")  # pragma: no cover
+
     cpdef list position_snapshots(self, PositionId position_id = None):
         """Abstract method (implement in subclass)."""
         raise NotImplementedError("method `position_snapshots` must be implemented in the subclass")  # pragma: no cover
 
+    cpdef list position_snapshot_bytes(self, PositionId position_id):
+        """Abstract method (implement in subclass)."""
+        raise NotImplementedError("method `position_snapshot_bytes` must be implemented in the subclass")  # pragma: no cover
+
     cpdef list positions(self, Venue venue = None, InstrumentId instrument_id = None, StrategyId strategy_id = None, PositionSide side = PositionSide.NO_POSITION_SIDE):
         """Abstract method (implement in subclass)."""
         raise NotImplementedError("method `positions` must be implemented in the subclass")  # pragma: no cover

nautilus_trader/cache/cache.pxd

@@ -94,6 +94,7 @@ cdef class Cache(CacheFacade):
     cdef dict _index_position_orders
     cdef dict _index_instrument_orders
     cdef dict _index_instrument_positions
+    cdef dict _index_instrument_position_snapshots
     cdef dict _index_strategy_orders
     cdef dict _index_strategy_positions
     cdef dict _index_exec_algorithm_orders

nautilus_trader/cache/cache.pyx

@@ -153,6 +153,7 @@ cdef class Cache(CacheFacade):
         self._index_position_orders: dict[PositionId, set[ClientOrderId]] = {}
         self._index_instrument_orders: dict[InstrumentId, set[ClientOrderId]] = {}
         self._index_instrument_positions: dict[InstrumentId, set[PositionId]] = {}
+        self._index_instrument_position_snapshots: dict[InstrumentId, set[PositionId]] = {}
         self._index_strategy_orders: dict[StrategyId, set[ClientOrderId]] = {}
         self._index_strategy_positions: dict[StrategyId, set[PositionId]] = {}
         self._index_exec_algorithm_orders: dict[ExecAlgorithmId, set[ClientOrderId]] = {}
@@ -959,6 +960,19 @@ cdef class Cache(CacheFacade):
         self._index_positions_open.discard(position_id)
         self._index_positions_closed.discard(position_id)
 
+        # Remove position snapshots and clean up index
+        cdef set[PositionId] snapshot_position_ids
+        cdef list[bytes] snapshots = self._position_snapshots.pop(position_id, None)
+
+        if snapshots is not None and position is not None:
+            snapshot_position_ids = self._index_instrument_position_snapshots.get(position.instrument_id)
+            if snapshot_position_ids:
+                snapshot_position_ids.discard(position_id)
+
+                # Clean up
+                if not snapshot_position_ids:
+                    self._index_instrument_position_snapshots.pop(position.instrument_id, None)
+
         # Delete from database if requested
         if purge_from_database and self._database is not None:
             self._database.delete_position(position_id)
@@ -1027,6 +1041,7 @@ cdef class Cache(CacheFacade):
         self._index_position_orders.clear()
         self._index_instrument_orders.clear()
         self._index_instrument_positions.clear()
+        self._index_instrument_position_snapshots.clear()
         self._index_strategy_orders.clear()
         self._index_strategy_positions.clear()
         self._index_exec_algorithm_orders.clear()
@@ -2226,6 +2241,14 @@ cdef class Cache(CacheFacade):
         else:
             self._position_snapshots[position_id] = [position_pickled]
 
+        # Update snapshot index
+        cdef InstrumentId instrument_id = position.instrument_id
+        cdef set position_ids = self._index_instrument_position_snapshots.get(instrument_id)
+        if position_ids is not None:
+            position_ids.add(position_id)
+        else:
+            self._index_instrument_position_snapshots[instrument_id] = {position_id}
+
         self._log.debug(f"Snapshot {repr(copied_position)}")
 
     cpdef void snapshot_position_state(
@@ -4927,6 +4950,44 @@ cdef class Cache(CacheFacade):
 
         return [pickle.loads(s) for s in snapshots]
 
+    cpdef set position_snapshot_ids(self, InstrumentId instrument_id = None):
+        """
+        Return all position IDs for position snapshots with the given instrument filter.
+
+        Parameters
+        ----------
+        instrument_id : InstrumentId, optional
+            The instrument ID query filter.
+
+        Returns
+        -------
+        set[PositionId]
+
+        """
+        if instrument_id is not None:
+            return self._index_instrument_position_snapshots.get(instrument_id, set())
+        else:
+            # Return all position IDs that have snapshots
+            return set(self._position_snapshots.keys())
+
+    cpdef list position_snapshot_bytes(self, PositionId position_id):
+        """
+        Return the raw pickled snapshot bytes for the given position ID.
+
+        Parameters
+        ----------
+        position_id : PositionId
+            The position ID to get snapshot bytes for.
+
+        Returns
+        -------
+        list[bytes]
+            The list of pickled snapshot bytes, or empty list if no snapshots exist.
+
+        """
+        Condition.not_none(position_id, "position_id")
+        return self._position_snapshots.get(position_id, [])
+
     cpdef list positions(
         self,
         Venue venue = None,

nautilus_trader/portfolio/portfolio.pxd

@@ -27,6 +27,7 @@ from nautilus_trader.model.data cimport QuoteTick
 from nautilus_trader.model.events.account cimport AccountState
 from nautilus_trader.model.events.order cimport OrderEvent
 from nautilus_trader.model.events.position cimport PositionEvent
+from nautilus_trader.model.identifiers cimport AccountId
 from nautilus_trader.model.identifiers cimport InstrumentId
 from nautilus_trader.model.identifiers cimport PositionId
 from nautilus_trader.model.identifiers cimport Venue
@@ -55,6 +56,7 @@ cdef class Portfolio(PortfolioFacade):
     cdef dict[InstrumentId, Money] _unrealized_pnls
     cdef dict[InstrumentId, Money] _realized_pnls
     cdef dict[InstrumentId, Money] _snapshot_realized_pnls
+    cdef dict[PositionId, int] _snapshot_processed_counts
     cdef dict[InstrumentId, Decimal] _net_positions
     cdef dict[PositionId, object] _bet_positions
     cdef object _index_bet_positions
@@ -84,7 +86,7 @@ cdef class Portfolio(PortfolioFacade):
     cdef void _update_instrument_id(self, InstrumentId instrument_id)
     cdef void _update_net_position(self, InstrumentId instrument_id, list positions_open)
     cdef object _net_position(self, InstrumentId instrument_id)
-    cdef void _ensure_snapshot_pnls_cached(self)
+    cdef void _ensure_snapshot_pnls_cached_for(self, InstrumentId instrument_id)
     cdef Money _calculate_realized_pnl(self, InstrumentId instrument_id)
     cdef Money _calculate_unrealized_pnl(self, InstrumentId instrument_id, Price price=*)
     cdef Price _get_price(self, Position position)

nautilus_trader/portfolio/portfolio.pyx

@@ -24,6 +24,7 @@ The portfolio can satisfy queries for account information, margin balances,
 total risk exposures and total net positions.
 """
 
+import pickle
 import warnings
 from collections import defaultdict
 from decimal import Decimal
@@ -147,6 +148,7 @@ cdef class Portfolio(PortfolioFacade):
         self._unrealized_pnls: dict[InstrumentId, Money] = {}
         self._realized_pnls: dict[InstrumentId, Money] = {}
         self._snapshot_realized_pnls: dict[InstrumentId, Money] = {}
+        self._snapshot_processed_counts: dict[PositionId, int] = {}
         self._net_positions: dict[InstrumentId, Decimal] = {}
         self._bet_positions: dict[InstrumentId, object] = {}
         self._index_bet_positions: dict[InstrumentId, set[PositionId]] = defaultdict(set)
@@ -683,6 +685,7 @@ cdef class Portfolio(PortfolioFacade):
         self._unrealized_pnls.clear()
         self._pending_calcs.clear()
         self._snapshot_realized_pnls.clear()
+        self._snapshot_processed_counts.clear()
         self.analyzer.reset()
 
         self.initialized = False
@@ -709,44 +712,120 @@ cdef class Portfolio(PortfolioFacade):
         self._reset()
         self._log.info("DISPOSED")
 
-    cdef void _ensure_snapshot_pnls_cached(self):
-        if self._snapshot_realized_pnls:
-            return  # Already cached
+    cdef void _ensure_snapshot_pnls_cached_for(self, InstrumentId instrument_id):
+        # Get all positions for this instrument (both open and closed)
+        cdef set[PositionId] instrument_position_ids = self._cache.position_ids(venue=None, instrument_id=instrument_id)
 
-        cdef list[Position] all_snapshots = self._cache.position_snapshots()
-        cdef Position snapshot
-        cdef InstrumentId instrument_id
-        cdef Money existing_pnl
+        # Get snapshot position IDs for this instrument from index
+        cdef set[PositionId] snapshot_position_ids = self._cache.position_snapshot_ids(instrument_id)
 
-        for snapshot in all_snapshots:
-            if snapshot.realized_pnl is None:
-                continue
+        # Combine all position IDs (active and snapshot)
+        cdef set[PositionId] position_ids = set()
 
-            instrument_id = snapshot.instrument_id
-            existing_pnl = self._snapshot_realized_pnls.get(instrument_id)
+        if instrument_position_ids:
+            position_ids.update(instrument_position_ids)
 
-            if existing_pnl is not None:
-                if existing_pnl.currency == snapshot.realized_pnl.currency:
-                    # Same currency, add the amounts
-                    self._snapshot_realized_pnls[instrument_id] = Money(
-                        existing_pnl.as_double() + snapshot.realized_pnl.as_double(),
-                        existing_pnl.currency,
-                    )
+        position_ids.update(snapshot_position_ids)
+
+        if not position_ids:
+            # Clear stale cached total when no positions or snapshots exist
+            self._snapshot_realized_pnls.pop(instrument_id, None)
+            return  # No positions or snapshots for this instrument
+
+        cdef bint rebuild = False
+
+        cdef:
+            PositionId position_id
+            list position_id_snapshots
+            int prev_count
+            int curr_count
+
+        # Detect purge/reset (count regression) to trigger full rebuild for this instrument
+        for position_id in position_ids:
+            position_id_snapshots = self._cache.position_snapshot_bytes(position_id)
+            curr_count = len(position_id_snapshots)
+            prev_count = self._snapshot_processed_counts.get(position_id, 0)
+            if prev_count > curr_count:
+                rebuild = True
+                break
+
+        cdef:
+            Money existing_pnl
+            Position snapshot
+
+        if rebuild:
+            # Rebuild from scratch for this instrument
+            self._snapshot_realized_pnls.pop(instrument_id, None)
+
+            for position_id in position_ids:
+                position_id_snapshots = self._cache.position_snapshot_bytes(position_id)
+                curr_count = len(position_id_snapshots)
+                if curr_count:
+                    for s in position_id_snapshots:
+                        snapshot = pickle.loads(s)
+                        if snapshot.realized_pnl is None:
+                            continue
+
+                        # Aggregate if same currency; otherwise log and keep existing
+                        existing_pnl = self._snapshot_realized_pnls.get(instrument_id)
+                        if existing_pnl is not None:
+                            if existing_pnl.currency == snapshot.realized_pnl.currency:
+                                self._snapshot_realized_pnls[instrument_id] = Money(
+                                    existing_pnl.as_double() + snapshot.realized_pnl.as_double(),
+                                    existing_pnl.currency,
+                                )
+                            else:
+                                self._log.warning(
+                                    f"Cannot aggregate snapshot PnLs with different currencies for {instrument_id}: "
+                                    f"{existing_pnl.currency} vs {snapshot.realized_pnl.currency}",
+                                )
+                        else:
+                            self._snapshot_realized_pnls[instrument_id] = snapshot.realized_pnl
+                self._snapshot_processed_counts[position_id] = curr_count
+            return
+
+        # Incremental path: only unpickle new entries
+        for position_id in position_ids:
+            position_id_snapshots = self._cache.position_snapshot_bytes(position_id)
+            curr_count = len(position_id_snapshots)
+            if curr_count == 0:
+                continue
+            prev_count = self._snapshot_processed_counts.get(position_id, 0)
+            if prev_count >= curr_count:
+                continue
+            for idx in range(prev_count, curr_count):
+                snapshot = pickle.loads(position_id_snapshots[idx])
+                if snapshot.realized_pnl is None:
+                    continue
+                existing_pnl = self._snapshot_realized_pnls.get(instrument_id)
+                if existing_pnl is not None:
+                    if existing_pnl.currency == snapshot.realized_pnl.currency:
+                        self._snapshot_realized_pnls[instrument_id] = Money(
+                            existing_pnl.as_double() + snapshot.realized_pnl.as_double(),
+                            existing_pnl.currency,
+                        )
+                    else:
+                        self._log.warning(
+                            f"Cannot aggregate snapshot PnLs with different currencies for {instrument_id}: "
+                            f"{existing_pnl.currency} vs {snapshot.realized_pnl.currency}",
+                        )
                 else:
-                    # Different currencies - log warning
-                    self._log.warning(
-                        f"Cannot aggregate snapshot PnLs with different currencies for {instrument_id}: "
-                        f"{existing_pnl.currency} vs {snapshot.realized_pnl.currency}",
-                    )
-            else:
-                # First snapshot for this instrument
-                self._snapshot_realized_pnls[instrument_id] = Money(
-                    snapshot.realized_pnl.as_double(),
-                    snapshot.realized_pnl.currency,
-                )
+                    self._snapshot_realized_pnls[instrument_id] = snapshot.realized_pnl
+            self._snapshot_processed_counts[position_id] = curr_count
+
+        # Prune stale entries from processed counts
+        cdef PositionId stale_position_id
+        cdef list[PositionId] stale_ids = []
 
-        if self._debug and self._snapshot_realized_pnls:
-            self._log.debug(f"Cached snapshot realized PnLs for {len(self._snapshot_realized_pnls)} instruments")
+        for stale_position_id in self._snapshot_processed_counts:
+            if stale_position_id not in position_ids:
+                stale_ids.append(stale_position_id)
+
+        for stale_position_id in stale_ids:
+            self._snapshot_processed_counts.pop(stale_position_id, None)
+
+        if self._debug and self._snapshot_realized_pnls.get(instrument_id) is not None:
+            self._log.debug(f"Cached snapshot realized PnL for {instrument_id}")
 
 # -- QUERIES --------------------------------------------------------------------------------------
 
@@ -1533,8 +1612,8 @@ cdef class Portfolio(PortfolioFacade):
         else:
             currency = instrument.get_cost_currency()
 
-        # Ensure snapshot PnLs are cached (one-time cost)
-        self._ensure_snapshot_pnls_cached()
+        # Ensure snapshot PnLs for this instrument are cached (incremental)
+        self._ensure_snapshot_pnls_cached_for(instrument_id)
 
         cdef:
             list[Position] positions
@@ -1551,6 +1630,9 @@ cdef class Portfolio(PortfolioFacade):
             instrument_id=instrument_id,
         )
 
+        if self._debug:
+            self._log.debug(f"Found {len(positions)} positions for {instrument_id}")
+
         if not positions:
             # Check if we have cached snapshot PnL for this instrument
             cached_snapshot_pnl = self._snapshot_realized_pnls.get(instrument_id)
@@ -1578,6 +1660,8 @@ cdef class Portfolio(PortfolioFacade):
         if cached_snapshot_pnl_for_instrument is None:
             pass  # No snapshot PnL to add
         elif cached_snapshot_pnl_for_instrument.currency == currency:
+            if self._debug:
+                self._log.debug(f"Adding cached snapshot PnL: {cached_snapshot_pnl_for_instrument}")
             total_pnl += cached_snapshot_pnl_for_instrument.as_double()
         else:
             snapshot_xrate = self._cache.get_xrate(