""" 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 ENV_FILE = Path(r"G:\AI TRADING ENGINE\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 = params or {} 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 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 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) -> 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"} 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) -> dict: """Set SL e/o TP sulla posizione aperta del symbol. - sl_price: trigger SL (mark price) — None = non toccare - tp_price: trigger TP (mark price) — None = non toccare Bybit V5 /v5/position/trading-stop richiede almeno UNO tra stopLoss/takeProfit. Se passi entrambi, setta entrambi; se passi solo uno, lascia l'altro invariato (Bybit tratta i campi omessi come 'mantieni valore attuale').""" if sl_price is None and tp_price is None: raise ValueError("set_trading_stop: servono sl_price o tp_price (almeno uno)") params = {"category": "linear", "symbol": symbol} if sl_price is not None: params["stopLoss"] = str(sl_price) params["slTriggerBy"] = "MarkPrice" if tp_price is not None: params["takeProfit"] = str(tp_price) params["tpTriggerBy"] = "MarkPrice" 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()))