wehub-resource-sync
241 lines
12 KiB
Python
241 lines
12 KiB
Python
import numpy as np
|
|
import pytest
|
|
import pandas as pd
|
|
import scipy.sparse as sp
|
|
|
|
from cleanlab.datalab.internal.issue_manager.underperforming_group import (
|
|
UnderperformingGroupIssueManager,
|
|
)
|
|
from sklearn.datasets import make_blobs, load_iris
|
|
|
|
SEED = 42
|
|
|
|
|
|
class TestUnderperformingGroupIssueManager:
|
|
def generate_pred_probs(self, N, K, labels, noisy=False):
|
|
pred_probs = np.full((N, K), 0.1)
|
|
pred_probs[np.arange(N), labels] = 0.9
|
|
pred_probs = pred_probs / np.sum(pred_probs, axis=-1, keepdims=True)
|
|
if noisy: # Swap columns of a class to generate incorrect predictions
|
|
pred_probs_slice = pred_probs[labels == 0]
|
|
pred_probs_slice[:, [0, 1]] = pred_probs_slice[:, [1, 0]]
|
|
pred_probs[labels == 0] = pred_probs_slice
|
|
return pred_probs
|
|
|
|
@pytest.fixture
|
|
def make_data(self, noisy=False):
|
|
def data(noisy=noisy):
|
|
N = 400
|
|
K = 4
|
|
features, labels = make_blobs(n_samples=N, centers=K, n_features=2, random_state=SEED)
|
|
pred_probs = self.generate_pred_probs(N, K, labels, noisy)
|
|
data = {"features": features, "pred_probs": pred_probs, "labels": labels}
|
|
return data
|
|
|
|
return data
|
|
|
|
@pytest.fixture
|
|
def iris_data(self):
|
|
iris_dataset = load_iris()
|
|
features, labels = iris_dataset.data, iris_dataset.target
|
|
K = len(iris_dataset.target_names)
|
|
N = features.shape[0]
|
|
pred_probs = self.generate_pred_probs(N, K, labels, noisy=True)
|
|
data = {"features": features, "pred_probs": pred_probs, "labels": labels}
|
|
return data
|
|
|
|
@pytest.fixture
|
|
def issue_manager(self, lab, make_data, monkeypatch):
|
|
data = make_data()
|
|
monkeypatch.setattr(lab._labels, "labels", data["labels"])
|
|
clustering_kwargs = {"eps": 2}
|
|
return UnderperformingGroupIssueManager(
|
|
datalab=lab, threshold=0.2, clustering_kwargs=clustering_kwargs
|
|
)
|
|
|
|
def test_find_issues_no_underperforming_group(self, issue_manager, make_data):
|
|
data = make_data()
|
|
features, labels, pred_probs = data["features"], data["labels"], data["pred_probs"]
|
|
N = len(labels)
|
|
issue_manager.find_issues(features=features, pred_probs=pred_probs)
|
|
issues, summary = issue_manager.issues, issue_manager.summary
|
|
assert np.sum(issues["is_underperforming_group_issue"]) == 0
|
|
expected_issue_mask = np.full(N, False, bool)
|
|
assert np.all(
|
|
issues["is_underperforming_group_issue"] == expected_issue_mask
|
|
), "Issue mask should be correct"
|
|
expected_scores = np.ones(N)
|
|
np.testing.assert_allclose(
|
|
issues["underperforming_group_score"],
|
|
expected_scores,
|
|
err_msg="Scores should be correct",
|
|
)
|
|
assert summary["issue_type"][0] == "underperforming_group"
|
|
assert summary["score"][0] == pytest.approx(1.0, abs=0.01)
|
|
# Check with cluster_ids param
|
|
issue_manager.find_issues(features=features, pred_probs=pred_probs, cluster_ids=labels)
|
|
issues_with_clabels, summary_with_clabels = issue_manager.issues, issue_manager.summary
|
|
pd.testing.assert_frame_equal(issues_with_clabels, issues)
|
|
pd.testing.assert_frame_equal(summary_with_clabels, summary)
|
|
|
|
def test_find_issues(self, issue_manager, make_data):
|
|
RELATIVE_TOLERANCE = 1e-3
|
|
data = make_data(noisy=True)
|
|
features, labels, pred_probs = data["features"], data["labels"], data["pred_probs"]
|
|
N = len(labels)
|
|
issue_manager.find_issues(features=features, pred_probs=pred_probs)
|
|
issues, summary = issue_manager.issues, issue_manager.summary
|
|
assert np.sum(issues["is_underperforming_group_issue"]) == 100
|
|
expected_issue_mask = np.zeros(N, bool)
|
|
expected_issue_mask[labels == 0] = True
|
|
assert np.all(
|
|
issues["is_underperforming_group_issue"] == expected_issue_mask
|
|
), "Issue mask should be correct"
|
|
expected_loss_ratio = 0.1429
|
|
expected_scores = np.ones(N)
|
|
expected_scores[labels == 0] = expected_loss_ratio
|
|
np.testing.assert_allclose(
|
|
issues["underperforming_group_score"],
|
|
expected_scores,
|
|
err_msg="Scores should be correct",
|
|
rtol=1e-3,
|
|
)
|
|
assert summary["issue_type"][0] == "underperforming_group"
|
|
assert summary["score"][0] == pytest.approx(expected_loss_ratio, rel=RELATIVE_TOLERANCE)
|
|
# Check with cluster_ids param
|
|
issue_manager.find_issues(features=features, pred_probs=pred_probs, cluster_ids=labels)
|
|
issues_with_clabels, summary_with_clabels = issue_manager.issues, issue_manager.summary
|
|
pd.testing.assert_frame_equal(issues_with_clabels, issues, rtol=RELATIVE_TOLERANCE)
|
|
pd.testing.assert_frame_equal(summary_with_clabels, summary, rtol=RELATIVE_TOLERANCE)
|
|
# With shifted cluster_ids
|
|
issue_manager.find_issues(features=features, pred_probs=pred_probs, cluster_ids=labels + 10)
|
|
issues_with_clabels, summary_with_clabels = issue_manager.issues, issue_manager.summary
|
|
pd.testing.assert_frame_equal(issues_with_clabels, issues, rtol=RELATIVE_TOLERANCE)
|
|
pd.testing.assert_frame_equal(summary_with_clabels, summary, rtol=RELATIVE_TOLERANCE)
|
|
|
|
def test_collect_info(self, issue_manager, make_data):
|
|
"""Test some values in the info dict.
|
|
|
|
Mainly focused on the clustering info.
|
|
"""
|
|
UNDERPERFORMING_CLUSTER_ID = 0
|
|
data = make_data(noisy=True)
|
|
features, pred_probs, labels = data["features"], data["pred_probs"], data["labels"]
|
|
issue_manager.find_issues(features=features, pred_probs=pred_probs)
|
|
info = issue_manager.info
|
|
assert "weighted_knn_graph" in info["statistics"]
|
|
assert "clustering" in info
|
|
# Check clustering info
|
|
clustering_info = info["clustering"]
|
|
assert clustering_info["algorithm"] == "DBSCAN"
|
|
assert clustering_info["params"]["metric"] == "precomputed"
|
|
assert clustering_info["stats"]["n_clusters"] == 4
|
|
# Test collect_info() with cluster_ids
|
|
issue_manager.find_issues(features=features, pred_probs=pred_probs, cluster_ids=labels)
|
|
info = issue_manager.info
|
|
assert "nearest_neighbor" not in info
|
|
assert "distance_to_nearest_neighbor" not in info
|
|
assert info["statistics"] == {}
|
|
# Check clustering info
|
|
clustering_info = info["clustering"]
|
|
assert clustering_info["algorithm"] is None
|
|
assert clustering_info["params"] == {}
|
|
assert clustering_info["stats"]["underperforming_cluster_id"] == UNDERPERFORMING_CLUSTER_ID
|
|
cluster_labels = clustering_info["stats"]["cluster_ids"]
|
|
issues = issue_manager.issues
|
|
issue_indices = issues.index[issues["is_underperforming_group_issue"]].values
|
|
assert np.all(
|
|
cluster_labels[issue_indices] == UNDERPERFORMING_CLUSTER_ID
|
|
), "All samples with issue should belong to underperforming cluster"
|
|
|
|
np.testing.assert_equal(clustering_info["stats"]["cluster_ids"], labels)
|
|
|
|
def test_no_meaningful_clusters(self, issue_manager, make_data, monkeypatch):
|
|
np.random.seed(SEED)
|
|
data = make_data()
|
|
k = 10
|
|
N = 20
|
|
# Generate sparse data that cannot be clustered by DBSCAN
|
|
features = np.random.uniform(-100, 100, (N, 2))
|
|
# Dummy pred_probs and labels for running issue manager
|
|
pred_probs = data["pred_probs"][:N]
|
|
monkeypatch.setattr(issue_manager.datalab._labels, "labels", data["labels"][:N])
|
|
monkeypatch.setattr(issue_manager, "k", k) # Ensure that k is smaller than N
|
|
exception_pattern = "No meaningful clusters"
|
|
with pytest.raises(ValueError, match=exception_pattern):
|
|
issue_manager.find_issues(features=features, pred_probs=pred_probs)
|
|
# Cluster labels passed containing all outliers
|
|
cluster_ids = np.full(N, -1) # -1 is the outlier label for DBSCAN
|
|
with pytest.raises(ValueError, match=exception_pattern):
|
|
issue_manager.find_issues(
|
|
features=features, pred_probs=pred_probs, cluster_ids=cluster_ids
|
|
)
|
|
# Empty cluster ids
|
|
with pytest.raises(ValueError, match=exception_pattern):
|
|
issue_manager.find_issues(
|
|
features=features, pred_probs=pred_probs, cluster_ids=np.array([], dtype=int)
|
|
)
|
|
|
|
def test_min_cluster_samples(self, lab, issue_manager, make_data):
|
|
data = make_data()
|
|
features, pred_probs, labels = data["features"], data["pred_probs"], data["labels"]
|
|
labels[:3] = max(labels) + 1 # New cluster with very few samples
|
|
n_clusters = len(set(labels))
|
|
# Check if small cluster is filtered
|
|
issue_manager.find_issues(features=features, pred_probs=pred_probs, cluster_ids=labels)
|
|
clustering_info = issue_manager.info["clustering"]
|
|
assert clustering_info["stats"]["n_clusters"] == n_clusters - 1
|
|
|
|
# New issue manager to consider small cluster as well
|
|
issue_manager = UnderperformingGroupIssueManager(
|
|
datalab=lab, threshold=0.2, min_cluster_samples=3
|
|
)
|
|
issue_manager.find_issues(features=features, pred_probs=pred_probs, cluster_ids=labels)
|
|
clustering_info = issue_manager.info["clustering"]
|
|
assert clustering_info["stats"]["n_clusters"] == n_clusters
|
|
|
|
def test_find_issues_feature_subset(self, issue_manager, iris_data, monkeypatch):
|
|
features, pred_probs, labels = (
|
|
iris_data["features"],
|
|
iris_data["pred_probs"],
|
|
iris_data["labels"],
|
|
)
|
|
# TODO: Better asserts required. Ideally -> 3 clusters, 50 samples with issue.
|
|
monkeypatch.setattr(issue_manager.datalab._labels, "labels", labels)
|
|
monkeypatch.setattr(issue_manager, "clustering_kwargs", {"eps": 0.5})
|
|
# Find underperforming group based on one feature
|
|
single_feature = features[:, 0].reshape(-1, 1)
|
|
issue_manager.find_issues(features=single_feature, pred_probs=pred_probs)
|
|
assert np.sum(issue_manager.issues["is_underperforming_group_issue"]) == 16
|
|
# Find underperforming group based on two features
|
|
monkeypatch.setattr(issue_manager, "info", {})
|
|
issue_manager.find_issues(features=features[:, [1, 3]], pred_probs=pred_probs)
|
|
assert np.sum(issue_manager.issues["is_underperforming_group_issue"]) == 49
|
|
# Find underperforming group based on all features
|
|
monkeypatch.setattr(issue_manager, "info", {})
|
|
issue_manager.find_issues(features=features, pred_probs=pred_probs)
|
|
assert np.sum(issue_manager.issues["is_underperforming_group_issue"]) == 48
|
|
|
|
def test_knn_graph_change(self, issue_manager):
|
|
dist_matrix = np.random.randint(1, 5, size=(10, 10))
|
|
np.fill_diagonal(dist_matrix, 0) # Make diagonal 0 to mimic distance matrix
|
|
knn_graph = sp.csr_matrix(dist_matrix)
|
|
nnz_before_clustering = knn_graph.nnz
|
|
issue_manager.perform_clustering(knn_graph)
|
|
nnz_after_clustering = knn_graph.nnz
|
|
assert nnz_before_clustering == nnz_after_clustering
|
|
|
|
def test_report(self, issue_manager, make_data):
|
|
data = make_data()
|
|
features, pred_probs = data["features"], data["pred_probs"]
|
|
issue_manager.find_issues(features=features, pred_probs=pred_probs)
|
|
report = issue_manager.report(
|
|
issues=issue_manager.issues,
|
|
summary=issue_manager.summary,
|
|
info=issue_manager.info,
|
|
)
|
|
assert isinstance(report, str)
|
|
assert (
|
|
"--------------- underperforming_group issues ---------------\n\nNumber of examples with this issue"
|
|
) in report
|