Add WebSocket batch order operations for Bybit (#2521)

Author: sunlei

RELEASES.md

@@ -10,7 +10,7 @@ None
 
 ### Internal Improvements
 - Implemented exponential backoff and jitter for the `RetryManager` (#2518), thanks @davidsblom
-- Improved reconnection robustness for Bybit private/trading channels (#2520), thanks @davidsblom
+- Improved reconnection robustness for Bybit private/trading channels (#2520), thanks @sunlei
 - Fixed some clippy lints (#2517), thanks @twitu
 - Upgraded `databento` crate to v0.23.0
 - Upgraded `sqlx` crate to v0.8.5

nautilus_trader/adapters/bybit/common/enums.py

@@ -137,6 +137,9 @@ class BybitWsOrderRequestMsgOP(Enum):
     CREATE = "order.create"
     AMEND = "order.amend"
     CANCEL = "order.cancel"
+    CREATE_BATCH = "order.create-batch"
+    AMEND_BATCH = "order.amend-batch"
+    CANCEL_BATCH = "order.cancel-batch"
 
 
 @unique

nautilus_trader/adapters/bybit/endpoints/trade/amend_order.py

@@ -21,8 +21,8 @@
 
 from nautilus_trader.adapters.bybit.common.enums import BybitEndpointType
 from nautilus_trader.adapters.bybit.common.enums import BybitProductType
-from nautilus_trader.adapters.bybit.common.enums import BybitTriggerType
 from nautilus_trader.adapters.bybit.endpoints.endpoint import BybitHttpEndpoint
+from nautilus_trader.adapters.bybit.endpoints.trade.batch_amend_order import BybitBatchAmendOrder
 from nautilus_trader.adapters.bybit.schemas.order import BybitAmendOrderResponse
 from nautilus_trader.core.nautilus_pyo3 import HttpMethod
 
@@ -31,23 +31,13 @@
     from nautilus_trader.adapters.bybit.http.client import BybitHttpClient
 
 
-class BybitAmendOrderPostParams(msgspec.Struct, omit_defaults=True, frozen=True):
+class BybitAmendOrderPostParams(
+    BybitBatchAmendOrder,
+    omit_defaults=True,
+    frozen=True,
+    kw_only=True,
+):
     category: BybitProductType
-    symbol: str
-    orderId: str | None = None
-    orderLinkId: str | None = None
-    orderIv: str | None = None
-    triggerPrice: str | None = None
-    qty: str | None = None
-    price: str | None = None
-    tpslMode: str | None = None
-    takeProfit: str | None = None
-    stopLoss: str | None = None
-    tpTriggerBy: str | None = None
-    slTriggerBy: str | None = None
-    triggerBy: BybitTriggerType | None = None
-    tpLimitPrice: str | None = None
-    slLimitPrice: str | None = None
 
 
 class BybitAmendOrderEndpoint(BybitHttpEndpoint):

nautilus_trader/adapters/bybit/endpoints/trade/batch_amend_order.py

@@ -21,6 +21,7 @@
 
 from nautilus_trader.adapters.bybit.common.enums import BybitEndpointType
 from nautilus_trader.adapters.bybit.common.enums import BybitProductType
+from nautilus_trader.adapters.bybit.common.enums import BybitTriggerType
 from nautilus_trader.adapters.bybit.endpoints.endpoint import BybitHttpEndpoint
 from nautilus_trader.adapters.bybit.schemas.order import BybitBatchAmendOrderResponse
 from nautilus_trader.core.nautilus_pyo3 import HttpMethod
@@ -43,6 +44,7 @@ class BybitBatchAmendOrder(msgspec.Struct, omit_defaults=True, frozen=True):
     stopLoss: str | None = None
     tpTriggerBy: str | None = None
     slTriggerBy: str | None = None
+    triggerBy: BybitTriggerType | None = None
     tpLimitPrice: str | None = None
     slLimitPrice: str | None = None
 

nautilus_trader/adapters/bybit/endpoints/trade/batch_cancel_order.py

@@ -34,6 +34,7 @@ class BybitBatchCancelOrder(msgspec.Struct, omit_defaults=True, frozen=True):
     symbol: str
     orderId: str | None = None
     orderLinkId: str | None = None
+    orderFilter: str | None = None  # Spot only
 
 
 class BybitBatchCancelOrderPostParams(msgspec.Struct, omit_defaults=True, frozen=True):

nautilus_trader/adapters/bybit/endpoints/trade/cancel_order.py

@@ -22,6 +22,7 @@
 from nautilus_trader.adapters.bybit.common.enums import BybitEndpointType
 from nautilus_trader.adapters.bybit.common.enums import BybitProductType
 from nautilus_trader.adapters.bybit.endpoints.endpoint import BybitHttpEndpoint
+from nautilus_trader.adapters.bybit.endpoints.trade.batch_cancel_order import BybitBatchCancelOrder
 from nautilus_trader.adapters.bybit.schemas.order import BybitCancelOrderResponse
 from nautilus_trader.core.nautilus_pyo3 import HttpMethod
 
@@ -30,12 +31,13 @@
     from nautilus_trader.adapters.bybit.http.client import BybitHttpClient
 
 
-class BybitCancelOrderPostParams(msgspec.Struct, omit_defaults=True, frozen=True):
+class BybitCancelOrderPostParams(
+    BybitBatchCancelOrder,
+    omit_defaults=True,
+    frozen=True,
+    kw_only=True,
+):
     category: BybitProductType
-    symbol: str
-    orderId: str | None = None
-    orderLinkId: str | None = None
-    orderFilter: str | None = None  # Spot only
 
 
 class BybitCancelOrderEndpoint(BybitHttpEndpoint):

nautilus_trader/adapters/bybit/endpoints/trade/place_order.py

@@ -38,6 +38,8 @@ class BybitPlaceOrderPostParams(
     kw_only=True,
 ):
     category: BybitProductType
+    slippageToleranceType: str | None = None
+    slippageTolerance: str | None = None
 
 
 class BybitPlaceOrderEndpoint(BybitHttpEndpoint):

nautilus_trader/adapters/bybit/execution.py

@@ -685,7 +685,7 @@ async def _cancel_order(self, command: CancelOrder) -> None:
 
         retry_manager = await self._retry_manager_pool.acquire()
         try:
-            await retry_manager.run(
+            response = await retry_manager.run(
                 "cancel_order",
                 [client_order_id, venue_order_id],
                 self._order_single_client.cancel_order,
@@ -703,10 +703,35 @@ async def _cancel_order(self, command: CancelOrder) -> None:
                     retry_manager.message,
                     self._clock.timestamp_ns(),
                 )
+
+            if response:
+                ret_code = response.retCode
+                if ret_code != 0:
+                    if ret_code == 110001:  # order not exists or too late to cancel
+                        if not order.is_closed:
+                            self.generate_order_canceled(
+                                strategy_id=order.strategy_id,
+                                instrument_id=order.instrument_id,
+                                client_order_id=order.client_order_id,
+                                venue_order_id=order.venue_order_id,
+                                ts_event=self._clock.timestamp_ns(),
+                            )
+                    else:
+                        self.generate_order_cancel_rejected(
+                            strategy_id=order.strategy_id,
+                            instrument_id=order.instrument_id,
+                            client_order_id=order.client_order_id,
+                            venue_order_id=order.venue_order_id,
+                            reason=response.retMsg,
+                            ts_event=self._clock.timestamp_ns(),
+                        )
         finally:
             await self._retry_manager_pool.release(retry_manager)
 
-    async def _batch_cancel_orders(self, command: BatchCancelOrders) -> None:
+    async def _batch_cancel_orders(  # noqa: C901 (too complex)
+        self,
+        command: BatchCancelOrders,
+    ) -> None:
         # https://bybit-exchange.github.io/docs/v5/order/batch-cancel
 
         bybit_symbol = BybitSymbol(command.instrument_id.symbol.value)
@@ -742,7 +767,7 @@ async def _batch_cancel_orders(self, command: BatchCancelOrders) -> None:
 
             retry_manager = await self._retry_manager_pool.acquire()
             try:
-                await retry_manager.run(
+                response = await retry_manager.run(
                     "batch_cancel_orders",
                     None,
                     self._order_batch_client.batch_cancel_orders,
@@ -755,13 +780,40 @@ async def _batch_cancel_orders(self, command: BatchCancelOrders) -> None:
                         if order is None or order.is_closed:
                             continue
                         self.generate_order_cancel_rejected(
-                            order.strategy_id,
-                            order.instrument_id,
-                            order.client_order_id,
-                            order.venue_order_id,
-                            retry_manager.message,
-                            self._clock.timestamp_ns(),
+                            strategy_id=order.strategy_id,
+                            instrument_id=order.instrument_id,
+                            client_order_id=order.client_order_id,
+                            venue_order_id=order.venue_order_id,
+                            reason=retry_manager.message,
+                            ts_event=self._clock.timestamp_ns(),
                         )
+
+                if response:
+                    for idx, ret_info in enumerate(response.retExtInfo.list):
+                        ret_code = ret_info.code
+                        if ret_code != 0:
+                            order = self._cache.order(
+                                ClientOrderId(response.data.list[idx].orderLinkId),
+                            )
+                            if order is None or order.is_closed:
+                                continue
+                            if ret_code == 110001:  # order not exists or too late to cancel
+                                self.generate_order_canceled(
+                                    strategy_id=order.strategy_id,
+                                    instrument_id=order.instrument_id,
+                                    client_order_id=order.client_order_id,
+                                    venue_order_id=order.venue_order_id,
+                                    ts_event=self._clock.timestamp_ns(),
+                                )
+                            else:
+                                self.generate_order_cancel_rejected(
+                                    strategy_id=order.strategy_id,
+                                    instrument_id=order.instrument_id,
+                                    client_order_id=order.client_order_id,
+                                    venue_order_id=order.venue_order_id,
+                                    reason=ret_info.msg,
+                                    ts_event=self._clock.timestamp_ns(),
+                                )
             finally:
                 await self._retry_manager_pool.release(retry_manager)
 
@@ -876,7 +928,10 @@ async def _submit_order(self, command: SubmitOrder) -> None:
         finally:
             await self._retry_manager_pool.release(retry_manager)
 
-    async def _submit_order_list(self, command: SubmitOrderList) -> None:
+    async def _submit_order_list(  # noqa: C901 (too complex)
+        self,
+        command: SubmitOrderList,
+    ) -> None:
         bybit_symbol = BybitSymbol(command.instrument_id.symbol.value)
         product_type = bybit_symbol.product_type
         command_orders = command.order_list.orders
@@ -912,13 +967,39 @@ async def _submit_order_list(self, command: SubmitOrderList) -> None:
 
             retry_manager = await self._retry_manager_pool.acquire()
             try:
-                await retry_manager.run(
+                response = await retry_manager.run(
                     "submit_order_list",
                     None,
                     self._order_batch_client.batch_place_orders,
                     product_type=product_type,
                     submit_orders=submit_orders,
                 )
+
+                if not retry_manager.result:
+                    for order in batch_submits:
+                        self.generate_order_rejected(
+                            strategy_id=order.strategy_id,
+                            instrument_id=order.instrument_id,
+                            client_order_id=order.client_order_id,
+                            reason=retry_manager.message,
+                            ts_event=self._clock.timestamp_ns(),
+                        )
+
+                if response:
+                    for idx, ret_info in enumerate(response.retExtInfo.list):
+                        if ret_info.code != 0:
+                            order = self._cache.order(
+                                ClientOrderId(response.data.list[idx].orderLinkId),
+                            )
+                            if order is None:
+                                continue
+                            self.generate_order_rejected(
+                                strategy_id=order.strategy_id,
+                                instrument_id=order.instrument_id,
+                                client_order_id=order.client_order_id,
+                                reason=ret_info.msg,
+                                ts_event=self._clock.timestamp_ns(),
+                            )
             finally:
                 await self._retry_manager_pool.release(retry_manager)
 
@@ -1148,8 +1229,7 @@ def _process_execution(
 
         if client_order_id is None:
             self._log.debug(
-                f"Cannot process order execution for {venue_order_id!r}: "
-                "no `ClientOrderId` found (most likely due to being an external order)",
+                f"Cannot process order execution for {venue_order_id!r}: no `ClientOrderId` found (most likely due to being an external order)",
             )
             return
 
@@ -1172,8 +1252,7 @@ def _process_execution(
             )
             if strategy_id is None:
                 self._log.warning(
-                    f"Cannot process order execution for {client_order_id!r}: "
-                    "no strategy ID found (most likely due to being an external order)",
+                    f"Cannot process order execution for {client_order_id!r}: no strategy ID found (most likely due to being an external order)",
                 )
                 return
         else:

nautilus_trader/adapters/bybit/http/account.py

@@ -64,6 +64,8 @@
 from nautilus_trader.adapters.bybit.endpoints.trade.trade_history import BybitTradeHistoryEndpoint
 from nautilus_trader.adapters.bybit.endpoints.trade.trade_history import BybitTradeHistoryGetParams
 from nautilus_trader.adapters.bybit.http.client import BybitHttpClient
+from nautilus_trader.adapters.bybit.schemas.order import BybitBatchCancelOrderResponse
+from nautilus_trader.adapters.bybit.schemas.order import BybitBatchPlaceOrderResponse
 from nautilus_trader.common.component import LiveClock
 from nautilus_trader.core.correctness import PyCondition
 
@@ -79,7 +81,7 @@
     from nautilus_trader.adapters.bybit.schemas.account.set_margin_mode import BybitSetMarginModeResponse
     from nautilus_trader.adapters.bybit.schemas.account.switch_mode import BybitSwitchModeResponse
     from nautilus_trader.adapters.bybit.schemas.order import BybitAmendOrder
-    from nautilus_trader.adapters.bybit.schemas.order import BybitCancelOrder
+    from nautilus_trader.adapters.bybit.schemas.order import BybitCancelOrderResponse
     from nautilus_trader.adapters.bybit.schemas.order import BybitOrder
     from nautilus_trader.adapters.bybit.schemas.order import BybitPlaceOrderResponse
     from nautilus_trader.adapters.bybit.schemas.order import BybitSetTradingStopResponse
@@ -409,7 +411,7 @@ async def cancel_order(
         client_order_id: str | None = None,
         venue_order_id: str | None = None,
         order_filter: str | None = None,
-    ) -> BybitCancelOrder:
+    ) -> BybitCancelOrderResponse:
         response = await self._endpoint_cancel_order.post(
             BybitCancelOrderPostParams(
                 category=product_type,
@@ -419,7 +421,7 @@ async def cancel_order(
                 orderFilter=order_filter,
             ),
         )
-        return response.result
+        return response
 
     async def cancel_all_orders(
         self,
@@ -438,24 +440,24 @@ async def batch_place_orders(
         self,
         product_type: BybitProductType,
         submit_orders: list[BybitBatchPlaceOrder],
-    ) -> list[Any]:
+    ) -> BybitBatchPlaceOrderResponse:
         response = await self._endpoint_batch_place_order.post(
             BybitBatchPlaceOrderPostParams(
                 category=product_type,
                 request=submit_orders,
             ),
         )
-        return response.result.list
+        return response
 
     async def batch_cancel_orders(
         self,
         product_type: BybitProductType,
         cancel_orders: list[BybitBatchCancelOrder],
-    ) -> list[Any]:
+    ) -> BybitBatchCancelOrderResponse:
         response = await self._endpoint_batch_cancel_order.post(
             BybitBatchCancelOrderPostParams(
                 category=product_type,
                 request=cancel_orders,
             ),
         )
-        return response.result.list
+        return response