Merge pull request #5 from christinaexyou/add-fairness-metrics

feat(RHOAIENG-21045): Add fairness metrics and tests

Author: RobGeada

pyproject.toml

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

src/core/metrics/fairness/fairness_metrics_utils.py

@@ -0,0 +1,12 @@
+from typing import Callable
+import numpy as np
+
+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:
+    tp = np.sum((test == positive_class) & (truth == positive_class))
+    tn = np.sum((test != positive_class) & (truth != positive_class))
+    fp = np.sum((test == positive_class) & (truth != positive_class))
+    fn = np.sum((test != positive_class) & (truth == positive_class))
+    return {"tp": tp, "tn": tn, "fp": fp, "fn": fn}

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

@@ -0,0 +1,49 @@
+# pylint: disable=line-too-long
+from typing import List, Any, Union
+
+import numpy as np
+
+class DisparateImpactRatio:
+    """
+    Calculate disparate impact ratio (DIR).
+    """
+    @staticmethod
+    def calculate_model(
+        samples: np.ndarray,
+        model: Any,
+        privilege_columns: List[int],
+        privilege_values: List[int],
+        favorable_output: np.ndarray
+        ) -> float:
+        """
+        Calculate disparate impact ratio (DIR) for model outputs.
+        :param samples a NumPy array 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 integers specifying the privileged values
+        :param favorable_output the outputs that are considered favorable / desirable
+        return DIR score
+        """
+        outputs = model.predict(samples)
+        data = np.append(samples, outputs, axis=1)
+        privileged = np.sum(data[:, privilege_columns] == privilege_values)
+        unprivileged = np.sum(data[:, privilege_columns] != privilege_values)
+
+        return DisparateImpactRatio.calculate(privileged, unprivileged, favorable_output)
+
+    @staticmethod
+    def calculate(
+        privileged: Union[int, np.ndarray],
+        unprivileged: Union[int, np.ndarray],
+        favorable_output: int
+        ) -> float:
+        """
+        Calculate disparate impact ratio (DIR) when the labels are pre-calculated.
+        :param privileged a NumPy array with the privileged groups
+        :param unprivileged a NumPy array with the unprivileged groups
+        :param favorableOutput an output that is considered favorable / desirable
+        return DIR, between 0 and 1
+        """
+        probability_privileged = np.sum(privileged[:, -1] == favorable_output) / len(privileged)
+        probability_unprivileged = np.sum(unprivileged[:, -1] == favorable_output) / len(unprivileged)
+        return probability_unprivileged / probability_privileged

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

@@ -0,0 +1,64 @@
+# pylint: disable=line-too-long, too-many-arguments
+from typing import List, Any
+
+import numpy as np
+
+from src.core.metrics.fairness.fairness_metrics_utils import filter_rows_by_inputs, calculate_confusion_matrix
+
+class GroupAverageOddsDifference:
+    """
+    Calculate group average odds difference.
+    """
+    @ staticmethod
+    def calculate_model(
+        samples: np.ndarray,
+        model: Any,
+        privilege_columns: List[int],
+        privilege_values: List[int],
+        postive_class: List[int],
+        output_column: int
+    ):
+        """
+        Calculate group average odds difference for model outputs.
+        :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 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
+        """
+        outputs = model.predict(samples)
+        truth = np.append(samples, outputs, axis=1)
+
+        return GroupAverageOddsDifference.calculate(samples, truth, privilege_columns, privilege_values, postive_class, output_column)
+
+    @staticmethod
+    def calculate(test, truth, privilege_columns, privilege_values, positive_class, output_column):
+        """
+        Calculate group average odds difference when the labels are pre-calculated.
+        :param test a NumPy array representing the test data
+        :param truth a NumPy array representing the truth data
+        :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 positive_class the favorable / desirable outputs
+        :param output_column the column where the output is located
+        return group average odds difference, between -1 and 1
+        """
+        def privilege_filter(row):
+            return np.array_equal(row[privilege_columns], privilege_values)
+
+        test_privileged = filter_rows_by_inputs(test, privilege_filter)
+        test_unprivileged = filter_rows_by_inputs(test, lambda row: not privilege_filter(row))
+
+        truth_privileged = filter_rows_by_inputs(truth, privilege_filter)
+        truth_unprivileged = filter_rows_by_inputs(truth, lambda row: not privilege_filter(row))
+
+        ucm = calculate_confusion_matrix(test_unprivileged[:, output_column], truth_unprivileged[:, output_column], positive_class)
+        pcm = calculate_confusion_matrix(test_privileged[:, output_column], truth_privileged[:, output_column], positive_class)
+
+        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 + ufn + 1e-10) - ptp / (ptp + pfn + 1e-10)) / 2 + \
+            (ufp / (ufp + utn + 1e-10) - pfp / (pfp + ptn + 1e-10)) / 2

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

