Hyperliquidの履歴データを取得する：APIの使い方

import requests

BASE_URL = "https://api.hyperliquid.xyz"

def fetch_candles(coin, interval, start_ms, end_ms):
    """
    coin      : 取引ペア。例: "BTC"
    interval  : 時間足。例: "1m", "5m", "1h", "1d"
    start_ms  : 開始時刻（ミリ秒タイムスタンプ）
    end_ms    : 終了時刻（ミリ秒タイムスタンプ）
    """
    payload = {
        "type": "candleSnapshot",
        "req": {
            "coin":      coin,
            "interval":  interval,
            "startTime": start_ms,
            "endTime":   end_ms
        }
    }

    resp = requests.post(f"{BASE_URL}/info", json=payload)
    resp.raise_for_status()
    return resp.json()

# 例：BTCの直近7日分の1時間足を取得
import time

end_ms   = int(time.time() * 1000)
start_ms = end_ms - 7 * 24 * 3600 * 1000

candles  = fetch_candles("BTC", "1h", start_ms, end_ms)

def fetch_candles_paginated(coin, interval, start_ms, end_ms, page_size_ms):
    """
    時間範囲を複数区間に分割し、順番に取得して結合する。

    page_size_ms：1回あたりの時間幅（ミリ秒）。intervalに応じて調整する。
    """
    all_candles = []
    cursor = start_ms

    while cursor < end_ms:
        batch_end = min(cursor + page_size_ms, end_ms)
        batch     = fetch_candles(coin, interval, cursor, batch_end)

        all_candles.extend(batch)

        if not batch:
            break

        # 次ページは最後のK線のタイムスタンプ + 1 から開始
        cursor = batch[-1]["t"] + 1

        # レート制限を避けるための待機
        time.sleep(0.2)

    return all_candles

def fetch_funding_history(coin, start_ms, end_ms):
    payload = {
        "type": "fundingHistory",
        "req": {
            "coin":      coin,
            "startTime": start_ms,
            "endTime":   end_ms
        }
    }

    resp = requests.post(f"{BASE_URL}/info", json=payload)
    resp.raise_for_status()
    return resp.json()

def fetch_user_fills(address, start_ms, end_ms):
    payload = {
        "type": "userFills",
        "req": {
            "user":      address,
            "startTime": start_ms,
            "endTime":   end_ms
        }
    }

    resp = requests.post(f"{BASE_URL}/info", json=payload)
    resp.raise_for_status()
    return resp.json()

def fetch_liquidations(start_ms, end_ms):
    payload = {
        "type": "liquidations",
        "req": {
            "startTime": start_ms,
            "endTime":   end_ms
        }
    }

    resp = requests.post(f"{BASE_URL}/info", json=payload)
    resp.raise_for_status()
    return resp.json()

import csv

def save_candles_to_csv(candles, filepath):
    if not candles:
        return

    fieldnames = ["t", "o", "h", "l", "c", "v", "n"]

    with open(filepath, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()

        for candle in candles:
            writer.writerow({k: candle.get(k, "") for k in fieldnames})

import sqlite3

def init_db(db_path):
    conn = sqlite3.connect(db_path)

    conn.execute("""
    CREATE TABLE IF NOT EXISTS candles (
        coin     TEXT,
        interval TEXT,
        t        INTEGER,
        o        REAL,
        h        REAL,
        l        REAL,
        c        REAL,
        v        REAL,
        n        INTEGER,
        PRIMARY KEY (coin, interval, t)
    )
    """)

    conn.commit()
    return conn

def insert_candles(conn, coin, interval, candles):
    rows = [
        (coin, interval, c["t"], c["o"], c["h"], c["l"], c["c"], c["v"], c["n"])
        for c in candles
    ]

    conn.executemany(
        "INSERT OR IGNORE INTO candles VALUES (?,?,?,?,?,?,?,?,?)",
        rows
    )

    conn.commit()