Fix consistent ordering of execution events

- Consolidated event publishing and snapshotting in ExecutionEngine
- Consolidated account state updates in Portfolio and simplified subscriptions
- Downstream components will now receive consistently ordered event types

Author: cjdsellers

RELEASES.md

@@ -37,6 +37,7 @@ Released on TBD (UTC).
 - Upgraded `pyo3` crate to v0.24.2
 
 ### Fixes
+- Fixed consistent ordering of execution events (#2513), thanks for reporting @stastnypremysl
 - Fixed memory leak in `RetryManager` by simplifying the acquire-release pattern, avoiding the asynchronous context manager protocol that led to state sharing, thanks for reporting @DeevsDeevs
 - Fixed locked balance and initial margin calculations for reduce-only orders (#2505), thanks for reporting @stastnypremysl
 - Fixed purging order events from position (these needed to be purged prior to removing cache index entry), thanks @DeevsDeevs

nautilus_trader/execution/engine.pxd

@@ -19,7 +19,6 @@ from nautilus_trader.common.component cimport TimeEvent
 from nautilus_trader.common.generators cimport PositionIdGenerator
 from nautilus_trader.core.rust.model cimport OmsType
 from nautilus_trader.core.rust.model cimport OrderSide
-from nautilus_trader.execution.algorithm cimport ExecAlgorithm
 from nautilus_trader.execution.client cimport ExecutionClient
 from nautilus_trader.execution.messages cimport BatchCancelOrders
 from nautilus_trader.execution.messages cimport CancelAllOrders
@@ -31,6 +30,7 @@ from nautilus_trader.execution.messages cimport SubmitOrderList
 from nautilus_trader.execution.messages cimport TradingCommand
 from nautilus_trader.model.events.order cimport OrderEvent
 from nautilus_trader.model.events.order cimport OrderFilled
+from nautilus_trader.model.events.position cimport PositionEvent
 from nautilus_trader.model.identifiers cimport InstrumentId
 from nautilus_trader.model.identifiers cimport PositionId
 from nautilus_trader.model.identifiers cimport StrategyId
@@ -52,6 +52,7 @@ cdef class ExecutionEngine(Component):
     cdef readonly dict[StrategyId, OmsType] _oms_overrides
     cdef readonly dict[InstrumentId, StrategyId] _external_order_claims
     cdef readonly str snapshot_positions_timer_name
+    cdef list[PositionEvent] _pending_position_events
 
     cdef readonly bint debug
     """If debug mode is active (will provide extra debug logging).\n\n:returns: `bool`"""

nautilus_trader/execution/engine.pyx

@@ -22,8 +22,7 @@ includes sending commands to, and receiving events from, the trading venue
 endpoints via its registered execution clients.
 
 The engine employs a simple fan-in fan-out messaging pattern to execute
-`TradingCommand` messages, and process `AccountState` or `OrderEvent` type
-messages.
+`TradingCommand` messages and `OrderEvent` messages.
 
 Alternative implementations can be written on top of the generic engine - which
 just need to override the `execute` and `process` methods.
@@ -48,7 +47,6 @@ from nautilus_trader.common.component cimport RECV
 from nautilus_trader.common.component cimport Clock
 from nautilus_trader.common.component cimport Component
 from nautilus_trader.common.component cimport LogColor
-from nautilus_trader.common.component cimport Logger
 from nautilus_trader.common.component cimport MessageBus
 from nautilus_trader.common.component cimport TimeEvent
 from nautilus_trader.common.generators cimport PositionIdGenerator
@@ -57,12 +55,9 @@ from nautilus_trader.core.fsm cimport InvalidStateTrigger
 from nautilus_trader.core.rust.core cimport secs_to_nanos
 from nautilus_trader.core.rust.model cimport ContingencyType
 from nautilus_trader.core.rust.model cimport OmsType
-from nautilus_trader.core.rust.model cimport OrderStatus
-from nautilus_trader.core.rust.model cimport OrderType
+from nautilus_trader.core.rust.model cimport OrderSide
 from nautilus_trader.core.rust.model cimport PositionSide
-from nautilus_trader.core.rust.model cimport TimeInForce
 from nautilus_trader.core.uuid cimport UUID4
-from nautilus_trader.execution.algorithm cimport ExecAlgorithm
 from nautilus_trader.execution.client cimport ExecutionClient
 from nautilus_trader.execution.messages cimport BatchCancelOrders
 from nautilus_trader.execution.messages cimport CancelAllOrders
@@ -91,11 +86,11 @@ from nautilus_trader.model.identifiers cimport PositionId
 from nautilus_trader.model.identifiers cimport StrategyId
 from nautilus_trader.model.identifiers cimport Venue
 from nautilus_trader.model.instruments.base cimport Instrument
-from nautilus_trader.model.instruments.currency_pair cimport CurrencyPair
 from nautilus_trader.model.objects cimport Money
 from nautilus_trader.model.objects cimport Price
 from nautilus_trader.model.objects cimport Quantity
 from nautilus_trader.model.orders.base cimport Order
+from nautilus_trader.trading.strategy cimport Strategy
 
 
 cdef class ExecutionEngine(Component):
@@ -151,6 +146,8 @@ cdef class ExecutionEngine(Component):
             clock=clock,
         )
 
+        self._pending_position_events = []
+
         # Configuration
         self.debug: bool = config.debug
         self.manage_own_order_books = config.manage_own_order_books
@@ -1050,6 +1047,25 @@ cdef class ExecutionEngine(Component):
         else:
             self._apply_event_to_order(order, event)
 
+        # Publish events
+        self._msgbus.publish_c(
+            topic=f"events.order.{event.strategy_id}",
+            msg=event,
+        )
+        if self.snapshot_orders:
+            self._create_order_state_snapshot(order)
+
+        for pos_event in self._pending_position_events:
+            self._msgbus.publish_c(
+                topic=f"events.position.{pos_event.strategy_id}",
+                msg=pos_event,
+            )
+            if self.snapshot_positions:
+                position = self.cache.position(pos_event.position_id)
+                self._create_position_state_snapshot(position, open_only=isinstance(pos_event, PositionOpened))
+
+        self._pending_position_events.clear()
+
     cpdef OmsType _determine_oms_type(self, OrderFilled fill):
         cdef ExecutionClient client
         # Check for strategy OMS override
@@ -1171,13 +1187,6 @@ cdef class ExecutionEngine(Component):
 
         self._cache.update_order(order)
 
-        self._msgbus.publish_c(
-            topic=f"events.order.{event.strategy_id}",
-            msg=event,
-        )
-        if self.snapshot_orders:
-            self._create_order_state_snapshot(order)
-
     cpdef void _handle_order_fill(self, Order order, OrderFilled fill, OmsType oms_type):
         cdef Instrument instrument = self._cache.load_instrument(fill.instrument_id)
         if instrument is None:
@@ -1222,8 +1231,6 @@ cdef class ExecutionEngine(Component):
         if position is None:
             position = Position(instrument, fill)
             self._cache.add_position(position, oms_type)
-            if self.snapshot_positions:
-                self._create_position_state_snapshot(position, open_only=True)
         else:
             try:
                 # Always snapshot opening positions to handle NETTING OMS
@@ -1242,10 +1249,7 @@ cdef class ExecutionEngine(Component):
             ts_init=self._clock.timestamp_ns(),
         )
 
