Binance Futures Backtest: Historical Perp 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"}

def book(market: str, at: str):
    r = requests.get(
        f"{API}/book",
        headers=H,
        params={"source": "binance-perps", "market": market, "at": at, "depth": 10},
        timeout=30,
    )
    r.raise_for_status()
    return r.json()

def trades(market: str, start: str, end: str):
    r = requests.get(
        f"{API}/trades",
        headers=H,
        params={
            "source": "binance-perps",
            "market": market,
            "start": start,
            "end": end,
            "limit": 500,
        },
        timeout=30,
    )
    r.raise_for_status()
    return r.json()["data"]

MARKET = "BTCUSDT"  # USDⓈ-M perpetual id on AymarData
AT = "2026-10-23T10:00:00Z"
START, END = "2026-10-23T10:00:00Z", "2026-10-23T10:05:00Z"

snap = book(MARKET, AT)
tape = trades(MARKET, START, END)
bids, asks = snap.get("bids") or [], snap.get("asks") or []
mid = (bids[0]["price"] + asks[0]["price"]) / 2 if bids and asks else None

buys = [t for t in tape if t.get("side") == "buy"]
sells = [t for t in tape if t.get("side") == "sell"]
buy_notional = sum((t["price"] or 0) * (t["size"] or 0) for t in buys)
sell_notional = sum((t["price"] or 0) * (t["size"] or 0) for t in sells)

# Simple flow signal: buy-pressure → lean long next minute (research sketch)
signal = "long" if buy_notional > sell_notional * 1.05 else "flat"
print({
    "market": MARKET,
    "mid": mid,
    "spread": snap.get("spread"),
    "n_trades": len(tape),
    "buy_notional": round(buy_notional, 2),
    "sell_notional": round(sell_notional, 2),
    "signal": signal,
})