diff --git a/src/rtichoke/_viz_spec_v2.py b/src/rtichoke/_viz_spec_v2.py index 1ec30b70..b8e85103 100644 --- a/src/rtichoke/_viz_spec_v2.py +++ b/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": [], } diff --git a/tests/test_viz_spec_v2.py b/tests/test_viz_spec_v2.py index 7ec46dea..e8b9dfec 100644 --- a/tests/test_viz_spec_v2.py +++ b/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"])