""" 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("/opt/charter/.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', etc.""" 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 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, trailing_stop: float = None, active_price: float = None, sl_trigger_by: str = "MarkPrice", tp_trigger_by: str = "MarkPrice") -> dict: """Bybit V5 /v5/position/trading-stop โ€” imposta SL, TP, trailing stop nativi sulla posizione corrente. Eseguiti DALL'EXCHANGE anche se il bot รจ offline. BUG_005 fix: aggiunti parametri opzionali per supportare Charter enforcement: - tp_price: take profit nativo (singolo prezzo, riduce TUTTA la posizione) - trailing_stop: distanza trailing (es. "0.5" = 0.5% sotto mark) - active_price: prezzo di attivazione trailing (es. entry*1.03 = dopo TP1) Mantiene backward-compat: set_trading_stop(symbol, sl_price) funziona come prima. NB: Bybit consente UN SOLO takeProfit nativo per posizione. Per split 50/50 si usa TP medio pesato, oppure si combinano TP nativo (riduzione totale 50%) + ordine reduceOnly residuo (NO โ€” vedi fix live_engine per la strategia scelta). """ params = {"category": "linear", "symbol": symbol, "positionIdx": 0} # one-way mode if sl_price is not None: params["stopLoss"] = str(sl_price) params["slTriggerBy"] = sl_trigger_by if tp_price is not None: params["takeProfit"] = str(tp_price) params["tpTriggerBy"] = tp_trigger_by if trailing_stop is not None: params["trailingStop"] = str(trailing_stop) if active_price is not None: params["activePrice"] = str(active_price) if len(params) <= 3: raise ValueError("set_trading_stop: specificare almeno sl_price, tp_price o trailing_stop") return self._request("POST", "/v5/position/trading-stop", params, signed=True) def cancel_order(self, symbol: str, order_id: str) -> dict: """Cancella un singolo ordine pending per symbol. Usato per cleanup reduceOnly.""" params = {"category": "linear", "symbol": symbol, "orderId": order_id} return self._request("POST", "/v5/order/cancel", params, signed=True) def get_open_orders(self, symbol: str = None) -> list: """Ritorna lista ordini pending (Untriggered/New). Se symbol=None, tutti.""" params = {"category": "linear", "settleCoin": "USDT"} if symbol: params["symbol"] = symbol data = self._request("GET", "/v5/order/realtime", params, signed=True) return data.get("result", {}).get("list", []) def cancel_reduce_orders(self, symbol: str, keep_order_ids: list = None) -> int: """BUG_005 FIX 3: cancella tutti i reduceOnly pendenti per symbol, tranne quelli in keep_order_ids. Usato come cleanup PRIMA di piazzare nuovi TP/SL, per evitare conflitti. Ritorna numero ordini cancellati. NB: NON cancella gli stop/positions Bybit V5 nativi (non sono ordini separate, sono campi sulla posizione). Cancella SOLO ordini 'order' veri con reduceOnly=True. """ keep = set(keep_order_ids or []) cancelled = 0 try: orders = self.get_open_orders(symbol) for o in orders: oid = o.get("orderId") is_reduce = str(o.get("reduceOnly", "")).lower() == "true" status = o.get("orderStatus", "") if oid in keep: continue if is_reduce and status in ("New", "Untriggered", "PartiallyFilled"): try: self.cancel_order(symbol, oid) cancelled += 1 except Exception as e: print(f" [WARN] cancel_order {oid} fallita: {e}") return cancelled except Exception as e: print(f" [WARN] cancel_reduce_orders: {e}") return 0 def set_leverage(self, symbol: str, leverage: int, side: str = "long", bypass_if_already: bool = True) -> dict: """Imposta la leva per il simbolo PRIMA di aprire un ordine. Charter Mattia 12/07: TARGET 3x per TUTTI gli asset. - Se l'asset e' GIA' a 3x su Bybit: BYPASSA, NON chiamare set_leverage. - Altrimenti (leva attuale diversa da 3x): chiama set_leverage(3). Comportamento: - bypass_if_already=True (default): legge get_leverage() prima; se gia' al target, ritorna {"retCode": 0, "retMsg": "bypassed (already at target)", "result": {}, "bypassed": True}. - bypass_if_already=False: chiama sempre set_leverage. Bybit ritorna 110043 "leverage not modified" se gia' al target (gia' gestito come OK). Bybit V5 richiede SIA buyLeverage CHE sellLeverage (err 10001 se manca uno). """ if bypass_if_already: # Check PRIMA: se leva attuale == target, bypassa try: current = self.get_leverage(symbol) current_lev = current.get("leverage") if current_lev is not None and str(current_lev) == str(leverage): return { "retCode": 0, "retMsg": f"bypassed (leverage already at {leverage}x)", "result": {}, "bypassed": True, } except Exception: # Se get_leverage fallisce, prosegui con set_leverage (fallback) pass 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 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()))