gate changes

.vscode/launch.json

@@ -1,14 +1,6 @@
 {
   "version": "0.2.0",
   "configurations": [
-    {
-      "name": "Python Debugger: Current File with Arguments",
-      "type": "debugpy",
-      "request": "launch",
-      "program": "${file}",
-      "console": "integratedTerminal",
-      "args": "${command:pickArgs}"
-    },
     {
       "name": ".NET Core Attach",
       "type": "coreclr",
@@ -38,12 +30,12 @@
       "name": "(gdb) Launch",
       "type": "cppdbg",
       "request": "launch",
-      "program": "/home/bts/spark/build/sparkd-debug-clang/server/sparkd",
+      "program": "${workspaceFolder}/build/sparkd-debug-clang/src/app_sparkd/sparkd",
       "args": [
         "run"
       ],
       "stopAtEntry": false,
-      "cwd": "/home/bts/spark/build/sparkd-debug-clang/server",
+      "cwd": "${workspaceFolder}/build/sparkd-debug-clang/src/app_sparkd/",
       "environment": [],
       "externalConsole": false,
       "MIMode": "gdb",

CMakePresets.json

@@ -76,7 +76,7 @@
         },
         "BUILD_ADMIN_WEB_APP": {
           "type": "BOOL",
-          "value": "ON"
+          "value": "OFF"
         },
         "SHARED_MEMORY_PREFIX": "sparkd"
       }

src/app_sparkd/configurations/gateio_dev/.env

@@ -1 +1,4 @@
-SPARKD__CONFIG_DIR=src/app_sparkd/configurations/py_dummy_md
+SPARKD__CONFIG_DIR=src/app_sparkd/configurations/gateio_dev
+SPARKD__PYTHON_REQUIREMENTS=src/app_sparkd/configurations/gateio_dev/requirements.txt
+SPARKD__IMAGE=sparkd_symbols
+SPARKD__RUNTIME_CONF_DIR=/home/bts/8000/sparkdata/conf

src/app_sparkd/configurations/gateio_dev/py_gateio_1.json

@@ -0,0 +1,19 @@
+{
+  "gateway_type": "gateio_execution_gateway.GateioExecutionGateway",
+  "account": "GATEIO_DEV",
+  "module_path": [
+    "python_interop_plugins"
+  ],
+  "venue": "VENUE_GATEIO",
+  "old_api_key":"b05a32a37fb21bd719a3a61901a8d0dd" ,
+  "old_secret_key": "ef776a256285823e98a8d1dd897324cdb05e933b9b39723c8f8d6bdb0af2dcb7",
+  "use_testnet": true,
+  "api_key": "be5478ae23219292fbe9597439b4201d",
+  "secret_key": "ecc8bf77a2a3109c553cd194d5bd066211bca9a51ffbcf4e3c3db285620065a8",
+  "ws_base_url":"wss://ws-testnet.gate.io",
+  "ws_usdt_futures":"/v4/ws/futures/usdt",
+  "logging": {
+    "console_level": "DEBUG",
+    "file_level": "DEBUG"
+  }
+}

src/app_sparkd/configurations/gateio_dev/requirements.txt

@@ -0,0 +1,11 @@
+certifi==2025.4.26
+charset-normalizer==3.4.2
+dotenv==0.9.9
+idna==3.10
+numpy==2.2.6
+pybit==5.11.0
+pycryptodome==3.23.0
+python-dotenv==1.1.0
+requests==2.32.4
+urllib3==2.4.0
+websocket-client==1.8.0

src/app_sparkd/configurations/gateio_dev/spark.conf

@@ -36,3 +36,30 @@ MarketDataConfig {
     }
   }
 }
+
+NotSimExecution {
+  value {
+    entries {
+      venue: VENUE_GATEIO
+    }
+  }
+}
+JetFIX {
+  value {
+    connection {
+      name: "gateio_1"
+      venue: VENUE_GATEIO
+      max_order_msgs_per_window: 300
+      window_duration_msecs: 1000
+    }
+  }
+}
+Ports {
+   value {
+      config_port: 8000
+      qbus_port: 8001
+      glbx_sim_fix_port: 8002
+      ice_sim_fix_port: 8003
+      replicator_port: 8080
+   }
+}

