Add execution.fast support for Bybit (#2165)

Author: sunlei

nautilus_trader/adapters/bybit/config.py

@@ -96,6 +96,8 @@ class BybitExecClientConfig(LiveExecClientConfig, frozen=True):
     use_gtd : bool, default False
         If False, then GTD time in force will be remapped to GTC
         (this is useful if managing GTD orders locally).
+    use_ws_execution_fast : bool, default False
+        If use fast execution stream.
     use_ws_trade_api : bool, default False
         If the client is using websocket to send order requests.
     use_http_batch_api : bool, default False
@@ -126,6 +128,7 @@ class BybitExecClientConfig(LiveExecClientConfig, frozen=True):
     demo: bool = False
     testnet: bool = False
     use_gtd: bool = False  # Not supported on Bybit
+    use_ws_execution_fast: bool = False
     use_ws_trade_api: bool = False
     use_http_batch_api: bool = False
     max_retries: PositiveInt | None = None

nautilus_trader/adapters/bybit/execution.py

@@ -15,7 +15,6 @@
 
 from __future__ import annotations
 
-from decimal import Decimal
 from typing import TYPE_CHECKING
 
 import msgspec
@@ -39,6 +38,8 @@
 from nautilus_trader.adapters.bybit.http.errors import should_retry
 from nautilus_trader.adapters.bybit.schemas.common import BYBIT_PONG
 from nautilus_trader.adapters.bybit.schemas.ws import BybitWsAccountExecution
+from nautilus_trader.adapters.bybit.schemas.ws import BybitWsAccountExecutionFast
+from nautilus_trader.adapters.bybit.schemas.ws import BybitWsAccountExecutionFastMsg
 from nautilus_trader.adapters.bybit.schemas.ws import BybitWsAccountExecutionMsg
 from nautilus_trader.adapters.bybit.schemas.ws import BybitWsAccountOrderMsg
 from nautilus_trader.adapters.bybit.schemas.ws import BybitWsAccountWalletMsg
@@ -170,11 +171,13 @@ def __init__(
         self._product_types = product_types
         self._use_gtd = config.use_gtd
         self._use_ws_trade_api = config.use_ws_trade_api
+        self._use_ws_execution_fast = config.use_ws_execution_fast
         self._use_http_batch_api = config.use_http_batch_api
 
         self._log.info(f"Account type: {account_type_to_str(account_type)}", LogColor.BLUE)
         self._log.info(f"Product types: {[p.value for p in product_types]}", LogColor.BLUE)
         self._log.info(f"{config.use_gtd=}", LogColor.BLUE)
+        self._log.info(f"{config.use_ws_execution_fast=}", LogColor.BLUE)
         self._log.info(f"{config.use_ws_trade_api=}", LogColor.BLUE)
         self._log.info(f"{config.use_http_batch_api=}", LogColor.BLUE)
         self._log.info(f"{config.ws_trade_timeout_secs=}", LogColor.BLUE)
@@ -248,6 +251,9 @@ def __init__(
 
         self._decoder_ws_account_order_update = msgspec.json.Decoder(BybitWsAccountOrderMsg)
         self._decoder_ws_account_execution_update = msgspec.json.Decoder(BybitWsAccountExecutionMsg)
+        self._decoder_ws_account_execution_fast_update = msgspec.json.Decoder(
+            BybitWsAccountExecutionFastMsg,
+        )
         # self._decoder_ws_account_position_update = msgspec.json.Decoder(BybitWsAccountPositionMsg)
         self._decoder_ws_account_wallet_update = msgspec.json.Decoder(BybitWsAccountWalletMsg)
 
@@ -269,10 +275,15 @@ async def _connect(self) -> None:
         await self._update_account_state()
 
         await self._ws_client.connect()
-        await self._ws_client.subscribe_executions_update()
+
         await self._ws_client.subscribe_orders_update()
         await self._ws_client.subscribe_wallet_update()
 
+        if self._use_ws_execution_fast:
+            await self._ws_client.subscribe_executions_fast_update()
+        else:
+            await self._ws_client.subscribe_executions_update()
+
         if self._use_ws_trade_api:
             await self._ws_order_client.connect()
 
@@ -991,6 +1002,8 @@ def _handle_ws_message_private(self, raw: bytes) -> None:
                 self._handle_account_order_update(raw)
             elif "execution" in ws_message.topic:
                 self._handle_account_execution_update(raw)
+            elif "execution.fast" in ws_message.topic:
+                self._handle_account_execution_fast_update(raw)
             elif "wallet" == ws_message.topic:
                 self._handle_account_wallet_update(raw)
             else:
@@ -1006,7 +1019,18 @@ def _handle_account_execution_update(self, raw: bytes) -> None:
         except Exception as e:
             self._log.exception(f"Failed to handle account execution update: {e}", e)
 
-    def _process_execution(self, execution: BybitWsAccountExecution) -> None:
+    def _handle_account_execution_fast_update(self, raw: bytes) -> None:
+        try:
+            msg = self._decoder_ws_account_execution_fast_update.decode(raw)
+            for trade in msg.data:
+                self._process_execution(trade)
+        except Exception as e:
+            self._log.exception(f"Failed to handle account execution update: {e}", e)
+
+    def _process_execution(
+        self,
+        execution: BybitWsAccountExecution | BybitWsAccountExecutionFast,
+    ) -> None:
         instrument_id = self._get_cached_instrument_id(execution.symbol, execution.category)
         client_order_id = ClientOrderId(execution.orderLinkId) if execution.orderLinkId else None
         venue_order_id = VenueOrderId(execution.orderId)
@@ -1047,6 +1071,13 @@ def _process_execution(self, execution: BybitWsAccountExecution) -> None:
         if instrument is None:
             raise ValueError(f"Cannot handle trade event: instrument {instrument_id} not found")
 
+        quote_currency = instrument.quote_currency
+        fee = instrument.maker_fee if execution.isMaker else instrument.taker_fee
+
+        last_qty = Quantity(float(execution.execQty), instrument.size_precision)
+        last_px = Price(float(execution.execPrice), instrument.price_precision)
+        commission_amount = last_qty * last_px * fee
+
         self.generate_order_filled(
             strategy_id=strategy_id,
             instrument_id=instrument_id,
@@ -1056,10 +1087,10 @@ def _process_execution(self, execution: BybitWsAccountExecution) -> None:
             trade_id=TradeId(execution.execId),
             order_side=order_side,
             order_type=order_type,
-            last_qty=Quantity(float(execution.execQty), instrument.size_precision),
-            last_px=Price(float(execution.execPrice), instrument.price_precision),
-            quote_currency=instrument.quote_currency,
-            commission=Money(Decimal(execution.execFee), instrument.quote_currency),
+            last_qty=last_qty,
+            last_px=last_px,
+            quote_currency=quote_currency,
+            commission=Money(commission_amount, quote_currency),
             liquidity_side=LiquiditySide.MAKER if execution.isMaker else LiquiditySide.TAKER,
             ts_event=millis_to_nanos(float(execution.execTime)),
         )

nautilus_trader/adapters/bybit/http/account.py

@@ -64,26 +64,26 @@
 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.account.balance import BybitWalletBalance
-from nautilus_trader.adapters.bybit.schemas.account.fee_rate import BybitFeeRate
-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 BybitOrder
-from nautilus_trader.adapters.bybit.schemas.order import BybitPlaceOrderResponse
-from nautilus_trader.adapters.bybit.schemas.order import BybitSetTradingStopResponse
-from nautilus_trader.adapters.bybit.schemas.position import BybitPositionStruct
-from nautilus_trader.adapters.bybit.schemas.trade import BybitExecution
 from nautilus_trader.common.component import LiveClock
 from nautilus_trader.core.correctness import PyCondition
 
 
 if TYPE_CHECKING:
     from nautilus_trader.adapters.bybit.common.enums import BybitMarginMode
     from nautilus_trader.adapters.bybit.http.client import BybitHttpClient
+    from nautilus_trader.adapters.bybit.schemas.account.balance import BybitWalletBalance
+    from nautilus_trader.adapters.bybit.schemas.account.fee_rate import BybitFeeRate
     from nautilus_trader.adapters.bybit.schemas.account.info import BybitAccountInfo
     from nautilus_trader.adapters.bybit.schemas.account.set_leverage import BybitSetLeverageResponse
     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 BybitOrder
+    from nautilus_trader.adapters.bybit.schemas.order import BybitPlaceOrderResponse
+    from nautilus_trader.adapters.bybit.schemas.order import BybitSetTradingStopResponse
+    from nautilus_trader.adapters.bybit.schemas.position import BybitPositionStruct
+    from nautilus_trader.adapters.bybit.schemas.trade import BybitExecution
     from nautilus_trader.common.component import LiveClock
 
 # fmt: on

nautilus_trader/adapters/bybit/schemas/ws.py

@@ -720,6 +720,34 @@ class BybitWsAccountExecutionMsg(msgspec.Struct):
     data: list[BybitWsAccountExecution]
 
 
+################################################################################
+# Private - Account Execution Fast
+################################################################################
+
+
+class BybitWsAccountExecutionFast(msgspec.Struct):
+    category: BybitProductType
+    symbol: str
+    orderId: str
+    isMaker: bool
+    orderLinkId: str
+    side: BybitOrderSide
+    execId: str
+    execPrice: str
+    execQty: str
+    execTime: str
+    seq: int
+
+    orderType: BybitOrderType = BybitOrderType.UNKNOWN
+    stopOrderType: BybitStopOrderType = BybitStopOrderType.NONE
+
+
+class BybitWsAccountExecutionFastMsg(msgspec.Struct):
+    topic: str
+    creationTime: int
+    data: list[BybitWsAccountExecutionFast]
+
+
 ################################################################################
 # Private - Account Wallet
 ################################################################################

nautilus_trader/adapters/bybit/websocket/client.py

@@ -375,7 +375,7 @@ async def subscribe_executions_update(self) -> None:
         subscription = "execution"
         await self._subscribe(subscription)
 
-    async def subscribe_executions_update_fast(self) -> None:
+    async def subscribe_executions_fast_update(self) -> None:
         subscription = "execution.fast"
         await self._subscribe(subscription)