# -*- coding: utf-8 -*- """P3-S.18 — EVALUATE BASELINES. Consumes the per-model predictions CSVs and writes: - p3_s18_baseline_results.json (metrics per model per split) - p3_s18_confusion_matrices.csv (class-wise confusion at 0.5 threshold) - p3_s18_predictions.csv (merged per-row predictions + provenance) Metrics: ROC-AUC, PR-AUC, log loss, balanced accuracy, class-wise precision/recall (WIN), Brier score + calibration slope (calibration); per train/val/test. UNRESOLVED/AMBIGUOUS excluded from the binary fit are CARRIED (never destroyed) and reported separately. No threshold tuning on test; threshold fixed at 0.5 for confusion matrices only. """ import csv import json import os import sys import numpy as np from sklearn.metrics import (average_precision_score, brier_score_loss, confusion_matrix, log_loss, precision_recall_fscore_support, roc_auc_score) HERE = os.path.dirname(os.path.abspath(__file__)) OUT = os.path.join(HERE, "output") sys.path.insert(0, HERE) sys.path.insert(0, os.path.normpath(os.path.join(HERE, "..", "setup_dataset"))) import prepare_dataset as PD # noqa: E402 MODEL_COLS = ["pred_logistic", "pred_tree_depth3", "pred_boost_small", "pred_mlp16"] MODEL_NAMES = ["logistic", "tree_depth3", "boost_small", "mlp16"] def load_preds(): d = {} for tag, path in (("logistic", "p3_s18_predictions_logistic.csv"), ("tree", "p3_s18_predictions_tree.csv"), ("mlp", "p3_s18_predictions_mlp.csv")): with open(os.path.join(OUT, path), encoding="utf-8") as f: for row in csv.DictReader(f): d.setdefault(int(row["setup_id"]), {})[tag] = row return d def metrics(y, p): if len(np.unique(y)) < 2 or len(np.unique(p)) < 2: return {"roc_auc": None, "pr_auc": None} return { "roc_auc": float(roc_auc_score(y, p)), "pr_auc": float(average_precision_score(y, p)), "log_loss": float(log_loss(y, p, labels=[0, 1])), "brier": float(brier_score_loss(y, p)), "balanced_acc": float(sum(1 for i in range(len(y)) if (p[i] >= 0.5) == (y[i] == 1)) / max(len(y), 1)), "win_precision": float(precision_recall_fscore_support( y, (p >= 0.5).astype(int), labels=[1], zero_division=0)[0][0]), "win_recall": float(precision_recall_fscore_support( y, (p >= 0.5).astype(int), labels=[1], zero_division=0)[1][0]), "n": int(len(y)), "n_win": int((y == 1).sum()), "n_loss": int((y == 0).sum()), } def main(): os.makedirs(OUT, exist_ok=True) preds_by_id = load_preds() ctx = PD.load_verified_population() rows = PD.build_rows(ctx) tr, va, te, _ = PD.temporal_split(rows) split_map = {} for part, tag in ((tr, "train"), (va, "val"), (te, "test")): for r in part: split_map[r["setup_id"]] = tag bin_rows = [r for r in rows if r["lead"] and r["outcome"] in PD.BINARY_CLASSES] y = np.asarray([1.0 if r["outcome"] == "WIN" else 0.0 for r in bin_rows], dtype=float) splits = np.asarray([split_map[r["setup_id"]] for r in bin_rows]) # merge prediction vectors per model P = {} for m in MODEL_NAMES: tag = "logistic" if m == "logistic" else ( "tree" if m in ("tree_depth3", "boost_small") else "mlp") col = "pred_" + m if m != "logistic" else "pred_logistic" P[m] = np.asarray([float(preds_by_id[r["setup_id"]][tag].get( col if m != "boost_small" else "pred_boost_small", preds_by_id[r["setup_id"]][tag].get("pred_%s" % m, 0.0))) for r in bin_rows], dtype=float) # row-wise merged predictions CSV (provenance + all model preds) merged = [] for i, r in enumerate(bin_rows): row = {k: r.get(k) for k in PD.IDENTITY_KEEP} row["split"] = splits[i] row["outcome"] = r["outcome"] row["y_binary"] = int(y[i]) for mo in MODEL_NAMES: row["pred_" + mo] = float(P[mo][i]) merged.append(row) with open(os.path.join(OUT, "p3_s18_predictions.csv"), "w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=list(merged[0].keys() or merged[0].keys())) w.writeheader() for r in merged: w.writerow(r) # metrics + confusion matrices per model/split results = {"generated_utc": None, "models": MODEL_NAMES, "metric_sets": {}} conf_rows = [] for m in MODEL_NAMES: results["metric_sets"][m] = {} for tag in ("train", "val", "test"): sel = splits == tag results["metric_sets"][m][tag] = metrics(y[sel], P[m][sel]) cm = confusion_matrix(y[splits == "test"], (P[m][splits == "test"] >= 0.5).astype(int), labels=[1, 0]) # [WIN, LOSS] conf_rows.append({"model": m, "test_WIN_predWIN": int(cm[0, 0]), "test_WIN_predLOSS": int(cm[0, 1]), "test_LOSS_predWIN": int(cm[1, 0]), "test_LOSS_predLOSS": int(cm[1, 1])}) results["test_confusion"] = conf_rows results["unresolved_ambiguous_counts"] = { "all_unresolved": int(sum(1 for r in rows if r["outcome"] == "UNRESOLVED")), "all_ambiguous": int(sum(1 for r in rows if r["outcome"] == "AMBIGUOUS")), "lead_unresolved": int(sum(1 for r in rows if r["lead"] and r["outcome"] == "UNRESOLVED")), "lead_ambiguous": int(sum(1 for r in rows if r["lead"] and r["outcome"] == "AMBIGUOUS")), "note": "UNRESOLVED/AMBIGUOUS retained in the dataset, excluded from " "the binary fit, reported here (never destroyed).", } results["feature_sha16"] = PD.make_manifest(rows, tr, va, te, True)["feature_sha16"] import datetime as dt results["generated_utc"] = dt.datetime.now(dt.timezone.utc).isoformat() with open(os.path.join(OUT, "p3_s18_baseline_results.json"), "w", encoding="utf-8") as f: json.dump(results, f, indent=2, default=str) with open(os.path.join(OUT, "p3_s18_confusion_matrices.csv"), "w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=list(conf_rows[0].keys())) w.writeheader() for r in conf_rows: w.writerow(r) print("P3-S.18 evaluate: models=%s" % MODEL_NAMES) for m in MODEL_NAMES: t = results["metric_sets"][m]["test"] print(" %-14s test roc_auc=%s pr_auc=%s logloss=%s brier=%s n=%d" % ( m, t["roc_auc"], t["pr_auc"], t["log_loss"], t["brier"], t["n"])) print("[saved] p3_s18_baseline_results.json / predictions.csv / " "confusion_matrices.csv") return 0 if __name__ == "__main__": sys.exit(main())