""" Bybit Demo REST client (custom, bypass ccxt bug con api-demo.bybit.com). Implementa solo i metodi necessari al live engine: - fetch_ohlcv(symbol, interval, limit) - fetch_balance() - fetch_positions(symbol) - create_order(symbol, side, qty) market entry - create_reduce_order(symbol, side, qty, price) - set_trading_stop(symbol, sl_price) Testato: HMAC manuale funziona, totalEquity 48462 USDT. """ import os import time import hmac import hashlib import json import requests from pathlib import Path from dotenv import load_dotenv from typing import Optional, List, Dict from order_dispatch_guard import validate_order_dispatch, audit_dispatch ENV_FILE = Path(r"/opt/charter-live/API_KEY_BYBIT.env") BASE_URL = "https://api-demo.bybit.com" class BybitDemoClient: def __init__(self): load_dotenv(ENV_FILE) self.api_key = os.getenv("BYBIT_DEMO_API_KEY") self.secret = os.getenv("BYBIT_DEMO_SECRET_KEY") if not self.api_key or not self.secret: raise RuntimeError(f"Credenziali mancanti in {ENV_FILE}") self.session = requests.Session() def _sign(self, body: str, ts: str) -> str: """Bybit V5 signature: timestamp + api_key + recv_window + body. body = querystring sorted (GET) oppure JSON body raw come inviato (POST).""" sign_str = f"{ts}{self.api_key}5000{body}" return hmac.new(self.secret.encode(), sign_str.encode(), hashlib.sha256).hexdigest() def _request(self, method: str, path: str, params: dict = None, signed: bool = False) -> dict: params = dict(params or {}) dispatch = None if method.upper() != "GET" and path.startswith("/v5/order/"): if method.upper() != "POST" or path != "/v5/order/create": raise RuntimeError("ORDER_DISPATCH_BLOCKED unsupported order mutation") if params.get("reduceOnly"): # Un ordine reduceOnly NON puo' aprire una posizione (Bybit lo rifiuterebbe): # e' esente dal vincolo di collegamento all'alert. Resta tracciato in audit. audit_dispatch("attempt", {"exempt": "reduceOnly", "symbol": params.get("symbol"), "side": params.get("side"), "qty": params.get("qty"), "reduceOnly": True, "orderLinkId": params.get("orderLinkId", ""), "pid": os.getpid(), "program": os.path.basename(__file__)}) else: dispatch = validate_order_dispatch(params) audit_dispatch("attempt", dispatch) headers = {"Content-Type": "application/json"} url = BASE_URL + path body_for_sign = "" body_raw = "" if signed: ts = str(int(time.time() * 1000)) if method == "GET": # GET: signature su querystring nell'ORDINE DEL DICT (non sort alfabetico). # Bybit V5 richiede match esatto con la URL (requests.get mantiene ordine dict). # Es: get_leverage vuole category=symbol=settleCoin, sort alfabetico le mette # in altro ordine -> err 10004 signature mismatch. body_for_sign = "&".join(f"{k}={v}" for k, v in params.items()) else: # POST: signature su JSON body raw. requests.post(json=params) usa json.dumps(params) # con default separators=(', ', ': ') e NIENTE sort_keys. Match esatto: body_for_sign = json.dumps(params, separators=(", ", ": ")) headers["X-BAPI-API-KEY"] = self.api_key headers["X-BAPI-SIGN"] = self._sign(body_for_sign, ts) headers["X-BAPI-TIMESTAMP"] = ts headers["X-BAPI-RECV-WINDOW"] = "5000" if method == "GET": r = self.session.get(url, params=params, headers=headers, timeout=10) elif method == "POST": r = self.session.post(url, json=params, headers=headers, timeout=10) else: raise ValueError(f"Method {method} not supported") data = r.json() if dispatch is not None: audit_dispatch("response", dict(dispatch, retCode=data.get("retCode"), orderId=(data.get("result") or {}).get("orderId"))) if data.get("retCode", 0) != 0: raise RuntimeError(f"Bybit err {data.get('retCode')}: {data.get('retMsg')} | body_for_sign={body_for_sign[:200]}") return data def fetch_ohlcv(self, symbol: str, interval: str, limit: int = 200) -> list: """symbol: 'ZECUSDT' (senza /). interval: '240' (4h), '60' (1h), '15', ecc.""" params = {"category": "linear", "symbol": symbol, "interval": str(interval), "limit": str(limit)} data = self._request("GET", "/v5/market/kline", params, signed=False) return [[int(c[0]), float(c[1]), float(c[2]), float(c[3]), float(c[4]), float(c[5])] for c in reversed(data["result"]["list"])] def get_qty_step(self, symbol: str) -> float: """Ritorna lo step minimo di qty per il simbolo (es. 1.0 per AERO, 0.001 per altri).""" try: data = self._request("GET", "/v5/market/instruments-info", {"category": "linear", "symbol": symbol}, signed=False) for inst in data["result"]["list"]: if inst.get("symbol") == symbol: lot = inst.get("lotSizeFilter", {}) return float(lot.get("qtyStep", 1.0)) except Exception as e: pass return 1.0 # default conservativo def get_price_tick(self, symbol: str) -> float: """Ritorna lo step minimo di price per il simbolo (es. 0.01 per BTC, 0.0001 per AERO). FIX 2026-07-20: serve per arrotondare SL/TP/activePrice ed evitare errore Bybit 10001 (es. trailingStop activePrice BTC: 64871.996999... vs Bybit che accetta solo 2 decimali).""" try: data = self._request("GET", "/v5/market/instruments-info", {"category": "linear", "symbol": symbol}, signed=False) for inst in data["result"]["list"]: if inst.get("symbol") == symbol: pf = inst.get("priceFilter", {}) return float(pf.get("tickSize", 0.01)) except Exception as e: pass return 0.01 # default conservativo def round_price(self, symbol: str, price: float) -> float: """Arrotonda price al multiplo del tickSize del simbolo. FIX 2026-07-20: previene errore Bybit 10001 'TrailingProfit should greater than session_average_price' causato da troppi decimali (es. 64871.996999... vs 64872 atteso). FIX 2026-08-08: math.log10 per gestire notazione scientifica Python (es. 0.00001 -> str '1e-05' che non contiene '.', decimali calcolati a 0). Bugfix: WIFUSDT tickSize=0.00001 causava round_price(0.13981) -> 0.0 -> 34040 not modified.""" import math tick = self.get_price_tick(symbol) if tick >= 1: return float(int(price)) if tick <= 0: return price decimals = max(0, -int(math.floor(math.log10(tick)))) return round(price - (price % tick), decimals) def round_qty(self, symbol: str, qty: float) -> float: """Arrotonda qty al multiplo del qtyStep del simbolo.""" step = self.get_qty_step(symbol) if step >= 1: return float(int(qty)) decimals = len(str(step).rstrip('0').split('.')[-1]) if '.' in str(step) else 0 return round(qty - (qty % step), decimals) def fetch_balance(self) -> dict: data = self._request("GET", "/v5/account/wallet-balance", {"accountType": "UNIFIED"}, signed=True) return data["result"]["list"][0] if data["result"]["list"] else {} def fetch_positions(self, symbol: str = None) -> list: params = {"category": "linear", "settleCoin": "USDT"} if symbol: params["symbol"] = symbol data = self._request("GET", "/v5/position/list", params, signed=True) return [p for p in data["result"]["list"] if float(p.get("size", 0) or 0) > 0] def get_last_price(self, symbol: str) -> float: """Ritorna l'ultimo prezzo (lastPrice) per un symbol linear. Per SL/TP usa markPrice.""" data = self._request("GET", "/v5/market/tickers", {"category": "linear", "symbol": symbol}) lst = data.get("result", {}).get("list", []) if not lst: raise RuntimeError(f"No ticker data for {symbol}") return float(lst[0]["lastPrice"]) def get_mark_price(self, symbol: str) -> float: """Ritorna il mark price (per SL/TP triggerBy=Mark).""" data = self._request("GET", "/v5/market/tickers", {"category": "linear", "symbol": symbol}) lst = data.get("result", {}).get("list", []) if not lst: raise RuntimeError(f"No ticker data for {symbol}") return float(lst[0]["markPrice"]) def create_market_order(self, symbol: str, side: str, qty: float, *, order_link_id: str = "", reduce_only: bool = False) -> dict: """side: 'long'/'short' oppure 'buy'/'sell'. Bybit V5 vuole 'Buy'/'Sell'.""" bybit_side = "Buy" if side.lower() in ("buy", "long") else "Sell" params = {"category": "linear", "symbol": symbol, "side": bybit_side, "orderType": "Market", "qty": str(qty), "timeInForce": "GTC", "orderLinkId": order_link_id, "reduceOnly": reduce_only} return self._request("POST", "/v5/order/create", params, signed=True) def create_limit_order(self, symbol: str, side: str, qty: float, price: float, reduce_only: bool = True) -> dict: bybit_side = "Buy" if side.lower() in ("buy", "long") else "Sell" params = {"category": "linear", "symbol": symbol, "side": bybit_side, "orderType": "Limit", "qty": str(qty), "price": str(price), "timeInForce": "GTC", "reduceOnly": reduce_only} return self._request("POST", "/v5/order/create", params, signed=True) def set_trading_stop(self, symbol: str, sl_price: float = None, tp_price: float = None, trailing_stop: float = None, active_price: float = None) -> dict: """Set SL/TP/TSL sulla posizione aperta del symbol. - sl_price: trigger SL fisso (mark price) — None = non toccare - tp_price: trigger TP fisso (mark price) — None = non toccare - trailing_stop: distanza trailing in PERCENTUALE (es. 0.5 = 0.5%). None = non toccare. - active_price: prezzo trigger a cui si attiva il trailing. None = attivo subito. Bybit V5 /v5/position/trading-stop richiede almeno UNO tra stopLoss/takeProfit/trailingStop. Se passi piu' parametri, setta tutti; se ne passi solo uno, lascia gli altri invariati. Aggiunto trailingStop + activePrice (Mavis 2026-07-20) per trailing stop nativo Bybit V5. FIX 2026-07-20: arrotonda tutti i prezzi al tickSize del symbol per evitare errore 10001 (es. BTC activePrice 64871.996999 → 64872.00).""" if sl_price is None and tp_price is None and trailing_stop is None: raise ValueError("set_trading_stop: servono sl_price, tp_price o trailing_stop (almeno uno)") params = {"category": "linear", "symbol": symbol} if sl_price is not None: params["stopLoss"] = str(self.round_price(symbol, sl_price)) params["slTriggerBy"] = "MarkPrice" if tp_price is not None: params["takeProfit"] = str(self.round_price(symbol, tp_price)) params["tpTriggerBy"] = "MarkPrice" if trailing_stop is not None: params["trailingStop"] = str(trailing_stop) if active_price is not None: params["activePrice"] = str(self.round_price(symbol, active_price)) return self._request("POST", "/v5/position/trading-stop", params, signed=True) def set_leverage(self, symbol: str, leverage: int, side: str = "long") -> dict: """Imposta la leva per il simbolo PRIMA di aprire un ordine. Bybit V5 richiede di passare SIA buyLeverage CHE sellLeverage (anche se apri solo long), altrimenti ritorna 10001 'buy leverage not equal sell leverage'. Se la leva e' gia' impostata a quel valore, Bybit ritorna 110043 'leverage not modified' (non e' un errore bloccante, gia' gestito).""" params = {"category": "linear", "symbol": symbol, "buyLeverage": str(leverage), "sellLeverage": str(leverage)} try: return self._request("POST", "/v5/position/set-leverage", params, signed=True) except RuntimeError as e: # 110043 = leverage not modified (gia' impostata a quel valore, OK) if "110043" in str(e): return {"retCode": 110043, "retMsg": "leverage not modified (already at target)", "result": {}} raise def get_leverage(self, symbol: str) -> dict: """Ritorna la leva corrente per il simbolo. Richiede posizione aperta o che la leva sia stata precedentemente impostata. Ritorna dict con 'leverage', 'avgPrice', 'size', 'side', 'unrealisedPnl'. NOTA firma: ordine querystring DEVE essere (category, symbol, settleCoin) — Bybit non accetta sort alfabetico (vedi err 10004). Manteniamo ordine esplicito.""" data = self._request("GET", "/v5/position/list", {"category": "linear", "symbol": symbol, "settleCoin": "USDT"}, signed=True) result = {} for p in data.get("result", {}).get("list", []): if p.get("symbol") == symbol: result["leverage"] = p.get("leverage") result["avgPrice"] = p.get("avgPrice") result["size"] = p.get("size") result["side"] = p.get("side") result["unrealisedPnl"] = p.get("unrealisedPnl") break return result def fetch_open_orders(self, symbol: str = None) -> list: """Lista ordini aperti (status New, PartiallyFilled, Untriggered, Created). Usato per verificare che TP1/TP2 riduzioneOnly siano effettivamente registrati. symbol: 'ZECUSDT' oppure None per tutti.""" params = {"category": "linear", "settleCoin": "USDT"} if symbol: params["symbol"] = symbol data = self._request("GET", "/v5/order/realtime", params, signed=True) live_status = ("New", "PartiallyFilled", "Untriggered", "Created") return [o for o in data.get("result", {}).get("list", []) if o.get("orderStatus") in live_status] def get_position_tpsl(self, symbol: str) -> dict: """Legge SL/TP correnti dalla posizione (campi posizione, non ordini separati). Ritorna dict con stopLoss, takeProfit, slTriggerBy, avgPrice, size, side, leverage. Se la posizione non esiste o size=0, ritorna {}.""" data = self._request("GET", "/v5/position/list", {"category": "linear", "symbol": symbol, "settleCoin": "USDT"}, signed=True) for p in data.get("result", {}).get("list", []): if p.get("symbol") == symbol and float(p.get("size", 0) or 0) > 0: return { "stopLoss": p.get("stopLoss"), "takeProfit": p.get("takeProfit"), "slTriggerBy": p.get("slTriggerBy"), "tpTriggerBy": p.get("tpTriggerBy"), "avgPrice": p.get("avgPrice"), "size": p.get("size"), "side": p.get("side"), "leverage": p.get("leverage"), } return {} if __name__ == "__main__": # Test rapido c = BybitDemoClient() print("Balance:", c.fetch_balance().get("totalEquity", "N/A"), "USDT") print("OHLCV ZEC 4H (last 2):", c.fetch_ohlcv("ZECUSDT", 240, 2)) print("Open positions:", len(c.fetch_positions()))