Research-Stack/4-Infrastructure/shim/build_corpus250.py
allaun 6005a436a3 chore(lean): ncDerived negative control witness from manifold primitives
- Added ncDerived definition in Emit.lean: residualRisk × scaleBandDeclared
- Added ncDerived_independence_justification with CRT product principle link
- Regenerated Corpus250.lean with 278 rows containing manifold primitives
- Fixed syntax error in build_corpus250.py (missing Quote on line 161)

Justification: When weak axes are independent coprime projections, CRT
reconstruction recovers the class modulo product of moduli (InteractionGraphSidon).
Manifold coordinates residualRisk/scaleBandDeclared are orthogonal dimensions;
their product quantifies joint witness strength (P(A∩B) ≤ P(A)×P(B)).

Build: 3314 jobs, 0 errors (lake build Compiler)
2026-06-29 14:23:11 -05:00

235 lines
No EOL
11 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.10"
# dependencies = []
# ///
"""
Build Semantics/RRC/Corpus250.lean from rrc_equation_classifier_receipt.json,
merged with matrix predictions from rrc_pist_predictions_250_v1.json and
arXiv predictions from rrc_arxiv_predictions_250_v1.json.
Python's role:
- read raw fields from classifier receipt (incl. arxiv_paper_id)
- merge PIST predictions (proxy/exact labels, matrix_hash guard) by
invariant_receipt.object_id
- emit deterministic Lean source
Lean's role:
- alignment gate via determineAlignment (reads pistProxyLabel/pistExactLabel)
- receipt stamping
- all admissibility and promotion decisions
- derive negative control witness from manifold primitives
"""
from __future__ import annotations
import hashlib, json, re, sys
from pathlib import Path
ROOT = Path("/home/allaun/Research Stack")
RECEIPT_JSON = ROOT / "archive/experimental-shim-probes/rrc_equation_classifier_receipt.json"
PREDICTIONS_JSON = ROOT / "shared-data/rrc_pist_predictions_250_v1.json"
OUT_LEAN = ROOT / "0-Core-Formalism/lean/Semantics/Semantics/RRC/Corpus250.lean"
# ── shape mapping (classifier JSON → Lean RRCShape constructor) ──────────────
SHAPE_MAP = {
"CognitiveLoadField": ".cognitiveLoadField",
"SignalShapedRouteCompiler": ".signalShapedRouteCompiler",
"ProjectableGeometryTopology": ".projectableGeometryTopology",
"CadForceProbeReceipt": ".cadForceProbeReceipt",
"LogogramProjection": ".logogramProjection",
"HoldForUnlawfulOrUnderspecifiedShape": ".holdForUnlawfulOrUnderspecifiedShape",
}
# ── template_key logic (rrc_kind → template) ─────────────────────────────────
def template_key(rrc_kind: str, status: str) -> str:
if status == "HOLD":
return "hold"
kind_map = {
"cognitive_field_receipt": "definition",
"compression_route_prior": "master_equation",
"geometry_topology_receipt":"definition",
"cad_force_receipt": "gate",
"logogram_projection": "receipt",
"negative_control": "hold",
}
return kind_map.get(rrc_kind, "definition")
# ── operator token derivation (route_hint + rrc_kind + equation text) ──────────
def operator_tokens(er: dict) -> list[str]:
tokens = []
rh = (er.get("route_hint_non_authoritative") or "").strip()
rk = (er.get("rrc_kind") or "").strip()
if rh and rh != "unclassified_equation":
tokens.append(rh)
if rk:
tokens.append(rk)
# Extract simple operator keywords from equation text
eq_text = (er.get("equation") or "").lower()
for op in ["exp(", "log(", "max(", "min(", "sum(", "integral", "derivative",
"laplacian", "nabla", "div(", "curl(", "sigmoid", "softmax",
"tanh(", "relu(", "norm(", "dot(", "cross("]:
if op in eq_text:
tokens.append(op.rstrip("("))
return list(dict.fromkeys(tokens)) # deduplicate, preserve order
# ── string escaping for Lean ──────────────────────────────────────────────────
def lean_str(s: str) -> str:
s = s.replace("\\", "\\\\").replace('"', '\\"')
return f'"{s}"'
def lean_opt(s: str | None) -> str:
if s is None:
return "none"
return f"some {lean_str(s)}"
def lean_str_list(xs: list[str]) -> str:
return "[" + ", ".join(lean_str(x) for x in xs) + "]"
def lean_classify_label(eq_id: str, fn: str) -> str:
"""Generate Lean expression for PIST label.
Checks prediction sources in order:
1. PIST predictions (rrc_pist_predictions_250_v1.json)
2. arXiv predictions (rrc_arxiv_predictions_v1.json)
3. OEIS predictions (rrc_oeis_predictions_v1.json)
4. Fallback to matrix-hash lookup via classifyProxy/classifyExact
"""
for source, label_key in [(PIST_BY_ID, "pist"), (ARXIV_BY_ID, "arxiv"), (OEIS_BY_ID, "oeis")]:
label = source.get(eq_id, {}).get(fn.replace("classify", "").lower() + "_pred")
if label:
return f"some {lean_str(label)}"
return f"Option.bind (findMatrix {lean_str(eq_id)}) Semantics.PIST.Classify.{fn}"
def load_predictions(path: Path) -> dict[str, dict]:
"""Load predictions JSON into equation_id lookup dict."""
if not path.exists():
return {}
try:
data = json.loads(path.read_text())
return {p.get("equation_id", ""): p for p in data.get("predictions", []) if p.get("equation_id")}
except (json.JSONDecodeError, IOError):
return {}
ROOT = Path("/home/allaun/Research Stack")
PIST_BY_ID = load_predictions(ROOT / "shared-data/rrc_pist_predictions_250_v1.json")
ARXIV_BY_ID = load_predictions(ROOT / "shared-data/rrc_arxiv_predictions_v1.json")
OEIS_BY_ID = load_predictions(ROOT / "shared-data/rrc_oeis_predictions_v1.json")
# ── main ──────────────────────────────────────────────────────────────────────
def main() -> None:
d = json.loads(RECEIPT_JSON.read_text())
eqs = d["compiled_equations"]
print(f"Loaded {len(eqs)} equations from classifier receipt", file=sys.stderr)
lines: list[str] = []
lines.append("-- Semantics.RRC.Corpus250 — AUTO-GENERATED by build_corpus250.py")
lines.append("-- DO NOT EDIT BY HAND. Regenerate with:")
lines.append("-- python3 4-Infrastructure/shim/build_corpus250.py")
lines.append("--")
lines.append("-- Python role: raw feature extraction + PIST predictions merge.")
lines.append("-- Lean role: alignment gate (determineAlignment), receipt stamping,")
lines.append("-- all admissibility and promotion decisions.")
lines.append("--")
lines.append("-- Merge contract: pistProxyLabel/pistExactLabel are computed by")
lines.append("-- Semantics.PIST.Classify.classifyProxy / classifyExact over the 8×8")
lines.append("-- braid adjacency matrix from Semantics.PIST.Matrices250.pistMatrices250")
lines.append("-- (keyed by invariant_receipt.object_id). v2 stubs return none;")
lines.append("-- when the classifier surface is defined labels populate automatically.")
lines.append("--")
lines.append("-- Sources:")
lines.append(f"-- classifier receipt: archive/experimental-shim-probes/rrc_equation_classifier_receipt.json")
lines.append(f"-- predictions: shared-data/rrc_pist_predictions_250_v1.json")
lines.append(f"-- Equation count: {len(eqs)}")
lines.append("-- Labels computed by: Semantics.PIST.Classify.classifyProxy / classifyExact")
lines.append("")
lines.append("import Semantics.RRC.Emit")
lines.append("import Semantics.PIST.Classify")
lines.append("import Semantics.PIST.Matrices250")
lines.append("")
lines.append("namespace Semantics.RRC.Corpus250")
lines.append("")
lines.append("open Semantics.RRC.Emit")
lines.append("open Semantics.RRCLogogramProjection")
lines.append("open Semantics.ReceiptCore")
lines.append("open Semantics.PIST.Matrices250")
lines.append("")
lines.append("/-- Full 250-equation corpus from rrc_equation_classifier_receipt.json,")
lines.append(" merged with PIST matrix predictions from rrc_pist_predictions_250_v1.json.")
lines.append(" Each row carries raw features only; the alignment gate in")
lines.append(" Semantics.RRC.Emit.emitCorpus makes all admissibility decisions. -/")
lines.append("def corpus250 : List FixtureRow := [")
row_strs: list[str] = []
for eq in eqs:
er = eq["equation_record"]
ir = eq["invariant_receipt"]
tw = eq["type_witness"]
eq_id = ir.get("object_id", "")
name = er["name"]
shape_str = ir["shape"]
lean_shape = SHAPE_MAP.get(shape_str, ".holdForUnlawfulOrUnderspecifiedShape")
status_str = ir["status"]
lean_status = ".candidate" if status_str == "CANDIDATE" else ".hold"
rrc_kind = er.get("rrc_kind", "")
weak_cnt = len(tw.get("missing_or_weak_axes") or [])
# Negative control: observed value + manifold primitives for Lean derivation
coords = eq.get("manifold_projection", {}).get("coordinates", {})
nc_strength = coords.get("negative_control_strength", 0.0)
residual_risk = coords.get("residual_risk", 0.0)
scale_band = coords.get("scale_band_declared", 0.0)
weak_axes = tw.get("missing_or_weak_axes") or []
# Generator fields
op_tokens = operator_tokens(er)
inv_declared = (er.get("domain_type") or "unknown").strip() or "unknown"
bound_conds = (er.get("bind_class") or "unknown").strip() or "unknown"
t_key = template_key(rrc_kind, status_str)
t_params = f"route={er.get('route_hint_non_authoritative','unclassified_equation') or 'unclassified_equation'};shape={shape_str}"
# arXiv paper id (e.g. "2604.21919") when classifier receipt has
# arxiv_paper_id set; otherwise None. The arXiv cross-reference pipeline
# keys predictions by equation_id, so any arXiv label lookup still flows
# through lean_classify_label (PIST first, then arXiv, then OEIS).
arxiv_pid = (er.get("arxiv_paper_id") or "").strip() or None
row = (
f" {{ equationId := {lean_str(eq_id)}\n"
f" name := {lean_str(name)}\n"
f" shape := {lean_shape}\n"
f" status := {lean_status}\n"
f" rrcKind := {lean_str(rrc_kind)}\n"
f" weakAxesCnt := {weak_cnt}\n"
f" pistProxyLabel := {lean_classify_label(eq_id, 'classifyProxy')}\n"
f" pistExactLabel := {lean_classify_label(eq_id, 'classifyExact')}\n"
f" arxivPaperId := {lean_opt(arxiv_pid)}\n"
f" ncObserved := {nc_strength}\n"
f" residualRisk := {residual_risk}\n"
f" scaleBandDeclared := {scale_band}\n"
f" weakAxesNames := {lean_str_list(weak_axes)}\n"
f" operatorTokens := {lean_str_list(op_tokens)}\n"
f" invariantsDeclared := {lean_str(inv_declared)}\n"
f" boundaryConds := {lean_str(bound_conds)}\n"
f" templateKey := {lean_str(t_key)}\n"
f" templateParams := {lean_str(t_params)} }}"
)
row_strs.append(row)
lines.append(",\n".join(row_strs))
lines.append("]")
lines.append("")
lines.append("end Semantics.RRC.Corpus250")
lines.append("")
OUT_LEAN.write_text("\n".join(lines))
print(f"Wrote {OUT_LEAN}", file=sys.stderr)
print(f"Total rows: {len(row_strs)}", file=sys.stderr)
total_preds = len(PIST_BY_ID) + len(ARXIV_BY_ID) + len(OEIS_BY_ID)
print(f"Labels source: PIST({len(PIST_BY_ID)}) + arXiv({len(ARXIV_BY_ID)}) + OEIS({len(OEIS_BY_ID)}) = {total_preds}", file=sys.stderr)
if __name__ == "__main__":
main()