Spaces:
Running on Zero
Running on Zero
| """Hyperliquid public REST API client — candle and price data.""" | |
| from __future__ import annotations | |
| import time | |
| from datetime import datetime, timezone | |
| from typing import Literal | |
| import httpx | |
| import pandas as pd | |
| HL_INFO_URL = "https://api.hyperliquid.xyz/info" | |
| Interval = Literal["15m", "1h", "4h", "1d"] | |
| _INTERVAL_MS: dict[str, int] = { | |
| "15m": 15 * 60 * 1_000, | |
| "1h": 60 * 60 * 1_000, | |
| "4h": 4 * 60 * 60 * 1_000, | |
| "1d": 24 * 60 * 60 * 1_000, | |
| } | |
| def fetch_candles( | |
| coin: str = "HYPE", | |
| interval: Interval = "4h", | |
| lookback: int = 200, | |
| timeout: float = 15.0, | |
| ) -> pd.DataFrame: | |
| """Return a DataFrame of OHLCV candles sorted oldest-first. | |
| Columns: open_time (UTC datetime), open, high, low, close, volume, trades | |
| """ | |
| now_ms = int(time.time() * 1_000) | |
| step_ms = _INTERVAL_MS[interval] | |
| start_ms = now_ms - lookback * step_ms | |
| payload = { | |
| "type": "candleSnapshot", | |
| "req": { | |
| "coin": coin, | |
| "interval": interval, | |
| "startTime": start_ms, | |
| "endTime": now_ms, | |
| }, | |
| } | |
| with httpx.Client(timeout=timeout) as client: | |
| resp = client.post(HL_INFO_URL, json=payload) | |
| resp.raise_for_status() | |
| raw = resp.json() | |
| if not raw: | |
| raise ValueError(f"No candles returned for {coin}/{interval}") | |
| rows = [] | |
| for c in raw: | |
| rows.append( | |
| { | |
| "open_time": datetime.fromtimestamp(c["t"] / 1_000, tz=timezone.utc), | |
| "open": float(c["o"]), | |
| "high": float(c["h"]), | |
| "low": float(c["l"]), | |
| "close": float(c["c"]), | |
| "volume": float(c["v"]), | |
| "trades": int(c["n"]), | |
| } | |
| ) | |
| df = pd.DataFrame(rows).sort_values("open_time").reset_index(drop=True) | |
| return df | |
| def fetch_mark_price(coin: str = "HYPE", timeout: float = 10.0) -> float: | |
| """Return the current mark price for *coin*.""" | |
| with httpx.Client(timeout=timeout) as client: | |
| resp = client.post(HL_INFO_URL, json={"type": "allMids"}) | |
| resp.raise_for_status() | |
| mids = resp.json() | |
| if coin not in mids: | |
| raise ValueError(f"{coin} not found in allMids response") | |
| return float(mids[coin]) | |