chore: Add config json for py plugins

Support overrides per venue in a separate file, so symbols can be
changed after deployment using arham's widget.

src/lib_python_interop/python_interop_plugins/bybit_execution_gateway.py

@@ -3,13 +3,13 @@ import time
 from typing import Optional
 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,
     format_quantity,
+    load_config,
     parse_float,
     StructuredLog as SLog,
 )
@@ -90,9 +90,11 @@ class BybitExecutionGateway(ExecutionGateway):
         super(BybitExecutionGateway, self).__init__()
         name = "bybit_1"
         self.name = name
-        self.logger = configure_logger(logging.getLogger(name), "OrderMan")
+        cfg = load_config([f"py_{name}"])
+        self.logger = configure_logger(
+            logging.getLogger(name), "OrderMan", **(cfg.get("logging") or {})
+        )
         self.logger.info("[bybit_gw_ctr.json]%s" % SLog(name=name))
-        cfg = load_gateway_config(name)
         self.account: str = cfg["account"]
         self.api_key: str = cfg["api_key"]
         self.secret_key: str = cfg["secret_key"]
@@ -333,4 +335,3 @@ class BybitExecutionGateway(ExecutionGateway):
             "[bybit_gw_sec.json][%s]" % ", ".join([str(sec) for sec in securities])
         )
         self.securities = {sec.name(): sec for sec in securities}
-

src/lib_python_interop/python_interop_plugins/bybit_md_provider.py

@@ -5,7 +5,8 @@ import logging
 import sys
 import threading
 import signal
-from typing import Optional
+from itertools import chain
+from typing import Iterable, Optional
 from pybit.unified_trading import WebSocket
 from python_interop_module import (
     MdBridge,
@@ -14,26 +15,25 @@ from python_interop_module import (
     PyBookList,
     PySecurityList,
 )
-from bybit_conf import CONTRACTS
 from spark_core import (
     configure_logger,
     get_contract_spec,
     get_price_info,
     get_quantity_info,
     get_security,
+    load_config,
 )
 from py_md_utils import PERP_EXP_STRING
 import requests
 
 logger = logging.getLogger("bybit_md_provider")
-configure_logger(logger, "md")
 
 def normalize_contract_name(contract: str) -> str:
     if len(contract) > 4:
         return contract[:-4] + "_" + contract[-4:]
     return contract
 
-def fetch_contract_specs() -> dict:
+def fetch_contract_specs(contracts: Iterable[str], use_testnet: bool = True) -> dict:
     default_specs = {
         "tick_unit": 1,
         "tick_base": 10,
@@ -41,16 +41,17 @@ def fetch_contract_specs() -> dict:
         "qty_tick_unit": 1,
         "qty_tick_base": 10000
     }
-    specs_dict = {contract: default_specs for contract in CONTRACTS}
+    specs_dict = {contract: default_specs for contract in contracts}
     try:
-        url = "https://api.bybit.com/v5/market/instruments-info?category=linear"
+        host = "api-testnet.bybit.com" if use_testnet else "api.bybit.com"
+        url = f"https://{host}/v5/market/instruments-info?category=linear"
         response = requests.get(url, timeout=5)
         response.raise_for_status()
         data = response.json()
         found_contracts = set()
         for item in data.get("result", {}).get("list", []):
             contract = item.get("symbol")
-            if contract in CONTRACTS:
+            if contract in contracts:
                 price_tick = float(item.get("priceFilter", {}).get("tickSize", "0.1"))
                 lot_size = float(item.get("lotSizeFilter", {}).get("qtyStep", "0.001"))
                 tick_base = int(1 / price_tick) if price_tick != 0 else 10
@@ -66,7 +67,7 @@ def fetch_contract_specs() -> dict:
                     "qty_tick_base": qty_tick_base
                 }
                 found_contracts.add(contract)
-        for contract in CONTRACTS:
+        for contract in contracts:
             if contract not in found_contracts:
                 logger.warning(f"Contract {contract} not found in API response, using default specs")
         return specs_dict
