Improve execution engine reconciliation robustness

- Add 1-unit precision tolerance for fill/position reconciliation
- Skip reconciliation during shutdown to prevent race conditions
- Add tests for tolerance and position discrepancy checks

Author: cjdsellers

examples/live/okx/okx_spot_swap_quoter.py

@@ -408,7 +408,7 @@ def on_stop(self) -> None:
     trader_id=TraderId("TESTER-001"),
     logging=LoggingConfig(
         log_level="INFO",
-        # log_level_file="DEBUG",
+        log_level_file="DEBUG",
         use_pyo3=True,
     ),
     exec_engine=LiveExecEngineConfig(

nautilus_trader/live/execution_engine.py

@@ -2011,6 +2011,9 @@ def _reconcile_order_report(
         trades: list[FillReport],
         is_external: bool = True,
     ) -> bool:
+        if self._is_shutting_down:
+            return True  # Skip reconciliation during shutdown
+
         client_order_id: ClientOrderId = report.client_order_id
 
         if client_order_id is None:
@@ -2146,6 +2149,15 @@ def _reconcile_order_report(
         if report.filled_qty > order.filled_qty:
             # Check if order is already closed to avoid duplicate inferred fills
             if order.is_closed:
+                # Use the higher precision for tolerance check
+                precision = max(report.filled_qty.precision, order.filled_qty.precision)
+                if self._is_within_single_unit_tolerance(
+                    report.filled_qty.as_decimal(),
+                    order.filled_qty.as_decimal(),
+                    precision,
+                ):
+                    return True
+
                 self._log.warning(  # TODO: Reduce level to debug after initial development phase
                     f"{order.instrument_id} {order.client_order_id!r} already {order.status_string()} but "
                     f"reported difference in filled_qty: "
@@ -2180,6 +2192,9 @@ def _reconcile_order_report(
         return True  # Reconciled
 
     def _reconcile_fill_report_single(self, report: FillReport) -> bool:
+        if self._is_shutting_down:
+            return True  # Skip reconciliation during shutdown
+
         if not self._consider_for_reconciliation(report.instrument_id):
             self._log_skipping_reconciliation_on_instrument_id(report)
             return True  # Filtered
@@ -2379,6 +2394,20 @@ def _fill_reports_equal(self, cached_fill: OrderFilled, report: FillReport) -> b
             and cached_fill.ts_event == report.ts_event
         )
 
+    def _is_within_single_unit_tolerance(
+        self,
+        value1: Decimal,
+        value2: Decimal,
+        precision: int,
+    ) -> bool:
+        # Handles rounding discrepancies from venues (e.g., OKX fillSz vs accFillSz)
+        # Only apply tolerance for fractional quantities (precision > 0)
+        if precision == 0:
+            return value1 == value2  # Integer quantities require exact match
+
+        tolerance = Decimal(10) ** -precision
+        return abs(value1 - value2) <= tolerance
+
     def _check_position_discrepancy(
         self,
         cached_positions: list[Position],
@@ -2394,6 +2423,19 @@ def _check_position_discrepancy(
         if venue_report is None:
             # We think we have a position, but venue says flat (or no report)
             if cached_qty != 0:
+                instrument = self._cache.instrument(instrument_id)
+                if instrument is not None:
+                    if self._is_within_single_unit_tolerance(
+                        cached_qty,
+                        Decimal(0),
+                        instrument.size_precision,
+                    ):
+                        return False
+                else:
+                    self._log.debug(
+                        f"Cannot apply tolerance check for {instrument_id}: instrument not in cache",
+                    )
+
                 self._log.warning(
                     f"Position discrepancy for {instrument_id}: "
                     f"cached_qty={cached_qty}, venue has no position report",
@@ -2409,9 +2451,25 @@ def _check_position_discrepancy(
         if cached_qty == venue_qty:
             return False
 
+        instrument = self._cache.instrument(instrument_id)
+        if instrument is not None:
+            if self._is_within_single_unit_tolerance(
+                cached_qty,
+                venue_qty,
+                instrument.size_precision,
+            ):
+                return False
+        else:
+            self._log.debug(
+                f"Cannot apply tolerance check for {instrument_id}: instrument not in cache",
+            )
+
         return True
 
     def _reconcile_position_report(self, report: PositionStatusReport) -> bool:
+        if self._is_shutting_down:
+            return True  # Skip reconciliation during shutdown
+
         if not self._consider_for_reconciliation(report.instrument_id):
             self._log_skipping_reconciliation_on_instrument_id(report)
             return True  # Filtered

nautilus_trader/test_kit/strategies/tester_exec.py

@@ -315,10 +315,12 @@ def open_position(self, net_qty: Decimal) -> None:
             self.log.warning(f"Open position with {net_qty}, skipping")
             return
 
+        quantity = self.instrument.make_qty(abs(net_qty))
+
         order: MarketOrder = self.order_factory.market(
             instrument_id=self.config.instrument_id,
             order_side=OrderSide.BUY if net_qty > 0 else OrderSide.SELL,
-            quantity=self.instrument.make_qty(self.config.order_qty),
+            quantity=quantity,
             time_in_force=self.config.open_position_time_in_force,
             quote_quantity=self.config.use_quote_quantity,
         )