Fix position activity timestamps in live execution engine

Author: cjdsellers

RELEASES.md

@@ -47,17 +47,21 @@ releases as feedback arrives, before the final `2.0.0` release.
 - Fixed v2 matching engine queue position for per-order deltas in L3 books (#4370), thanks for reporting @warmi024
 - Fixed v2 own order book sizes to track remaining quantity after partial fills
 - Fixed v2 interval book snapshots blocking order submission from `on_book` handlers
+- Fixed v2 position reconciliation grace to measure on the monotonic clock (#4366), thanks @folknor
+- Fixed live execution engine position activity to stamp receipt time instead of venue `ts_event`
 - Fixed Redis message bus startup with Python v2 configs (#4356), thanks for reporting @davidgreyme
 - Fixed Binance Futures order reports omitting external limit order prices (#4346), thanks for reporting @linimin
 - Fixed Binance Futures external algo order materialization (#4348), thanks for reporting @linimin
 - Fixed Derive perpetual quote and settlement currency to USDC (venue reports quote as `USD`)
 - Fixed Derive option `scheduled_activation` parsing as UNIX seconds (was parsed as milliseconds)
 - Fixed Polymarket RTDS retained-subscription recovery after reconnects (#4353), thanks @graceyangfan
 - Fixed Tardis replay trades directory to `trades/` for catalog compatibility (#4373), thanks @AdvancedUno
+- Fixed Tardis replay bars directory to `bars/` for catalog compatibility (#4378), thanks @AdvancedUno
 - Fixed Hyperliquid `l2Book` resubscribe options and shared stream teardown (#4298)
 
 ### Internal Improvements
 - Improved portfolio statistics test coverage with canonical worked examples
+- Made portfolio reference-count clones explicit (#4364), thanks @ChrisAB
 - Upgraded Rust (MSRV) to 1.96.1
 - Upgraded Cython to v3.2.8
 - Upgraded `redis` crate to v1.3.0

nautilus_trader/live/execution_engine.py

@@ -3724,10 +3724,12 @@ def _handle_event_with_tracking(self, event: OrderEvent) -> None:
         self._record_local_activity(event)
 
         if isinstance(event, OrderFilled):
-            self._recent_fills_cache[event.trade_id] = self._clock.timestamp_ns()
-            self._position_local_activity_ns[(event.instrument_id, event.account_id)] = (
-                event.ts_event
-            )
+            ts_now = self._clock.timestamp_ns()
+            self._recent_fills_cache[event.trade_id] = ts_now
+
+            # Stamp receipt time: a venue ts_event ahead of this clock would make
+            # the grace delta negative and suppress the position check.
+            self._position_local_activity_ns[(event.instrument_id, event.account_id)] = ts_now
 
             # Track inferred fill timestamps to prevent duplicate historical fills
             if event.reconciliation:

tests/unit_tests/live/test_execution_recon.py

@@ -4634,6 +4634,75 @@ async def counting_query(instrument_id, clients):
     assert len(queries) == 1
 
 
+@pytest.mark.asyncio
+async def test_position_activity_stamped_from_receipt_time_for_venue_ahead_fill(
+    live_exec_engine,
+    exec_client,
+    cache,
+):
+    # Arrange
+    live_exec_engine.register_client(exec_client)
+    live_exec_engine.generate_missing_orders = False
+    live_exec_engine._position_check_threshold_ns = 0
+
+    account = AccountId("SIM-A")
+    order = TestExecStubs.limit_order(instrument=AUDUSD_SIM, order_side=OrderSide.BUY)
+    order.apply(TestEventStubs.order_submitted(order, account_id=account))
+    order.apply(TestEventStubs.order_accepted(order, account_id=account))
+    cache.add_order(order, position_id=None)
+
+    venue_ahead_ns = live_exec_engine._clock.timestamp_ns() + 900 * 86_400 * 1_000_000_000
+    fill = TestEventStubs.order_filled(
+        order,
+        instrument=AUDUSD_SIM,
+        account_id=account,
+        trade_id=TradeId("T-AXIS-1"),
+        last_qty=Quantity.from_int(1000),
+        last_px=Price.from_str("1.00000"),
+        ts_event=venue_ahead_ns,
+    )
+
+    # Act
+    live_exec_engine._handle_event_with_tracking(fill)
+
+    # Assert - stamp is on the local clock axis
+    stamp = live_exec_engine._position_local_activity_ns[(AUDUSD_SIM.id, account)]
+    assert stamp <= live_exec_engine._clock.timestamp_ns()
+    assert stamp != venue_ahead_ns
+
+    # A cached position with no venue report: a genuine discrepancy
+    order_b = TestExecStubs.limit_order(instrument=AUDUSD_SIM, order_side=OrderSide.BUY)
+    fill_b = TestEventStubs.order_filled(
+        order_b,
+        instrument=AUDUSD_SIM,
+        account_id=account,
+        trade_id=TradeId("T-AXIS-2"),
+        last_qty=Quantity.from_int(1000),
+        last_px=Price.from_str("1.00000"),
+        position_id=PositionId("P-AXIS-B"),
+    )
+    position = Position(instrument=AUDUSD_SIM, fill=fill_b)
+    cache.add_position(position, OmsType.HEDGING)
+
+    queries = []
+
+    async def counting_query(instrument_id, clients):
+        queries.append(instrument_id)
+        return [], False
+
+    live_exec_engine._query_and_find_missing_fills = counting_query
+
+    # Act - zero threshold: a receipt-time stamp is already expired, a venue-axis
+    # stamp would make the delta negative and suppress forever
+    await live_exec_engine._process_cached_position_discrepancies(
+        {(AUDUSD_SIM.id, account): [position]},
+        {},
+    )
+
+    # Assert - the grace does not suppress the discrepancy query
+    assert len(queries) == 1
+
+
 @pytest.mark.asyncio
 async def test_check_positions_consistency_processes_only_discrepant_account(
     live_exec_engine,