#!/usr/bin/env python3 """ Bybit V5 demo client (testnet). Wrapper leggero: set_leverage, get_last_price, place_order, get_instrument_info. Usa le env BYBIT_API_KEY e BYBIT_API_SECRET. NOTA: richiede `pybit` (pip install pybit). """ import os import time import hmac import hashlib import json import logging from urllib.parse import urlencode from urllib.request import Request, urlopen from urllib.error import HTTPError, URLError log = logging.getLogger("bybit") class BybitDemoClient: BASE_DEMO = "https://api-demo.bybit.com" # Bybit V5 demo (Mattia, chiavi demo personali) BASE_TESTNET = "https://api-testnet.bybit.com" # Bybit V5 testnet ufficiale (richiede registrazione) BASE_MAINNET = "https://api.bybit.com" # Bybit V5 mainnet (soldi veri) def __init__(self, api_key: str, api_secret: str, testnet: bool = True, demo: bool = True): self.api_key = api_key self.api_secret = api_secret # FIX 2026-07-17 (Verifier): default ora e' BASE_DEMO per matchare le chiavi demo di Mattia. # Mattia usa chiavi di api-demo.bybit.com (non testnet ufficiale). Priorita': demo > testnet > mainnet. if demo: self.base = self.BASE_DEMO elif testnet: self.base = self.BASE_TESTNET else: self.base = self.BASE_MAINNET # ----- low-level signed request ----- def _sign(self, payload: str) -> str: """Bybit V5: firma HMAC-SHA256 su timestamp + api_key + recv_window + payload. payload = query_string per GET, body_json per POST.""" return hmac.new( self.api_secret.encode("utf-8"), payload.encode("utf-8"), hashlib.sha256 ).hexdigest() def _request(self, method: str, path: str, params: dict | None = None, body: dict | None = None) -> dict: params = params or {} timestamp = str(int(time.time() * 1000)) params["api_key"] = self.api_key params["timestamp"] = timestamp params["recv_window"] = "5000" # FIX 2026-07-17 (Verifier): Bybit V5 firma correttamente per GET vs POST # origin_string per GET = `timestamp + api_key + recv_window + urlencode(sorted_all_params_con_auth_incluso)` # origin_string per POST = `timestamp + api_key + recv_window + body_json` # L'urlencode per GET DEVE includere api_key, timestamp, recv_window (oltre ai business params) # per matchare l'origin_string restituito da Bybit in caso di errore firma. # sign NON va incluso (siamo noi a calcolarlo). # Vedi: https://bybit-exchange.github.io/docs/v5/guide#authentication if method == "GET": # urlencode di TUTTI i params TRANNE sign all_but_sign = {k: v for k, v in params.items() if k != "sign"} sign_payload = f"{timestamp}{self.api_key}5000{urlencode(sorted(all_but_sign.items()))}" else: # POST body_json = json.dumps(body, separators=(",", ":")) if body else "" sign_payload = f"{timestamp}{self.api_key}5000{body_json}" params["sign"] = self._sign(sign_payload) url = f"{self.base}{path}?{urlencode(params)}" data = json.dumps(body, separators=(",", ":")).encode("utf-8") if body else None req = Request(url, data=data, method=method) req.add_header("Content-Type", "application/json") # FIX 2026-07-17 (Verifier): Bybit V5 richiede gli auth headers X-BAPI-* per le POST. # Senza, le POST con body (set_leverage, place_order) falliscono con "apiKey is missing" # e poi "apiTimestamp is missing". Aggiungo tutti e 3 (api_key, timestamp, sign, recv_window). # Per le GET questi sono gia' passati come query param, quindi innocuo duplicarli come header. req.add_header("X-BAPI-API-KEY", self.api_key) req.add_header("X-BAPI-TIMESTAMP", timestamp) req.add_header("X-BAPI-SIGN", params["sign"]) req.add_header("X-BAPI-RECV-WINDOW", "5000") try: with urlopen(req, timeout=10) as resp: return json.loads(resp.read().decode("utf-8")) except HTTPError as e: body = e.read().decode("utf-8", errors="replace") log.error("Bybit HTTP error %s: %s", e.code, body) raise except URLError as e: log.error("Bybit URL error: %s", e) raise # ----- public helpers ----- def get_last_price(self, symbol: str) -> float: """Ritorna l'ultimo prezzo (mark price) di un simbolo.""" r = self._request("GET", "/v5/market/tickers", params={"category": "linear", "symbol": symbol}) result = r.get("result", {}) lst = result.get("list", []) if not lst: raise RuntimeError(f"no ticker for {symbol}") return float(lst[0]["markPrice"]) def get_instrument_info(self, symbol: str) -> dict: """Ritorna info strumento (lot size, tick size, ecc.) per un simbolo.""" r = self._request("GET", "/v5/market/instruments-info", params={"category": "linear", "symbol": symbol}) result = r.get("result", {}) lst = result.get("list", []) if not lst: raise RuntimeError(f"no instrument info for {symbol}") return lst[0] def set_leverage(self, symbol: str, leverage: int) -> dict: """Imposta leverage per il simbolo (linear).""" body = { "category": "linear", "symbol": symbol, "buyLeverage": str(leverage), "sellLeverage": str(leverage), } # NB: questo endpoint NON vuole la firma perché è un'azione utente # (alcune versioni di Bybit la richiedono, gestiamo entrambi i casi) try: return self._request("POST", "/v5/position/set-leverage", body=body) except Exception as e: log.warning("set_leverage failed (potrebbe essere già impostata): %s", e) return {} def place_order( self, symbol: str, side: str, qty: float, leverage: int = 3, order_type: str = "Market", price: float | None = None, sl_price: float | None = None, tp_price: float | None = None, ) -> dict: """Invia un ordine Market (o Limit se price è dato).""" body = { "category": "linear", "symbol": symbol, "side": side.capitalize(), # "Buy" / "Sell" "orderType": order_type, "qty": str(qty), "timeInForce": "GTC" if order_type == "Limit" else "IOC", } if price is not None and order_type == "Limit": body["price"] = str(price) if sl_price is not None: body["stopLoss"] = str(sl_price) if tp_price is not None: body["takeProfit"] = str(tp_price) return self._request("POST", "/v5/order/create", body=body)