refactor: Manage all py plugins (md and exe) in one class

All calls to/from python are performed on one thread via a simple
message queue (necessary because of the global interpreter lock).

src/lib_python_interop/python_interop_plugins/bybit_execution_gateway.py

@@ -5,6 +5,7 @@ import uuid
 from pybit.unified_trading import WebSocket, WebSocketTrading
 from py_execution_utils import load_gateway_config
 from spark_core import (
+    JsonElement,
     as_bool,
     configure_logger,
     format_price,
@@ -23,6 +24,7 @@ from python_interop_module import (
     PySecurity,
     PySecurityList,
 )
+from spark_enums import OrderStatus, ExecType
 import threading
 
 
@@ -36,7 +38,8 @@ class OrderData:
         self.last_filled_quantity: float = 0
         self.avg_filled_price: float = 0
         self.last_filled_price: float = 0
-        self.exec_type = "PENDING_NEW"
+        self.exec_type: ExecType = ExecType.PENDING_NEW
+        self.status: OrderStatus = OrderStatus.PENDING_NEW
 
     def on_update(
         self,
@@ -63,19 +66,29 @@ class OrderData:
         self.filled_quantity = filled_quantity
         self.avg_filled_price = avg_filled_price
         match status:
+            case "New":
+                self.exec_type = ExecType.NEW
+                self.status = OrderStatus.NEW
             case "Filled":
-                self.exec_type = "FILL"
-            case "Canceled":
-                self.exec_type = "CANCELED"
+                self.exec_type = ExecType.FILL
+                self.status = OrderStatus.FILLED
+            case "PartiallyFilled":
+                self.exec_type = ExecType.PARTIAL_FILL
+                self.status = OrderStatus.PARTIALLY_FILLED
+            case "Cancelled":
+                self.exec_type = ExecType.CANCELED
+                self.status = OrderStatus.CANCELED
             case "Rejected":
-                self.exec_type = "REJECTED"
+                self.exec_type = ExecType.REJECTED
+                self.status = OrderStatus.REJECTED
             case _:
-                self.exec_type = "NEW" if filled_quantity == 0 else "PARTIAL_FILL"
+                raise ValueError(f"Unexpected order status '{status}'")
 
 
 class BybitExecutionGateway(ExecutionGateway):
-    def __init__(self, name: str):
+    def __init__(self):
         super(BybitExecutionGateway, self).__init__()
+        name = "bybit_1"
         self.name = name
         self.logger = configure_logger(logging.getLogger(name), "OrderMan")
         self.logger.info("[bybit_gw_ctr.json]%s" % SLog(name=name))
@@ -135,15 +148,6 @@ class BybitExecutionGateway(ExecutionGateway):
                     total_filled_qty = parse_float(d["cumExecQty"])
                     avg_filled_price = parse_float(d["avgPrice"])
                     status: str = d["orderStatus"]
-                    # brits can't spell
-                    if status == "Cancelled":
-                        status = "Canceled"
-                    exec_type = ord_data.exec_type
-                    match ord_data.exec_type:
-                        case "PARTIAL_FILL":
-                            exec_type = "PARTIALLY_FILLED"
-                        case _:
-                            pass
                     px = parse_float(d["price"])
                     qty = parse_float(d["qty"])
                     ord_data.on_update(
@@ -162,8 +166,8 @@ class BybitExecutionGateway(ExecutionGateway):
                         price=px,
                         quantity=qty,
                         time_in_force=d["timeInForce"],
-                        exec_type=exec_type,
-                        order_status=status,
+                        exec_type=ord_data.exec_type.name,
+                        order_status=ord_data.status.name,
                         last_filled_price=ord_data.last_filled_price,
                         last_filled_quantity=ord_data.last_filled_quantity,
                         filled_quantity=total_filled_qty,
@@ -198,7 +202,7 @@ class BybitExecutionGateway(ExecutionGateway):
         params = {
             "api_key": self.api_key,
             "api_secret": self.secret_key,
-            "testnet": True,
+            "testnet": self.use_testnet,
         }
         self.logger.info("[bybit_gw_start_trd_cli.json]%s" % json.dumps(params))
         self.trade_cli = WebSocketTrading(**params)
@@ -206,6 +210,25 @@ class BybitExecutionGateway(ExecutionGateway):
             "category": "linear",
         }
         self.bridge.on_is_ready(True)
+        # Sleep loop to keep thread alive and detect connection status changes
+        conn_statuses = {
+            "stream": self.stream_cli.is_connected(),
+            "trade": self.trade_cli.is_connected(),
+        }
+        while not self.stop_event.is_set():
+            time.sleep(1)
+            new_conn_statuses = {
+                "stream": self.stream_cli.is_connected(),
+                "trade": self.trade_cli.is_connected(),
+            }
+            if conn_statuses != new_conn_statuses:
+                conn_statuses = new_conn_statuses
+                self.logger.log(
+                    logging.INFO
+                    if all(map(lambda kv: kv[1], conn_statuses.items()))
+                    else logging.WARNING,
+                    "[bybit_gw_connection_status.json]%s" % JsonElement(conn_statuses),
+                )
 
     def start(self, bridge: ExecutionGatewayBridge):
         def start_wrapper():

src/lib_python_interop/python_interop_plugins/spark_enums.py

@@ -0,0 +1,19 @@
+from enum import Enum
+
+
+class OrderStatus(Enum):
+    NEW = 1
+    PENDING_NEW = 2
+    PARTIALLY_FILLED = 3
+    FILLED = 4
+    CANCELED = 5
+    REJECTED = 6
+
+
+class ExecType(Enum):
+    NEW = 1
+    PENDING_NEW = 2
+    PARTIAL_FILL = 3
+    FILL = 4
+    CANCELED = 5
+    REJECTED = 6