Spaces:
Running
Running
| #!/usr/bin/env python3 | |
| """Hyperliquid API client — candle fetching, order placement, and live execution helpers.""" | |
| import math | |
| import os | |
| import time | |
| import logging | |
| import requests | |
| from typing import Dict, Optional, List, Any | |
| import urllib3 | |
| import certifi | |
| logger = logging.getLogger(__name__) | |
| class HyperliquidAPI: | |
| def __init__( | |
| self, | |
| verify_ssl=True, | |
| min_request_interval_ms: int = 250, | |
| max_retries: int = 5, | |
| max_backoff_seconds: int = 30, | |
| ): | |
| self.base_url = "https://api.hyperliquid.xyz" | |
| self.verify_ssl = verify_ssl | |
| self.min_request_interval_ms = max(0, int(min_request_interval_ms)) | |
| self.base_min_request_interval_ms = self.min_request_interval_ms | |
| self.max_retries = max(1, int(max_retries)) | |
| self.max_backoff_seconds = max(1, int(max_backoff_seconds)) | |
| self._last_request_ts = 0.0 | |
| self.session = requests.Session() | |
| self.session.headers.update({ | |
| 'Content-Type': 'application/json', | |
| 'User-Agent': 'HyperliquidBot/1.0' | |
| }) | |
| if not self.verify_ssl: | |
| self._disable_ssl_verification() | |
| else: | |
| self.session.verify = self._get_ca_bundle() | |
| def _disable_ssl_verification(self): | |
| """Disable SSL verification for the session permanently""" | |
| self.verify_ssl = False | |
| self.session.verify = False | |
| urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) | |
| logger.debug("SSL verification disabled for this session.") | |
| def _get_ca_bundle(self): | |
| cert_paths = [ | |
| '/etc/ssl/certs/ca-certificates.crt', | |
| '/etc/ssl/ca-bundle.pem', | |
| '/usr/local/etc/openssl/cert.pem', | |
| '/opt/homebrew/etc/openssl/cert.pem', | |
| ] | |
| for path in cert_paths: | |
| if os.path.exists(path): | |
| return path | |
| return True | |
| def _make_request(self, endpoint: str, data: dict) -> dict: | |
| url = f"{self.base_url}/{endpoint}" | |
| for attempt in range(self.max_retries): | |
| try: | |
| # Global pacing | |
| elapsed_ms = (time.time() - self._last_request_ts) * 1000.0 | |
| if elapsed_ms < self.min_request_interval_ms: | |
| time.sleep((self.min_request_interval_ms - elapsed_ms) / 1000.0) | |
| # Explicitly pass verify to ensure it's respected | |
| response = self.session.post(url, json=data, timeout=30, verify=self.session.verify) | |
| self._last_request_ts = time.time() | |
| response.raise_for_status() | |
| return response.json() | |
| except (requests.exceptions.SSLError, requests.exceptions.ConnectionError) as e: | |
| if self.verify_ssl: | |
| logger.warning(f"SSL/Connection failure. Retrying with verification disabled... Error: {e}") | |
| self._disable_ssl_verification() | |
| continue | |
| logger.error(f"Request failed: {e}") | |
| raise | |
| except requests.exceptions.HTTPError as e: | |
| if e.response.status_code == 429: | |
| time.sleep(min(self.max_backoff_seconds, 2 ** (attempt + 1))) | |
| continue | |
| raise | |
| except Exception as e: | |
| if attempt == self.max_retries - 1: | |
| raise | |
| time.sleep(min(self.max_backoff_seconds, 2 ** attempt)) | |
| def get_meta(self) -> Dict[str, Any]: | |
| return self._make_request("info", {"type": "meta"}) | |
| def get_candles(self, coin: str, interval: str, start_time: int) -> List[dict]: | |
| try: | |
| data = {"type": "candleSnapshot", "req": {"coin": coin, "interval": interval, "startTime": start_time}} | |
| return self._make_request("info", data) | |
| except Exception as e: | |
| logger.error(f"Error fetching candles for {coin}: {e}") | |
| return [] | |
| def get_asset_ctx(self, coin: str) -> Optional[Dict[str, Any]]: | |
| """Return the perp asset context for coin (funding rate, open interest, mark price, etc.).""" | |
| try: | |
| data = self._make_request("info", {"type": "metaAndAssetCtxs"}) | |
| meta, ctxs = data[0], data[1] | |
| for asset, ctx in zip(meta.get("universe", []), ctxs): | |
| if asset.get("name") == coin: | |
| return ctx | |
| except Exception as e: | |
| logger.warning(f"Failed to fetch asset ctx for {coin}: {e}") | |
| return None | |
| def create_live_execution_clients(self, private_key: str, account_address: Optional[str] = None, timeout: float = 15.0, skip_ws: bool = True) -> Dict[str, Any]: | |
| from eth_account import Account | |
| from hyperliquid.exchange import Exchange | |
| from hyperliquid.info import Info | |
| wallet = Account.from_key(private_key) | |
| meta = self.get_meta() | |
| info = Info(self.base_url, skip_ws=skip_ws, meta=meta) | |
| exchange = Exchange(wallet, self.base_url, meta=meta, account_address=account_address) | |
| # Sync SSL settings to SDK clients | |
| for client in (info, exchange, exchange.info): | |
| client.session.verify = self.session.verify | |
| if not self.verify_ssl: | |
| client.session.mount('https://', requests.adapters.HTTPAdapter(pool_connections=10, pool_maxsize=10)) | |
| return {'wallet': wallet, 'info': info, 'exchange': exchange} | |
| def get_sz_decimals(self, coin: str) -> int: | |
| """Return the number of decimal places allowed for order SIZE on this coin.""" | |
| try: | |
| meta = self.get_meta() | |
| for asset in meta.get("universe", []): | |
| if asset.get("name") == coin: | |
| return int(asset.get("szDecimals", 4)) | |
| except Exception: | |
| pass | |
| return 4 # safe fallback | |
| def get_px_decimals(self, coin: str) -> int: | |
| """ | |
| Return the number of decimal places allowed for order PRICE on this coin. | |
| Uses allMids for the current mid price (universeContext is not present | |
| in the perps meta response). | |
| Rule: max_decimals = max(0, 5 - floor(log10(price)) - 1) | |
| Examples: | |
| BTC ~95000 → 1 decimal | |
| ETH ~3500 → 2 decimals | |
| SOL ~90 → 2 decimals | |
| TIA ~5 → 4 decimals | |
| DOGE ~0.15 → 5 decimals | |
| """ | |
| try: | |
| all_mids = self._make_request("info", {"type": "allMids"}) | |
| mid_px = float(all_mids.get(coin, 0)) | |
| if mid_px > 0: | |
| mag = math.floor(math.log10(mid_px)) | |
| return max(0, 5 - mag - 1) | |
| except Exception: | |
| pass | |
| return 2 # safe fallback for mid-range coins like SOL | |
| def place_market_order( | |
| self, | |
| exchange: Any, | |
| coin: str, | |
| is_buy: bool, | |
| sz: float, | |
| px: float, | |
| sl: Optional[float] = None, | |
| tp: Optional[float] = None, | |
| limit_sl: bool = False, | |
| ) -> Any: | |
| """ | |
| Taker entry (IOC) + exchange-native SL/TP. | |
| sl_type: | |
| limit_sl=False → SL is market trigger (taker, guaranteed fill) | |
| limit_sl=True → SL is limit trigger (maker, may not fill on gap) | |
| Fee profile (your account rates 0.0432% taker / 0.0144% maker): | |
| Hype/Scalp: taker entry + limit TP + market SL | |
| win: 0.0432 + 0.0144 = 0.0576% round-trip | |
| loss: 0.0432 + 0.0432 = 0.0864% round-trip | |
| """ | |
| sz_dec = self.get_sz_decimals(coin) | |
| px_dec = self.get_px_decimals(coin) | |
| sz = round(float(sz), sz_dec) | |
| if sz <= 0: | |
| logger.error(f"Size rounded to zero for {coin} (sz_dec={sz_dec}). Skipping.") | |
| return None | |
| # ── Check notional after rounding ─────────────────────────────────── | |
| notional_after_rounding = sz * px | |
| if notional_after_rounding < 10.0: | |
| logger.error( | |
| f"[{coin}] Notional ${notional_after_rounding:.2f} falls below $10 after rounding " | |
| f"(sz={sz}, px={px:.{px_dec}f}). Skipping." | |
| ) | |
| return None | |
| limit_px = round(px * 1.02, px_dec) | |
| logger.info(f"Placing {'BUY' if is_buy else 'SELL'} limit-IOC {sz} {coin} @ {limit_px} (mark={round(px, px_dec)}) [px_dec={px_dec} sz_dec={sz_dec}]") | |
| try: | |
| result = exchange.order(coin, is_buy, sz, limit_px, {"limit": {"tif": "Ioc"}}) | |
| logger.info(f"Order response for {coin}: {result}") | |
| except Exception as e: | |
| logger.error(f"Entry order failed: {e}") | |
| return None | |
| try: | |
| statuses = result.get("response", {}).get("data", {}).get("statuses", []) | |
| filled_sz = sum(float(s["filled"].get("totalSz", 0)) for s in statuses if "filled" in s) | |
| errors = [s["error"] for s in statuses if "error" in s] | |
| if errors: | |
| logger.error(f"Entry rejected for {coin}: {errors}") | |
| return result | |
| if filled_sz <= 0: | |
| logger.warning(f"Entry IOC for {coin} did not fill. Skipping SL/TP.") | |
| return result | |
| logger.info(f"Entry filled {filled_sz} {coin}. Placing SL/TP on exchange...") | |
| sz = round(filled_sz, sz_dec) | |
| except Exception as e: | |
| logger.warning(f"Could not parse fill status for {coin}: {e}. Attempting SL/TP anyway.") | |
| # ── Stop-Loss ───────────────────────────────────────────────────────── | |
| if sl is not None: | |
| try: | |
| sl_px = round(float(sl), px_dec) | |
| sl_side = not is_buy | |
| is_mkt = not limit_sl | |
| exchange.order( | |
| coin, sl_side, sz, sl_px, | |
| {"trigger": {"triggerPx": sl_px, "isMarket": is_mkt, "tpsl": "sl"}}, | |
| reduce_only=True, | |
| ) | |
| sl_type = "market trigger" if is_mkt else "limit trigger" | |
| logger.info(f"SL placed for {coin} @ {sl_px} ({sl_type})") | |
| except Exception as e: | |
| logger.error(f"SL order failed for {coin}: {e}") | |
| # ── Take-Profit (limit trigger — maker exit) ────────────────────────── | |
| if tp is not None: | |
| try: | |
| tp_px = round(float(tp), px_dec) | |
| tp_side = not is_buy | |
| exchange.order( | |
| coin, tp_side, sz, tp_px, | |
| {"trigger": {"triggerPx": tp_px, "isMarket": False, "tpsl": "tp"}}, | |
| reduce_only=True, | |
| ) | |
| logger.info(f"TP placed for {coin} @ {tp_px} (limit trigger — maker exit)") | |
| except Exception as e: | |
| logger.error(f"TP order failed for {coin}: {e}") | |
| return result | |