#!/usr/bin/env python3
"""Turns raw snapshots into docs/data/metrics.json and docs/data/charts.json.

Stdlib only. Every number on the site and in THESIS.md comes from a metric registered here, with its source.
Inference is labelled in the metric's `note`. Nothing is estimated silently.
"""
import csv, datetime, gzip, json, math, pathlib, random, re, statistics, sys, time
from collections import Counter, defaultdict

HERE = pathlib.Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))
from common import ROOT, load  # noqa: E402

BQ = ROOT / "raw" / "bigquery"
OUT = ROOT / "docs" / "data"
GH = "https://code.swiftbotlab.workers.dev/near-toll-audit/blob/main"
METRICS, CHARTS = {}, {}


def bq(name):
    path = next(BQ.glob(f"{name}*.csv.gz"))
    with gzip.open(path, "rt") as fh:
        return list(csv.DictReader(fh))


def bq_merged(name, recent, key):
    """Full-history snapshot, with the rolling recent query replacing any overlapping rows."""
    rows = bq(name)
    if not list(BQ.glob(f"{recent}*.csv.gz")):
        return rows
    fresh = bq(recent)
    if key == "wk":  # the first week of a rolling window is partial: keep the snapshot's copy of it
        weeks = sorted({r["wk"] for r in fresh})
        fresh = [r for r in fresh if r["wk"] != weeks[0]]
    seen = {r[key] for r in fresh}
    return [r for r in rows if r[key] not in seen] + fresh


def m(key, value, unit, label, source, note=""):
    if isinstance(value, float):
        value = round(value, 6 if abs(value) < 1 else 4 if abs(value) < 1000 else 2)
    METRICS[key] = {"value": value, "unit": unit, "label": label, "source": source, "note": note}
    return value


def d(s):
    return datetime.date.fromisoformat(s[:10])


def monday(day):
    return day - datetime.timedelta(days=day.weekday())


def from_ts(t):
    return datetime.datetime.fromtimestamp(t, datetime.timezone.utc).date()


def ols(y, xs):
    """OLS with intercept. Returns beta, R squared. Small systems, plain Gaussian elimination."""
    n, k = len(y), len(xs) + 1
    x = [[1.0] + [c[i] for c in xs] for i in range(n)]
    a = [[sum(x[r][i] * x[r][j] for r in range(n)) for j in range(k)] + [sum(x[r][i] * y[r] for r in range(n))] for i in range(k)]
    for i in range(k):
        p = max(range(i, k), key=lambda r: abs(a[r][i]))
        a[i], a[p] = a[p], a[i]
        if abs(a[i][i]) < 1e-12:
            return [0.0] * k, 0.0
        for r in range(k):
            if r != i:
                f = a[r][i] / a[i][i]
                a[r] = [u - f * v for u, v in zip(a[r], a[i])]
    beta = [a[i][k] / a[i][i] for i in range(k)]
    ybar = sum(y) / n
    sst = sum((v - ybar) ** 2 for v in y) or 1e-18
    sse = sum((y[r] - sum(b * c for b, c in zip(beta, x[r]))) ** 2 for r in range(n))
    return beta, 1 - sse / sst


def corr(a, b):
    if len(a) < 3:
        return 0.0
    ma, mb = sum(a) / len(a), sum(b) / len(b)
    sa = math.sqrt(sum((v - ma) ** 2 for v in a))
    sb = math.sqrt(sum((v - mb) ** 2 for v in b))
    if sa == 0 or sb == 0:
        return 0.0
    return sum((u - ma) * (v - mb) for u, v in zip(a, b)) / (sa * sb)


# ---------------------------------------------------------------- market
def market():
    cg = load("cg_near")
    price, mcap = cg["price"], cg["mcap"]
    m("price", price, "usd", "NEAR price", "CoinGecko")
    m("mcap", mcap, "usd", "Market cap", "CoinGecko")
    m("fdv", cg["fdv"], "usd", "Fully diluted value", "CoinGecko")
    m("mcap_rank", cg["rank"], "n", "Market cap rank", "CoinGecko")
    m("ath_price", cg["ath"], "usd", "All-time high price", "CoinGecko", cg["ath_date"][:10])
    pulled = ((load("_fetch_status") or {}).get("cg_near") or {}).get("pulled_at") or ""
    m("market_as_of", pulled[:10] or "n/a", "date", "Day of the market and fee pull (prices, DefiLlama, revenue page)", "raw/latest/_fetch_status")
    px = load("price_near_daily")
    series = [(from_ts(p["t"]), p["p"]) for p in px]
    year_ago = [p for day, p in series if day <= series[-1][0] - datetime.timedelta(days=365)]
    m("price_1y_change_pct", (price / year_ago[-1] - 1) * 100, "pct", "NEAR price change over one year", "DefiLlama coins")
    cb = load("coinbase_near") or []
    if cb:
        by_day = {from_ts(r["t"]): r["open"] for r in cb}  # the coins series stamps each price near 00:00 UTC, so compare with the daily open
        gaps = [abs(p / by_day[day] - 1) for day, p in series if day in by_day][-180:]
        m("price_crosscheck_median_gap_pct", statistics.median(gaps) * 100, "pct", "Median daily gap, coins API price vs Coinbase open", "DefiLlama coins vs Coinbase")
    CHARTS["price"] = [[day.isoformat(), round(p, 4)] for day, p in series if day >= datetime.date(2024, 1, 1)]
    return price, mcap


# ---------------------------------------------------------------- supply
def supply(price):
    rows = sorted(bq_merged("01_", "01b_", "block_date"), key=lambda r: r["block_date"])
    today = datetime.datetime.now(datetime.timezone.utc).date()
    full = [r for r in rows if d(r["block_date"]) < today and int(r["blocks"]) > 60000 or d(r["block_date"]) < datetime.date(2025, 1, 1)]
    full = [r for r in full if int(r["blocks"]) > 8000]
    # the last row of a snapshot can be a part day: drop it
    last_blocks = statistics.median(int(r["blocks"]) for r in full[-30:])
    if int(full[-1]["blocks"]) < last_blocks * 0.8:
        full = full[:-1]
    as_of = full[-1]["block_date"]
    t30 = full[-30:]
    mint30 = sum(float(r["minted_near"]) for r in t30)
    burn30 = sum(float(r["burned_near"]) for r in t30)
    sup = float(full[-1]["supply_near"])
    year = 365 / 30
    m("as_of", as_of, "date", "Last complete day of chain data", "BigQuery blocks")
    m("supply", sup, "near", "Total supply", "BigQuery blocks (block headers)")
    m("mint_30d_per_day", mint30 / 30, "near", "New NEAR per day, last 30 days", "BigQuery blocks")
    m("burn_30d_per_day", burn30 / 30, "near", "NEAR burned per day, last 30 days", "BigQuery blocks")
    gross = mint30 * year / sup * 100
    m("gross_issuance_pct", gross, "pct", "Gross issuance, annualised", "BigQuery blocks")
    m("burn_pct", burn30 * year / sup * 100, "pct", "Burn, annualised share of supply", "BigQuery blocks")
    m("net_issuance_pct", (mint30 - burn30) * year / sup * 100, "pct", "Net issuance before buybacks", "BigQuery blocks")
    m("burn_share_of_issuance_pct", burn30 / mint30 * 100, "pct", "Burn as a share of issuance", "BigQuery blocks")
    m("issuance_near_yr", mint30 * year, "near", "Annual issuance", "BigQuery blocks")
    m("issuance_usd_yr", mint30 * year * price, "usd", "Annual issuance at today's price", "BigQuery blocks + CoinGecko")
    m("burn_usd_yr", burn30 * year * price, "usd", "Annual burn at today's price", "BigQuery blocks + CoinGecko")
    m("mint_365d", sum(float(r["minted_near"]) for r in full[-365:]), "near", "NEAR minted in the last 365 days", "BigQuery blocks")
    m("burn_all_time", sum(float(r["burned_near"]) for r in full), "near", "NEAR burned since genesis", "BigQuery blocks")
    m("mint_all_time", sum(float(r["minted_near"]) for r in full), "near", "NEAR minted since genesis", "BigQuery blocks")
    before = [r for r in full if "2025-10-01" <= r["block_date"] < "2025-10-25"]
    after = [r for r in full if "2025-11-05" <= r["block_date"] < "2025-12-05"]
    m("mint_per_day_before_cut", statistics.mean(float(r["minted_near"]) for r in before), "near", "New NEAR per day before the 2.5% cap", "BigQuery blocks")
    m("mint_per_day_after_cut", statistics.mean(float(r["minted_near"]) for r in after), "near", "New NEAR per day after the 2.5% cap", "BigQuery blocks")
    since = [r for r in full if r["block_date"] >= "2026-02-23"]
    top = max(since, key=lambda r: float(r["burned_near"]))
    med = statistics.median(float(r["burned_near"]) for r in since)
    m("largest_daily_burn_since_buyback", float(top["burned_near"]), "near", "Largest single-day burn since the buyback began", "BigQuery blocks", f"on {top['block_date']}, against a median of {med:.0f}")
    # the April break in burn
    mar = [float(r["burned_near"]) for r in full if r["block_date"][:7] == "2026-03"]
    apr = [float(r["burned_near"]) for r in full if r["block_date"][:7] == "2026-04"]
    m("burn_march_per_day", statistics.mean(mar), "near", "Burn per day, March 2026", "BigQuery blocks")
    m("burn_april_per_day", statistics.mean(apr), "near", "Burn per day, April 2026", "BigQuery blocks")
    monthly = defaultdict(lambda: [0.0, 0.0, 0])
    for r in full:
        rec = monthly[r["block_date"][:7]]
        rec[0] += float(r["minted_near"])
        rec[1] += float(r["burned_near"])
        rec[2] += 1
    CHARTS["mint_burn_monthly"] = [{"m": k, "mint_per_day": round(v[0] / v[2]), "burn_per_day": round(v[1] / v[2], 1)} for k, v in sorted(monthly.items()) if k >= "2024-01"]
    cross = load("chain_supply_crosscheck") or []
    by_day = {r["block_date"]: float(r["supply_near"]) for r in rows}
    gaps = [abs(c["rpc_supply_near"] - by_day[c["date"]]) for c in cross if c["date"] in by_day]
    if gaps:
        m("supply_crosscheck_max_gap_near", max(gaps), "near", "Largest gap, BigQuery supply vs archival RPC header, five dates", "BigQuery vs archival RPC")
    circ = bq("02_")
    c0 = circ[-1]
    first = circ[0]
    CHARTS["locked_supply"] = [{"d": r["block_date"], "locked": round(float(r["total_near"]) - float(r["circulating_near"]))} for r in circ[::14]] + [{"d": c0["block_date"], "locked": round(float(c0["total_near"]) - float(c0["circulating_near"]))}]
    m("circulating_bq", float(c0["circulating_near"]), "near", "Circulating supply, NEAR public dataset", "BigQuery circulating_supply")
    m("locked_bq", float(c0["total_near"]) - float(c0["circulating_near"]), "near", "Not yet circulating, NEAR public dataset", "BigQuery circulating_supply")
    m("locked_bq_2023", float(first["total_near"]) - float(first["circulating_near"]), "near", "Not yet circulating on 2023-01-01", "BigQuery circulating_supply")
    nb = load("nearblocks_stats")
    m("circulating_nearblocks", int(nb["circulating_supply"]) / 1e24, "near", "Circulating supply, NearBlocks", "NearBlocks stats", "NearBlocks and the public dataset disagree; CoinGecko counts everything as circulating")
    return as_of, mint30, burn30, sup


# ---------------------------------------------------------------- activity
APP = {"tg", "sweat", "kaiching"}


