Implement Bybit reduce_only and batch cancel

Author: cjdsellers

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

@@ -40,6 +40,7 @@ class BybitPlaceOrderPostParams(msgspec.Struct, omit_defaults=True, frozen=True)
     triggerBy: BybitTriggerType | None = None
     timeInForce: BybitTimeInForce | None = None
     orderLinkId: str | None = None
+    reduceOnly: bool | None = None
 
 
 class BybitPlaceOrderEndpoint(BybitHttpEndpoint):

nautilus_trader/adapters/bybit/execution.py

@@ -532,53 +532,58 @@ async def _cancel_order(self, command: CancelOrder) -> None:
 
     async def _cancel_all_orders(self, command: CancelAllOrders) -> None:
         bybit_symbol = BybitSymbol(command.instrument_id.symbol.value)
-        await self._http_account.cancel_all_orders(
-            bybit_symbol.product_type,
-            bybit_symbol.raw_symbol,
+
+        if bybit_symbol.product_type == BybitProductType.INVERSE:
+            # Batch cancel not implemented for INVERSE
+            self._log.warning(
+                f"Batch cancel not implemented for INVERSE, "
+                f"canceling all for symbol {command.instrument_id.symbol.value}",
+            )
+            await self._http_account.cancel_all_orders(
+                bybit_symbol.product_type,
+                bybit_symbol.raw_symbol,
+            )
+            return
+
+        open_orders_strategy: list[Order] = self._cache.orders_open(
+            instrument_id=command.instrument_id,
+            strategy_id=command.strategy_id,
         )
 
-    # TODO: Determine signing issue for batch requests
-    # async def _cancel_all_orders(self, command: CancelAllOrders) -> None:
-    #     open_orders_strategy: list[Order] = self._cache.orders_open(
-    #         instrument_id=command.instrument_id,
-    #         strategy_id=command.strategy_id,
-    #     )
-    #
-    #     bybit_symbol = BybitSymbol(command.instrument_id.symbol.value)
-    #
-    #     # Check total orders for instrument
-    #     open_orders_total_count = self._cache.orders_open_count(
-    #         instrument_id=command.instrument_id,
-    #     )
-    #     if open_orders_total_count > 10:
-    #         # This could be reimplemented later to group requests into batches of 10
-    #         self._log.warning(
-    #             f"Total {command.instrument_id.symbol.value} orders open exceeds 10, "
-    #             f"is {open_orders_total_count}: canceling all for symbol",
-    #         )
-    #         await self._http_account.cancel_all_orders(
-    #             bybit_symbol.product_type,
-    #             bybit_symbol.raw_symbol,
-    #         )
-    #         return
-    #
-    #     cancel_batch: list[Order] = []
-    #     for order in open_orders_strategy:
-    #         cancel_batch.append(order)
-    #
-    #     await self._http_account.batch_cancel_orders(
-    #         product_type=bybit_symbol.product_type,
-    #         symbol=bybit_symbol.raw_symbol,
-    #         orders=cancel_batch,
-    #     )
+        # Check total orders for instrument
+        open_orders_total_count = self._cache.orders_open_count(
+            instrument_id=command.instrument_id,
+        )
+        if open_orders_total_count > 10:
+            # This could be reimplemented later to group requests into batches of 10
+            self._log.warning(
+                f"Total {command.instrument_id.symbol.value} orders open exceeds 10, "
+                f"is {open_orders_total_count}: canceling all for symbol",
+            )
+            await self._http_account.cancel_all_orders(
+                bybit_symbol.product_type,
+                bybit_symbol.raw_symbol,
+            )
+            return
+
+        cancel_batch: list[Order] = []
+        for order in open_orders_strategy:
+            cancel_batch.append(order)
+
+        await self._http_account.batch_cancel_orders(
+            product_type=bybit_symbol.product_type,
+            symbol=bybit_symbol.raw_symbol,
+            orders=cancel_batch,
+        )
 
     async def _submit_order(self, command: SubmitOrder) -> None:
         order = command.order
         if order.is_closed:
             self._log.warning(f"Order {order} is already closed")
             return
 
-        if not self._check_order_validity(order):
+        bybit_symbol = BybitSymbol(command.instrument_id.symbol.value)
+        if not self._check_order_validity(order, bybit_symbol.product_type):
             return
 
         self._log.debug(f"Submitting order {order}")
@@ -600,20 +605,28 @@ async def _submit_order(self, command: SubmitOrder) -> None:
             except BybitError as e:
                 self._log.error(repr(e))
 
-    def _check_order_validity(self, order: Order) -> bool:
+    def _check_order_validity(self, order: Order, product_type: BybitProductType) -> bool:
         # Check order type valid
         if order.order_type not in self._enum_parser.valid_order_types:
             self._log.error(
                 f"Cannot submit {order} has invalid order type {order.order_type}, unsupported on Bybit",
             )
             return False
+
         # Check post only
         if order.is_post_only and order.order_type != OrderType.LIMIT:
             self._log.error(
                 f"Cannot submit {order} has invalid post only {order.is_post_only}, unsupported on Bybit",
             )
             return False
 
+        # Check reduce only
+        if order.is_reduce_only and product_type == BybitProductType.SPOT:
+            self._log.error(
+                f"Cannot submit {order} is reduce_only, unsupported on Bybit SPOT",
+            )
+            return False
+
         return True
 
     async def _submit_market_order(self, order: MarketOrder) -> None:
@@ -630,6 +643,7 @@ async def _submit_market_order(self, order: MarketOrder) -> None:
             quote_quantity=order.is_quote_quantity,
             time_in_force=time_in_force,
             client_order_id=str(order.client_order_id),
+            reduce_only=order.is_reduce_only if order.is_reduce_only else None,
         )
 
     async def _submit_limit_order(self, order: LimitOrder) -> None:
