import numpy as np import pandas as pd import pytest from hypothesis import example, given, settings from hypothesis import strategies as st from sklearn.linear_model import LogisticRegression as LogReg from sklearn.neighbors import NearestNeighbors from cleanlab import count, outlier from cleanlab.benchmarking.noise_generation import ( generate_noise_matrix_from_trace, generate_noisy_labels, ) from cleanlab.count import get_confident_thresholds from cleanlab.internal.label_quality_utils import get_normalized_entropy from cleanlab.outlier import OutOfDistribution def make_data( means=[[3, 2], [7, 7], [0, 8]], covs=[[[5, -1.5], [-1.5, 1]], [[1, 0.5], [0.5, 4]], [[5, 1], [1, 5]]], sizes=[80, 40, 40], avg_trace=0.8, seed=1, # set to None for non-reproducible randomness ): np.random.seed(seed=seed) m = len(means) # number of classes n = sum(sizes) local_data = [] labels = [] test_data = [] test_labels = [] for idx in range(m): local_data.append( np.random.multivariate_normal(mean=means[idx], cov=covs[idx], size=sizes[idx]) ) test_data.append( np.random.multivariate_normal(mean=means[idx], cov=covs[idx], size=sizes[idx]) ) labels.append(np.array([idx for i in range(sizes[idx])])) test_labels.append(np.array([idx for i in range(sizes[idx])])) X_train = np.vstack(local_data) true_labels_train = np.hstack(labels) X_test = np.vstack(test_data) true_labels_test = np.hstack(test_labels) # Compute p(true_label=k) py = np.bincount(true_labels_train) / float(len(true_labels_train)) noise_matrix = generate_noise_matrix_from_trace( m, trace=avg_trace * m, py=py, valid_noise_matrix=True, seed=seed, ) # Generate our noisy labels using the noise_matrix. s = generate_noisy_labels(true_labels_train, noise_matrix) ps = np.bincount(s) / float(len(s)) # Compute inverse noise matrix inv = count.compute_inv_noise_matrix(py, noise_matrix, ps=ps) # Estimate pred_probs latent = count.estimate_py_noise_matrices_and_cv_pred_proba( X=X_train, labels=s, cv_n_folds=3, ) label_errors_mask = s != true_labels_train return { "X_train": X_train, "true_labels_train": true_labels_train, "X_test": X_test, "true_labels_test": true_labels_test, "labels": s, "label_errors_mask": label_errors_mask, "ps": ps, "py": py, "noise_matrix": noise_matrix, "inverse_noise_matrix": inv, "est_py": latent[0], "est_nm": latent[1], "est_inv": latent[2], "cj": latent[3], "pred_probs": latent[4], "m": m, "n": n, } # Global to be used by all test methods. Only compute this once for speed. data = make_data() def test_class_wrong_info_assert_valid_inputs(): features = data["X_train"] pred_probs = data["pred_probs"] OOD = OutOfDistribution() # TESTING: _assert_valid_inputs() asserts correct errors in fit try: OOD.fit() except Exception as e: assert "Not enough information to compute scores" in str(e) with pytest.raises(ValueError) as e: OOD.fit() try: OOD.fit(features=features, pred_probs=pred_probs) except Exception as e: assert "Cannot fit to OOD Estimator to both features and pred_probs" in str(e) with pytest.raises(ValueError) as e: OOD.fit(features=features, pred_probs=pred_probs) OOD = OutOfDistribution() features_flat = np.ravel(features) features_extra_dim = features[np.newaxis] try: OOD.fit(features=features_flat) except Exception as e: assert "array needs to be of shape (N, M)" in str(e) with pytest.raises(ValueError) as e: OOD.fit(features=features_flat) try: OOD.fit(features=features_extra_dim) except Exception as e: assert "array needs to be of shape (N, M)" in str(e) with pytest.raises(ValueError) as e: OOD.fit(features=features_extra_dim) # TODO: DO WE NEED TO TESTING: _assert_valid_inputs() asserts correct errors in score? def test_class_wrong_info_fit_ood(): features = data["X_test"] pred_probs = data["pred_probs"] labels = data["labels"] # TESTING: wrong param in params dict try: OOD = OutOfDistribution(params={"strange_param": -1}) except Exception as e: assert "strange_param" in str(e) with pytest.raises(ValueError) as e: OOD = OutOfDistribution(params={"strange_param": -1}) #### SCORE wrong info # TESTING: calling score before any fitting OOD = OutOfDistribution() try: OOD.score(features=features) except Exception as e: assert "OOD estimator needs to be fit on features first" in str(e) with pytest.raises(ValueError) as e: OOD.score(features=features) try: OOD.score(pred_probs=pred_probs) except Exception as e: assert "OOD estimator needs to be fit on pred_probs first" in str(e) with pytest.raises(ValueError) as e: OOD.score(pred_probs=pred_probs) # TESTING: calling scoring with opposite fitting OOD_outlier = OutOfDistribution() OOD_outlier.fit(features=features) try: OOD_outlier.score(pred_probs=pred_probs) except Exception as e: assert "OOD estimator needs to be fit on pred_probs first" in str(e) with pytest.raises(ValueError) as e: OOD_outlier.score(pred_probs=pred_probs) OOD_ood = OutOfDistribution() OOD_ood.fit(pred_probs=pred_probs, labels=labels) try: OOD_ood.score(features=features) except Exception as e: assert "OOD estimator needs to be fit on features first" in str(e) with pytest.raises(ValueError) as e: OOD_ood.score(features=features) def test_class_params_logic(): features = data["X_test"] pred_probs = data["pred_probs"] # TESTING: params dict is a copy params_dict = {"k": 10, "t": 5} OOD = OutOfDistribution(params=params_dict) OOD.fit(features=features) ood_params = OOD.params params_dict = params_dict.update({"k": 20}) assert ood_params == OOD.params # test calling functions with different params performs differently @pytest.mark.filterwarnings("ignore::UserWarning") # Should be 7 warnings def test_class_public_func(): features = data["X_test"] pred_probs = data["pred_probs"] labels = data["true_labels_test"] # Fit Logistic Regression model on X_train and estimate train pred_probs logreg = LogReg(solver="lbfgs") logreg.fit(data["X_train"], data["true_labels_train"]) train_pred_probs = logreg.predict_proba(data["X_train"]) # Get knn and confident_thresholds and pass them into OOD object initialization for testing already fitted logic knn = NearestNeighbors(n_neighbors=7).fit(data["X_train"]) confident_thresholds = get_confident_thresholds( pred_probs=train_pred_probs, labels=data["true_labels_train"] ) OOD_outlier_already_fit = OutOfDistribution(params={"knn": knn}) OOD_ood_already_fit = OutOfDistribution(params={"confident_thresholds": confident_thresholds}) #### TESTING INITIALIZATION # TESTING knn and confident_thresholds passed during initialization correctly assert OOD_outlier_already_fit.params["knn"].n_neighbors == 7 assert (OOD_ood_already_fit.params["confident_thresholds"] == confident_thresholds).all() #### TESTING FIT: # Test fitting OOD object without labels and adjust_pred_probs=False OOD_ood = OutOfDistribution(params={"adjust_pred_probs": False}) OOD_ood.fit(pred_probs=data["pred_probs"], labels=None) # Warning assert OOD_ood.params["adjust_pred_probs"] is False OOD_ood.score( pred_probs=pred_probs ) # This should be ok without passing in labels since we are not adjusting # Testing regular fit OOD_ood = OutOfDistribution() print(OOD_ood.params) OOD_ood.fit(pred_probs=pred_probs, labels=labels) print(OOD_ood.params) OOD_outlier = OutOfDistribution() OOD_outlier.fit(features=features) print(OOD_outlier.params) assert OOD_ood.params["confident_thresholds"] is not None and OOD_ood.params["knn"] is None assert ( OOD_outlier.params["knn"] is not None and OOD_outlier.params["confident_thresholds"] is None ) assert OOD_ood.params is not None and OOD_outlier.params is not None # Testing calling fit on already fitted function (should not overwrite but warn) OOD_outlier_already_fit.fit(features=features) # Warning assert OOD_outlier_already_fit.params["knn"].n_neighbors == 7 # Assert not overwritten OOD_ood_already_fit.fit(pred_probs=pred_probs, labels=labels) # Warning assert ( OOD_ood_already_fit.params["confident_thresholds"] == confident_thresholds ).all() # Assert not overwritten # Testing fit uses correct metrics given feature dimensionality X_small = np.random.rand(20, 3) OOD_euclidean = OutOfDistribution() OOD_euclidean.fit(features=X_small) # The metric attribute is the pairwise distance function implemented in scipy, use __name__ to get the name of the function assert OOD_euclidean.params["knn"].metric.__name__ == "euclidean" X_small_with_ood = np.vstack([X_small, [999999.0] * 3]) euclidean_score = OOD_euclidean.score(features=X_small_with_ood) assert (np.max(euclidean_score) <= 1) and (np.min(euclidean_score) >= 0) assert np.argmin(euclidean_score) == (euclidean_score.shape[0] - 1) # Re-run tests with high dimensional dataset X_large = np.hstack([np.zeros((200, 400)), np.random.rand(200, 1)]) OOD_cosine = OutOfDistribution() OOD_cosine.fit(features=X_large) assert OOD_cosine.params["knn"].metric == "cosine" X_large_with_ood = np.vstack([X_large, [999999.0] * 401]) cosine_score = OOD_cosine.score(features=X_large_with_ood) assert (np.max(cosine_score) <= 1) and (np.min(cosine_score) >= 0) assert np.argmin(cosine_score) == (cosine_score.shape[0] - 1) #### TESTING SCORE ood_score = OOD_ood.score(pred_probs=pred_probs) outlier_score = OOD_outlier.score(features=features) assert ood_score is not None and outlier_score is not None assert np.sum(ood_score) != np.sum(outlier_score) #### TESTING FIT SCORE OOD_ood_fs = OutOfDistribution() ood_score_fs = OOD_ood_fs.fit_score(pred_probs=pred_probs, labels=labels) OOD_outlier_fs = OutOfDistribution() outlier_score_fs = OOD_outlier_fs.fit_score(features=features) assert ( OOD_ood_fs.params["confident_thresholds"] is not None and OOD_ood_fs.params["knn"] is None ) assert ( OOD_ood_fs.params["confident_thresholds"] == OOD_ood.params["confident_thresholds"] ).all() assert ( OOD_outlier.params["knn"] is not None and OOD_outlier.params["confident_thresholds"] is None ) assert ood_score_fs is not None and outlier_score_fs is not None assert np.sum(outlier_score_fs) - np.sum(outlier_score) < 1 # scores are similar assert np.sum(ood_score_fs) - np.sum(ood_score) < 1 # scores are similar # Testing calling fit_score on already fitted function score_outlier_fs = OOD_outlier_already_fit.fit_score(features=features) # Warning assert OOD_outlier_already_fit.params["knn"].n_neighbors == 7 score_ood_fs = OOD_ood_already_fit.fit_score(pred_probs=pred_probs, labels=labels) # Warning assert (OOD_ood_already_fit.params["confident_thresholds"] == confident_thresholds).all() assert ( score_outlier_fs is not None and score_ood_fs is not None ) # Assert scores still calculated # Testing calling fit_score repeatedly on already fitted function does not fit it more, just scores same as before score_outlier_fs1 = OOD_outlier_already_fit.fit_score(features=features) # Warning score_ood_fs1 = OOD_ood_already_fit.fit_score(pred_probs=pred_probs, labels=labels) # Warning assert (score_outlier_fs == score_outlier_fs1).all() and (score_ood_fs1 == score_ood_fs).all() # Testing scores calculated during already fitted fit_score identical to scores calculated during score. score_outlier_s = OOD_outlier_already_fit.score(features=features) score_ood_s = OOD_ood_already_fit.score(pred_probs=pred_probs) assert (score_outlier_fs == score_outlier_s).all() and (score_ood_fs == score_ood_s).all() def test_get_ood_features_scores(): ood = OutOfDistribution() X_train = data["X_train"] X_test = data["X_test"] # Create OOD datapoint X_ood = np.array([[999999999.0, 999999999.0]]) # Add OOD datapoint to X_test X_test_with_ood = np.vstack([X_test, X_ood]) # Fit nearest neighbors on X_train knn = NearestNeighbors(n_neighbors=5, metric="euclidean").fit(X_train) # Get KNN distance as outlier score k = 5 knn_distance_to_score, _ = ood._get_ood_features_scores(features=X_test_with_ood, knn=knn, k=k) # Checking that X_ood has the smallest outlier score among all the datapoints assert np.argmin(knn_distance_to_score) == (knn_distance_to_score.shape[0] - 1) # Get KNN distance as outlier score without passing k # By default k=10 is used or k = n_neighbors when k > n_neighbors extracted from the knn knn_distance_to_score, _ = ood._get_ood_features_scores(features=X_test_with_ood, knn=knn) # Checking that X_ood has the smallest outlier score among all the datapoints assert np.argmin(knn_distance_to_score) == (knn_distance_to_score.shape[0] - 1) # Get KNN distance as outlier score passing k and t > 1 large_t_knn_distance_to_score, _ = ood._get_ood_features_scores( features=X_test_with_ood, knn=knn, k=k, t=5 ) # Checking that X_ood has the smallest outlier score among all the datapoints assert np.argmin(large_t_knn_distance_to_score) == (large_t_knn_distance_to_score.shape[0] - 1) # Get KNN distance as outlier score passing k and t < 1 small_t_knn_distance_to_score, _ = ood._get_ood_features_scores( features=X_test_with_ood, knn=knn, k=k, t=0.002 ) # Checking that X_ood has the smallest outlier score among all the datapoints assert np.argmin(small_t_knn_distance_to_score) == (small_t_knn_distance_to_score.shape[0] - 1) assert np.sum(small_t_knn_distance_to_score) >= np.sum(large_t_knn_distance_to_score) @pytest.mark.filterwarnings("ignore::UserWarning") def test_default_k_and_model_get_ood_features_scores(): # Testing using 'None' as model param and correct setting of default k as max_k # Create dataset with OOD example X = data["X_test"] X_ood = np.array([[999999999.0, 999999999.0]]) X_with_ood = np.vstack([X, X_ood]) instantiated_k = 10 # Create NN class object with small instantiated k and fit on data knn = NearestNeighbors(n_neighbors=instantiated_k, metric="euclidean").fit(X_with_ood) ood = OutOfDistribution() avg_knn_distances_default_model, _ = ood._get_ood_features_scores( features=X_with_ood, k=instantiated_k, # this should use default estimator (same as above) and k = instantiated_k ) avg_knn_distances_default_k, knn2 = ood._get_ood_features_scores( features=X_with_ood, # default k should be set to 10 == instantiated_k ) assert isinstance(knn2, type(knn)) avg_knn_distances, _ = ood._get_ood_features_scores( features=None, knn=knn, k=25, # this should throw user warn, k should be set to instantiated_k ) # Score sums should be equal because the three estimators used have identical params and fit assert avg_knn_distances.sum() == avg_knn_distances_default_model.sum() assert avg_knn_distances_default_k.sum() == avg_knn_distances.sum() avg_knn_distances_large_k, _ = ood._get_ood_features_scores( features=X_with_ood, k=25, # this should use default estimator and k = 25 ) avg_knn_distances_tiny_k, _ = ood._get_ood_features_scores( features=None, knn=knn, k=1, # this should use knn estimator and k = 1 ) avg_knn_distances_tiny_k_default, _ = ood._get_ood_features_scores( features=X_with_ood, k=1, # this should use default estimator and k = 1 ) # Score sums should be different because k = user param for estimators and k != 10. assert avg_knn_distances_tiny_k.sum() != avg_knn_distances.sum() assert avg_knn_distances_large_k.sum() != avg_knn_distances.sum() assert avg_knn_distances_tiny_k_default.sum() != avg_knn_distances_default_model.sum() # Test that when knn is None ValueError raised if passed in k > len(features) try: ood._get_ood_features_scores( features=X_with_ood, knn=None, k=len(X_with_ood) + 1, # this should throw ValueError, k ! > len(features) ) except Exception as e: assert "nearest neighbors" in str(e) with pytest.raises(ValueError) as e: ood._get_ood_features_scores( features=X_with_ood, knn=None, k=len(X_with_ood) + 1, # this should throw ValueError, k ! > len(features) ) def test_not_enough_info_get_ood_features_scores(): # Testing calling function with not enough information to calculate outlier scores ood = OutOfDistribution() try: ood._get_ood_features_scores( features=None, knn=None, # this should throw TypeError because knn=None and features=None ) except Exception as e: assert "Both knn and features arguments" in str(e) with pytest.raises(ValueError) as e: ood._get_ood_features_scores( features=None, knn=None, # this should throw TypeError because knn=None and features=None ) def test_ood_predictions_scores(): # Create and add OOD datapoint to test set X = data["X_test"] means = [[3, 2], [7, 7], [0, 8]] X_ood = np.array(means).mean(axis=0) X_with_ood = np.vstack([X, X_ood]) y = data["true_labels_test"] y_with_ood = np.hstack([y, data["true_labels_train"][1]]) # Fit Logistic Regression model on X_train and estimate pred_probs logreg = LogReg(solver="lbfgs") logreg.fit(data["X_train"], data["true_labels_train"]) pred_probs = logreg.predict_proba(X_with_ood) ### Test non-adjusted OOD score logic ood_predictions_scores_entropy, _ = outlier._get_ood_predictions_scores( pred_probs=pred_probs, adjust_pred_probs=False, ) # adjust pred probs should be False by default ood_predictions_scores_least_confidence, _ = outlier._get_ood_predictions_scores( pred_probs=pred_probs, method="least_confidence", adjust_pred_probs=False, ) ood_predictions_scores_gen, _ = outlier._get_ood_predictions_scores( pred_probs=pred_probs, method="gen", adjust_pred_probs=False, M=3, # Totally three classes ) # check OOD scores calculated correctly assert (1.0 - get_normalized_entropy(pred_probs) == ood_predictions_scores_entropy).all() assert (pred_probs.max(axis=1) == ood_predictions_scores_least_confidence).all() assert ood_predictions_scores_gen.max() < 1 assert ood_predictions_scores_gen.min() > 0 assert np.where(np.sort(ood_predictions_scores_entropy) == ood_predictions_scores_entropy[-1])[ 0 ] < 0.02 * len(ood_predictions_scores_entropy) assert np.where( np.sort(ood_predictions_scores_least_confidence) == ood_predictions_scores_least_confidence[-1] )[0] < 0.02 * len(ood_predictions_scores_least_confidence) assert np.where(np.sort(ood_predictions_scores_gen) == ood_predictions_scores_gen[-1])[ 0 ] < 0.02 * len(ood_predictions_scores_gen) ### Test adjusted OOD score logic ( ood_predictions_scores_adj_entropy, confident_thresholds_adj_entropy, ) = outlier._get_ood_predictions_scores( pred_probs=pred_probs, labels=y_with_ood, adjust_pred_probs=True, method="entropy", ) ( ood_predictions_scores_adj_least_confidence, confident_thresholds_adj_least_confidence, ) = outlier._get_ood_predictions_scores( pred_probs=pred_probs, labels=y_with_ood, adjust_pred_probs=True, method="least_confidence", ) # test confident thresholds calculated correctly confident_thresholds = get_confident_thresholds( labels=y_with_ood, pred_probs=pred_probs, multi_label=False ) assert (confident_thresholds == confident_thresholds_adj_entropy).all() assert (confident_thresholds_adj_least_confidence == confident_thresholds_adj_entropy).all() # check adjusted OOD scores different from non adjust OOD scores assert not (ood_predictions_scores_adj_entropy == ood_predictions_scores_entropy).all() assert not ( ood_predictions_scores_adj_least_confidence == ood_predictions_scores_least_confidence ).all() ### Test pre-calculated confident thresholds logic ood_predictions_scores_2, confident_thresholds_2 = outlier._get_ood_predictions_scores( pred_probs=pred_probs, confident_thresholds=confident_thresholds, adjust_pred_probs=True, ) assert (confident_thresholds_2 == confident_thresholds).all() assert (ood_predictions_scores_2 == ood_predictions_scores_adj_entropy).all() # test using labels list type works y_with_ood_list = y_with_ood.tolist() ( ood_predictions_scores_adj_entropy_list, confident_thresholds_adj_entropy_list, ) = outlier._get_ood_predictions_scores( pred_probs=pred_probs, labels=y_with_ood_list, adjust_pred_probs=True, method="entropy", ) # test using labels series type works y_with_ood_series = pd.Series(y_with_ood) ( ood_predictions_scores_adj_entropy_series, confident_thresholds_adj_entropy_series, ) = outlier._get_ood_predictions_scores( pred_probs=pred_probs, labels=y_with_ood_series, adjust_pred_probs=True, method="entropy", ) assert (confident_thresholds_adj_entropy_list == confident_thresholds_adj_entropy).all() assert (confident_thresholds_adj_entropy_series == confident_thresholds_adj_entropy).all() @pytest.mark.filterwarnings("ignore::UserWarning") def test_wrong_info_get_ood_predictions_scores(): # Test calling function with not enough information to calculate ood scores try: outlier._get_ood_predictions_scores( pred_probs=data["pred_probs"], labels=None, adjust_pred_probs=True, # this should throw ValueError because knn=None and features=None ) except Exception as e: assert "Cannot calculate adjust_pred_probs without labels" in str(e) with pytest.raises(ValueError) as e: outlier._get_ood_predictions_scores( pred_probs=data["pred_probs"], labels=None, adjust_pred_probs=True, # this should throw ValueError because knn=None and features=None ) # Test calling function with not enough information to calculate ood scores try: outlier._get_ood_predictions_scores( pred_probs=data["pred_probs"], adjust_pred_probs=True, # this should throw ValueError because knn=None and features=None ) except Exception as e: assert "Cannot calculate adjust_pred_probs without labels" in str(e) with pytest.raises(ValueError) as e: outlier._get_ood_predictions_scores( pred_probs=data["pred_probs"], adjust_pred_probs=True, # this should throw ValueError because not enough data provided ) # Test calling function with not a real method try: outlier._get_ood_predictions_scores( pred_probs=data["pred_probs"], labels=data["labels"], adjust_pred_probs=True, method="not_a_real_method", # this should throw ValueError because method not real method ) except Exception as e: assert "not a valid OOD scoring" in str(e) with pytest.raises(ValueError) as e: outlier._get_ood_predictions_scores( pred_probs=data["pred_probs"], labels=data["labels"], adjust_pred_probs=True, method="not_a_real_method", # this should throw ValueError because method not real method ) # Test calling function with too much information to calculate ood scores outlier._get_ood_predictions_scores( pred_probs=data["pred_probs"], labels=data["labels"], adjust_pred_probs=False, # this should user warning because provided info is not used ) @given( fill_value=st.floats( min_value=5 * float(np.finfo(np.float64).eps), max_value=5, exclude_min=False, allow_subnormal=False, allow_infinity=False, allow_nan=False, ), K=st.integers(min_value=2, max_value=100), ) @example(K=1, fill_value=0.0) @settings(deadline=None) def test_scores_for_identical_examples(fill_value, K): N = 100 features = np.full((N, K), fill_value=fill_value) ood = OutOfDistribution() scores = ood.fit_score(features=features, verbose=False) # Dataset with only expected_score = np.full(N, 1.0) np.testing.assert_array_equal( scores, expected_score, err_msg=f"The calculated distances were {ood.params['knn'].kneighbors()}", ) @given(K=st.integers(min_value=2, max_value=100)) @settings(max_examples=10000, deadline=None) def test_scores_for_identical_examples_across_rows(K): N = 100 fill_value = np.random.random(K) features = np.full((N, K), fill_value=fill_value) ood = OutOfDistribution() scores = ood.fit_score(features=features, verbose=False) # Dataset with only expected_score = np.full(N, 1.0) np.testing.assert_array_equal( scores, expected_score, err_msg=f"The calculated distances were {ood.params['knn'].kneighbors()}", ) if K < 4: # This little changes should not affect euclidean calculation features += np.random.random(features.shape) * 1e-10 ood = OutOfDistribution() scores = ood.fit_score(features=features, verbose=False) np.testing.assert_array_equal( scores, expected_score, err_msg=f"The calculated distances were {ood.params['knn'].kneighbors()}", )