diff --git a/tests/unit_tests/model_validation/sklearn/test_PopulationStabilityIndex.py b/tests/unit_tests/model_validation/sklearn/test_PopulationStabilityIndex.py new file mode 100644 index 000000000..5329f78a1 --- /dev/null +++ b/tests/unit_tests/model_validation/sklearn/test_PopulationStabilityIndex.py @@ -0,0 +1,108 @@ +import unittest + +import pandas as pd +import plotly.graph_objects as go +from sklearn.datasets import make_classification +from sklearn.linear_model import LogisticRegression +from sklearn.model_selection import train_test_split + +import validmind as vm +from validmind.tests.model_validation.sklearn.PopulationStabilityIndex import ( + PopulationStabilityIndex, +) + + +def _build(n_classes=3, drop_class=None, drop_class_initial=None, drop_class_new=None): + """Fit a model on ``n_classes`` and build initial/new VM datasets. + + ``drop_class`` removes that class from both dataset slices while leaving the + model fit on the full class set -- the missing-class scenario. Pass + ``drop_class_initial``/``drop_class_new`` instead to drop a class from only + one of the two slices -- the class-membership-differs-between-slices scenario. + """ + if drop_class is not None: + drop_class_initial = drop_class + drop_class_new = drop_class + + X, y = make_classification( + n_samples=800, + n_features=5, + n_informative=4, + n_redundant=0, + n_classes=n_classes, + n_clusters_per_class=1, + random_state=7, + ) + X_train, X_eval, y_train, y_eval = train_test_split( + X, y, test_size=0.5, random_state=7 + ) + model = LogisticRegression(max_iter=1000).fit(X_train, y_train) + vm_model = vm.init_model(input_id="psi_model", model=model, __log=False) + + # Two evaluation slices to compare (initial vs new). + X_a, X_b, y_a, y_b = train_test_split(X_eval, y_eval, test_size=0.5, random_state=7) + + def _ds(input_id, X_, y_, drop_cls): + df = pd.DataFrame(X_, columns=[f"f{i}" for i in range(5)]) + df["target"] = y_ + if drop_cls is not None: + df = df[df["target"] != drop_cls].reset_index(drop=True) + ds = vm.init_dataset( + input_id=input_id, dataset=df, target_column="target", __log=False + ) + ds.assign_predictions(vm_model) + return ds + + return ( + vm_model, + _ds("psi_initial", X_a, y_a, drop_class_initial), + _ds("psi_new", X_b, y_b, drop_class_new), + ) + + +class TestPopulationStabilityIndexMulticlass(unittest.TestCase): + def test_one_vs_rest_tables_and_traces(self): + model, ds_initial, ds_new = _build(n_classes=3) + tables, fig, raw = PopulationStabilityIndex([ds_initial, ds_new], model) + + self.assertIsInstance(tables, dict) + self.assertIsInstance(fig, go.Figure) + self.assertIsInstance(raw, vm.RawData) + + # One table and one raw entry per class; 3 traces per class subplot. + self.assertEqual(len(tables), 3) + self.assertEqual(set(raw.psi_raw), {"0", "1", "2"}) + self.assertEqual(len(fig.data), 9) + + +class TestPopulationStabilityIndexMissingClass(unittest.TestCase): + def test_missing_class_computes_present_tables(self): + # Model trained on 4 classes; both slices omit class 3. Previously the + # shape guard skipped; alignment on classes_ makes it compute now. + model, ds_initial, ds_new = _build(n_classes=4, drop_class=3) + tables, fig, raw = PopulationStabilityIndex([ds_initial, ds_new], model) + + # Only the 3 present classes get tables/raw entries; absent class 3 is gone. + self.assertEqual(len(tables), 3) + self.assertEqual(set(raw.psi_raw), {"0", "1", "2"}) + self.assertNotIn("3", set(raw.psi_raw)) + self.assertEqual(len(fig.data), 9) + + +class TestPopulationStabilityIndexClassMembershipDiffers(unittest.TestCase): + def test_new_dataset_missing_class_uses_initial_present_set(self): + # Model trained on 4 classes; initial slice has all 4, new slice is + # missing class 3. The present-class set is derived from the initial + # dataset only (the new dataset only gets a column-width check, since PSI + # compares score distributions rather than positives), so this should + # compute all 4 per-class tables without skipping. + model, ds_initial, ds_new = _build(n_classes=4, drop_class_new=3) + tables, fig, raw = PopulationStabilityIndex([ds_initial, ds_new], model) + + self.assertEqual(len(tables), 4) + self.assertEqual(set(raw.psi_raw), {"0", "1", "2", "3"}) + self.assertEqual(len(fig.data), 12) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit_tests/model_validation/sklearn/test_PrecisionRecallCurve.py b/tests/unit_tests/model_validation/sklearn/test_PrecisionRecallCurve.py index efc2d7cc9..8f291140b 100644 --- a/tests/unit_tests/model_validation/sklearn/test_PrecisionRecallCurve.py +++ b/tests/unit_tests/model_validation/sklearn/test_PrecisionRecallCurve.py @@ -3,6 +3,8 @@ import numpy as np import pandas as pd import plotly.graph_objects as go +from sklearn.datasets import make_classification +from sklearn.linear_model import LogisticRegression from sklearn.model_selection import train_test_split import validmind as vm @@ -16,6 +18,55 @@ XGBClassifier = None # type: ignore[misc,assignment] +class TestPrecisionRecallCurveMulticlassSklearn(unittest.TestCase): + """Runtime coverage for the multiclass PR refactor without the xgboost extra.""" + + def _build(self, drop_class=None): + X, y = make_classification( + n_samples=400, + n_features=5, + n_informative=4, + n_redundant=0, + n_classes=4 if drop_class is not None else 3, + n_clusters_per_class=1, + random_state=42, + ) + model = LogisticRegression(max_iter=1000).fit(X, y) + df = pd.DataFrame(X, columns=[f"f{i}" for i in range(5)]) + df["target"] = y + if drop_class is not None: + df = df[df["target"] != drop_class].reset_index(drop=True) + ds = vm.init_dataset( + input_id=f"mc_pr_sk_{drop_class}", + dataset=df, + target_column="target", + __log=False, + ) + vm_model = vm.init_model( + input_id=f"mc_pr_sk_model_{drop_class}", model=model, __log=False + ) + ds.assign_predictions(vm_model) + return vm_model, ds + + def test_one_vs_rest_traces(self): + model, ds = self._build() + fig, raw = PrecisionRecallCurve(model, ds) + self.assertIsInstance(fig, go.Figure) + self.assertIsInstance(raw, vm.RawData) + # 3 per-class curves + micro-average = 4 traces (no random baseline for PR). + self.assertEqual(len(fig.data), 4) + self.assertEqual(set(raw.average_precision) - {"micro"}, {"0", "1", "2"}) + + def test_missing_class_computes_present_curves(self): + # Model trained on 4 classes; dataset omits class 3. Previously skipped. + model, ds = self._build(drop_class=3) + fig, raw = PrecisionRecallCurve(model, ds) + names = [t.name for t in fig.data] + self.assertEqual(len(fig.data), 4) + self.assertEqual(sum(n.startswith("Class ") for n in names), 3) + self.assertEqual(set(raw.average_precision), {"0", "1", "2", "micro"}) + + @unittest.skipUnless(XGBClassifier is not None, "xgboost optional extra required") class TestPrecisionRecallCurveBinary(unittest.TestCase): def setUp(self): diff --git a/tests/unit_tests/model_validation/sklearn/test_ROCCurve.py b/tests/unit_tests/model_validation/sklearn/test_ROCCurve.py index 8c474a741..1960f3c98 100644 --- a/tests/unit_tests/model_validation/sklearn/test_ROCCurve.py +++ b/tests/unit_tests/model_validation/sklearn/test_ROCCurve.py @@ -3,6 +3,8 @@ import pandas as pd import validmind as vm import plotly.graph_objects as go +from sklearn.datasets import make_classification +from sklearn.linear_model import LogisticRegression from sklearn.model_selection import train_test_split from validmind.tests.model_validation.sklearn.ROCCurve import ROCCurve @@ -12,6 +14,53 @@ XGBClassifier = None # type: ignore[misc,assignment] +class TestROCCurveMulticlassMissingClass(unittest.TestCase): + """A training class absent from the evaluated slice now computes (sklearn). + + Uses LogisticRegression so it runs without the xgboost extra. The model is fit + on four classes; the dataset omits one, which previously tripped the shape + guard and skipped. Alignment on estimator.classes_ makes it compute instead. + """ + + def setUp(self): + X, y = make_classification( + n_samples=400, + n_features=5, + n_informative=4, + n_redundant=0, + n_classes=4, + n_clusters_per_class=1, + random_state=42, + ) + model = LogisticRegression(max_iter=1000).fit(X, y) + + df = pd.DataFrame(X, columns=[f"f{i}" for i in range(5)]) + df["target"] = y + df = df[df["target"] != 3].reset_index(drop=True) + + self.ds = vm.init_dataset( + input_id="mc_roc_missing", dataset=df, target_column="target", __log=False + ) + self.model = vm.init_model( + input_id="mc_roc_missing_model", model=model, __log=False + ) + self.ds.assign_predictions(self.model) + + def test_missing_class_produces_present_curves_plus_micro(self): + fig, raw = ROCCurve(self.model, self.ds) + self.assertIsInstance(fig, go.Figure) + self.assertIsInstance(raw, vm.RawData) + + names = [t.name for t in fig.data] + # 3 present-class curves + micro-average + random baseline = 5 traces. + self.assertEqual(len(fig.data), 5) + self.assertEqual(sum(n.startswith("Class ") for n in names), 3) + self.assertTrue(any(n.startswith("Micro-average") for n in names)) + + # Per-class RawData keyed by the present classes (no absent class 3) + micro. + self.assertEqual(set(raw.auc), {"0", "1", "2", "micro"}) + + @unittest.skipUnless(XGBClassifier is not None, "xgboost optional extra required") class TestROCCurve(unittest.TestCase): def setUp(self): diff --git a/tests/unit_tests/model_validation/sklearn/test_multiclass_proba.py b/tests/unit_tests/model_validation/sklearn/test_multiclass_proba.py new file mode 100644 index 000000000..120dc3d79 --- /dev/null +++ b/tests/unit_tests/model_validation/sklearn/test_multiclass_proba.py @@ -0,0 +1,198 @@ +import types +import unittest + +import numpy as np +import pandas as pd +from sklearn.datasets import make_classification +from sklearn.linear_model import LogisticRegression + +import validmind as vm +from validmind.errors import SkipTestError +from validmind.tests.model_validation.sklearn._multiclass_proba import ( + MulticlassProba, + multiclass_proba, + proba_matrix, + resolve_class_list, +) + + +def _fit_model_and_dataset(n_classes=3, drop_class=None, random_state=42): + """Fit LogisticRegression on ``n_classes`` classes and build a VM dataset. + + ``drop_class`` builds the dataset from a slice with that class removed, while + the model is still fit on the full class set -- the missing-class scenario. + """ + X, y = make_classification( + n_samples=400, + n_features=5, + n_informative=4, + n_redundant=0, + n_classes=n_classes, + n_clusters_per_class=1, + random_state=random_state, + ) + model = LogisticRegression(max_iter=1000).fit(X, y) + + df = pd.DataFrame(X, columns=[f"f{i}" for i in range(5)]) + df["target"] = y + if drop_class is not None: + df = df[df["target"] != drop_class].reset_index(drop=True) + + vm_dataset = vm.init_dataset( + input_id=f"ds_{n_classes}_{drop_class}", + dataset=df, + target_column="target", + __log=False, + ) + vm_model = vm.init_model( + input_id=f"model_{n_classes}_{drop_class}", model=model, __log=False + ) + return vm_model, vm_dataset + + +class TestResolveClassList(unittest.TestCase): + def test_uses_estimator_classes_(self): + # Model trained on 4 classes, dataset slice missing class 3: the column + # order must follow estimator.classes_ (all four), not np.unique(y). + model, dataset = _fit_model_and_dataset(n_classes=4, drop_class=3) + np.testing.assert_array_equal(resolve_class_list(model, dataset), [0, 1, 2, 3]) + # np.unique of the slice would have missed class 3. + self.assertEqual(sorted(np.unique(dataset.y).tolist()), [0, 1, 2]) + + def test_falls_back_to_unique_without_classes_(self): + # A model object without ``classes_`` falls back to np.unique(dataset.y). + _, dataset = _fit_model_and_dataset(n_classes=3) + fake_model = types.SimpleNamespace(model=object()) + np.testing.assert_array_equal( + resolve_class_list(fake_model, dataset), [0, 1, 2] + ) + + +class TestMulticlassProbaHappyPath(unittest.TestCase): + def test_all_classes_present(self): + model, dataset = _fit_model_and_dataset(n_classes=3) + result = multiclass_proba(model, dataset, "ROC Curve") + + self.assertIsInstance(result, MulticlassProba) + np.testing.assert_array_equal(result.class_list, [0, 1, 2]) + self.assertEqual([int(c) for c in result.classes_present], [0, 1, 2]) + self.assertEqual(result.present_indices, [0, 1, 2]) + + n = len(dataset.y) + self.assertEqual(result.y_bin.shape, (n, 3)) + self.assertEqual(result.y_prob.shape, (n, 3)) + # Probability rows sum to 1 (full per-class matrix, not a single column). + np.testing.assert_allclose(result.y_prob.sum(axis=1), np.ones(n), atol=1e-6) + + +class TestMulticlassProbaMissingClass(unittest.TestCase): + def test_missing_class_computes_and_aligns(self): + # Model trained on 4 classes, dataset slice missing class 3. + model, dataset = _fit_model_and_dataset(n_classes=4, drop_class=3) + result = multiclass_proba(model, dataset, "GINITable") + + # Full training class list drives column alignment / binarization ... + np.testing.assert_array_equal(result.class_list, [0, 1, 2, 3]) + self.assertEqual(result.y_prob.shape[1], 4) + self.assertEqual(result.y_bin.shape[1], 4) + + # ... but only the present classes are surfaced for per-class output. + self.assertEqual([int(c) for c in result.classes_present], [0, 1, 2]) + self.assertEqual(result.present_indices, [0, 1, 2]) + # The dropped class's column carries no positives. + self.assertEqual(int(result.y_bin[:, 3].sum()), 0) + + +class TestMulticlassProbaSkips(unittest.TestCase): + def test_no_predict_proba_skips(self): + model, dataset = _fit_model_and_dataset(n_classes=3) + model.model.predict_proba = None + with self.assertRaises(SkipTestError): + multiclass_proba(model, dataset, "ROC Curve") + + def test_raising_predict_proba_skips(self): + model, dataset = _fit_model_and_dataset(n_classes=3) + + def _boom(_X): + raise RuntimeError("no probabilities here") + + model.model.predict_proba = _boom + with self.assertRaises(SkipTestError): + multiclass_proba(model, dataset, "ROC Curve") + + def test_wrong_width_matrix_skips(self): + model, dataset = _fit_model_and_dataset(n_classes=3) + # A matrix with fewer columns than the class list must skip, not crash. + model.model.predict_proba = lambda X: np.zeros((len(X), 2)) + with self.assertRaises(SkipTestError): + multiclass_proba(model, dataset, "GINITable") + + def test_proba_matrix_wrong_width_skips(self): + # proba_matrix guards width directly against the supplied class list. + model, dataset = _fit_model_and_dataset(n_classes=3) + with self.assertRaises(SkipTestError): + proba_matrix(model, dataset, np.array([0, 1, 2, 3]), "PSI") + + def test_binary_model_on_multiclass_dataset_skips(self): + # Model trained on only 2 classes but evaluated against a 3-class dataset: + # the width check alone would pass (2 == 2), but one-vs-rest multiclass + # metrics don't apply to a binary model, so this must skip rather than + # crash when label_binarize's single-column output gets indexed as [:, 1]. + model, dataset = _fit_model_and_dataset(n_classes=3) + model.model.classes_ = np.array([0, 1]) + model.model.predict_proba = lambda X: np.tile([0.5, 0.5], (len(X), 1)) + with self.assertRaises(SkipTestError): + multiclass_proba(model, dataset, "ROC Curve") + + def test_unseen_label_in_dataset_skips(self): + # Model trained on classes {0, 1, 2}; dataset's target column includes a + # label (5) the model never saw. label_binarize would silently give those + # rows an all-zero column instead of raising, so this must skip explicitly. + model, dataset = _fit_model_and_dataset(n_classes=3) + dataset._df.loc[dataset._df.index[:5], dataset.target_column] = 5 + with self.assertRaises(SkipTestError): + multiclass_proba(model, dataset, "GINITable") + + def test_string_labels_with_int_trained_model_skip(self): + # Model trained on integer labels; the dataset's target column loads as + # strings (e.g. a CSV round-trip). String labels compare unequal to the + # integer class_list, so this must not quietly misalign label_binarize or + # mark classes absent -- it should skip loudly via the unknown-label guard, + # naming every label as unseen. + X, y = make_classification( + n_samples=400, + n_features=5, + n_informative=4, + n_redundant=0, + n_classes=3, + n_clusters_per_class=1, + random_state=42, + ) + model = LogisticRegression(max_iter=1000).fit(X, y) + + df = pd.DataFrame(X, columns=[f"f{i}" for i in range(5)]) + df["target"] = y + df["target"] = df["target"].astype(str) + + vm_dataset = vm.init_dataset( + input_id="ds_string_labels", + dataset=df, + target_column="target", + __log=False, + ) + vm_model = vm.init_model( + input_id="model_string_labels", model=model, __log=False + ) + + with self.assertRaises(SkipTestError) as ctx: + multiclass_proba(vm_model, vm_dataset, "ROC Curve") + + message = str(ctx.exception) + self.assertIn("not trained on", message) + self.assertIn("0", message) + self.assertIn("1", message) + self.assertIn("2", message) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit_tests/model_validation/statsmodels/test_GINITable.py b/tests/unit_tests/model_validation/statsmodels/test_GINITable.py index f16918f75..97fc4ae20 100644 --- a/tests/unit_tests/model_validation/statsmodels/test_GINITable.py +++ b/tests/unit_tests/model_validation/statsmodels/test_GINITable.py @@ -9,6 +9,58 @@ from validmind.tests.model_validation.statsmodels.GINITable import GINITable +class TestGINITableMissingClass(unittest.TestCase): + """A class in training but absent from the evaluated slice now computes. + + Previously the shape guard (4 predict_proba columns vs 3 dataset classes) + raised SkipTestError even though the one-vs-rest metrics are computable. With + training-class alignment the test produces a row per present class plus micro. + """ + + def setUp(self): + X, y = make_classification( + n_samples=400, + n_features=5, + n_informative=4, + n_redundant=0, + n_classes=4, + n_clusters_per_class=1, + random_state=42, + ) + # Model sees all four classes. + model = LogisticRegression(max_iter=1000).fit(X, y) + + # Dataset is a slice with class 3 removed. + df = pd.DataFrame(X, columns=[f"f{i}" for i in range(5)]) + df["target"] = y + df = df[df["target"] != 3].reset_index(drop=True) + + self.vm_dataset = vm.init_dataset( + input_id="mc_missing_dataset", + dataset=df, + target_column="target", + __log=False, + ) + self.vm_model = vm.init_model( + input_id="mc_missing_model", model=model, __log=False + ) + self.vm_dataset.assign_predictions(self.vm_model) + + def test_missing_class_returns_present_rows_plus_micro(self): + results, raw_data = GINITable(self.vm_dataset, self.vm_model) + + self.assertIsInstance(results, pd.DataFrame) + self.assertIsInstance(raw_data, RawData) + + # 3 present classes (0, 1, 2) + micro; the absent class 3 has no row. + self.assertEqual(set(results["Class"]), {"0", "1", "2", "micro"}) + self.assertEqual(len(results), 4) + self.assertNotIn("3", set(results["Class"])) + + self.assertEqual(set(raw_data.fpr), {"0", "1", "2", "micro"}) + self.assertEqual(set(raw_data.tpr), {"0", "1", "2", "micro"}) + + class TestGINITable(unittest.TestCase): def setUp(self): """Set up test fixtures before each test method.""" diff --git a/validmind/tests/model_validation/sklearn/PopulationStabilityIndex.py b/validmind/tests/model_validation/sklearn/PopulationStabilityIndex.py index 8815cddb4..513d6f538 100644 --- a/validmind/tests/model_validation/sklearn/PopulationStabilityIndex.py +++ b/validmind/tests/model_validation/sklearn/PopulationStabilityIndex.py @@ -14,6 +14,8 @@ from validmind.logging import get_logger from validmind.vm_models import VMDataset, VMModel +from ._multiclass_proba import multiclass_proba, proba_matrix + logger = get_logger(__name__) _PSI_PALETTE = ["#DE257E", "#1F77B4", "#2CA02C", "#FF7F0E", "#9467BD", "#8C564B"] @@ -104,54 +106,37 @@ def _psi_table_rows(psi_results): return rows_with_total, table_rows -def _multiclass_psi(datasets, model, classes, num_bins, mode): +def _multiclass_psi(datasets, model, num_bins, mode): """One-vs-rest PSI for a multiclass model. PSI needs a 1-D score distribution to compare across the two datasets. The - stored single probability column cannot represent every class, so we ask the - underlying estimator for the full per-class probability matrix (mirroring the - ROC/PR curve tests). Models without a usable ``predict_proba`` (metadata-only - / precomputed single-column predictions) are skipped rather than crashed. + stored single probability column cannot represent every class, so the shared + helper reaches the underlying estimator for the full per-class probability + matrix (mirroring the ROC/PR curve tests), aligning columns to the training + class order. The initial dataset drives the class alignment and the set of + classes actually present; the new dataset's matrix is validated against the + same class list. Models that cannot supply a matching matrix are skipped. """ - raw_model = getattr(model, "model", None) - proba_fn = getattr(raw_model, "predict_proba", None) - if not callable(proba_fn): - raise SkipTestError( - "Multiclass Population Stability Index requires per-class " - "probabilities from the underlying model's predict_proba, which is " - "not available for this model (e.g. metadata-only / precomputed " - "predictions). Skipping." - ) - try: - prob_initial = np.asarray(proba_fn(datasets[0].x_df())) - prob_new = np.asarray(proba_fn(datasets[1].x_df())) - except Exception as e: - raise SkipTestError( - "Multiclass Population Stability Index could not compute per-class " - f"probabilities ({type(e).__name__}). Skipping." - ) from e - - n_classes = len(classes) - for prob in (prob_initial, prob_new): - if prob.ndim != 2 or prob.shape[1] != n_classes: - raise SkipTestError( - "Multiclass Population Stability Index requires a per-class " - f"probability matrix with one column per class (got shape " - f"{getattr(prob, 'shape', None)} for {n_classes} classes). Skipping." - ) - - # predict_proba columns are ordered by sorted class label == np.unique. + aligned = multiclass_proba(model, datasets[0], "Population Stability Index") + prob_initial = aligned.y_prob + prob_new = proba_matrix( + model, datasets[1], aligned.class_list, "Population Stability Index" + ) + + classes_present = aligned.classes_present + n_present = len(classes_present) + fig = make_subplots( - rows=n_classes, + rows=n_present, cols=1, - specs=[[{"secondary_y": True}] for _ in range(n_classes)], - subplot_titles=[f"Class {cls}" for cls in classes], + specs=[[{"secondary_y": True}] for _ in range(n_present)], + subplot_titles=[f"Class {cls}" for cls in classes_present], vertical_spacing=0.08, ) tables = {} raw_per_class = {} - for i, cls in enumerate(classes): + for plot_i, (i, cls) in enumerate(zip(aligned.present_indices, classes_present)): psi_results = calculate_psi( prob_initial[:, i].copy(), prob_new[:, i].copy(), @@ -159,17 +144,17 @@ def _multiclass_psi(datasets, model, classes, num_bins, mode): mode=mode, ) x = list(range(len(psi_results))) - color = _PSI_PALETTE[i % len(_PSI_PALETTE)] + color = _PSI_PALETTE[plot_i % len(_PSI_PALETTE)] fig.add_trace( go.Bar( x=x, y=[d["percent_initial"] for d in psi_results], name="Initial", marker=dict(color="#DE257E"), - showlegend=i == 0, + showlegend=plot_i == 0, legendgroup="initial", ), - row=i + 1, + row=plot_i + 1, col=1, secondary_y=False, ) @@ -179,10 +164,10 @@ def _multiclass_psi(datasets, model, classes, num_bins, mode): y=[d["percent_new"] for d in psi_results], name="New", marker=dict(color="#E8B1F8"), - showlegend=i == 0, + showlegend=plot_i == 0, legendgroup="new", ), - row=i + 1, + row=plot_i + 1, col=1, secondary_y=False, ) @@ -192,10 +177,10 @@ def _multiclass_psi(datasets, model, classes, num_bins, mode): y=[d["psi"] for d in psi_results], name="PSI", line=dict(color=color), - showlegend=i == 0, + showlegend=plot_i == 0, legendgroup="psi", ), - row=i + 1, + row=plot_i + 1, col=1, secondary_y=True, ) @@ -211,7 +196,7 @@ def _multiclass_psi(datasets, model, classes, num_bins, mode): fig.update_layout( title="Population Stability Index (PSI) — one-vs-rest per class", barmode="group", - height=300 * n_classes, + height=300 * n_present, ) return ( @@ -291,7 +276,7 @@ def PopulationStabilityIndex( classes = np.unique(datasets[0].y) if len(classes) > 2: - return _multiclass_psi(datasets, model, classes, num_bins, mode) + return _multiclass_psi(datasets, model, num_bins, mode) psi_results = calculate_psi( datasets[0].y_prob(model).copy(), diff --git a/validmind/tests/model_validation/sklearn/PrecisionRecallCurve.py b/validmind/tests/model_validation/sklearn/PrecisionRecallCurve.py index 1843d92b7..4cb50d5fd 100644 --- a/validmind/tests/model_validation/sklearn/PrecisionRecallCurve.py +++ b/validmind/tests/model_validation/sklearn/PrecisionRecallCurve.py @@ -7,13 +7,14 @@ import numpy as np import plotly.graph_objects as go from sklearn.metrics import average_precision_score, precision_recall_curve -from sklearn.preprocessing import label_binarize from validmind import RawData, tags, tasks from validmind.errors import SkipTestError from validmind.models import FoundationModel from validmind.vm_models import VMDataset, VMModel +from ._multiclass_proba import multiclass_proba + @tags( "sklearn", @@ -75,7 +76,7 @@ def PrecisionRecallCurve( classes = np.unique(y_true) if len(classes) > 2: - return _multiclass_pr_curve(model, dataset, classes) + return _multiclass_pr_curve(model, dataset) precision, recall, _ = precision_recall_curve(y_true, dataset.y_prob(model)) @@ -105,53 +106,27 @@ def PrecisionRecallCurve( def _multiclass_pr_curve( - model: VMModel, dataset: VMDataset, classes: np.ndarray + model: VMModel, dataset: VMDataset ) -> Tuple[go.Figure, RawData]: """One-vs-rest precision-recall curves for a multiclass model. Needs the full per-class probability matrix, which the stored single - probability column cannot provide, so we ask the model for it directly. - Models without a usable ``predict_proba`` (Foundation/metadata-only, - precomputed single-column probabilities) are skipped rather than crashed. + probability column cannot provide; the shared helper reaches the underlying + estimator, aligns the probability columns to the training class order and + skips models that cannot supply a matching matrix. """ - # The VMModel wrapper's predict_proba is binary-only (it returns just the - # positive-class column), so reach the underlying estimator for the full - # per-class probability matrix. - raw_model = getattr(model, "model", None) - proba_fn = getattr(raw_model, "predict_proba", None) - if not callable(proba_fn): - raise SkipTestError( - "Multiclass Precision-Recall Curve requires per-class probabilities " - "from the underlying model's predict_proba, which is not available " - "for this model (e.g. Foundation / metadata-only / precomputed " - "predictions). Skipping." - ) - try: - y_prob = np.asarray(proba_fn(dataset.x_df())) - except Exception as e: - raise SkipTestError( - "Multiclass Precision-Recall Curve could not compute per-class " - f"probabilities ({type(e).__name__}). Skipping." - ) from e - - n_classes = len(classes) - if y_prob.ndim != 2 or y_prob.shape[1] != n_classes: - raise SkipTestError( - "Multiclass Precision-Recall Curve requires a per-class probability " - f"matrix with one column per class (got shape " - f"{getattr(y_prob, 'shape', None)} for {n_classes} classes). Skipping." - ) - - # One-hot the true labels in the same class order predict_proba columns use - # (sklearn orders predict_proba columns by sorted class label == np.unique). - y_bin = label_binarize(dataset.y.flatten(), classes=classes) + aligned = multiclass_proba(model, dataset, "Precision-Recall Curve") + y_bin = aligned.y_bin + y_prob = aligned.y_prob traces = [] raw_precision = {} raw_recall = {} raw_ap = {} palette = ["#DE257E", "#1F77B4", "#2CA02C", "#FF7F0E", "#9467BD", "#8C564B"] - for i, cls in enumerate(classes): + for plot_i, (i, cls) in enumerate( + zip(aligned.present_indices, aligned.classes_present) + ): precision, recall, _ = precision_recall_curve(y_bin[:, i], y_prob[:, i]) ap = average_precision_score(y_bin[:, i], y_prob[:, i]) key = str(cls) @@ -164,11 +139,14 @@ def _multiclass_pr_curve( y=precision, mode="lines", name=f"Class {key} (AP = {ap:.2f})", - line=dict(color=palette[i % len(palette)]), + line=dict(color=palette[plot_i % len(palette)]), ) ) - # Micro-average across all one-vs-rest decisions. + # Micro-average across the one-vs-rest decisions of the present classes. + present = aligned.present_indices + y_bin = y_bin[:, present] + y_prob = y_prob[:, present] micro_precision, micro_recall, _ = precision_recall_curve( y_bin.ravel(), y_prob.ravel() ) diff --git a/validmind/tests/model_validation/sklearn/ROCCurve.py b/validmind/tests/model_validation/sklearn/ROCCurve.py index ab2086a7f..f79054276 100644 --- a/validmind/tests/model_validation/sklearn/ROCCurve.py +++ b/validmind/tests/model_validation/sklearn/ROCCurve.py @@ -7,12 +7,12 @@ import numpy as np import plotly.graph_objects as go from sklearn.metrics import roc_auc_score, roc_curve -from sklearn.preprocessing import label_binarize from validmind import RawData, tags, tasks -from validmind.errors import SkipTestError from validmind.vm_models import VMDataset, VMModel +from ._multiclass_proba import multiclass_proba + @tags( "sklearn", @@ -76,7 +76,7 @@ def ROCCurve(model: VMModel, dataset: VMDataset) -> Tuple[go.Figure, RawData]: classes = np.unique(dataset.y) if len(classes) > 2: - return _multiclass_roc_curve(model, dataset, classes) + return _multiclass_roc_curve(model, dataset) y_prob = dataset.y_prob(model) y_true = dataset.y.astype(y_prob.dtype).flatten() @@ -117,53 +117,27 @@ def ROCCurve(model: VMModel, dataset: VMDataset) -> Tuple[go.Figure, RawData]: def _multiclass_roc_curve( - model: VMModel, dataset: VMDataset, classes: np.ndarray + model: VMModel, dataset: VMDataset ) -> Tuple[go.Figure, RawData]: """One-vs-rest ROC curves for a multiclass model. Needs the full per-class probability matrix, which the stored single - probability column cannot provide, so we ask the model for it directly. - Models without a usable ``predict_proba`` (metadata-only, precomputed - single-column probabilities) are skipped rather than crashed. + probability column cannot provide; the shared helper reaches the underlying + estimator, aligns the probability columns to the training class order and + skips models that cannot supply a matching matrix. """ - # The VMModel wrapper's predict_proba is binary-only (it returns just the - # positive-class column), so reach the underlying estimator for the full - # per-class probability matrix. - raw_model = getattr(model, "model", None) - proba_fn = getattr(raw_model, "predict_proba", None) - if not callable(proba_fn): - raise SkipTestError( - "Multiclass ROC Curve requires per-class probabilities from the " - "underlying model's predict_proba, which is not available for this " - "model (e.g. metadata-only / precomputed predictions). Skipping." - ) - try: - y_prob = np.asarray(proba_fn(dataset.x_df())) - except Exception as e: - raise SkipTestError( - "Multiclass ROC Curve could not compute per-class probabilities " - f"({type(e).__name__}). Skipping." - ) from e - - n_classes = len(classes) - if y_prob.ndim != 2 or y_prob.shape[1] != n_classes: - raise SkipTestError( - "Multiclass ROC Curve requires a per-class probability matrix with " - f"one column per class (got shape {getattr(y_prob, 'shape', None)} " - f"for {n_classes} classes). Skipping." - ) - - # One-hot the true labels in the same class order predict_proba columns use - # (sklearn orders predict_proba columns by sorted class label == np.unique). - y_true = dataset.y.flatten() - y_bin = label_binarize(y_true, classes=classes) + aligned = multiclass_proba(model, dataset, "ROC Curve") + y_bin = aligned.y_bin + y_prob = aligned.y_prob traces = [] raw_fpr = {} raw_tpr = {} raw_auc = {} palette = ["#DE257E", "#1F77B4", "#2CA02C", "#FF7F0E", "#9467BD", "#8C564B"] - for i, cls in enumerate(classes): + for plot_i, (i, cls) in enumerate( + zip(aligned.present_indices, aligned.classes_present) + ): fpr, tpr, _ = roc_curve(y_bin[:, i], y_prob[:, i], drop_intermediate=False) auc = roc_auc_score(y_bin[:, i], y_prob[:, i]) key = str(cls) @@ -176,11 +150,14 @@ def _multiclass_roc_curve( y=tpr, mode="lines", name=f"Class {key} (AUC = {auc:.2f})", - line=dict(color=palette[i % len(palette)]), + line=dict(color=palette[plot_i % len(palette)]), ) ) - # Micro-average across all one-vs-rest decisions. + # Micro-average across the one-vs-rest decisions of the present classes. + present = aligned.present_indices + y_bin = y_bin[:, present] + y_prob = y_prob[:, present] micro_fpr, micro_tpr, _ = roc_curve(y_bin.ravel(), y_prob.ravel()) micro_auc = roc_auc_score(y_bin, y_prob, average="micro", multi_class="ovr") raw_fpr["micro"] = micro_fpr diff --git a/validmind/tests/model_validation/sklearn/_multiclass_proba.py b/validmind/tests/model_validation/sklearn/_multiclass_proba.py new file mode 100644 index 000000000..60404a368 --- /dev/null +++ b/validmind/tests/model_validation/sklearn/_multiclass_proba.py @@ -0,0 +1,150 @@ +# Copyright © 2023-2026 ValidMind Inc. All rights reserved. +# Refer to the LICENSE file in the root of this repository for details. +# SPDX-License-Identifier: AGPL-3.0 AND ValidMind Commercial + +"""Shared preamble for the one-vs-rest multiclass classification tests. + +ROCCurve, PrecisionRecallCurve, PopulationStabilityIndex and GINITable all need +the full per-class probability matrix for a multiclass model. The VMModel wrapper +only stores/exposes the positive-class column, so every one of those tests reaches +the underlying estimator's ``predict_proba``, guards the cases where it is missing, +shaped wrong, trained on fewer than 3 classes, or evaluated against labels the +model never saw (raising ``SkipTestError`` with a test-specific message), and +one-hot encodes the true labels. This module holds that preamble in one place. + +Column alignment matters: scikit-learn orders ``predict_proba`` columns by +``estimator.classes_`` (the classes seen in training), which is not necessarily +``np.unique(dataset.y)`` — a dataset slice can be missing a training class. We +align on ``classes_`` when available (falling back to ``np.unique``) so the +probability columns line up, binarize against that full training class list to +keep the columns aligned, and expose the subset of classes actually present in the +evaluated dataset's ``y`` (a class with no positives has no defined ROC) so callers +emit per-class output only for those. Consequently, callers that compute a +micro-average pool only the present classes' aligned columns, so when a training +class is absent from the evaluated slice the micro number deliberately excludes +that class's column rather than pooling the full training-class matrix. +""" + +from typing import Any, List, NamedTuple + +import numpy as np +from sklearn.preprocessing import label_binarize + +from validmind.errors import SkipTestError +from validmind.vm_models import VMDataset, VMModel + + +class MulticlassProba(NamedTuple): + """Aligned per-class probabilities for a multiclass one-vs-rest test. + + ``class_list`` is the full training class order used for column alignment and + binarization; ``classes_present``/``present_indices`` select the classes that + actually appear in the evaluated dataset's ``y`` (paired: ``present_indices[k]`` + is the ``class_list``/``y_prob`` column for ``classes_present[k]``). ``y_bin`` is + the true labels one-hot encoded against ``class_list``; ``y_prob`` is the + estimator's per-class probability matrix in ``class_list`` column order. + """ + + class_list: np.ndarray + classes_present: List[Any] + present_indices: List[int] + y_bin: np.ndarray + y_prob: np.ndarray + + +def resolve_class_list(model: VMModel, dataset: VMDataset) -> np.ndarray: + """Return the ``predict_proba`` column order for a multiclass model. + + scikit-learn orders columns by ``estimator.classes_`` (training classes), so + prefer that; fall back to ``np.unique(dataset.y)`` for models that do not + expose it. + """ + raw_model = getattr(model, "model", None) + classes = getattr(raw_model, "classes_", None) + if classes is None: + classes = np.unique(dataset.y) + return np.asarray(classes) + + +def proba_matrix( + model: VMModel, dataset: VMDataset, class_list: np.ndarray, test_name: str +) -> np.ndarray: + """Return the estimator's per-class probability matrix, or ``SkipTestError``. + + The VMModel wrapper's ``predict_proba`` is binary-only (positive-class column + only), so reach the underlying estimator for the full per-class matrix. Skips + (rather than crashes) for models without a usable ``predict_proba`` + (metadata-only / precomputed single-column predictions) or a matrix whose width + does not match ``class_list``. + """ + raw_model = getattr(model, "model", None) + proba_fn = getattr(raw_model, "predict_proba", None) + if not callable(proba_fn): + raise SkipTestError( + f"Multiclass {test_name} requires per-class probabilities from the " + "underlying model's predict_proba, which is not available for this " + "model (e.g. metadata-only / precomputed predictions). Skipping." + ) + try: + y_prob = np.asarray(proba_fn(dataset.x_df())) + except Exception as e: + raise SkipTestError( + f"Multiclass {test_name} could not compute per-class probabilities " + f"({type(e).__name__}). Skipping." + ) from e + + n_classes = len(class_list) + if y_prob.ndim != 2 or y_prob.shape[1] != n_classes: + raise SkipTestError( + f"Multiclass {test_name} requires a per-class probability matrix with " + f"one column per class (got shape {getattr(y_prob, 'shape', None)} for " + f"{n_classes} classes). Skipping." + ) + return y_prob + + +def multiclass_proba( + model: VMModel, dataset: VMDataset, test_name: str +) -> MulticlassProba: + """Resolve aligned per-class probabilities for a multiclass one-vs-rest test. + + Aligns on the estimator's training classes, binarizes the true labels against + that full class list (so columns stay aligned even when the dataset is missing a + class), and reports which classes are actually present in the evaluated ``y``. + Raises ``SkipTestError`` for models that cannot supply a matching per-class + probability matrix, that were trained on fewer than 3 classes (one-vs-rest + multiclass metrics don't apply to a binary model), or when the evaluated + dataset's ``y`` contains a label the model was never trained on. + """ + class_list = resolve_class_list(model, dataset) + if len(class_list) < 3: + raise SkipTestError( + f"Multiclass {test_name} requires a model trained on at least 3 " + "classes for one-vs-rest multiclass metrics to apply, but this model " + f"appears to be binary (trained on {len(class_list)} classes). " + "Skipping." + ) + + y_prob = proba_matrix(model, dataset, class_list, test_name) + + y_true = np.asarray(dataset.y).flatten() + y_bin = label_binarize(y_true, classes=class_list) + + unknown_labels = sorted(set(np.unique(y_true)) - set(class_list)) + if unknown_labels: + raise SkipTestError( + f"Multiclass {test_name} found label(s) {unknown_labels} in the " + "evaluated dataset that the model was not trained on (not present in " + "the estimator's classes_). Skipping." + ) + + present_indices = [i for i, cls in enumerate(class_list) if np.any(y_true == cls)] + classes_present = [class_list[i] for i in present_indices] + + return MulticlassProba( + class_list=class_list, + classes_present=classes_present, + present_indices=present_indices, + y_bin=y_bin, + y_prob=y_prob, + ) diff --git a/validmind/tests/model_validation/statsmodels/GINITable.py b/validmind/tests/model_validation/statsmodels/GINITable.py index 5aa01d205..835634818 100644 --- a/validmind/tests/model_validation/statsmodels/GINITable.py +++ b/validmind/tests/model_validation/statsmodels/GINITable.py @@ -7,10 +7,9 @@ import numpy as np import pandas as pd from sklearn.metrics import roc_auc_score, roc_curve -from sklearn.preprocessing import label_binarize from validmind import RawData, tags, tasks -from validmind.errors import SkipTestError +from validmind.tests.model_validation.sklearn._multiclass_proba import multiclass_proba from validmind.vm_models import VMDataset, VMModel @@ -71,7 +70,7 @@ def GINITable(dataset: VMDataset, model: VMModel) -> Tuple[pd.DataFrame, RawData classes = np.unique(dataset.y) if len(classes) > 2: - return _multiclass_gini_table(model, dataset, classes) + return _multiclass_gini_table(model, dataset) y_true = np.ravel(dataset.y) # Flatten y_true to make it one-dimensional y_prob = dataset.y_prob(model) @@ -99,50 +98,24 @@ def GINITable(dataset: VMDataset, model: VMModel) -> Tuple[pd.DataFrame, RawData def _multiclass_gini_table( - model: VMModel, dataset: VMDataset, classes: np.ndarray + model: VMModel, dataset: VMDataset ) -> Tuple[pd.DataFrame, RawData]: """One-vs-rest AUC/GINI/KS for a multiclass model. Needs the full per-class probability matrix, which the stored single - probability column cannot provide, so we ask the underlying estimator for - it directly. Models without a usable ``predict_proba`` (metadata-only, - precomputed single-column probabilities) are skipped rather than crashed. + probability column cannot provide; the shared helper reaches the underlying + estimator, aligns the probability columns to the training class order and + skips models that cannot supply a matching matrix. """ - # The VMModel wrapper's y_prob is binary-only (positive-class column only), - # so reach the underlying estimator for the full per-class probability matrix. - raw_model = getattr(model, "model", None) - proba_fn = getattr(raw_model, "predict_proba", None) - if not callable(proba_fn): - raise SkipTestError( - "Multiclass GINITable requires per-class probabilities from the " - "underlying model's predict_proba, which is not available for this " - "model (e.g. metadata-only / precomputed predictions). Skipping." - ) - try: - y_prob = np.asarray(proba_fn(dataset.x_df())) - except Exception as e: - raise SkipTestError( - "Multiclass GINITable could not compute per-class probabilities " - f"({type(e).__name__}). Skipping." - ) from e - - n_classes = len(classes) - if y_prob.ndim != 2 or y_prob.shape[1] != n_classes: - raise SkipTestError( - "Multiclass GINITable requires a per-class probability matrix with " - f"one column per class (got shape {getattr(y_prob, 'shape', None)} " - f"for {n_classes} classes). Skipping." - ) - - # One-hot the true labels in the same class order predict_proba columns use - # (sklearn orders predict_proba columns by sorted class label == np.unique). - y_true = dataset.y.flatten() - y_bin = label_binarize(y_true, classes=classes) + aligned = multiclass_proba(model, dataset, "GINITable") + y_true = np.asarray(dataset.y).flatten() + y_bin = aligned.y_bin + y_prob = aligned.y_prob rows = [] raw_fpr = {} raw_tpr = {} - for i, cls in enumerate(classes): + for i, cls in zip(aligned.present_indices, aligned.classes_present): fpr, tpr, _ = roc_curve(y_bin[:, i], y_prob[:, i]) auc = roc_auc_score(y_bin[:, i], y_prob[:, i]) key = str(cls) @@ -152,9 +125,14 @@ def _multiclass_gini_table( {"Class": key, "AUC": auc, "GINI": 2 * auc - 1, "KS": max(tpr - fpr)} ) - # Micro-average across all one-vs-rest decisions. - micro_fpr, micro_tpr, _ = roc_curve(y_bin.ravel(), y_prob.ravel()) - micro_auc = roc_auc_score(y_bin, y_prob, average="micro", multi_class="ovr") + # Micro-average across the one-vs-rest decisions of the present classes. + present = aligned.present_indices + micro_fpr, micro_tpr, _ = roc_curve( + y_bin[:, present].ravel(), y_prob[:, present].ravel() + ) + micro_auc = roc_auc_score( + y_bin[:, present], y_prob[:, present], average="micro", multi_class="ovr" + ) raw_fpr["micro"] = micro_fpr raw_tpr["micro"] = micro_tpr rows.append(