feat(RHOAIENG-24177): Add unit test for Individual Consistency for TrustyAI Python Service

Author: artemsa223

pyproject.toml

@@ -15,6 +15,8 @@ dependencies = [
     "requests>=2.31.0,<3",
     "cryptography>=44.0.2,<45",
     "h5py>=3.13.0,<4",
+    "scikit-learn",
+    "aif360",
 ]
 
 [project.optional-dependencies]

src/core/metrics/fairness/fairness_metrics_utils.py

@@ -1,6 +1,7 @@
+from typing import Callable
 import numpy as np
 
-def filter_rows_by_inputs(data, filter_func):
+def filter_rows_by_inputs(data: np.ndarray, filter_func: Callable[[np.ndarray], bool]):
     return data[np.apply_along_axis(filter_func, 1, data)]
 
 def calculate_confusion_matrix(test: np.array, truth: np.array, positive_class: int) -> dict:

src/core/metrics/fairness/group/disparate_impact_ratio.py

@@ -1,5 +1,5 @@
 # pylint: disable=line-too-long
-from typing import List, Any
+from typing import List, Any, Union
 
 import numpy as np
 
@@ -33,8 +33,8 @@ def calculate_model(
 
     @staticmethod
     def calculate(
-        privileged: np.ndarray,
-        unprivileged: np.ndarray,
+        privileged: Union[int, np.ndarray],
+        unprivileged: Union[int, np.ndarray],
         favorable_output: int
         ) -> float:
         """

src/core/metrics/fairness/group/group_average_odds_difference.py

@@ -23,8 +23,8 @@ def calculate_model(
         :param samples a NumPy arrary of inputs to be used for testing fairness
         :param model the model to be tested for fairness
         :param privilege_columns a list of integers specifying the indices of the privileged columns
-        :param privilege_values a list of intergers specifying the privileged values
-        :param postive_class the favorable / desirable outputs
+        :param privilege_values a list of integers specifying the privileged values
+        :param positive_class the favorable / desirable outputs
         :param output_column the column index where the output is located
         return group average odds difference score
         """

src/core/metrics/fairness/group/group_average_predictive_value_difference.py

@@ -58,5 +58,5 @@ def privilege_filter(row):
         utp, utn, ufp, ufn = ucm["tp"], ucm["tn"], ucm["fp"], ucm["fn"]
         ptp, ptn, pfp, pfn = pcm["tp"], pcm["tn"], pcm["fp"], pcm["fn"]
 
-        return (utp / (utp + ufp) - ptp / (ptp + pfp + 1e-10)) / 2 + \
-            (ufn / (ufn + utn) - pfn / (pfn + ptn + 1e-10)) / 2
+        return (utp / (utp + ufp + 1e-10) - ptp / (ptp + pfp + 1e-10)) / 2 + \
+            (ufn / (ufn + utn + 1e-10) - pfn / (pfn + ptn + 1e-10)) / 2

src/core/metrics/fairness/group/group_statistical_parity_difference.py

@@ -26,8 +26,8 @@ def calculate_model(
         """
         outputs = model.predict(samples)
         data = np.append(samples, outputs, axis=1)
-        privileged = data[np.where(data[:, privilege_columns] == privilege_values)]
-        unprivileged = data[np.where(data[:, privilege_columns] != privilege_values)]
+        privileged = np.all(data[:, privilege_columns] == privilege_values, axis=1)
+        unprivileged = np.all(data[:, privilege_columns] != privilege_values, axis=1)
 
         return GroupStatisticalParityDifference.calculate(privileged, unprivileged, favorable_output)
 

tests/data/bank_churn_train.csv

@@ -1,27 +1,28 @@
 # pylint: disable=line-too-long, missing-function-docstring
-from typing import List, Optional
-
 from pytest import approx
 import numpy as np
 import pandas as pd
 
 from sklearn.linear_model import LogisticRegression
+from sklearn.neighbors import NearestNeighbors
 from sklearn.preprocessing import LabelEncoder
 
 from aif360.sklearn.metrics import (
     disparate_impact_ratio,
     statistical_parity_difference,
     average_odds_difference,
     average_predictive_value_difference,
+    consistency_score
 )
 
 from src.core.metrics.fairness.group.disparate_impact_ratio import DisparateImpactRatio
 from src.core.metrics.fairness.group.group_average_odds_difference import GroupAverageOddsDifference
 from src.core.metrics.fairness.group.group_average_predictive_value_difference import GroupAveragePredictiveValueDifference
 from src.core.metrics.fairness.group.group_statistical_parity_difference import GroupStatisticalParityDifference
+from src.core.metrics.fairness.individual.individual_consistency import IndividualConsistency
 
 df = pd.read_csv(
-    "https://raw.githubusercontent.com/trustyai-explainability/model-collection/8aa8e2e762c6d2b41dbcbe8a0035d50aa5f58c93/bank-churn/data/train.csv",
+    "tests/data/bank_churn_train.csv",
 )
 X = df.drop(columns=["Exited"], axis=1)
 y = df["Exited"]
@@ -34,8 +35,7 @@ def train_model():
         X[feature] = label_encoders[feature].fit_transform(X[feature])
     lr = LogisticRegression().fit(X, y)
 
-    y_pred = pd.DataFrame(lr.predict(X))
-    return y_pred
+    return pd.DataFrame(lr.predict(X))
 
 def truth_predict_output():
     y.index = X["Gender"]
@@ -58,20 +58,92 @@ def get_labeled_data():
     data_pred[:, -1] = y_pred.to_numpy().flatten()
     return data, data_pred
 
+
+def get_k_neighbors_function(k_value=5):
+    """Create a function that returns k nearest neighbors for a given input."""
+
+    def find_neighbors(sample, samples):
+        """Find k nearest neighbors for a given sample."""
+        if isinstance(sample, np.ndarray) and sample.ndim > 1:
+            sample = sample.flatten()
+
+        nbrs = NearestNeighbors(n_neighbors=k_value + 1, algorithm='ball_tree').fit(samples)
+        distances, indices = nbrs.kneighbors([sample])
+
+        neighbor_indices = indices[0][1:k_value + 1]
+        return samples[neighbor_indices]
+
+    return find_neighbors
+
+
+def get_processed_data(sample_size=None):
+    """Process data for testing individual consistency."""
+    categorical_features = ['Geography', 'Gender', 'Card Type', 'HasCrCard', 'IsActiveMember', 'Complain']
+    X_processed = X.copy()
+    for feature in categorical_features:
+        if feature in X_processed.columns:
+            le = LabelEncoder()
+            X_processed[feature] = le.fit_transform(X_processed[feature])
+
+    if sample_size is not None:
+        return X_processed.to_numpy()[:sample_size]
+    return X_processed.to_numpy()
+
+
+class MockPredictionProvider:
+    """Mock prediction provider for testing."""
+
+    def __init__(self, predictions):
+        self.predictions = predictions
+
+    def predict(self, x):
+        """Return prediction for input."""
+        if isinstance(x, np.ndarray) and x.ndim == 1:
+            x = x.reshape(1, -1)
+
+        result = []
+        for i in range(x.shape[0]):
+            if i < len(self.predictions):
+                result.append([self.predictions[i][0]])
+            else:
+                result.append([0])
+        return result
+
+
+class PerfectConsistencyProvider:
+    """Provider that always returns the same prediction."""
+
+    def predict(self, x):
+        if isinstance(x, np.ndarray) and x.ndim == 1:
+            x = x.reshape(1, -1)
+        return [[1] for _ in range(x.shape[0])]
+
+
+class RandomPredictionProvider:
+    """Provider that returns random predictions."""
+
+    def __init__(self, seed=42):
+        self.rng = np.random.RandomState(seed)
+
+    def predict(self, x):
+        if isinstance(x, np.ndarray) and x.ndim == 1:
+            x = x.reshape(1, -1)
+        return [[self.rng.randint(0, 2)] for _ in range(x.shape[0])]
+
 y, y_pred = truth_predict_output()
 privileged, unprivileged = get_privileged_unprivleged_split()
 data, data_pred = get_labeled_data()
 
 
 def test_disparate_impact_ratio():
-    dir = disparate_impact_ratio(y, prot_attr="Gender", priv_group="Male", pos_label=1)
+    dir_result = disparate_impact_ratio(y, prot_attr="Gender", priv_group="Male", pos_label=1)
 
     score = DisparateImpactRatio.calculate(
         privileged=privileged,
         unprivileged=unprivileged,
         favorable_output=1
     )
-    assert score == approx(dir, abs=1e-5)
+    assert score == approx(dir_result, abs=1e-5)
 
 
 def test_statistical_parity_difference():
@@ -98,7 +170,7 @@ def test_average_odds_difference():
         output_column=-1
     )
 
-    assert score == approx(aod, abs=0.2)
+    assert score == approx(aod, abs=1e-5)
 
 
 def test_average_predictive_value_difference():
@@ -114,3 +186,62 @@ def test_average_predictive_value_difference():
     )
 
     assert score == approx(apvd, abs=0.2)
+
+
+def test_individual_consistency():
+    """Test individual consistency calculation using AIF360's consistency_score as ground truth."""
+    X_sample = get_processed_data(sample_size=50)
+    y_pred_sample = y_pred.iloc[:50].to_numpy()
+
+    k = 5
+    cs_score = consistency_score(X_sample, y_pred_sample.flatten())
+
+    prediction_provider = MockPredictionProvider(y_pred_sample)
+    proximity_function = get_k_neighbors_function(k)
+
+    score = IndividualConsistency.calculate(
+        proximity_function=proximity_function,
+        samples=X_sample,
+        prediction_provider=prediction_provider
+    )
+
+    assert score == approx(cs_score, abs=0.2)
+
+
+def test_individual_consistency_perfect():
+    """Test individual consistency with a perfect consistency model."""
+    X_sample = get_processed_data(sample_size=20)
+
+    perfect_predictions = np.ones(20)
+
+    cs_score = consistency_score(X_sample, perfect_predictions)
+
+    proximity_function = get_k_neighbors_function(3)
+
+    consistency = IndividualConsistency.calculate(
+        proximity_function=proximity_function,
+        samples=X_sample,
+        prediction_provider=PerfectConsistencyProvider()
+    )
+
+    assert consistency == approx(cs_score, abs=0.2)
+
+
+def test_individual_consistency_imperfect():
+    """Test individual consistency with an inconsistent model."""
+    X_sample = get_processed_data(sample_size=20)
+
+    rng = np.random.RandomState(42)
+    random_predictions = rng.randint(0, 2, size=20)
+
+    cs_score = consistency_score(X_sample, random_predictions)
+
+    proximity_function = get_k_neighbors_function(3)
+
+    consistency = IndividualConsistency.calculate(
+        proximity_function=proximity_function,
+        samples=X_sample,
+        prediction_provider=RandomPredictionProvider(seed=42)
+    )
+
+    assert consistency == approx(cs_score, abs=0.2)