Spaces:
Running on Zero
Running on Zero
File size: 2,275 Bytes
aa66616 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 | """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])
|