ctt artifacts 2026-07-02 workspace/scripts/eval_metrics.py

workspace/scripts/eval_metrics.py

@@ -0,0 +1,337 @@
+#!/usr/bin/env python
+from __future__ import annotations
+
+import argparse
+import json
+import math
+import sys
+from collections import defaultdict
+from pathlib import Path
+from typing import Any
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+from cil.metrics import (  # noqa: E402
+    MetricInputError,
+    branch_car,
+    macro_micro_summary,
+    measured_support_gap,
+    negative_near_at_threshold,
+    outcome_ptr_at_k,
+    pairwise_causal_dominance_ece,
+    positives_closer_than_negatives,
+    proxy_positive_tangent_coverage_at_k,
+    proxy_support_distance,
+    selector_regret_at_k,
+)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Evaluate CIL/CTT metrics while keeping measured outcome metrics "
+            "separate from distance-only proxy metrics."
+        )
+    )
+    parser.add_argument("--input", type=Path, required=True)
+    parser.add_argument("--out-dir", type=Path, required=True)
+    parser.add_argument("--mode", choices=("measured", "proxy"), required=True)
+    parser.add_argument("--k", type=int, default=16)
+    parser.add_argument("--epsilon", type=float, default=0.0)
+    parser.add_argument("--thresholds", default="0.20,0.40")
+    parser.add_argument("--bootstrap-samples", type=int, default=1000)
+    parser.add_argument("--confidence", type=float, default=0.95)
+    args = parser.parse_args(argv)
+
+    if args.k <= 0:
+        parser.error("--k must be positive")
+    thresholds = _parse_thresholds(args.thresholds)
+    payload = json.loads(args.input.read_text())
+    rows = payload.get("rows", payload) if isinstance(payload, dict) else payload
+    if not isinstance(rows, list):
+        parser.error("input must be a JSON list or an object with a rows list")
+
+    metric_rows = []
+    for index, row in enumerate(rows):
+        if not isinstance(row, dict):
+            raise MetricInputError(f"row {index} must be an object")
+        metric_rows.append(
+            _measured_row(row, k=args.k, epsilon=args.epsilon)
+            if args.mode == "measured"
+            else _proxy_row(row, k=args.k, thresholds=thresholds)
+        )
+
+    metric_names = sorted(
+        {
+            key
+            for row in metric_rows
+            for key, value in row.items()
+            if key not in {"task_id", "seed", "chart_id", "mode"}
+            and isinstance(value, (int, float))
+            and math.isfinite(float(value))
+        }
+    )
+    summary = {
+        name: macro_micro_summary(
+            metric_rows,
+            name,
+            bootstrap_samples=args.bootstrap_samples,
+            confidence=args.confidence,
+        )
+        for name in metric_names
+    }
+
+    out_dir = args.out_dir
+    out_dir.mkdir(parents=True, exist_ok=True)
+    (out_dir / "metrics.json").write_text(
+        json.dumps(
+            {
+                "mode": args.mode,
+                "k": args.k,
+                "epsilon": args.epsilon,
+                "thresholds": thresholds,
+                "num_rows": len(metric_rows),
+                "rows": metric_rows,
+                "summary": summary,
+            },
+            indent=2,
+            sort_keys=True,
+        )
+        + "\n"
+    )
+    (out_dir / "metrics_by_task.json").write_text(
+        json.dumps(_group_means(metric_rows, "task_id", metric_names), indent=2, sort_keys=True)
+        + "\n"
+    )
+    (out_dir / "metrics_by_seed.json").write_text(
+        json.dumps(_group_means(metric_rows, "seed", metric_names), indent=2, sort_keys=True)
+        + "\n"
+    )
+    (out_dir / "table.tex").write_text(_latex_table(summary) + "\n")
+    (out_dir / "report.md").write_text(_markdown_report(args.mode, args.k, summary) + "\n")
+    print(json.dumps({"out_dir": str(out_dir), "num_rows": len(metric_rows)}, indent=2))
+    return 0
+
+
+def _measured_row(row: dict[str, Any], *, k: int, epsilon: float) -> dict[str, Any]:
+    if not bool(row.get("candidates_evaluated", False)):
+        raise MetricInputError(
+            "measured mode requires candidates_evaluated=true for every row; "
+            "distance-only rows must use --mode proxy"
+        )
+    utilities = _numbers(row, "generated_utilities")
+    if not utilities:
+        raise MetricInputError("measured rows require generated_utilities")
+    selected_index = int(row.get("selected_index", 0))
+    hidden = _numbers(row, "hidden_chart_utilities", required=False)
+    selected_utility = utilities[selected_index]
+    prefix = utilities[:k]
+    output = _base_row(row, mode="measured")
+    output[f"outcome_ptr_at_{k}"] = outcome_ptr_at_k(
+        utilities,
+        _number(row, "base_utility"),
+        epsilon=epsilon,
+        k=k,
+        candidates_evaluated=True,
+    )
+    output[f"selector_regret_at_{k}"] = selector_regret_at_k(
+        utilities,
+        selected_index=selected_index,
+        k=k,
+        candidates_evaluated=True,
+    )
+    output[f"branch_car_at_{k}"] = branch_car(max(prefix), selected_utility)
+    if hidden:
+        output[f"support_gap_at_{k}"] = measured_support_gap(
+            max(hidden),
+            max(prefix),
+            candidates_evaluated=True,
+        )
+    predicted = _numbers(row, "predicted_scores", required=False)
+    if predicted and len(predicted) >= len(utilities):
+        ece = pairwise_causal_dominance_ece(predicted[: len(utilities)], utilities)
+        output["pairwise_causal_calibration_ece"] = ece["ece"]
+    return output
+
+
+def _proxy_row(row: dict[str, Any], *, k: int, thresholds: list[float]) -> dict[str, Any]:
+    generated = _matrix(row, "generated_tangents")
+    positives = _matrix(row, "positive_tangents")
+    negatives = _matrix(row, "negative_tangents", required=False)
+    output = _base_row(row, mode="proxy")
+    for threshold in thresholds:
+        suffix = _threshold_suffix(threshold)
+        output[f"pptc_at_{k}_thr_{suffix}"] = proxy_positive_tangent_coverage_at_k(
+            generated,
+            positives,
+            threshold=threshold,
+            k=k,
+        )
+        output[f"negative_near_at_{k}_thr_{suffix}"] = negative_near_at_threshold(
+            generated,
+            negatives,
+            threshold=threshold,
+            k=k,
+        )
+    distance = proxy_support_distance(generated, positives, k=k)
+    if distance is not None:
+        output[f"proxy_support_distance_at_{k}"] = distance
+    closer = positives_closer_than_negatives(generated, positives, negatives, k=k)
+    if closer is not None:
+        output[f"pos_closer_than_neg_at_{k}"] = closer
+    output[f"candidate_diversity_at_{k}"] = _mean_pairwise_distance(generated[:k])
+    output[f"collapse_rate_at_{k}"] = _collapse_rate(generated[:k])
+    return output
+
+
+def _base_row(row: dict[str, Any], *, mode: str) -> dict[str, Any]:
+    return {
+        "mode": mode,
+        "chart_id": str(row.get("chart_id", row.get("group_id", "unknown"))),
+        "task_id": str(row.get("task_id", "unknown")),
+        "seed": str(row.get("seed", "unknown")),
+    }
+
+
+def _numbers(row: dict[str, Any], key: str, *, required: bool = True) -> list[float]:
+    values = row.get(key)
+    if values is None:
+        if required:
+            raise MetricInputError(f"row requires {key}")
+        return []
+    if not isinstance(values, list):
+        raise MetricInputError(f"{key} must be a list")
+    return [float(value) for value in values]
+
+
+def _number(row: dict[str, Any], key: str) -> float:
+    if key not in row:
+        raise MetricInputError(f"row requires {key}")
+    return float(row[key])
+
+
+def _matrix(row: dict[str, Any], key: str, *, required: bool = True) -> list[list[float]]:
+    values = row.get(key)
+    if values is None:
+        if required:
+            raise MetricInputError(f"row requires {key}")
+        return []
+    if not isinstance(values, list):
+        raise MetricInputError(f"{key} must be a list of vectors")
+    return [[float(item) for item in vector] for vector in values]
+
+
+def _parse_thresholds(raw: str) -> list[float]:
+    values = [float(item.strip()) for item in raw.split(",") if item.strip()]
+    if not values or any(value < 0.0 for value in values):
+        raise ValueError("--thresholds must contain non-negative values")
+    return values
+
+
+def _threshold_suffix(value: float) -> str:
+    return f"{value:.2f}".replace(".", "p")
+
+
+def _group_means(
+    rows: list[dict[str, Any]],
+    key: str,
+    metric_names: list[str],
+) -> dict[str, dict[str, float]]:
+    grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
+    for row in rows:
+        grouped[str(row.get(key, "unknown"))].append(row)
+    output: dict[str, dict[str, float]] = {}
+    for group, group_rows in sorted(grouped.items()):
+        payload: dict[str, float] = {}
+        for metric in metric_names:
+            values = [
+                float(row[metric])
+                for row in group_rows
+                if isinstance(row.get(metric), (int, float))
+                and math.isfinite(float(row[metric]))
+            ]
+            if values:
+                payload[metric] = sum(values) / len(values)
+        output[group] = payload
+    return output
+
+
+def _latex_table(summary: dict[str, Any]) -> str:
+    lines = [
+        "% Auto-generated by scripts/eval_metrics.py",
+        "\\begin{tabular}{lrrrr}",
+        "\\toprule",
+        "Metric & N & Micro mean & CI low & CI high \\\\",
+        "\\midrule",
+    ]
+    for name, payload in sorted(summary.items()):
+        micro = payload["micro"]
+        lines.append(
+            f"{_latex_escape(name)} & {micro['n']} & {_fmt(micro['mean'])} & "
+            f"{_fmt(micro['low'])} & {_fmt(micro['high'])} \\\\"
+        )
+    lines.extend(["\\bottomrule", "\\end{tabular}"])
+    return "\n".join(lines)
+
+
+def _markdown_report(mode: str, k: int, summary: dict[str, Any]) -> str:
+    lines = [
+        f"# Metric Evaluation ({mode})",
+        "",
+        f"K: `{k}`",
+        "",
+        "| Metric | N | Micro mean | 95% CI | Task macro | Seed macro |",
+        "| --- | ---: | ---: | ---: | ---: | ---: |",
+    ]
+    for name, payload in sorted(summary.items()):
+        micro = payload["micro"]
+        task_mean = payload["macro_by_task"]["mean"]
+        seed_mean = payload["macro_by_seed"]["mean"]
+        lines.append(
+            f"| {name} | {micro['n']} | {_fmt(micro['mean'])} | "
+            f"[{_fmt(micro['low'])}, {_fmt(micro['high'])}] | "
+            f"{_fmt(task_mean)} | {_fmt(seed_mean)} |"
+        )
+    return "\n".join(lines)
+
+
+def _mean_pairwise_distance(vectors: list[list[float]]) -> float:
+    if len(vectors) < 2:
+        return 0.0
+    distances = []
+    for left_index, left in enumerate(vectors):
+        for right in vectors[left_index + 1 :]:
+            distances.append(_rms_l2(left, right))
+    return sum(distances) / len(distances)
+
+
+def _collapse_rate(vectors: list[list[float]], *, threshold: float = 1.0e-6) -> float:
+    if not vectors:
+        return 0.0
+    first = vectors[0]
+    collapsed = sum(1 for vector in vectors if _rms_l2(first, vector) <= threshold)
+    return collapsed / len(vectors)
+
+
+def _rms_l2(left: list[float], right: list[float]) -> float:
+    if len(left) != len(right):
+        raise MetricInputError("vectors must have matching dimensions")
+    if not left:
+        return 0.0
+    return math.sqrt(sum((a - b) ** 2 for a, b in zip(left, right, strict=True)) / len(left))
+
+
+def _fmt(value: Any) -> str:
+    if not isinstance(value, (int, float)):
+        return "n/a"
+    return f"{float(value):.4f}"
+
+
+def _latex_escape(value: str) -> str:
+    return value.replace("_", "\\_").replace("%", "\\%").replace("&", "\\&")
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())