# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca """Evaluate every run under results/runs/ (one predictions.parquet per run) against the truth table. Outputs (results/evaluation/): summary.csv (all arm × committee × aggregation × run_id), summary_main.csv (one row per arm, main configuration), contrasts.csv, coverage.csv, strata.csv, faithfulness.csv (+ faithfulness_turns.csv, faithfulness_committee.csv), pit.csv, records.parquet (per-record errors of the main configuration), manifest.json. LaTeX fragments + CSV twins in results/tables/: main_results, contrasts, coverage, strata, cost, faithfulness. Run ids starting with `_` (e.g. `_fake_demo`) are SKIPPED unless --include-fake is given. python3 scripts/20_evaluate.py # real data: data/processed/analytic.parquet + splits.parquet python3 scripts/20_evaluate.py --synthetic --include-fake --B 500 # smoke run on the fake demo """ from __future__ import annotations import argparse import json import sys from datetime import datetime, timezone from pathlib import Path import numpy as np import pandas as pd import polars as pl from agentic_valuation.config import PROCESSED, RESULTS, SEED, SYNTHETIC from agentic_valuation.evaluation import (audit_run, compare_arms, coverage_by, pit_from_quantiles, pit_histogram, reliability, stratified_metrics, summarise_by, write_contrasts, write_cost, write_coverage, write_faithfulness, write_main_results, write_strata) from agentic_valuation.evaluation._common import join_truth, prepare_truth, to_pandas from agentic_valuation.evaluation.faithfulness import summarise_faithfulness from agentic_valuation.evaluation.metrics import PRIMARY_METRIC, per_record from agentic_valuation.evaluation.stratify import add_strata # Pre-registered primary family (docs/preregistration.md H1–H4): (A, B) with Δ = metric(B) − metric(A) PRIMARY_PAIRS = [("D_avm", "G_debate_r2"), ("F_independent", "G_debate_r2"), ("J_cost_matched", "G_debate_r2"), ("I_placebo", "G_debate_r2")] PRIMARY_METRICS = [PRIMARY_METRIC] TAIL_PAIRS = [("I_placebo", "G_debate_r2")] # H4b: P95 |e| SECONDARY_PAIRS = [("E_single_llm", "G_debate_r2"), ("H_debate_skeptic", "G_debate_r2"), ("A_hedonic", "G_debate_r2"), ("C_gbm", "G_debate_r2"), ("D_avm", "F_independent"), ("E_single_llm", "F_independent"), ("G_debate_r1", "G_debate_r2"), ("G_debate_r2", "G_debate_r4")] SECONDARY_METRICS = [PRIMARY_METRIC, "mdape", "p95_abs_log_error", "severe_rate", "coverage_90", "pinball_log"] def collect_runs(runs_dir: Path, include_fake: bool) -> tuple[pd.DataFrame, list[str]]: frames, used = [], [] for d in sorted(p for p in runs_dir.iterdir() if p.is_dir()): if d.name.startswith("_") and not include_fake: continue f = d / "predictions.parquet" if not f.exists(): continue df = pl.read_parquet(f).to_pandas() if "run_id" not in df.columns: df["run_id"] = d.name if not include_fake: df = df[~df["run_id"].astype(str).str.startswith("_")] if df.empty: continue frames.append(df) used.append(d.name) if not frames: return pd.DataFrame(), used return pd.concat(frames, ignore_index=True), used def load_truth(args: argparse.Namespace) -> tuple[pd.DataFrame, str]: if args.synthetic: path = SYNTHETIC / "synthetic_transactions.parquet" t = prepare_truth(pl.read_parquet(path)) return t, f"SYNTHETIC {path}" path = Path(args.truth) t = prepare_truth(pl.read_parquet(path)) sp = Path(args.splits) if sp.exists(): s = pl.read_parquet(sp).to_pandas() t = t.merge(s[["tx_id", "split"]], on="tx_id", how="left") return t, str(path) def select_main(summary: pd.DataFrame, committee: str, aggregation: str) -> pd.DataFrame: """One row per arm, in order of preference: (committee, aggregation) → (any committee, aggregation) → baseline config (none/model) → first available. The choice is recorded in `main_selection`; ties broken by the last run_id.""" rows = [] for arm, g in summary.groupby("arm", sort=True): candidates = [ ("preferred", g[(g["committee"] == committee) & (g["aggregation"] == aggregation)]), ("aggregation_any_committee", g[g["aggregation"] == aggregation]), ("baseline", g[(g["committee"] == "none") & (g["aggregation"] == "model")]), ("fallback_first", g), ] label, pick = next((lab, c) for lab, c in candidates if not c.empty) row = pick.sort_values(["committee", "run_id"]).iloc[-1].to_dict() row["main_selection"] = label rows.append(row) return pd.DataFrame(rows) def main_predictions(preds: pd.DataFrame, main: pd.DataFrame) -> pd.DataFrame: keys = main[["arm", "committee", "aggregation", "run_id"]] return preds.merge(keys, on=["arm", "committee", "aggregation", "run_id"], how="inner") def run_contrasts(preds_main: pd.DataFrame, truth: pd.DataFrame, pairs, metrics, B, seed, family: str, holm_family: bool) -> pd.DataFrame: arms = set(preds_main["arm"].unique()) ok_pairs = [(a, b) for a, b in pairs if a in arms and b in arms] if not ok_pairs: return pd.DataFrame() c = compare_arms(preds_main, truth, ok_pairs, metric=metrics, B=B, seed=seed, holm_family=holm_family) c.insert(0, "family", family) return c def main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--runs-dir", default=str(RESULTS / "runs")) ap.add_argument("--truth", default=str(PROCESSED / "analytic.parquet")) ap.add_argument("--splits", default=str(PROCESSED / "splits.parquet")) ap.add_argument("--synthetic", action="store_true", help="use data/synthetic as truth (tests / smoke only)") ap.add_argument("--split", default="test", help="partition to evaluate (rows of predictions with this split)") ap.add_argument("--include-fake", action="store_true", help="also collect run ids starting with `_`") ap.add_argument("--B", type=int, default=2000) ap.add_argument("--seed", type=int, default=SEED) ap.add_argument("--main-committee", default="mixed_random") ap.add_argument("--main-aggregation", default="median") ap.add_argument("--out", default=str(RESULTS / "evaluation")) ap.add_argument("--tables", default=str(RESULTS / "tables")) ap.add_argument("--min-cell", type=int, default=30) args = ap.parse_args(argv) out, tables = Path(args.out), Path(args.tables) out.mkdir(parents=True, exist_ok=True) tables.mkdir(parents=True, exist_ok=True) runs_dir = Path(args.runs_dir) preds, used = collect_runs(runs_dir, args.include_fake) if preds.empty: print(f"no predictions found under {runs_dir} (fake runs skipped={not args.include_fake}); nothing to evaluate", file=sys.stderr) return 1 preds = preds[preds["split"].astype(str) == args.split].copy() if preds.empty: print(f"no predictions for split={args.split!r}", file=sys.stderr) return 1 truth, truth_src = load_truth(args) if "split" in truth.columns: truth = truth[truth["split"].astype(str) == args.split] truth = add_strata(truth) n_before = len(preds) preds = preds[preds["tx_id"].astype(str).isin(set(truth["tx_id"]))] print(f"runs: {used} | prediction rows: {n_before} (matched to truth: {len(preds)}) | truth: {truth_src}") # 1. summaries summary = summarise_by(preds, truth, by=["arm", "committee", "aggregation", "run_id"]) summary.to_csv(out / "summary.csv", index=False) main_sel = select_main(summary, args.main_committee, args.main_aggregation) main_sel.to_csv(out / "summary_main.csv", index=False) pm = main_predictions(preds, main_sel) records = per_record(pm, truth) records = add_strata(records) records.to_parquet(out / "records.parquet", index=False) # 2. contrasts parts = [ run_contrasts(pm, truth, PRIMARY_PAIRS, PRIMARY_METRICS, args.B, args.seed, "primary", True), run_contrasts(pm, truth, TAIL_PAIRS, ["p95_abs_log_error"], args.B, args.seed, "primary_tail", False), run_contrasts(pm, truth, SECONDARY_PAIRS, SECONDARY_METRICS, args.B, args.seed, "secondary", False), ] contrasts = pd.concat([p for p in parts if not p.empty], ignore_index=True) if any(not p.empty for p in parts) else pd.DataFrame() contrasts.to_csv(out / "contrasts.csv", index=False) # 3. coverage / calibration cov_rows, pit_rows = [], [] for arm, g in records.groupby("arm", sort=True): if not g["has_interval"].any(): continue r = reliability(g) r.insert(0, "arm", arm) r.insert(1, "group", "all") r.insert(2, "level_value", "all") cov_rows.append(r) for strat in ("region_group", "property_class", "price_tercile"): if strat in g.columns: cb = coverage_by(g, [strat], min_n=args.min_cell) cb = cb.rename(columns={strat: "level_value", "coverage": "empirical"}) cb.insert(0, "arm", arm) cb.insert(1, "group", strat) cb["kind"] = "interval" cov_rows.append(cb[["arm", "group", "level_value", "kind", "nominal", "empirical", "n", "suppressed"]]) ph = pit_histogram(pit_from_quantiles(g)) ph.insert(0, "arm", arm) pit_rows.append(ph) coverage = pd.concat(cov_rows, ignore_index=True) if cov_rows else pd.DataFrame() coverage.to_csv(out / "coverage.csv", index=False) (pd.concat(pit_rows, ignore_index=True) if pit_rows else pd.DataFrame()).to_csv(out / "pit.csv", index=False) # 4. strata strata = stratified_metrics(pm, truth, min_n=args.min_cell) strata.to_csv(out / "strata.csv", index=False) # 5. faithfulness turns, comms = [], [] for name in used: t, c = audit_run(runs_dir / name) if not t.empty: turns.append(t) if not c.empty: comms.append(c) turns_df = pd.concat(turns, ignore_index=True) if turns else pd.DataFrame() comms_df = pd.concat(comms, ignore_index=True) if comms else pd.DataFrame() faith = summarise_faithfulness(turns_df, comms_df) if not turns_df.empty else pd.DataFrame() turns_df.to_csv(out / "faithfulness_turns.csv", index=False) comms_df.to_csv(out / "faithfulness_committee.csv", index=False) faith.to_csv(out / "faithfulness.csv", index=False) if not comms_df.empty: rounds = comms_df.groupby(["arm", "round"], as_index=False).agg(iqr_log=("iqr_log", "mean"), convergence=("convergence", "mean"), anchoring_index=("anchoring_index", "mean"), share_toward_median=("share_toward_median", "mean"), n=("tx_id", "nunique")) rounds.to_csv(out / "committee_rounds.csv", index=False) # 6. LaTeX fragments write_main_results(main_sel, tables / "main_results.tex") write_coverage(main_sel, tables / "coverage.tex") write_cost(main_sel, tables / "cost.tex") write_contrasts(contrasts if not contrasts.empty else pd.DataFrame(columns=["arm_a", "arm_b", "metric", "n", "value_a", "value_b", "diff", "ci_lo", "ci_hi", "p_boot", "p_holm", "dm_p"]), tables / "contrasts.tex") write_strata(strata, tables / "strata.tex") write_faithfulness(faith, tables / "faithfulness.tex") manifest = { "generated_at": datetime.now(timezone.utc).isoformat(timespec="seconds"), "script": "scripts/20_evaluate.py", "runs": used, "truth": truth_src, "split": args.split, "n_prediction_rows": int(len(preds)), "n_truth": int(len(truth)), "B": args.B, "seed": args.seed, "main_committee": args.main_committee, "main_aggregation": args.main_aggregation, "min_cell": args.min_cell, "arms": sorted(preds["arm"].unique().tolist()), "fake_included": args.include_fake, "note": "SYNTHETIC / FAKE — not an empirical result" if (args.synthetic or args.include_fake) else "", } (out / "manifest.json").write_text(json.dumps(manifest, indent=1, ensure_ascii=False), encoding="utf-8") print(main_sel[["arm", "committee", "aggregation", "n", PRIMARY_METRIC, "coverage_90", "main_selection"]].to_string(index=False)) if not contrasts.empty: print(contrasts[["family", "arm_a", "arm_b", "metric", "diff", "ci_lo", "ci_hi", "p_boot", "p_holm", "dm_p"]].to_string(index=False)) print(f"wrote {out} and {tables}") return 0 if __name__ == "__main__": raise SystemExit(main())