ctt metrics api

workspace/cil/__init__.py

@@ -0,0 +1,32 @@
+"""Canonical CIL research APIs.
+
+This lightweight package is intentionally separate from the implementation-heavy
+``dovla_cil`` modules. It gives scripts, evaluator code, and paper artifacts a
+stable place to import benchmark metrics from.
+"""
+
+from cil.metrics import (
+    bootstrap_ci,
+    branch_causal_action_regret,
+    car_decomposition,
+    macro_micro_summary,
+    negative_near_at_threshold,
+    pairwise_causal_dominance_ece,
+    positive_tangent_recall_at_k,
+    positives_closer_than_negatives,
+    selector_regret_at_k,
+    support_gap,
+)
+
+__all__ = [
+    "bootstrap_ci",
+    "branch_causal_action_regret",
+    "car_decomposition",
+    "macro_micro_summary",
+    "negative_near_at_threshold",
+    "pairwise_causal_dominance_ece",
+    "positive_tangent_recall_at_k",
+    "positives_closer_than_negatives",
+    "selector_regret_at_k",
+    "support_gap",
+]

workspace/cil/metrics.py

@@ -0,0 +1,367 @@
+from __future__ import annotations
+
+import math
+import random
+from collections import defaultdict
+from collections.abc import Iterable, Mapping, Sequence
+from typing import Any
+
+
+Number = int | float
+Vector = Sequence[Number]
+Matrix = Sequence[Vector]
+
+
+def branch_causal_action_regret(best_measured_branch: Number, selected_branch: Number) -> float:
+    """Regret to the best measured branch in a same-state action chart.
+
+    The metric is clipped at zero so a generated action that exceeds the hidden chart
+    oracle does not produce negative regret.
+    """
+
+    return max(0.0, float(best_measured_branch) - float(selected_branch))
+
+
+def positive_tangent_recall_at_k(
+    generated_utilities: Sequence[Number],
+    base_utility: Number,
+    *,
+    epsilon: float = 0.0,
+    k: int | None = None,
+) -> float:
+    """Return 1.0 if any generated proposal improves over the base action."""
+
+    prefix = _prefix(generated_utilities, k)
+    threshold = float(base_utility) + float(epsilon)
+    return 1.0 if any(float(value) > threshold for value in prefix) else 0.0
+
+
+def selector_regret_at_k(
+    generated_utilities: Sequence[Number],
+    *,
+    selected_index: int = 0,
+    k: int | None = None,
+) -> float:
+    """Selector regret inside the generated proposal prefix."""
+
+    prefix = _prefix(generated_utilities, k)
+    if not prefix:
+        return 0.0
+    if selected_index < 0 or selected_index >= len(prefix):
+        raise IndexError("selected_index must refer to an element inside the K-prefix")
+    return branch_causal_action_regret(max(float(value) for value in prefix), prefix[selected_index])
+
+
+def support_gap(best_hidden_chart: Number, best_generated_set: Number) -> float:
+    """Proposal-support gap to a hidden same-state chart oracle."""
+
+    return max(0.0, float(best_hidden_chart) - float(best_generated_set))
+
+
+def car_decomposition(
+    *,
+    best_hidden_chart: Number,
+    best_generated_set: Number,
+    selected: Number,
+    base: Number | None = None,
+) -> dict[str, float | None]:
+    """Decompose total CAR into support gap and selector gap.
+
+    ``best_hidden_chart`` is the hidden same-state oracle. ``best_generated_set`` is
+    the oracle over deployment-clean generated candidates. ``selected`` is the action
+    actually executed by the method.
+    """
+
+    support = support_gap(best_hidden_chart, best_generated_set)
+    selector = branch_causal_action_regret(best_generated_set, selected)
+    payload: dict[str, float | None] = {
+        "base": None if base is None else float(base),
+        "selected": float(selected),
+        "best_generated_set": float(best_generated_set),
+        "best_hidden_chart": float(best_hidden_chart),
+        "support_gap": support,
+        "selector_gap": selector,
+        "total_car": branch_causal_action_regret(best_hidden_chart, selected),
+        "support_fraction_of_total": None,
+        "selector_fraction_of_total": None,
+        "clean_gain": None,
+        "closed_fraction_of_base_to_chart_gap": None,
+    }
+    total = support + selector
+    if total > 0.0:
+        payload["support_fraction_of_total"] = support / total
+        payload["selector_fraction_of_total"] = selector / total
+    if base is not None:
+        base_value = float(base)
+        clean_gain = float(selected) - base_value
+        chart_gap = float(best_hidden_chart) - base_value
+        payload["clean_gain"] = clean_gain
+        payload["closed_fraction_of_base_to_chart_gap"] = (
+            clean_gain / chart_gap if chart_gap > 0.0 else None
+        )
+    return payload
+
+
+def negative_near_at_threshold(
+    generated_tangents: Matrix,
+    negative_tangents: Matrix,
+    *,
+    threshold: float,
+    k: int | None = None,
+) -> float:
+    """Fraction of generated tangents within ``threshold`` of any negative tangent."""
+
+    if threshold < 0:
+        raise ValueError("threshold must be non-negative")
+    generated = _prefix(generated_tangents, k)
+    if not generated or not negative_tangents:
+        return 0.0
+    count = 0
+    for tangent in generated:
+        nearest_negative = min(_rms_l2(tangent, negative) for negative in negative_tangents)
+        if nearest_negative <= threshold:
+            count += 1
+    return count / len(generated)
+
+
+def positives_closer_than_negatives(
+    generated_tangents: Matrix,
+    positive_tangents: Matrix,
+    negative_tangents: Matrix,
+    *,
+    k: int | None = None,
+) -> float | None:
+    """Fraction of generated tangents closer to positive support than negative support."""
+
+    generated = _prefix(generated_tangents, k)
+    if not generated or not positive_tangents or not negative_tangents:
+        return None
+    count = 0
+    for tangent in generated:
+        positive_distance = min(_rms_l2(tangent, positive) for positive in positive_tangents)
+        negative_distance = min(_rms_l2(tangent, negative) for negative in negative_tangents)
+        if positive_distance < negative_distance:
+            count += 1
+    return count / len(generated)
+
+
+def pairwise_causal_dominance_ece(
+    predicted_scores: Sequence[Number],
+    utilities: Sequence[Number],
+    *,
+    n_bins: int = 10,
+    temperature: float = 1.0,
+    min_utility_margin: float = 0.0,
+) -> dict[str, Any]:
+    """Expected calibration error for pairwise causal dominance predictions."""
+
+    if len(predicted_scores) != len(utilities):
+        raise ValueError("predicted_scores and utilities must have the same length")
+    if n_bins <= 0:
+        raise ValueError("n_bins must be positive")
+    if temperature <= 0.0:
+        raise ValueError("temperature must be positive")
+
+    bins = [
+        {
+            "count": 0,
+            "accuracy_sum": 0.0,
+            "confidence_sum": 0.0,
+            "lower": index / n_bins,
+            "upper": (index + 1) / n_bins,
+        }
+        for index in range(n_bins)
+    ]
+    total = 0
+    correct_sum = 0.0
+    confidence_sum = 0.0
+    for left in range(len(predicted_scores)):
+        for right in range(left + 1, len(predicted_scores)):
+            utility_margin = float(utilities[left]) - float(utilities[right])
+            if abs(utility_margin) <= min_utility_margin:
+                continue
+            score_margin = (float(predicted_scores[left]) - float(predicted_scores[right])) / (
+                float(temperature)
+            )
+            probability_left_dominates = _sigmoid(score_margin)
+            confidence = max(probability_left_dominates, 1.0 - probability_left_dominates)
+            prediction_left_dominates = probability_left_dominates >= 0.5
+            truth_left_dominates = utility_margin > 0.0
+            correct = float(prediction_left_dominates == truth_left_dominates)
+            bin_index = min(n_bins - 1, int(confidence * n_bins))
+            bucket = bins[bin_index]
+            bucket["count"] += 1
+            bucket["accuracy_sum"] += correct
+            bucket["confidence_sum"] += confidence
+            total += 1
+            correct_sum += correct
+            confidence_sum += confidence
+
+    ece = 0.0
+    rendered_bins: list[dict[str, float | int]] = []
+    for bucket in bins:
+        count = int(bucket["count"])
+        accuracy = bucket["accuracy_sum"] / count if count else 0.0
+        confidence = bucket["confidence_sum"] / count if count else 0.0
+        if total:
+            ece += (count / total) * abs(accuracy - confidence)
+        rendered_bins.append(
+            {
+                "lower": float(bucket["lower"]),
+                "upper": float(bucket["upper"]),
+                "count": count,
+                "accuracy": accuracy,
+                "confidence": confidence,
+                "abs_gap": abs(accuracy - confidence),
+            }
+        )
+    return {
+        "ece": ece,
+        "num_pairs": total,
+        "accuracy": correct_sum / total if total else None,
+        "mean_confidence": confidence_sum / total if total else None,
+        "bins": rendered_bins,
+    }
+
+
+def bootstrap_ci(
+    values: Iterable[Number],
+    *,
+    num_samples: int = 1000,
+    confidence: float = 0.95,
+    seed: int = 0,
+) -> dict[str, float | int | None]:
+    """Bootstrap confidence interval for the mean."""
+
+    clean = [float(value) for value in values if math.isfinite(float(value))]
+    if num_samples <= 0:
+        raise ValueError("num_samples must be positive")
+    if not 0.0 < confidence < 1.0:
+        raise ValueError("confidence must be in (0, 1)")
+    if not clean:
+        return {
+            "n": 0,
+            "mean": None,
+            "low": None,
+            "high": None,
+            "confidence": confidence,
+            "num_samples": num_samples,
+        }
+
+    rng = random.Random(seed)
+    means = []
+    for _ in range(num_samples):
+        sample = [clean[rng.randrange(len(clean))] for _ in range(len(clean))]
+        means.append(sum(sample) / len(sample))
+    means.sort()
+    alpha = (1.0 - confidence) / 2.0
+    low_index = min(len(means) - 1, max(0, int(math.floor(alpha * len(means)))))
+    high_index = min(len(means) - 1, max(0, int(math.ceil((1.0 - alpha) * len(means))) - 1))
+    return {
+        "n": len(clean),
+        "mean": sum(clean) / len(clean),
+        "low": means[low_index],
+        "high": means[high_index],
+        "confidence": confidence,
+        "num_samples": num_samples,
+    }
+
+
+def macro_micro_summary(
+    rows: Iterable[Mapping[str, Any]],
+    value_key: str,
+    *,
+    task_key: str = "task_id",
+    seed_key: str = "seed",
+    bootstrap_samples: int = 1000,
+    confidence: float = 0.95,
+    seed: int = 0,
+) -> dict[str, Any]:
+    """Summarize metric rows with micro, task-macro, and seed-macro means."""
+
+    materialized = list(rows)
+    values = [_maybe_float(row.get(value_key)) for row in materialized]
+    values = [value for value in values if value is not None]
+    by_task: dict[str, list[float]] = defaultdict(list)
+    by_seed: dict[str, list[float]] = defaultdict(list)
+    for row in materialized:
+        value = _maybe_float(row.get(value_key))
+        if value is None:
+            continue
+        by_task[str(row.get(task_key, "unknown"))].append(value)
+        by_seed[str(row.get(seed_key, "unknown"))].append(value)
+
+    task_means = {key: _mean(items) for key, items in sorted(by_task.items())}
+    seed_means = {key: _mean(items) for key, items in sorted(by_seed.items())}
+    task_macro_values = [value for value in task_means.values() if value is not None]
+    seed_macro_values = [value for value in seed_means.values() if value is not None]
+    return {
+        "metric": value_key,
+        "n": len(values),
+        "micro": bootstrap_ci(
+            values,
+            num_samples=bootstrap_samples,
+            confidence=confidence,
+            seed=seed,
+        ),
+        "macro_by_task": {
+            "mean": _mean(task_macro_values),
+            "per_task": task_means,
+            "ci": bootstrap_ci(
+                task_macro_values,
+                num_samples=bootstrap_samples,
+                confidence=confidence,
+                seed=seed + 1,
+            ),
+        },
+        "macro_by_seed": {
+            "mean": _mean(seed_macro_values),
+            "per_seed": seed_means,
+            "ci": bootstrap_ci(
+                seed_macro_values,
+                num_samples=bootstrap_samples,
+                confidence=confidence,
+                seed=seed + 2,
+            ),
+        },
+    }
+
+
+def _prefix(values: Sequence[Any], k: int | None) -> list[Any]:
+    if k is None:
+        return list(values)
+    if k <= 0:
+        raise ValueError("k must be positive when provided")
+    return list(values[:k])
+
+
+def _rms_l2(left: Vector, right: Vector) -> float:
+    if len(left) != len(right):
+        raise ValueError("tangent vectors must have the same length")
+    if not left:
+        return 0.0
+    squared = [(float(a) - float(b)) ** 2 for a, b in zip(left, right, strict=True)]
+    return math.sqrt(sum(squared) / len(squared))
+
+
+def _sigmoid(value: float) -> float:
+    if value >= 0.0:
+        z = math.exp(-value)
+        return 1.0 / (1.0 + z)
+    z = math.exp(value)
+    return z / (1.0 + z)
+
+
+def _maybe_float(value: Any) -> float | None:
+    if value is None:
+        return None
+    try:
+        numeric = float(value)
+    except (TypeError, ValueError):
+        return None
+    return numeric if math.isfinite(numeric) else None
+
+
+def _mean(values: Iterable[float]) -> float | None:
+    clean = [float(value) for value in values if math.isfinite(float(value))]
+    return sum(clean) / len(clean) if clean else None