@@ -647,6 +661,7 @@ async def _submit_limit_order(self, order: LimitOrder) -> None:
             price=str(order.price),
             time_in_force=time_in_force,
             client_order_id=str(order.client_order_id),
+            reduce_only=order.is_reduce_only if order.is_reduce_only else None,
         )
 
     def _handle_ws_message(self, raw: bytes) -> None:

nautilus_trader/adapters/bybit/http/account.py

@@ -219,6 +219,7 @@ async def place_order(
         price: str | None = None,
         time_in_force: BybitTimeInForce | None = None,
         client_order_id: str | None = None,
+        reduce_only: bool | None = None,
     ) -> BybitPlaceOrderResponse:
         market_unit = "baseCoin" if not quote_quantity else "quoteCoin"
         result = await self._endpoint_place_order.post(
@@ -232,6 +233,7 @@ async def place_order(
                 price=price,
                 timeInForce=time_in_force,
                 orderLinkId=client_order_id,
+                reduceOnly=reduce_only,
             ),
         )
         return result
@@ -313,4 +315,4 @@ async def batch_cancel_orders(
                 request=request,
             ),
         )
-        return response.result
+        return response.result.list

nautilus_trader/adapters/bybit/http/client.py

@@ -31,17 +31,6 @@
 from nautilus_trader.core.nautilus_pyo3 import Quota
 
 
-def create_string_from_dict(data):
-    property_strings = []
-
-    for key, value in data.items():
-        property_string = f'"{key}":"{value}"'
-        property_strings.append(property_string)
-
-    result_string = "{" + ",".join(property_strings) + "}"
-    return result_string
-
-
 class ResponseCode(msgspec.Struct):
     retCode: int
 
@@ -80,7 +69,7 @@ def __init__(
         self._log: Logger = Logger(name=type(self).__name__)
         self._api_key: str = api_key
         self._api_secret: str = api_secret
-        self._recv_window: int = 8000
+        self._recv_window: int = 5000
 
         self._base_url: str = base_url
         self._headers: dict[str, Any] = {
@@ -178,7 +167,7 @@ async def sign_request(
 
     def _sign_post_request(self, payload: dict[str, Any]) -> list[str]:
         timestamp = str(self._clock.timestamp_ms())
-        payload_str = create_string_from_dict(payload)
+        payload_str = msgspec.json.encode(payload).decode()
         result = timestamp + self._api_key + str(self._recv_window) + payload_str
         signature = hmac.new(
             self._api_secret.encode(),

nautilus_trader/adapters/bybit/schemas/order.py

@@ -216,6 +216,15 @@ class BybitAmendOrderResponse(msgspec.Struct):
 ################################################################################
 
 
+class BybitPlaceResult(msgspec.Struct):
+    code: int  # Success/error code
+    msg: str  # Success/error message
+
+
+class BybitBatchPlaceOrderExtInfo(msgspec.Struct):
+    list: list[BybitPlaceResult]
+
+
 class BybitBatchPlaceOrder(msgspec.Struct):
     category: BybitProductType
     symbol: str
@@ -224,10 +233,15 @@ class BybitBatchPlaceOrder(msgspec.Struct):
     createAt: str
 
 
+class BybitBatchPlaceOrderResult(msgspec.Struct):
+    list: list[BybitBatchPlaceOrder]
+
+
 class BybitBatchPlaceOrderResponse(msgspec.Struct):
     retCode: int
     retMsg: str
-    result: list[BybitBatchPlaceOrder]
+    result: BybitBatchPlaceOrderResult
+    retExtInfo: BybitBatchPlaceOrderExtInfo
     time: int
 
 
@@ -241,18 +255,26 @@ class BybitCancelResult(msgspec.Struct):
     msg: str  # Success/error message
 
 
+class BybitBatchCancelOrderExtInfo(msgspec.Struct):
+    list: list[BybitCancelResult]
+
+
 class BybitBatchCancelOrder(msgspec.Struct):
     category: BybitProductType
     symbol: str
     orderId: str
     orderLinkId: str
 
 
+class BybitBatchCancelOrderResult(msgspec.Struct):
+    list: list[BybitBatchCancelOrder]
+
+
 class BybitBatchCancelOrderResponse(msgspec.Struct):
     retCode: int
     retMsg: str
-    result: list[BybitBatchCancelOrder]
-    retExtInfo: list[BybitCancelResult]
+    result: BybitBatchCancelOrderResult
+    retExtInfo: BybitBatchCancelOrderExtInfo
     time: int
 
 
@@ -266,16 +288,24 @@ class BybitAmendResult(msgspec.Struct):
     msg: str  # Success/error message
 
 
-class BybitBatchAmend(msgspec.Struct):
+class BybitBatchAmendOrderExtInfo(msgspec.Struct):
+    list: list[BybitAmendResult]
+
+
+class BybitBatchAmendOrder(msgspec.Struct):
     category: BybitProductType
     symbol: str
     orderId: str
     orderLinkId: str
 
 
+class BybitBatchAmendOrderResult(msgspec.Struct):
+    list: list[BybitBatchAmendOrder]
+
+
 class BybitBatchAmendOrderResponse(msgspec.Struct):
     retCode: int
     retMsg: str
-    result: list[BybitBatchAmend]
-    retExtInfo: list[BybitAmendResult]
+    result: BybitBatchAmendOrderResult
+    retExtInfo: BybitBatchAmendOrderExtInfo
     time: int