1# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca2"""Tests of the evaluation layer: hand-computed metrics, bootstrap coverage, Holm, interval coverage,3conformal rescaling, faithfulness audit on a Pydantic-built transcript, LaTeX fragment hygiene."""4from __future__ import annotations56import math7from datetime import date, timedelta8from pathlib import Path910import numpy as np11import pandas as pd12import pytest13from scipy import stats1415from agentic_valuation.evaluation import (apply_conformal, block_bootstrap_contrast, compare_arms, conformalise, coverage_by,16 diebold_mariano, holm, per_record, pit_from_quantiles, reliability, stratified_metrics,17 summarise, summarise_by)18from agentic_valuation.evaluation.faithfulness import audit_estimate, audit_run, audit_transcript, summarise_faithfulness19from agentic_valuation.evaluation.tables import arm_pretty, escape, num, write_contrasts, write_coverage, write_main_results, write_strata20from agentic_valuation.schemas import (AgentEstimate, AgentTurn, Adjustment, Comparable, DebateTranscript, EvidencePacket,21 MarketContext, SubjectProperty)2223LOG125 = math.log(1.25)242526# ----------------------------------------------------------------------------------------------- helpers2728def _truth(n: int, seed: int = 1) -> pd.DataFrame:29 rng = np.random.default_rng(seed)30 d0 = date(2025, 4, 1)31 return pd.DataFrame({32 "tx_id": [f"T{i:05d}" for i in range(n)],33 "price": np.exp(rng.normal(math.log(400_000), 0.4, n)).round(-3),34 "date": [d0 + timedelta(days=int(x)) for x in rng.integers(0, 270, n)],35 "mun_code": rng.choice([f"M{k}" for k in range(12)], n),36 "region_name": rng.choice(["Montréal", "Capitale-Nationale", "Outaouais", "Estrie"], n),37 "property_class": rng.choice(["house", "condo", "plex"], n),38 "urban_class": rng.choice(["metro", "suburban", "rural"], n),39 "year_built": rng.integers(1900, 2024, n),40 "floor_area_m2": rng.uniform(60, 250, n),41 })424344def _preds(truth: pd.DataFrame, arm: str, sigma: float, bias: float = 0.0, seed: int = 2, intervals: bool = True,45 shared: np.ndarray | None = None, iscale: float = 1.0) -> pd.DataFrame:46 rng = np.random.default_rng(seed)47 e = rng.normal(bias, sigma, len(truth)) + (shared if shared is not None else 0.0)48 point = truth["price"].to_numpy() * np.exp(e)49 z = stats.norm.ppf([0.05, 0.25, 0.5, 0.75, 0.95])50 s = sigma * iscale51 q = np.exp(np.log(point)[:, None] + s * z[None, :]) if intervals else np.full((len(truth), 5), np.nan)52 return pd.DataFrame({"tx_id": truth["tx_id"], "split": "test", "arm": arm, "committee": "mixed_random", "aggregation": "median",53 "point": point, "q05": q[:, 0], "q25": q[:, 1], "q50": q[:, 2], "q75": q[:, 3], "q95": q[:, 4],54 "cost_usd": 0.5, "latency_s": 10.0, "tokens": 1000, "run_id": "r1"})555657# ----------------------------------------------------------------------------------------------- metrics5859def test_metrics_hand_computed():60 truth = pd.DataFrame({"tx_id": ["a", "b", "c", "d"], "price": [100.0, 200.0, 300.0, 400.0]})61 preds = pd.DataFrame({62 "tx_id": ["a", "b", "c", "d"], "arm": "X", "point": [110.0, 180.0, 300.0, 600.0],63 "q05": [90.0, 150.0, 250.0, 500.0], "q25": [100.0, 170.0, 280.0, 550.0], "q50": [110.0, 180.0, 300.0, 600.0],64 "q75": [120.0, 190.0, 320.0, 650.0], "q95": [130.0, 210.0, 350.0, 700.0], "cost_usd": [1, 1, 1, 1],65 })66 e = np.log(np.array([110, 180, 300, 600]) / np.array([100, 200, 300, 400]))67 s = summarise(preds, truth)68 assert s["n"] == 469 assert s["mean_abs_log_error"] == pytest.approx(np.mean(np.abs(e)))70 assert s["mae_cad"] == pytest.approx((10 + 20 + 0 + 200) / 4)71 assert s["mdape"] == pytest.approx(np.median([0.1, 0.1, 0.0, 0.5]))72 assert s["rmse_log"] == pytest.approx(np.sqrt(np.mean(e**2)))73 assert s["bias_log"] == pytest.approx(np.mean(e))74 assert s["severe_rate"] == pytest.approx(0.25) # only d: log(1.5) > log(1.25)75 assert s["over_rate"] == pytest.approx(0.25) # d: +50 %76 assert s["under_rate"] == pytest.approx(0.0) # b: −10 % only77 # raw coverage: 50 % → price in [q25,q75]: a yes (100∈[100,120]), b no (200∉[170,190]), c yes, d no → 0.578 assert s["coverage_50"] == pytest.approx(0.5)79 # 90 %: a yes, b no (200∉[150,210]? yes it is), c yes (300∈[250,350]), d no (400∉[500,700]) → 0.7580 assert s["coverage_90"] == pytest.approx(0.75)81 assert s["mean_log_width_90"] == pytest.approx(np.mean(np.log([130 / 90, 210 / 150, 350 / 250, 700 / 500])))82 assert s["cost_usd_total"] == pytest.approx(4.0)83 # pinball for record c (price 300): q=[250,280,300,320,350] → τ·(y−q) for y≥q, (τ−1)(y−q) otherwise, in log space84 y = math.log(300)85 lq = np.log([250, 280, 300, 320, 350])86 taus = [0.05, 0.25, 0.5, 0.75, 0.95]87 pb = np.mean([t * (y - q) if y >= q else (t - 1) * (y - q) for t, q in zip(taus, lq)])88 rec = per_record(preds, truth)89 assert rec.loc[rec["tx_id"] == "c", "pinball_log"].iloc[0] == pytest.approx(pb)909192def test_interval_metrics_nan_without_quantiles():93 truth = _truth(50)94 p = _preds(truth, "A", 0.1, intervals=False)95 s = summarise(p, truth)96 assert np.isnan(s["coverage_90"]) and np.isnan(s["pinball_log"]) and np.isnan(s["crps_log"])97 assert np.isfinite(s["mean_abs_log_error"])9899100def test_normal_fit_coverage_matches_raw_on_normal_quantiles():101 truth = _truth(4000)102 p = _preds(truth, "D", 0.15)103 s = summarise(p, truth)104 # quantiles are exactly log-normal → fitted 90 % band coincides with [q05,q95] and coverage ≈ nominal105 assert s["coverage_90_normal_fit"] == pytest.approx(s["coverage_90"], abs=1e-9)106 assert abs(s["coverage_90"] - 0.90) < 0.03107 assert abs(s["coverage_50"] - 0.50) < 0.04108 assert abs(s["coverage_80"] - 0.80) < 0.03109 assert abs(s["coverage_95"] - 0.95) < 0.03110111112def test_summarise_by_groups():113 truth = _truth(100)114 p = pd.concat([_preds(truth, "A", 0.1), _preds(truth, "B", 0.2, seed=3)])115 out = summarise_by(p, truth)116 assert set(out["arm"]) == {"A", "B"} and (out["n"] == 100).all()117118119# ----------------------------------------------------------------------------------------------- inference120121def test_bootstrap_ci_covers_true_difference():122 """Simulation: arm B has |e| smaller by a known amount on average; CI should cover the population difference."""123 truth = _truth(600, seed=11)124 rng = np.random.default_rng(5)125 shared = rng.normal(0, 0.05, len(truth))126 a = _preds(truth, "A", 0.16, shared=shared, seed=21)127 b = _preds(truth, "B", 0.10, shared=shared, seed=22)128 # population mean |e| for N(0, s²+0.05²): s_tot·sqrt(2/π)129 true_diff = (math.hypot(0.10, 0.05) - math.hypot(0.16, 0.05)) * math.sqrt(2 / math.pi)130 c = compare_arms(pd.concat([a, b]), truth, [("A", "B")], B=400, seed=1)131 row = c.iloc[0]132 assert row["ci_lo"] < true_diff < row["ci_hi"]133 assert row["diff"] < 0 and row["p_boot"] < 0.05 and row["dm_p"] < 0.05134 assert row["n_blocks"] > 10135136137def test_bootstrap_no_difference_under_null():138 truth = _truth(300, seed=3)139 a = _preds(truth, "A", 0.15, seed=1)140 b = a.assign(arm="B")141 rec_a, rec_b = per_record(a, truth), per_record(b, truth)142 r = block_bootstrap_contrast(rec_a, rec_b, B=200, seed=0)143 assert r["diff"] == 0.0 and r["ci_lo"] <= 0 <= r["ci_hi"]144145146def test_holm_matches_statsmodels_and_is_monotone():147 from statsmodels.stats.multitest import multipletests148 p = [0.01, 0.04, 0.03, 0.20, 0.005]149 ours = holm(p)150 ref = multipletests(p, method="holm")[1]151 assert np.allclose(ours, ref)152 assert holm([0.5])[0] == 0.5153 assert np.isnan(holm([0.1, np.nan])[1]) and holm([0.1, np.nan])[0] == 0.1154155156def test_diebold_mariano_detects_difference_and_respects_order():157 rng = np.random.default_rng(0)158 la = np.abs(rng.normal(0, 0.2, 500))159 lb = np.abs(rng.normal(0, 0.1, 500))160 r = diebold_mariano(la, lb)161 assert r["dm_stat"] < 0 and r["dm_p"] < 0.01162 same = diebold_mariano(la, la)163 assert np.isnan(same["dm_stat"]) or abs(same["dm_mean_diff"]) < 1e-12164165166def test_compare_arms_requires_unambiguous_spec():167 truth = _truth(40)168 a = _preds(truth, "A", 0.1)169 a2 = a.assign(aggregation="mean")170 with pytest.raises(ValueError, match="ambiguous"):171 compare_arms(pd.concat([a, a2, _preds(truth, "B", 0.1)]), truth, [("A", "B")], B=20)172 c = compare_arms(pd.concat([a, a2, _preds(truth, "B", 0.1)]), truth, [({"arm": "A", "aggregation": "mean"}, "B")], B=20)173 assert len(c) == 1174175176# ----------------------------------------------------------------------------------------------- calibration177178def test_coverage_by_suppresses_small_cells():179 truth = _truth(200)180 rec = per_record(_preds(truth, "D", 0.15), truth)181 rec["grp"] = ["big"] * 180 + ["tiny"] * 20182 cb = coverage_by(rec, ["grp"], min_n=30)183 assert cb.loc[cb["grp"] == "tiny", "suppressed"].all()184 assert cb.loc[cb["grp"] == "tiny", "coverage"].isna().all()185 assert cb.loc[cb["grp"] == "big", "coverage"].notna().all()186187188def test_reliability_and_pit_uniform_for_calibrated_forecast():189 truth = _truth(3000, seed=9)190 rec = per_record(_preds(truth, "D", 0.15, seed=4), truth)191 r = reliability(rec)192 q = r[r["kind"] == "quantile"]193 assert (q["gap"].abs() < 0.03).all()194 pit = pit_from_quantiles(rec)195 assert np.isfinite(pit).all() and ((pit > 0) & (pit < 1)).all()196 # Kolmogorov–Smirnov against U(0,1) should not reject at 1 %197 assert stats.kstest(pit, "uniform").pvalue > 0.01198199200def test_conformal_scale_achieves_nominal_on_validation_and_improves_test():201 truth_v, truth_t = _truth(1000, seed=31), _truth(1000, seed=32)202 # too-narrow intervals (half the true spread) → factor must be ≈ 2 and coverage ≈ 0.90 on validation203 pv = _preds(truth_v, "D", 0.15, seed=41, iscale=0.5)204 pt = _preds(truth_t, "D", 0.15, seed=42, iscale=0.5)205 sc = conformalise(pv, truth_v)206 assert 1.6 < sc.scale_90 < 2.5207 assert sc.achieved_valid_90 >= 0.90208 adj_v = apply_conformal(pv, sc)209 assert summarise(adj_v, truth_v)["coverage_90"] >= 0.90 - 1e-9210 before, after = summarise(pt, truth_t)["coverage_90"], summarise(adj_v := apply_conformal(pt, sc), truth_t)["coverage_90"]211 assert before < 0.75 and abs(after - 0.90) < 0.04212 # point and q50 untouched, ordering preserved213 assert np.allclose(adj_v["point"], pt["point"]) and np.allclose(adj_v["q50"], pt["q50"])214 assert (adj_v["q05"] <= adj_v["q25"]).all() and (adj_v["q75"] <= adj_v["q95"]).all()215 # too-wide intervals → factor < 1 (narrowing)216 sc2 = conformalise(_preds(truth_v, "D", 0.15, seed=43, iscale=2.0), truth_v)217 assert sc2.scale_90 < 0.7218219220# ----------------------------------------------------------------------------------------------- strata221222def test_stratified_metrics_long_table_and_suppression():223 truth = _truth(400)224 st = stratified_metrics(_preds(truth, "A", 0.1), truth, strata=["region_group", "year_built_decade"], metrics=["mean_abs_log_error"])225 assert set(st.columns) >= {"arm", "stratum", "level", "n", "metric", "value", "suppressed"}226 assert set(st["stratum"]) == {"region_group", "year_built_decade"}227 small = st[st["n"] < 30]228 assert small["suppressed"].all() and small["value"].isna().all()229 big = st[st["n"] >= 30]230 assert (~big["suppressed"]).all() and big["value"].notna().all()231 assert set(st[st["stratum"] == "region_group"]["level"]) <= {"Montréal", "Québec", "Outaouais", "other"}232233234# ----------------------------------------------------------------------------------------------- faithfulness235236def _packet(tx_id: str = "T1") -> EvidencePacket:237 d = date(2025, 6, 1)238 comps = [Comparable(evidence_id=f"C{k:02d}", sale_date=d - timedelta(days=60 * k), price=p, distance_km=0.5 * k, property_class="house",239 municipality="Gatineau", months_before_valuation=2.0 * k) for k, p in ((1, 400_000.0), (2, 440_000.0), (3, 380_000.0))]240 subj = SubjectProperty(tx_id=tx_id, valuation_date=d, information_cutoff=d - timedelta(days=1), property_class="house", region="Outaouais",241 municipality="Gatineau", lat=45.477, lng=-75.701)242 return EvidencePacket(packet_id="pk1", subject=subj, comparables=comps, market=MarketContext(as_of=d - timedelta(days=1)))243244245def _estimate(p: float, cited, adj, words: int = 50, assumptions=()) -> AgentEstimate:246 return AgentEstimate(point_estimate=p, q05=p * 0.85, q25=p * 0.95, q50=p, q75=p * 1.05, q95=p * 1.15, evidence_ids_used=cited,247 adjustments=adj, key_assumptions=list(assumptions), reasoning_summary=" ".join(["w"] * words))248249250def test_audit_estimate_hand_built():251 pk = _packet()252 good = _estimate(410_000, ["P", "C01", "C02", "M1"], [Adjustment(comparable_id="C01", adjusted_price=410_000, reason="r"),253 Adjustment(comparable_id="C02", adjusted_price=420_000, reason="r")])254 r = audit_estimate(good, pk)255 assert r["cited_share_valid"] == 1.0 and r["adj_share_within_40"] == 1.0 and r["point_in_hull"] is True256 assert r["quantile_sane"] is True and r["summary_words"] == 50 and r["summary_within_limit"]257 bad = _estimate(700_000, ["P", "C01", "C77"], [Adjustment(comparable_id="C01", adjusted_price=200_000, reason="r"),258 Adjustment(comparable_id="C03", adjusted_price=390_000, reason="r")], words=150, assumptions=["unique view"])259 r = audit_estimate(bad, pk)260 assert r["cited_share_valid"] == pytest.approx(2 / 3) and r["n_cited_invalid"] == 1261 assert r["adj_share_within_40"] == pytest.approx(0.5) # 200k is −50 % of 400k262 assert r["point_in_hull"] is False and r["hull_deviation_justified"] is True # 700k > 390k × 1.25263 assert r["hull_log_deviation"] == pytest.approx(math.log(700_000 / (390_000 * 1.25)))264 assert not r["summary_within_limit"]265 # without a packet the id checks are NaN266 assert np.isnan(audit_estimate(good, None)["cited_share_valid"])267268269def test_audit_transcript_committee_dynamics(tmp_path: Path):270 pk = _packet()271 roles = ["r1", "r2", "r3", "r4"]272 r0 = {"r1": 380_000.0, "r2": 400_000.0, "r3": 420_000.0, "r4": 500_000.0}273 med0 = float(np.median(np.log(list(r0.values()))))274 turns = []275 for role in roles:276 turns.append(AgentTurn(role=role, round=0, kind="estimate", model="m", provider="mock", estimate=_estimate(r0[role], ["P", "C01"], [])))277 # round 1: everyone moves halfway (in log) toward the round-0 median → anchoring index 1, share toward median 1278 r1 = {role: float(np.exp(np.log(v) + 0.5 * (med0 - np.log(v)))) for role, v in r0.items()}279 for role in roles:280 turns.append(AgentTurn(role=role, round=1, kind="revision", model="m", provider="mock", estimate=_estimate(r1[role], ["P", "C01"], [])))281 tr = DebateTranscript(tx_id="T1", arm="G_debate_r1", committee="all_anthropic", protocol="structured_debate", rounds=1,282 packet_hash=pk.content_hash(), turns=turns, final_estimates=r1)283 t, c = audit_transcript(tr, pk)284 assert len(t) == 8 and t["packet_hash_match"].all()285 c1 = c[c["round"] == 1].iloc[0]286 assert c1["anchoring_index"] == pytest.approx(1.0)287 assert c1["share_toward_median"] == 1.0288 assert c1["convergence"] == pytest.approx(0.5, abs=1e-6)289 # round-trip through files + audit_run + summary290 run = tmp_path / "run_x"291 (run / "transcripts").mkdir(parents=True)292 (run / "packets").mkdir()293 (run / "transcripts" / "T1.json").write_text(tr.to_json())294 (run / "packets" / "T1.json").write_text(pk.model_dump_json())295 turns_df, comm_df = audit_run(run)296 assert len(turns_df) == 8 and (turns_df["run_id"] == "run_x").all()297 summ = summarise_faithfulness(turns_df, comm_df)298 assert summ.iloc[0]["cited_share_valid"] == 1.0 and summ.iloc[0]["anchoring_index"] == pytest.approx(1.0)299300301# ----------------------------------------------------------------------------------------------- tables302303def test_latex_fragments_have_no_trailing_backslashes(tmp_path: Path):304 truth = _truth(80)305 preds = pd.concat([_preds(truth, "A_hedonic", 0.1, intervals=False), _preds(truth, "G_debate_r2", 0.1, seed=7)])306 summ = summarise_by(preds, truth)307 tex = write_main_results(summ, tmp_path / "main.tex")308 text = (tmp_path / "main.tex").read_text()309 assert not text.rstrip().endswith("\\\\")310 assert text.rstrip().endswith(r"\end{tabular}")311 assert r"\toprule" in tex and r"\bottomrule" in tex and r"\num{" in tex312 assert "A (hedonic)" in tex and "G (debate r2)" in tex313 assert (tmp_path / "main.csv").exists()314 cov = write_coverage(summ, tmp_path / "coverage.tex")315 assert "{--}" in cov and not cov.rstrip().endswith("\\\\") # arm without intervals prints --316 c = compare_arms(preds, truth, [("A_hedonic", "G_debate_r2")], B=30)317 write_contrasts(c, tmp_path / "contrasts.tex")318 assert not (tmp_path / "contrasts.tex").read_text().rstrip().endswith("\\\\")319 st = stratified_metrics(preds, truth, strata=["region_group"], metrics=["mean_abs_log_error"])320 write_strata(st, tmp_path / "strata.tex")321 s = (tmp_path / "strata.tex").read_text()322 assert not s.rstrip().endswith("\\\\") and "Region:" in s323324325def test_escape_and_num():326 assert escape("a_b & 50%") == r"a\_b \& 50\%"327 assert num(float("nan")) == "{--}" and num(0.12345, 2) == r"\num{0.12}" and num(None) == "{--}"328 assert arm_pretty("J_cost_matched") == "J (cost matched)" and arm_pretty("human") == "human"329