ADD: BYBIT MD Working, NaN's also Added.

src/app_sparkd/configurations/bybit_dev/symbols.conf

@@ -3,6 +3,61 @@ SymbolLists {
     list {
       key: "Bybit"
       symbol: "BTC"
+      symbol: "ETH"    
+      symbol: "BNB"
+      symbol: "XRP"
+      symbol: "ADA"
+      symbol: "PRIME"
+      # symbol: "AAVE"
+      # symbol: "ADA"
+      # symbol: "ALGO"
+      # symbol: "APT"
+      # symbol: "ARB"
+      # symbol: "ATOM"
+      # symbol: "AVAX"
+      # symbol: "AXS"
+      # symbol: "BCH"
+      # symbol: "BNB"
+      # symbol: "BONK"
+      # symbol: "BTC"
+      # symbol: "CHZ"
+      # symbol: "CRV"
+      # symbol: "DOGE"
+      # symbol: "DOT"
+      # symbol: "EGLD"
+      # symbol: "ETH"
+      # symbol: "FLOKI"
+      # symbol: "FTM"
+      # symbol: "GALA"
+      # symbol: "GRT"
+      # symbol: "HBAR"
+      # symbol: "ICP"
+      # symbol: "INJ"
+      # symbol: "JASMY"
+      # symbol: "KAVA"
+      # symbol: "LINK"
+      # symbol: "LTC"
+      # symbol: "MANA"
+      # symbol: "MKR"
+      # symbol: "NEAR"
+      # symbol: "NOT"
+      # symbol: "OP"
+      # symbol: "PENDLE"
+      # symbol: "PEPE"
+      # symbol: "POL"
+      # symbol: "RUNE"
+      # symbol: "SAND"
+      # symbol: "SHIB"
+      # symbol: "SOL"
+      # symbol: "SUI"
+      # symbol: "TON"
+      # symbol: "TRX"
+      # symbol: "UNI"
+      # symbol: "VET"
+      # symbol: "WLD"
+      # symbol: "XLM"
+      # symbol: "XRP"
+      # symbol: "ZIL"
     }
   }
 }

src/lib_python_md/python_md_providers/bybit_conf.py

@@ -1,10 +1,76 @@
+# # Bybit API configuration
+# BYBIT_API_KEY = 'your_api_key'  # Not needed for public orderbooks
+# BYBIT_API_SECRET = 'your_api_secret'  # Not needed for public orderbooks
+# BYBIT_TESTNET = False  # Use mainnet
+# WS_URL = "wss://stream.bybit.com/v5/public/linear"  # Mainnet linear perpetual futures
+# REST_API_URL = "https://api.bybit.com"  # Mainnet REST API
+
+# # Contracts with underscores for C++ frontend compatibility
+# CONTRACTS = ["BTC_USDT", "ETH_USDT", "BNB_USDT", "XRP_USDT", "ADA_USDT"]
+
+
 # Bybit API configuration
 BYBIT_API_KEY = 'your_api_key'
 BYBIT_API_SECRET = 'your_api_secret'
 BYBIT_TESTNET = True  # Set to False for mainnet
 WS_URL = "wss://stream-testnet.bybit.com/v5/public/linear"
 
-CONTRACTS = ['BTCUSDT']
+CONTRACTS = ["BTCUSDT", "ETHUSDT", "BNBUSDT", "XRPUSDT", "ADAUSDT", "PRIMEUSDT"]
+
+# CONTRACTS = [
+#     'BTCUSDT',
+#     'ETHUSDT',
+#     'SOLUSDT',
+#     'BNBUSDT',
+#     'ADAUSDT',
+#     'XRPUSDT',  # Added based on prior example
+#     'DOGEUSDT',
+#     'TRXUSDT',
+#     'LINKUSDT',
+#     'DOTUSDT',
+#     'AVAXUSDT',
+#     'MATICUSDT',  # POLUSDT on Bybit
+#     'SHIB1000USDT',
+#     'LTCUSDT',
+#     'BCHUSDT',
+#     'NEARUSDT',
+#     'APTUSDT',
+#     'INJUSDT',
+#     'FTMUSDT',  # Not in list but commonly paired; assuming available
+#     'UNIUSDT',
+#     'AAVEUSDT',
+#     'OPUSDT',
+#     'ARBUSDT',
+#     'SUIUSDT',
+#     'TONUSDT',
+#     'ATOMUSDT',  # Not in list but commonly paired; assuming available
+#     'XLMUSDT',   # Not in list but commonly paired; assuming available
+#     'ALGOUSDT',  # Not in list but commonly paired; assuming available
+#     'VETUSDT',   # Not in list but commonly paired; assuming available
+#     'ICPUSDT',
+#     'HBARUSDT',
+#     'GRTUSDT',
+#     'EGLDUSDT',
+#     'SANDUSDT',
+#     'MANAUSDT',
+#     'AXSUSDT',
+#     'CHZUSDT',
+#     'CRVUSDT',
+#     'MKRUSDT',
+#     'PEPE1000USDT',
+#     'FLOKI1000USDT',
+#     'BONK1000USDT',
+#     'WLDUSDT',   # Not in list but commonly paired; assuming available
+#     'NOTUSDT',
+#     'JASMYUSDT',
+#     'GALAUSDT',
+#     'PENDLEUSDT',
+#     'ZILUSDT',   # Not in list but commonly paired; assuming available
+#     'KAVAUSDT',
+#     'RUNEUSDT'
+# ]
+
+
 # CONTRACTS = [
 #     '1000000BABYDOGEUSDT',
 #     '1000000CHEEMSUSDT',