-        self._msgbus.publish_c(
-            topic=f"events.position.{event.strategy_id}",
-            msg=event,
-        )
+        self._pending_position_events.append(event)
 
         return position
 
@@ -1258,8 +1262,6 @@ cdef class ExecutionEngine(Component):
             return  # Not re-raising to avoid crashing engine
 
         self._cache.update_position(position)
-        if self.snapshot_positions:
-            self._create_position_state_snapshot(position, open_only=False)
 
         cdef PositionEvent event
         if position.is_closed_c():
@@ -1277,10 +1279,7 @@ cdef class ExecutionEngine(Component):
                 ts_init=self._clock.timestamp_ns(),
             )
 
-        self._msgbus.publish_c(
-            topic=f"events.position.{event.strategy_id}",
-            msg=event,
-        )
+        self._pending_position_events.append(event)
 
     cpdef bint _will_flip_position(self, Position position, OrderFilled fill):
         return (

nautilus_trader/portfolio/portfolio.pxd

@@ -13,6 +13,8 @@
 #  limitations under the License.
 # -------------------------------------------------------------------------------------------------
 
+from libc.stdint cimport uint64_t
+
 from nautilus_trader.accounting.accounts.base cimport Account
 from nautilus_trader.accounting.manager cimport AccountsManager
 from nautilus_trader.cache.cache cimport Cache
@@ -25,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
@@ -74,6 +77,7 @@ cdef class Portfolio(PortfolioFacade):
 
 # -- INTERNAL -------------------------------------------------------------------------------------
 
+    cdef AccountState _update_position(self, InstrumentId instrument_id, AccountId account_id, uint64_t ts_event)
     cdef object _net_position(self, InstrumentId instrument_id)
     cdef void _update_instrument_id(self, InstrumentId instrument_id)
     cdef void _update_net_position(self, InstrumentId instrument_id, list positions_open)

nautilus_trader/portfolio/portfolio.pyx

@@ -24,6 +24,8 @@ The portfolio can satisfy queries for account information, margin balances,
 total risk exposures and total net positions.
 """
 
+from libc.stdint cimport uint64_t
+
 from collections import defaultdict
 from decimal import Decimal
 
@@ -56,6 +58,7 @@ from nautilus_trader.model.events.order cimport OrderRejected
 from nautilus_trader.model.events.order cimport OrderUpdated
 from nautilus_trader.model.events.position cimport PositionEvent
 from nautilus_trader.model.functions cimport position_side_to_str
+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
@@ -167,7 +170,6 @@ cdef class Portfolio(PortfolioFacade):
 
         # Required subscriptions
         self._msgbus.subscribe(topic="events.order.*", handler=self.update_order, priority=10)
-        self._msgbus.subscribe(topic="events.position.*", handler=self.update_position, priority=10)
         self._msgbus.subscribe(topic="events.account.*", handler=self.update_account, priority=10)
 
         if config.use_mark_prices:
@@ -526,6 +528,11 @@ cdef class Portfolio(PortfolioFacade):
             ts_event=event.ts_event,
         )
 
+        if isinstance(event, OrderFilled):
+            maybe_account_state = self._update_position(event.instrument_id, event.account_id, event.ts_event)
+            if maybe_account_state is not None:
+                account_state = maybe_account_state
+
         if account_state is None:
             self._log.debug(f"Added pending calculation for {instrument.id}")
             self._pending_calcs.add(instrument.id)
@@ -547,48 +554,49 @@ cdef class Portfolio(PortfolioFacade):
             The event to update with.
 
         """
-        Condition.not_none(event, "event")
+        self._update_position(event.instrument_id, event.account_id, event.ts_event)
 
+    cdef AccountState _update_position(self, InstrumentId instrument_id, AccountId account_id, uint64_t ts_event):
         cdef list positions_open = self._cache.positions_open(
             venue=None,  # Faster query filtering
-            instrument_id=event.instrument_id,
+            instrument_id=instrument_id,
         )
         self._update_net_position(
-            instrument_id=event.instrument_id,
+            instrument_id=instrument_id,
             positions_open=positions_open,
         )
 
-        self._realized_pnls[event.instrument_id] = self._calculate_realized_pnl(
-            instrument_id=event.instrument_id,
+        self._realized_pnls[instrument_id] = self._calculate_realized_pnl(
+            instrument_id=instrument_id,
         )
-        self._unrealized_pnls[event.instrument_id] = self._calculate_unrealized_pnl(
-            instrument_id=event.instrument_id,
+        self._unrealized_pnls[instrument_id] = self._calculate_unrealized_pnl(
+            instrument_id=instrument_id,
         )
 
-        cdef Account account = self._cache.account(event.account_id)
+        cdef Account account = self._cache.account(account_id)
         if account is None:
             self._log.error(
                 f"Cannot update position: "
-                f"no account registered for {event.account_id}",
+                f"no account registered for {account_id}",
             )
             return  # No account registered
 
         if account.type != AccountType.MARGIN or not account.calculate_account_state:
             return  # Nothing to calculate
 
-        cdef Instrument instrument = self._cache.instrument(event.instrument_id)
+        cdef Instrument instrument = self._cache.instrument(instrument_id)
         if instrument is None:
             self._log.error(
                 f"Cannot update position: "
-                f"no instrument found for {event.instrument_id}",
+                f"no instrument found for {instrument_id}",
             )
             return  # No instrument found
 
-        self._accounts.update_positions(
+        return self._accounts.update_positions(
             account=account,
             instrument=instrument,
             positions_open=positions_open,
-            ts_event=event.ts_event,
+            ts_event=ts_event,
         )
 
     def _reset(self) -> None: