Backfill the Full Hyperliquid OHLCV Archive

Build me a resumable Python script that backfills the full Hyperliquid candle archive from Dwellir for one market and one interval at a time, using a multi-process HTTP/2 client that can sustain several thousand requests per second.

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`
- After installation, ask me to authenticate it by running `dwellir auth login`.
- Check authentication with `dwellir auth status`.
- Get an enabled API key with `dwellir keys list --toon`. If there are multiple good candidates, ask me which key to use.
- Discover the Hyperliquid Index endpoint with `dwellir endpoints search hyperliquid --ecosystem hyperliquid --network mainnet --key <name> --toon`.
- Use the Dwellir Hyperliquid REST candles endpoint at `api-hyperliquid-index.n.dwellir.com`.
- Export one market and one interval at a time.
- Use `httpx` with `http2=True` (not `requests`). A single Python asyncio process caps around `1000 RPS` regardless of transport; HTTP/2 multi-process is what scales past that.
- Split the time range into N worker processes using `multiprocessing.spawn`. Each worker runs its own asyncio event loop and its own `httpx.AsyncClient(http2=True)` pool.
- Warm separate HTTP/2 connections per worker so they pick up distinct `DWSESSION` cookies from the platform's backend router.
- Within each worker, run many concurrent async tasks that share the worker's HTTP/2 connections; HTTP/2 multiplexing lets one connection carry many in-flight streams.
- Apply a token-bucket rate limiter per worker so the aggregate request rate matches the plan ceiling (Scale: `5000 RPS`, Growth: `500 RPS`, Developer: `100 RPS`, Free: `20 RPS`).
- For the Scale plan, 8 workers is a good default. For Free/Developer/Growth, a single worker is enough.
- Treat the data as sparse: if a bucket returns `404`, count it and skip it rather than fabricating a candle.
- The OHLCV archive starts at `2025-07-27T08:00:00Z`.
- Iterate bucket-open timestamps between a start and end time.
- Support `1s`, `1m`, and `5m`.
- Write per-shard CSVs with columns: `s, i, t, T, o, h, l, c, v, q, n, x`.
- Persist resume state per shard so the export can continue after an interruption.
- Make the resume logic safe if the process stops between writing a CSV row and updating the shard state files.
- Write the shard state files atomically so a partial write does not break the next run.
- Merge the shard outputs back into one CSV sorted by the candle open time `t` when the run completes.
- Show me how to run a short validation window first, then how to run a larger backfill for BTC `1m` and `1s`.

{
  "s": "BTC",
  "i": "1m",
  "t": 1774911600000,
  "T": 1774911659999,
  "o": "66745",
  "h": "66746",
  "l": "66687",
  "c": "66687",
  "v": "4.78586",
  "q": "319316.665",
  "n": 256,
  "x": true
}

"""Resumable Hyperliquid OHLCV backfill via the Dwellir Hyperliquid Index endpoint.

One market and one interval per run. Splits the time range into N worker
processes, each running an asyncio + HTTP/2 (httpx) loop with sticky-session
discovery for backend pinning. Writes one CSV part per shard plus an atomic
resume state file. Merges parts into one ordered CSV when the run completes.

Usage:

    python backfill_hyperliquid_ohlcv.py \\
        --api-key $DWELLIR_API_KEY \\
        --market BTC --interval 1m \\
        --start 2026-03-01T00:00:00Z --end 2026-04-01T00:00:00Z \\
        --output btc-1m.csv \\
        --rate 5000 --workers 8

Per-plan recommended settings:

    Free       --rate 20    --workers 1
    Developer  --rate 100   --workers 1
    Growth     --rate 500   --workers 1
    Scale      --rate 5000  --workers 8
"""
from __future__ import annotations

import argparse
import asyncio
import csv
import json
import multiprocessing as mp
import os
import sys
import time
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path

import httpx

ENDPOINT_HOST = "https://api-hyperliquid-index.n.dwellir.com"
ARCHIVE_FLOOR = "2025-07-27T08:00:00Z"

INTERVAL_STEP = {
    "1s": timedelta(seconds=1),
    "1m": timedelta(minutes=1),
    "5m": timedelta(minutes=5),
}

CSV_FIELDS = ["s", "i", "t", "T", "o", "h", "l", "c", "v", "q", "n", "x"]


def parse_utc(value: str) -> datetime:
    return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(timezone.utc)


def format_utc(value: datetime) -> str:
    return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")


def split_range(start: datetime, end: datetime, step: timedelta, n_shards: int):
    total = max(1, int((end - start) / step))
    n_shards = max(1, min(n_shards, total))
    base, rem = divmod(total, n_shards)
    shards = []
    cursor = start
    for i in range(n_shards):
        size = base + (1 if i < rem else 0)
        shard_end = cursor + step * size
        if shard_end > end:
            shard_end = end
        shards.append((cursor, shard_end))
        cursor = shard_end
    return shards


def atomic_write(path: Path, payload: str):
    tmp = path.with_name(path.name + ".tmp")
    tmp.write_text(payload)
    tmp.replace(path)


def write_state(path: Path, payload: dict):
    atomic_write(path, json.dumps(payload, indent=2, sort_keys=True) + "\n")


def load_state(path: Path):
    if not path.exists():
        return None
    try:
        return json.loads(path.read_text())
    except json.JSONDecodeError:
        return None


class TokenBucketAsync:
    def __init__(self, rate_per_second: float):
        self.rate = float(rate_per_second)
        self.capacity = max(self.rate, 1.0)
        self.tokens = self.capacity
        self.last = time.monotonic()

    async def acquire(self):
        while True:
            now = time.monotonic()
            self.tokens = min(self.capacity, self.tokens + (now - self.last) * self.rate)
            self.last = now
            if self.tokens >= 1.0:
                self.tokens -= 1.0
                return
            await asyncio.sleep((1.0 - self.tokens) / self.rate)


async def warm_clients(base: str, market: str, interval: str, probe_time: datetime,
                       desired: int, max_attempts: int):
    seen = set()
    out = []
    for _ in range(max_attempts):
        if len(out) >= desired:
            break
        limits = httpx.Limits(max_connections=200, max_keepalive_connections=200)
        client = httpx.AsyncClient(http2=True, timeout=30.0, limits=limits)
        try:
            r = await client.get(
                f"{base}/v1/candles",
                params={"market": market, "interval": interval, "time": format_utc(probe_time)},
            )
        except Exception:
            await client.aclose()
            continue
        if r.status_code not in (200, 404):
            await client.aclose()
            continue
        backend = r.cookies.get("DWSESSION") or "unknown"
        if backend in seen:
            await client.aclose()
            continue
        seen.add(backend)
        out.append((backend, client))
    if not out:
        out.append(("unknown", httpx.AsyncClient(http2=True, timeout=30.0)))
    return out


@dataclass
class ShardResult:
    shard_index: int
    fetched: int
    skipped: int
    errors: int
    seconds: float
    sticky_backends: list


async def shard_worker(shard_index, base, market, interval, shard_start, shard_end,
                       rate_per_worker, sticky_per_worker, workers_per_client,
                       csv_path, state_path):
    """Process one time shard with crash-safe resume.

    Correctness invariants:
    - The state file's `next_time` is always backed by the highest *contiguous*
      completed bucket, never just the max in-flight completion. Async tasks
      complete out of order, so a max-watermark would skip holes on resume.
    - Candles reach the CSV before they are counted in state. Consumers only
      append to in-memory buffers; a periodic flusher writes the CSV first,
      then updates counters and advances the frontier, then writes state.
    - The merge step deduplicates by candle open time `t` to absorb the rare
      duplicate that can occur if the process is killed between flushing the
      CSV and writing the new state.
    """
    step = INTERVAL_STEP[interval]
    state = load_state(state_path)
    cursor = shard_start
    fetched = skipped = errors = 0
    if state:
        cursor = parse_utc(state["next_time"])
        fetched = state.get("fetched", 0)
        skipped = state.get("skipped", 0)
        errors = state.get("errors", 0)
    if cursor >= shard_end:
        return ShardResult(shard_index, fetched, skipped, errors, 0.0, [])

    clients = await warm_clients(base, market, interval, shard_end - step,
                                 sticky_per_worker, sticky_per_worker * 3)
    backends = [b for b, _ in clients]

    if not csv_path.exists() or csv_path.stat().st_size == 0:
        with csv_path.open("w", newline="") as f:
            csv.DictWriter(f, fieldnames=CSV_FIELDS).writeheader()

    queue: asyncio.Queue = asyncio.Queue()
    cur = cursor
    while cur < shard_end:
        queue.put_nowait(cur)
        cur += step
    if queue.qsize() == 0:
        for _, c in clients:
            await c.aclose()
        return ShardResult(shard_index, fetched, skipped, errors, 0.0, backends)

    limiter = None if rate_per_worker is None else TokenBucketAsync(float(rate_per_worker))

    write_buffer: list = []
    completion_buffer: list = []  # (ts, "ok" | "404" | "err")
    completed_kind: dict = {}  # ts -> kind, for completions past the frontier
    counters = {"fetched": fetched, "skipped": skipped, "errors": errors}
    frontier = cursor - step  # last contiguous-completed bucket
    stop_event = asyncio.Event()

    def write_checkpoint(*, completed: bool = False) -> None:
        write_state(state_path, {
            "shard_index": shard_index,
            "interval": interval, "market": market,
            "next_time": format_utc(shard_end if completed else frontier + step),
            "fetched": counters["fetched"],
            "skipped": counters["skipped"],
            "errors": counters["errors"],
            "range_start": format_utc(shard_start),
            "range_end": format_utc(shard_end),
            **({"completed": True} if completed else {}),
        })

    def flush_once() -> None:
        # Snapshot buffers atomically — these statements have no awaits, so
        # asyncio cannot interleave a consumer between the snapshot and the
        # clear, and a consumer's two buffer appends always land together.
        nonlocal frontier
        to_write = write_buffer[:]
        write_buffer.clear()
        to_complete = completion_buffer[:]
        completion_buffer.clear()
        if not to_write and not to_complete:
            return

        # Make candles durable BEFORE counting them — otherwise an interrupt
        # between the count and the disk write would silently lose data.
        if to_write:
            with csv_path.open("a", newline="") as f:
                w = csv.DictWriter(f, fieldnames=CSV_FIELDS)
                for row in to_write:
                    w.writerow(row)

        # Defer counting until the frontier advances through each timestamp,
        # so the persisted counters match exactly what the frontier covers.
        # Without this, an interrupt would persist counts for ahead-of-frontier
        # completions; resume would re-fetch and double-count them.
        for ts, kind in to_complete:
            completed_kind[ts] = kind

        next_ts = frontier + step
        while next_ts in completed_kind:
            kind = completed_kind.pop(next_ts)
            if kind == "ok":
                counters["fetched"] += 1
            elif kind == "404":
                counters["skipped"] += 1
            else:
                counters["errors"] += 1
            frontier = next_ts
            next_ts += step

        write_checkpoint()

    async def flush_loop():
        while not stop_event.is_set():
            try:
                await asyncio.wait_for(stop_event.wait(), timeout=1.0)
                break
            except asyncio.TimeoutError:
                pass
            try:
                flush_once()
            except Exception:
                pass

    async def consumer(client):
        url = f"{base}/v1/candles"
        while True:
            try:
                ts = queue.get_nowait()
            except asyncio.QueueEmpty:
                return
            if limiter is not None:
                await limiter.acquire()
            params = {"market": market, "interval": interval, "time": format_utc(ts)}
            try:
                r = await client.get(url, params=params)
                if r.status_code == 200:
                    try:
                        candle = r.json()
                        # These two appends MUST be back-to-back with no await
                        # between them so the flusher cannot snapshot a CSV
                        # candle without its matching completion entry.
                        write_buffer.append(candle)
                        completion_buffer.append((ts, "ok"))
                    except Exception:
                        completion_buffer.append((ts, "err"))
                elif r.status_code == 404:
                    completion_buffer.append((ts, "404"))
                else:
                    completion_buffer.append((ts, "err"))
            except Exception:
                completion_buffer.append((ts, "err"))

    t0 = time.monotonic()
    flusher = asyncio.create_task(flush_loop())
    tasks = []
    for _, c in clients:
        for _ in range(workers_per_client):
            tasks.append(asyncio.create_task(consumer(c)))
    try:
        await asyncio.gather(*tasks)
    finally:
        try:
            flush_once()
        except Exception:
            pass
        stop_event.set()
        try:
            await flusher
        except Exception:
            pass
        for _, c in clients:
            try:
                await c.aclose()
            except Exception:
                pass

    write_checkpoint(completed=True)

    return ShardResult(shard_index, counters["fetched"], counters["skipped"],
                      counters["errors"], time.monotonic() - t0, backends)


def shard_entry(queue_mp, shard_index, api_key, market, interval, shard_start_iso,
                shard_end_iso, rate_per_worker, sticky_per_worker, workers_per_client,
                csv_path, state_path):
    base = f"{ENDPOINT_HOST}/{api_key}"
    try:
        result = asyncio.run(shard_worker(
            shard_index, base, market, interval,
            parse_utc(shard_start_iso), parse_utc(shard_end_iso),
            rate_per_worker, sticky_per_worker, workers_per_client,
            Path(csv_path), Path(state_path),
        ))
        queue_mp.put(("ok", shard_index, {
            "fetched": result.fetched,
            "skipped": result.skipped,
            "errors": result.errors,
            "seconds": result.seconds,
            "sticky_backends": result.sticky_backends,
        }))
    except Exception as exc:
        queue_mp.put(("err", shard_index, repr(exc)))


def merge_parts(part_paths, final_output):
    """Merge per-shard CSVs into one file ordered by candle open time `t`.

    Shards write rows concurrently so they arrive unsorted. We collect the
    combined rows in memory, deduplicate by `t` (the rare duplicate can occur
    when a process is killed between flushing the CSV and writing the new
    state — the row is durable but the state still asks for it on resume),
    and sort before writing the final file. A month of `1s` with ~30%
    materialized is roughly 800k rows.
    """
    by_t: dict = {}
    for p in part_paths:
        if not p.exists():
            continue
        with p.open(newline="") as src:
            for row in csv.DictReader(src):
                by_t[int(row["t"])] = row
    rows = [by_t[k] for k in sorted(by_t)]
    tmp = final_output.with_name(final_output.name + ".tmp")
    with tmp.open("w", newline="") as out:
        w = csv.DictWriter(out, fieldnames=CSV_FIELDS)
        w.writeheader()
        w.writerows(rows)
    tmp.replace(final_output)


def discover_api_key(env_var="DWELLIR_API_KEY", key_name="support_bot"):
    val = os.environ.get(env_var)
    if val:
        return val
    import subprocess
    try:
        out = subprocess.check_output(["dwellir", "keys", "list", "--toon"], text=True)
    except (FileNotFoundError, subprocess.CalledProcessError):
        raise SystemExit(f"set {env_var} or install/auth the `dwellir` CLI")
    for line in out.splitlines():
        if f",{key_name}," in line and ",true," in line:
            return line.split(",", 1)[0].strip()
    raise SystemExit(f"no enabled key '{key_name}' via dwellir CLI; pass --api-key or set {env_var}")


def main():
    parser = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("--api-key", default=None,
                        help="Dwellir API key UUID; defaults to $DWELLIR_API_KEY or `dwellir keys list`")
    parser.add_argument("--api-key-name", default="support_bot",
                        help="key name to look up via `dwellir keys list` if --api-key not given")
    parser.add_argument("--market", required=True)
    parser.add_argument("--interval", required=True, choices=sorted(INTERVAL_STEP))
    parser.add_argument("--start", required=True,
                        help=f"ISO-8601 UTC start (inclusive); archive floor is {ARCHIVE_FLOOR}")
    parser.add_argument("--end", required=True, help="ISO-8601 UTC end (exclusive)")
    parser.add_argument("--output", required=True, type=Path)
    parser.add_argument("--rate", type=int, default=5000,
                        help="aggregate request-rate cap (RPS); match your plan ceiling")
    parser.add_argument("--workers", type=int, default=8,
                        help="worker processes (Scale=8, Growth/Developer/Free=1)")
    parser.add_argument("--sticky-per-worker", type=int, default=4)
    parser.add_argument("--workers-per-client", type=int, default=128)
    args = parser.parse_args()

    api_key = args.api_key or discover_api_key(key_name=args.api_key_name)
    start = parse_utc(args.start)
    end = parse_utc(args.end)
    step = INTERVAL_STEP[args.interval]
    if end <= start:
        raise SystemExit("--end must be after --start")
    if end > datetime.now(tz=timezone.utc) - step:
        raise SystemExit(f"--end must be at most one bucket-step ({step}) before now")

    workers = max(1, args.workers)
    rate_per_worker = None if args.rate <= 0 else args.rate / workers
    shards = split_range(start, end, step, workers)

    parts_dir = args.output.with_suffix(args.output.suffix + ".parts")
    parts_dir.mkdir(parents=True, exist_ok=True)
    print(f"backfill: market={args.market} interval={args.interval} "
          f"buckets={int((end - start)/step)} shards={len(shards)} "
          f"rate={args.rate} workers={workers}", file=sys.stderr)

    ctx = mp.get_context("spawn")
    queue_mp = ctx.Queue()
    procs = []
    csv_paths = []
    for i, (s, e) in enumerate(shards):
        csv_path = parts_dir / f"shard-{i:03d}.csv"
        state_path = parts_dir / f"shard-{i:03d}.state.json"
        csv_paths.append(csv_path)
        p = ctx.Process(
            target=shard_entry,
            args=(queue_mp, i, api_key, args.market, args.interval,
                  format_utc(s), format_utc(e), rate_per_worker,
                  args.sticky_per_worker, args.workers_per_client,
                  str(csv_path), str(state_path)),
        )
        p.start()
        procs.append(p)

    t0 = time.monotonic()
    results = []
    failed = []
    for _ in range(len(procs)):
        tag, sid, payload = queue_mp.get()
        if tag == "ok":
            results.append((sid, payload))
            r = payload
            print(f"  shard {sid:>3} done: fetched={r['fetched']} skipped={r['skipped']} "
                  f"errors={r['errors']} {r['seconds']:.1f}s backends={r['sticky_backends']}",
                  file=sys.stderr)
        else:
            failed.append((sid, payload))
            print(f"  shard {sid:>3} ERROR: {payload}", file=sys.stderr)
    for p in procs:
        p.join(timeout=10.0)

    if failed:
        print(f"FAILED: {len(failed)} shard(s) errored — partial CSV in {parts_dir}",
              file=sys.stderr)
        sys.exit(1)

    total_fetched = sum(r["fetched"] for _, r in results)
    total_skipped = sum(r["skipped"] for _, r in results)
    total_errors = sum(r["errors"] for _, r in results)
    wall = time.monotonic() - t0
    useful = total_fetched + total_skipped

    merge_parts(csv_paths, args.output)
    print(f"merged {len(csv_paths)} shards -> {args.output}", file=sys.stderr)
    print(f"summary: fetched={total_fetched} skipped(404)={total_skipped} "
          f"errors={total_errors} wall={wall:.1f}s useful_rps={useful/max(wall,1e-9):.1f}",
          file=sys.stderr)


if __name__ == "__main__":
    main()