Kalshi Backtest & Historical Data: Order Books & Trades

import os
import requests

API = "https://api.aymardata.com/v1"
KEY = os.environ["AYMARDATA_API_KEY"]  # export AYMARDATA_API_KEY=ad_live_…
H = {"Authorization": f"Bearer {KEY}", "User-Agent": "aymardata-backtest/1.0"}

# Kalshi books/trades over REST are expanding; daily files are queryable now.
# End-to-end: list Kalshi files for a date, prefer trades, fall back to assets.

def list_files(source: str, date: str):
    r = requests.get(
        f"{API}/files",
        headers=H,
        params={"source": source, "date": date},
        timeout=60,
    )
    r.raise_for_status()
    return r.json()["data"]

DATE = "2026-10-24"
files = list_files("kalshi", DATE)
trades = [f for f in files if "/table=trades/" in f["path"]]
assets = [f for f in files if "/table=assets/" in f["path"]]
pick = (trades or assets)[0]
print(f"kalshi files={len(files)} trades={len(trades)} assets={len(assets)}")
print("fetching", pick["path"])
blob = requests.get(pick["url"], timeout=120)
blob.raise_for_status()
out = f"/tmp/kalshi-{DATE}.parquet"
open(out, "wb").write(blob.content)
print("saved", out, "bytes", len(blob.content))

try:
    import pyarrow.parquet as pq
    table = pq.read_table(out)
    print("columns", table.column_names[:12])
    print("rows", table.num_rows)
except Exception as e:
    print("parquet inspect skipped:", e)