diff --git a/src/app_sparkd/configurations/bybit_dev/symbols.conf b/src/app_sparkd/configurations/bybit_dev/symbols.conf index 4942b8294..184ade8e9 100644 --- a/src/app_sparkd/configurations/bybit_dev/symbols.conf +++ b/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" } } } diff --git a/src/lib_python_md/python_md_providers/bybit_conf.py b/src/lib_python_md/python_md_providers/bybit_conf.py index 36b9c37ea..87acff5ce 100644 --- a/src/lib_python_md/python_md_providers/bybit_conf.py +++ b/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', diff --git a/src/lib_python_md/python_md_providers/bybit_md_provider.py b/src/lib_python_md/python_md_providers/bybit_md_provider.py index 9300722d0..6380e15a5 100644 --- a/src/lib_python_md/python_md_providers/bybit_md_provider.py +++ b/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 ()', '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.") \ No newline at end of file + 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) diff --git a/src/lib_python_md/python_md_providers/config.py b/src/lib_python_md/python_md_providers/config.py index f3b5ef10a..b0e10333d 100644 --- a/src/lib_python_md/python_md_providers/config.py +++ b/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"