near-toll-audit / pipeline/grade.py · raw
#!/usr/bin/env python3"""Grades the kill criteria against the latest metrics and appends to a hash-chained ledger.A tripped row shows on the page with no human step. Each ledger line carries the hash of the previous line, so editingone old grade breaks the chain. The chain has no key, so anyone could recompute all of it: public git history is whatmakes a rewrite visible, and CI refuses a new ledger that does not start with the old one byte for byte."""import datetime, hashlib, json, pathlib, sysROOT = pathlib.Path(__file__).resolve().parent.parentdef line_hash(prev, payload): return hashlib.sha256((prev + json.dumps(payload, sort_keys=True, separators=(",", ":"))).encode()).hexdigest()def verify_chain(path): prev = "0" * 64 if not path.exists(): raise SystemExit("ledger file is missing") n = 0 for raw in path.read_text().splitlines(): rec = json.loads(raw) got = rec.pop("hash") if rec["prev"] != prev or line_hash(prev, rec) != got: raise SystemExit(f"ledger chain broken at line {n + 1}") prev = got n += 1 return ndef main(argv): metrics = json.loads((ROOT / "docs" / "data" / "metrics.json").read_text()) model_path = ROOT / "docs" / "data" / "model.json" if model_path.exists(): # A5 watches the market's premium over our own base case, which only the model knows. model = json.loads(model_path.read_text()) base = model["results"]["base"]["price"] metrics["price_over_base_pct"] = {"value": round((model["price"] / base - 1) * 100, 2) if base else None} spec_path = ROOT / "thesis" / "kill_criteria.json" spec = json.loads(spec_path.read_text()) spec_hash = hashlib.sha256(spec_path.read_bytes()).hexdigest() today = datetime.datetime.now(datetime.timezone.utc).date().isoformat() as_of = metrics["as_of"]["value"] rows = [] for c in spec["criteria"]: val = (metrics.get(c["metric"]) or {}).get("value") tripped = None if isinstance(val, (int, float)): tripped = val < c["threshold"] if c["op"] == "<" else val > c["threshold"] expired = today > c["deadline"] status = "unknown" if tripped is None else "tripped" if tripped else "survived" if expired else "live" room = None if isinstance(val, (int, float)) and c["threshold"]: room = (val - c["threshold"]) / abs(c["threshold"]) * 100 room = room if c["op"] == "<" else -room rows.append({**c, "value": val, "status": status, "room_pct": None if room is None else round(room, 1)}) out = {"as_of": as_of, "graded_on": today, "spec_sha256": spec_hash, "rows": rows, "tripped": [r["id"] for r in rows if r["status"] == "tripped"], "note": spec["note"]} (ROOT / "docs" / "data" / "grades.json").write_text(json.dumps(out, indent=1)) ledger = ROOT / "ledger" / "grades.jsonl" ledger.parent.mkdir(exist_ok=True) if not ledger.exists() and "--offline" not in argv: ledger.write_text("") count = verify_chain(ledger) if ledger.exists() else 0 if "--offline" not in argv: prev = "0" * 64 if ledger.exists() and ledger.read_text().strip(): prev = json.loads(ledger.read_text().strip().splitlines()[-1])["hash"] rec = {"prev": prev, "graded_on": today, "as_of": as_of, "spec_sha256": spec_hash, "statuses": {r["id"]: r["status"] for r in rows}, "values": {r["id"]: r["value"] for r in rows}} last = json.loads(ledger.read_text().strip().splitlines()[-1]) if count else None if not last or last["statuses"] != rec["statuses"] or last["graded_on"] != today: rec["hash"] = line_hash(prev, rec) with ledger.open("a", encoding="utf8") as fh: fh.write(json.dumps(rec, sort_keys=True) + "\n") count += 1 print(f"graded {len(rows)} criteria, tripped: {out['tripped'] or 'none'}, ledger lines: {count}")if __name__ == "__main__": sys.exit(main(sys.argv))