Add internal canonical gains v2 builder

src/rtichoke/_viz_spec_v2.py

@@ -19,39 +19,139 @@
     "sensitivity",
     "specificity",
 }
+_REQUIRED_GAINS_COLUMNS = {
+    "reference_group",
+    "chosen_cutoff",
+    "sensitivity",
+    "ppcr",
+    "real_positives",
+    "n",
+}
 
 
 def _roc_v2_spec_from_performance_data(
     performance_data: pl.DataFrame,
     evaluation_metadata: Mapping[str, _EvaluationMetadata],
 ) -> dict[str, object]:
-    """Build a canonical ROC-v2 spec without recalculating statistics.
+    """Build a canonical ROC-v2 spec without recalculating statistics."""
+    return _curve_v2_spec_from_performance_data(
+        performance_data,
+        evaluation_metadata,
+        chart_type="roc",
+    )
+
+
+def _gains_v2_spec_from_performance_data(
+    performance_data: pl.DataFrame,
+    evaluation_metadata: Mapping[str, _EvaluationMetadata],
+) -> dict[str, object]:
+    """Build a canonical gains-v2 spec from production performance quantities."""
+    spec = _curve_v2_spec_from_performance_data(
+        performance_data,
+        evaluation_metadata,
+        chart_type="gains",
+    )
+    prevalence = _gains_population_prevalence(performance_data, evaluation_metadata)
+
+    populations = list(
+        dict.fromkeys(metadata.population for metadata in evaluation_metadata.values())
+    )
+    spec["references"] = [
+        {"type": "identity", "scope": "global", "label": "Random"},
+        *[
+            {
+                "type": "path",
+                "scope": "population",
+                "population": population,
+                "label": "Perfect Model",
+                "points": [
+                    {"x": 0, "y": 0},
+                    {"x": prevalence[population], "y": 1},
+                    {"x": 1, "y": 1},
+                ],
+            }
+            for population in populations
+        ],
+    ]
+    return spec
+
+
+def _gains_population_prevalence(
+    performance_data: pl.DataFrame,
+    evaluation_metadata: Mapping[str, _EvaluationMetadata],
+) -> dict[str, float]:
+    """Map production event prevalence from compatibility groups to populations."""
+    group_prevalence = {
+        str(row["reference_group"]): float(row["prevalence"])
+        for row in (
+            performance_data.select(
+                "reference_group",
+                (pl.col("real_positives") / pl.col("n")).alias("prevalence"),
+            )
+            .unique()
+            .to_dicts()
+        )
+    }
+    population_values: dict[str, set[float]] = {}
+    for group, metadata in evaluation_metadata.items():
+        if group not in group_prevalence:
+            continue
+        population_values.setdefault(metadata.population, set()).add(
+            group_prevalence[group]
+        )
+
+    prevalence: dict[str, float] = {}
+    for population in dict.fromkeys(
+        metadata.population for metadata in evaluation_metadata.values()
+    ):
+        values = population_values.get(population, set())
+        if len(values) != 1:
+            raise ValueError(
+                "Gains performance data must have one prevalence per population: "
+                f"{population}"
+            )
+        prevalence[population] = next(iter(values))
+    return prevalence
+
+
+def _curve_v2_spec_from_performance_data(
+    performance_data: pl.DataFrame,
+    evaluation_metadata: Mapping[str, _EvaluationMetadata],
+    *,
+    chart_type: str,
+) -> dict[str, object]:
+    """Build common canonical curve semantics without recalculating statistics.
 
     ``reference_group`` is used only to join existing performance rows to the
     semantic metadata established by the high-level production inputs. Stable
     evaluation/series IDs are ordinal over that semantic metadata, so they do
