fix: regenerate PIST predictions pipeline, fix Matrices250 kernel blowup

- generate_predictions.py: reads 278 equations from classifier receipt,
  builds 8×8 strand adjacency matrices, deduplicates to 250 by equation_id
- build_pist_matrices_250.py: emit each matrix as a separate Lean def
  (match-based findMatrix) to avoid kernel term-size blowup
- rrc_pist_predictions_250_v1.json regenerated in Research Stack shared-data/
- Shakeout finding: text-derived matrices have spectral radius < 2.0,
  so classifyExact returns none (expected — 250 non-zero matrices, weak features)
This commit is contained in:
allaun 2026-06-30 04:28:23 -05:00
parent e104df6cc3
commit 862992ec98
2 changed files with 153 additions and 32 deletions

View file

@ -7,11 +7,9 @@
Build formal/SilverSight/PIST/Matrices250.lean from
shared-data/rrc_pist_predictions_250_v1.json (or a local copy).
Python's role: read the predictions artifact, serialize each 8×8 matrix as a
Lean Array (Array Int) literal, emit an association list keyed by rrc_eq_<hex>.
The generated Lean file is consumed by SilverSight.PIST.Classify
(classifyProxy/classifyExact) to produce pistProxyLabel/pistExactLabel.
Each 8×8 matrix is emitted as its own named `def` to avoid overwhelming
Lean's kernel with a single large term. `findMatrix` is a `match` on the
equation_id string that delegates to the individual def.
Usage:
python3 python/build_pist_matrices_250.py
@ -20,23 +18,25 @@ Usage:
--out-lean formal/SilverSight/PIST/Matrices250.lean
"""
from __future__ import annotations
import argparse, hashlib, json, sys
import argparse, hashlib, json, re, sys
from pathlib import Path
def lean_str(s: str) -> str:
s = s.replace("\\", "\\\\").replace('"', '\\"')
return f'"{s}"'
def lean_matrix_def_name(eid: str) -> str:
"""Sanitize equation_id to a valid Lean identifier."""
name = re.sub(r"[^a-zA-Z0-9_]", "_", eid)
if name[0].isdigit():
name = "m_" + name
return name
def lean_array_int(xs: list[int]) -> str:
inner = ", ".join(str(x) for x in xs)
return f"#[{inner}]"
return "#[" + ", ".join(str(x) for x in xs) + "]"
def lean_matrix_rows(rows: list[list[int]]) -> str:
inner = ",\n ".join(lean_array_int(r) for r in rows)
return f"#[\n {inner}\n ]"
return "#[\n " + inner + "\n ]"
def main() -> int:
@ -61,24 +61,23 @@ def main() -> int:
raw = preds.get("predictions", [])
print(f"Loaded {len(raw)} predictions from {args.predictions}", file=sys.stderr)
# Deterministic order by equation_id for reproducible builds.
raw_sorted = sorted(raw, key=lambda p: p.get("equation_id", ""))
entries: list[str] = []
defs: list[str] = []
find_cases: list[str] = []
for p in raw_sorted:
eid = p.get("equation_id", "")
mat = p.get("matrix_8x8", [])
if not eid or len(mat) != 8:
print(f"WARNING: skipping malformed entry {eid}", file=sys.stderr)
continue
matrix_lean = lean_matrix_rows(mat)
entries.append(f" ({lean_str(eid)}, {matrix_lean})")
name = lean_matrix_def_name(eid)
defs.append(f"def {name} : Array (Array Int) :=\n {lean_matrix_rows(mat)}")
find_cases.append(f" | \"{eid}\" => some {name}")
# Include a content hash so the generated file can be validated.
content_blob = json.dumps(
{p.get("equation_id"): p.get("matrix_8x8") for p in raw_sorted},
sort_keys=True,
separators=(",", ":"),
sort_keys=True, separators=(",", ":"),
).encode("utf-8")
content_hash = hashlib.sha256(content_blob).hexdigest()
@ -90,26 +89,20 @@ def main() -> int:
"-- Source: rrc_pist_predictions_250_v1.json",
"-- Content hash (SHA-256 canonical matrices): " + content_hash,
"--",
"-- This file is consumed by SilverSight.PIST.Classify (classifyProxy/",
"-- classifyExact) at compile time to produce pistProxyLabel/",
"-- pistExactLabel for each invariant equation_id.",
"-- This file is consumed by SilverSight.RRC.Q16_16Manifold at compile time",
"-- to produce pistProxyLabel/pistExactLabel for each equation_id.",
"",
"namespace SilverSight.PIST.Matrices250",
"",
"/-- All matrices in this list have dimension 8×8 (the canonical strand count).",
" Stored as association list (key → matrix). Generated from",
" rrc_pist_predictions_250_v1.json.",
f" Entries: {len(entries)}. -/",
f"def pistMatrixDim : Nat := 8",
"",
"def pistMatrices250 : List (String × Array (Array Int)) :=",
" [",
",\n".join(entries),
" ]",
*defs,
"",
"/-- Look up an 8×8 matrix by invariant equation_id. -/",
"def findMatrix (eqId : String) : Option (Array (Array Int)) :=",
" pistMatrices250.find? (fun (k, _) => k = eqId) |>.map (fun (_, v) => v)",
" match eqId with",
*find_cases,
" | _ => none",
"",
"end SilverSight.PIST.Matrices250",
"",
@ -117,7 +110,7 @@ def main() -> int:
args.out_lean.parent.mkdir(parents=True, exist_ok=True)
args.out_lean.write_text("\n".join(lines))
print(f"Wrote {args.out_lean} ({len(entries)} entries)", file=sys.stderr)
print(f"Wrote {args.out_lean} ({len(defs)} entries)", file=sys.stderr)
return 0

View file

@ -0,0 +1,128 @@
#!/usr/bin/env python3
"""Regenerate rrc_pist_predictions_250_v1.json from the classifier receipt.
Reads all 278 compiled equations, builds 8×8 strand adjacency matrices,
deduplicates by equation_id per the AGENTS.md merge rules, and writes
the predictions artifact.
Usage:
python3 python/generate_predictions.py
python3 python/generate_predictions.py \
--receipt /path/to/rrc_equation_classifier_receipt.json \
--output /path/to/rrc_pist_predictions_250_v1.json
"""
from __future__ import annotations
import argparse
import hashlib
import json
import re
import sys
from pathlib import Path
from datetime import datetime, timezone
from collections import OrderedDict
def tokenize(text: str) -> list[str]:
text = text.lower()
return re.findall(r"[a-z0-9]+|[+\-*/=^(){}\[\]]", text)
def build_matrix(tokens: list[str]) -> list[list[int]]:
vocab = sorted(set(tokens))
matrix = [[0 for _ in range(8)] for _ in range(8)]
for t_i, t_j in zip(tokens, tokens[1:]):
matrix[vocab.index(t_i) % 8][vocab.index(t_j) % 8] += 1
return matrix
def canonical_json_matrix(matrix: list[list[int]]) -> str:
return json.dumps(matrix, separators=(",", ":"))
def build_record(text: str, equation_id: str, name: str) -> dict:
tokens = tokenize(text)
vocab = sorted(set(tokens))
matrix = build_matrix(tokens)
matrix_json = canonical_json_matrix(matrix)
return {
"equation_id": equation_id,
"name": name,
"equation": text,
"schema": "rrc_pist_predictions_250_v1",
"claim_boundary": "matrix-only;no-classifier;no-lean-spectral",
"proxy_pred": None,
"exact_pred": None,
"matrix_schema": "token_strand_adjacency_8x8_v1",
"matrix_hash": hashlib.sha256(matrix_json.encode("utf-8")).hexdigest(),
"matrix_8x8": matrix,
"global_vocab_hash": hashlib.sha256(
json.dumps(vocab, separators=(",", ":")).encode("utf-8")
).hexdigest(),
"source_records": [],
}
def main():
parser = argparse.ArgumentParser(description="Regenerate PIST predictions")
parser.add_argument("--receipt", type=Path,
default=Path("/home/allaun/Research Stack/archive/experimental-shim-probes/rrc_equation_classifier_receipt.json"))
parser.add_argument("--output", type=Path,
default=Path("/home/allaun/Research Stack/shared-data/rrc_pist_predictions_250_v1.json"))
args = parser.parse_args()
receipt = json.loads(args.receipt.read_text())
equations = receipt["compiled_equations"]
print(f"Loaded {len(equations)} source equations", file=sys.stderr)
# Dedup by equation_id: deterministic representative selection per AGENTS.md
# representative = min(records, key = equation_record.equation_id)
merged: dict[str, dict] = OrderedDict()
for eq in equations:
rec = eq.get("equation_record", eq)
eid = rec["equation_id"]
name = rec.get("name", eid)
text = rec.get("equation", "")
if eid not in merged:
merged[eid] = build_record(text, eid, name)
merged[eid]["source_records"] = []
merged[eid]["source_records"].append({
"equation_record_id": eid,
"name": name,
})
predictions = list(merged.values())
# Build global vocab
all_tokens = set()
for p in predictions:
all_tokens.update(tokenize(p.get("equation", "")))
global_vocab = sorted(all_tokens)
output = {
"schema": "rrc_pist_predictions_250_v1",
"generated_at": datetime.now(timezone.utc).isoformat(),
"claim_boundary": "matrix-only;no-classifier;no-lean-spectral",
"global_vocab_hash": hashlib.sha256(
json.dumps(global_vocab, separators=(",", ":")).encode("utf-8")
).hexdigest(),
"global_vocab": global_vocab,
"matrix_schema": "token_strand_adjacency_8x8_v1",
"summary": {
"total_source_records": len(equations),
"unique_equation_ids": len(predictions),
},
"predictions": predictions,
}
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(json.dumps(output, indent=2) + "\n")
print(f"Wrote {args.output} ({len(predictions)} unique predictions from {len(equations)} source records)", file=sys.stderr)
if __name__ == "__main__":
main()