From 094a174ff2ebd47b6aa04fd4a8471fa546c23088 Mon Sep 17 00:00:00 2001 From: bigsong <35025755+bigsongeth@users.noreply.github.com> Date: Thu, 9 Jul 2026 00:13:58 +0800 Subject: [PATCH] Fix mock trading order submission broken by the OMS refactor The OMS refactor (4cdb370) routed every EMS submit handler through private_connector._oms.(...), and the follow-up removed the per-exchange components that consumed the "{exchange}.order" msgbus endpoint. MockLinearConnector was never migrated: - It has no _oms attribute, so any order submitted via Strategy.create_order() dies in the EMS with AttributeError: 'MockLinearConnector' object has no attribute '_oms' and the engine's cancel-open-orders shutdown path hits the same error. - Its fill reports still go to the "{exchange}.order" endpoint, which nothing consumes anymore, so even without the crash no fill would reach the cache, the registry, or the strategy callbacks. Migrate the mock connector to the new architecture: - Serve as its own OMS (self._oms = self): its create_order / cancel_order / cancel_all_orders match the interface the EMS calls - Accept the EMS-generated oid (and reduce_only) in create_order instead of always minting a fresh UUID, so registry and inflight tracking reconcile - Dispatch PENDING/FILLED/FAILED updates the way OrderManagementSystem.order_status_update does: validate against the registry, update the cache (which clears inflight state), emit the strategy callback endpoints, and unregister closed orders - Log-and-return stubs for the operations mock trading does not support (modify/batch/tp-sl/ws), because the EMS submit loop awaits some of them directly and an exception would kill the queue consumer - Pass the engine's OrderRegistry to the mock connector; direct calls without a registered oid behave as before Reproduce: run strategy/binance/mock_trading.py; the first scheduled market order (~10s) crashes the submit task and the strategy never receives on_filled_order. --- nexustrader/base/connector.py | 73 ++++++++++++-- nexustrader/engine.py | 1 + test/base/test_mock_linear_connector.py | 124 +++++++++++++++++++++++- 3 files changed, 188 insertions(+), 10 deletions(-) diff --git a/nexustrader/base/connector.py b/nexustrader/base/connector.py index af8934ae..e82398eb 100644 --- a/nexustrader/base/connector.py +++ b/nexustrader/base/connector.py @@ -25,6 +25,7 @@ from nexustrader.constants import ExchangeType, AccountType from nexustrader.core.cache import AsyncCache from nexustrader.core.entity import TaskManager +from nexustrader.core.registry import OrderRegistry from nexustrader.error import OrderError from nexustrader.constants import ( OrderSide, @@ -497,6 +498,7 @@ def __init__( quote_currency: str = "USDT", update_interval: int = 60, # seconds leverage: int = 1, + registry: OrderRegistry | None = None, ): self._account_type = account_type self._market = exchange.market @@ -514,6 +516,11 @@ def __init__( self._task_manager = task_manager self._leverage = leverage self._log = Logger(name=type(self).__name__) + self._registry = registry or OrderRegistry() + # The EMS submit handlers and the engine dispatch order operations + # through `private_connector._oms`. The mock connector executes + # orders itself, so it serves as its own OMS. + self._oms = self async def _init_position(self): for _, position in self._cache._get_all_positions_from_db( @@ -537,8 +544,9 @@ async def _init_balance(self): self._cache._apply_balance(self._account_type, balances) await self._cache.sync_balances() - async def cancel_order(self, symbol: str, order_id: str, **kwargs) -> Order: - """Cancel an order""" + async def cancel_order(self, oid: str, symbol: str, **kwargs) -> Order: + """Cancel an order. Mock orders fill immediately, so there is never + anything to cancel.""" pass async def cancel_all_orders(self, symbol: str) -> bool: @@ -553,8 +561,11 @@ async def create_order( amount: Decimal, price: Decimal | None = None, time_in_force: TimeInForce = TimeInForce.GTC, + oid: str | None = None, + reduce_only: bool = False, **kwargs, ) -> Order: + oid = oid or UUID4().value try: if amount <= 0: raise OrderError(f"Invalid order amount {amount}") @@ -614,15 +625,13 @@ async def create_order( fee = amount * Decimal(str(price)) * Decimal(str(self._fee_rate)) fee_currency = market.quote - reduce_only = kwargs.get("reduce_only", False) - cost = amount * Decimal(str(price)) order = Order( exchange=self._exchange_id, symbol=symbol, status=OrderStatus.PENDING, - oid=UUID4().value, + oid=oid, amount=amount, filled=Decimal(0), timestamp=self._clock.timestamp_ms(), @@ -661,14 +670,17 @@ async def create_order( ) self._apply_position(order) - self._msgbus.send( - endpoint=f"{self._exchange_id.value}.order", msg=order_filled - ) + # Nothing consumes the "{exchange_id}.order" endpoint since the + # OMS refactor; dispatch the status updates directly so the cache + # and the strategy callbacks stay in sync. + self._order_status_update(order) + self._order_status_update(order_filled) return order except OrderError as e: self._log.error(f"Error creating order: {e}") - return Order( + failed_order = Order( exchange=self._exchange_id, + oid=oid, timestamp=self._clock.timestamp_ms(), symbol=symbol, status=OrderStatus.FAILED, @@ -680,6 +692,49 @@ async def create_order( filled=Decimal(0), remaining=amount, ) + self._order_status_update(failed_order) + return failed_order + + def _order_status_update(self, order: Order): + """Dispatch an order status update the way OrderManagementSystem does. + + Orders submitted outside the EMS path (oid not registered) update + neither the cache nor the strategy callbacks, matching the previous + behaviour of direct calls. + """ + if not self._registry.is_registered(order.oid): + return + if not self._cache.update_order_status(order): + return + match order.status: + case OrderStatus.PENDING: + self._msgbus.send(endpoint="pending", msg=order) + case OrderStatus.FAILED: + self._msgbus.send(endpoint="failed", msg=order) + case OrderStatus.FILLED: + self._msgbus.send(endpoint="filled", msg=order) + if order.is_closed: + self._registry.unregister_order(order.oid) + self._registry.unregister_tmp_order(order.oid) + + async def modify_order(self, *args, **kwargs) -> None: + self._log.error("modify_order is not supported by the mock connector") + + async def create_tp_sl_order(self, *args, **kwargs) -> None: + self._log.error("create_tp_sl_order is not supported by the mock connector") + + async def create_batch_orders(self, *args, **kwargs) -> None: + self._log.error("create_batch_orders is not supported by the mock connector") + + async def create_order_ws(self, *args, **kwargs) -> None: + self._log.error( + "create_order_ws is not supported by the mock connector; use create_order" + ) + + async def cancel_order_ws(self, *args, **kwargs) -> None: + self._log.error( + "cancel_order_ws is not supported by the mock connector; use cancel_order" + ) @property def pnl(self) -> float: diff --git a/nexustrader/engine.py b/nexustrader/engine.py index 5d05ed92..a484a57a 100644 --- a/nexustrader/engine.py +++ b/nexustrader/engine.py @@ -252,6 +252,7 @@ def _build_private_connectors(self): quote_currency=mock_conn_config.quote_currency, update_interval=mock_conn_config.update_interval, leverage=mock_conn_config.leverage, + registry=self._registry, ) self._private_connectors[account_type] = private_connector elif mock_conn_config.account_type.is_inverse_mock: diff --git a/test/base/test_mock_linear_connector.py b/test/base/test_mock_linear_connector.py index ed5b8128..583982bf 100644 --- a/test/base/test_mock_linear_connector.py +++ b/test/base/test_mock_linear_connector.py @@ -2,7 +2,7 @@ from decimal import Decimal from typing import Dict from nexustrader.schema import PositionSide -from nexustrader.constants import OrderStatus, OrderSide, OrderType +from nexustrader.constants import OrderStatus, OrderSide, OrderType, TimeInForce from nexustrader.exchange.binance.constants import BinanceAccountType from nexustrader.core.nautilius_core import LiveClock from nexustrader.base import MockLinearConnector @@ -397,3 +397,125 @@ async def test_initialize_overwrite_check( assert ( len(position) == 0 ) # since we overwrite the balance, the position is not in db + + +############################ TEST EMS SUBMIT PATH ############################ + +# Regression tests: the OMS refactor routed the EMS submit handlers through +# `private_connector._oms.(...)`, but MockLinearConnector had no `_oms` +# attribute. Every order submitted via Strategy.create_order() crashed with +# AttributeError and no fill ever reached the cache or strategy callbacks. + + +async def test_ems_create_order_path( + mock_linear_connector: MockLinearConnector, message_bus +): + connector = mock_linear_connector + await connector._cache._init_storage() + await connector._init_balance() + await connector._init_position() + + received = [] + message_bus.register(endpoint="pending", handler=received.append) + message_bus.register(endpoint="filled", handler=received.append) + + symbol = "BTCUSDT-PERP.BINANCE" + oid = "ems-oid-1" + # mimic what the EMS does before dispatching to connector._oms + connector._registry.register_order(oid) + connector._cache.add_inflight_order(symbol, oid) + + order = await connector._oms.create_order( + oid=oid, + symbol=symbol, + side=OrderSide.BUY, + type=OrderType.MARKET, + amount=Decimal("1"), + price=None, + time_in_force=TimeInForce.GTC, + reduce_only=False, + ) + + assert order.oid == oid # the EMS-generated oid is honored + assert [o.status for o in received] == [OrderStatus.PENDING, OrderStatus.FILLED] + assert all(o.oid == oid for o in received) + + cached = connector._cache.get_order(oid) + assert cached is not None + assert cached.status == OrderStatus.FILLED + assert connector._cache.get_inflight_orders(symbol) == set() + assert not connector._registry.is_registered(oid) + + position = connector._cache.get_position(symbol) + assert position.amount == Decimal("1") + + +async def test_ems_failed_order_reaches_strategy( + mock_linear_connector: MockLinearConnector, message_bus +): + connector = mock_linear_connector + await connector._cache._init_storage() + await connector._init_balance() + await connector._init_position() + + received = [] + message_bus.register(endpoint="failed", handler=received.append) + + oid = "ems-oid-failed-1" + connector._registry.register_order(oid) + order = await connector._oms.create_order( + oid=oid, + symbol="BTC-USD", # unknown market -> OrderError + side=OrderSide.BUY, + type=OrderType.MARKET, + amount=Decimal("1"), + price=None, + time_in_force=TimeInForce.GTC, + reduce_only=False, + ) + + assert order.status == OrderStatus.FAILED + assert order.oid == oid + assert [o.oid for o in received] == [oid] + assert not connector._registry.is_registered(oid) + + +async def test_direct_create_order_generates_oid( + mock_linear_connector: MockLinearConnector, message_bus +): + """Direct calls (bypassing the EMS) keep working: an oid is generated + and no strategy callbacks fire for unregistered orders.""" + connector = mock_linear_connector + await connector._cache._init_storage() + await connector._init_balance() + await connector._init_position() + + received = [] + message_bus.register(endpoint="filled", handler=received.append) + + order = await connector.create_order( + symbol="BTCUSDT-PERP.BINANCE", + side=OrderSide.BUY, + type=OrderType.LIMIT, + amount=Decimal("1"), + ) + assert order.status == OrderStatus.PENDING + assert order.oid + assert received == [] + + +async def test_unsupported_ems_operations_do_not_raise( + mock_linear_connector: MockLinearConnector, +): + """The EMS submit loop awaits some of these directly; raising would kill + the queue consumer task.""" + connector = mock_linear_connector + await connector._oms.modify_order( + oid="x", symbol="s", side=None, price=None, amount=None + ) + await connector._oms.create_tp_sl_order(oid="x", symbol="s") + await connector._oms.create_batch_orders(orders=[]) + await connector._oms.create_order_ws(oid="x", symbol="s") + await connector._oms.cancel_order_ws(oid="x", symbol="s") + await connector._oms.cancel_order(oid="x", symbol="s") + await connector._oms.cancel_all_orders("s")