@@ -0,0 +1,62 @@
+# pylint: disable=line-too-long, too-many-arguments
+from typing import List, Any
+
+import numpy as np
+
+from src.core.metrics.fairness.fairness_metrics_utils import filter_rows_by_inputs, calculate_confusion_matrix
+
+class GroupAveragePredictiveValueDifference:
+    """
+    Calculate group average predictive value difference.
+    """
+    @staticmethod
+    def calculate_model(
+        samples: np.ndarray,
+        model: Any,
+        privilege_columns: List[int],
+        privilege_values: List[int],
+        positive_class: int,
+        output_column: int
+    ) -> float:
+        """
+        Calculate group average predictive value difference for model outputs.
+        :param samples a Numpy array 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 integers specifying the privileged values
+        :param positive_class the favorable / desirable outputs
+        :param output_column the column index where the output is located
+        """
+        outputs = model.predict(samples)
+        truth = np.append(samples, outputs, axis=1)
+        return GroupAveragePredictiveValueDifference.calculate(samples, truth, privilege_columns, privilege_values, positive_class, output_column)
+
+    @staticmethod
+    def calculate(test, truth, privilege_columns, privilege_values, positive_class, output_column):
+        """
+        Calculate group average predictive value difference when the labels are pre-calculated.
+        :param test a NumPy array representing the test data
+        :param truth a NumPy array representing the truth data
+        :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 positive_class the favorable / desirable outputs
+        :param output_column the column where the output is located
+        return group average predictive value difference, between -1 and 1
+        """
+        def privilege_filter(row):
+            return np.array_equal(row[privilege_columns], privilege_values)
+
+        test_privileged = filter_rows_by_inputs(test, privilege_filter)
+        test_unprivileged = filter_rows_by_inputs(test, lambda row: not privilege_filter(row))
+
+        truth_privileged = filter_rows_by_inputs(truth, privilege_filter)
+        truth_unprivileged = filter_rows_by_inputs(truth, lambda row: not privilege_filter(row))
+
+        ucm = calculate_confusion_matrix(test_unprivileged[:, output_column], truth_unprivileged[:, output_column], positive_class)
+        pcm = calculate_confusion_matrix(test_privileged[:, output_column], truth_privileged[:, output_column], positive_class)
+
+        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 + 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

@@ -0,0 +1,49 @@
+# pylint: disable=line-too-long
+from typing import List
+
+import numpy as np
+
+class GroupStatisticalParityDifference:
+    """
+    Calculate group statistical parity difference (SPD).
+    """
+    @staticmethod
+    def calculate_model(
+        samples: np.ndarray,
+        model,
+        privilege_columns: List[int],
+        privilege_values: List[int],
+        favorable_output,
+        ) -> float:
+        """
+        Calculate group statistical parity difference (SPD) for model outputs.
+        :param samples a NumPy array 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 integers specifying the privileged values
+        :param favorable_output the outputs that are considered favorable / desirable
+        return SPD score
+        """
+        outputs = model.predict(samples)
+        data = np.append(samples, outputs, axis=1)
+        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)
+
+    @staticmethod
+    def calculate(
+        privileged,
+        unprivileged,
+        favorable_output,
+        ) -> float:
+        """
+        Calculate statistical/demographic parity difference (SPD) when the labels are pre-calculated.
+        :param priviledged numPy array with the privileged groups
+        :param unpriviledged numPy array with the unpriviledged groups
+        :param favorableOutput an output that is considered favorable / desirable
+        return SPD, between 0 and 1
+        """
+        probability_privileged = np.sum(privileged[:, -1] == favorable_output) / len(privileged)
+        probability_unprivileged = np.sum(unprivileged[:, -1] == favorable_output) / len(unprivileged)
+        return probability_unprivileged - probability_privileged

src/core/metrics/fairness/individual/individual_consistency.py

@@ -0,0 +1,37 @@
+# pylint: disable=too-few-public-methods, line-too-long
+from typing import Any
+
+import numpy as np
+
+class IndividualConsistency:
+    """
+    Calculate individual fairness in terms of consistency of predictions across similar inputs.
+    :param proximity_function: a function that finds the top k similar inputs, given a reference input and a list of inputs
+    :param samples a list of inputs to be tested for consistency
+    :param predictionProvider the model under inspection
+    return the consistency measure
+    """
+    @staticmethod
+    def calculate(
+        proximity_function: Any,
+        samples: np.ndarray,
+        prediction_provider: Any
+    ) -> float:
+        """
+        Calculate individual fairness.
+        :param proximity_function: a function that finds the top k similar inputs, given a reference input and a list of inputs
+        :param samples a list of inputs to be tested for consistency
+        :param prediction_provider the model under inspection
+        return the consistency measure
+        """
+        consistency =  1
+        for sample in samples:
+            prediction_outputs = prediction_provider.predict(sample)
+            prediction_output = prediction_outputs[0]
+            neighbors = proximity_function(sample, samples)
+            neighbors_outputs = prediction_provider.predict(neighbors)
+            for output in prediction_outputs:
+                for neighbor_output in neighbors_outputs:
+                    if neighbor_output != output:
+                        consistency -= 1 / (len(neighbors) * len(prediction_output) * len(samples))
+        return consistency