refactor: Consolidate all py interop into single lib
It was just too hard trying to make it work with separate libs. The main issue was getting the python md and execution modules to recognize PySecurityList and other types defined in the core module.
This commit is contained in:
John Colvin committed
1 parent
7c9330650d
commit
bdd5e08b60
87 files changed
+1093
-2926
No files matched your search
@@ -0,0 +1,313 @@
|
||||
import json
|
||||
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 (
|
||||
as_bool,
|
||||
configure_logger,
|
||||
format_price,
|
||||
format_quantity,
|
||||
parse_float,
|
||||
StructuredLog as SLog,
|
||||
)
|
||||
import logging
|
||||
from python_interop_module import (
|
||||
ExecutionGateway,
|
||||
ExecutionGatewayBridge,
|
||||
PyCreateOrderRequest,
|
||||
PyCancelReplaceOrderRequest,
|
||||
PyCancelOrderRequest,
|
||||
PyExecutionReport,
|
||||
PySecurity,
|
||||
PySecurityList,
|
||||
)
|
||||
import threading
|
||||
|
||||
|
||||
class OrderData:
|
||||
def __init__(self, bts_id: int, exch_id: str, price: float, quantity: float):
|
||||
self.bts_id = bts_id
|
||||
self.exch_id = exch_id
|
||||
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 = "PENDING_NEW"
|
||||
|
||||
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 "Filled":
|
||||
self.exec_type = "FILL"
|
||||
case "Canceled":
|
||||
self.exec_type = "CANCELED"
|
||||
case "Rejected":
|
||||
self.exec_type = "REJECTED"
|
||||
case _:
|
||||
self.exec_type = "NEW" if filled_quantity == 0 else "PARTIAL_FILL"
|
||||
|
||||
|
||||
class BybitExecutionGateway(ExecutionGateway):
|
||||
def __init__(self, name: str):
|
||||
super(BybitExecutionGateway, self).__init__()
|
||||
self.name = name
|
||||
self.logger = configure_logger(logging.getLogger(name), "OrderMan")
|
||||
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"]
|
||||
self.use_testnet: bool = cfg.get("use_testnet", True)
|
||||
self.use_trace_logging: bool = cfg.get("use_trace_logging", False)
|
||||
self.stop_event = threading.Event()
|
||||
self.stream_cli: Optional[WebSocket] = None
|
||||
self.trade_cli: Optional[WebSocketTrading] = None
|
||||
# bts id -> exch id
|
||||
self.orders: dict[int, OrderData] = {}
|
||||
self.securities: dict[str, PySecurity] = {}
|
||||
|
||||
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 _on_order_ack(self, typ, msg: object):
|
||||
self.logger.info(
|
||||
'[bybit_gw_ord_ack.json]{ "typ": "%s", "msg": %s}' % (typ, msg)
|
||||
)
|
||||
|
||||
def _on_abort(
|
||||
self,
|
||||
key: int,
|
||||
req_type: str,
|
||||
rsn: str,
|
||||
):
|
||||
self.logger.error(
|
||||
'[bybit_gw_abort.json]{ "key": %d, "typ": "%s", "rsn": "%s", "ts": %d }'
|
||||
% (key, req_type, rsn, time.time_ns())
|
||||
)
|
||||
|
||||
def start_impl(self, bridge: ExecutionGatewayBridge):
|
||||
def on_order_update(msg: dict):
|
||||
def on_error(err: str):
|
||||
self.logger.error(
|
||||
'[bybit_gw_ord_upd_err.json]{ "err": "%s", "ts": %d }'
|
||||
% (err, time.time_ns())
|
||||
)
|
||||
|
||||
self.logger.info("[bybit_gw_ord_upd.json]%s" % msg)
|
||||
try:
|
||||
upds: list[dict] = msg["data"]
|
||||
for d in upds:
|
||||
er = PyExecutionReport()
|
||||
exch_id: str = d["orderLinkId"]
|
||||
ord_data: Optional[OrderData] = self._get_order_data(exch_id)
|
||||
if not ord_data:
|
||||
on_error("Did not resolve exch order id %s" % exch_id)
|
||||
return
|
||||
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(
|
||||
px, qty, total_filled_qty, avg_filled_price, status
|
||||
)
|
||||
transact_time = int(d["updatedTime"]) * 1_000_000
|
||||
er.set_data(
|
||||
key=ord_data.bts_id,
|
||||
exchange_order_id=d["orderLinkId"],
|
||||
venue="BYBIT",
|
||||
security_type="FUTURE",
|
||||
exchange_security_id=d["symbol"],
|
||||
account=self.account,
|
||||
order_type=d["orderType"],
|
||||
is_buy=d["side"] == "Buy",
|
||||
price=px,
|
||||
quantity=qty,
|
||||
time_in_force=d["timeInForce"],
|
||||
exec_type=exec_type,
|
||||
order_status=status,
|
||||
last_filled_price=ord_data.last_filled_price,
|
||||
last_filled_quantity=ord_data.last_filled_quantity,
|
||||
filled_quantity=total_filled_qty,
|
||||
avg_filled_price=avg_filled_price,
|
||||
# There is no transact_id in Bybit, so we use updatedTime
|
||||
transact_id=str(transact_time),
|
||||
transact_time=transact_time,
|
||||
)
|
||||
self.logger.info("[bybit_gw_exe_rep_out.json]%s" % er)
|
||||
bridge.on_execution_report(er)
|
||||
except Exception as e:
|
||||
self.logger.error(
|
||||
'[bybit_gw_ord_upd_err.json]{ "err": %s }' % e, exc_info=True
|
||||
)
|
||||
|
||||
def on_exe_report(msg):
|
||||
self.logger.info("[bybit_gw_exe_rep.json]%s" % msg)
|
||||
|
||||
# Stream client (receives order updates)
|
||||
params = {
|
||||
"api_key": self.api_key,
|
||||
"api_secret": self.secret_key,
|
||||
"testnet": self.use_testnet,
|
||||
"channel_type": "private",
|
||||
"trace_logging": False,
|
||||
}
|
||||
self.logger.info("[bybit_gw_stream_sub.json]%s" % json.dumps(params))
|
||||
self.stream_cli = WebSocket(**params)
|
||||
self.stream_cli.order_stream(on_order_update)
|
||||
self.stream_cli.execution_stream(on_exe_report)
|
||||
# Trade client (sends orders)
|
||||
params = {
|
||||
"api_key": self.api_key,
|
||||
"api_secret": self.secret_key,
|
||||
"testnet": True,
|
||||
}
|
||||
self.logger.info("[bybit_gw_start_trd_cli.json]%s" % json.dumps(params))
|
||||
self.trade_cli = WebSocketTrading(**params)
|
||||
params = {
|
||||
"category": "linear",
|
||||
}
|
||||
self.bridge.on_is_ready(True)
|
||||
|
||||
def start(self, bridge: ExecutionGatewayBridge):
|
||||
def start_wrapper():
|
||||
try:
|
||||
self.start_impl(bridge)
|
||||
except Exception as e:
|
||||
self.logger.error(
|
||||
'[bybit_gw_start_err.json]{ "err": "%s" }' % str(e), exc_info=True
|
||||
)
|
||||
|
||||
self.logger.info("[bybit_gw_start.json]%s" % json.dumps({"ts": time.time_ns()}))
|
||||
self.bridge = bridge
|
||||
threading.Thread(
|
||||
target=start_wrapper,
|
||||
name=f"{self.name}_main",
|
||||
daemon=True,
|
||||
).start()
|
||||
|
||||
def stop(self):
|
||||
self.logger.info("[bybit_gw_stop.json]%s" % json.dumps({"ts": time.time_ns()}))
|
||||
self.stop_event.set()
|
||||
|
||||
def create_order(self, req: PyCreateOrderRequest):
|
||||
self.logger.info("[bybit_gw_crt_ord.json]%s" % str(req))
|
||||
if not self.stream_cli or not self.trade_cli or not self.bridge:
|
||||
self._on_abort(
|
||||
req.key(),
|
||||
"crt",
|
||||
"WebSocket not initialized (stream_cli: %s, trade_cli: %s, bridge: %s)"
|
||||
% as_bool(self.stream_cli, self.trade_cli, self.bridge),
|
||||
)
|
||||
return
|
||||
exch_ord_id = uuid.uuid1().hex
|
||||
bts_id = req.key()
|
||||
px = req.price()
|
||||
qty = req.quantity()
|
||||
self.orders[req.key()] = OrderData(
|
||||
bts_id=bts_id, exch_id=exch_ord_id, price=px, quantity=qty
|
||||
)
|
||||
sec_name = req.exchange_security_id()
|
||||
sec = self.securities.get(sec_name)
|
||||
if not sec:
|
||||
return
|
||||
params = {
|
||||
"orderLinkId": exch_ord_id,
|
||||
"category": "linear",
|
||||
"symbol": sec_name,
|
||||
"side": "Buy" if req.is_buy() else "Sell",
|
||||
"orderType": req.order_type().title(),
|
||||
"price": format_price(sec, px),
|
||||
"qty": format_quantity(sec, qty),
|
||||
}
|
||||
self.logger.info("[bybit_gw_crt_ord_send.json]%s" % json.dumps(params))
|
||||
self.trade_cli.place_order(lambda msg: self._on_order_ack("crt", msg), **params)
|
||||
|
||||
def cancel_replace_order(self, req: PyCancelReplaceOrderRequest):
|
||||
self.logger.info("[bybit_gw_cxlrep_ord.json]%s" % req)
|
||||
if not self.stream_cli or not self.trade_cli:
|
||||
self._on_abort(req.key(), "cxlrep", "WebSocket client not created.")
|
||||
return
|
||||
bts_id = req.key()
|
||||
ord_data = self.orders.get(bts_id)
|
||||
if not ord_data:
|
||||
self._on_abort(req.key(), "cxlrep", "Did not resolve exch order id")
|
||||
return
|
||||
sec_name = req.exchange_security_id()
|
||||
sec = self.securities.get(sec_name)
|
||||
if not sec:
|
||||
return
|
||||
params = {
|
||||
"orderLinkId": ord_data.exch_id,
|
||||
"category": "linear",
|
||||
"symbol": req.exchange_security_id(),
|
||||
"price": format_price(sec, req.price()),
|
||||
"qty": format_quantity(sec, req.quantity()),
|
||||
}
|
||||
self.trade_cli.amend_order(
|
||||
lambda msg: self._on_order_ack("cxlrep", msg), **params
|
||||
)
|
||||
|
||||
def cancel_order(self, req: PyCancelOrderRequest):
|
||||
self.logger.info("[bybit_gw_cxl_ord.json]%s" % str(req))
|
||||
if not self.stream_cli or not self.trade_cli:
|
||||
self._on_abort(req.key(), "cxl", "WebSocket client not created.")
|
||||
return
|
||||
bts_id = req.key()
|
||||
ord_data = self.orders.get(bts_id)
|
||||
if not ord_data:
|
||||
self._on_abort(req.key(), "cxl", "Did not resolve exch order id")
|
||||
return
|
||||
params = {
|
||||
"orderLinkId": ord_data.exch_id,
|
||||
"category": "linear",
|
||||
"symbol": req.exchange_security_id(),
|
||||
}
|
||||
self.trade_cli.cancel_order(
|
||||
lambda msg: self._on_order_ack("cxl", msg), **params
|
||||
)
|
||||
|
||||
def on_securities(self, securities: PySecurityList):
|
||||
self.logger.info(
|
||||
"[bybit_gw_sec.json][%s]" % ", ".join([str(sec) for sec in securities])
|
||||
)
|
||||
self.securities = {sec.name(): sec for sec in securities}
|
||||
|
||||
@@ -0,0 +1,263 @@
|
||||
import json
|
||||
import numpy as np
|
||||
import time
|
||||
import logging
|
||||
import sys
|
||||
import threading
|
||||
import signal
|
||||
from typing import Optional
|
||||
from pybit.unified_trading import WebSocket
|
||||
from python_interop_module import (
|
||||
MdBridge,
|
||||
MdProvider,
|
||||
PyBook,
|
||||
PyBookList,
|
||||
PySecurityList,
|
||||
)
|
||||
from bybit_conf import CONTRACTS
|
||||
from spark_core import (
|
||||
configure_logger,
|
||||
get_contract_spec,
|
||||
get_price_info,
|
||||
get_quantity_info,
|
||||
get_security,
|
||||
)
|
||||
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:
|
||||
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:
|
||||
logger.warning(f"Contract {contract} not found in API response, using default specs")
|
||||
return specs_dict
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to fetch contract specs: {e}", exc_info=True)
|
||||
return specs_dict
|
||||
|
||||
class BybitMdProvider(MdProvider):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.ws = None
|
||||
self.ws_thread: Optional[threading.Thread] = None
|
||||
self.bridge: Optional[MdBridge] = 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):
|
||||
logger.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_impl(self, bridge: MdBridge):
|
||||
self.bridge = bridge
|
||||
logger.info("Starting BybitMdProvider")
|
||||
|
||||
contract_specs = fetch_contract_specs()
|
||||
sec_list = PySecurityList()
|
||||
|
||||
for contract in CONTRACTS:
|
||||
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)
|
||||
)
|
||||
except Exception as e:
|
||||
logger.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:
|
||||
logger.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=PERP_EXP_STRING,
|
||||
option_kind=None,
|
||||
strike=None,
|
||||
spec=spec,
|
||||
)
|
||||
sec_list.append(sec)
|
||||
logger.info(f"Created security: {contract} with specs: {specs}")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to create security for contract: {contract}, error: {e}")
|
||||
continue
|
||||
|
||||
bridge.on_securities(sec_list)
|
||||
logger.info("Sent securities list--bridge.on_securities called.")
|
||||
|
||||
try:
|
||||
self.ws = WebSocket(
|
||||
testnet=False,
|
||||
channel_type="linear",
|
||||
ping_interval=20,
|
||||
)
|
||||
logger.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
|
||||
)
|
||||
logger.info(f"Subscribed to orderbook.50.{contract}")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to subscribe to orderbook for {contract}, error: {e}")
|
||||
|
||||
logger.info("Subscribed to orderbook channels on Bybit unified trading WS.")
|
||||
|
||||
for t in threading.enumerate():
|
||||
logger.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:
|
||||
logger.info("KeyboardInterrupt received — shutting down.")
|
||||
self.stop()
|
||||
except Exception as e:
|
||||
logger.error(f"Bybit WebSocket fatal error: {e}", exc_info=True)
|
||||
self.stop()
|
||||
|
||||
def start(self, bridge: MdBridge):
|
||||
def start_wrapper():
|
||||
try:
|
||||
self.start_impl(bridge)
|
||||
except Exception as e:
|
||||
logger.error(f"Error in BybitMdProvider start_impl: {e}", exc_info=True)
|
||||
|
||||
self.ws_thread = threading.Thread(target=start_wrapper)
|
||||
self.ws_thread.start()
|
||||
|
||||
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 += [(np.float64('nan'), np.float64('nan'))]
|
||||
sorted_asks = sorted(raw_asks, key=lambda x: x[0])[:5]
|
||||
sorted_asks += [(np.float64('nan'), np.float64('nan'))]
|
||||
|
||||
logger.debug(f"📘 BOOK DATA for {contract}: bids={sorted_bids} asks={sorted_asks}")
|
||||
|
||||
book = PyBook()
|
||||
book.set_data(
|
||||
contract,
|
||||
"Bybit",
|
||||
"FUTURE",
|
||||
np.array(sorted_bids, dtype=np.float64),
|
||||
np.array(sorted_asks, dtype=np.float64),
|
||||
)
|
||||
book_list = PyBookList()
|
||||
book_list.append(book)
|
||||
if self.bridge:
|
||||
self.bridge.on_books(book_list)
|
||||
logger.debug(f"✔ on_books called for {contract}")
|
||||
else:
|
||||
logger.warning("Bridge is None, cannot call on_books")
|
||||
except Exception as e:
|
||||
logger.error(f"Error processing book update for {contract}: {e}", exc_info=True)
|
||||
|
||||
def stop(self):
|
||||
logger.info("Stopping BybitMdProvider...")
|
||||
self.stop_event.set()
|
||||
|
||||
logger.debug("Threads before stopping WebSocket:")
|
||||
for t in threading.enumerate():
|
||||
logger.debug(f"Thread(stop): {t.name}, Alive: {t.is_alive()}, Daemon: {t.daemon}, Target={t._target}")
|
||||
|
||||
if self.ws:
|
||||
try:
|
||||
self.ws.exit()
|
||||
logger.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():
|
||||
logger.info(f"Joining pybit thread: {t.name}")
|
||||
t.join(timeout=1.0)
|
||||
except Exception as e:
|
||||
logger.error(f"Error while closing WebSocket: {e}", exc_info=True)
|
||||
self.ws = None
|
||||
|
||||
if self.ws_thread:
|
||||
self.ws_thread.join()
|
||||
self.bridge = None
|
||||
logger.info("✔ Stopped BybitMdProvider.")
|
||||
|
||||
logger.debug("Threads after stop:")
|
||||
for t in threading.enumerate():
|
||||
logger.debug(f"Thread(after stop): {t.name}, Alive: {t.is_alive()}, Daemon: {t.daemon}, Target={t._target}")
|
||||
|
||||
import os
|
||||
os._exit(0)
|
||||
@@ -0,0 +1,30 @@
|
||||
# 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.gateio.ws/v4/ws/usdt" # Mainnet url for perp futures
|
||||
|
||||
# API credentials
|
||||
API_KEY = "e4a0c6b198549042b13faa7645640d60"
|
||||
API_SECRET = "527c37877efc247bb0b6158cf3d5d18b84db720da894307f1e2ab88df3aaf0ee"
|
||||
|
||||
# Futures contract to subscribe to:
|
||||
|
||||
CONTRACTS = ['MELON_USDT', 'CAKE_USDT', 'ETH_USDT', 'KLAY_USDT', 'RON_USDT',
|
||||
'MANA_USDT', 'BZZ_USDT', 'KEEP_USDT', 'OMG_USDT', 'GMX_USDT', 'ICP_USDT',
|
||||
'LRC_USDT', 'MINA_USDT', 'ZIL_USDT', 'TURBO_USDT', 'OP_USDT', 'ZRX_USDT',
|
||||
'VET_USDT', 'AR_USDT', 'IMX_USDT', 'HIGH_USDT', 'ARB_USDT', 'YFI_USDT',
|
||||
'NEAR_USDT', 'OXY_USDT', 'WAVES_USDT', 'AMP_USDT', 'CELO_USDT', 'CRV_USDT',
|
||||
'FIL_USDT', 'PEOPLE_USDT', 'IOTX_USDT', 'ATOM_USDT', 'NOT_USDT', 'HBAR_USDT',
|
||||
'GMT_USDT', 'KAVA_USDT', 'WIF_USDT', 'UNI_USDT', 'GALA_USDT', 'AAVE_USDT',
|
||||
'CRU_USDT', 'PEPE_USDT', 'TWT_USDT', 'IRIS_USDT', 'APE_USDT', 'BNB_USDT',
|
||||
'ATLAS_USDT', 'BCHA_USDT', 'ETHW_USDT', 'DOT_USDT', 'TONCOIN_USDT', 'DOGS_USDT',
|
||||
'MER_USDT', 'ADA_USDT', 'TRIBE_USDT', 'THETA_USDT', 'BTC_USDT', 'ORDI_USDT',
|
||||
'ANKR_USDT', 'KIN_USDT', 'AMPL_USDT', 'TRX_USDT', 'SAND_USDT', 'SERO_USDT',
|
||||
'LUNC_USDT', 'BAT_USDT', 'KSM_USDT', 'SOL_USDT', 'AXS_USDT', 'BCD_USDT',
|
||||
'TIA_USDT', 'WOO_USDT', 'QTUM_USDT', 'OLDLUNA_USDT', 'AVAX_USDT', 'EOS_USDT',
|
||||
'SNX_USDT', 'MKR_USDT', 'ZKS_USDT', 'XRP_USDT', 'STEP_USDT', 'COMP_USDT',
|
||||
'UST_USDT', 'WSB_USDT', 'LINK_USDT', 'LUNA_USDT', 'ETC_USDT', 'JST_USDT',
|
||||
'LON_USDT', 'PEARL_USDT', 'BTT_USDT', 'XLM_USDT', 'GRT_USDT', 'STX_USDT',
|
||||
'NU_USDT', 'ALGO_USDT', 'ROSE_USDT', 'SHIB_USDT', 'BSV_USDT', 'ENJ_USDT',
|
||||
'USTC_USDT', 'NEO_USDT', 'FLOW_USDT', 'RUNE_USDT', 'FIL6_USDT', 'APT_USDT',
|
||||
'DOGE_USDT', 'CHZ_USDT', '1INCH_USDT', 'CSPR_USDT', 'XEC_USDT', 'ENS_USDT',
|
||||
'IOST_USDT', 'LDO_USDT', 'XTZ_USDT']
|
||||
@@ -0,0 +1,241 @@
|
||||
import dotenv
|
||||
import json
|
||||
import logging
|
||||
import math
|
||||
import sys
|
||||
import traceback
|
||||
from typing import Any, Optional
|
||||
import uuid
|
||||
import websocket
|
||||
|
||||
from binance_client import BinanceClient
|
||||
import python_interop_module as pyinter
|
||||
|
||||
|
||||
BINANCE_API_KEY = "2RJ9H02ztFGWxeY2oRpnFYZNU8mki4V1yMvKrDU08HCiEhO4VLe2ePSUFqEYTNxc"
|
||||
BINANCE_USDS_FUTURES_API_KEY = (
|
||||
"b838c108cd2648e34a4f088a7fc635a9cbccf3c2faa6bcaabcb532ffacc087b0"
|
||||
)
|
||||
BINANCE_USDS_FUTURES_SECRET_KEY = (
|
||||
"a68f0ab27403c42fa0306b6341b2ddd51d99ac42916f849404155449457d0492"
|
||||
)
|
||||
|
||||
|
||||
def to_bts_exec_type(exec_type: str, ord_status: str):
|
||||
match exec_type:
|
||||
case "TRADE_PREVENTION":
|
||||
return "REJECTED"
|
||||
case "NONE":
|
||||
return "UNDEFINED"
|
||||
case "TRADE":
|
||||
return "FILL" if ord_status == "FILLED" else "PARTIAL_FILL"
|
||||
case _:
|
||||
return exec_type
|
||||
|
||||
|
||||
def to_bts_order_status(ord_status: str):
|
||||
match ord_status:
|
||||
case "EXPIRED_IN_MATCH":
|
||||
return "EXPIRED"
|
||||
case _:
|
||||
return ord_status
|
||||
|
||||
|
||||
class DummyExecutionGateway(pyinter.ExecutionGateway):
|
||||
def __init__(self):
|
||||
super(DummyExecutionGateway, self).__init__()
|
||||
self.bridge = None
|
||||
self.client = None
|
||||
self.start_th = None
|
||||
self.ws = None
|
||||
self.orders = {}
|
||||
self.logger = logging.getLogger(__name__)
|
||||
stdout_handler = logging.StreamHandler(sys.stdout)
|
||||
stdout_handler.formatter = logging.Formatter(
|
||||
"%(asctime)s | %(levelname)7s | %(filename)s:%(lineno)s | %(message)s"
|
||||
)
|
||||
self.logger.addHandler(stdout_handler)
|
||||
self.logger.addHandler(
|
||||
logging.FileHandler(
|
||||
"/home/bts/sparkdata/data/logging.python_execution/dummy_log"
|
||||
)
|
||||
)
|
||||
self.logger.setLevel(logging.INFO)
|
||||
env_path = "/workspaces/spark/tools/binance/binance.env"
|
||||
self.logger.info("DummyExecutionGateway ctr. Loading env")
|
||||
dotenv.load_dotenv(env_path, interpolate=False)
|
||||
|
||||
def _index_transaction(self, bts_key: int):
|
||||
exch_key = uuid.uuid1().hex
|
||||
self.orders[exch_key] = bts_key
|
||||
return exch_key
|
||||
|
||||
def _on_user_data_msg_core(self, ws: websocket.WebSocket, msg: Any):
|
||||
self.logger.info("Recv usr data: %s", str(msg))
|
||||
if not msg:
|
||||
self.logger.error("Recv empty usr data msg")
|
||||
return
|
||||
if not self.bridge:
|
||||
self.logger.error("Cannot process usr data. Bridge not created")
|
||||
return
|
||||
upd = json.loads(str(msg))
|
||||
if "e" in upd:
|
||||
if upd["e"] == "executionReport":
|
||||
# Binance spots update
|
||||
bts_ord_id = self.orders.get(upd["c"])
|
||||
is_old_ord = False
|
||||
if not bts_ord_id:
|
||||
bts_ord_id = self.orders.get(upd["C"])
|
||||
if not bts_ord_id:
|
||||
self.logger.warning(
|
||||
"Received update with unknown order id: %s", msg
|
||||
)
|
||||
return
|
||||
is_old_ord = True
|
||||
er = pyinter.PyExecutionReport()
|
||||
er.set_key(bts_ord_id)
|
||||
er.set_exchange_security_id(upd["s"])
|
||||
er.set_security_type("SPOT")
|
||||
er.set_exec_type(to_bts_exec_type(upd["x"], upd["X"]))
|
||||
er.set_is_buy(upd["S"] == "BUY")
|
||||
er.set_price(float(upd["p"]))
|
||||
er.set_quantity(float(upd["q"]))
|
||||
er.set_last_filled_price(upd["L"])
|
||||
er.set_last_filled_quantity(upd["l"])
|
||||
total_filled_qty = float(upd["z"])
|
||||
er.set_filled_quantity(total_filled_qty)
|
||||
total_filled_notional = float(upd["Z"])
|
||||
avg_filled_price = (
|
||||
total_filled_notional / total_filled_qty
|
||||
if total_filled_qty > 0
|
||||
else math.nan
|
||||
)
|
||||
er.set_avg_filled_price(avg_filled_price)
|
||||
er.set_order_status(to_bts_order_status(upd["X"]))
|
||||
er.set_order_type(upd["o"])
|
||||
match upd["X"]:
|
||||
case "CANCELED":
|
||||
if not is_old_ord:
|
||||
del self.orders[bts_ord_id]
|
||||
case "FILLED":
|
||||
del self.orders[bts_ord_id]
|
||||
case "REJECTED":
|
||||
del self.orders[bts_ord_id]
|
||||
self.bridge.on_execution_report(er)
|
||||
elif upd["e"] == "ORDER_TRADE_UPDATE":
|
||||
data = upd["o"]
|
||||
# Binance Usds Futures order update
|
||||
bts_ord_id = self.orders.get(data["c"])
|
||||
is_old_ord = False
|
||||
if not bts_ord_id:
|
||||
bts_ord_id = self.orders.get(data["C"])
|
||||
if not bts_ord_id:
|
||||
self.logger.warning(
|
||||
"Received dataate with unknown order id: %s", msg
|
||||
)
|
||||
return
|
||||
is_old_ord = True
|
||||
|
||||
def parse_float(s: Optional[str], default_val=math.nan):
|
||||
if s is not None:
|
||||
try:
|
||||
return float(s)
|
||||
except Exception:
|
||||
pass
|
||||
return default_val
|
||||
|
||||
er = pyinter.PyExecutionReport()
|
||||
er.set_key(bts_ord_id)
|
||||
er.set_venue("BINANCE_USDS_FUTURES")
|
||||
er.set_security_type("FUTURE")
|
||||
er.set_time_in_force("GTC")
|
||||
er.set_exchange_order_id(str(data.get("i") or 0))
|
||||
# er.set_exchange_order_id(data["i"])
|
||||
er.set_exchange_security_id(data["s"])
|
||||
er.set_exec_type(to_bts_exec_type(data["x"], data["X"]))
|
||||
er.set_is_buy(data["S"] == "BUY")
|
||||
er.set_price(parse_float(data["p"]))
|
||||
er.set_quantity(parse_float(data["q"]))
|
||||
er.set_last_filled_price(parse_float(data.get("L")))
|
||||
er.set_last_filled_quantity(parse_float(data["l"]))
|
||||
er.set_filled_quantity(parse_float(data["z"]))
|
||||
er.set_avg_filled_price(parse_float(data["ap"]))
|
||||
er.set_order_status(to_bts_order_status(data["X"]))
|
||||
er.set_order_type(data["o"])
|
||||
match data["X"]:
|
||||
case "CANCELED":
|
||||
if not is_old_ord:
|
||||
del self.orders[bts_ord_id]
|
||||
case "FILLED":
|
||||
del self.orders[bts_ord_id]
|
||||
case "REJECTED":
|
||||
del self.orders[bts_ord_id]
|
||||
print("Firing this exe report: %s" % str(er), flush=True)
|
||||
self.bridge.on_execution_report(er)
|
||||
|
||||
def _on_user_data_msg(self, ws: websocket.WebSocket, msg: Any):
|
||||
try:
|
||||
self._on_user_data_msg_core(ws, msg)
|
||||
except Exception:
|
||||
self.logger.error("on_user_data_msg. %s" % traceback.format_exc())
|
||||
|
||||
def _on_user_data_open(self, ws: websocket.WebSocket):
|
||||
self.ws = ws
|
||||
if self.bridge:
|
||||
self.bridge.on_is_ready(True)
|
||||
|
||||
def _on_user_data_err(self, ws: websocket.WebSocket, err: str):
|
||||
self.logger.error("on_user_data_err: %s", err)
|
||||
|
||||
def _on_user_data_close(self, ws: websocket.WebSocket, code: int, msg: str):
|
||||
self.logger.warning("on_user_data_close: %s %s", str(code), msg)
|
||||
if self.bridge:
|
||||
self.bridge.on_is_ready(False)
|
||||
|
||||
def start(self, bridge: pyinter.ExecutionGatewayBridge):
|
||||
print("DummyExecutionGateway.start")
|
||||
self.bridge = bridge
|
||||
self.client = BinanceClient(venue="BINANCE_USDS_FUTURES")
|
||||
print("DummyExecutionGateway.start. Subscribing to binance user data")
|
||||
self.client.subscribe_user_data(
|
||||
on_message=self._on_user_data_msg,
|
||||
on_open=self._on_user_data_open,
|
||||
on_close=self._on_user_data_close,
|
||||
)
|
||||
print("DummyExecutionGateway.start. Subscribed to binance user data")
|
||||
|
||||
def cancel_order(self, req: pyinter.PyCancelOrderRequest):
|
||||
if not self.client:
|
||||
self.logger.error("Cannot cancel order. BinanceClient not created.")
|
||||
return
|
||||
self.logger.info("[cancel_order.json]%s", str(req))
|
||||
try:
|
||||
self.client.request(
|
||||
"CANCEL_ORDER",
|
||||
symbol=req.exchange_security_id(),
|
||||
orig_client_order_id=req.exchange_order_id(),
|
||||
new_client_order_id=self._index_transaction(req.key()),
|
||||
)
|
||||
except Exception as ex:
|
||||
self.logger.error("Create order request threw. %s" % str(ex))
|
||||
|
||||
def create_order(self, req: pyinter.PyCreateOrderRequest):
|
||||
if not self.client:
|
||||
self.logger.error("Cannot create order. BinanceClient not created.")
|
||||
return
|
||||
print("hey bud")
|
||||
self.logger.info("[create_order.json]%s", str(req))
|
||||
side = "BUY" if req.is_buy() else "SELL"
|
||||
try:
|
||||
self.client.request(
|
||||
op="CREATE_ORDER",
|
||||
symbol=req.exchange_security_id(),
|
||||
price=req.price(),
|
||||
quantity=f"{req.quantity():.6f}",
|
||||
side=side,
|
||||
time_in_force="GTC",
|
||||
type="LIMIT",
|
||||
new_client_order_id=self._index_transaction(req.key()),
|
||||
)
|
||||
except Exception as ex:
|
||||
self.logger.error("Create order request threw. %s" % str(ex))
|
||||
@@ -0,0 +1,79 @@
|
||||
from python_interop_module import (
|
||||
MdBridge,
|
||||
MdProvider,
|
||||
PySecurityList,
|
||||
PyBook,
|
||||
PyBookList,
|
||||
PyTradeList,
|
||||
)
|
||||
from spark_core import (
|
||||
get_security,
|
||||
get_contract_spec,
|
||||
get_price_info,
|
||||
get_quantity_info,
|
||||
)
|
||||
from py_md_utils import get_book_side, get_trade
|
||||
|
||||
|
||||
class DummyMdProvider(MdProvider):
|
||||
def __init__(self):
|
||||
super(DummyMdProvider, self).__init__()
|
||||
pass
|
||||
|
||||
def start(self, bridge: MdBridge):
|
||||
# start subscriptions, bind callbacks etc.
|
||||
def add_fake_securities():
|
||||
sec_list = PySecurityList()
|
||||
sec_list.append(
|
||||
get_security(
|
||||
venue="Glbx",
|
||||
name="ESH8",
|
||||
asset="ES",
|
||||
group="ES",
|
||||
currency="USD",
|
||||
expiration="2026-03-28",
|
||||
spec=get_contract_spec(
|
||||
price_info=get_price_info(
|
||||
tick_unit=25,
|
||||
tick_base=1000,
|
||||
cash_per_point=50,
|
||||
),
|
||||
qty_info=get_quantity_info(tick_unit=1, tick_base=1),
|
||||
),
|
||||
)
|
||||
)
|
||||
bridge.on_securities(sec_list)
|
||||
|
||||
def add_fake_books():
|
||||
bids = get_book_side([(99.75, 8), (99.50, 11)], is_bid_side=True)
|
||||
asks = get_book_side([(100.25, 3)], is_bid_side=False)
|
||||
book = PyBook()
|
||||
book.set_venue("Glbx")
|
||||
book.set_security_type("FUTURE")
|
||||
book.set_exchange_security_id("ESH8")
|
||||
book.set_bids(bids)
|
||||
book.set_asks(asks)
|
||||
# pass book to sink
|
||||
book_list = PyBookList()
|
||||
book_list.append(book)
|
||||
bridge.on_books(book_list)
|
||||
|
||||
def add_fake_trades():
|
||||
trade = get_trade(
|
||||
venue="Glbx",
|
||||
security_type="FUTURE",
|
||||
exchange_security_id="ESH8",
|
||||
price=100,
|
||||
quantity=2,
|
||||
aggressor_side="Buy",
|
||||
)
|
||||
trade_list = PyTradeList()
|
||||
trade_list.append(trade)
|
||||
bridge.on_trades(trade_list)
|
||||
|
||||
add_fake_securities()
|
||||
add_fake_books()
|
||||
add_fake_trades()
|
||||
|
||||
def stop(self):
|
||||
print("Stopping python DummyMdProvider")
|
||||
@@ -0,0 +1,226 @@
|
||||
import json
|
||||
import numpy as np
|
||||
import time
|
||||
import logging
|
||||
import sys
|
||||
import threading
|
||||
import requests
|
||||
from websocket import create_connection, WebSocketConnectionClosedException
|
||||
from typing import Optional, Any
|
||||
from python_interop_module import (
|
||||
MdBridge,
|
||||
MdProvider,
|
||||
PyBook,
|
||||
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,
|
||||
)
|
||||
|
||||
logger = logging.getLogger("gateio_md_provider")
|
||||
configure_logger(logger, "md")
|
||||
|
||||
def fetch_contract_specs() -> dict:
|
||||
"""Fetch specs for all contracts in one request."""
|
||||
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.gateio.ws/api/v4/futures/usdt/contracts"
|
||||
response = requests.get(url, timeout=5)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
for item in data:
|
||||
contract = item.get("name")
|
||||
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
|
||||
qty_tick_base = int(1 / contract_size) if contract_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
|
||||
}
|
||||
return specs_dict
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to fetch batch specs: {e}", exc_info=True)
|
||||
return specs_dict
|
||||
|
||||
class GateIoMdProvider(MdProvider):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.ws = None
|
||||
self.ws_thread: Optional[threading.Thread] = None
|
||||
self.bridge: Optional[MdBridge] = None
|
||||
self.stop_event = threading.Event()
|
||||
self.ping_thread = None
|
||||
|
||||
def send_ping(self):
|
||||
"""Send ping messages every 15 seconds to keep connection alive."""
|
||||
while not self.stop_event.is_set():
|
||||
try:
|
||||
ping_msg = {
|
||||
"time": int(time.time() * 1000),
|
||||
"channel": "futures.ping",
|
||||
"event": "",
|
||||
"payload": []
|
||||
}
|
||||
self.ws.send(json.dumps(ping_msg))
|
||||
logger.debug("Sent ping")
|
||||
except Exception as e:
|
||||
logger.error(f"Ping error: {e}", exc_info=True)
|
||||
break
|
||||
time.sleep(15)
|
||||
|
||||
def start_impl(self, bridge: MdBridge):
|
||||
self.bridge = bridge
|
||||
logger.info(f"Connecting to {WS_URL}")
|
||||
self.ws = create_connection(WS_URL)
|
||||
self.ws.settimeout(5)
|
||||
logger.info("Connected.")
|
||||
|
||||
# Start ping thread
|
||||
self.ping_thread = threading.Thread(target=self.send_ping, daemon=True)
|
||||
self.ping_thread.start()
|
||||
logger.info("Started ping thread")
|
||||
|
||||
# Fetch all contract specs at once
|
||||
contract_specs = fetch_contract_specs()
|
||||
|
||||
# 1) Send security info
|
||||
sec_list = PySecurityList()
|
||||
for contract in CONTRACTS:
|
||||
asset, quote = contract.split("_")
|
||||
specs = contract_specs[contract]
|
||||
spec = get_contract_spec(
|
||||
price_info=get_price_info(
|
||||
tick_unit=specs["tick_unit"],
|
||||
tick_base=specs["tick_base"],
|
||||
cash_per_point=specs["cash_per_point"]
|
||||
),
|
||||
qty_info=get_quantity_info(
|
||||
tick_unit=specs["qty_tick_unit"],
|
||||
tick_base=specs["qty_tick_base"]
|
||||
)
|
||||
)
|
||||
sec = get_security(
|
||||
venue="GateIO",
|
||||
name=contract,
|
||||
asset=asset,
|
||||
group=asset,
|
||||
currency=quote,
|
||||
expiration=None,
|
||||
option_kind=None,
|
||||
strike=None,
|
||||
spec=spec,
|
||||
)
|
||||
sec_list.append(sec)
|
||||
logger.info(f"Created security: {contract} with specs: {specs}")
|
||||
bridge.on_securities(sec_list)
|
||||
logger.info("Sent securities list--sink.on_securities called.")
|
||||
|
||||
# 2) Subscribe to order books
|
||||
for contract in CONTRACTS:
|
||||
sub_msg = {
|
||||
"time": int(time.time() * 1000),
|
||||
"channel": "futures.order_book",
|
||||
"event": "subscribe",
|
||||
"payload": [contract, "5", "0"]
|
||||
}
|
||||
self.ws.send(json.dumps(sub_msg))
|
||||
logger.info(f"✔ Sent subscribe for {contract}: {sub_msg}")
|
||||
|
||||
# 3) Listen for incoming messages in main thread
|
||||
logger.info("▶ Listening for data")
|
||||
try:
|
||||
while not self.stop_event.is_set():
|
||||
try:
|
||||
msg = self.ws.recv()
|
||||
except Exception as e:
|
||||
logger.debug(f"⚠ recv() error or timeout: {e}")
|
||||
continue
|
||||
|
||||
logger.debug(f"⟵ raw WS: {msg}")
|
||||
try:
|
||||
data = json.loads(msg)
|
||||
logger.debug(f"Parsed JSON: {data}")
|
||||
except Exception as e:
|
||||
logger.error(f"JSON parse error: {e}", exc_info=True)
|
||||
continue
|
||||
|
||||
event = data.get("event")
|
||||
channel = data.get("channel")
|
||||
|
||||
# Handle subscription acknowledgements
|
||||
if event == "subscribe":
|
||||
logger.info(f"🔔 Subscribe ACK: {data}")
|
||||
continue
|
||||
|
||||
# Handle order book updates
|
||||
if channel == "futures.order_book" and event in ["all", "update"]:
|
||||
logger.debug(f"Handling '{event}' event")
|
||||
payload = data.get("result", {})
|
||||
if isinstance(payload, dict) and ("bids" in payload or "asks" in payload):
|
||||
contract = payload.get("contract", "<no-contract>")
|
||||
bids = [(np.float64(l["p"]), np.float64(l["s"])) for l in payload.get("bids", [])]
|
||||
asks = [(np.float64(l["p"]), np.float64(l["s"])) for l in payload.get("asks", [])]
|
||||
logger.debug(f"📘 BOOK DATA for {contract}: bids={bids} asks={asks}")
|
||||
|
||||
book = PyBook()
|
||||
book.set_data(
|
||||
contract,
|
||||
"GateIO",
|
||||
"SPOT",
|
||||
np.array(bids, dtype=np.float64),
|
||||
np.array(asks, dtype=np.float64),
|
||||
)
|
||||
book_list = PyBookList()
|
||||
book_list.append(book)
|
||||
self.bridge.on_books(book_list)
|
||||
logger.debug(f"✔ on_books called for {contract}")
|
||||
else:
|
||||
logger.debug("… no bids/asks in payload; skipping")
|
||||
|
||||
except WebSocketConnectionClosedException:
|
||||
logger.info("WebSocket closed.")
|
||||
except Exception as e:
|
||||
logger.error("Fatal error in listen loop:", exc_info=True)
|
||||
|
||||
def start(self, bridge: MdBridge):
|
||||
self.ws_thread = threading.Thread(target=lambda: self.start_impl(bridge))
|
||||
self.ws_thread.start()
|
||||
|
||||
def stop(self):
|
||||
"""Disconnect and cleanup."""
|
||||
logger.info("Stopping GateIoMdProvider...")
|
||||
self.stop_event.set()
|
||||
|
||||
if self.ping_thread and self.ping_thread.is_alive():
|
||||
self.ping_thread.join()
|
||||
logger.info("Ping thread stopped")
|
||||
|
||||
if self.ws:
|
||||
try:
|
||||
self.ws.close()
|
||||
except Exception:
|
||||
pass
|
||||
self.ws = None
|
||||
|
||||
if self.ws_thread:
|
||||
self.ws_thread.join()
|
||||
self.bridge = None
|
||||
logger.info("✔ Stopped GateIoMdProvider.")
|
||||
@@ -0,0 +1,12 @@
|
||||
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)
|
||||
@@ -0,0 +1,40 @@
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
|
||||
from typing import Optional
|
||||
from python_interop_module import PyBookSide, PyBookLevel, PyTrade
|
||||
|
||||
|
||||
PERP_EXP_STRING = "2100-01-01T00:00"
|
||||
|
||||
|
||||
def get_book_side(levels: list[tuple[float, float]], is_bid_side: bool):
|
||||
book_side = PyBookSide()
|
||||
book_side.set_is_ascending(not is_bid_side)
|
||||
for price, qty in levels:
|
||||
book_level = PyBookLevel()
|
||||
book_level.set_price(price)
|
||||
book_level.set_quantity(qty)
|
||||
book_side.set_level(book_level)
|
||||
return book_side
|
||||
|
||||
|
||||
def get_trade(
|
||||
venue: str,
|
||||
security_type: str,
|
||||
exchange_security_id: str,
|
||||
price: float,
|
||||
quantity: float,
|
||||
aggressor_side: str,
|
||||
):
|
||||
trade = PyTrade()
|
||||
trade.set_venue(venue)
|
||||
trade.set_security_type(security_type)
|
||||
trade.set_exchange_security_id(exchange_security_id)
|
||||
trade.set_price(price)
|
||||
trade.set_quantity(quantity)
|
||||
trade.set_aggressor_side(aggressor_side)
|
||||
return trade
|
||||
@@ -0,0 +1,372 @@
|
||||
import json
|
||||
import math
|
||||
import string
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
from typing import Callable, Optional
|
||||
import websocket
|
||||
from python_interop_module import (
|
||||
PyContractPriceInfo,
|
||||
PyContractPriceInfoList,
|
||||
PyContractQuantityInfo,
|
||||
PyContractQuantityInfoList,
|
||||
PyContractSpecification,
|
||||
PyFuture,
|
||||
PyOption,
|
||||
PySecurity,
|
||||
PySpot,
|
||||
)
|
||||
|
||||
|
||||
class JsonElement:
|
||||
def __init__(self, value):
|
||||
self.value = value
|
||||
|
||||
def __str__(self):
|
||||
return json.dumps(self.value)
|
||||
|
||||
|
||||
class StructuredLog:
|
||||
def __init__(self, **kwargs):
|
||||
self.kwargs = kwargs
|
||||
if "ts" not in kwargs:
|
||||
kwargs["ts"] = time.time_ns()
|
||||
|
||||
def __str__(self):
|
||||
return json.dumps(self.kwargs)
|
||||
|
||||
|
||||
def _log_level_name(level: int):
|
||||
match level:
|
||||
case logging.DEBUG:
|
||||
return "Diag"
|
||||
case logging.INFO:
|
||||
return "Info"
|
||||
case logging.WARNING:
|
||||
return "Warn"
|
||||
case logging.ERROR:
|
||||
return "Error"
|
||||
case logging.CRITICAL:
|
||||
return "Fatal"
|
||||
case _:
|
||||
return str(level)
|
||||
|
||||
|
||||
class FileLogFormatter(logging.Formatter):
|
||||
def __init__(self, subsystem: str):
|
||||
self.subsystem = subsystem
|
||||
|
||||
def formatTime(self, record, datefmt=None):
|
||||
ct = self.converter(record.created)
|
||||
if datefmt:
|
||||
return time.strftime(datefmt, ct)
|
||||
else:
|
||||
t = time.strftime("%Y-%m-%d %H:%M:%S", ct)
|
||||
return f"{t}.{int(record.msecs * 1000):06d}"
|
||||
|
||||
def format(self, record):
|
||||
env = "dev"
|
||||
thread_id = threading.get_ident()
|
||||
filename_line = f"{record.filename}:{record.lineno}"
|
||||
timestamp = self.formatTime(record)
|
||||
timestamp_ns = int(record.created * 1e9)
|
||||
return f"{timestamp}|{timestamp_ns}|{self.subsystem:<8}|{_log_level_name(record.levelno):<5}|{os.getpid():<3}|{record.threadName:<20}|{thread_id}|{filename_line:<35}|{env}| {record.getMessage()}"
|
||||
|
||||
|
||||
class ConsoleLogFormatter(logging.Formatter):
|
||||
def __init__(self, subsystem: str):
|
||||
self.subsystem = subsystem
|
||||
|
||||
def formatTime(self, record, datefmt=None):
|
||||
ct = self.converter(record.created)
|
||||
if datefmt:
|
||||
return time.strftime(datefmt, ct)
|
||||
else:
|
||||
t = time.strftime("%Y-%m-%d %H:%M:%S", ct)
|
||||
return f"{t}.{int(record.msecs * 1000):06d}"
|
||||
|
||||
def format(self, record):
|
||||
filename_line = f"{record.filename}:{record.lineno}"
|
||||
timestamp = self.formatTime(record)
|
||||
return f"{timestamp}|{self.subsystem:<8}|{_log_level_name(record.levelno):<5}|{filename_line:<35}| {record.getMessage()}"
|
||||
|
||||
|
||||
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"
|
||||
)
|
||||
stdout_handler.formatter = ConsoleLogFormatter(subsystem=subsystem)
|
||||
file_handler.formatter = FileLogFormatter(subsystem=subsystem)
|
||||
logger.handlers.clear()
|
||||
logger.addHandler(stdout_handler)
|
||||
logger.addHandler(file_handler)
|
||||
logger.setLevel(logging.INFO)
|
||||
return logger
|
||||
|
||||
|
||||
def parse_float(s: Optional[str], default_val=math.nan):
|
||||
if s is not None:
|
||||
try:
|
||||
return float(s)
|
||||
except Exception:
|
||||
pass
|
||||
return default_val
|
||||
|
||||
|
||||
def run_websocket(
|
||||
url: str,
|
||||
stop_event: threading.Event,
|
||||
logger: logging.Logger,
|
||||
on_message: Callable[[str], None],
|
||||
send_ping: Optional[Callable[[websocket.WebSocket], None]] = None,
|
||||
ping_interval_secs: Optional[int] = 25,
|
||||
):
|
||||
conn_num = 0
|
||||
if send_ping is None:
|
||||
|
||||
def send_empty_ping(ws):
|
||||
return ws.ping()
|
||||
|
||||
send_ping = send_empty_ping
|
||||
|
||||
while not stop_event.is_set():
|
||||
if conn_num > 0:
|
||||
# Gradually increase sleep time between attempts, capping at 5min.
|
||||
sleep_secs = min(conn_num * 5, 300)
|
||||
logger.warning("Reconnecting to %s in %ds." % (url, sleep_secs))
|
||||
time.sleep(sleep_secs)
|
||||
logger.info("Connecting to %s (attempt no %d)" % (url, conn_num))
|
||||
ws = websocket.create_connection(url)
|
||||
ws.settimeout(5)
|
||||
logger.info("Connected to %s. Listening for data" % url)
|
||||
if ping_interval_secs:
|
||||
|
||||
def ping_loop(orig_conn_num):
|
||||
time.sleep(ping_interval_secs)
|
||||
while not stop_event.is_set() and conn_num == orig_conn_num:
|
||||
send_ping(ws)
|
||||
time.sleep(ping_interval_secs)
|
||||
|
||||
threading.Thread(target=ping_loop(conn_num), daemon=True).start()
|
||||
try:
|
||||
msg = ws.recv()
|
||||
if isinstance(msg, bytes):
|
||||
on_message(msg.decode("utf-8"))
|
||||
elif isinstance(msg, str):
|
||||
on_message(msg)
|
||||
else:
|
||||
logger.error("Websocket recv unexpected msg type %s" % str(type(msg)))
|
||||
|
||||
except Exception:
|
||||
logging.error("Websocket recv threw.", exc_info=True)
|
||||
continue
|
||||
|
||||
|
||||
def get_tick_factors(tick_sz: float):
|
||||
"""
|
||||
Returns a tuple of (tick unit, tick base) that represents the tick size as a
|
||||
fraction. For example, if tick_sz is 0.01, returns (1, 100), as 1/100 == 0.01.
|
||||
"""
|
||||
if not (tick_sz > 0):
|
||||
raise ValueError(f"Invalid tick size {tick_sz}")
|
||||
tick_denom = 1.0 / tick_sz
|
||||
match tick_denom:
|
||||
case 1:
|
||||
return (1, 1)
|
||||
case 10:
|
||||
return (10, 100)
|
||||
case 100:
|
||||
return (1, 100)
|
||||
case 1_000:
|
||||
return (1, 1_000)
|
||||
case 10_000:
|
||||
return (1, 10_000)
|
||||
case 100_000:
|
||||
return (1, 100_000)
|
||||
case 1_000_000:
|
||||
return (1, 1_000_000)
|
||||
case 10_000_000:
|
||||
return (1, 10_000_000)
|
||||
case 100_000_000:
|
||||
return (1, 100_000_000)
|
||||
case 1_000_000_000:
|
||||
return (1, 1_000_000_000)
|
||||
case 4:
|
||||
return (25, 100)
|
||||
case 40:
|
||||
return (25, 1_000)
|
||||
case 400:
|
||||
return (25, 10_000)
|
||||
case 4_000:
|
||||
return (25, 100_000)
|
||||
case 20:
|
||||
return (5, 100)
|
||||
case 200:
|
||||
return (5, 1_000)
|
||||
case 2_000:
|
||||
return (5, 10_000)
|
||||
case 20_000:
|
||||
return (5, 100_000)
|
||||
case _:
|
||||
raise ValueError(f"Invalid tick size {tick_sz}")
|
||||
|
||||
|
||||
def get_num_decimal_places(tick_sz: float):
|
||||
return math.pow(10, -math.floor(math.log10(tick_sz)))
|
||||
|
||||
|
||||
def get_qty_decimal_places(sec: PySecurity) -> int:
|
||||
qty_infos = sec.contract_spec().quantity_infos()
|
||||
if not qty_infos:
|
||||
return 2
|
||||
tick_sz = qty_infos[0].tick_unit() / qty_infos[0].tick_base()
|
||||
return round(-math.floor(math.log10(tick_sz)))
|
||||
|
||||
|
||||
def _get_decimal_places(tick_base: float) -> int:
|
||||
# tick base of 100 -> 2 decimals, tick base of 1000 -> 3 decimals, etc.
|
||||
tick_sz = 1 / tick_base
|
||||
return round(-math.floor(math.log10(tick_sz)))
|
||||
|
||||
|
||||
def get_price_decimal_places(sec: PySecurity) -> int:
|
||||
price_infos = sec.contract_spec().price_infos()
|
||||
if not price_infos:
|
||||
return 2
|
||||
return _get_decimal_places(price_infos[0].tick_base())
|
||||
|
||||
|
||||
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}"
|
||||
|
||||
|
||||
def get_price_info(
|
||||
tick_unit: float,
|
||||
tick_base: float,
|
||||
cash_per_point: Optional[float] = None,
|
||||
multiplier: Optional[float] = None,
|
||||
max_price: Optional[float] = None,
|
||||
):
|
||||
# Use 'not' checks to account for nans (for which every op evaluates to false)
|
||||
if not (tick_unit > 0) or not (tick_base >= tick_unit):
|
||||
raise ValueError(f"Invalid tick factors {tick_unit}/{tick_base}")
|
||||
price_info = PyContractPriceInfo()
|
||||
price_info.set_tick_unit(tick_unit)
|
||||
price_info.set_tick_base(tick_base)
|
||||
if cash_per_point:
|
||||
price_info.set_cash_per_point(cash_per_point)
|
||||
if max_price:
|
||||
price_info.set_max_price(max_price)
|
||||
if multiplier:
|
||||
price_info.set_multiplier(multiplier)
|
||||
return price_info
|
||||
|
||||
|
||||
def get_quantity_info(
|
||||
tick_unit: float,
|
||||
tick_base: float,
|
||||
min_value: Optional[float] = None,
|
||||
max_value: Optional[float] = None,
|
||||
multiplier: Optional[float] = None,
|
||||
max_price: Optional[float] = None,
|
||||
):
|
||||
# Use 'not' checks to account for nans (for which every op evaluates to false)
|
||||
if not (tick_unit > 0) or not (tick_base >= tick_unit):
|
||||
raise ValueError(f"Invalid tick factors {tick_unit}/{tick_base}")
|
||||
if not min_value:
|
||||
min_value = tick_unit / tick_base
|
||||
elif min_value < (tick_unit / tick_base):
|
||||
# min_value should at least be 1 tick
|
||||
raise ValueError(f"min_value ({min_value}) < tick ({tick_unit}/{tick_base})")
|
||||
qty_info = PyContractQuantityInfo()
|
||||
qty_info.set_tick_unit(tick_unit)
|
||||
qty_info.set_tick_base(tick_base)
|
||||
qty_info.set_min_value(min_value)
|
||||
if max_value:
|
||||
if not (max_value > min_value):
|
||||
raise ValueError(
|
||||
f"max_value ({max_value}) not greater than min_value ({min_value})"
|
||||
)
|
||||
qty_info.set_max_value(max_value)
|
||||
if max_price:
|
||||
qty_info.set_max_price(max_price)
|
||||
if multiplier:
|
||||
qty_info.set_multiplier(multiplier)
|
||||
return qty_info
|
||||
|
||||
|
||||
def get_contract_spec(
|
||||
price_info: Optional[PyContractPriceInfo] = None,
|
||||
qty_info: Optional[PyContractQuantityInfo] = None,
|
||||
notional_qty_info: Optional[PyContractQuantityInfo] = None,
|
||||
):
|
||||
price_infos = PyContractPriceInfoList()
|
||||
if not price_info:
|
||||
# Default to 3 decimal places
|
||||
price_info = get_price_info(1, 1000)
|
||||
price_infos.append(price_info)
|
||||
qty_infos = PyContractQuantityInfoList()
|
||||
if not qty_info:
|
||||
# Default to 3 decimal places
|
||||
qty_info = get_quantity_info(1, 1000)
|
||||
qty_infos.append(qty_info)
|
||||
notional_qty_infos = PyContractQuantityInfoList()
|
||||
if notional_qty_info:
|
||||
notional_qty_infos.append(notional_qty_info)
|
||||
spec = PyContractSpecification()
|
||||
spec.set_price_infos(price_infos)
|
||||
spec.set_quantity_infos(qty_infos)
|
||||
spec.set_notional_quantity_infos(notional_qty_infos)
|
||||
return spec
|
||||
|
||||
|
||||
def get_security(
|
||||
venue: str,
|
||||
name: str,
|
||||
asset: str,
|
||||
group: str,
|
||||
currency: str,
|
||||
expiration: Optional[str] = None,
|
||||
option_kind: Optional[str] = None,
|
||||
strike: Optional[float] = None,
|
||||
spec: Optional[PyContractSpecification] = None,
|
||||
):
|
||||
sec = PySecurity()
|
||||
sec.set_venue(venue)
|
||||
sec.set_exchange_id(name)
|
||||
sec.set_name(name)
|
||||
sec.set_currency(currency)
|
||||
sec.set_asset(asset)
|
||||
sec.set_group(group)
|
||||
if not spec:
|
||||
spec = get_contract_spec()
|
||||
sec.set_contract_spec(spec)
|
||||
if expiration and strike and option_kind:
|
||||
opt = PyOption()
|
||||
opt.set_expiration(expiration)
|
||||
opt.set_is_call(option_kind[0].lower() == "c")
|
||||
opt.set_strike(strike)
|
||||
sec.set_option(opt)
|
||||
elif expiration and not strike and not option_kind:
|
||||
fut = PyFuture()
|
||||
fut.set_expiration(expiration)
|
||||
sec.set_future(fut)
|
||||
elif not expiration and not strike and not option_kind:
|
||||
sec.set_spot(PySpot())
|
||||
else:
|
||||
raise ValueError("Invalid parameters (unable to deduce security type)")
|
||||
return sec
|
||||
Reference in new issue
Block a user