@@ -77,6 +78,7 @@ def fetch_contract_specs() -> dict:
 class BybitMdProvider(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
@@ -93,14 +95,40 @@ class BybitMdProvider(MdProvider):
         signal.signal(signal.SIGTERM, signal_handler)
         signal.signal(signal.SIGTSTP, signal_handler)
 
+    @property
+    def config(self):
+        if self._config is None:
+            self._config = load_config(["py_bybit_md_provider", "bybit_overrides"])
+        return self._config
+
+    @property
+    def use_testnet(self):
+        return self.config.get("use_testnet", True)
+
+    @property
+    def currencies(self):
+        return self.config.get("currencies") or ["USDT"]
+
+    @property
+    def futures(self):
+        return self.config.get("futures", [])
+
+    @property
+    def spots(self):
+        return self.config.get("spots", [])
+
     def start_impl(self, bridge: MdBridge):
+        configure_logger(logger, "md", **self.config.get("logging", {}))
         self.bridge = bridge
         logger.info("Starting BybitMdProvider")
 
-        contract_specs = fetch_contract_specs()
+        contracts = list(
+            chain(*[map(lambda f: f"{f}{c}", self.futures) for c in self.currencies])
+        )
+        contract_specs = fetch_contract_specs(contracts, use_testnet=self.use_testnet)
         sec_list = PySecurityList()
 
-        for contract in CONTRACTS:
+        for contract in contracts:
             normalized_contract = normalize_contract_name(contract)
             asset, quote = normalized_contract.split("_")
             specs = contract_specs.get(contract, {})
@@ -150,13 +178,13 @@ class BybitMdProvider(MdProvider):
 
         try:
             self.ws = WebSocket(
-                testnet=False,
+                testnet=self.use_testnet,
                 channel_type="linear",
                 ping_interval=20,
             )
             logger.info("Connecting to Bybit WebSocket (mainnet, linear)")
 
-            for contract in CONTRACTS:
+            for contract in contracts:
                 try:
                     self.ws.orderbook_stream(
                         symbol=contract,

src/lib_python_interop/python_interop_plugins/gateio_md_provider.py

@@ -5,8 +5,9 @@ import logging
 import sys
 import threading
 import requests
+from itertools import chain
 from websocket import create_connection, WebSocketConnectionClosedException
-from typing import Optional, Any
+from typing import Iterable, Optional, Any
 from python_interop_module import (
     MdBridge,
     MdProvider,
@@ -14,19 +15,22 @@ from python_interop_module import (
     PyBookList,
     PySecurityList,
 )
-from config import WS_URL, CONTRACTS
 from spark_core import (
     configure_logger,
     get_contract_spec,
     get_price_info,
     get_quantity_info,
     get_security,
+    load_config,
 )
 
-logger = logging.getLogger("gateio_md_provider")
-configure_logger(logger, "md")
+# Urls:
+#   prod perps (ws) - "wss://fx-ws.gateio.ws/v4/ws/usdt"
+#   testnet perps (ws) - "wss://fx-ws-testnet.gateio.ws/v4/ws/usdt"
 
-def fetch_contract_specs() -> dict:
+logger = logging.getLogger("gateio_md_provider")
+
+def fetch_contract_specs(contracts: Iterable[str]) -> dict:
     """Fetch specs for all contracts in one request."""
     default_specs = {
         "tick_unit": 1,
@@ -35,7 +39,7 @@ def fetch_contract_specs() -> dict:
         "qty_tick_unit": 1,
         "qty_tick_base": 10000
     }
-    specs_dict = {contract: default_specs for contract in CONTRACTS}
+    specs_dict = {contract: default_specs for contract in contracts}
     try:
         url = "https://api.gateio.ws/api/v4/futures/usdt/contracts"
         response = requests.get(url, timeout=5)
@@ -43,7 +47,7 @@ def fetch_contract_specs() -> dict:
         data = response.json()
         for item in data:
             contract = item.get("name")
-            if contract in CONTRACTS:
+            if contract in contracts:
                 price_tick = float(item.get("price_tick", "0.1"))
                 contract_size = float(item.get("contract_size", "0.0001"))
                 tick_base = int(1 / price_tick) if price_tick != 0 else 10
@@ -87,9 +91,12 @@ class GateIoMdProvider(MdProvider):
             time.sleep(15)
 
     def start_impl(self, bridge: MdBridge):
+        cfg = load_config(["gateio_md_provider", "gateio_overrides"])
+        configure_logger(logger, "md", **cfg.get("logging", {}))
         self.bridge = bridge
-        logger.info(f"Connecting to {WS_URL}")
-        self.ws = create_connection(WS_URL)
+        ws_url = cfg["ws_url"]
+        logger.info(f"Connecting to {ws_url}")
+        self.ws = create_connection(ws_url)
         self.ws.settimeout(5)
         logger.info("Connected.")
 
@@ -99,11 +106,14 @@ class GateIoMdProvider(MdProvider):
         logger.info("Started ping thread")
 
         # Fetch all contract specs at once
-        contract_specs = fetch_contract_specs()
+        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]))
+        contract_specs = fetch_contract_specs(contracts)
 
         # 1) Send security info
         sec_list = PySecurityList()
