fix: Fire reject report if order ack has errors

Rejects are ONLY observable in the order ack, not the execution report
stream from which all other updates are observed. While the pybit sdk
does expose ack callbacks, it doesn't invoke them on rejects/errors.

src/lib_python_interop/python_interop_plugins/bybit_execution_gateway.py

@@ -2,7 +2,7 @@ import json
 import time
 from typing import Optional
 import uuid
-from pybit.unified_trading import WebSocket, WebSocketTrading
+from pybit_unified_trading import WebSocket, WebSocketTrading
 from spark_core import (
     JsonElement,
     as_bool,
@@ -24,14 +24,32 @@ from python_interop_module import (
     PySecurity,
     PySecurityList,
 )
-from spark_enums import OrderStatus, ExecType
+from spark_enums import OrderStatus, ExecType, OrderType, SecurityType, TimeInForce
 import threading
 
 
 class OrderData:
-    def __init__(self, bts_id: int, exch_id: str, price: float, quantity: float):
+    def __init__(
+        self,
+        account: str,
+        bts_id: int,
+        exch_id: str,
+        exch_sec_id: str,
+        sec_type: SecurityType,
+        is_buy: bool,
+        order_type: OrderType,
+        time_in_force: TimeInForce,
+        price: float,
+        quantity: float,
+    ):
+        self.account = account
         self.bts_id = bts_id
         self.exch_id = exch_id
+        self.exch_sec_id = exch_sec_id
+        self.sec_type = sec_type
+        self.is_buy = is_buy
+        self.order_type = order_type
+        self.time_in_force = time_in_force
         self.price = price
         self.quantity = quantity
         self.filled_quantity: float = 0
@@ -41,6 +59,53 @@ class OrderData:
         self.exec_type: ExecType = ExecType.PENDING_NEW
         self.status: OrderStatus = OrderStatus.PENDING_NEW
 
+    def to_json(self):
+        return {
+            "account": self.account,
+            "bts_id": self.bts_id,
+            "exch_id": self.exch_id,
+            "exch_sec_id": self.exch_sec_id,
+            "sec_type": self.sec_type.name,
+            "is_buy": self.is_buy,
+            "order_type": self.order_type.name,
+            "time_in_force": self.time_in_force.name,
+            "price": self.price,
+            "quantity": self.quantity,
+            "filled_quantity": self.filled_quantity,
+            "last_filled_quantity": self.last_filled_quantity,
+            "avg_filled_price": self.avg_filled_price,
+            "last_filled_price": self.last_filled_price,
+            "exec_type": self.exec_type.name,
+            "status": self.status.name,
+        }
+
+    def to_execution_report(self):
+        transact_time = time.time_ns()
+        er = PyExecutionReport()
+        er.set_data(
+            key=self.bts_id,
+            exchange_order_id=self.exch_id,
+            venue="BYBIT",
+            security_type=self.sec_type.name,
+            exchange_security_id=self.exch_sec_id,
+            account=self.account,
+            order_type=self.order_type.name,
+            is_buy=self.is_buy,
+            price=self.price,
+            quantity=self.quantity,
+            time_in_force=self.time_in_force.name,
+            exec_type=self.exec_type.name,
+            order_status=self.status.name,
+            last_filled_price=self.last_filled_price,
+            last_filled_quantity=self.last_filled_quantity,
+            filled_quantity=self.filled_quantity,
+            avg_filled_price=self.avg_filled_price,
+            # There is no transact_id in Bybit, so we use updatedTime
+            transact_id=str(transact_time),
+            transact_time=transact_time,
+        )
+        return er
+
     def on_update(
         self,
         price: float,
@@ -83,6 +148,7 @@ class OrderData:
                 self.status = OrderStatus.REJECTED
             case _:
                 raise ValueError(f"Unexpected order status '{status}'")
+        return self.to_execution_report()
 
 
 class BybitExecutionGateway(ExecutionGateway):
@@ -113,10 +179,53 @@ class BybitExecutionGateway(ExecutionGateway):
                 return v
         return None
 
-    def _on_order_ack(self, typ, msg: object):
-        self.logger.info(
-            '[bybit_gw_ord_ack.json]{ "typ": "%s", "msg": %s}' % (typ, msg)
-        )
+    def _order_handlers(self, typ: str, exch_id: str):
+        def get_error_desc(msg: dict):
+            return "%s (code %s)" % (
+                (msg.get("retMsg") or "Unknown", msg.get("retCode") or "Unknown")
+            )
+
+        def on_err(msg: dict):
+            self.logger.error(
+                '[bybit_gw_ord_ack_err.json]{ "typ": "%s", "exch_id": "%s", "err": %s, "known_orders": %s }'
+                % (
+                    typ,
+                    exch_id,
+                    JsonElement(msg),
+                    JsonElement({k: v.to_json() for k, v in self.orders.items()}),
+                )
+            )
+            req_id = msg.get("reqId")
+            if not req_id:
+                self.logger.error(
+                    '[bybit_gw_ord_ack_err.json]{ "typ": "%s", "exch_id": "%s", "err": "%s" }'
+                    % (typ, exch_id, msg)
+                )
+            ord_data = self._get_order_data(exch_id)
+            if ord_data:
+                er = ord_data.to_execution_report()
+                match er.order_status():
+                    case OrderStatus.PENDING_NEW.name:
+                        er.set_order_status(OrderStatus.REJECTED.name)
+                        er.set_exec_type(ExecType.REJECTED.name)
+                        er.set_error_msg(get_error_desc(msg))
+                        self.bridge.on_execution_report(er)
+                    case _:
+                        pass
+            else:
+                self.logger.error(
+                    '[bybit_gw_ord_ack_err.json]{"typ": "%s", "exch_id": "%s", "err": "Unknown exch id" }'
+                )
+
+        def on_ack(msg: dict):
+            ord_data = self._get_order_data(exch_id)
+            if ord_data is None:
+                self.logger.error(
+                    '[bybit_gw_ord_ack_err.json]{ "typ": "%s", "exch_id": "%s", "err": "No order data found with exch id", "msg": "%s" }'
+                    % (typ, exch_id, msg)
+                )
+
+        return (on_ack, on_err)
 
     def _on_abort(
         self,
@@ -210,9 +319,6 @@ class BybitExecutionGateway(ExecutionGateway):
         }
         self.logger.info("[bybit_gw_start_trd_cli.json]%s" % JsonElement(params))
         self.trade_cli = WebSocketTrading(**params)
-        params = {
-            "category": "linear",
-        }
         self.bridge.on_is_ready(True)
         # Sleep loop to keep thread alive and detect connection status changes
         conn_statuses = {
@@ -269,13 +375,22 @@ class BybitExecutionGateway(ExecutionGateway):
         bts_id = req.key()
         px = req.price()
         qty = req.quantity()
-        self.orders[req.key()] = OrderData(
-            bts_id=bts_id, exch_id=exch_ord_id, price=px, quantity=qty
-        )
         sec_name = req.exchange_security_id()
+        self.orders[req.key()] = OrderData(
+            account=self.account,
+            bts_id=bts_id,
+            exch_id=exch_ord_id,
+            exch_sec_id=req.exchange_security_id(),
+            sec_type=SecurityType.FUTURE,
+            is_buy=req.is_buy(),
+            order_type=OrderType.parse(req.order_type()),
+            time_in_force=TimeInForce.parse(req.time_in_force()),
+            price=px,
+            quantity=qty,
+        )
         sec = self.securities.get(sec_name)
         if not sec:
-            return
+            raise ValueError(f"Unknown security '{sec_name}'")
         params = {
             "orderLinkId": exch_ord_id,
             "category": "linear",
@@ -285,8 +400,9 @@ class BybitExecutionGateway(ExecutionGateway):
             "price": format_price(sec, px),
             "qty": format_quantity(sec, qty),
         }
+        on_ack, on_err = self._order_handlers("crt", exch_ord_id)
         self.logger.info("[bybit_gw_crt_ord_send.json]%s" % SLog(params=params))
-        self.trade_cli.place_order(lambda msg: self._on_order_ack("crt", msg), **params)
+        self.trade_cli.place_order(on_ack, **params, error_callback=on_err)
 
     def cancel_replace_order(self, req: PyCancelReplaceOrderRequest):
         self.logger.info("[bybit_gw_cxlrep_ord.json]%s" % req)
@@ -309,10 +425,9 @@ class BybitExecutionGateway(ExecutionGateway):
             "price": format_price(sec, req.price()),
             "qty": format_quantity(sec, req.quantity()),
         }
+        on_ack, on_err = self._order_handlers("cxlrep", ord_data.exch_id)
         self.logger.info("[bybit_gw_cxlrep_ord_send.json]%s" % SLog(params=params))
-        self.trade_cli.amend_order(
-            lambda msg: self._on_order_ack("cxlrep", msg), **params
-        )
+        self.trade_cli.amend_order(on_ack, **params, error_callback=on_err)
 
     def cancel_order(self, req: PyCancelOrderRequest):
         self.logger.info("[bybit_gw_cxl_ord.json]%s" % req)
@@ -329,10 +444,9 @@ class BybitExecutionGateway(ExecutionGateway):
             "category": "linear",
             "symbol": req.exchange_security_id(),
         }
+        on_ack, on_err = self._order_handlers("cxl", ord_data.exch_id)
         self.logger.info("[bybit_gw_cxl_ord_send.json]%s" % SLog(params=params))
-        self.trade_cli.cancel_order(
-            lambda msg: self._on_order_ack("cxl", msg), **params
-        )
+        self.trade_cli.cancel_order(on_ack, **params, error_callback=on_err)
 
     def on_securities(self, securities: PySecurityList):
         self.logger.info("[bybit_gw_sec.json]%s" % securities)

src/lib_python_interop/python_interop_plugins/pybit_unified_trading.py

@@ -1,27 +1,28 @@
 import logging
 from dataclasses import dataclass
+from typing import Any
 from pybit.exceptions import (
     InvalidChannelTypeError,
     TopicMismatchError,
     UnauthorizedExceptionError,
 )
-from ._v5_misc import MiscHTTP
-from ._v5_market import MarketHTTP
-from ._v5_trade import TradeHTTP
-from ._v5_account import AccountHTTP
-from ._v5_asset import AssetHTTP
-from ._v5_position import PositionHTTP
-from ._v5_pre_upgrade import PreUpgradeHTTP
-from ._v5_spot_leverage_token import SpotLeverageHTTP
-from ._v5_spot_margin_trade import SpotMarginTradeHTTP
-from ._v5_user import UserHTTP
-from ._v5_broker import BrokerHTTP
-from ._v5_institutional_loan import InstitutionalLoanHTTP
-from ._v5_crypto_loan import CryptoLoanHTTP
-from ._v5_earn import EarnHTTP
-from ._websocket_stream import _V5WebSocketManager
-from ._websocket_trading import _V5TradeWebSocketManager
-from ._v5_spread import (
+from pybit._v5_misc import MiscHTTP
+from pybit._v5_market import MarketHTTP
+from pybit._v5_trade import TradeHTTP
+from pybit._v5_account import AccountHTTP
+from pybit._v5_asset import AssetHTTP
+from pybit._v5_position import PositionHTTP
+from pybit._v5_pre_upgrade import PreUpgradeHTTP
+from pybit._v5_spot_leverage_token import SpotLeverageHTTP
+from pybit._v5_spot_margin_trade import SpotMarginTradeHTTP
+from pybit._v5_user import UserHTTP
+from pybit._v5_broker import BrokerHTTP
+from pybit._v5_institutional_loan import InstitutionalLoanHTTP
+from pybit._v5_crypto_loan import CryptoLoanHTTP
+from pybit._v5_earn import EarnHTTP
+from pybit_websocket_stream import _V5WebSocketManager
+from pybit_websocket_trading import _V5TradeWebSocketManager
+from pybit._v5_spread import (
     SpreadHTTP,
     _V5WebSocketSpreadTrading,
 )
@@ -370,29 +371,46 @@ class WebSocketTrading(_V5TradeWebSocketManager):
     def __init__(self, recv_window=0, referral_id="", **kwargs):
         super().__init__(recv_window, referral_id, **kwargs)
 
+    @staticmethod
+    def _extract_error_callback(req_params: dict[str, Any]):
+        err_cb = req_params.pop("error_callback") if "error_callback" in req_params else None
+        return (req_params, err_cb)
+
     def place_order(self, callback, **kwargs):
         operation = "order.create"
-        self._send_order_operation(operation, callback, kwargs)
+        self._send_order_operation(
+            operation, callback, *self._extract_error_callback(kwargs)
+        )
 
     def amend_order(self, callback, **kwargs):
         operation = "order.amend"
-        self._send_order_operation(operation, callback, kwargs)
+        self._send_order_operation(
+            operation, callback, *self._extract_error_callback(kwargs)
+        )
 
     def cancel_order(self, callback, **kwargs):
         operation = "order.cancel"
-        self._send_order_operation(operation, callback, kwargs)
+        self._send_order_operation(
+            operation, callback, *self._extract_error_callback(kwargs)
+        )
 
     def place_batch_order(self, callback, **kwargs):
         operation = "order.create-batch"
-        self._send_order_operation(operation, callback, kwargs)
+        self._send_order_operation(
+            operation, callback, *self._extract_error_callback(kwargs)
+        )
 
     def amend_batch_order(self, callback, **kwargs):
         operation = "order.amend-batch"
-        self._send_order_operation(operation, callback, kwargs)
+        self._send_order_operation(
+            operation, callback, *self._extract_error_callback(kwargs)
+        )
 
     def cancel_batch_order(self, callback, **kwargs):
         operation = "order.cancel-batch"
-        self._send_order_operation(operation, callback, kwargs)
+        self._send_order_operation(
+            operation, callback, *self._extract_error_callback(kwargs)
+        )
 
 
 class WebsocketSpreadTrading(_V5WebSocketSpreadTrading):

src/lib_python_interop/python_interop_plugins/pybit_websocket_stream.py

@@ -2,11 +2,11 @@ import websocket
 import threading
 import time
 import json
-from ._http_manager import generate_signature
+from pybit._http_manager import generate_signature
 import logging
 import copy
 from uuid import uuid4
-from . import _helpers
+from pybit import _helpers
 
 
 logger = logging.getLogger(__name__)
@@ -62,6 +62,7 @@ class _WebSocketManager:
         #       "topic_name": function
         #   }
         self.callback_directory = {}
+        self.error_callback_directory = {}
 
         # Record the subscriptions made so that we can resubscribe if the WSS
         # connection is broken.

src/lib_python_interop/python_interop_plugins/pybit_websocket_trading.py

@@ -2,8 +2,8 @@ from dataclasses import dataclass, field
 import json
 import uuid
 import logging
-from ._websocket_stream import _WebSocketManager
-from . import _helpers
+from pybit_websocket_stream import _WebSocketManager
+from pybit import _helpers
 
 
 logger = logging.getLogger(__name__)
@@ -37,7 +37,10 @@ class _V5TradeWebSocketManager(_WebSocketManager):
             f"WebSocket request {message['reqId']} hit an error. Enabling "
             f"traceLogging to reproduce the issue. Raw error: {message}"
         )
-        self._pop_callback(message["reqId"])
+        req_id = message["reqId"]
+        self._pop_callback(req_id)
+        if req_id in self.error_callback_directory:
+            self.error_callback_directory.pop(req_id)(message)
 
     def _handle_incoming_message(self, message):
         def is_auth_message():
@@ -63,10 +66,16 @@ class _V5TradeWebSocketManager(_WebSocketManager):
     def _set_callback(self, topic, callback_function):
         self.callback_directory[topic] = callback_function
 
+    def _set_error_callback(self, topic, callback_function):
+        self.error_callback_directory[topic] = callback_function
+
     def _pop_callback(self, topic):
         return self.callback_directory.pop(topic)
 
-    def _send_order_operation(self, operation, callback, request):
+    def _pop_error_callback(self, topic):
+        return self.error_callback_directory.pop(topic)
+
+    def _send_order_operation(self, operation, callback, request, error_callback=None):
         request_id = str(uuid.uuid4())
 
         message = {
@@ -87,3 +96,5 @@ class _V5TradeWebSocketManager(_WebSocketManager):
 
         self.ws.send(json.dumps(message))
         self._set_callback(request_id, callback)
+        if error_callback:
+            self._set_error_callback(request_id, error_callback)

src/lib_python_interop/python_interop_plugins/spark_enums.py

@@ -17,3 +17,27 @@ class ExecType(Enum):
     FILL = 4
     CANCELED = 5
     REJECTED = 6
+
+class OrderType(Enum):
+    LIMIT = 1
+    MARKET = 2
+    ICEBERG = 3
+    SPREADER = 4
+
+    @staticmethod
+    def parse(s: str):
+        return OrderType.__members__[s]
+
+
+class TimeInForce(Enum):
+    FOK = 1
+    GTC = 2
+
+    @staticmethod
+    def parse(s: str):
+        return TimeInForce.__members__[s]
+
+class SecurityType(Enum):
+    SPOT = 1
+    FUTURE = 2
+    OPTION = 3