def activity():
    tiers = bq("05_")
    months = defaultdict(lambda: defaultdict(float))
    for r in tiers:
        rec = months[r["m"][:7]]
        for f in ("signers", "txs", "signers_2d", "signers_engaged", "signers_engaged_nobot", "tx_intents", "signers_intents", "tx_top1", "tx_top10", "tx_top100"):
            rec[f] += float(r[f])
        if r["kind"] == "kaiching":
            rec["kaiching_signers"] += float(r["signers"])
        if r["kind"] == "tg":
            rec["tg_engaged"] += float(r["signers_engaged"])
        if r["kind"] in APP:
            rec["app_signers"] += float(r["signers"])
            rec["app_engaged"] += float(r["signers_engaged"])
            rec["app_txs"] += float(r["txs"])
        else:
            rec["self_signers"] += float(r["signers"])
            rec["self_engaged"] += float(r["signers_engaged_nobot"])
            rec["self_txs"] += float(r["txs"])
    keys = [k for k in sorted(months) if k < "2026-11"]
    CHARTS["signer_tiers"] = [{"m": k, "signers": int(months[k]["signers"] - months[k]["kaiching_signers"]), "repeat": int(months[k]["signers_2d"]),
                               "engaged": int(months[k]["signers_engaged"]), "self_engaged": int(months[k]["self_engaged"]),
                               "app_engaged": int(months[k]["app_engaged"]), "txs": int(months[k]["txs"]),
                               "top1_share": round(months[k]["tx_top1"] / months[k]["txs"] * 100, 1),
                               "top10_share": round(months[k]["tx_top10"] / months[k]["txs"] * 100, 1)} for k in keys]
    last, prev_year = keys[-1], "2025-10"
    real_last = months[last]["signers"] - months[last]["kaiching_signers"]
    m("signers_last", int(real_last), "n", f"Distinct signers, {last}, without kaiching accounts", "BigQuery transactions", "560,000 kaiching accounts each signed one DELETE_ACCOUNT in October")
    m("signers_last_raw", int(months[last]["signers"]), "n", f"Distinct signers, {last}, all accounts", "BigQuery transactions")
    m("signers_engaged_last", int(months[last]["signers_engaged"]), "n", f"Engaged signers, {last}", "BigQuery transactions", "signed on 3+ days and touched 2+ contracts")
    m("signers_engaged_self_last", int(months[last]["self_engaged"]), "n", f"Engaged self-custody signers, {last}", "BigQuery transactions", "excludes .tg, .sweat, kaiching namespaces and signers over 3,000 tx a month")
    m("signers_engaged_jul25", int(months["2025-07"]["signers_engaged"]), "n", "Engaged signers, 2025-07", "BigQuery transactions")
    m("signers_engaged_change_pct", (months[last]["signers_engaged"] / months["2025-07"]["signers_engaged"] - 1) * 100, "pct", "Change in engaged signers since 2025-07", "BigQuery transactions")
    m("tg_engaged_change_pct", (months[last]["tg_engaged"] / months["2025-07"]["tg_engaged"] - 1) * 100, "pct", "Change in engaged .tg signers since 2025-07", "BigQuery transactions")
    m("self_engaged_change_pct", (months[last]["self_engaged"] / months["2025-07"]["self_engaged"] - 1) * 100, "pct", "Change in engaged self-custody signers since 2025-07", "BigQuery transactions")
    m("self_engaged_aug26", int(months["2026-08"]["self_engaged"]), "n", "Engaged self-custody signers, 2026-08", "BigQuery transactions")
    m("self_engaged_rise_since_aug", months[last]["self_engaged"] / months["2026-08"]["self_engaged"], "x", "Rise in engaged self-custody signers since 2026-08", "BigQuery transactions", "cause not established")
    m("engaged_share_of_signers_pct", months[last]["signers_engaged"] / real_last * 100, "pct", f"Engaged share of signers, {last}", "BigQuery transactions")
    m("top10_tx_share_last", months[last]["tx_top10"] / months[last]["txs"] * 100, "pct", f"Top 10 signers' share of transactions, {last}", "BigQuery transactions")
    feb = months["2026-02"]
    m("signers_feb26", int(feb["signers"]), "n", "Distinct signers, 2026-02", "BigQuery transactions", "a one-month airdrop spike")
    m("engaged_feb26", int(feb["signers_engaged"]), "n", "Engaged signers, 2026-02", "BigQuery transactions")
    top = bq("06_")
    mar_signers = [r for r in top if r["m"].startswith("2026-03") and r["role"] == "signer"]
    leader = max(mar_signers, key=lambda r: int(r["txs"]))
    m("top_signer_mar26", leader["account"], "text", "Largest signer, March 2026", "BigQuery transactions")
    m("top_signer_share_mar26", int(leader["txs"]) / months["2026-03"]["txs"] * 100, "pct", "That signer's share of March 2026 transactions", "BigQuery transactions")
    exit_rows = bq("17_")
    daily_exit = [r for r in exit_rows if r["series"] == "signer_daily"]
    m("top_signer_last_active", daily_exit[-1]["block_date"], "date", "Last day that signer sent transactions, April 2026", "BigQuery transactions")
    m("kaiching_deletes_oct26", sum(int(r["txs"]) for r in exit_rows if r["series"] == "delete_account"), "n", "kaiching accounts deleted, 2-8 October 2026", "BigQuery transaction_actions")
    m("tx_mar26", int(months["2026-03"]["txs"]), "n", "Transactions, 2026-03", "BigQuery transactions")
    m("tx_apr26", int(months["2026-04"]["txs"]), "n", "Transactions, 2026-04", "BigQuery transactions")
    m("tx_drop_apr26_pct", (months["2026-04"]["txs"] / months["2026-03"]["txs"] - 1) * 100, "pct", "Change in transactions, March to April 2026", "BigQuery transactions")
    # NearBlocks long-run daily series, weekly mean
    nb = load("nearblocks_charts") or []
    weekly = defaultdict(lambda: [0, 0, 0])
    for r in nb:
        day = d(r["date"])
        if day < datetime.date(2024, 1, 1):
            continue
        w = weekly[monday(day).isoformat()]
        w[0] += int(r["txns"] or 0)
        w[1] += int(r["active_accounts"] or 0)
        w[2] += 1
    CHARTS["tx_weekly"] = [{"w": k, "tx": v[0] // v[2], "active": v[1] // v[2]} for k, v in sorted(weekly.items()) if v[2] == 7]
    # funders
    fund = bq("07_")
    out = {}
    for mon in ("2026-03", "2026-10"):
        rows = sorted([r for r in fund if r["m"].startswith(mon)], key=lambda r: -int(r["accounts_touched"]))
        total = sum(int(r["accounts_touched"]) for r in rows)
        signed = sum(int(r["accounts_that_signed"]) for r in rows)
        shares = [int(r["accounts_touched"]) / total for r in rows]
        out[mon] = {"accounts": total, "signed": signed, "top1": rows[0]["funder"], "top1_accounts": int(rows[0]["accounts_touched"]),
                    "top1_signed": int(rows[0]["accounts_that_signed"]), "top1_share": round(shares[0] * 100, 1),
                    "top10_share": round(sum(shares[:10]) * 100, 1), "effective_funders": round(1 / sum(s * s for s in shares), 1)}
    CHARTS["funders"] = out
    m("funder_top1_mar26", out["2026-03"]["top1"], "text", "Largest funder of accounts, March 2026", "BigQuery receipt_actions")
    m("funder_top1_accounts_mar26", out["2026-03"]["top1_accounts"], "n", "Accounts that funder touched", "BigQuery receipt_actions")
    m("funder_top1_signed_mar26", out["2026-03"]["top1_signed"], "n", "Of those, accounts that signed anything that month", "BigQuery receipt_actions")
    m("funder_top1_share_mar26", out["2026-03"]["top1_share"], "pct", "Its share of all accounts funded", "BigQuery receipt_actions")
    m("effective_funders_oct26", out["2026-10"]["effective_funders"], "n", "Effective number of funders, October 2026", "BigQuery receipt_actions", "inverse Herfindahl")
    # cohorts
    coh = bq("04_")
    size = defaultdict(lambda: defaultdict(int))
    active = defaultdict(lambda: defaultdict(lambda: defaultdict(int)))
    for r in coh:
        grp = "app" if r["kind"] in APP else "self"
        f, a = r["first_m"][:7], r["mth"][:7]
        active[grp][f][a] += int(r["signers"])
        if f == a:
            size[grp][f] += int(r["signers"])

    def add_months(s, k):
        y, mon = int(s[:4]), int(s[5:7]) + k
        y += (mon - 1) // 12
        mon = (mon - 1) % 12 + 1
        return f"{y}-{mon:02d}"
    ret = {}
    for grp in ("self", "app"):
        rows = []
        for f in sorted(size[grp]):
            if not "2024-01" <= f <= "2026-09":
                continue
            base = size[grp][f]
            rows.append({"cohort": f, "size": base, **{f"m{k}": (round(active[grp][f].get(add_months(f, k), 0) / base * 100, 1) if add_months(f, k) <= "2026-10" and base else None) for k in (1, 3, 6, 12)}})
        ret[grp] = rows

    def pooled(grp, k, lo, hi):
        num = den = 0
        for f, base in size[grp].items():
            if lo <= f <= hi and add_months(f, k) <= "2026-10":
                num += active[grp][f].get(add_months(f, k), 0)
                den += base
        return num / den * 100 if den else None
    named, hexa = defaultdict(lambda: defaultdict(int)), defaultdict(lambda: defaultdict(int))
    for r in coh:
        if r["kind"] == "named_near":
            named[r["first_m"][:7]][r["mth"][:7]] += int(r["signers"])
        elif r["kind"] in ("implicit", "evm_implicit"):
            hexa[r["first_m"][:7]][r["mth"][:7]] += int(r["signers"])

    def group_retention(table, k, lo, hi):
        num = sum(v.get(add_months(f, k), 0) for f, v in table.items() if lo <= f <= hi and add_months(f, k) <= "2026-10")
        den = sum(v.get(f, 0) for f, v in table.items() if lo <= f <= hi and add_months(f, k) <= "2026-10")
        return num / den * 100 if den else None
    ret["named"] = [{"cohort": f, "size": named[f].get(f, 0), **{f"m{k}": (round(named[f].get(add_months(f, k), 0) / named[f][f] * 100, 1) if add_months(f, k) <= "2026-10" and named[f].get(f) else None) for k in (1, 3, 6, 12)}} for f in sorted(named) if "2025-07" <= f <= "2026-09"]
    m("ret_named_m1", group_retention(named, 1, "2025-07", "2026-09"), "pct", "Named .near signers back one month later", "BigQuery transactions")
    m("ret_hex_m1", group_retention(hexa, 1, "2025-07", "2026-09"), "pct", "Hex accounts back one month later", "BigQuery transactions", "over 90% of self-custody cohorts are hex accounts, many of them disposable")
    CHARTS["retention"] = ret
    CHARTS["new_signers"] = [{"m": f, "app": size["app"].get(f, 0), "self": size["self"].get(f, 0)} for f in sorted(set(size["app"]) | set(size["self"])) if "2024-01" <= f <= "2026-10"]
    m("new_app_signers_peak", max(size["app"].get(f, 0) for f in size["app"] if f >= "2024-01"), "n", "New app-managed signers, peak month since 2024", "BigQuery transactions")
    m("new_app_signers_recent", statistics.mean(size["app"].get(f, 0) for f in ("2026-05", "2026-06", "2026-07", "2026-08")), "n", "New app-managed signers a month, May to Aug 2026", "BigQuery transactions")
    self_m1 = sorted(r["m1"] for r in ret["self"] if r["cohort"] >= "2025-07" and r["cohort"] <= "2026-09" and r["m1"] is not None)
    m("ret_self_m1_median", statistics.median(self_m1), "pct", "Self-custody signers back one month later, median cohort", "BigQuery transactions")
    m("ret_self_m3", pooled("self", 3, "2025-01", "2026-06"), "pct", "Self-custody signers still active 3 months on (2025-26 cohorts)", "BigQuery transactions")
    m("ret_self_m12", pooled("self", 12, "2025-01", "2025-10"), "pct", "Self-custody signers still active 12 months on", "BigQuery transactions")
    m("ret_app_m3", pooled("app", 3, "2025-01", "2026-06"), "pct", "App-managed signers still active 3 months on", "BigQuery transactions")
    m("ret_app_m12", pooled("app", 12, "2025-01", "2025-10"), "pct", "App-managed signers still active 12 months on", "BigQuery transactions")
    # tenure of active signers
    tenure = defaultdict(lambda: Counter())
    for r in coh:
        f, a = r["first_m"][:7], r["mth"][:7]
        if a < "2025-07" or a > "2026-10":
            continue
        age = (int(a[:4]) - int(f[:4])) * 12 + int(a[5:7]) - int(f[5:7])
        band = "new" if age == 0 else "1-6 mo" if age <= 6 else "7-24 mo" if age <= 24 else "over 2 yr"
        tenure[a][band] += int(r["signers"])
    CHARTS["tenure"] = [{"m": k, **v} for k, v in sorted(tenure.items())]
    # burn by contract
    burn = bq("08_")
    out = {}
    for mon in ("2026-03", "2026-10"):
        rows = [r for r in burn if r["m"].startswith(mon)]
        total = sum(float(r["tokens_burnt_near"]) for r in rows)
        out[mon] = [{"account": r["account"], "near": round(float(r["tokens_burnt_near"]), 1), "share": round(float(r["tokens_burnt_near"]) / total * 100, 1)} for r in rows[:14]]
    CHARTS["burn_by_contract"] = out
    kai = sum(r["share"] for r in out["2026-03"] if "kaiching" in r["account"])
    m("burn_kaiching_share_mar26", kai, "pct", "Kaiching contracts' share of top-60 gas spend, March 2026", "BigQuery execution_outcomes")
    intents_oct = next((r for r in burn if r["m"].startswith("2026-10") and r["account"] == "intents.near"), None)
    m("intents_gas_near_oct26", float(intents_oct["tokens_burnt_near"]), "near", "Gas spent by intents.near, October 2026", "BigQuery execution_outcomes")
    return months


# ---------------------------------------------------------------- intents
TOK = None


def tokens():
    global TOK
    if TOK is None:
        TOK = {r["token"]: r for r in csv.DictReader((HERE / "sql" / "token_prices.csv").open())}
    return TOK


def llama_series(obj):
    return {from_ts(t): v for t, v in obj.get("totalDataChart") or []}


def asset_class(token, meta):
    sym = (meta or {}).get("symbol", "").upper()
    if sym in ("USDC", "USDT", "USDT0", "DAI", "USD1", "USDG", "USDC.E", "XDAI", "FRAX", "USDE"):
        return "Stablecoins"
    if "BTC" in sym:
        return "Bitcoin"
    if sym in ("ETH", "WETH", "STETH", "WSTETH"):
        return "Ether"
    if sym in ("SOL", "WSOL"):
        return "Solana"
    if sym == "ZEC":
        return "Zcash"
    if sym in ("WNEAR", "NEAR"):
        return "NEAR"
    return "Other"


def official_series():
    """The official dashboard's own daily volume and fees and monthly revenue. DefiLlama is the fallback, and the cross-check."""
    page = load("official_revenue_page") or {}
    vol = {d(r["date_at"]): r["volume_usd"] for r in page.get("intentVolumeSeries") or []}
    fees = {d(r["date_at"]): r["fees_usd"] for r in page.get("totalFeesSeries") or []}
    rev = {r["period_month"][:7]: r for r in page.get("revenueSeries") or []}
    return vol, fees, rev


def last_full_months(as_of, n):
    day = d(as_of) + datetime.timedelta(days=1)
    y, mon = day.year, day.month
    out = []
    for _ in range(n):
        mon -= 1
        if mon == 0:
            y, mon = y - 1, 12
        out.append(f"{y}-{mon:02d}")
    return out[::-1]


def intents(price, mint30, as_of):
    fees = load("llama_intents_fees")
    rails = load("llama_rail_volumes")
    llama_vol = llama_series(rails["near-intents"])
    llama_fee = llama_series(fees["dailyFees"])
    llama_rev = llama_series(fees["dailyRevenue"])
    vol, gross, rev = official_series()
    if not vol or not gross or not rev:  # page layout changed: fall back to DefiLlama, which publishes the same volume since September
        vol, gross = llama_vol, llama_fee
        rev = {}
        for day, val in llama_rev.items():
            key = day.isoformat()[:7]
            rec = rev.setdefault(key, {"revenue_usd": 0.0, "revenue_near": 0.0})
            rec["revenue_usd"] += val
        m("revenue_source", "DefiLlama fallback", "text", "Source of revenue series", "pipeline")
    else:
        m("revenue_source", "revenue.near.org", "text", "Source of revenue series", "pipeline")
    end = d(as_of) - datetime.timedelta(days=1)  # the newest day is reported late
    days = sorted(x for x in vol if x <= end)
    m1 = last_full_months(as_of, 1)
    m3 = last_full_months(as_of, 3)

    def month_sum(series, keys):
        return sum(v for day, v in series.items() if day.isoformat()[:7] in keys)
    v1, f1, v3, f3 = month_sum(vol, m1), month_sum(gross, m1), month_sum(vol, m3), month_sum(gross, m3)
    r1 = sum(rev.get(k, {}).get("revenue_usd", 0) for k in m1)
    r3 = sum(rev.get(k, {}).get("revenue_usd", 0) for k in m3)
    rn1 = sum(rev.get(k, {}).get("revenue_near", 0) for k in m1)
    SRC = "revenue.near.org"
    m("window_month", m1[0], "text", "Last full month", SRC)
    m("vol_30d", v1, "usd", "Intents volume, last full month", SRC)
    m("vol_90d", v3, "usd", "Intents volume, last three full months", SRC)
    m("vol_annualised", v3 * 4, "usd", "Intents volume, annualised from three months", SRC)
    m("vol_all_time", sum(vol.values()), "usd", "Intents volume, all time", SRC)
    m("fees_30d", f1, "usd", "Gross fees, last full month", SRC)
    m("fees_90d", f3, "usd", "Gross fees, last three full months", SRC)
    m("rev_30d", r1, "usd", "Captured revenue, last full month", SRC)
    m("rev_90d", r3, "usd", "Captured revenue, last three full months", SRC)
    m("rev_annualised", r3 * 4, "usd", "Captured revenue, annualised from three months", SRC)
    m("rev_run_rate", r1 * 12, "usd", "Captured revenue, last month times twelve", SRC)
    m("rev_all_time", sum(r.get("revenue_usd", 0) for r in rev.values()), "usd", "Captured revenue, all time", SRC)
    m("rev_all_time_llama", sum(llama_rev.values()), "usd", "Captured revenue, all time, DefiLlama", "DefiLlama", "counts NEAR swept into wallets, so it is lumpy")
    m("gross_bps_90d", f3 / v3 * 1e4, "bps", "Gross fee rate, three months", SRC)
    m("net_bps_90d", r3 / v3 * 1e4, "bps", "Captured revenue rate, three months", SRC)
    m("gross_bps_30d", f1 / v1 * 1e4, "bps", "Gross fee rate, last full month", SRC)
    m("net_bps_30d", r1 / v1 * 1e4, "bps", "Captured revenue rate, last full month", SRC)
    page = load("official_revenue_page") or {}
    split = page.get("capture_split")
    capture = (split["pp_fe_d30"] + split["pp_b2b_d30"] + split["pp_qi_d30"]) * 100 if split else r1 / f1 * 100
    m("capture_30d_pct", capture, "pct", "Capture rate, 30 days", SRC)
    m("capture_all_time_pct", (split["pp_fe_all_time"] + split["pp_b2b_all_time"] + split["pp_qi_all_time"]) * 100 if split else None, "pct", "Capture rate, all time", SRC)
    monthly = defaultdict(lambda: [0.0, 0.0])
    for x in days:
        rec = monthly[x.isoformat()[:7]]
        rec[0] += vol.get(x, 0)
        rec[1] += gross.get(x, 0)
    table = []
    for k, v in sorted(monthly.items()):
        if k < "2025-05":
            continue
        r = rev.get(k, {}).get("revenue_usd", 0)
        table.append({"m": k, "vol": round(v[0]), "fees": round(v[1]), "rev": round(r), "gross_bps": round(v[1] / v[0] * 1e4, 1) if v[0] else None,
                      "net_bps": round(r / v[0] * 1e4, 2) if v[0] else None, "capture": round(r / v[1] * 100, 1) if v[1] else None})
    CHARTS["intents_monthly"] = table
    # how far the official series and DefiLlama agree
    if vol is not llama_vol:
        recent = sorted(vol)[-61:-1]
        same = sum(1 for x in recent if abs(llama_vol.get(x, 0) - vol[x]) < 2)
        m("official_llama_same_days", same, "n", "Of the last 60 days, days where DefiLlama equals the official volume to the dollar", "revenue.near.org vs DefiLlama")
        run = None
        for x in sorted(vol, reverse=True)[1:]:
            if abs(llama_vol.get(x, 0) - vol[x]) >= 2:
                break
            run = x
        m("official_llama_match_since", run.isoformat() if run else "never", "date", "Daily volume matches DefiLlama to the dollar since", "revenue.near.org vs DefiLlama", "before this date the two diverge, and revenue differs by month")
        m("official_llama_match_days_all", sum(1 for x in vol if abs(llama_vol.get(x, 0) - vol[x]) < 2), "n", "Days matching to the dollar, full history", "revenue.near.org vs DefiLlama")
        m("official_llama_days_all", len(vol), "n", "Days in the official series", "revenue.near.org")
    # weekly rail shares (DefiLlama for every rail, so the four are measured alike)
    weekly = defaultdict(lambda: Counter())
    for slug, obj in rails.items():
        for day, val in llama_series(obj).items():
            if day <= end:
                weekly[monday(day).isoformat()][obj["label"]] += val
    weeks = sorted(w for w in weekly if w >= "2025-01-06" and monday(end).isoformat() > w)
    shares = []
    for w in weeks:
        tot = sum(weekly[w].values())
        shares.append({"w": w, "total": round(tot), **{k: round(v / tot * 100, 1) for k, v in weekly[w].items()}})
    CHARTS["rail_share"] = shares

    def pooled_share(selected):
        ours = sum(weekly[w]["NEAR Intents"] for w in selected)
        return ours / (sum(sum(weekly[w].values()) for w in selected) or 1) * 100
    m("rail_share_13w", pooled_share(weeks[-13:]), "pct", "NEAR Intents share of native swap rails, 13 weeks, pooled", "DefiLlama", "rails: NEAR Intents, THORChain, Chainflip, Maya")
    m("rail_share_4w", pooled_share(weeks[-4:]), "pct", "Same, last four weeks", "DefiLlama")
    m("rail_share_jun26", pooled_share([w for w in weeks if w.startswith("2026-06")]), "pct", "Same, June 2026", "DefiLlama")
    leaders = []
    for r in shares:
        name = max(("NEAR Intents", "THORChain", "Chainflip", "Maya"), key=lambda k: r.get(k, 0))
        if not leaders or leaders[-1][0] != name:
            leaders.append([name, r["w"]])
    CHARTS["rail_leaders"] = leaders
    m("rail_leader_changes", len(leaders) - 1, "n", "Changes of weekly leader among rails since 2025-01", "DefiLlama")
    # on-chain reconciliation
    meta = tokens()
    px = load("token_prices_weekly") or {}
    flows = defaultdict(lambda: Counter())
    classes = Counter()
    chains = Counter()
    priced_ev = unpriced_ev = 0
    for r in bq_merged("09_", "09b_", "wk"):
        if r["account"] != "intents.near":
            continue
        info = meta.get(r["token"])
        if not info:
            unpriced_ev += int(r["events"])
            continue
        series = px.get(info["coingecko_id"])
        when = datetime.datetime.fromisoformat(r["wk"] + "T12:00:00+00:00").timestamp() + 3 * 86400
        p = min(series, key=lambda q: abs(q[0] - when))[1] if series else float(info["price"])
        priced_ev += int(r["events"])
        scale = p / 10 ** int(info["decimals"])
        g, delta = float(r["gross_raw"]) * scale, float(r["delta_raw"]) * scale
        wk = flows[r["wk"]]
        if r["cause"] == "ft_burn":
            wk["withdrawn"] += g
        elif r["cause"] == "ft_transfer":
            wk["in"] += (g + delta) / 2
            wk["out"] += (g - delta) / 2
        elif r["cause"] == "ft_mint":
            wk["in"] += g
        if r["wk"] >= "2026-08-03":
            klass = asset_class(r["token"], info)
            classes[klass] += g
            chains[info["blockchain"]] += g
    m("flow_price_coverage_pct", priced_ev / (priced_ev + unpriced_ev) * 100, "pct", "Share of token events we could price", "BigQuery ft_events + DefiLlama coins")
    wvol = Counter()
    for day, val in vol.items():
        wvol[monday(day).isoformat()] += val
    rec = []
    done_weeks = [w for w in sorted(flows) if w < monday(end).isoformat() and wvol[w] > 0]
    for w in done_weeks:
        f = flows[w]
        rec.append({"w": w, "reported": round(wvol[w]), "deposits": round(f["in"]), "withdrawn": round(f["withdrawn"]), "moved_out": round(f["out"]),
                    "ratio": round((f["in"] + f["withdrawn"] + f["out"]) / 2 / wvol[w], 3)})
    CHARTS["flow_reconciliation"] = rec
    last26 = rec[-26:]
    ratios = sorted(r["ratio"] for r in last26)
    m("flow_ratio_median", statistics.median(ratios), "x", "One-way on-chain flow as a share of reported volume, median of 26 weeks", "BigQuery ft_events vs revenue.near.org", "one-way flow = (deposits + withdrawals + transfers out) / 2; includes solver rebalancing")
    m("flow_ratio_low", ratios[0], "x", "Same ratio, lowest of 26 weeks", "BigQuery ft_events vs revenue.near.org")
    m("flow_ratio_high", ratios[-1], "x", "Same ratio, highest of 26 weeks", "BigQuery ft_events vs revenue.near.org")
    outlier = max(rec, key=lambda r: r["ratio"])
    m("flow_outlier_week", outlier["w"], "date", "Week with the largest flow ratio", "BigQuery ft_events", "a 47M SOL deposit and same-day withdrawal: raw flow is not automatically economic")
    m("flow_outlier_ratio", outlier["ratio"], "x", "Its ratio", "BigQuery ft_events")
    tot = sum(classes.values())
    CHARTS["flow_assets"] = [{"k": k, "share": round(v / tot * 100, 1)} for k, v in classes.most_common()]
    tot = sum(chains.values())
    CHARTS["flow_chains"] = [{"k": k, "share": round(v / tot * 100, 1)} for k, v in chains.most_common(12)]
    # destinations
    dest = bq("10_")
    CHARTS["destinations"] = [{"m": r["m"][:7], "dests": int(r["destinations"]), "withdrawals": int(r["withdrawals"]),
                               "top10": round(int(r["top10"]) / int(r["withdrawals"]) * 100, 1), "top100": round(int(r["top100"]) / int(r["withdrawals"]) * 100, 1)} for r in dest]
    lastd = dest[-1]
    m("dest_7d", int(lastd["destinations"]), "n", "Distinct withdrawal addresses in the first 7 days of October 2026", "BigQuery ft_events")
    m("dest_top10_count_share", int(lastd["top10"]) / int(lastd["withdrawals"]) * 100, "pct", "Top 10 addresses' share of withdrawal count", "BigQuery ft_events")
    size_rows = bq("13_")
    prev_rows = bq("13b_")
    buckets = [r for r in size_rows if r["kind"] == "bucket"]
    tops = {r["k"]: r for r in size_rows if r["kind"] == "top"}
    prev = {r["k"]: r for r in prev_rows if r["kind"] == "top"}
    total_usd = float(tops["all"]["usd"])
    prev_usd = float(prev["all"]["usd"])
    total_n = sum(int(r["n"]) for r in buckets)
    CHARTS["withdrawal_sizes"] = [{"k": r["k"][2:], "n_share": round(int(r["n"]) / total_n * 100, 1), "usd_share": round(float(r["usd"]) / total_usd * 100, 1)} for r in buckets]
    m("wd_week_usd", total_usd, "usd", "Priced withdrawals, week of 2026-10-19", "BigQuery ft_events")
    m("wd_top1_value_share", float(tops["top1"]["usd"]) / total_usd * 100, "pct", "Largest address's share of withdrawal value", "BigQuery ft_events")
    m("wd_top10_value_share", float(tops["top10"]["usd"]) / total_usd * 100, "pct", "Top 10 addresses' share of withdrawal value, week of 2026-10-19", "BigQuery ft_events")
    m("wd_top10_value_share_prev", float(prev["top10"]["usd"]) / prev_usd * 100, "pct", "Same, week of 2026-10-12", "BigQuery ft_events")
    m("wd_top100_value_share", float(tops["top100"]["usd"]) / total_usd * 100, "pct", "Top 100 addresses' share of withdrawal value", "BigQuery ft_events")
    floor = total_usd - float(tops["top100"]["usd"])
    floor_prev = prev_usd - float(prev["top100"]["usd"])
    reported = wvol.get("2026-10-19")
    reported_prev = wvol.get("2026-10-12")
    m("wd_floor_week_usd", floor, "usd", "Withdrawals outside the top 100 addresses, week of 2026-10-19", "BigQuery ft_events", "a rough lower bound for end-user flow: large addresses may be solvers, exchanges or real whales, and the tail may include bots")
    m("wd_floor_week_prev_usd", floor_prev, "usd", "Same, week of 2026-10-12", "BigQuery ft_events")
    m("wd_floor_annual_usd", floor * 52, "usd", "That lower bound, annualised", "BigQuery ft_events")
    if reported:
        m("wd_floor_share_of_reported_pct", floor / reported * 100, "pct", "That lower bound as a share of reported volume", "BigQuery ft_events vs revenue.near.org")
    if reported_prev:
        m("wd_floor_share_prev_pct", floor_prev / reported_prev * 100, "pct", "Same, week of 2026-10-12", "BigQuery ft_events vs revenue.near.org")
    if reported:
        m("wd_ex_top10_share_of_reported_pct", (total_usd - float(tops["top10"]["usd"])) / reported * 100, "pct", "Withdrawals outside the 10 largest addresses, as a share of reported volume, week of 2026-10-19", "BigQuery ft_events vs revenue.near.org", "an upper estimate of end-user flow: it leaves out the largest solver and exchange addresses but keeps smaller ones")
        m("wd_all_share_of_reported_pct", total_usd / reported * 100, "pct", "All priced withdrawals as a share of reported volume, week of 2026-10-19", "BigQuery ft_events vs revenue.near.org", "includes solver and exchange withdrawals")
    big = next(r for r in buckets if r["k"].startswith("e"))
    small = [r for r in buckets if r["k"][0] in "ab"]
    m("wd_big_count_share", int(big["n"]) / total_n * 100, "pct", "Withdrawals over $100k, share of count", "BigQuery ft_events")
    m("wd_big_value_share", float(big["usd"]) / total_usd * 100, "pct", "Withdrawals over $100k, share of value", "BigQuery ft_events")
    m("wd_small_count_share", sum(int(r["n"]) for r in small) / total_n * 100, "pct", "Withdrawals under $1k, share of count", "BigQuery ft_events")
    m("wd_small_value_share", sum(float(r["usd"]) for r in small) / total_usd * 100, "pct", "Withdrawals under $1k, share of value", "BigQuery ft_events")
    ret = defaultdict(dict)
    for r in bq("14_"):
        ret[r["m"][:7]][r["first_m"][:7]] = int(r["destinations"])
    table = []
    for mon in sorted(ret):
        tot = sum(ret[mon].values())
        table.append({"m": mon, "dests": tot, "returning": tot - ret[mon].get(mon, 0), "returning_share": round((tot - ret[mon].get(mon, 0)) / tot * 100, 1)})
    CHARTS["dest_returning"] = table
    m("dest_returning_share", statistics.mean(r["returning_share"] for r in table[-6:]), "pct", "Addresses seen in an earlier sampled week, 6-month mean", "BigQuery ft_events", "seven-day samples, so true repeat use is higher")
    # following $100, with one capture rate so the shares add up
    gas = METRICS["intents_gas_near_oct26"]["value"] * price
    gross_100 = f1 / v1 * 100
    captured_100 = gross_100 * capture / 100
    per100 = {"gross": gross_100, "captured": captured_100, "others": gross_100 - captured_100, "protocol_fee": 0.0001, "gas": gas / v1 * 100}
    CHARTS["per_100"] = {k: round(v, 5) for k, v in per100.items()}
    m("per100_gross", per100["gross"], "usd", "Fees on $100 of volume", SRC)
    m("per100_captured", per100["captured"], "usd", "Of which reaches NEAR-side wallets", SRC)
    m("per100_gas", per100["gas"], "usd", "Gas spent by intents.near per $100", "BigQuery execution_outcomes")
    return {"v30": v1, "f30": f1, "r30": r1, "v90": v3, "r90": r3, "f90": f3, "rn30": rn1, "vol": vol, "gross": gross, "rev": rev, "end": end, "month": m1[0], "llama_vol": llama_vol}


# ---------------------------------------------------------------- accrual
def accrual(price, mint30, burn30, sup, it):
    official = load("official_revenue_page") or {}
    wallets = load("chain_wallets") or {}
    control = load("chain_control") or {}
    hist = load("wallet_history") or []
    bal = {k: int(v["amount"]) / 1e24 for k, v in wallets.items()}
    held = bal.get("buybacks.multisignature.near", 0) + bal.get("fefundsadmin.sputnik-dao.near", 0) + bal.get("1csfundsadmin.sputnik-dao.near", 0)
    m("buyback_wallet_near", bal.get("buybacks.multisignature.near"), "near", "NEAR in the buyback multisig", "RPC view_account")
    m("fe_wallet_near", bal.get("fefundsadmin.sputnik-dao.near"), "near", "NEAR in the front-end fee DAO", "RPC view_account")
    m("revenue_wallets_near", held, "near", "NEAR held across the three revenue wallets", "RPC view_account")
    m("revenue_wallets_usd", held * price, "usd", "Value of those holdings", "RPC + CoinGecko")
    m("revenue_wallets_pct_supply", held / sup * 100, "pct", "Holdings as a share of supply", "RPC")
    m("treasury_near", bal.get("treasury.near"), "near", "NEAR in treasury.near", "RPC view_account")
    m("intents_fee_pips", control.get("intents_fee_pips"), "n", "On-chain protocol fee, in pips (1 pip = 0.0001%)", "RPC intents.near fee()")
    m("buyback_confirmations", control.get("buyback_confirmations"), "n", "Signatures needed to move buyback funds", "RPC get_num_confirmations")
    m("buyback_multisig_requests", control.get("buyback_request_nonce"), "n", "Requests ever raised on that multisig", "RPC get_request_nonce")
    m("venear_supply", control.get("venear_supply_near"), "near", "veNEAR supply", "RPC venear.dao")
    m("venear_accounts", control.get("venear_accounts"), "n", "veNEAR accounts", "RPC venear.dao")
    series = official.get("revenueSeries") or []
    if series:
        m("official_rev_month_near", it["rn30"], "near", f"Official captured revenue in NEAR, {it['month']}", "revenue.near.org")
        m("official_rev_all_time_usd", series[-1]["cumulative_revenue_usd"], "usd", "Official captured revenue, all time", "revenue.near.org")
        CHARTS["official_rev_monthly"] = [{"m": r["period_month"][:7], "near": round(r["revenue_near"]), "usd": round(r["revenue_usd"])} for r in series]
    inflows = sum(float(str(v).replace(",", "")) for _, v in official.get("wallets") or [])
    if inflows:
        m("revenue_wallet_inflows_near", inflows, "near", "Official all-time inflows to the three revenue wallets", "revenue.near.org")
        m("revenue_held_share_pct", held / inflows * 100, "pct", "Share of those inflows still held", "revenue.near.org vs RPC")
    split = official.get("capture_split")
    if split:
        for win, label in (("d30", "30 days"), ("all_time", "all time")):
            fe, b2b, qi = split[f"pp_fe_{win}"] * 100, split[f"pp_b2b_{win}"] * 100, split[f"pp_qi_{win}"] * 100
            m(f"capture_fe_{win}", fe, "pct", f"Gross fees NEAR captured through its own front end, {label}", "revenue.near.org")
            m(f"capture_b2b_{win}", b2b, "pct", f"Gross fees NEAR captured from partners, {label}", "revenue.near.org")
            m(f"capture_qi_{win}", qi, "pct", f"Gross fees NEAR captured from quote improvement, {label}", "revenue.near.org")
            m(f"capture_others_{win}", 100 - fe - b2b - qi, "pct", f"Gross fees kept by partners, wallets and solvers, {label}", "revenue.near.org")
        CHARTS["capture_split"] = [{"k": k, "d30": round(METRICS[f"{a}_d30"]["value"], 1), "all_time": round(METRICS[f"{a}_all_time"]["value"], 1)} for k, a in
                                   (("NEAR, own front end", "capture_fe"), ("NEAR, from partners", "capture_b2b"), ("NEAR, quote improvement", "capture_qi"), ("Partners, wallets and solvers (the rest)", "capture_others"))]
        CHARTS["capture_split"].insert(0, {"k": "NEAR, all three channels", "d30": round(METRICS["capture_30d_pct"]["value"], 1), "all_time": round(METRICS["capture_all_time_pct"]["value"], 1)})
    month = it["month"]
    mint_m = sum(float(r["minted_near"]) for r in bq_merged("01_", "01b_", "block_date") if r["block_date"][:7] == month)
    burn_m = sum(float(r["burned_near"]) for r in bq_merged("01_", "01b_", "block_date") if r["block_date"][:7] == month)
    buy_near = it["rn30"] or it["r30"] / price
    m("mint_month_near", mint_m, "near", f"NEAR minted, {month}", "BigQuery blocks")
    m("burn_month_near", burn_m, "near", f"NEAR burned, {month}", "BigQuery blocks")
    m("offset_30d_pct", (burn_m + buy_near) / mint_m * 100, "pct", f"Burn plus captured revenue as a share of issuance, {month}", "BigQuery + revenue.near.org", "assumes every captured dollar buys NEAR that is never sold")
    m("offset_90d_rev_pct", (it["r90"] * 4) / METRICS["issuance_usd_yr"]["value"] * 100, "pct", "Annualised captured revenue as a share of issuance value", "revenue.near.org + BigQuery")
    m("net_issuance_after_buyback_pct", (mint_m - burn_m - buy_near) * 12 / sup * 100, "pct", "Net issuance if all revenue buys NEAR and locks it", "BigQuery + revenue.near.org", "best case for holders")
    need = METRICS["issuance_usd_yr"]["value"] / (METRICS["net_bps_30d"]["value"] / 1e4)
    m("breakeven_volume_yr", need, "usd", "Annual volume at which captured revenue equals issuance", "derived", "at today's price and last month's take")
    m("breakeven_multiple", need / (it["v30"] * 12), "x", "Multiple of last month's volume needed for that", "derived")
    CHARTS["flows_near_per_day"] = {"issuance": round(mint_m / 31), "burn": round(burn_m / 31), "revenue_equiv": round(buy_near / 31), "days": 31}
    moved = bq("18_")
    total_moved = sum(float(r["near"]) for r in moved)
    m("treasury_moved_near", total_moved, "near", "NEAR moved out of treasury.near in April and May 2026", "BigQuery receipt_actions")
    m("treasury_moved_usd", total_moved * price, "usd", "Its value at today's price", "BigQuery + CoinGecko")
    m("treasury_moved_multiple", total_moved / held, "x", "That amount over the NEAR held in the revenue wallets", "BigQuery + RPC")
    CHARTS["treasury_moves"] = [{"d": r["block_date"], "to": r["receiver"][:12] + "..." if len(r["receiver"]) > 40 else r["receiver"], "near": round(float(r["near"]))} for r in moved if float(r["near"]) > 1000]
    by_acct = defaultdict(list)
    for r in hist:
        by_acct[r["account"]].append([r["date"], round(r["near"], 1)])
    CHARTS["wallet_history"] = dict(by_acct)
    bb = by_acct.get("buybacks.multisignature.near") or []
    drops = sum(1 for a, b in zip(bb, bb[1:]) if b[1] < a[1] - 1000)
    m("buyback_weekly_declines", drops, "n", "Weeks in which the buyback wallet's balance fell by more than 1,000 NEAR", "archival RPC weekly balances")
    roles = {}
    for dao in ("fefundsadmin.sputnik-dao.near", "1csfundsadmin.sputnik-dao.near"):
        info = control.get(dao) or {}
        admin = next((r for r in info.get("roles", []) if r["name"] == "Admin"), {})
        members = admin.get("members") or []
        roles[dao] = {"admins": members if len(members) < 8 else members[:8], "proposals": info.get("last_proposal_id")}
    CHARTS["wallet_control"] = roles


# ---------------------------------------------------------------- staking
def staking(price, sup):
    v = load("chain_validators")
    rows = sorted(v["validators"], key=lambda r: -int(r["stake"]))
    stake = [int(r["stake"]) / 1e24 for r in rows]
    total = sum(stake)
    m("stake_total", total, "near", "NEAR staked with active validators", "RPC validators")
    m("stake_ratio_pct", total / sup * 100, "pct", "Staked share of supply", "RPC validators")
    m("validators", len(rows), "n", "Active validators", "RPC validators")
    run, nak13, nak12 = 0.0, None, None
    for i, s in enumerate(stake, 1):
        run += s
        if nak13 is None and run > total / 3:
            nak13 = i
        if nak12 is None and run > total / 2:
            nak12 = i
    m("nakamoto_third", nak13, "n", "Validators needed for one third of stake", "RPC validators", "accounts, not grouped by operator")
    m("nakamoto_half", nak12, "n", "Validators needed for half of stake", "RPC validators")
    m("top10_stake_pct", sum(stake[:10]) / total * 100, "pct", "Top 10 validators' share of stake", "RPC validators")
    m("smallest_stake", stake[-1], "near", "Smallest active stake", "RPC validators")
    m("median_stake", statistics.median(stake), "near", "Median active stake", "RPC validators")
    fees = [(int(r["stake"]) / 1e24, r["fee"]["numerator"] / r["fee"]["denominator"], r["account"]) for r in rows if r.get("fee") and r["fee"].get("denominator")]
    # pools at 100% commission keep every reward: that is self-stake by large holders, not a price charged to delegators
    self_pools = [f for f in fees if f[1] >= 0.5]
    open_pools = [f for f in fees if f[1] < 0.5]
    wavg = sum(s * f for s, f, _ in open_pools) / sum(s for s, _, _ in open_pools)
    m("commission_wavg_pct", wavg * 100, "pct", "Stake-weighted commission, pools open to delegators", "RPC get_reward_fee_fraction", "excludes pools at 50% or more")
    m("self_stake_pools", len(self_pools), "n", "Pools at 50%+ commission", "RPC get_reward_fee_fraction")
    m("self_stake_near", sum(s for s, _, _ in self_pools), "near", "Stake in those pools", "RPC get_reward_fee_fraction", "mostly 100% commission: large holders staking with themselves")
    m("self_stake_share_pct", sum(s for s, _, _ in self_pools) / total * 100, "pct", "Their share of all stake", "RPC get_reward_fee_fraction")
    issuance = METRICS["issuance_near_yr"]["value"]
    to_stakers = issuance * 0.9
    gross = to_stakers / total * 100
    m("staking_yield_gross_pct", gross, "pct", "Gross staking yield", "derived from headers and stake", "90% of issuance over stake; 10% goes to treasury.near")
    net = gross * (1 - wavg)
    m("staking_yield_net_pct", net, "pct", "Staking yield after typical commission", "derived")
    m("staking_real_yield_pct", net - METRICS["net_issuance_pct"]["value"], "pct", "Yield after commission minus net issuance", "derived", "what a staker gains against supply growth")
    m("treasury_share_of_issuance_pct", 10.0, "pct", "Treasury share of issuance", "nearcore reward_calculator.rs (hardcoded 1/10 on mainnet); RPC field reads 0", "the protocol_reward_rate RPC field is a display artefact")
    m("non_staker_dilution_pct", METRICS["net_issuance_pct"]["value"], "pct", "Yearly dilution of an unstaked holder", "BigQuery blocks")
    gov = load("chain_governance") or []
    ramp = next((g for g in gov if "1.6%" in (g.get("title") or "")), None)
    if ramp:
        votes = ramp.get("votes") or []
        yes = int(votes[0]["total_venear"]) / 1e24 if votes else 0
        total = int((ramp.get("total_votes") or {}).get("total_venear", 0)) / 1e24
        m("issuance_vote_status", ramp.get("status"), "text", "Status of the vote to cut issuance to 1.6%", "RPC vote.dao")
        m("issuance_vote_for_pct", yes / total * 100 if total else None, "pct", "Share of votes in favour", "RPC vote.dao")
        m("issuance_vote_voters", (ramp.get("total_votes") or {}).get("total_votes"), "n", "Accounts that voted", "RPC vote.dao")
    proto = load("chain_protocol") or {}
    m("protocol_version", proto.get("protocol_version"), "n", "Live protocol version", "RPC EXPERIMENTAL_protocol_config")
    m("gas_rebate_removed", "yes" if (proto.get("protocol_version") or 0) >= 87 else "no", "text", "30% contract gas rebate removed (protocol 87)", "RPC plus nearcore CHANGELOG")
    hist = bq("15_")
    CHARTS["stake_history"] = [{"d": r["block_date"], "near": round(float(r["staked_near"])), "accounts": int(r["accounts"])} for r in hist]
    etf = load("etf_nrr") or {}
    if etf.get("near_held") and etf["near_held"] < 200e6:
        m("etf_near", etf["near_held"], "near", "NEAR held by the Bitwise ETF", "nrretf.com")
        m("etf_pct_supply", etf["near_held"] / sup * 100, "pct", "ETF share of supply", "nrretf.com")
        m("etf_to_buyback_ratio", etf["near_held"] / METRICS["revenue_wallets_near"]["value"], "x", "ETF holdings over NEAR held in the revenue wallets", "nrretf.com + RPC")
        m("absorbed_pct_of_year_mint", (etf["near_held"] + METRICS["revenue_wallets_near"]["value"]) / METRICS["mint_365d"]["value"] * 100, "pct", "ETF plus revenue wallets as a share of a year's new NEAR", "nrretf.com + RPC + BigQuery", "an equivalence: holdings were not necessarily bought from new supply")
    top = bq("11_")
    tot = sum(float(r["total"]) for r in top)
    m("top10_holders_pct", sum(float(r["total"]) for r in top[:10]) / sup * 100, "pct", "Top 10 balances, share of supply", "BigQuery ft_balances_daily", "includes exchanges, wrap.near and liquid staking pools, so not ten owners")
    m("top100_holders_pct", sum(float(r["total"]) for r in top[:100]) / sup * 100, "pct", "Top 100 balances, share of supply", "BigQuery ft_balances_daily")
    return wavg


# ---------------------------------------------------------------- peers
def peers(mcap):
    caps = {x["id"]: x for x in load("cg_markets")}
    fees = {x["name"]: x for x in load("llama_peer_chain_fees")["chains"]}
    tvl = {x["name"]: x for x in load("llama_chain_tvl")}
    stable = {x["name"]: x["usd"] for x in load("llama_stable_chains")}
    names = [("NEAR", "near", "Near"), ("Solana", "solana", "Solana"), ("Avalanche", "avalanche-2", "Avalanche"), ("Sui", "sui", "Sui"), ("Aptos", "aptos", "Aptos"),
             ("Arbitrum", "arbitrum", "Arbitrum"), ("ICP", "internet-computer", "ICP"), ("Injective", "injective-protocol", "Injective")]
    rows = []
    for label, cid, key in names:
        cap = (caps.get(cid) or {}).get("market_cap")
        fee = (fees.get(key) or {}).get("total30d")
        chg = (caps.get(cid) or {}).get("price_change_percentage_1y_in_currency")
        rows.append({"chain": label, "mcap": cap, "fdv": (caps.get(cid) or {}).get("fully_diluted_valuation"), "fees_30d": fee, "tvl": (tvl.get(key) or {}).get("tvl"), "stable": stable.get(key),
                     "mcap_to_fees": round(cap / (fee * 12)) if cap and fee else None, "chg_1y": round(chg, 1) if chg is not None else None})
    CHARTS["l1_peers"] = rows
    near_row = rows[0]
    others = [r for r in rows[1:] if r["mcap"]]
    m("l1_peer_median_mcap", statistics.median(r["mcap"] for r in others), "usd", "Median market cap of seven peer chains", "CoinGecko")
    m("l1_peer_median_fdv", statistics.median(r["fdv"] for r in others if r["fdv"]), "usd", "Median fully diluted value of seven peer chains", "CoinGecko", "NEAR is fully circulating, so FDV is the like-for-like basis")
    peer_chg = [r["chg_1y"] for r in others if r["chg_1y"] is not None]
    if peer_chg and near_row["chg_1y"] is not None:
        m("near_chg_1y_pct", near_row["chg_1y"], "pct", "NEAR price change, one year", "CoinGecko")
        m("peer_chg_1y_median_pct", statistics.median(peer_chg), "pct", "Median one-year change of seven peer chains", "CoinGecko")
    mcap_with = near_row["mcap"] / ((near_row["fees_30d"] + METRICS["fees_30d"]["value"]) * 12)
    m("near_mcap_to_all_fees", mcap_with, "x", "NEAR market cap over L1 fees plus Intents gross fees, annualised", "CoinGecko + DefiLlama", "counts fees NEAR does not keep")
    m("near_mcap_to_kept_fees", near_row["mcap"] / (near_row["fees_30d"] * 12 + METRICS["rev_run_rate"]["value"]), "x", "NEAR market cap over L1 fees plus captured revenue, annualised", "CoinGecko + DefiLlama + revenue.near.org")
    m("near_kept_annual", near_row["fees_30d"] * 12 + METRICS["rev_run_rate"]["value"], "usd", "What NEAR keeps a year: L1 fees plus captured Intents revenue, last 30 days times twelve", "DefiLlama + revenue.near.org")
    m("peer_mcap_to_fees_median", statistics.median(r["mcap_to_fees"] for r in others if r["mcap_to_fees"]), "x", "Median market cap to fees, seven peer chains", "CoinGecko + DefiLlama")
    m("near_chain_fees_30d", near_row["fees_30d"], "usd", "NEAR L1 fees, 30 days", "DefiLlama")
    m("near_stable_float", near_row["stable"], "usd", "Stablecoins on NEAR", "DefiLlama")
    m("near_tvl", near_row["tvl"], "usd", "DeFi TVL on NEAR", "DefiLlama")
    m("near_stable_rank", 1 + sum(1 for k, v in stable.items() if (v or 0) > (near_row["stable"] or 0)), "n", "NEAR's rank by stablecoin float", "DefiLlama")
    series = load("llama_stable_near")
    CHARTS["stable_near"] = [{"d": from_ts(r["t"]).isoformat(), "usd": round(r["usd"] or 0)} for r in series if from_ts(r["t"]) >= datetime.date(2024, 1, 1) and from_ts(r["t"]).weekday() == 0]
    comps = load("comps_holder_revenue") or []
    table = []
    for r in comps:
        ann = r["holders_rev_30d"] * 12
        table.append({"symbol": r["symbol"], "mcap": r["mcap"], "ann_holder_rev": round(ann), "multiple": round(r["mcap"] / ann, 1) if ann else None,
                      "multiple_1y": round(r["mcap"] / r["holders_rev_1y"], 1) if r["holders_rev_1y"] else None})
    CHARTS["comps"] = table
    near = next(t for t in table if t["symbol"] == "NEAR")
    run = METRICS["rev_run_rate"]["value"]
    near["ann_holder_rev"] = round(run)
    near["multiple"] = round(near["mcap"] / run, 1)
    others = sorted(t["multiple"] for t in table if t["symbol"] != "NEAR" and t["multiple"])
    m("comps_median_multiple", statistics.median(others), "x", "Median market cap / annualised holder revenue, ten comps", "DefiLlama + CoinGecko")
    m("near_multiple", near["multiple"], "x", "NEAR market cap / captured revenue, last month times twelve", "CoinGecko + revenue.near.org", "whole-chain market cap against app tokens")
    m("near_multiple_ex_chain", (near["mcap"] - METRICS["l1_peer_median_fdv"]["value"]) / run, "x", "Same, after subtracting a peer-median chain value", "CoinGecko + revenue.near.org")
    rail_fees = load("llama_rail_fees") or {}
    rails = load("llama_rail_volumes")
    cg = {x["id"]: x for x in load("cg_markets")}
    rows = []
    for label, slug, fee_slug, coin in (("NEAR Intents", "near-intents", "near-intents", "near"), ("THORChain", "thorchain-dex", None, "thorchain"), ("Chainflip", "chainflip-amm", "chainflip", "chainflip"), ("Maya", "maya-protocol", None, None)):
        v30 = rails[slug].get("total30d")
        fee = (rail_fees.get(fee_slug) or {}).get("total30d") if fee_slug else None
        cap = (cg.get(coin) or {}).get("market_cap") if coin else None
        rows.append({"rail": label, "vol_30d": v30, "fees_30d": fee, "take_bps": round(fee / v30 * 1e4, 1) if fee and v30 else None, "mcap": cap,
                     "mcap_per_annual_volume": round(cap / (v30 * 12), 4) if cap and v30 else None})
    CHARTS["rail_table"] = rows
    n, t, c = rows[0], rows[1], rows[2]
    m("mcap_per_volume_near", n["mcap_per_annual_volume"], "x", "NEAR market cap per dollar of annual rail volume", "CoinGecko + DefiLlama")
    m("mcap_per_volume_rune", t["mcap_per_annual_volume"], "x", "RUNE, same measure", "CoinGecko + DefiLlama")
    m("mcap_per_volume_flip", c["mcap_per_annual_volume"], "x", "FLIP, same measure", "CoinGecko + DefiLlama")
    dex = load("llama_dex_market") or {}
    m("spot_dex_30d", dex.get("total30d"), "usd", "All DEX volume, 30 days", "DefiLlama")
    gh = load("github_activity") or {}
    rows = [{k: v for k, v in r.items() if k != "weekly"} for r in gh.values() if "error" not in r]
    CHARTS["github"] = rows
    by_chain = defaultdict(lambda: [0, 0])
    for r in rows:
        by_chain[r["chain"]][0] += r.get("commits_13w") or 0
        by_chain[r["chain"]][1] += r.get("active_authors_90d") or 0
    CHARTS["github_chain"] = [{"chain": k, "commits_13w": v[0], "authors_90d": v[1]} for k, v in sorted(by_chain.items(), key=lambda t: -t[1][0])]
    if "near/nearcore" in gh:
        m("nearcore_commits_13w", gh["near/nearcore"].get("commits_13w"), "n", "nearcore commits, 13 weeks", "GitHub")
        m("nearcore_authors_90d", gh["near/nearcore"].get("active_authors_90d"), "n", "nearcore commit authors, 90 days", "GitHub")
    if "near/intents" in gh:
        m("intents_repo_authors_90d", gh["near/intents"].get("active_authors_90d"), "n", "near/intents commit authors, 90 days", "GitHub")


# ---------------------------------------------------------------- swap-level data (Explorer API aggregates)
EX = ROOT / "raw" / "explorer"


def ex(name):
    path = EX / f"{name}.csv.gz"
    if not path.exists():
        return []
    with gzip.open(path, "rt") as fh:
        return list(csv.DictReader(fh))


NEAR_SIDE = {"buybacks.multisignature.near", "fefundsadmin.sputnik-dao.near", "1csfundsadmin.sputnik-dao.near"}


def fee_owner(name):
    if name in NEAR_SIDE:
        return "NEAR wallets"
    if "swapkit" in name or name in ("thorswap.near",):
        return "Wallets via SwapKit"
    if name == "(other)":
        return "Other partners"
    return "Other partners"


def explorer(it):
    daily = ex("panel_daily")
    if not daily:
        m("ex_days", 0, "n", "Swap-level sample days", "Intents Explorer API")
        return
    vol = it["vol"]
    days = [r["day"] for r in daily]
    m("ex_days", len(days), "n", "Swap-level sample days", "Intents Explorer API")
    m("ex_first_day", days[0], "date", "First sample day", "Intents Explorer API")
    m("ex_last_day", days[-1], "date", "Last sample day", "Intents Explorer API")
    cut = (d(days[-1]) - datetime.timedelta(days=91)).isoformat()
    recent = [r for r in daily if r["day"] >= cut]
    rdays = {r["day"] for r in recent}
    usd = sum(float(r["usd_in"]) for r in recent)
    fee = sum(float(r["app_fee_usd"]) for r in recent)
    near_fee = sum(float(r["near_wallet_fee_usd"]) for r in recent)
    swaps = sum(int(r["swaps"]) for r in recent)
    official = sum(vol.get(d(x), 0) for x in rdays)
    m("ex_swaps_sample", sum(int(r["swaps"]) for r in daily), "n", "Successful swaps in the sample", "Intents Explorer API")
    m("ex_recent_days", len(recent), "n", "Sample days in the last 13 weeks", "Intents Explorer API")
    m("ex_coverage_pct", usd / official * 100 if official else None, "pct", "Explorer swap volume as a share of reported volume, same days", "Intents Explorer API vs DefiLlama")
    official_fees = sum(it["gross"].get(d(x), 0) for x in rdays)
    m("ex_fee_coverage_pct", fee / official_fees * 100 if official_fees else None, "pct", "App fees on Explorer swaps as a share of all reported fees, same days", "Intents Explorer API vs DefiLlama")
    rest_vol = max(official - usd, 0)
    rest_fee = max(official_fees - fee, 0)
    m("ex_rest_bps", rest_fee / rest_vol * 1e4 if rest_vol else None, "bps", "Implied fee on volume outside the Explorer, an upper bound", "derived", "reported fees minus Explorer app fees, over the remaining volume; 1Click's unlisted share also sits in this remainder")
    m("ex_avg_swap_usd", usd / swaps, "usd", "Average swap", "Intents Explorer API")
    m("ex_app_fee_bps", fee / usd * 1e4, "bps", "Average app fee on 1Click swaps, weighted by dollars", "Intents Explorer API")
    m("ex_near_fee_share_pct", near_fee / fee * 100 if fee else None, "pct", "Share of stated app fees naming a NEAR wallet as recipient", "Intents Explorer API", "understates NEAR's take: 1Click also takes a separate share that is not listed in appFees")
    m("ex_refund_rate_pct", sum(int(r["refunded"]) + int(r["failed"]) for r in recent) / (swaps + sum(int(r["refunded"]) + int(r["failed"]) for r in recent)) * 100, "pct", "Swaps refunded or failed", "Intents Explorer API")
    m("ex_senders_day", statistics.mean(int(r["senders"]) for r in recent), "n", "Distinct senders on a sample day", "Intents Explorer API")
    m("ex_blank_sender_pct", sum(float(r["blank_sender_usd"]) for r in recent) / usd * 100, "pct", "Volume with no sender field", "Intents Explorer API")
    m("ex_top10_senders_day_pct", sum(float(r["top10_sender_usd"]) for r in recent) / usd * 100, "pct", "Top 10 senders' share of a day's volume", "Intents Explorer API")
    weekly = [{"day": r["day"], "usd": float(r["usd_in"]), "swaps": int(r["swaps"]), "fee_bps": round(float(r["app_fee_usd"]) / float(r["usd_in"]) * 1e4, 1) if float(r["usd_in"]) else None,
               "near_bps": round(float(r["near_wallet_fee_usd"]) / float(r["usd_in"]) * 1e4, 2) if float(r["usd_in"]) else None, "official": round(vol.get(d(r["day"]), 0)), "senders": int(r["senders"])} for r in daily]
    CHARTS["ex_daily"] = weekly
    partners = ex("panel_partner")
    pu, pf, pn = Counter(), Counter(), Counter()
    for r in partners:
        if r["day"] in rdays:
            pu[r["partner"]] += float(r["usd"])
            pf[r["partner"]] += float(r["fee_usd"])
            pn[r["partner"]] += int(r["swaps"])
    rows = [{"partner": k, "vol_share": round(v / usd * 100, 1), "fee_share": round(pf[k] / fee * 100, 1) if fee else 0, "bps": round(pf[k] / v * 1e4, 1) if v else 0, "swaps": pn[k]} for k, v in pu.most_common(12)]
    CHARTS["ex_partners"] = rows
    if rows:
        by_fee = max(rows, key=lambda r: r["fee_share"])
        m("ex_top_fee_partner", by_fee["partner"], "text", "Partner that collects the most fees", "Intents Explorer API")
        m("ex_top_fee_partner_vol", by_fee["vol_share"], "pct", "Its share of 1Click volume", "Intents Explorer API")
        m("ex_top_fee_partner_fee", by_fee["fee_share"], "pct", "Its share of app fees", "Intents Explorer API")
        m("ex_top_fee_partner_bps", by_fee["bps"], "bps", "Its average fee", "Intents Explorer API")
        m("ex_top3_vol_share", sum(sorted((r["vol_share"] for r in rows if r["partner"] != "(none)"), reverse=True)[:3]), "pct", "Top three named partners' share of volume", "Intents Explorer API")
    rec = ex("panel_recipient")
    ru = Counter()
    for r in rec:
        if r["day"] in rdays:
            ru[r["recipient"]] += float(r["fee_usd"])
    total_rec = sum(ru.values()) or 1
    CHARTS["ex_recipients"] = [{"recipient": k, "share": round(v / total_rec * 100, 1), "owner": fee_owner(k)} for k, v in ru.most_common(10)]
    owners = Counter()
    for k, v in ru.items():
        owners[fee_owner(k)] += v
    CHARTS["ex_fee_owners"] = [{"k": k, "share": round(v / total_rec * 100, 1)} for k, v in owners.most_common()]
    m("ex_wallets_fee_share_pct", owners["Wallets via SwapKit"] / total_rec * 100, "pct", "Share of app fees going to wallets through SwapKit", "Intents Explorer API")
    fam = Counter()
    cha = Counter()
    for r in ex("panel_routes"):
        if r["day"] in rdays:
            (fam if r["kind"] == "family" else cha)[r["route"]] += float(r["usd"])
    tot = sum(fam.values()) or 1
    CHARTS["ex_routes"] = [{"k": k.replace(">", " to "), "share": round(v / tot * 100, 1)} for k, v in fam.most_common(10)]
    tot_c = sum(cha.values()) or 1
    CHARTS["ex_chain_routes"] = [{"k": k.replace(">", " to "), "share": round(v / tot_c * 100, 1)} for k, v in cha.most_common(10)]
    m("ex_stable_stable_pct", fam["stable>stable"] / tot * 100, "pct", "Stablecoin-to-stablecoin share of volume", "Intents Explorer API")
    m("ex_zec_pct", sum(v for k, v in fam.items() if "zec" in k) / tot * 100, "pct", "Swaps with Zcash on either side, share of volume", "Intents Explorer API")
    size_u, size_n, cost = Counter(), Counter(), defaultdict(list)
    for r in ex("panel_sizes"):
        if r["day"] in rdays:
            size_u[r["bucket"]] += float(r["usd"])
            size_n[r["bucket"]] += int(r["n"])
            if r["cost_p50"] not in ("", "None"):
                cost[r["bucket"]].append(float(r["cost_p50"]))
    order = ["under $100", "$100-1k", "$1k-10k", "$10k-100k", "over $100k"]
    tot_n = sum(size_n.values()) or 1
    tot_u = sum(size_u.values()) or 1
    CHARTS["ex_sizes"] = [{"k": k, "n_share": round(size_n[k] / tot_n * 100, 1), "usd_share": round(size_u[k] / tot_u * 100, 1), "cost": round(statistics.median(cost[k]), 1) if cost[k] else None} for k in order]
    by_day_partner = defaultdict(Counter)
    for r in partners:
        by_day_partner[r["day"]][r["partner"]] += float(r["usd"])
    by_day_route = defaultdict(Counter)
    for r in ex("panel_routes"):
        if r["kind"] == "family":
            by_day_route[r["day"]][r["route"]] += float(r["usd"])
    series = []
    for r in daily:
        day, total_usd = r["day"], float(r["usd_in"])
        if total_usd < 100000:
            continue
        routes = by_day_route[day]
        series.append({"day": day, "swapkit": round(by_day_partner[day]["swapkit"] / total_usd * 100, 1),
                       "zec": round(sum(v for k, v in routes.items() if "zec" in k) / total_usd * 100, 1),
                       "stable": round(routes["stable>stable"] / total_usd * 100, 1)})
    CHARTS["ex_mix_series"] = series
    wk = ex("recent_weekly")
    if wk:
        CHARTS["ex_users"] = [{k: (float(v) if k not in ("week",) and v != "" else (None if v == "" else v)) for k, v in r.items()} for r in wk]
        last = [r for r in wk if r["ret_4w"] != ""]
        m("ex_users_week", statistics.mean(int(r["users"]) for r in wk), "n", "Distinct users a week, recent eight weeks", "Intents Explorer API")
        m("ex_top10_users_week_pct", statistics.mean(float(r["top10_share"]) for r in wk), "pct", "Top 10 users' share of a week's volume", "Intents Explorer API")
        m("ex_top100_users_week_pct", statistics.mean(float(r["top100_share"]) for r in wk), "pct", "Top 100 users' share of a week's volume", "Intents Explorer API")
        if last:
            m("ex_ret_4w_pct", statistics.mean(float(r["ret_4w"]) for r in last), "pct", "Users seen again four weeks later", "Intents Explorer API")
    # Pricing-power test two. A partner's average fee moves with its route mix, so we use the posted fee: the most common
    # fee across its swaps that day. A step in the posted fee that holds is a real price change.
    day_total = Counter()
    for r in partners:
        day_total[r["day"]] += float(r["usd"])
    per = defaultdict(list)
    for r in partners:
        if r["partner"] in ("(other)", "(none)") or int(r["swaps"]) < 20 or not day_total[r["day"]]:
            continue
        per[r["partner"]].append((r["day"], float(r["posted_bps"]), float(r["usd"]) / day_total[r["day"]]))
    events = []
    for name, series in per.items():
        series.sort()
        i = 4
        while i <= len(series) - 4:
            before, after = series[i - 4:i], series[i:i + 4]
            f0 = statistics.median(x[1] for x in before)
            f1 = statistics.median(x[1] for x in after)
            steady = sum(1 for x in after if abs(x[1] - f1) <= 1) >= 3 and sum(1 for x in before if abs(x[1] - f0) <= 1) >= 3
            if steady and abs(f1 - f0) >= 3:
                s0, s1 = statistics.mean(x[2] for x in before), statistics.mean(x[2] for x in after)
                if s0 > 0 and s1 > 0:
                    events.append({"partner": name, "day": series[i][0], "fee_before": f0, "fee_after": f1, "share_before": round(s0 * 100, 2), "share_after": round(s1 * 100, 2),
                                   "log_share_change": round(math.log(s1 / s0), 3)})
                    i += 4
                    continue
            i += 1
    CHARTS["fee_events"] = sorted(events, key=lambda e: e["day"])
    m("fee_events", len(events), "n", "Lasting steps of 3 bps or more in a partner's posted fee", "Intents Explorer API")
    rises = [e for e in events if e["fee_after"] > e["fee_before"]]
    cuts = [e for e in events if e["fee_after"] < e["fee_before"]]
    m("fee_events_rises", len(rises), "n", "Of those, rises", "Intents Explorer API")
    m("fee_events_cuts", len(cuts), "n", "Of those, cuts", "Intents Explorer API")
    rng = random.Random(20261101)

    def mean_ci(rows):
        vals = [r["log_share_change"] for r in rows]
        if len(vals) < 3:
            return None
        boots = sorted(statistics.mean(rng.choice(vals) for _ in vals) for _ in range(4000))
        return statistics.mean(vals), boots[100], boots[3899]
    for name, rows in (("rise", rises), ("cut", cuts)):
        res = mean_ci(rows)
        if res:
            m(f"fee_{name}_share_change_pct", (math.exp(res[0]) - 1) * 100, "pct", f"Average change in partner share after a fee {name}", "Intents Explorer API", "four sample weeks after against four before")
            m(f"fee_{name}_lo_pct", (math.exp(res[1]) - 1) * 100, "pct", f"95% interval, low, fee {name}", "Intents Explorer API")
            m(f"fee_{name}_hi_pct", (math.exp(res[2]) - 1) * 100, "pct", f"95% interval, high, fee {name}", "Intents Explorer API")


def fee_switch(it):
    """Natural experiments on NEAR's take, with rival rails as controls. Rates are pooled from official monthly revenue."""
    rails = load("llama_rail_volumes")
    vol = {k: llama_series(v) for k, v in rails.items()}
    weekly = defaultdict(Counter)
    for k, series in vol.items():
        for day, val in series.items():
            weekly[monday(day)][k] += val
    rows = []
    for w in sorted(weekly):
        rec = weekly[w]
        peers_vol = rec["thorchain-dex"] + rec["chainflip-amm"] + rec["maya-protocol"]
        if rec["near-intents"] <= 0 or peers_vol <= 0 or w >= monday(it["end"]):
            continue
        rows.append({"w": w, "ours": rec["near-intents"], "peers": peers_vol, "share": rec["near-intents"] / (rec["near-intents"] + peers_vol) * 100})
    CHARTS["pricing_series"] = [{"w": r["w"].isoformat(), "share": round(r["share"], 1)} for r in rows if r["w"] >= datetime.date(2025, 9, 1)]
    monthly = {r["m"]: r for r in CHARTS["intents_monthly"]}

    def pooled_rate(keys):
        vol_sum = sum(monthly[k]["vol"] for k in keys if k in monthly)
        return (sum(monthly[k]["rev"] for k in keys if k in monthly) / vol_sum * 1e4 if vol_sum else None,
                sum(monthly[k]["fees"] for k in keys if k in monthly) / vol_sum * 1e4 if vol_sum else None)
    out = []
    rng = random.Random(20261101)
    events = (("Partner revenue share goes live, 23 Feb 2026", datetime.date(2026, 2, 23), 13, ["2025-12", "2026-01", "2026-02"], ["2026-03", "2026-04", "2026-05"]),
              ("Capture steps up, 31 Aug 2026", datetime.date(2026, 8, 31), 8, ["2026-06", "2026-07", "2026-08"], ["2026-09", "2026-10"]))
    for label, start, span, before, after in events:
        pre = [r for r in rows if start - datetime.timedelta(weeks=span) <= r["w"] < start]
        post = [r for r in rows if start <= r["w"] < start + datetime.timedelta(weeks=span)]

        def ratio(sample):
            return sum(r["ours"] for r in sample) / sum(r["peers"] for r in sample)
        boots = sorted(ratio([rng.choice(post) for _ in post]) / ratio([rng.choice(pre) for _ in pre]) for _ in range(4000))
        net0, gross0 = pooled_rate(before)
        net1, gross1 = pooled_rate(after)
        share = lambda sample: sum(r["ours"] for r in sample) / sum(r["ours"] + r["peers"] for r in sample) * 100
        out.append({"label": label, "share_before": round(share(pre), 1), "share_after": round(share(post), 1), "gross_before": round(gross0, 1), "gross_after": round(gross1, 1),
                    "net_before": round(net0, 2), "net_after": round(net1, 2), "relative": round(ratio(post) / ratio(pre), 2), "lo": round(boots[100], 2), "hi": round(boots[3899], 2)})
    CHARTS["pricing_events"] = out
    a, b = out
    m("switch_net_before", a["net_before"], "bps", "Captured rate, three months before the partner share went live", "revenue.near.org")
    m("switch_net_after", a["net_after"], "bps", "Captured rate, three months after", "revenue.near.org")
    m("switch_gross_before", a["gross_before"], "bps", "Gross fee, three months before", "revenue.near.org")
    m("switch_gross_after", a["gross_after"], "bps", "Gross fee, three months after", "revenue.near.org")
    m("switch_share_before", a["share_before"], "pct", "Rail share, 13 weeks before, pooled", "DefiLlama")
    m("switch_share_after", a["share_after"], "pct", "Rail share, 13 weeks after, pooled", "DefiLlama")
    m("switch_rel_lo", a["lo"], "x", "Volume against rivals, after over before, 95% low", "DefiLlama")
    m("switch_rel_hi", a["hi"], "x", "Volume against rivals, after over before, 95% high", "DefiLlama")
    m("step_net_before", b["net_before"], "bps", "Captured rate, June to August 2026", "revenue.near.org")
    m("step_net_after", b["net_after"], "bps", "Captured rate, September and October 2026", "revenue.near.org")
    m("step_share_before", b["share_before"], "pct", "Rail share, eight weeks before 31 August, pooled", "DefiLlama")
    m("step_share_after", b["share_after"], "pct", "Rail share, eight weeks after, pooled", "DefiLlama")
    m("step_rel_lo", b["lo"], "x", "Volume against rivals, after over before, 95% low", "DefiLlama")
    m("step_rel_hi", b["hi"], "x", "Volume against rivals, after over before, 95% high", "DefiLlama")


# ---------------------------------------------------------------- agents
def agents(months):
    rows = bq("12_")
    pat = re.compile(r"(^|[.\-_])(agent|agents|ai|shade|outlayer|nearai|llm|gpt|bot)([.\-_]|$)")
    named = [r for r in rows if pat.search(r["receiver"])]
    oct_tx = months["2026-10"]["txs"]
    tx_named = sum(int(r["txs"]) for r in named)
    biggest = max(named, key=lambda r: int(r["txs"]))
    m("ai_named_tx_oct", tx_named, "n", "Transactions to contracts with AI or agent names, October 2026", "BigQuery transactions")
    m("ai_named_share_pct", tx_named / oct_tx * 100, "pct", "Their share of all transactions", "BigQuery transactions")
    m("ai_named_ex_top_share_pct", (tx_named - int(biggest["txs"])) / oct_tx * 100, "pct", "Share without the largest such contract", "BigQuery transactions", biggest["receiver"])
    sign = {r["receiver"]: r for r in rows}
    if "v1.signer" in sign:
        m("chain_sig_tx_oct", int(sign["v1.signer"]["txs"]), "n", "Direct transactions to v1.signer (Chain Signatures), October 2026", "BigQuery transactions")
    groups = {"Games and rewards": ("game.hot.tg", "spin.sweat", "jars-oracle.sweat", "oracle.sweat", "learn.sweat", "sweat_welcome.near", "wallet.kaiching", "firedrop.hot.tg", "cooldogsclub.near", "embr.playember_reserve.near"),
              "Intents and bridges": ("intents.near", "omni.bridge.near", "omft.near", "btc-connector.bridge.near", "zcash-connector.bridge.near", "v2_1.omni.hot.tg", "v1.signer"),
              "DeFi and tokens": ("v2.ref-finance.near", "wrap.near", "17208628f84f5d6ad33f0da3bbbeb27ffcb398eac501a31bd6ad2011e36133a1", "usdt.tether-token.near", "aurora")}
    out = []
    seen = 0
    for name, members in groups.items():
        n = sum(int(sign[a]["txs"]) for a in members if a in sign)
        seen += n
        out.append({"k": name, "txs": n, "share": round(n / oct_tx * 100, 1)})
    out.append({"k": "AI and agent names", "txs": tx_named, "share": round(tx_named / oct_tx * 100, 1)})
    out.append({"k": "Everything else", "txs": int(oct_tx - seen - tx_named), "share": round((oct_tx - seen - tx_named) / oct_tx * 100, 1)})
    CHARTS["contract_mix"] = out
    top = bq("16_")
    total = sum(int(r["txs"]) for r in top)
    all_tx = months["2026-10"]["tx_intents"]
    lead = top[0]
    escrow = next((r for r in top if r["signer"] == "escrow.ai.near"), None)
    m("intents_top_signer", lead["signer"], "text", "Largest signer into intents.near, October 2026", "BigQuery transactions")
    m("intents_top_signer_share_pct", int(lead["txs"]) / all_tx * 100, "pct", "Its share of direct calls", "BigQuery transactions")
    m("intents_top5_signers_pct", sum(int(r["txs"]) for r in top[:5]) / all_tx * 100, "pct", "Top five signers' share of direct calls into intents.near", "BigQuery transactions")
    if escrow:
        m("escrow_ai_share_pct", int(escrow["txs"]) / all_tx * 100, "pct", "escrow.ai.near share of direct calls into intents.near, October 2026", "BigQuery transactions", "what these calls move is not measured")
    CHARTS["intents_signers"] = [{"signer": r["signer"], "txs": int(r["txs"]), "share": round(int(r["txs"]) / all_tx * 100, 1)} for r in top[:10]]


# ---------------------------------------------------------------- lead-lag
def lead_lag(it):
    px = {from_ts(p["t"]): p["p"] for p in load("price_near_daily")}
    burn = {d(r["block_date"]): float(r["burned_near"]) for r in bq_merged("01_", "01b_", "block_date")}
    nb = {d(r["date"]): int(r["active_accounts"] or 0) for r in load("nearblocks_charts")}
    weeks = defaultdict(lambda: {"v": 0.0, "b": 0.0, "a": 0.0, "p": [], "n": 0})
    for day, val in it["vol"].items():
        if datetime.date(2025, 2, 3) <= day <= it["end"]:
            w = weeks[monday(day)]
            w["v"] += val
            w["b"] += burn.get(day, 0)
            w["a"] += nb.get(day, 0)
            w["n"] += 1
            if day in px:
                w["p"].append(px[day])
    keys = [k for k in sorted(weeks) if weeks[k]["n"] == 7 and weeks[k]["p"]]
    series = [(k, math.log(weeks[k]["v"]), math.log(statistics.mean(weeks[k]["p"])), math.log(weeks[k]["b"] or 1), math.log(weeks[k]["a"] or 1)) for k in keys]
    diff = [(b[0], b[1] - a[1], b[2] - a[2], b[3] - a[3], b[4] - a[4]) for a, b in zip(series, series[1:])]
    break_weeks = {datetime.date(2026, 3, 30), datetime.date(2026, 4, 6)}
    diff = [r for r in diff if r[0] not in break_weeks]
    n = len(diff)
    ccf = []
    for lag in range(-8, 9):
        pairs = [(diff[i][1], diff[i + lag][2]) for i in range(n) if 0 <= i + lag < n]
        ccf.append({"lag": lag, "r": round(corr([p[0] for p in pairs], [p[1] for p in pairs]), 3)})
    CHARTS["ccf_volume_price"] = {"points": ccf, "band": round(1.96 / math.sqrt(n), 3), "n": n}
    results = {}
    rng = random.Random(20261101)
    for name, idx in (("burn", 3), ("active", 4)):
        for h in (1, 2, 4):
            rows = [(sum(diff[i + j][idx] for j in range(1, h + 1)), diff[i][1], diff[i][2], diff[i][idx]) for i in range(n - h)]
            y = [r[0] for r in rows]
            _, r_own = ols(y, [[r[3] for r in rows]])
            _, r_v = ols(y, [[r[3] for r in rows], [r[1] for r in rows]])
            _, r_p = ols(y, [[r[3] for r in rows], [r[2] for r in rows]])
            gain = r_v - r_p
            boots = []
            block = 6
            for _ in range(1500):
                pick = []
                while len(pick) < len(rows):
                    start = rng.randrange(0, len(rows) - block + 1)
                    pick.extend(rows[start:start + block])
                pick = pick[:len(rows)]
                yy = [r[0] for r in pick]
                _, a = ols(yy, [[r[3] for r in pick], [r[1] for r in pick]])
                _, b = ols(yy, [[r[3] for r in pick], [r[2] for r in pick]])
                boots.append(a - b)
            boots.sort()
            lo, hi = boots[int(0.025 * len(boots))], boots[int(0.975 * len(boots))]
            raw_lo, raw_hi = lo, hi
            # walk-forward: does adding the predictor beat the own-history forecast out of sample
            err_own, err_v, err_p = [], [], []
            for t in range(40, len(rows)):
                # a row's target spans h weeks, so drop the h - 1 training rows whose targets overlap the test week
                train, test = rows[:t - h + 1], rows[t]
                yy = [r[0] for r in train]
                bo, _ = ols(yy, [[r[3] for r in train]])
                bv, _ = ols(yy, [[r[3] for r in train], [r[1] for r in train]])
                bp, _ = ols(yy, [[r[3] for r in train], [r[2] for r in train]])
                err_own.append(test[0] - (bo[0] + bo[1] * test[3]))
                err_v.append(test[0] - (bv[0] + bv[1] * test[3] + bv[2] * test[1]))
                err_p.append(test[0] - (bp[0] + bp[1] * test[3] + bp[2] * test[2]))
            rmse = lambda e: math.sqrt(sum(x * x for x in e) / len(e)) if e else None
            results[f"{name}_{h}w"] = {"r2_extra_volume": round(r_v - r_own, 3), "r2_extra_price": round(r_p - r_own, 3), "gap": round(gain, 3), "gap_lo": round(lo, 3), "gap_hi": round(hi, 3), "raw_lo": raw_lo, "raw_hi": raw_hi,
                                       "oos_rmse_own": round(rmse(err_own), 4), "oos_rmse_volume": round(rmse(err_v), 4), "oos_rmse_price": round(rmse(err_p), 4), "n": len(rows)}
    CHARTS["lead_lag"] = results
    wins = sum(1 for r in results.values() if r.pop("raw_lo") > 0)
    losses = sum(1 for r in results.values() if r.pop("raw_hi") < 0)
    oos = sum(1 for r in results.values() if r["oos_rmse_volume"] < r["oos_rmse_own"])
    oos_price = sum(1 for r in results.values() if r["oos_rmse_price"] < r["oos_rmse_own"])
    m("leadlag_oos_price_beats_own", oos_price, "n", "Of six tests, price improves on own-history forecasts out of sample", "pipeline/analyze.py lead_lag")
    m("leadlag_weeks", n, "n", "Weekly observations in the lead-lag test", "DefiLlama + BigQuery + NearBlocks")
    m("leadlag_volume_wins", wins, "n", "Of six tests, volume beats price with the interval above zero", "pipeline/analyze.py lead_lag")
    m("leadlag_price_wins", losses, "n", "Of six tests, price beats volume with the interval below zero", "pipeline/analyze.py lead_lag")
    m("leadlag_oos_beats_own", oos, "n", "Of six tests, volume improves on own-history forecasts out of sample", "pipeline/analyze.py lead_lag")
    peak = max(ccf, key=lambda r: abs(r["r"]))
    m("ccf_peak_lag", peak["lag"], "n", "Lag of largest volume-price correlation (weeks, positive = volume first)", "pipeline/analyze.py lead_lag")
    m("ccf_peak_r", peak["r"], "x", "That correlation", "pipeline/analyze.py lead_lag")


def incidents(as_of):
    """Kill rule T6 reads a hand-kept list of security incidents (thesis/incidents.json), each with its public source.
    The list is the one manual input here: code grades it, but a person has to add the next incident."""
    spec = json.loads((ROOT / "thesis" / "incidents.json").read_text())
    end = d(as_of)
    rows = [r for r in spec["incidents"] if 0 <= (end - d(r["date"])).days < 365]
    unrecovered = sum(max(0.0, r["usd_lost"] - r["usd_recovered"]) for r in rows)
    m("exploit_unrecovered_365d", unrecovered, "usd", "User funds lost to Intents or bridge exploits in the last 365 days and not returned", "thesis/incidents.json (hand-kept, each row cites a source)",
      f"{len(rows)} incident(s) in the window")
    m("exploit_incidents_365d", len(rows), "n", "Security incidents in the last 365 days", "thesis/incidents.json")
    CHARTS["incidents"] = spec["incidents"]


def main():
    t0 = time.time()
    price, mcap = market()
    as_of, mint30, burn30, sup = supply(price)
    months = activity()
    it = intents(price, mint30, as_of)
    accrual(price, mint30, burn30, sup, it)
    staking(price, sup)
    peers(mcap)
    agents(months)
    lead_lag(it)
    fee_switch(it)
    explorer(it)
    incidents(as_of)
    OUT.mkdir(parents=True, exist_ok=True)
    (OUT / "metrics.json").write_text(json.dumps(METRICS, indent=1, sort_keys=True, allow_nan=False))
    (OUT / "charts.json").write_text(json.dumps(CHARTS, separators=(",", ":"), sort_keys=True, allow_nan=False))
    print(f"{len(METRICS)} metrics, {len(CHARTS)} chart data sets, {time.time() - t0:.1f}s")


if __name__ == "__main__":
    main()