#!/usr/bin/env python3 # -*- coding: utf-8 -*- """PDG Regression Suite for the Observed Particle Spectrum Closure companion. This engine makes the companion document *testable*. It loads a frozen set of theory outputs and PDG/evidence CSVs, runs eight validation tests, and emits seven machine-readable + human-readable reports. It fails closed: any missing required field, unit, uncertainty, claim class, or comparison value causes the offending row to FAIL and the process to exit non-zero. Safe wording (also written into every report): The regression suite does not prove the theory true. It prevents the document from making claims that are unsupported by its own tables, methods, and declared evidence. Design constraints (PM-AGENT governance, RUNNABLE-FIRST): * Pure Python 3 standard library: csv, json, hashlib, pathlib, argparse, dataclasses, datetime, re, sys, math, typing. * PyYAML is used if importable, but a stdlib fallback parses the flat ``validation_config.yaml`` (and a ``config.json`` mirror), so the suite runs with zero pip installs. * Every PDG constant in the CSVs must carry a 'source' / '*_source' column; rows without one fail Test 1. CLI: python pdg_regression.py --data data --out out --config validation_config.yaml Exit codes: 0 every row passed validation AND the release gate is satisfiable (subject to config; a pure WARN does not flip the exit code). 1 one or more rows FAILED a required-field / claim-class / residual / status invariant, OR a required input file was missing / unparseable, OR the release gate failed. 2 CLI / configuration usage error. """ from __future__ import annotations import argparse import csv import hashlib import json import math import re import sys from dataclasses import dataclass, field, asdict from datetime import date from pathlib import Path from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple # --------------------------------------------------------------------------- # # Constants # # --------------------------------------------------------------------------- # SAFE_WORDING = ( "The regression suite does not prove the theory true. It prevents the " "document from making claims that are unsupported by its own tables, " "methods, and declared evidence." ) # The seven outputs this engine must always produce. OUTPUT_FILES: Tuple[str, ...] = ( "validation_report.md", "validation_report.json", "failed_claims.csv", "pending_claims.csv", "release_gate_summary.csv", "residuals_by_sector.csv", "claim_class_audit.csv", ) # The six frozen input CSVs. INPUT_CSVS: Tuple[str, ...] = ( "claims_registry.csv", "evidence_register.csv", "theory_outputs.csv", "pdg_comparison_master.csv", "open_items_register.csv", "falsification_targets.csv", ) # Canonical claim classes (Test 2). Aliases (spaces / slashes) are normalised # onto these canonical tokens before validation. DEFAULT_ALLOWED_CLAIM_CLASSES: Tuple[str, ...] = ( "predicted", "imported", "fitted", "compatible_only", "pending", "out_of_scope", "anomaly", ) CLAIM_CLASS_ALIASES: Dict[str, str] = { "compatible": "compatible_only", "compatible only": "compatible_only", "compatible-only": "compatible_only", "out of scope": "out_of_scope", "out-of-scope": "out_of_scope", "outofscope": "out_of_scope", "anomaly/falsification target": "anomaly", "anomaly / falsification target": "anomaly", "falsification_target": "anomaly", "falsification target": "anomaly", } # Status vocabulary (Stage 3 protocol). Aliases normalised to canonical tokens. STATUS_ALIASES: Dict[str, str] = { "out of scope": "out_of_scope", "out-of-scope": "out_of_scope", "fail/tension": "fail", "fail / tension": "fail", "tension": "fail", "anomaly/falsification target": "anomaly", } # Claim classes whose numerical rows must carry a residual (Test 4). RESIDUAL_REQUIRED_CLASSES: Tuple[str, ...] = ("predicted", "imported", "computed") # Markers that flag a row as derived from a tentative PDG state (Test 6). TENTATIVE_MARKERS = ("tentative", "unconfirmed", "single-experiment", "needs confirmation") # Markers that flag a closure as "final" (so a tentative state needs a note). FINAL_CLOSURE_MARKERS = ("pass", "final", "closed", "closure") # Markers that flag a row as a broad resonance (Test 7). BROAD_RESONANCE_MARKERS = ("broad", "resonance", "f_0(500)", "sigma meson", "f0(500)", "kappa") # Words that suggest a value is being treated as an exact / stable mass. EXACT_MASS_MARKERS = ("exact", "stable", "pole-exact", "precise mass") # --------------------------------------------------------------------------- # # Configuration # # --------------------------------------------------------------------------- # @dataclass class ReleaseGate: """Release-gate switches (mirrors validation_config.yaml release_gate).""" allow_pending: bool = True require_units: bool = True require_claim_ids: bool = True require_evidence_ids: bool = True fail_on_unlabeled_fit: bool = True fail_on_prediction_fit_conflict: bool = True @dataclass class Config: """Parsed validation configuration.""" pdg_version: str = "PDG 2024" data_freeze_date: str = "" allowed_claim_classes: Tuple[str, ...] = DEFAULT_ALLOWED_CLAIM_CLASSES release_gate: ReleaseGate = field(default_factory=ReleaseGate) # Sectors that the release gate requires Stage-1/2/3 coverage for. required_sectors: Tuple[str, ...] = ( "leptons", "quarks", "gauge", "scalar", "mesons", "baryons", ) # Treat Test 7 (broad resonance) as hard FAIL (True) or WARN (False). fail_on_broad_resonance_exact: bool = False # --------------------------------------------------------------------------- # # Minimal flat-YAML parser (stdlib fallback when PyYAML is absent) # # --------------------------------------------------------------------------- # def _coerce_scalar(raw: str) -> Any: """Coerce a YAML/JSON-ish scalar string into a Python value.""" text = raw.strip() if text == "" or text.lower() in ("null", "~", "none"): return None low = text.lower() if low in ("true", "yes", "on"): return True if low in ("false", "no", "off"): return False # Strip matching quotes. if len(text) >= 2 and text[0] == text[-1] and text[0] in ("'", '"'): return text[1:-1] # Numbers. try: if re.fullmatch(r"[+-]?\d+", text): return int(text) return float(text) except ValueError: return text def _parse_flat_yaml(text: str) -> Dict[str, Any]: """Parse the small, flat subset of YAML used by validation_config.yaml. Supports: * top-level ``key: value`` scalars, * one level of nesting (``key:`` then indented ``child: value``), * block sequences (``key:`` then indented ``- item`` lines). This is intentionally tiny; it is only ever asked to parse *our own* config file, whose shape is fixed by the Stage-5 spec. """ root: Dict[str, Any] = {} lines = text.splitlines() i = 0 n = len(lines) def indent_of(s: str) -> int: return len(s) - len(s.lstrip(" ")) while i < n: raw = lines[i] stripped = raw.strip() i += 1 if not stripped or stripped.startswith("#"): continue if stripped.startswith("- "): # stray top-level sequence item; skip continue if ":" not in stripped: continue key, _, after = stripped.partition(":") key = key.strip() after = after.split("#", 1)[0].strip() if "#" in after else after.strip() base_indent = indent_of(raw) if after: # inline scalar value root[key] = _coerce_scalar(after) continue # Block: gather child lines at deeper indentation. seq: List[Any] = [] mapping: Dict[str, Any] = {} while i < n: child_raw = lines[i] child_stripped = child_raw.strip() if not child_stripped or child_stripped.startswith("#"): i += 1 continue if indent_of(child_raw) <= base_indent: break i += 1 if child_stripped.startswith("- "): seq.append(_coerce_scalar(child_stripped[2:])) elif ":" in child_stripped: ck, _, cv = child_stripped.partition(":") cv = cv.split("#", 1)[0].strip() mapping[ck.strip()] = _coerce_scalar(cv) if seq: root[key] = seq elif mapping: root[key] = mapping else: root[key] = None return root def load_config(config_path: Path) -> Config: """Load a validation config from YAML (PyYAML or fallback) or JSON mirror. Resolution order: 1. The given ``--config`` path, parsed via PyYAML if installed, else the flat-YAML fallback. 2. If the YAML path is absent, a sibling ``config.json`` mirror. 3. If neither exists, fail closed. """ raw: Optional[Dict[str, Any]] = None if config_path.exists(): text = config_path.read_text(encoding="utf-8") if config_path.suffix.lower() == ".json": raw = json.loads(text) else: try: import yaml # type: ignore raw = yaml.safe_load(text) except Exception: # PyYAML missing or unhappy -> stdlib fallback. raw = _parse_flat_yaml(text) else: mirror = config_path.with_name("config.json") if mirror.exists(): raw = json.loads(mirror.read_text(encoding="utf-8")) if not isinstance(raw, dict): raise SystemExit( f"[FAIL-CLOSED] config not found or unparseable: {config_path} " f"(and no config.json mirror). A validation config is required." ) gate_raw = raw.get("release_gate") or {} if not isinstance(gate_raw, dict): gate_raw = {} gate = ReleaseGate( allow_pending=bool(gate_raw.get("allow_pending", True)), require_units=bool(gate_raw.get("require_units", True)), require_claim_ids=bool(gate_raw.get("require_claim_ids", True)), require_evidence_ids=bool(gate_raw.get("require_evidence_ids", True)), fail_on_unlabeled_fit=bool(gate_raw.get("fail_on_unlabeled_fit", True)), fail_on_prediction_fit_conflict=bool( gate_raw.get("fail_on_prediction_fit_conflict", True) ), ) classes_raw = raw.get("allowed_claim_classes") or list(DEFAULT_ALLOWED_CLAIM_CLASSES) allowed = tuple(normalize_claim_class(str(c)) for c in classes_raw) sectors_raw = raw.get("required_sectors") if isinstance(sectors_raw, (list, tuple)) and sectors_raw: required_sectors = tuple(str(s).strip().lower() for s in sectors_raw) else: required_sectors = Config().required_sectors return Config( pdg_version=str(raw.get("pdg_version", "PDG 2024")), data_freeze_date=str(raw.get("data_freeze_date", "")), allowed_claim_classes=allowed, release_gate=gate, required_sectors=required_sectors, fail_on_broad_resonance_exact=bool(raw.get("fail_on_broad_resonance_exact", False)), ) # --------------------------------------------------------------------------- # # Normalisation helpers # # --------------------------------------------------------------------------- # def normalize_claim_class(value: str) -> str: key = (value or "").strip().lower() key = key.replace("/", " / ") # so 'a/b' aliases match key = re.sub(r"\s+", " ", key).strip() if key in CLAIM_CLASS_ALIASES: return CLAIM_CLASS_ALIASES[key] return key.replace(" ", "_") def normalize_status(value: str) -> str: key = (value or "").strip().lower() key = re.sub(r"\s+", " ", key).strip() if key in STATUS_ALIASES: return STATUS_ALIASES[key] return key.replace(" ", "_") _PENDING_TOKENS = ("pending", "tbd", "n/a", "na", "", "-", "—", "none", "to_be_determined") def is_pending_marker(value: str) -> bool: return (value or "").strip().lower() in _PENDING_TOKENS def parse_float(value: str) -> Optional[float]: """Parse a float from a CSV cell, tolerating PDG-style annotations. Strips +/- error attachments, scientific notation prefixes, commas and surrounding text. Returns None if no numeric value is present. """ if value is None: return None text = str(value).strip() if text == "" or is_pending_marker(text): return None # Take the leading numeric token, allowing sign, decimal, exponent. m = re.search(r"[+-]?\d*\.?\d+(?:[eE][+-]?\d+)?", text.replace(",", "")) if not m: return None try: return float(m.group(0)) except ValueError: return None # --------------------------------------------------------------------------- # # CSV loading # # --------------------------------------------------------------------------- # def load_csv(path: Path) -> List[Dict[str, str]]: """Load a CSV into a list of dict rows with normalised (lower) headers. Fails closed if the file is missing or has no header. """ if not path.exists(): raise SystemExit(f"[FAIL-CLOSED] required input file missing: {path}") with path.open("r", encoding="utf-8-sig", newline="") as fh: reader = csv.DictReader(fh) if reader.fieldnames is None: raise SystemExit(f"[FAIL-CLOSED] input file has no header: {path}") rows: List[Dict[str, str]] = [] for raw in reader: row = { (k or "").strip().lower(): (v.strip() if isinstance(v, str) else v) for k, v in raw.items() if k is not None } rows.append(row) return rows def get(row: Dict[str, str], *names: str) -> str: """First non-empty value among the given column aliases (lower-cased).""" for name in names: if name in row and row[name] not in (None, ""): return str(row[name]).strip() return "" def sector_of(row: Dict[str, str]) -> str: """Best-effort sector label for a row, from explicit sector or family.""" explicit = get(row, "sector", "category") if explicit: return explicit.strip().lower() family = get(row, "particle_family", "particle", "family").lower() table = { "lepton": "leptons", "neutrino": "leptons", "quark": "quarks", "meson": "mesons", "pion": "mesons", "kaon": "mesons", "baryon": "baryons", "proton": "baryons", "neutron": "baryons", "hyperon": "baryons", "gauge": "gauge", "boson": "gauge", "photon": "gauge", "higgs": "scalar", "scalar": "scalar", "resonance": "resonances", "exotic": "exotic", } for needle, sector in table.items(): if needle in family: return sector return family or "unclassified" # --------------------------------------------------------------------------- # # Findings model # # --------------------------------------------------------------------------- # @dataclass class Finding: """A single validation outcome for one row + one test.""" test_id: int test_name: str severity: str # "FAIL" | "WARN" | "PASS" file: str row_ref: str # human-readable row identity sector: str reason: str required_fix: str = "" def to_dict(self) -> Dict[str, Any]: return asdict(self) @dataclass class Results: findings: List[Finding] = field(default_factory=list) # residual rows for residuals_by_sector.csv residual_rows: List[Dict[str, Any]] = field(default_factory=list) # claim-class audit rows claim_audit_rows: List[Dict[str, Any]] = field(default_factory=list) # pending claims pending_rows: List[Dict[str, Any]] = field(default_factory=list) # release-gate per-sector rows gate_rows: List[Dict[str, Any]] = field(default_factory=list) def add(self, f: Finding) -> None: self.findings.append(f) @property def fails(self) -> List[Finding]: return [f for f in self.findings if f.severity == "FAIL"] @property def warns(self) -> List[Finding]: return [f for f in self.findings if f.severity == "WARN"] # --------------------------------------------------------------------------- # # The eight validation tests # # --------------------------------------------------------------------------- # # Required fields per the spec for theory_outputs rows (Test 1). THEORY_REQUIRED = [ ("particle/family", ("particle_family", "particle", "family")), ("observable", ("observable",)), ("value", ("value_or_pending", "value", "t_value")), ("units", ("units", "unit")), ("method", ("method",)), ("claim class", ("claim_class",)), ("status", ("status",)), ] def test1_required_fields(theory: List[Dict[str, str]], res: Results) -> None: """Test 1: every theory-output row carries the required fields. A pending value is acceptable for the 'value' field (it is an explicit pending marker) but every other field must be present, and units must be present even for pending rows whose observable is numeric. """ for idx, row in enumerate(theory, start=2): ref = row_ref(row, idx) sector = sector_of(row) for label, aliases in THEORY_REQUIRED: val = get(row, *aliases) if label == "value": # value may be a pending marker, but must be *some* token. if val == "" and not any( a in row for a in aliases ): res.add(Finding(1, "required_fields_present", "FAIL", "theory_outputs.csv", ref, sector, "missing value-or-pending marker", "add a numeric value or an explicit 'pending' marker")) continue if val == "": res.add(Finding(1, "required_fields_present", "FAIL", "theory_outputs.csv", ref, sector, f"missing required field: {label}", f"populate '{label}' for this row")) # Uncertainty availability: a numeric, non-pending value should carry a # 'source' citation (CITE EVERY CONSTANT). Missing source => FAIL. value = get(row, "value_or_pending", "value") if value and not is_pending_marker(value) and parse_float(value) is not None: if get(row, "source", "pdg_source") == "": res.add(Finding(1, "required_fields_present", "FAIL", "theory_outputs.csv", ref, sector, "numeric value lacks a 'source' citation", "add a 'source' column value (e.g. companion table ref)")) def test2_claim_class_validity( rows_by_file: Dict[str, List[Dict[str, str]]], cfg: Config, res: Results ) -> None: """Test 2: every claim_class is one of the allowed canonical classes.""" allowed = set(cfg.allowed_claim_classes) for fname in ("theory_outputs.csv", "pdg_comparison_master.csv", "claims_registry.csv", "falsification_targets.csv"): for idx, row in enumerate(rows_by_file.get(fname, []), start=2): cc_raw = get(row, "claim_class") if cc_raw == "": continue # absence handled by Test 1 / Test 8 where required cc = normalize_claim_class(cc_raw) if cc not in allowed: res.add(Finding(2, "claim_class_validity", "FAIL", fname, row_ref(row, idx), sector_of(row), f"invalid claim_class '{cc_raw}' (normalised '{cc}')", f"use one of: {', '.join(sorted(allowed))}")) def test3_prediction_fit_separation( theory: List[Dict[str, str]], cfg: Config, res: Results ) -> None: """Test 3: a 'predicted' observable must not also be a fit input. A row is flagged if claim_class normalises to 'predicted' yet the row carries a truthy fit-input marker (a 'fit_input' / 'is_fit_input' column, or the method/notes naming it as a fit input). """ if not cfg.release_gate.fail_on_prediction_fit_conflict: return for idx, row in enumerate(theory, start=2): cc = normalize_claim_class(get(row, "claim_class")) if cc != "predicted": continue fit_flag = get(row, "fit_input", "is_fit_input", "used_in_fit").lower() method = get(row, "method", "notes").lower() is_fit = fit_flag in ("true", "yes", "1", "y") or "fit input" in method or "fitted" in method if is_fit: res.add(Finding(3, "prediction_fit_separation", "FAIL", "theory_outputs.csv", row_ref(row, idx), sector_of(row), "claim_class='predicted' but observable is marked as a fit input", "reclassify as 'fitted' or remove the fit-input marker")) def test4_residual_calculation( comp: List[Dict[str, str]], res: Results ) -> None: """Test 4: residual Δ=T−O (and z where σ available) for numerical rows. Recomputes Δ and z from T,O,σ. Fails when a row whose claim_class is computed/predicted/imported carries numeric T and O but no residual can be formed, and verifies any stored delta/z agree with the recomputation. """ for idx, row in enumerate(comp, start=2): ref = row_ref(row, idx) sector = sector_of(row) cc = normalize_claim_class(get(row, "claim_class")) status = normalize_status(get(row, "status")) t = parse_float(get(row, "t_value", "theory_value", "t")) o = parse_float(get(row, "o_value", "pdg_value", "observed", "o")) st = parse_float(get(row, "t_sigma", "theory_uncertainty", "sigma_t")) so = parse_float(get(row, "o_sigma", "pdg_uncertainty", "sigma_o")) units = get(row, "units", "unit") numeric = t is not None and o is not None requires_residual = cc in RESIDUAL_REQUIRED_CLASSES and status not in ("pending", "out_of_scope") if requires_residual and not numeric: res.add(Finding(4, "residual_calculation", "FAIL", "pdg_comparison_master.csv", ref, sector, f"claim_class='{cc}' needs residual but T or O is non-numeric", "supply numeric T_value and O_value (PDG)")) continue if not numeric: continue if units == "" and status not in ("out_of_scope",): res.add(Finding(4, "residual_calculation", "FAIL", "pdg_comparison_master.csv", ref, sector, "numeric comparison row lacks units", "record explicit units for T and O")) delta = t - o z: Optional[float] = None if st is not None and so is not None: denom = math.sqrt(st * st + so * so) if denom > 0: z = delta / denom rel = abs(delta) / abs(o) if o != 0 else None # Cross-check any stored delta / z. stored_delta = parse_float(get(row, "delta", "residual")) if stored_delta is not None and abs(stored_delta - delta) > _tol(delta, stored_delta): res.add(Finding(4, "residual_calculation", "WARN", "pdg_comparison_master.csv", ref, sector, f"stored delta {stored_delta} != recomputed {delta:.6g}", "recompute delta = T - O")) stored_z = parse_float(get(row, "z", "normalized_residual", "z_score")) if z is not None and stored_z is not None and abs(stored_z - z) > 0.05 * max(1.0, abs(z)): res.add(Finding(4, "residual_calculation", "WARN", "pdg_comparison_master.csv", ref, sector, f"stored z {stored_z} != recomputed {z:.4g}", "recompute z = (T-O)/sqrt(sigT^2+sigO^2)")) res.residual_rows.append({ "particle": get(row, "particle", "particle_family"), "sector": sector, "observable": get(row, "observable"), "units": units, "T_value": t, "O_value": o, "T_sigma": st if st is not None else "", "O_sigma": so if so is not None else "", "delta": round(delta, 12), "z": round(z, 6) if z is not None else "", "rel_error": round(rel, 9) if rel is not None else "", "claim_class": cc, "status": status, "pdg_source": get(row, "pdg_source", "source"), }) def _tol(a: float, b: float) -> float: """Relative+absolute tolerance for float cross-checks.""" return 1e-6 + 1e-3 * max(abs(a), abs(b)) def test5_pending_not_pass( rows_by_file: Dict[str, List[Dict[str, str]]], res: Results ) -> None: """Test 5: a row with status 'pass' must not have claim_class 'pending'.""" for fname in ("theory_outputs.csv", "pdg_comparison_master.csv", "claims_registry.csv"): for idx, row in enumerate(rows_by_file.get(fname, []), start=2): cc = normalize_claim_class(get(row, "claim_class")) status = normalize_status(get(row, "status")) if status == "pass" and cc == "pending": res.add(Finding(5, "pending_cannot_be_pass", "FAIL", fname, row_ref(row, idx), sector_of(row), "status='pass' while claim_class='pending'", "downgrade status to 'pending' or reclassify the claim")) def test6_tentative_needs_note( rows_by_file: Dict[str, List[Dict[str, str]]], res: Results ) -> None: """Test 6: a tentative PDG state marked as final closure needs a note.""" for fname in ("theory_outputs.csv", "pdg_comparison_master.csv", "claims_registry.csv"): for idx, row in enumerate(rows_by_file.get(fname, []), start=2): blob = " ".join(str(v) for v in row.values()).lower() is_tentative = any(m in blob for m in TENTATIVE_MARKERS) if not is_tentative: continue status = normalize_status(get(row, "status")) statement = get(row, "statement", "notes", "observable").lower() final = status in ("pass",) or any(m in statement for m in FINAL_CLOSURE_MARKERS) note = get(row, "confidence_note", "confidence", "note", "notes") has_note = bool(note) and any(m in blob for m in ("confidence", "tentative", "caution")) if final and not has_note: res.add(Finding(6, "tentative_needs_confidence_note", "FAIL", fname, row_ref(row, idx), sector_of(row), "tentative PDG state marked as final closure without a confidence note", "add a confidence_note or downgrade status from final closure")) def test7_broad_resonance( rows_by_file: Dict[str, List[Dict[str, str]]], cfg: Config, res: Results ) -> None: """Test 7: broad resonances must not be treated as exact stable masses.""" severity = "FAIL" if cfg.fail_on_broad_resonance_exact else "WARN" for fname in ("theory_outputs.csv", "pdg_comparison_master.csv"): for idx, row in enumerate(rows_by_file.get(fname, []), start=2): blob = " ".join(str(v) for v in row.values()).lower() is_broad = any(m in blob for m in BROAD_RESONANCE_MARKERS) if not is_broad: continue treated_exact = any(m in blob for m in EXACT_MASS_MARKERS) # Zero theory/observed sigma on a broad resonance == treated exact. st = parse_float(get(row, "t_sigma", "theory_uncertainty")) so = parse_float(get(row, "o_sigma", "pdg_uncertainty")) zero_width_claim = (st == 0.0) or (so == 0.0) if treated_exact or zero_width_claim: res.add(Finding(7, "broad_resonance_not_exact", severity, fname, row_ref(row, idx), sector_of(row), "broad resonance treated as an exact stable-particle mass", "report a pole/width band, not an exact mass; carry uncertainty")) def test8_release_gate( rows_by_file: Dict[str, List[Dict[str, str]]], cfg: Config, res: Results, ) -> bool: """Test 8: release-gate coverage per required sector. A sector passes the gate when it has, across the loaded tables: * at least one theory-output row (Stage-1/2 status present), * at least one PDG-comparison row OR an out_of_scope declaration (Stage-3 status present), * claim IDs and evidence IDs where the config requires them, * reviewer-safe wording present in the registry statements (no bare 'matches PDG' / 'proves' style overclaim). Returns True if the overall release gate is satisfiable. """ gate = cfg.release_gate theory = rows_by_file.get("theory_outputs.csv", []) comp = rows_by_file.get("pdg_comparison_master.csv", []) claims = rows_by_file.get("claims_registry.csv", []) evidence = rows_by_file.get("evidence_register.csv", []) evidence_ids = {get(r, "evidence_id") for r in evidence if get(r, "evidence_id")} # Index by sector. def by_sector(rows: List[Dict[str, str]]) -> Dict[str, List[Dict[str, str]]]: out: Dict[str, List[Dict[str, str]]] = {} for r in rows: out.setdefault(sector_of(r), []).append(r) return out theory_s = by_sector(theory) comp_s = by_sector(comp) claims_s = by_sector(claims) overall_ok = True overclaim_re = re.compile(r"\b(matches pdg|proves|proven|guaranteed|exactly explains all)\b", re.I) for sector in cfg.required_sectors: stage1 = bool(theory_s.get(sector)) stage2 = stage1 # quantum-number/status presence rides on theory rows sector_claims = claims_s.get(sector, []) stage3 = bool(comp_s.get(sector)) or any( normalize_status(get(c, "status")) == "out_of_scope" for c in sector_claims ) # Evidence completeness. ev_ok = True if gate.require_evidence_ids: for c in sector_claims: eid = get(c, "evidence_id") if eid == "" or (evidence_ids and eid not in evidence_ids): ev_ok = False break if not sector_claims: ev_ok = False # Claim-id completeness. claim_id_ok = True if gate.require_claim_ids: claim_id_ok = bool(sector_claims) and all(get(c, "claim_id") for c in sector_claims) # Units completeness across that sector's theory rows. units_ok = True if gate.require_units: for r in theory_s.get(sector, []): val = get(r, "value_or_pending", "value") if val and not is_pending_marker(val) and parse_float(val) is not None: if get(r, "units", "unit") == "": units_ok = False break # Reviewer-safe wording. wording_ok = not any(overclaim_re.search(get(c, "statement")) for c in sector_claims) sector_ok = stage1 and stage2 and stage3 and ev_ok and claim_id_ok and units_ok and wording_ok missing: List[str] = [] if not stage1: missing.append("stage1_status") if not stage2: missing.append("stage2_status") if not stage3: missing.append("stage3_status") if not ev_ok: missing.append("evidence_completeness") if not claim_id_ok: missing.append("claim_ids") if not units_ok: missing.append("units") if not wording_ok: missing.append("reviewer_safe_wording") if not sector_ok: overall_ok = False res.add(Finding(8, "release_gate_coverage", "FAIL", "release_gate", f"sector={sector}", sector, "release-gate coverage incomplete: " + ", ".join(missing), "supply the missing stage statuses / evidence / wording")) res.gate_rows.append({ "sector": sector, "stage1_status": "present" if stage1 else "MISSING", "stage2_status": "present" if stage2 else "MISSING", "stage3_status": "present" if stage3 else "MISSING", "evidence_complete": "yes" if ev_ok else "NO", "claim_ids_present": "yes" if claim_id_ok else "NO", "units_present": "yes" if units_ok else "NO", "reviewer_safe_wording": "yes" if wording_ok else "NO", "gate": "PASS" if sector_ok else "FAIL", }) return overall_ok # --------------------------------------------------------------------------- # # Cross-cutting checks (referential integrity) + audit/pending population # # --------------------------------------------------------------------------- # def check_referential_integrity( rows_by_file: Dict[str, List[Dict[str, str]]], cfg: Config, res: Results ) -> None: """Missing evidence IDs / claim IDs / open-item & falsifier propagation.""" claims = rows_by_file.get("claims_registry.csv", []) evidence = rows_by_file.get("evidence_register.csv", []) falsifiers = rows_by_file.get("falsification_targets.csv", []) claim_ids = {get(c, "claim_id") for c in claims if get(c, "claim_id")} evidence_ids = {get(e, "evidence_id") for e in evidence if get(e, "evidence_id")} falsifier_ids = {get(f, "target_id", "falsifier_id") for f in falsifiers if get(f, "target_id", "falsifier_id")} for idx, c in enumerate(claims, start=2): ref = row_ref(c, idx) sector = sector_of(c) if cfg.release_gate.require_claim_ids and get(c, "claim_id") == "": res.add(Finding(1, "required_fields_present", "FAIL", "claims_registry.csv", ref, sector, "claim row lacks claim_id", "assign a claim_id")) eid = get(c, "evidence_id") if cfg.release_gate.require_evidence_ids and eid == "": res.add(Finding(8, "release_gate_coverage", "FAIL", "claims_registry.csv", ref, sector, "claim lacks evidence_id", "link an evidence_id")) elif eid and evidence_ids and eid not in evidence_ids: res.add(Finding(8, "release_gate_coverage", "FAIL", "claims_registry.csv", ref, sector, f"evidence_id '{eid}' not in evidence_register", "register the evidence or correct the id")) fid = get(c, "falsifier_id") # An anomaly/falsification-target claim must point at a real falsifier. if normalize_claim_class(get(c, "claim_class")) == "anomaly": if fid == "": res.add(Finding(8, "release_gate_coverage", "FAIL", "claims_registry.csv", ref, sector, "anomaly claim lacks falsifier_id", "link a falsification_targets row")) elif falsifier_ids and fid not in falsifier_ids: res.add(Finding(8, "release_gate_coverage", "FAIL", "claims_registry.csv", ref, sector, f"falsifier_id '{fid}' not in falsification_targets", "register the falsification target or correct the id")) # Evidence rows must cite an EXACT source (never "the relevant paper"). for idx, e in enumerate(evidence, start=2): src = get(e, "exact_source", "source") if src == "": res.add(Finding(1, "required_fields_present", "FAIL", "evidence_register.csv", row_ref(e, idx), sector_of(e), "evidence row lacks exact_source", "cite an exact source (file+section / PDG listing)")) elif re.search(r"\b(the relevant paper|see paper|tbd|somewhere)\b", src, re.I): res.add(Finding(1, "required_fields_present", "FAIL", "evidence_register.csv", row_ref(e, idx), sector_of(e), f"evidence source is non-exact: '{src}'", "replace with an exact file/section or PDG listing")) def populate_audit_and_pending( rows_by_file: Dict[str, List[Dict[str, str]]], res: Results ) -> None: """Build the claim-class-audit and pending-claims output rows.""" # claim_class_audit from claims_registry (falls back to theory_outputs). claims = rows_by_file.get("claims_registry.csv", []) source_rows = claims if claims else rows_by_file.get("theory_outputs.csv", []) for r in source_rows: cc = normalize_claim_class(get(r, "claim_class")) status = normalize_status(get(r, "status")) res.claim_audit_rows.append({ "claim_id": get(r, "claim_id", "particle_family", "particle"), "sector": sector_of(r), "claim_class": cc, "status": status, "computed": "yes" if cc in ("predicted", "imported", "fitted") else "no", "fitted": "yes" if cc == "fitted" else "no", "imported": "yes" if cc == "imported" else "no", "evidence_id": get(r, "evidence_id"), "statement": get(r, "statement", "observable"), }) if cc == "pending" or status == "pending": res.pending_rows.append({ "claim_id": get(r, "claim_id", "particle_family", "particle"), "sector": sector_of(r), "observable": get(r, "observable", "statement"), "claim_class": cc, "status": status, "blocking_input": get(r, "blocking_input", "gate_condition", "method"), }) # --------------------------------------------------------------------------- # # Row reference helper # # --------------------------------------------------------------------------- # def row_ref(row: Dict[str, str], line_no: int) -> str: parts = [ get(row, "claim_id"), get(row, "particle", "particle_family", "family"), get(row, "observable"), ] label = " / ".join(p for p in parts if p) return f"row{line_no}" + (f": {label}" if label else "") # --------------------------------------------------------------------------- # # Output writers # # --------------------------------------------------------------------------- # def write_csv(path: Path, rows: Sequence[Dict[str, Any]], columns: Sequence[str]) -> None: with path.open("w", encoding="utf-8", newline="") as fh: writer = csv.DictWriter(fh, fieldnames=list(columns)) writer.writeheader() for r in rows: writer.writerow({c: r.get(c, "") for c in columns}) def sector_summary( rows_by_file: Dict[str, List[Dict[str, str]]], res: Results, cfg: Config ) -> List[Dict[str, Any]]: """Per-sector tally: total / pass / partial / pending / fail.""" claims = rows_by_file.get("claims_registry.csv", []) source_rows = claims if claims else rows_by_file.get("theory_outputs.csv", []) fails_by_sector: Dict[str, int] = {} for f in res.fails: fails_by_sector[f.sector] = fails_by_sector.get(f.sector, 0) + 1 tally: Dict[str, Dict[str, int]] = {} for r in source_rows: sec = sector_of(r) status = normalize_status(get(r, "status")) d = tally.setdefault(sec, {"total": 0, "pass": 0, "partial": 0, "pending": 0, "fail": 0}) d["total"] += 1 if status in d: d[status] += 1 elif status == "out_of_scope": pass gate_pass = {g["sector"]: g["gate"] for g in res.gate_rows} out: List[Dict[str, Any]] = [] for sec in sorted(set(list(tally.keys()) + list(cfg.required_sectors))): d = tally.get(sec, {"total": 0, "pass": 0, "partial": 0, "pending": 0, "fail": 0}) sector_fails = fails_by_sector.get(sec, 0) release = gate_pass.get(sec, "n/a") out.append({ "sector": sec, "total_claims": d["total"], "pass": d["pass"], "partial": d["partial"], "pending": d["pending"], "fail": d["fail"] + sector_fails, "release_status": release, }) return out def write_outputs( out_dir: Path, rows_by_file: Dict[str, List[Dict[str, str]]], res: Results, cfg: Config, gate_ok: bool, ) -> List[str]: out_dir.mkdir(parents=True, exist_ok=True) written: List[str] = [] summary = sector_summary(rows_by_file, res, cfg) # failed_claims.csv failed_rows = [{ "test_id": f.test_id, "test_name": f.test_name, "file": f.file, "row": f.row_ref, "sector": f.sector, "reason": f.reason, "required_fix": f.required_fix, } for f in res.fails] p = out_dir / "failed_claims.csv" write_csv(p, failed_rows, ["test_id", "test_name", "file", "row", "sector", "reason", "required_fix"]) written.append(str(p)) # pending_claims.csv p = out_dir / "pending_claims.csv" write_csv(p, res.pending_rows, ["claim_id", "sector", "observable", "claim_class", "status", "blocking_input"]) written.append(str(p)) # release_gate_summary.csv p = out_dir / "release_gate_summary.csv" write_csv(p, res.gate_rows, ["sector", "stage1_status", "stage2_status", "stage3_status", "evidence_complete", "claim_ids_present", "units_present", "reviewer_safe_wording", "gate"]) written.append(str(p)) # residuals_by_sector.csv (sorted by sector then particle) res.residual_rows.sort(key=lambda r: (str(r.get("sector", "")), str(r.get("particle", "")))) p = out_dir / "residuals_by_sector.csv" write_csv(p, res.residual_rows, ["sector", "particle", "observable", "units", "T_value", "O_value", "T_sigma", "O_sigma", "delta", "z", "rel_error", "claim_class", "status", "pdg_source"]) written.append(str(p)) # claim_class_audit.csv p = out_dir / "claim_class_audit.csv" write_csv(p, res.claim_audit_rows, ["claim_id", "sector", "claim_class", "status", "computed", "fitted", "imported", "evidence_id", "statement"]) written.append(str(p)) # validation_report.json report = { "pdg_version": cfg.pdg_version, "data_freeze_date": cfg.data_freeze_date, "safe_wording": SAFE_WORDING, "totals": { "findings": len(res.findings), "fail": len(res.fails), "warn": len(res.warns), }, "release_gate_pass": gate_ok and not res.fails, "sector_summary": summary, "failed_claims": failed_rows, "warnings": [f.to_dict() for f in res.warns], "release_gate_summary": res.gate_rows, "pending_claims": res.pending_rows, "allowed_claim_classes": list(cfg.allowed_claim_classes), } p = out_dir / "validation_report.json" p.write_text(json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8") written.append(str(p)) # validation_report.md p = out_dir / "validation_report.md" p.write_text(render_markdown(report, res), encoding="utf-8") written.append(str(p)) return written def render_markdown(report: Dict[str, Any], res: Results) -> str: L: List[str] = [] L.append("# PDG Regression Validation Report") L.append("") L.append(f"- PDG version: `{report['pdg_version']}`") L.append(f"- Data freeze date: `{report['data_freeze_date']}`") L.append(f"- Findings: {report['totals']['findings']} " f"(FAIL={report['totals']['fail']}, WARN={report['totals']['warn']})") verdict = "PASS — releasable for external review" if report["release_gate_pass"] \ else "FAIL — not releasable; resolve findings below" L.append(f"- Release gate: **{verdict}**") L.append("") L.append("## Sector Summary") L.append("") L.append("| Sector | Total claims | Pass | Partial | Pending | Fail | Release status |") L.append("|---|---:|---:|---:|---:|---:|---|") for s in report["sector_summary"]: L.append(f"| {s['sector']} | {s['total_claims']} | {s['pass']} | {s['partial']} " f"| {s['pending']} | {s['fail']} | {s['release_status']} |") L.append("") L.append("## Failed Claims") L.append("") if report["failed_claims"]: L.append("| Failed claim | Reason | Required fix |") L.append("|---|---|---|") for f in report["failed_claims"]: L.append(f"| {f['file']} {f['row']} (test {f['test_id']} " f"{f['test_name']}) | {f['reason']} | {f['required_fix']} |") else: L.append("_No failed claims._") L.append("") L.append("## Warnings") L.append("") if report["warnings"]: for w in report["warnings"]: L.append(f"- (test {w['test_id']} {w['test_name']}) {w['file']} " f"{w['row_ref']}: {w['reason']}") else: L.append("_No warnings._") L.append("") L.append("## Pending Claims") L.append("") if report["pending_claims"]: for pc in report["pending_claims"]: L.append(f"- `{pc['claim_id']}` [{pc['sector']}] {pc['observable']} " f"(blocking: {pc['blocking_input']})") else: L.append("_No pending claims._") L.append("") L.append("## Disclaimer") L.append("") L.append("> " + report["safe_wording"]) L.append("") return "\n".join(L) # --------------------------------------------------------------------------- # # Orchestration # # --------------------------------------------------------------------------- # def run(data_dir: Path, out_dir: Path, config_path: Path) -> int: cfg = load_config(config_path) rows_by_file: Dict[str, List[Dict[str, str]]] = {} for name in INPUT_CSVS: rows_by_file[name] = load_csv(data_dir / name) res = Results() theory = rows_by_file["theory_outputs.csv"] comp = rows_by_file["pdg_comparison_master.csv"] # The eight validation tests. test1_required_fields(theory, res) test2_claim_class_validity(rows_by_file, cfg, res) test3_prediction_fit_separation(theory, cfg, res) test4_residual_calculation(comp, res) test5_pending_not_pass(rows_by_file, res) test6_tentative_needs_note(rows_by_file, res) test7_broad_resonance(rows_by_file, cfg, res) gate_ok = test8_release_gate(rows_by_file, cfg, res) # Cross-cutting referential integrity + audit/pending population. check_referential_integrity(rows_by_file, cfg, res) populate_audit_and_pending(rows_by_file, res) written = write_outputs(out_dir, rows_by_file, res, cfg, gate_ok) # Console summary. print(f"[pdg_regression] inputs: {data_dir} outputs: {out_dir}") print(f"[pdg_regression] findings: {len(res.findings)} " f"(FAIL={len(res.fails)}, WARN={len(res.warns)})") for fpath in written: print(f"[pdg_regression] wrote {fpath}") print(f"[pdg_regression] {SAFE_WORDING}") # FAIL-CLOSED: any FAIL finding, or an unsatisfiable release gate, => exit 1. if res.fails or not gate_ok: print(f"[pdg_regression] RESULT: FAIL ({len(res.fails)} failing rows; " f"release_gate_ok={gate_ok})", file=sys.stderr) return 1 print("[pdg_regression] RESULT: PASS") return 0 def build_arg_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( prog="pdg_regression.py", description="PDG regression suite for the Observed Particle Spectrum Closure companion. " + SAFE_WORDING, ) p.add_argument("--data", default="data", help="directory holding the 6 frozen input CSVs") p.add_argument("--out", default="out", help="directory for the 7 output files") p.add_argument("--config", default="validation_config.yaml", help="validation config (.yaml or .json; .yaml falls back to config.json mirror)") return p def main(argv: Optional[Sequence[str]] = None) -> int: args = build_arg_parser().parse_args(argv) base = Path.cwd() data_dir = (base / args.data).resolve() if not Path(args.data).is_absolute() else Path(args.data) out_dir = (base / args.out).resolve() if not Path(args.out).is_absolute() else Path(args.out) config_path = (base / args.config).resolve() if not Path(args.config).is_absolute() else Path(args.config) try: return run(data_dir, out_dir, config_path) except SystemExit as exc: # fail-closed messages raised as SystemExit if isinstance(exc.code, int): return exc.code print(str(exc.code), file=sys.stderr) return 1 if __name__ == "__main__": sys.exit(main())