-        for contract in CONTRACTS:
+        for contract in contracts:
             asset, quote = contract.split("_")
             specs = contract_specs[contract]
             spec = get_contract_spec(
@@ -134,7 +144,7 @@ class GateIoMdProvider(MdProvider):
         logger.info("Sent securities list--sink.on_securities called.")
 
         # 2) Subscribe to order books
-        for contract in CONTRACTS:
+        for contract in contracts:
             sub_msg = {
                 "time": int(time.time() * 1000),
                 "channel": "futures.order_book",

src/lib_python_interop/python_interop_plugins/py_execution_utils.py

@@ -1,12 +0,0 @@
-import json
-import string
-import os
-
-
-def load_gateway_config(name: str, expand_vars: bool = True):
-    s = ""
-    with open(f"/home/bts/sparkdata/conf/py_{name}.json") as f:
-        s = f.read()
-    if expand_vars:
-        s = string.Template(s).substitute(os.environ)
-    return json.loads(s)

src/lib_python_interop/python_interop_plugins/spark_core.py

@@ -6,7 +6,8 @@ import os
 import sys
 import threading
 import time
-from typing import Callable, Optional
+from functools import reduce
+from typing import Any, Callable, Optional
 import websocket
 from python_interop_module import (
     PyContractPriceInfo,
@@ -98,20 +99,59 @@ def as_bool(*args):
     (True if arg else False for arg in args)
 
 
-def configure_logger(logger: logging.Logger, subsystem: str):
-    stdout_handler = logging.StreamHandler(sys.stdout)
-    file_handler = logging.FileHandler(
-        f"/home/bts/sparkdata/data/logging.sparkd/{logger.name}_log"
+def _load_config_json(file_name: str, expand_vars: bool = True) -> Optional[dict]:
+    if not file_name.endswith(".json"):
+        file_name = file_name + ".json"
+    s = ""
+    try:
+        with open(f"/home/bts/sparkdata/conf/{file_name}") as f:
+            s = f.read()
+    except FileNotFoundError:
+        return None
+    if expand_vars:
+        try:
+            s = string.Template(s).substitute(os.environ)
+        except KeyError:
+            return None
+    return json.loads(s)
+
+
+def load_config(names: list[str], expand_vars: bool = True):
+    def apply_overrides(cfg: dict, overrides: dict):
+        cfg.update(overrides)
+        return cfg
+
+    return reduce(
+        apply_overrides, [_load_config_json(n, expand_vars) or {} for n in names]
     )
+
+
+def configure_logger(
+    logger: logging.Logger,
+    subsystem: str,
+    **kwargs,
+):
+    logger.propagate = False
+    # Console
+    stdout_handler = logging.StreamHandler(sys.stdout)
+    stdout_handler.setLevel(kwargs.get("console_level", logging.WARNING))
     stdout_handler.formatter = ConsoleLogFormatter(subsystem=subsystem)
+    # File
+    file_handler = logging.FileHandler(
+        f"/home/bts/sparkdata/data/logging.sparkd/{logger.name or 'py'}_log"
+    )
+    file_handler.setLevel(kwargs.get("file_level", logging.INFO))
     file_handler.formatter = FileLogFormatter(subsystem=subsystem)
     logger.handlers.clear()
     logger.addHandler(stdout_handler)
     logger.addHandler(file_handler)
-    logger.setLevel(logging.INFO)
+    logger.setLevel(min(map(lambda h: h.level, logger.handlers)))
     return logger
 
 
+def configure_default_logger(**kwargs):
+    configure_logger(logging.getLogger(), "Gen", **kwargs)
+
 def parse_float(s: Optional[str], default_val=math.nan):
     if s is not None:
         try: