near-toll-audit / pipeline/model.py · raw
#!/usr/bin/env python3"""Valuation model. Same maths as docs/model.js (tests/test_model.py proves they agree to 1e-9).Reads scenarios from thesis/scenarios.json, fills live inputs from docs/data/metrics.json, writes docs/data/model.json.Every assumption lives in that one JSON file with its reason."""import json, pathlib, sysROOT = pathlib.Path(__file__).resolve().parent.parentdef path(a, b, t):    return a + (b - a) * (min(t, 5) - 1) / 4def issuance_rate(p, t):    ramp = max(1, p["ramp_years"])    return p["issuance_start"] + (p["issuance_end"] - p["issuance_start"]) * min(t, ramp) / rampdef value(p):    ownership, pv, rows, earnings, vol = 1.0, 0.0, [], 0.0, p["vol0"]    for t in range(1, 6):        vol = vol * (1 + p["cagr"])        gross = vol * path(p["gross_bps_start"], p["gross_bps_end"], t) / 1e4        captured = gross * path(p["capture_start"], p["capture_end"], t)        l1 = p["l1_burn_usd"] * (1 + p["l1_growth"]) ** t        earnings = captured * p["claim"] + l1        leak = issuance_rate(p, t) * (p["treasury_share"] + (1 - p["treasury_share"]) * p["commission"])        ownership *= 1 - leak        pv += earnings * ownership / (1 + p["discount"]) ** t        rows.append({"year": t, "volume": vol, "gross": gross, "captured": captured, "earnings": earnings, "ownership": ownership})    terminal = earnings * p["exit_multiple"] * ownership / (1 + p["discount"]) ** 5    total = pv + terminal + p["l1_value"]    return {"price": total / p["supply"], "equity": total, "pv_cash": pv, "pv_terminal": terminal, "l1_value": p["l1_value"],            "y5_volume": vol, "y5_captured": rows[4]["captured"], "y5_earnings": earnings, "rows": rows}def implied_volume(p, target):    lo, hi = -0.95, 20.0    q = dict(p)    def at(g):        q["cagr"] = g        return value(q)["price"]    if at(hi) < target:        return None    if at(lo) > target:        return {"cagr": lo, "y5_volume": p["vol0"] * (1 + lo) ** 5}    for _ in range(80):        mid = (lo + hi) / 2        if at(mid) < target:            lo = mid        else:            hi = mid    g = (lo + hi) / 2    return {"cagr": g, "y5_volume": p["vol0"] * (1 + g) ** 5}def weighted(scenarios):    return sum(s["probability"] * value(s["inputs"])["price"] for s in scenarios.values())def implied_bull(scenarios, target):    bear = value(scenarios["bear"]["inputs"])["price"]    base = value(scenarios["base"]["inputs"])["price"]    bull = value(scenarios["bull"]["inputs"])["price"]    a, b = scenarios["bear"]["probability"], scenarios["base"]["probability"]    rest = (a * bear + b * base) / (a + b)    return None if bull == rest else (target - rest) / (bull - rest)def fill(scenarios, metrics):    """Replaces "$metric" strings in the assumptions with live values, so the model refreshes with the data."""    out = {}    for name, sc in scenarios.items():        inputs = {}        for key, val in sc["inputs"].items():            if isinstance(val, str) and val.startswith("$"):                expr = val[1:]                scale = 1.0                if "*" in expr:                    expr, factor = expr.split("*")                    scale = float(factor)                inputs[key] = metrics[expr]["value"] * scale            else:                inputs[key] = val        out[name] = {**sc, "inputs": inputs}    return outdef tornado(base, ranges):    out = []    mid = value(base)["price"]    for key, (lo, hi, label) in ranges.items():        a, b = dict(base), dict(base)        a[key], b[key] = lo, hi        out.append({"key": key, "label": label, "low_input": lo, "high_input": hi, "low": value(a)["price"], "high": value(b)["price"], "mid": mid})    return sorted(out, key=lambda r: -abs(r["high"] - r["low"]))def grid(base, vols, nets):    rows = []    for v in vols:        row = []        for net in nets:            q = dict(base)            q["cagr"] = (v / q["vol0"]) ** 0.2 - 1            q["gross_bps_start"] = q["gross_bps_end"] = net            q["capture_start"] = q["capture_end"] = 1.0            row.append(value(q)["price"])        rows.append(row)    return {"volumes": vols, "net_bps": nets, "prices": rows}def main():    metrics = json.loads((ROOT / "docs" / "data" / "metrics.json").read_text())    spec = json.loads((ROOT / "thesis" / "scenarios.json").read_text())    scenarios = fill(spec["scenarios"], metrics)    total_p = sum(s["probability"] for s in scenarios.values())    if abs(total_p - 1) > 1e-9:        raise SystemExit(f"scenario probabilities sum to {total_p}")    price = metrics["price"]["value"]    results = {k: value(s["inputs"]) for k, s in scenarios.items()}    base = scenarios["base"]["inputs"]    what_if = {}    for name, change in spec["what_if"].items():        q = dict(base)        q.update({k: (metrics[v[1:]]["value"] if isinstance(v, str) else v) for k, v in change.items()})        what_if[name] = value(q)["price"]    implied_rich = {}    for name, change in {"exit_40x": {"exit_multiple": 40}, "chain_3_9b": {"l1_value": 3.9e9}}.items():        q = dict(base)        q.update(change)        implied_rich[name] = implied_volume(q, price)    reweighted = {}    for bull_weight in (0.1, 0.2, 0.3, 0.4):        rest = 1 - scenarios["bull"]["probability"]        reweighted[f"{int(bull_weight * 100)}"] = sum(value(s["inputs"])["price"] * (bull_weight if k == "bull" else s["probability"] / rest * (1 - bull_weight)) for k, s in scenarios.items())    out = {"price": price, "scenarios": scenarios, "results": results, "weighted": weighted(scenarios), "what_if": what_if, "implied_rich": implied_rich, "reweighted": reweighted,           "implied_base": implied_volume(base, price), "implied_bull_probability": implied_bull(scenarios, price),           "tornado": tornado(base, {k: tuple(v) for k, v in spec["tornado"].items()}),           "grid": grid(base, spec["grid"]["volumes"], spec["grid"]["net_bps"]), "reasons": spec["reasons"]}    (ROOT / "docs" / "data" / "model.json").write_text(json.dumps(out, indent=1, allow_nan=False))    print(json.dumps({"price": price, "weighted": round(out["weighted"], 3), **{k: round(v["price"], 3) for k, v in results.items()},                      "implied_cagr": out["implied_base"] and round(out["implied_base"]["cagr"], 3), "implied_y5": out["implied_base"] and round(out["implied_base"]["y5_volume"]),                      "implied_bull_p": out["implied_bull_probability"]}))if __name__ == "__main__":    sys.exit(main())