src/app_sparkd/sparkd/md/databento/DatabentoDataHandler.cpp

@@ -271,7 +271,6 @@ void DatabentoDatahandler::run(
    auto process_bbo = [&](const databento::Record &record) {
       const auto &msg = record.Get<databento::Mbp1Msg>();
       debug_log(msg);
-
       // std::vector<std::tuple<uint64_t,std::string>>& times = all_times.emplace_back();
       // times.reserve(8);
       // return databento::KeepGoing::Continue; //only added so the same functions work with the threaded builder, not really need in the blocking version

src/lib_python_interop/python_interop/plugin_manager.cpp

@@ -102,8 +102,11 @@ struct __attribute__((visibility("hidden"))) PluginManager_Impl {
             sd.import("sys").attr("path").attr("insert")(
                 0, "python_interop_plugins");
             sd.import("spark_core").attr("configure_default_logger")();
-            sd.import("bybit_md_provider");
-            sd.import("bybit_execution_gateway");
+            // sd.import("bybit_md_provider");
+            // sd.import("bybit_execution_gateway");
+            sd.import("gateio_md_provider");
+            sd.import("gateio_execution_gateway");
+            
             py::gil_scoped_release gil_rel;
             Stopwatch sw;
             int n = 0;

src/lib_python_interop/python_interop_plugins/gateio_execution_gateway.py

@@ -0,0 +1,521 @@
+# gateio_execution_gateway.py
+
+import json
+import hmac
+import hashlib
+import time
+import uuid
+import threading
+import websocket
+import logging
+
+from python_interop_module import (
+    ExecutionGateway,
+    ExecutionGatewayBridge,
+    PyCreateOrderRequest,
+    PyCancelReplaceOrderRequest,
+    PyCancelOrderRequest,
+    PyExecutionReport,
+    PySecurity,
+    PySecurityList,
+)
+from spark_enums import SecurityType, OrderType, TimeInForce, OrderStatus, ExecType
+from spark_core import JsonElement, as_bool, configure_logger, format_price, format_quantity , load_config, StructuredLog as SLog
+from typing import Optional
+
+# WS_URL = "wss://fx-ws-testnet.gateio.ws/v4/ws/usdt"
+# WS_URL = "wss://ws-testnet.gate.io/v4/ws/futures/usdt"
+class OrderData:
+    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
+        self.last_filled_quantity: float = 0
+        self.avg_filled_price: float = 0
+        self.last_filled_price: float = 0
+        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="GATEIO",
+            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 gateio, so we use updatedTime
+            transact_id=str(transact_time),
+            transact_time=transact_time,
+        )
+        return er
+
+    def on_update(
+        self,
+        price: float,
+        quantity: float,
+        filled_quantity: float,
+        avg_filled_price: float,
+        status: str,
+    ):
+        self.price = price
+        self.filled_quantity = 0
+        self.quantity = quantity
+        lfq = filled_quantity - self.filled_quantity
+        if lfq > 0:
+            if self.filled_quantity == 0:
+                self.last_filled_quantity = filled_quantity
+                self.last_filled_price = avg_filled_price
+            else:
+                price_diff = avg_filled_price - self.avg_filled_price
+                lfq_weight = filled_quantity / (self.filled_quantity + filled_quantity)
+                lfp = self.avg_filled_price + (price_diff / lfq_weight)
+                self.last_filled_price = lfp
+                self.last_filled_quantity = lfq
+        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 = 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 = ExecType.REJECTED
+                self.status = OrderStatus.REJECTED
+            case _:
+                raise ValueError(f"Unexpected order status '{status}'")
+        return self.to_execution_report()
+
+
+class GateioExecutionGateway(ExecutionGateway):
+    def __init__(self):
+        super(GateioExecutionGateway,self).__init__()
+        name = "gateio_1"
+        self.name = "gateio_1"
+        self.prefix = self.make_session_prefix()
+        cfg = load_config([f"py_{name}"]) 
+        self.logger = configure_logger(logging.getLogger(self.name), "OrderMan", **cfg.get("logging", {}))
+        self.logger.info("[gateio_gw_ctr.json]%s" % SLog(name=name))
+        self.account: str = cfg["account"]
+        self.api_key: str = cfg["api_key"]
+        self.ws_usdt_futures_url =cfg["ws_base_url"]+cfg["ws_usdt_futures"]
+        self.secret_key: str = cfg["secret_key"]
+        self.use_testnet: bool = cfg.get("use_testnet", True)
+        self.use_trace_logging: bool = cfg.get("use_trace_logging", False)
+
+        self.orders = {}
+        self.securities = {}
+        self.ws = None
+        self.stop_event = threading.Event()
+        self.orders: dict[int, OrderData] = {}
+        self.securities: dict[str, PySecurity] = {}
+        # self.bridge: Optional[ExecutionGatewayBridge] = None
+
+        self.ping_thread = None
+        self.ping_thread_stop = threading.Event()
+
+    def start(self, bridge: ExecutionGatewayBridge):
+        self.bridge = bridge
+        self.logger.info("[gateio_gw_start.json]%s" % json.dumps({"ts": time.time_ns()}))
+        thread = threading.Thread(target=self._run_ws, daemon=True)
+        thread.start()
+
+    def stop(self):
+        self.logger.info("[gateio_gw_stop]")
+        self.stop_event.set()
+        self.ping_thread_stop.set()
+        if self.ws:
+            self.ws.close()
+        if self.ping_thread:
+            self.ping_thread.join(timeout=2)
+    def _ws_send(self, msg: dict, label: str = "send"):
+        formatted = json.dumps(msg, separators=(",", ":"), ensure_ascii=False)
+        self.logger.info(f"[gateio_gw_ws_{label}] >> {formatted}")
+        if self.ws:
+            self.ws.send(formatted)
+    def normalize_message(self,msg: dict) -> dict:
+        if "header" in msg:
+            base = {
+                "type": "response",
+                "channel": msg["header"].get("channel"),
+                "event": msg["header"].get("event"),
+                "status": msg["header"].get("status"),
+                "trace_id": msg["header"].get("trace_id"),
+                "response_time": int(msg["header"].get("response_time")),
+                "request_id": msg.get("request_id"),
+            }
+            if "errs" in msg.get("data", {}):
+                base["error"] = msg["data"]["errs"]
+            else:
+                base["data"] = msg["data"].get("result", {})
+            return base
+        else:
+            return {
+                "type": "update",
+                "time": msg.get("time"),
+                "channel": msg.get("channel"),
+                "event": msg.get("event"),
+                "trace_id": msg.get("trace_id"),
+                "data": msg.get("result", msg.get("payload", {})),
+            }
+    def _ws_receive(self, raw_msg: str):
+        if "futures.pong" in raw_msg:
+            # self.logger.debug(f"[gateio_gw_pong] {raw_msg}")
+            return None
+        self.logger.info(f"[gateio_gw_ws_recv] << {raw_msg}")
+        try:
+            msg = json.loads(raw_msg)
+        except json.JSONDecodeError as e:
+            self.logger.error(f"[gateio_gw_ws_recv_error] Failed to parse message: {e}")
+            return None
+        return self.normalize_message(msg)
+    def _ping_loop(self):
+            self.logger.info(f"[gateio_gw_ping] ping loop started")
+            while not self.ping_thread_stop.is_set():
+                try:
+                    ts = int(time.time())
+                    ping_msg = {"time": ts, "event": "ping", "channel" : "futures.ping"}
+                    if self.ws:
+                        self.ws.send(json.dumps(ping_msg))
+                        # self.logger.debug(f"[gateio_gw_ping] {ping_msg}")
+                except Exception as e:
+                    self.logger.warning(f"[gateio_gw_ping_err] {e}", exc_info=True)
+                self.ping_thread_stop.wait(10)  # Ping every 10 seconds
+    def start_ping(self):
+            self.ping_thread_stop.clear()
+            self.ping_thread = threading.Thread(target=self._ping_loop, daemon=True)
+            self.ping_thread.start()
+    def make_session_prefix(self):
+        alphabet = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789._"
+        base = len(alphabet)
+
+        def custom_base64_encode(value: int) -> str:
+            if value == 0:
+                return alphabet[0]
+            result = []
+            while value:
+                result.append(alphabet[value % base])
+                value //= base
+            return ''.join(reversed(result))
+
+    # Use current time in milliseconds
+        now_millis = int(time.time() * 1_000)
+        return "t-"+custom_base64_encode(now_millis) +"-"
+    def _get_order_data(self, exch_id: str):
+        for _, v in self.orders.items():
+            if v.exch_id == exch_id:
+                return v
+        return None
+    def _run_ws(self):
+        def on_open(ws):
+            self.logger.info("WebSocket opened")
+            payload = self._build_login_payload()
+            self.logger.info(f"Sending login: {payload}")
+            self._ws_send(payload, label="login")
+        
+        
+        def on_message(ws, msg):
+            data = self._ws_receive(msg)
+            if not data:
+                return  # Invalid message format
+
+            # Handle login
+            if data.get("channel") == "futures.login" and data.get("event") == "api":
+                if data.get("status") != "200":
+                    self.logger.error(f"[gateio_gw_login_err] cant login msg {data}")
+                    return
+                self._subscribe_to_orders()
+                self.start_ping()
+                return
+
+            # Handle order subscription success
+            if data.get("channel") == "futures.orders":
+                if data.get("event") == "subscribe":
+                    if  data.get("data",{}).get("status",{})=="success":
+                        self.bridge.on_is_ready(True)
+                        return
+                    else:#log for now decide what to do later
+                        self.logger.error(f"[gateio_gw_ord_sub_err.json] Could not subscribe to orders, msg: {data}")
+                        
+
+                # Handle order update
+                if data.get("event") == "update":
+                    self.logger.info(f"[gateio_gw_order_update.json]{json.dumps(data)}")
+                    try:
+                        for upd in data.get("result", []):
+                            exch_id = upd.get("text")
+                            ord_data = self._get_order_data(exch_id)
+                            if not ord_data:
+                                self.logger.error(f"[gateio_gw_ord_upd_err.json] Unknown exch_id {exch_id}")
+                                continue
+
+                            total_filled_qty = float(ord_data.quantity - upd["left"])
+                            last_filled_qty = total_filled_qty - ord_data.filled_quantity
+                            last_filled_px = float(upd["fill_price"]) if last_filled_qty > 0 else 0
+                            ord_data.on_update(
+                                price=float(upd["price"]),
+                                quantity=float(upd["size"]),
+                                filled_quantity=total_filled_qty,
+                                avg_filled_price=float(upd.get("fill_price", 0.0)),
+                                status=upd["status"]
+                            )
+
+                            transact_time = int(upd.get("finish_time_ms", upd["create_time_ms"])) * 1_000_000
+
+                            er = PyExecutionReport()
+                            er.set_data(
+                                key=ord_data.bts_id,
+                                exchange_order_id=str(upd["id"]),
+                                venue="GATE",
+                                security_type="FUTURE",
+                                exchange_security_id=upd["contract"],
+                                account=self.account,
+                                order_type=str(ord_data.order_type),
+                                is_buy=ord_data.is_buy,
+                                price=float(upd["price"]),
+                                quantity=float(upd["size"]),
+                                time_in_force=upd["tif"],
+                                exec_type=ord_data.exec_type.name,
+                                order_status=ord_data.status.name,
+                                last_filled_price=last_filled_px,
+                                last_filled_quantity=last_filled_qty,
+                                filled_quantity=total_filled_qty,
+                                avg_filled_price=float(upd.get("fill_price", 0.0)),
+                                transact_id=str(transact_time),
+                                transact_time=transact_time,
+                            )
+                            self.logger.info(f"[gateio_gw_exe_rep_out.json]{er}")
+                            self.bridge.on_execution_report(er)
+                    except Exception as e:
+                        self.logger.error(f"[gateio_gw_ord_upd_err.json]{{\"err\": \"{e}\"}}", exc_info=True)
+                    return
+
+            # Handle order ack or error from futures.order_place
+            if data.get("channel") == "futures.order_place" and data.get("event") == "api":
+                req_id  = data.get("request_id", "")
+                exch_id = self.prefix + req_id.split("-", 1)[0]
+                
+                ord_data = self._get_order_data(exch_id)
+                if not ord_data:
+                    self.logger.error(f"[gateio_gw_ord_ack_err.json] No order data found for {exch_id}, msg: {data}")
+                    return
+
+                status = data.get("status")
+                if status == "200":
+                    # ack is true and valid
+                    er = ord_data.to_execution_report()
+                    er.set_order_status(OrderStatus.NEW.name)
+                    er.set_exec_type(ExecType.NEW.name)
+                    self.bridge.on_execution_report(er)
+                    self.logger.info(f"[gateio_gw_ord_ack.json]{json.dumps(data)}")
+                    return
+
+                elif status == "400":
+                    err_info = data.get("error", {})
+                    er = ord_data.to_execution_report()
+                    er.set_order_status(OrderStatus.REJECTED.name)
+                    er.set_exec_type(ExecType.REJECTED.name)
+                    er.set_error_msg(f'{err_info.get("label")}: {err_info.get("message")}')
+                    self.bridge.on_execution_report(er)
+                    self.logger.error(f"[gateio_gw_ord_rej.json]{json.dumps(data)}")
+                    return
+
+            self.logger.error(f"[gateio_gw_ping_err] unhandled message type {data}")
+
+        def on_error(ws, error):
+            self.logger.error(f"WebSocket error: {error}")
+
+        def on_close(ws, code, msg):
+            self.logger.warning(f"WebSocket closed: {code} {msg}")
+
+        self.ws = websocket.WebSocketApp(
+            self.ws_usdt_futures_url,
+            on_open=on_open,
+            on_message=on_message,
+            on_error=on_error,
+            on_close=on_close
+        )
+
+        self.ws.run_forever()
+    def gen_sign(self,channel, event, timestamp):
+
+        s = 'channel=%s&event=%s&time=%d' % (channel, event, timestamp)
+        sign = hmac.new(self.secret_key.encode('utf-8'), s.encode('utf-8'), hashlib.sha512).hexdigest()
+        return {'method': 'api_key', 'KEY': self.api_key, 'SIGN': sign}
+    def _subscribe_to_orders(self):
+        sub_payload = {
+            "time": int(time.time()),
+            "channel": "futures.orders",
+            "event": "subscribe",
+            "payload": ['!all']
+        }
+        sub_payload['auth'] = self.gen_sign(sub_payload['channel'], sub_payload['event'], sub_payload['time'])
+        self.logger.info(f"Subscribing to order updates: {sub_payload}")
+        if self.ws:
+            self._ws_send(sub_payload, label="subscribe_orders")
+    def _build_login_payload(self):
+        ts = int(time.time()) - 4
+        ts_ms = int(time.time() * 1000) - 4000
+        payload = b""
+        key = f"api\nfutures.login\n{payload.decode()}\n{ts}"
+        sign = hmac.new(self.secret_key.encode(), key.encode(), hashlib.sha512).hexdigest()
+
+        return {
+            "time": ts,
+            "channel": "futures.login",
+            "event": "api",
+            "payload": {
+                "api_key": self.api_key,
+                "signature": sign,
+                "timestamp": str(ts),
+                "req_id": f"{ts_ms}-1"
+            }
+        }
+    def _on_abort(
+        self,
+        key: int,
+        req_type: str,
+        rsn: str,
+    ):
+        self.logger.error(
+            '[gateio_gw_abort.json]{ "key": %d, "typ": "%s", "rsn": "%s", "ts": %d }'
+            % (key, req_type, rsn, time.time_ns())
+        )
+    def create_order(self, req):
+        self.logger.info(f"[gateio_gw_create_order] {req}")
+        if not self.bridge:
+            self._on_abort(
+                req.key(),
+                "crt",
+                "WebSocket not initialized ( bridge: %s) req: "+req.__str__()
+                % as_bool(self.bridge),
+            )
+            return
+        bts_id = req.key()
+        px = req.price()
+        qty = req.quantity()
+        sec_name = req.exchange_security_id()
+
+        self.orders[bts_id] = OrderData(
+            account=self.account,
+            bts_id=bts_id,
+            exch_id=self.prefix+str(bts_id),
+            exch_sec_id=sec_name,
+            sec_type=SecurityType.parse(req.security_type()),
+            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:
+            self._on_abort(
+                bts_id,
+                "crt",
+                "sec not found req: " + req.__str__()
+            )
+            return
+
+        # Construct order parameters
+        order_payload = {
+            "contract": sec_name,
+            "size": int(qty),  # Gate expects integer size
+            "price": format_price(sec, px),
+            "tif": req.time_in_force().lower(),
+            "iceberg": 0,
+            "text": self.prefix+str(bts_id)
+        }
+        if not req.is_buy():
+            order_payload["size"] = -order_payload["size"]
+
+        ts = int(time.time())
+        msg = {
+            "time": ts,
+            "channel": "futures.order_place",
+            "event": "api",
+            "payload": {
+                "req_id": f"{bts_id}-{ts}",
+                "req_param": order_payload
+            }
+        }
+        msg["auth"] = self.gen_sign("futures.order_place", "api", ts)
+
+        self.logger.info("[gateio_gw_crt_ord_send.json]%s" % SLog(params=msg))
+        self._ws_send(msg, label="create_order")
+
+    def cancel_order(self, req):
+        self.logger.info(f"[gateio_gw_cancel_order] {req}")
+        # TODO: cancel via order ID
+
+    def cancel_replace_order(self, req):
+        self.logger.info(f"[gateio_gw_cancel_replace_order] {req}")
+        # TODO: implement amend order logic
+
+    def on_securities(self, sec_list):
+        self.logger.info(f"[gateio_gw_securities] {sec_list}")
+        self.securities = {sec.name(): sec for sec in sec_list}

src/lib_python_interop/python_interop_plugins/gateio_md_provider.py

@@ -1,5 +1,4 @@
 import json
-import numpy as np
 import time
 import logging
 import sys
@@ -28,6 +27,7 @@ from spark_core import (
 #   prod perps (ws) - "wss://fx-ws.gateio.ws/v4/ws/usdt"
 #   testnet perps (ws) - "wss://fx-ws-testnet.gateio.ws/v4/ws/usdt"
 
+import numpy as np
 logger = logging.getLogger("gateio_md_provider")
 
 def fetch_contract_specs(contracts: Iterable[str]) -> dict:
@@ -67,12 +67,17 @@ def fetch_contract_specs(contracts: Iterable[str]) -> dict:
 class GateIoMdProvider(MdProvider):
     def __init__(self):
         super().__init__()
+        self._config: Optional[dict] = None
         self.ws = None
         self.ws_thread: Optional[threading.Thread] = None
         self.bridge: Optional[MdBridge] = None
         self.stop_event = threading.Event()
         self.ping_thread = None
-
+    @property
+    def config(self):
+        if self._config is None:
+            self._config = load_config(["py_gateio_md_provider", "gateio_overrides"])
+        return self._config
     def send_ping(self):
         """Send ping messages every 15 seconds to keep connection alive."""
         while not self.stop_event.is_set():
@@ -91,7 +96,7 @@ class GateIoMdProvider(MdProvider):
             time.sleep(15)
 
     def start_impl(self, bridge: MdBridge):
-        cfg = load_config(["gateio_md_provider", "gateio_overrides"])
+        cfg = self.config
         configure_logger(logger, "md", **cfg.get("logging", {}))
         self.bridge = bridge
         ws_url = cfg["ws_url"]
@@ -108,7 +113,7 @@ class GateIoMdProvider(MdProvider):
         # Fetch all contract specs at once
         spots: list[str] = cfg["spots"]
         currencies: list[str] = cfg.get("currencies") or ["USDT"]
-        contracts = list(chain(*[map(lambda s: f"{s}{c}", spots) for c in currencies]))
+        contracts = list(chain(*[map(lambda s: f"{s}_{c}", spots) for c in currencies]))
         contract_specs = fetch_contract_specs(contracts)
 
         # 1) Send security info
@@ -191,6 +196,9 @@ class GateIoMdProvider(MdProvider):
                         logger.debug(f"📘 BOOK DATA for {contract}: bids={bids} asks={asks}")
 
                         book = PyBook()
+                      
+                        bids+= [(np.float64('nan'), np.float64('nan'))]
+                        asks+= [(np.float64('nan'), np.float64('nan'))]
                         book.set_data(
                             contract,
                             "GateIO",
@@ -208,7 +216,7 @@ class GateIoMdProvider(MdProvider):
         except WebSocketConnectionClosedException:
             logger.info("WebSocket closed.")
         except Exception as e:
-            logger.error("Fatal error in listen loop:", exc_info=True)
+            logger.error("Fatal error in listen loop: {}",e, exc_info=True)
 
     def start(self, bridge: MdBridge):
         self.ws_thread = threading.Thread(target=lambda: self.start_impl(bridge))
@@ -234,3 +242,47 @@ class GateIoMdProvider(MdProvider):
             self.ws_thread.join()
         self.bridge = None
         logger.info("✔ Stopped GateIoMdProvider.")
+
+if __name__ == "__main__":
+    import argparse
+    from fake_md_bridge import FakeMdBridge
+
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--currencies", nargs="+")
+    ap.add_argument("--futures", nargs="+")
+    log_levels = list(logging._nameToLevel)
+    ap.add_argument("--console-level", choices=log_levels)
+    ap.add_argument("--file-level", choices=log_levels)
+    ap.add_argument("--dev", action="store_true", dest="use_testnet")
+    ap.add_argument("--prod", action="store_false", dest="use_testnet")
+    ap.set_defaults(
+        currencies=None,
+        futures=None,
+        console_level=None,
+        file_level=None,
+        use_testnet=None,
+    )
+    args = ap.parse_args()
+    print(args)
+    logger = logging.getLogger("gateio_md_provider_app")
+    md_pro = GateIoMdProvider()
+    config = md_pro.config
+    if args.currencies:
+        config["currencies"] = args.currencies
+    if args.futures:
+        config["futures"] = args.futures
+    logging_overrides = {}
+    if args.console_level:
+        logging_overrides["console_level"] = args.console_level
+    if args.file_level:
+        logging_overrides["file_level"] = args.file_level
+    if logging_overrides:
+        config["logging"].update(logging_overrides)
+    if args.use_testnet is not None:
+        config["use_testnet"] = args.use_testnet
+    # print(json.dumps(config))
+    md_bridge = FakeMdBridge(logger)
+    md_pro.start(md_bridge)  # type: ignore
+    while True:
+        time.sleep(1)
+

src/lib_python_interop/python_interop_plugins/spark_core.py

@@ -301,7 +301,6 @@ def format_price(sec: PySecurity, value: float) -> str:
     decimals = get_price_decimal_places(sec)
     return f"{value:.{decimals}f}"
 
-
 def format_quantity(sec: PySecurity, value: float) -> str:
     decimals = get_qty_decimal_places(sec)
     return f"{value:.{decimals}f}"

src/lib_python_interop/python_interop_plugins/trade_client_cli.py

@@ -4,6 +4,8 @@ import re
 import sys
 import time
 import traceback
+from gateio_execution_gateway import GateioExecutionGateway
+from gateio_md_provider import GateIoMdProvider
 from rich.console import Console
 from typing import Optional
 
@@ -142,12 +144,16 @@ class VenueService:
             match venue.lower():
                 case "bybit":
                     self.md_provider = BybitMdProvider()
+                case "gateio":
+                    self.md_provider= GateIoMdProvider()
                 case _:
                     pass
         if is_execution_enabled:
             match venue.lower():
                 case "bybit":
                     self.execution_gw = BybitExecutionGateway()
+                case "gateio":
+                    self.execution_gw= GateioExecutionGateway()
                 case _:
                     pass
         if self.md_provider:
@@ -216,7 +222,8 @@ def run(
     handler = PromptToolkitHandler()
     handler.setFormatter(ConsoleLogFormatter("Gen"))
     logger.addHandler(handler)
-
+    print("hello",venues)
+    assert(False)
     svcs = {
         v: VenueService(
             v,