@@ -308,9 +374,9 @@ CONTRACTS = ['BTCUSDT']
 #     'MOCAUSDT',
 #     'MOODENGUSDT',
 #     'MORPHOUSDT',
-#     'MOTHER1USDT',
-#     'MOTHER2USDT',
-#     'MOTHER3USDT',
+#     # 'MOTHER1USDT',
+#     # 'MOTHER2USDT',
+#     # 'MOTHER3USDT',
 #     'MOVEUSDT',
 #     'MOVRUSDT',
 #     'MTLUSDT',
@@ -333,7 +399,7 @@ CONTRACTS = ['BTCUSDT']
 #     'OBOLUSDT',
 #     'OBTUSDT',
 #     'OGUSDT',
-#     'OKBUSDT',
+#     # 'OKBUSDT',
 #     'OLUSDT',
 #     'OMGUSDT',
 #     'OMUSDT',
@@ -354,7 +420,7 @@ CONTRACTS = ['BTCUSDT']
 #     'PEAQUSDT',
 #     'PENDLEUSDT',
 #     'PEOPLEUSDT',
-#     'PERPUSDT',
+#     # 'PERPUSDT',
 #     'PHAUSDT',
 #     'PHBUSDT',
 #     'PIPPINUSDT',
@@ -371,8 +437,8 @@ CONTRACTS = ['BTCUSDT']
 #     'POWRUSDT',
 #     'PRAIUSDT',
 #     'PRCLUSDT',
-#     'PRE05EOSUSDT',
-#     'PRE08EOSUSDT',
+#     # 'PRE05EOSUSDT',
+#     # 'PRE08EOSUSDT',
 #     'PRIMEUSDT',
 #     'PROMPTUSDT',
 #     'PROMUSDT',

src/lib_python_md/python_md_providers/bybit_md_provider.py

@@ -1,8 +1,14 @@
+import json
+import time
 import logging
 import sys
-from bybit_conf import CONTRACTS
+import threading
+import signal
+from pybit.unified_trading import WebSocket
 import python_md_module as pymd
-from py_md_utils import get_contract_spec, get_price_info, get_quantity_info, get_security
+from bybit_conf import CONTRACTS
+from py_md_utils import get_contract_spec, get_price_info, get_quantity_info, get_security, get_book_side
+import requests
 
 # Logging setup
 LOG_PATH = "/home/bts/sparkdata/data/bybit_md_provider.log"
@@ -17,63 +23,222 @@ formatter = logging.Formatter("%(asctime)s|%(levelname)s|%(message)s")
 console.setFormatter(formatter)
 logging.getLogger().addHandler(console)
 
+def normalize_contract_name(contract: str) -> str:
+    if len(contract) > 4:
+        return contract[:-4] + "_" + contract[-4:]
+    return contract
+
+def fetch_contract_specs() -> dict:
+    default_specs = {
+        "tick_unit": 1,
+        "tick_base": 10,
+        "cash_per_point": 0.1,
+        "qty_tick_unit": 1,
+        "qty_tick_base": 10000
+    }
+    specs_dict = {contract: default_specs for contract in CONTRACTS}
+    try:
+        url = "https://api.bybit.com/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:
+                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
+                if lot_size >= 1:
+                    qty_tick_base = int(lot_size)
+                else:
+                    qty_tick_base = int(1 / lot_size) if lot_size != 0 else 10000
+                specs_dict[contract] = {
+                    "tick_unit": 1,
+                    "tick_base": tick_base,
+                    "cash_per_point": price_tick,
+                    "qty_tick_unit": 1,
+                    "qty_tick_base": qty_tick_base
+                }
+                found_contracts.add(contract)
+        for contract in CONTRACTS:
+            if contract not in found_contracts:
+                logging.warning(f"Contract {contract} not found in API response, using default specs")
+        return specs_dict
+    except Exception as e:
+        logging.error(f"Failed to fetch contract specs: {e}", exc_info=True)
+        return specs_dict
+
 class BybitMdProvider(pymd.MdProvider):
     def __init__(self):
         super().__init__()
