diff --git a/src/rtichoke/_performance_table_spec.py b/src/rtichoke/_performance_table_spec.py new file mode 100644 index 00000000..620bc075 --- /dev/null +++ b/src/rtichoke/_performance_table_spec.py @@ -0,0 +1,222 @@ +"""Internal canonical PerformanceTableSpec builders. + +These helpers translate already-calculated production performance data plus +semantic evaluation metadata. They do not calculate or render statistics. +""" + +from __future__ import annotations + +import math +from collections.abc import Mapping +from typing import Any, TypedDict, cast + +import polars as pl + +from rtichoke.processing.evaluation_semantics import _EvaluationMetadata + +_METRICS: tuple[tuple[str, str], ...] = ( + ("true_positives", "True Positives"), + ("true_negatives", "True Negatives"), + ("false_positives", "False Positives"), + ("false_negatives", "False Negatives"), + ("sensitivity", "Sensitivity"), + ("specificity", "Specificity"), + ("false_positive_rate", "False Positive Rate"), + ("ppv", "PPV"), + ("npv", "NPV"), + ("lift", "Lift"), + ("predicted_positives", "Predicted Positives"), + ("ppcr", "PPCR"), + ("net_benefit", "Net Benefit"), + ("net_benefit_interventions_avoided", "Interventions Avoided"), +) + + +class _EvaluationSpec(TypedDict, total=False): + id: str + model: str + population: str + + +class _MetricDefinition(TypedDict): + id: str + label: str + + +class _OperatingPoint(TypedDict): + type: str + value: float + + +class _MetricValue(TypedDict): + metricId: str + estimate: float | int | None + + +class _EvaluationContext(TypedDict): + censoringHeuristic: str + competingEventHeuristic: str + + +class _PerformanceTableRow(TypedDict, total=False): + evaluationId: str + operatingPoint: _OperatingPoint + values: list[_MetricValue] + horizon: float + context: _EvaluationContext + + +class _PerformanceTableSpec(TypedDict): + schemaVersion: str + type: str + evaluations: list[_EvaluationSpec] + metrics: list[_MetricDefinition] + rows: list[_PerformanceTableRow] + + +def _performance_table_spec_from_performance_data( + performance_data: pl.DataFrame, + evaluation_metadata: Mapping[str, _EvaluationMetadata], +) -> _PerformanceTableSpec: + """Build the canonical static PerformanceTableSpec.""" + return _build_performance_table_spec( + performance_data, evaluation_metadata, time_dependent=False + ) + + +def _performance_table_times_spec_from_performance_data( + performance_data: pl.DataFrame, + evaluation_metadata: Mapping[str, _EvaluationMetadata], +) -> _PerformanceTableSpec: + """Build the canonical time-dependent PerformanceTableSpec.""" + return _build_performance_table_spec( + performance_data, evaluation_metadata, time_dependent=True + ) + + +def _build_performance_table_spec( + performance_data: pl.DataFrame, + evaluation_metadata: Mapping[str, _EvaluationMetadata], + *, + time_dependent: bool, +) -> _PerformanceTableSpec: + required = {"reference_group", "stratified_by", "chosen_cutoff"} + if time_dependent: + required |= { + "fixed_time_horizon", + "censoring_heuristic", + "competing_heuristic", + } + missing = required.difference(performance_data.columns) + if missing: + raise ValueError( + "Performance-table data is missing columns: " + ", ".join(sorted(missing)) + ) + + metric_definitions: list[_MetricDefinition] = [ + cast(_MetricDefinition, {"id": metric_id, "label": label}) + for metric_id, label in _METRICS + if metric_id in performance_data.columns + ] + metric_ids = [definition["id"] for definition in metric_definitions] + + rows = performance_data.select( + [ + column + for column in ( + "reference_group", + "fixed_time_horizon", + "censoring_heuristic", + "competing_heuristic", + "stratified_by", + "chosen_cutoff", + *metric_ids, + ) + if column in performance_data.columns + ] + ).to_dicts() + row_groups = {str(row["reference_group"]) for row in rows} + missing_metadata = row_groups.difference(evaluation_metadata) + if missing_metadata: + raise ValueError( + "Performance-table rows are missing evaluation metadata: " + + ", ".join(sorted(missing_metadata)) + ) + + ordered_groups = [group for group in evaluation_metadata if group in row_groups] + evaluation_ids = { + group: f"evaluation-{index}" + for index, group in enumerate(ordered_groups, start=1) + } + evaluations: list[_EvaluationSpec] = [] + for group in ordered_groups: + metadata = evaluation_metadata[group] + evaluation: _EvaluationSpec = { + "id": evaluation_ids[group], + "population": metadata.population, + } + if metadata.model is not None: + evaluation["model"] = metadata.model + evaluations.append(evaluation) + + canonical_rows: list[_PerformanceTableRow] = [] + for row in rows: + group = str(row["reference_group"]) + stratified_by = str(row["stratified_by"]) + operating_point: _OperatingPoint + if stratified_by == "probability_threshold": + operating_point = { + "type": "probability_threshold", + "value": _number(row["chosen_cutoff"]), + } + elif stratified_by == "ppcr": + operating_point = {"type": "ppcr", "value": _number(row["ppcr"])} + else: + raise ValueError( + "Canonical PerformanceTableSpec supports probability_threshold " + f"or ppcr operating points, not {stratified_by!r}" + ) + + canonical_row: _PerformanceTableRow = { + "evaluationId": evaluation_ids[group], + "operatingPoint": operating_point, + "values": cast( + list[_MetricValue], + [ + { + "metricId": metric_id, + "estimate": _nullable_number(row[metric_id]), + } + for metric_id in metric_ids + ], + ), + } + if time_dependent: + canonical_row["horizon"] = _number(row["fixed_time_horizon"]) + canonical_row["context"] = { + "censoringHeuristic": str(row["censoring_heuristic"]), + "competingEventHeuristic": str(row["competing_heuristic"]), + } + canonical_rows.append(canonical_row) + + return { + "schemaVersion": "2.0", + "type": "performance_table", + "evaluations": evaluations, + "metrics": metric_definitions, + "rows": canonical_rows, + } + + +def _nullable_number(value: Any) -> float | int | None: + if value is None: + return None + if isinstance(value, float) and math.isnan(value): + return None + return value + + +def _number(value: Any) -> float: + if value is None: + raise ValueError("Operating point and horizon values must not be null") + return float(value) diff --git a/src/rtichoke/processing/adjustments.py b/src/rtichoke/processing/adjustments.py index eafde726..519b7823 100644 --- a/src/rtichoke/processing/adjustments.py +++ b/src/rtichoke/processing/adjustments.py @@ -1,8 +1,9 @@ +from collections.abc import Sequence + import pandas as pd import polars as pl -from polarstate import predict_aj_estimates -from polarstate import prepare_event_table -from collections.abc import Sequence +from polarstate import predict_aj_estimates, prepare_event_table + from rtichoke.processing.transforms import assign_and_explode_polars @@ -529,7 +530,7 @@ def _aj_estimates_by_cutoff_per_horizon( df.filter(pl.col("fixed_time_horizon") == h) .group_by("strata") .map_groups( - lambda group: extract_aj_estimate_by_cutoffs( + lambda group, h=h: extract_aj_estimate_by_cutoffs( group, [h], breaks, stratified_by, full_event_table=False ) ) @@ -547,7 +548,7 @@ def _aj_estimates_per_horizon( df.filter(pl.col("fixed_time_horizon") == h) .group_by("strata") .map_groups( - lambda group: extract_aj_estimate_for_strata( + lambda group, h=h: extract_aj_estimate_for_strata( group, [h], full_event_table ) ) @@ -650,6 +651,8 @@ def _aj_adjusted_events( adjusted = extract_aj_estimate_by_cutoffs( non_competing, horizons, breaks, stratified_by, full_event_table ) + else: + raise ValueError(f"Unsupported risk-set scope: {risk_set_scope!r}") adjusted = adjusted.with_columns( [ @@ -704,6 +707,8 @@ def _aj_adjusted_events( adjusted = _aj_estimates_by_cutoff_per_horizon( base_df, horizons, breaks, stratified_by ) + else: + raise ValueError(f"Unsupported risk-set scope: {risk_set_scope!r}") adjusted = adjusted.with_columns( pl.lit(risk_set_scope) @@ -730,6 +735,8 @@ def _aj_adjusted_events( adjusted = extract_aj_estimate_by_cutoffs( base_df, horizons, breaks, stratified_by, full_event_table ) + else: + raise ValueError(f"Unsupported risk-set scope: {risk_set_scope!r}") adjusted = adjusted.with_columns( [ diff --git a/tests/test_cutoff_grid_endpoint.py b/tests/test_cutoff_grid_endpoint.py index 0436c4dd..d8980c97 100644 --- a/tests/test_cutoff_grid_endpoint.py +++ b/tests/test_cutoff_grid_endpoint.py @@ -1,3 +1,5 @@ +from typing import cast + import numpy as np import pytest @@ -17,7 +19,7 @@ def test_probability_threshold_breaks_match_r_seq_semantics(by, expected): breaks = create_breaks_values(None, "probability_threshold", by) assert breaks.tolist() == expected - assert np.max(breaks) <= 1.0 + assert float(np.max(breaks)) <= 1.0 @pytest.mark.parametrize( @@ -33,7 +35,7 @@ def test_binary_performance_data_matches_r_cutoff_endpoint(by, expected_max): assert cutoffs.min() == 0.0 assert cutoffs.max() == expected_max - assert cutoffs.max() <= 1.0 + assert cast(float, cutoffs.max()) <= 1.0 @pytest.mark.parametrize( @@ -56,4 +58,4 @@ def test_time_performance_data_matches_r_cutoff_endpoint(by, expected_max): assert cutoffs.min() == 0.0 assert cutoffs.max() == expected_max - assert cutoffs.max() <= 1.0 + assert cast(float, cutoffs.max()) <= 1.0 diff --git a/tests/test_performance_table.py b/tests/test_performance_table.py index 8b329ed1..3b5cdfe1 100644 --- a/tests/test_performance_table.py +++ b/tests/test_performance_table.py @@ -1,11 +1,11 @@ +from typing import cast + import numpy as np import pytest from great_tables import GT from reactable import Reactable import rtichoke.performance_table as performance_table_module -from rtichoke.performance_table_reactable import _bar_style, _net_benefit_style - from rtichoke import ( create_performance_table, create_performance_table_times, @@ -13,6 +13,8 @@ prepare_performance_data_times, render_performance_table, ) +from rtichoke.performance_table import PerformanceTableRenderer +from rtichoke.performance_table_reactable import _bar_style, _net_benefit_style def _example(): @@ -215,7 +217,9 @@ def test_invalid_renderer_is_rejected(): probs, reals = _example() data = prepare_performance_data(probs, reals, by=0.1) with pytest.raises(ValueError, match="renderer"): - render_performance_table(data, renderer="unknown") + render_performance_table( + data, renderer=cast(PerformanceTableRenderer, "unknown") + ) def test_reactable_metric_bar_matches_r_colors_and_geometry(): diff --git a/tests/test_performance_table_spec.py b/tests/test_performance_table_spec.py new file mode 100644 index 00000000..de7d0d73 --- /dev/null +++ b/tests/test_performance_table_spec.py @@ -0,0 +1,176 @@ +from collections.abc import Sequence + +import numpy as np +import polars as pl + +from rtichoke._performance_table_spec import ( + _performance_table_spec_from_performance_data, + _performance_table_times_spec_from_performance_data, +) +from rtichoke.performance_data.performance_data import prepare_performance_data +from rtichoke.performance_data.performance_data_times import ( + prepare_performance_data_times, +) +from rtichoke.processing.evaluation_semantics import _build_evaluation_metadata + + +def _static( + probs: dict[str, np.ndarray], + reals: np.ndarray | dict[str, np.ndarray], + *, + by: float = 0.5, + stratified_by: Sequence[str] = ("probability_threshold",), +): + data = prepare_performance_data(probs, reals, by=by, stratified_by=stratified_by) + metadata = _build_evaluation_metadata(probs, reals, np.array([])) + return _performance_table_spec_from_performance_data(data, metadata) + + +def test_static_one_model_one_population_and_metric_ids(): + probs = {"Model A": np.array([0.1, 0.4, 0.8, 0.9])} + spec = _static(probs, np.array([0, 0, 1, 1])) + + assert spec["evaluations"] == [ + { + "id": "evaluation-1", + "model": "Model A", + "population": "__shared_population__", + } + ] + assert [metric["id"] for metric in spec["metrics"]] == [ + "true_positives", + "true_negatives", + "false_positives", + "false_negatives", + "sensitivity", + "specificity", + "false_positive_rate", + "ppv", + "npv", + "lift", + "predicted_positives", + "ppcr", + "net_benefit", + "net_benefit_interventions_avoided", + ] + assert "seriesId" not in str(spec) + + +def test_static_multiple_models_share_population_and_ids_are_deterministic(): + probs = { + "Model A": np.array([0.1, 0.4, 0.8, 0.9]), + "Model B": np.array([0.2, 0.3, 0.7, 0.95]), + } + reals = np.array([0, 0, 1, 1]) + first = _static(probs, reals) + second = _static(probs, reals) + + assert first == second + assert [item["id"] for item in first["evaluations"]] == [ + "evaluation-1", + "evaluation-2", + ] + assert {item["population"] for item in first["evaluations"]} == { + "__shared_population__" + } + models = {item["model"] for item in first["evaluations"]} + assert models == {"Model A", "Model B"} + + +def test_keyed_inputs_are_distinct_populations_with_unknown_model(): + probs = { + "Population A": np.array([0.1, 0.4, 0.8, 0.9]), + "Population B": np.array([0.1, 0.4, 0.8, 0.9]), + } + reals = { + "Population A": np.array([0, 0, 1, 1]), + "Population B": np.array([0, 0, 1, 1]), + } + spec = _static(probs, reals) + + assert spec["evaluations"] == [ + {"id": "evaluation-1", "population": "Population A"}, + {"id": "evaluation-2", "population": "Population B"}, + ] + assert {row["evaluationId"] for row in spec["rows"]} == { + "evaluation-1", + "evaluation-2", + } + + +def test_static_probability_threshold_and_ppcr_operating_points(): + probs = {"Model A": np.array([0.1, 0.4, 0.8, 0.9])} + reals = np.array([0, 0, 1, 1]) + threshold = _static(probs, reals) + ppcr = _static(probs, reals, stratified_by=("ppcr",)) + + assert {row["operatingPoint"]["type"] for row in threshold["rows"]} == { + "probability_threshold" + } + assert {row["operatingPoint"]["type"] for row in ppcr["rows"]} == {"ppcr"} + assert all(0 <= row["operatingPoint"]["value"] <= 1 for row in ppcr["rows"]) + + +def test_zero_is_preserved_and_missing_metric_is_null(): + data = pl.DataFrame( + { + "reference_group": ["Model A"], + "stratified_by": ["probability_threshold"], + "chosen_cutoff": [0.5], + "sensitivity": [0.0], + "ppv": [None], + }, + schema_overrides={"ppv": pl.Float64}, + ) + metadata = _build_evaluation_metadata( + {"Model A": np.array([0.2])}, np.array([0]), np.array([]) + ) + spec = _performance_table_spec_from_performance_data(data, metadata) + values = { + value["metricId"]: value["estimate"] for value in spec["rows"][0]["values"] + } + + assert values == {"sensitivity": 0.0, "ppv": None} + + +def test_time_table_maps_horizon_and_heuristic_context(): + probs = {"Model A": np.array([0.1, 0.3, 0.7, 0.9])} + reals = np.array([0, 1, 0, 1]) + times = np.array([2.0, 3.0, 7.0, 8.0]) + heuristics = [ + { + "censoring_heuristic": "adjusted", + "competing_heuristic": "adjusted_as_negative", + }, + { + "censoring_heuristic": "excluded", + "competing_heuristic": "excluded", + }, + ] + data = prepare_performance_data_times( + probs, + reals, + times, + fixed_time_horizons=[5.0, 10.0], + heuristics_sets=heuristics, + by=0.5, + ) + metadata = _build_evaluation_metadata(probs, reals, times) + spec = _performance_table_times_spec_from_performance_data(data, metadata) + + assert {row["horizon"] for row in spec["rows"]} == {5.0, 10.0} + assert {tuple(row["context"].values()) for row in spec["rows"]} == { + ("adjusted", "adjusted_as_negative"), + ("excluded", "excluded"), + } + + +def test_equal_valued_distinct_evaluations_remain_distinct(): + values = np.array([0.1, 0.4, 0.8, 0.9]) + outcomes = np.array([0, 0, 1, 1]) + probs = {"Population A": values, "Population B": values.copy()} + reals = {"Population A": outcomes, "Population B": outcomes.copy()} + spec = _static(probs, reals) + + assert len(spec["evaluations"]) == 2 + assert len({row["evaluationId"] for row in spec["rows"]}) == 2 diff --git a/tests/test_time_gains_v2.py b/tests/test_time_gains_v2.py index 3d01c8bd..cd6a41ed 100644 --- a/tests/test_time_gains_v2.py +++ b/tests/test_time_gains_v2.py @@ -1,4 +1,5 @@ from pathlib import Path +from typing import Any, cast import matplotlib.figure import numpy as np @@ -116,6 +117,7 @@ def test_censoring_and_competing_risk_reference_comes_from_performance_layer(): spec = _gains_times_v2_spec_from_performance_data( performance, _build_evaluation_metadata(probs, reals, times) ) + references = cast(list[dict[str, Any]], spec["references"]) calculated = { float(row["fixed_time_horizon"]): float(row["real_positives"] / row["n"]) for row in performance.filter(performance["chosen_cutoff"] == 0) @@ -126,7 +128,7 @@ def test_censoring_and_competing_risk_reference_comes_from_performance_layer(): assert { reference["horizon"]: reference["points"][1]["x"] - for reference in spec["references"][1:] + for reference in references[1:] } == calculated diff --git a/tests/test_time_lift_v2.py b/tests/test_time_lift_v2.py index 257ee6ee..fa6b5a4e 100644 --- a/tests/test_time_lift_v2.py +++ b/tests/test_time_lift_v2.py @@ -1,3 +1,5 @@ +from typing import Any, cast + import matplotlib.figure import numpy as np import plotly.graph_objects as go @@ -143,6 +145,7 @@ def test_time_lift_censoring_and_competing_risk_reference_comes_from_performance spec = _lift_times_v2_spec_from_performance_data( performance, _build_evaluation_metadata(probs, reals, times) ) + references = cast(list[dict[str, Any]], spec["references"]) calculated_risks = { float(row["fixed_time_horizon"]): float(row["real_positives"] / row["n"]) for row in performance.filter(performance["chosen_cutoff"] == 0) @@ -151,7 +154,7 @@ def test_time_lift_censoring_and_competing_risk_reference_comes_from_performance .to_dicts() } - for reference in spec["references"][1:]: + for reference in references[1:]: horizon = reference["horizon"] risk = calculated_risks[horizon] assert reference["points"] == [ diff --git a/tests/test_viz_browser.py b/tests/test_viz_browser.py index 8e23e52c..bc5097ed 100644 --- a/tests/test_viz_browser.py +++ b/tests/test_viz_browser.py @@ -1,4 +1,5 @@ from pathlib import Path +from typing import Any, cast import numpy as np @@ -35,6 +36,7 @@ def test_real_roc_output_maps_to_canonical_spec(): performance_data = _real_roc_performance_data() spec = _roc_spec_from_performance_data(performance_data) + spec = cast(dict[str, Any], spec) assert spec["schemaVersion"] == "1.0" assert spec["type"] == "roc" @@ -47,6 +49,7 @@ def test_real_roc_output_maps_to_canonical_spec(): def test_real_calibration_output_maps_to_canonical_spec(): spec = _calibration_spec_from_curve_list(_real_calibration_curve_list()) + spec = cast(dict[str, Any], spec) assert spec["schemaVersion"] == "1.0" assert spec["type"] == "calibration" diff --git a/tests/test_viz_spec_v2.py b/tests/test_viz_spec_v2.py index e8b9dfec..8422cd67 100644 --- a/tests/test_viz_spec_v2.py +++ b/tests/test_viz_spec_v2.py @@ -1,3 +1,5 @@ +from typing import Any, cast + import numpy as np from rtichoke._viz_spec_v2 import ( @@ -24,6 +26,7 @@ def test_roc_v2_one_model_one_population(): performance_data, _static_metadata(probs, reals), ) + spec = cast(dict[str, Any], spec) assert spec["schemaVersion"] == "2.0" assert spec["type"] == "roc" @@ -57,6 +60,7 @@ def test_roc_v2_multiple_models_share_one_population(): performance_data, _static_metadata(probs, reals), ) + spec = cast(dict[str, Any], spec) evaluations = spec["evaluations"] assert [evaluation["id"] for evaluation in evaluations] == [ @@ -87,6 +91,7 @@ def test_roc_v2_keyed_populations_keep_model_identity_unknown(): performance_data, _static_metadata(probs, reals), ) + spec = cast(dict[str, Any], spec) assert spec["evaluations"] == [ {"id": "evaluation-1", "population": "Population A"}, @@ -112,6 +117,7 @@ def test_roc_v2_ids_do_not_encode_compatibility_group_labels(): prepare_performance_data(probs, reals, by=0.25), _static_metadata(probs, reals), ) + spec = cast(dict[str, Any], spec) renamed_reals = { "Cohort X": reals["Population A"], @@ -125,6 +131,7 @@ def test_roc_v2_ids_do_not_encode_compatibility_group_labels(): prepare_performance_data(renamed_probs, renamed_reals, by=0.25), _static_metadata(renamed_probs, renamed_reals), ) + renamed_spec = cast(dict[str, Any], renamed_spec) assert [evaluation["id"] for evaluation in spec["evaluations"]] == [ evaluation["id"] for evaluation in renamed_spec["evaluations"] @@ -142,6 +149,7 @@ def test_gains_v2_uses_production_prevalence_for_perfect_path(): spec = _gains_v2_spec_from_performance_data( performance_data, _static_metadata(probs, reals) ) + spec = cast(dict[str, Any], spec) assert spec["type"] == "gains" assert spec["x"] == "ppcr" @@ -171,6 +179,7 @@ def test_gains_v2_shares_one_perfect_path_across_models(): prepare_performance_data(probs, reals, by=0.25), _static_metadata(probs, reals), ) + spec = cast(dict[str, Any], spec) assert len(spec["series"]) == 2 assert len(spec["references"]) == 2 @@ -191,6 +200,7 @@ def test_gains_v2_keeps_equal_prevalence_populations_distinct(): prepare_performance_data(probs, reals, by=0.25), _static_metadata(probs, reals), ) + spec = cast(dict[str, Any], spec) perfect = spec["references"][1:] assert [reference["population"] for reference in perfect] == [