Research notebook in Python

"""Research notebook: do hours with unusually large liquidations ("flushes") mark turning points?

Python 3.9+, standard library only (pandas optional, see the last cell). Run it as a script, or open it in Jupyter or
VS Code: every "# %%" line starts a notebook cell. Joins the hourly liquidation series (GET /liquidations/summary)
with 1h candles (GET /candles/{coin}) and compares the forward returns after flush hours with all hours.

    export BLOCKMETRIC_API_KEY=bm_live_...            # free key: 30 days of history; Pro or API Dev: 1 year
    export BLOCKMETRIC_BASE_URL=https://<domain>/api/v1
    python research.py

Options: COIN (BTC), DAYS (30), PERCENTILE (95), CSV (write the joined table, optional).
Descriptive statistics over a short window, not a trading signal.
"""
# %% Setup
import csv
import json
import os
import statistics
import urllib.error
import urllib.parse
import urllib.request

BASE = os.environ.get("BLOCKMETRIC_BASE_URL", "").rstrip("/")
KEY = os.environ.get("BLOCKMETRIC_API_KEY")
COIN = os.environ.get("COIN", "BTC")
DAYS = int(os.environ.get("DAYS", "30"))
PERCENTILE = float(os.environ.get("PERCENTILE", "95"))


def api(path, **params):
    query = urllib.parse.urlencode({k: v for k, v in params.items() if v not in (None, "")})
    headers = {"Authorization": f"Bearer {KEY}"} if KEY else {"X-Blockmetric-Access": "public"}
    try:
        with urllib.request.urlopen(urllib.request.Request(f"{BASE}{path}?{query}", headers=headers), timeout=60) as res:
            return json.load(res)
    except urllib.error.HTTPError as e:
        err = json.load(e).get("error", {})
        raise SystemExit(f"{e.code} {err.get('code')}: {err.get('message')}")


# %% Fetch: hourly liquidations and hourly candles of the same window
summary = api("/liquidations/summary", coin=COIN, hours=DAYS * 24)
series = summary["data"]["series"]  # [{t, longUsd, shortUsd}], starts at coveredSince
candles = api(f"/candles/{urllib.parse.quote(COIN)}", tf="1h", limit=5000, **{"from": series[0]["t"]})["data"]["candles"]
close = {c["t"]: c["close"] for c in candles}
for w in summary["meta"].get("warnings") or []:
    print("note:", w.get("message"))

# %% Join and compute forward returns after 1, 4 and 24 hours
HOURS = [1, 4, 24]
rows = []
for i, s in enumerate(series):
    if s["t"] not in close or s["longUsd"] is None or s["shortUsd"] is None:  # null = an hour that was not collected
        continue
    row = {"t": s["t"], "long_usd": s["longUsd"], "short_usd": s["shortUsd"], "close": close[s["t"]]}
    for h in HOURS:
        later = series[i + h]["t"] if i + h < len(series) else None
        row[f"ret_{h}h"] = (close[later] / close[s["t"]] - 1) if later in close else None
    rows.append(row)


def threshold(values, pct):
    v = sorted(values)
    return v[min(len(v) - 1, int(len(v) * pct / 100))] if v else 0


long_cut = threshold([r["long_usd"] for r in rows], PERCENTILE)
short_cut = threshold([r["short_usd"] for r in rows], PERCENTILE)

# %% Compare: mean return and share of hours that went up afterwards
groups = {
    "all hours": rows,
    f"long flush (>= ${long_cut / 1e6:.1f}M)": [r for r in rows if long_cut > 0 and r["long_usd"] >= long_cut],
    f"short flush (>= ${short_cut / 1e6:.1f}M)": [r for r in rows if short_cut > 0 and r["short_usd"] >= short_cut],
}
print(f"{COIN}, {len(rows)} hours since {rows[0]['t'][:10] if rows else '-'}, flush = top {100 - PERCENTILE:g} % of hours")
print(f"{'group':<30} {'hours':>5}" + "".join(f" {'mean ' + str(h) + 'h':>9} {'up ' + str(h) + 'h':>6}" for h in HOURS))
for name, g in groups.items():
    cells = []
    for h in HOURS:
        r = [x[f"ret_{h}h"] for x in g if x[f"ret_{h}h"] is not None]
        cells.append(f" {statistics.mean(r) * 100:>8.2f}% {sum(1 for x in r if x > 0) / len(r) * 100:>5.0f}%" if r else f" {'-':>9} {'-':>6}")
    print(f"{name:<30} {len(g):>5}" + "".join(cells))
print(summary["meta"]["attribution"]["text"])

# %% Optional: save the table, or continue in pandas
if os.environ.get("CSV"):
    with open(os.environ["CSV"], "w", newline="") as f:
        w = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
        w.writeheader()
        w.writerows(rows)
    print("wrote", os.environ["CSV"])
# import pandas as pd
# df = pd.DataFrame(rows).set_index("t")
# df[["long_usd", "short_usd"]].plot()