+        self.ws = None
         self.sink = None
+        self.stop_event = threading.Event()
+        self._setup_signal_handlers()
+
+    def _setup_signal_handlers(self):
+        """Set up signal handlers for SIGINT, SIGTERM, and SIGTSTP."""
+        def signal_handler(sig, frame):
+            logging.info(f"Received signal {sig}, stopping...")
+            self.stop()
+        
+        signal.signal(signal.SIGINT, signal_handler)
+        signal.signal(signal.SIGTERM, signal_handler)
+        signal.signal(signal.SIGTSTP, signal_handler)
 
     def start(self, sink: pymd.MdSink):
         self.sink = sink
         logging.info("Starting BybitMdProvider")
 
-        # Use default contract specs for all contracts
-        default_specs = {
-            "tick_unit": 1,
-            "tick_base": 10,
-            "cash_per_point": 0.1,
-            "qty_tick_unit": 1,
-            "qty_tick_base": 10000
-        }
-
-        # Send security info
+        contract_specs = fetch_contract_specs()
         sec_list = pymd.PySecurityList()
+
         for contract in CONTRACTS:
-            # Handle contracts with multipliers or PERP suffix
-            asset = contract.replace("USDT", "").replace("PERP", "")
-            # Remove numeric prefixes (e.g., 1000PEPE -> PEPE)
-            for prefix in ["1000000", "10000", "1000"]:
-                if asset.startswith(prefix):
-                    asset = asset[len(prefix):]
-            quote = "USDT"
-            spec = get_contract_spec(
-                price_info=get_price_info(
-                    tick_unit=default_specs["tick_unit"],
-                    tick_base=default_specs["tick_base"],
-                    cash_per_point=default_specs["cash_per_point"]
-                ),
-                qty_info=get_quantity_info(
-                    tick_unit=default_specs["qty_tick_unit"],
-                    tick_base=default_specs["qty_tick_base"]
+            normalized_contract = normalize_contract_name(contract)
+            asset, quote = normalized_contract.split("_")
+            specs = contract_specs.get(contract, {})
+
+            try:
+                qty_info = get_quantity_info(
+                    tick_unit=specs.get("qty_tick_unit", 1),
+                    tick_base=specs.get("qty_tick_base", 10000)
                 )
-            )
-            sec = get_security(
-                venue="Bybit",
-                name=contract,
-                asset=asset,
-                group=asset,
-                currency=quote,
-                expiration=None,
-                option_kind=None,
-                strike=None,
-                spec=spec,
-            )
-            sec_list.append(sec)
-            logging.info(f"Created security: {contract} with specs: {default_specs}")
+            except Exception as e:
+                logging.error(f"Failed to get quantity info for contract: {contract}, error: {e}")
+                continue
+
+            try:
+                spec = get_contract_spec(
+                    price_info=get_price_info(
+                        tick_unit=specs.get("tick_unit", 1),
+                        tick_base=specs.get("tick_base", 10),
+                        cash_per_point=specs.get("cash_per_point", 0.1)
+                    ),
+                    qty_info=qty_info
+                )
+            except Exception as e:
+                logging.error(f"Failed to get contract spec for contract: {contract}, error: {e}")
+                continue
+
+            try:
+                sec = get_security(
+                    venue="Bybit",
+                    name=contract,
+                    asset=asset,
+                    group=asset,
+                    currency=quote,
+                    expiration=None,
+                    option_kind=None,
+                    strike=None,
+                    spec=spec,
+                )
+                sec_list.append(sec)
+                logging.info(f"Created security: {contract} with specs: {specs}")
+            except Exception as e:
+                logging.error(f"Failed to create security for contract: {contract}, error: {e}")
+                continue
+
         sink.on_securities(sec_list)
         logging.info("Sent securities list--sink.on_securities called.")
 
+        try:
+            self.ws = WebSocket(
+                testnet=False,
+                channel_type="linear",
+                ping_interval=20,
+            )
+            logging.info("Connecting to Bybit WebSocket (mainnet, linear)")
+
+            for contract in CONTRACTS:
+                try:
+                    self.ws.orderbook_stream(
+                        symbol=contract,
+                        depth=50,
+                        callback=self._on_book_update
+                    )
+                    logging.info(f"Subscribed to orderbook.50.{contract}")
+                except Exception as e:
+                    logging.error(f"Failed to subscribe to orderbook for {contract}, error: {e}")
+
+            logging.info("Subscribed to orderbook channels on Bybit unified trading WS.")
+
+            for t in threading.enumerate():
+                logging.debug(f"Thread (start): Name={t.name}, Alive={t.is_alive()}, Daemon={t.daemon}, Target={t._target}")
+
+            while not self.stop_event.is_set():
+                time.sleep(0.05)
+
+        except KeyboardInterrupt:
+            logging.info("KeyboardInterrupt received — shutting down.")
+            self.stop()
+        except Exception as e:
+            logging.error(f"Bybit WebSocket fatal error: {e}", exc_info=True)
+            self.stop()
+
+    def _on_book_update(self, data: dict):
+        try:
+            topic = data.get("topic", "")
+            if not topic.startswith("orderbook.50."):
+                return
+
+            contract = topic.split(".")[-1]
+            payload = data.get("data", {})
+            raw_bids = [(float(x[0]), float(x[1])) for x in payload.get("b", [])]
+            raw_asks = [(float(x[0]), float(x[1])) for x in payload.get("a", [])]
+
+            # Sort and take top 5, padding with [float('nan'), float('nan')] if needed
+            sorted_bids = sorted(raw_bids, key=lambda x: -x[0])[:5]
+            sorted_bids += [[float('nan'), float('nan')]] * (5 - len(sorted_bids))
+            sorted_asks = sorted(raw_asks, key=lambda x: x[0])[:5]
+            sorted_asks += [[float('nan'), float('nan')]] * (5 - len(sorted_asks))
+
+            logging.debug(f"📘 BOOK DATA for {contract}: bids={sorted_bids} asks={sorted_asks}")
+
+            book = pymd.PyBook()
+            book.set_venue("Bybit")
+            book.set_security_type("SPOT")
+            book.set_exchange_security_id(contract)
+            book.set_bids(get_book_side(sorted_bids, True))
+            book.set_asks(get_book_side(sorted_asks, False))
+
+            book_list = pymd.PyBookList()
+            book_list.append(book)
+            if self.sink:
+                self.sink.on_books(book_list)
+                logging.info(f"✔ on_books called for {contract}")
+            else:
+                logging.warning("Sink is None, cannot call on_books")
+        except Exception as e:
+            logging.error(f"Error processing book update for {contract}: {e}", exc_info=True)
+
     def stop(self):
-        """Cleanup."""
         logging.info("Stopping BybitMdProvider...")
+        self.stop_event.set()
+
+        logging.debug("Threads before stopping WebSocket:")
+        for t in threading.enumerate():
+            logging.debug(f"Thread(stop): {t.name}, Alive: {t.is_alive()}, Daemon: {t.daemon}, Target={t._target}")
+
+        if self.ws:
+            try:
+                self.ws.exit()
+                logging.info("WebSocket exit called")
+                # Clean up pybit threads with longer timeout
+                for t in threading.enumerate():
+                    if t.name in ['Thread-1 (<lambda>)', 'Thread-2 (_send_ping)'] and t.is_alive():
+                        logging.info(f"Joining pybit thread: {t.name}")
+                        t.join(timeout=1.0)
+            except Exception as e:
+                logging.error(f"Error while closing WebSocket: {e}", exc_info=True)
+            self.ws = None
+
         self.sink = None
-        logging.info("✔ Stopped BybitMdProvider.")
+        logging.info("✔ Stopped BybitMdProvider.")
+
+        logging.debug("Threads after stop:")
+        for t in threading.enumerate():
+            logging.debug(f"Thread(after stop): {t.name}, Alive: {t.is_alive()}, Daemon: {t.daemon}, Target={t._target}")
+
+        import os
+        os._exit(0)

src/lib_python_md/python_md_providers/config.py

@@ -1,5 +1,6 @@
 # Gate.io WebSocket configuration
-WS_URL = "wss://fx-ws-testnet.gateio.ws/v4/ws/usdt"  # Testnet url for perp futures 
+# WS_URL = "wss://fx-ws-testnet.gateio.ws/v4/ws/usdt"  # Testnet url for perp futures 
+WS_URL = "wss://fx-ws.gateio.ws/v4/ws/usdt"  # Mainnet url for perp futures 
 
 # API credentials
 API_KEY = "e4a0c6b198549042b13faa7645640d60"