Backfill the Full Hyperliquid Candle 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.

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.
- Treat the data as sparse: if a bucket returns 404, 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 a CSV with columns: market, interval, bucket_start, open, high, low, close, volume, trades_count, vwap.
- Persist resume state to a sidecar file 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 state file.
- Write the state file atomically so a partial write does not break the next run.
- Show me how to run a short validation window first, then how to run a larger backfill for BTC `1m`.

{
  "market": "BTC",
  "interval": "1m",
  "bucket_start": "2026-03-30T23:00:00Z",
  "open": "66745",
  "high": "66746",
  "low": "66687",
  "close": "66687",
  "volume": "4.78586",
  "trades_count": 256,
  "vwap": "66733.634306059987321311"
}

import argparse
import csv
import json
from datetime import datetime, timedelta, timezone
from pathlib import Path

import requests


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


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 load_state(state_path: Path):
    if not state_path.exists():
        return None
    try:
        return json.loads(state_path.read_text())
    except json.JSONDecodeError:
        print(f"Ignoring malformed state file at {state_path}")
        return None


def save_state(state_path: Path, payload):
    temp_path = state_path.with_name(state_path.name + ".tmp")
    temp_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
    temp_path.replace(state_path)


def fetch_candle(session: requests.Session, api_key: str, market: str, interval: str, bucket_start: datetime):
    response = session.get(
        f"https://api-hyperliquid-index.n.dwellir.com/{api_key}/v1/candles",
        params={
            "market": market,
            "interval": interval,
            "time": format_utc(bucket_start),
        },
        timeout=30,
    )

    if response.status_code == 404:
        return None

    response.raise_for_status()
    return response.json()


def ensure_csv_header(output_path: Path):
    if output_path.exists() and output_path.stat().st_size > 0:
        return

    with output_path.open("w", newline="") as handle:
        writer = csv.DictWriter(
            handle,
            fieldnames=[
                "market",
                "interval",
                "bucket_start",
                "open",
                "high",
                "low",
                "close",
                "volume",
                "trades_count",
                "vwap",
            ],
        )
        writer.writeheader()


def read_last_bucket_start(output_path: Path):
    if not output_path.exists() or output_path.stat().st_size == 0:
        return None

    with output_path.open(newline="") as handle:
        rows = csv.DictReader(handle)
        last_row = None
        for row in rows:
            last_row = row

    if last_row is None:
        return None

    return last_row["bucket_start"]


def append_row(output_path: Path, candle):
    with output_path.open("a", newline="") as handle:
        writer = csv.DictWriter(
            handle,
            fieldnames=[
                "market",
                "interval",
                "bucket_start",
                "open",
                "high",
                "low",
                "close",
                "volume",
                "trades_count",
                "vwap",
            ],
        )
        writer.writerow(candle)


def backfill(api_key: str, market: str, interval: str, start: str, end: str, output_path: Path):
    if interval not in STEP_BY_INTERVAL:
        raise ValueError(f"unsupported interval: {interval}")

    start_dt = parse_utc(start)
    end_dt = parse_utc(end)
    step = STEP_BY_INTERVAL[interval]
    state_path = output_path.with_suffix(output_path.suffix + ".state.json")
    state = load_state(state_path)

    current = start_dt
    fetched = 0
    skipped = 0

    ensure_csv_header(output_path)
    last_bucket_start = read_last_bucket_start(output_path)

    if state:
        current = parse_utc(state["next_time"])
        fetched = state["fetched"]
        skipped = state["skipped"]
        print(f"Resuming from {format_utc(current)}")

    if last_bucket_start:
        last_written_time = parse_utc(last_bucket_start) + step
        if last_written_time > current:
            current = last_written_time
            print(f"Continuing after last written candle at {last_bucket_start}")

    with requests.Session() as session:
        while current <= end_dt:
            candle = fetch_candle(session, api_key, market, interval, current)

            if candle is None:
                skipped += 1
            else:
                append_row(output_path, candle)
                fetched += 1

            next_time = current + step
            save_state(
                state_path,
                {
                    "market": market,
                    "interval": interval,
                    "next_time": format_utc(next_time),
                    "fetched": fetched,
                    "skipped": skipped,
                    "output_path": str(output_path),
                },
            )

            processed = fetched + skipped
            if processed % 100 == 0 or current == end_dt:
                print(
                    f"processed={processed} fetched={fetched} skipped={skipped} current={format_utc(current)}"
                )

            current = next_time

    save_state(
        state_path,
        {
            "market": market,
            "interval": interval,
            "next_time": format_utc(current),
            "fetched": fetched,
            "skipped": skipped,
            "output_path": str(output_path),
            "completed": True,
        },
    )


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("api_key")
    parser.add_argument("market")
    parser.add_argument("interval", choices=sorted(STEP_BY_INTERVAL))
    parser.add_argument("start")
    parser.add_argument("end")
    parser.add_argument("output_path")
    args = parser.parse_args()

    backfill(
        api_key=args.api_key,
        market=args.market,
        interval=args.interval,
        start=args.start,
        end=args.end,
        output_path=Path(args.output_path),
    )


if __name__ == "__main__":
    main()