Export Hyperliquid OHLCV Candles to Parquet

Build me a Python script that exports Hyperliquid OHLCV candles from Dwellir into Parquet.

Requirements:
- First check whether the `dwellir` CLI is installed.
- If it is not installed, recommend installing it with one of these options:
  - `brew tap dwellir-public/homebrew-tap && brew install dwellir`
  - `curl -fsSL https://raw.githubusercontent.com/dwellir-public/cli/main/scripts/install.sh | sh`
- Ask me to authenticate with `dwellir auth login`.
- Check authentication with `dwellir auth status`.
- Get an enabled API key name with `dwellir keys list --toon`.
- Use `dwellir endpoints search hyperliquid --ecosystem hyperliquid --network mainnet --key <name> --json` to discover both:
  - `Hyperliquid Index` for OHLCV range requests
  - `Hyperliquid HyperCore Info` for market discovery
- Support `--market BTC` for one market.
- Support `--all-markets` for all markets discovered from Info endpoint calls:
  - `{"type":"meta"}` for currently listed native perp names
  - `{"type":"spotMeta"}` for spot symbols, using canonical names such as `PURR/USDC` and `@INDEX` names for non-canonical pairs
  - `{"type":"perpDexs"}` for HIP-3 `prefix:ASSET` symbols
- Support `--max-markets` for short smoke tests before a full all-markets run.
- Use `GET /v1/candles/range` with `limit=5000`, follow `nextCursor`, and keep `end` exclusive.
- Support `1s`, `1m`, and `5m`.
- Stream pages into Parquet instead of collecting all rows in memory.
- Use columns: s, i, t, T, o, h, l, c, v, q, n, x.
- Show me how to run a one-market export and an all-markets export for one interval.

import argparse
import json
import subprocess
import sys
from typing import Any

import pyarrow as pa
import pyarrow.parquet as pq
import requests

VALID_INTERVALS = {"1s", "1m", "5m"}
SCHEMA = pa.schema(
    [
        ("s", pa.string()),
        ("i", pa.string()),
        ("t", pa.int64()),
        ("T", pa.int64()),
        ("o", pa.string()),
        ("h", pa.string()),
        ("l", pa.string()),
        ("c", pa.string()),
        ("v", pa.string()),
        ("q", pa.string()),
        ("n", pa.int64()),
        ("x", pa.bool_()),
    ]
)


def load_endpoint_catalog(key_name: str) -> dict[str, Any]:
    try:
        raw = subprocess.check_output(
            [
                "dwellir",
                "endpoints",
                "search",
                "hyperliquid",
                "--ecosystem",
                "hyperliquid",
                "--network",
                "mainnet",
                "--key",
                key_name,
                "--json",
            ],
            text=True,
        )
    except FileNotFoundError as exc:
        raise SystemExit("Install the dwellir CLI first, then run `dwellir auth login`.") from exc
    return json.loads(raw)


def endpoint_url(catalog: dict[str, Any], endpoint_name: str) -> str:
    for item in catalog.get("data", []):
        if item.get("name") == endpoint_name:
            return item["networks"][0]["nodes"][0]["https"]
    raise RuntimeError(f"Could not find endpoint named {endpoint_name!r}")


def post_info(session: requests.Session, info_url: str, payload: dict[str, Any]) -> Any:
    response = session.post(info_url, json=payload, timeout=30)
    response.raise_for_status()
    return response.json()


def discover_markets(session: requests.Session, info_url: str) -> list[str]:
    meta = post_info(session, info_url, {"type": "meta"})
    spot_meta = post_info(session, info_url, {"type": "spotMeta"})
    perp_dexs = post_info(session, info_url, {"type": "perpDexs"})

    perps = [
        asset["name"]
        for asset in meta.get("universe", [])
        if isinstance(asset, dict) and asset.get("name") and not asset.get("isDelisted")
    ]
    spots = [
        pair["name"] if pair.get("isCanonical") else f"@{pair['index']}"
        for pair in spot_meta.get("universe", [])
        if isinstance(pair, dict) and pair.get("name") and "index" in pair
    ]

    hip3 = []
    dex_items = perp_dexs if isinstance(perp_dexs, list) else perp_dexs.get("dexs", [])
    for dex in dex_items:
        if not isinstance(dex, dict):
            continue
        for pair in dex.get("assetToStreamingOiCap", []) or []:
            if isinstance(pair, list) and pair:
                hip3.append(pair[0])

    markets = []
    seen = set()
    for market in perps + spots + hip3:
        if market not in seen:
            seen.add(market)
            markets.append(market)
    return markets


def candle_pages(
    session: requests.Session,
    index_base_url: str,
    market: str,
    interval: str,
    start: str | None,
    end: str | None,
):
    cursor = None
    while True:
        params = {"market": market, "interval": interval, "limit": "5000"}
        if cursor:
            params["cursor"] = cursor
        elif start:
            params["start"] = start
        if end:
            params["end"] = end

        response = session.get(
            f"{index_base_url}/v1/candles/range",
            params=params,
            timeout=60,
        )
        response.raise_for_status()
        page = response.json()
        rows = page.get("data", [])
        if rows:
            yield rows
        if not page.get("hasMore"):
            break
        cursor = page["nextCursor"]


def table_from_rows(rows: list[dict[str, Any]]) -> pa.Table:
    return pa.Table.from_pylist(rows, schema=SCHEMA)


def export_parquet(args: argparse.Namespace) -> int:
    if args.interval not in VALID_INTERVALS:
        raise SystemExit(f"Unsupported interval: {args.interval}")
    if bool(args.market) == bool(args.all_markets):
        raise SystemExit("Pass exactly one of --market or --all-markets")

    catalog = load_endpoint_catalog(args.key_name)
    index_base_url = endpoint_url(catalog, "Hyperliquid Index")
    info_url = endpoint_url(catalog, "Hyperliquid HyperCore Info")

    session = requests.Session()
    markets = [args.market] if args.market else discover_markets(session, info_url)
    if args.max_markets is not None:
        markets = markets[: args.max_markets]
    writer = pq.ParquetWriter(args.output, SCHEMA, compression="zstd")
    written = 0

    try:
        for index, market in enumerate(markets, start=1):
            print(f"exporting {market} ({index}/{len(markets)})", file=sys.stderr)
            for rows in candle_pages(session, index_base_url, market, args.interval, args.start, args.end):
                writer.write_table(table_from_rows(rows))
                written += len(rows)
        if written == 0:
            writer.write_table(pa.Table.from_pylist([], schema=SCHEMA))
    finally:
        writer.close()

    print(f"wrote {written} candles for {len(markets)} market(s) to {args.output}")
    return 0


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--key-name", required=True, help="Dwellir API key name from `dwellir keys list --toon`")
    parser.add_argument("--market", help="One market symbol, for example BTC, @142, or hyna:ETH")
    parser.add_argument("--all-markets", action="store_true", help="Discover and export all currently listed markets")
    parser.add_argument("--max-markets", type=int, help="Limit discovered markets for smoke tests")
    parser.add_argument("--interval", required=True, choices=sorted(VALID_INTERVALS))
    parser.add_argument("--start", help="Inclusive ISO-8601 UTC start, for example 2026-03-01T00:00:00Z")
    parser.add_argument("--end", help="Exclusive ISO-8601 UTC end, for example 2026-04-01T00:00:00Z")
    parser.add_argument("--output", required=True)
    return export_parquet(parser.parse_args())


if __name__ == "__main__":
    raise SystemExit(main())