near-toll-audit / pipeline/build.py · raw
#!/usr/bin/env python3"""Renders site/template.html and THESIS.md into docs/, so the page carries real numbers even with JavaScript off."""import datetime, gzip, hashlib, html, json, pathlib, re, shutil, sysROOT = pathlib.Path(__file__).resolve().parent.parentDOCS = ROOT / "docs"def compact(v, digits=1):    a = abs(v)    for size, suffix in ((1e12, "T"), (1e9, "B"), (1e6, "M"), (1e3, "K")):        if a >= size:            text = f"{v / size:.{digits}f}"            return (text.rstrip("0").rstrip(".") if "." in text else text) + suffix    return f"{v:.0f}"FORMATS = {    "text": str,    "n": lambda v: f"{int(v):,}",    "usd": lambda v: "$" + compact(v),    "usd2": lambda v: f"${v:.2f}",    "usd3": lambda v: f"${v:.3f}",    "usdm": lambda v: "$" + compact(v, 0),    "nearm": lambda v: compact(v, 2 if abs(v) < 1e8 else 0) + " NEAR",    "near0": lambda v: f"{v:,.0f} NEAR",    "nearb2": lambda v: f"{v / 1e9:.2f}B NEAR",    "pct0": lambda v: f"{v:.0f}%",    "pct1": lambda v: f"{v:.1f}%",    "pct2": lambda v: f"{v:.2f}%",    "pct_from_x": lambda v: f"{v * 100:.0f}%",    "x0": lambda v: f"{v:.0f}x",    "x1": lambda v: f"{v:.1f}x",    "usd0": lambda v: f"${v:,.0f}",    "num": lambda v: compact(v),    "signed0": lambda v: f"{v:+.0f}%",    "abs0": lambda v: f"{abs(v):.0f}%",    "abs1": lambda v: f"{abs(v):.1f}%",    "bps1": lambda v: f"{v:.1f} bps",    "num2": lambda v: f"{v:.2f}",    "bps0": lambda v: f"{v:.0f} bps",    "bps2": lambda v: f"{v:.2f} bps",}def derived(metrics):    """Display-only values built from registered metrics."""    v = lambda k: metrics[k]["value"]    out = dict(metrics)    add = lambda k, val: out.__setitem__(k, {"value": val})    add("tx_drop_apr26_abs", abs(v("tx_drop_apr26_pct")))    add("signers_engaged_change_abs", abs(v("signers_engaged_change_pct")))    add("as_of_month", datetime.date.fromisoformat(v("as_of")).strftime("%B %Y"))    add("rail_premium_rune", v("mcap_per_volume_near") / v("mcap_per_volume_rune"))    add("dex_annual", v("spot_dex_30d") * 12)    add("window_month_name", datetime.date.fromisoformat(v("window_month") + "-01").strftime("%B %Y"))    add("tg_engaged_change_abs", abs(v("tg_engaged_change_pct")))    add("self_engaged_change_abs", abs(v("self_engaged_change_pct")))    add("peer_chg_1y_median_abs", abs(v("peer_chg_1y_median_pct")))    add("capture_fraction_pct", v("capture_30d_pct"))    add("per100_captured_bps", v("per100_captured") * 100)    add("capture_month_pct", v("net_bps_30d") / v("gross_bps_30d") * 100)    return outdef render(text, metrics, model, grades, escape=True):    def sub_metric(m):        key, fmt = m.group(1), m.group(2)        val = (metrics.get(key) or {}).get("value")        if val is None:            return "n/a"        out = FORMATS[fmt](val)        return html.escape(out) if escape else out    def sub_model(m):        key = m.group(1)        implied = model["implied_base"] or {}        vals = {"weighted": f"${model['weighted']:.2f}",                "gap": f"{abs(model['weighted'] / model['price'] - 1) * 100:.0f}%",                "bear": f"${model['results']['bear']['price']:.2f}", "base": f"${model['results']['base']['price']:.2f}", "bull": f"${model['results']['bull']['price']:.2f}",                "implied_multiple_low": f"{model['implied_rich']['chain_3_9b']['y5_volume'] / metrics['vol_annualised']['value']:.0f}x",                "implied_multiple_40x": f"{model['implied_rich']['exit_40x']['y5_volume'] / metrics['vol_annualised']['value']:.0f}x",                "weighted_bull_30": f"${model['reweighted']['30']:.2f}", "weighted_bull_10": f"${model['reweighted']['10']:.2f}",                "base_chain_share": f"{model['results']['base']['l1_value'] / model['results']['base']['equity'] * 100:.0f}%",                "base_chain_value": "$" + compact(model['results']['base']['l1_value'], 2),                "premium_over_base": f"{(model['price'] / model['results']['base']['price'] - 1) * 100:.0f}%",                "base_equity": "$" + compact(model['results']['base']['equity'], 2),                "supply": compact(model['scenarios']['base']['inputs']['supply'], 2) + " NEAR",                "base_chain_per_near": f"${model['results']['base']['l1_value'] / model['scenarios']['base']['inputs']['supply']:.2f}",                "implied_cagr_raw": f"{implied.get('cagr', 0):.3f}" if implied else "n/a",                "implied_y5_share": f"{implied.get('y5_volume', 0) / metrics['dex_annual']['value'] * 100:.0f}%" if implied else "n/a",                **{f"if_{k}": f"${v:.2f}" for k, v in model.get("what_if", {}).items()},                "implied_cagr": f"{implied.get('cagr', 0) * 100:.0f}%" if implied else "n/a",                "implied_y5": "$" + compact(implied.get("y5_volume", 0), 0) if implied else "n/a",                "implied_multiple": f"{implied.get('y5_volume', 0) / metrics['vol_annualised']['value']:.0f}x" if implied else "n/a",                "implied_bull": f"{model['implied_bull_probability'] * 100:.0f}%",                "bull_weight": f"{model['scenarios']['bull']['probability'] * 100:.0f}%"}        return html.escape(vals[key]) if escape else vals[key]    def sub_static(m):        kind = m.group(1)        out = KILL_STATIC(grades) if kind == "kill" else ELAST_TEXT(metrics) if kind == "elast" else html.escape(grades["note"])        return out if escape else re.sub(r"<[^>]+>", "", out)    def sub_grades(m):        live = sum(1 for r in grades["rows"] if r["status"] == "live")        tripped = len(grades["tripped"])        return f"{tripped} of {len(grades['rows'])} kill criteria tripped."    # Asset versions come from their content, so a visitor never mixes a new page with an old cached script.    if "{{build}}" in text:        digest = hashlib.sha256(b"".join((DOCS / n).read_bytes() for n in ("app.js", "charts.js", "model.js", "styles.css"))).hexdigest()[:10]        text = text.replace("{{build}}", digest)    text = re.sub(r"\{\{m:([a-z0-9_]+):([a-z0-9_]+)\}\}", sub_metric, text)    text = re.sub(r"\{\{model:([a-z0-9_]+)\}\}", sub_model, text)    def sub_case(m):        case, field, fmt = m.group(1), m.group(2), m.group(3)        sc = model["scenarios"][case]        val = sc["probability"] if field == "probability" else sc["inputs"][field]        return FORMATS[fmt](val * (100 if fmt.startswith("pct") else 1))    text = re.sub(r"\{\{s:([a-z]+):([a-z0-9_]+):([a-z0-9_]+)\}\}", sub_case, text)    text = re.sub(r"\{\{static:(killnote|kill|elast)\}\}", sub_static, text)    return re.sub(r"\{\{grades:summary\}\}", sub_grades, text)def _kill_value(r, key):    v = r[key]    if v is None:        return "n/a"    if r["unit"] == "usd":        return "$" + compact(v, 0 if key == "threshold" else 1)    if r["unit"] == "NEAR":        return compact(v, 0 if key == "threshold" else 1)    if key == "threshold":        num = f"{v:g}"    else:        num = f"{v:.2f}" if abs(v) < 10 else f"{round(v)}"    return num + ("%" if r["unit"] == "%" else "x" if r["unit"] == "x" else " " + r["unit"])def KILL_STATIC(grades):    """The kill list as plain HTML, so a reader with JavaScript off (or a crawler) sees every row. app.js redraws it."""    e = html.escape    out = ['<figcaption><h3>Criteria, graded by code at each build</h3><p>Thresholds and dates are fixed. Values are from the build '           f'graded on {e(grades["graded_on"])} (data through {e(grades["as_of"])}).</p></figcaption>']    for side, head in (("toll", "Would kill the toll road"), ("token", "Would kill our caution on the token")):        out.append(f'<h4 class="kill-head">{head}</h4><div class="kill-list">')        for r in grades["rows"]:            if r["side"] != side:                continue            added = f' (added {e(r["added"])})' if r.get("added") else ""            out.append(f'<div class="kill-row"><div class="kill-top"><span class="kill-id">{e(r["id"])}</span>'                       f'<span class="kill-claim">{e(r["claim"])}{added}</span><span class="pill {e(r["status"])}">{e(r["status"])}</span></div>'                       f'<div class="kill-facts"><span>Now {e(_kill_value(r, "value"))}</span><span>Trips if {e(r["op"])} {e(_kill_value(r, "threshold"))}</span>'                       f'<span>By {e(r["deadline"])}</span></div><div class="kill-effect">{e(r["effect"])}</div></div>')        out.append("</div>")    return "".join(out)def ELAST_TEXT(metrics):    """Test three in words, the same sentence app.js writes."""    v = lambda k: (metrics.get(k) or {}).get("value")    rise = v("fee_rise_share_change_pct")    if rise is None:        return "the partner panel is still filling in."    return (f"across {v('fee_events_rises')} lasting rises in a partner's posted fee, its share of swaps changed "            f"{rise:+.0f}% on average, with a 95% interval of {v('fee_rise_lo_pct'):.0f}% to {v('fee_rise_hi_pct'):+.0f}%. "            "Too wide to read, and most of these partners are small.")def inline(text):    text = html.escape(text, quote=True)    text = re.sub(r"`([^`]+)`", r"<code>\1</code>", text)    text = re.sub(r"\*\*([^*]+)\*\*", r"<strong>\1</strong>", text)    return re.sub(r"\[([^\]]+)\]\((https?://[^)\s\"'<>&]+|(?!//)[a-zA-Z0-9_./#-]+)\)", r'<a href="\2">\1</a>', text)def markdown(src):    """A small Markdown subset: headings, paragraphs, bulleted and numbered lists, tables. The memo uses nothing else."""    out, para, lst, tbl = [], [], False, []    def flush():        nonlocal para, lst, tbl        if para:            out.append("<p>" + inline(" ".join(para)) + "</p>")            para = []        if lst:            out.append("</%s>" % lst)            lst = False        if tbl:            rows = [[c.strip() for c in r.strip().strip("|").split("|")] for r in tbl if not re.match(r"^\|?[\s:|-]+\|?$", r)]            out.append('<div class="scroll"><table class="data"><thead><tr>' + "".join(f"<th>{inline(c)}</th>" for c in rows[0]) + "</tr></thead><tbody>" +                       "".join("<tr>" + "".join(f'<td class="wrap-cell">{inline(c)}</td>' for c in r) + "</tr>" for r in rows[1:]) + "</tbody></table></div>")            tbl = []    for line in src.splitlines():        s = line.rstrip()        if not s.strip():            flush()        elif s.startswith("|"):            if para or lst:                flush()            tbl.append(s)        elif s.startswith("#"):            flush()            level = len(s) - len(s.lstrip("#"))            out.append(f"<h{level}>{inline(s[level:].strip())}</h{level}>")        elif s.startswith("- ") or re.match(r"^\d+\. ", s):            kind = "ul" if s.startswith("- ") else "ol"            if para or (lst and lst != kind):                flush()            if not lst:                out.append('<ul class="clean">' if kind == "ul" else "<ol>")                lst = kind            out.append(f"<li>{inline(s[2:] if kind == 'ul' else s.split('. ', 1)[1])}</li>")        else:            para.append(s.strip())    flush()    return "\n".join(out)CSP = "default-src 'none'; script-src 'self'; style-src 'self'; connect-src 'self'; img-src 'self' data:; base-uri 'none'; form-action 'none'"def thesis_page():    src = (ROOT / "THESIS.md").read_text()    body = markdown(src)    (DOCS / "thesis.html").write_text(f"""<!doctype html><html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1"><title>NEAR Toll Audit Memo</title><meta name="description" content="The NEAR Toll Audit memo: thesis, evidence, kill criteria and the strongest case against it."><link rel="stylesheet" href="styles.css?v=memo2"><meta http-equiv="Content-Security-Policy" content="{CSP}"><script src="theme.js"></script></head><body><main class="memo"><p class="small"><a href="./">Back to the dashboard</a></p>{body}</main></body></html>""")def main():    metrics = derived(json.loads((DOCS / "data" / "metrics.json").read_text()))    model = json.loads((DOCS / "data" / "model.json").read_text())    grades = json.loads((DOCS / "data" / "grades.json").read_text())    shutil.copy(ROOT / "site" / "styles.css", DOCS / "styles.css")    (DOCS / "index.html").write_text(render((ROOT / "site" / "template.html").read_text(), metrics, model, grades))    if (ROOT / "THESIS.md").exists():        thesis_page()    status = {}    path = ROOT / "raw" / "latest" / "_fetch_status.json.gz"    if path.exists():        status = json.loads(gzip.open(path, "rb").read())    bq_pulled = "n/a"    manifest = ROOT / "raw" / "bigquery" / "MANIFEST.jsonl"    if manifest.exists():        lines = manifest.read_text().strip().splitlines()        if lines:            bq_pulled = json.loads(lines[-1])["pulled_at"]    (DOCS / "data" / "status.json").write_text(json.dumps({        "built_at": datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%d %H:%M UTC"), "bigquery_pulled": bq_pulled, "sources": status}, indent=1))    print("site built")if __name__ == "__main__":    sys.exit(main())