-    not encode labels, compatibility grouping names, prevalence, coordinates,
-    colors, or other presentation/statistical values.
+    not encode compatibility grouping names or presentation values.
     """
-    missing = _REQUIRED_ROC_COLUMNS.difference(performance_data.columns)
+    if chart_type == "roc":
+        required = _REQUIRED_ROC_COLUMNS
+        selected = ["reference_group", "chosen_cutoff", "sensitivity", "specificity"]
+    elif chart_type == "gains":
+        required = _REQUIRED_GAINS_COLUMNS
+        selected = ["reference_group", "chosen_cutoff", "sensitivity", "ppcr"]
+    else:
+        raise ValueError(f"Unsupported v2 curve type: {chart_type}")
+
+    missing = required.difference(performance_data.columns)
     if missing:
         missing_columns = ", ".join(sorted(missing))
-        raise ValueError(f"ROC performance data is missing columns: {missing_columns}")
-
-    rows = performance_data.select(
-        "reference_group",
-        "chosen_cutoff",
-        "sensitivity",
-        "specificity",
-    ).to_dicts()
+        raise ValueError(
+            f"{chart_type.upper()} performance data is missing columns: {missing_columns}"
+        )
+
+    rows = performance_data.select(*selected).to_dicts()
     row_groups = {str(row["reference_group"]) for row in rows}
     metadata_groups = set(evaluation_metadata)
-
     missing_metadata = row_groups.difference(metadata_groups)
     if missing_metadata:
         groups = ", ".join(sorted(missing_metadata))
         raise ValueError(
-            f"ROC performance rows are missing evaluation metadata: {groups}"
+            f"{chart_type.upper()} performance rows are missing evaluation metadata: "
+            f"{groups}"
         )
 
     ordered_groups = [group for group in evaluation_metadata if group in row_groups]
@@ -65,7 +165,6 @@ def _roc_v2_spec_from_performance_data(
 
     evaluations: list[dict[str, object]] = []
     series: list[dict[str, object]] = []
-
     for group in ordered_groups:
         metadata = evaluation_metadata[group]
         evaluation: dict[str, object] = {
@@ -82,7 +181,6 @@ def _roc_v2_spec_from_performance_data(
         else:
             display_value = metadata.population
             display_role = "population"
-
         series.append(
             {
                 "id": series_ids[group],
@@ -95,23 +193,42 @@ def _roc_v2_spec_from_performance_data(
             }
         )
 
+    data = []
+    for row in rows:
+        datum = {
+            "seriesId": series_ids[str(row["reference_group"])],
+            "cutoff": row["chosen_cutoff"],
+            "sensitivity": row["sensitivity"],
+        }
+        if chart_type == "roc":
+            datum["specificity"] = row["specificity"]
+        else:
+            datum["ppcr"] = row["ppcr"]
+        data.append(datum)
+
+    if chart_type == "roc":
+        return {
+            "schemaVersion": "2.0",
+            "type": "roc",
+            "evaluations": evaluations,
+            "series": series,
+            "data": data,
+            "x": "false_positive_rate",
+            "y": "sensitivity",
+            "xAxis": {"label": "1 - Specificity", "domain": [0, 1]},
+            "yAxis": {"label": "Sensitivity", "domain": [0, 1]},
+            "references": [{"type": "identity", "scope": "global"}],
+        }
+
     return {
         "schemaVersion": "2.0",
-        "type": "roc",
+        "type": "gains",
         "evaluations": evaluations,
         "series": series,
-        "data": [
-            {
-                "seriesId": series_ids[str(row["reference_group"])],
-                "cutoff": row["chosen_cutoff"],
-                "sensitivity": row["sensitivity"],
-                "specificity": row["specificity"],
-            }
-            for row in rows
-        ],
-        "x": "false_positive_rate",
+        "data": data,
+        "x": "ppcr",
         "y": "sensitivity",
-        "xAxis": {"label": "1 - Specificity", "domain": [0, 1]},
+        "xAxis": {"label": "Predicted Positives (Rate)", "domain": [0, 1]},
         "yAxis": {"label": "Sensitivity", "domain": [0, 1]},
-        "references": [{"type": "identity", "scope": "global"}],
+        "references": [],
     }

tests/test_viz_spec_v2.py

@@ -1,6 +1,9 @@
 import numpy as np
 
-from rtichoke._viz_spec_v2 import _roc_v2_spec_from_performance_data
+from rtichoke._viz_spec_v2 import (
+    _gains_v2_spec_from_performance_data,
+    _roc_v2_spec_from_performance_data,
+)
 from rtichoke.performance_data.performance_data import prepare_performance_data
 from rtichoke.processing.evaluation_semantics import (
     _SHARED_POPULATION,
@@ -129,3 +132,70 @@ def test_roc_v2_ids_do_not_encode_compatibility_group_labels():
     assert [series["id"] for series in spec["series"]] == [
         series["id"] for series in renamed_spec["series"]
     ]
+
+
+def test_gains_v2_uses_production_prevalence_for_perfect_path():
+    probs = {"Model A": np.array([0.05, 0.2, 0.7, 0.95])}
+    reals = np.array([0, 0, 0, 1])
+    performance_data = prepare_performance_data(probs, reals, by=0.25)
+
+    spec = _gains_v2_spec_from_performance_data(
+        performance_data, _static_metadata(probs, reals)
+    )
+
+    assert spec["type"] == "gains"
+    assert spec["x"] == "ppcr"
+    assert spec["y"] == "sensitivity"
+    assert spec["references"][0] == {
+        "type": "identity",
+        "scope": "global",
+        "label": "Random",
+    }
+    perfect = spec["references"][1]
+    assert perfect["scope"] == "population"
+    assert perfect["population"] == _SHARED_POPULATION
+    assert perfect["points"] == [
+        {"x": 0, "y": 0},
+        {"x": 0.25, "y": 1},
+        {"x": 1, "y": 1},
+    ]
+
+
+def test_gains_v2_shares_one_perfect_path_across_models():
+    probs = {
+        "Model A": np.array([0.05, 0.2, 0.7, 0.95]),
+        "Model B": np.array([0.1, 0.4, 0.6, 0.9]),
+    }
+    reals = np.array([0, 0, 1, 1])
+    spec = _gains_v2_spec_from_performance_data(
+        prepare_performance_data(probs, reals, by=0.25),
+        _static_metadata(probs, reals),
+    )
+
+    assert len(spec["series"]) == 2
+    assert len(spec["references"]) == 2
+    assert spec["references"][1]["population"] == _SHARED_POPULATION
+    assert spec["references"][1]["points"][1]["x"] == 0.5
+
+
+def test_gains_v2_keeps_equal_prevalence_populations_distinct():
+    probs = {
+        "Population A": np.array([0.05, 0.2, 0.7, 0.95]),
+        "Population B": np.array([0.1, 0.4, 0.6, 0.9]),
+    }
+    reals = {
+        "Population A": np.array([0, 0, 1, 1]),
+        "Population B": np.array([0, 1, 0, 1]),
+    }
+    spec = _gains_v2_spec_from_performance_data(
+        prepare_performance_data(probs, reals, by=0.25),
+        _static_metadata(probs, reals),
+    )
+
+    perfect = spec["references"][1:]
+    assert [reference["population"] for reference in perfect] == [
+        "Population A",
+        "Population B",
+    ]
+    assert perfect[0]["points"] == perfect[1]["points"]
+    assert all("model" not in evaluation for evaluation in spec["evaluations"])