Did cohort positioning predict price? Backtest without lookahead

export VIKE_API_KEY=...
python hl-cohort-backtest.py             # real run: needs the key
python hl-cohort-backtest.py --dry-run   # prints the requests it would make, calls nothing

"""Does the smart cohort's BTC positioning say anything about the next hour's return? A point-in-time check.

Run:  VIKE_API_KEY=... python hl-cohort-backtest.py    (--dry-run lists the requests and makes none)
"""
import os
import sys
from datetime import datetime
from statistics import correlation, mean
from urllib.parse import urlencode

NAME = "hl-cohort-backtest"
BASE = "https://data.vike.io"
DRY = "--dry-run" in sys.argv
COHORT = "3x smart"
HOUR_MS = 3_600_000


def fail(why):
    sys.exit(f"{NAME}: {why}")


def call(path, method="GET", **params):
    url = BASE + path + ("?" + urlencode(params) if params else "")
    if DRY:
        print(method, url)
        return None
    key = os.environ.get("VIKE_API_KEY") or fail("set VIKE_API_KEY")
    import requests

    r = requests.request(method, url, headers={"X-API-Key": key}, timeout=60)
    if r.status_code != 200:
        fail(f"{method} {path} answered HTTP {r.status_code}: {r.text[:200]}")
    return r.json()


def main():
    # labelBasis=point_in_time: each row carries the label the wallet had THAT hour. The default
    # resolves labels at query time, which hands a backtest today's grading (lookahead).
    matrix = call(
        "/v1/hl-cohorts/symbols/BTC/cohort-metrics",
        axis="pnl", labelBasis="point_in_time", hours=200,
        start="2026-08-10T00:00:00Z", end="2026-08-17T00:00:00Z",
    )
    candles = call(
        "/v1/hyperliquid/ohlcv",
        symbol="BTC", interval="1h", limit=200,
        start="2026-08-10T00:00:00Z", end="2026-08-17T02:00:00Z",
    )
    if DRY:
        return
    if matrix["label_basis"] != "point_in_time":
        fail("labels are not point-in-time; a backtest on them would look ahead")

    ret = {c["ts"]: c["close"] / c["open"] - 1 for c in candles["candles"]}
    pairs = []  # (bias in hour H, return of hour H+1): positioning is read before the move it is compared to
    for m in matrix["metrics"]:
        if m["cohort"] != COHORT or m["bias"] is None:
            continue
        hour_ms = int(datetime.fromisoformat(m["ts"]).timestamp() * 1000)
        if hour_ms + HOUR_MS in ret:
            pairs.append((m["bias"], ret[hour_ms + HOUR_MS]))
    if len(pairs) < 10:
        fail(f"only {len(pairs)} usable hours for cohort {COHORT!r}")

    long_side = [r for b, r in pairs if b > 0]
    short_side = [r for b, r in pairs if b <= 0]
    print(f"{COHORT}, BTC, {len(pairs)} hours (labels: point_in_time)")
    print(f"  corr(bias, next-hour return) = {correlation([b for b, _ in pairs], [r for _, r in pairs]):+.3f}")
    if long_side:
        print(f"  next-hour return after net-long hours : {mean(long_side) * 1e4:+.1f} bp over {len(long_side)} hours")
    if short_side:
        print(f"  next-hour return after net-short hours: {mean(short_side) * 1e4:+.1f} bp over {len(short_side)} hours")


if __name__ == "__main__":
    try:
        main()
    except Exception as e:  # a shape the recipe does not expect fails by name, not by traceback
        fail(f"{type(e).__name__}: {e}")