feature: equalised odds

.gitignore

@@ -1,4 +1,5 @@
 VERSION
+*.swp
 
 tmp/
 .idea/

aequitas/core/metrics.py

@@ -3,6 +3,7 @@
 import numpy as np
 import typing
 
+
 Probability = float
 Condition = typing.Callable[[np.array], np.array]
 ConditionOrScalar = typing.Union[Condition, Scalar]
@@ -28,11 +29,12 @@ def probability(x: np.array, x_cond: ConditionOrScalar) -> Probability:
 
 
 def conditional_probability(
-        y: np.array,
-        y_cond: ConditionOrScalar,
-        x: np.array,
-        x_cond: ConditionOrScalar,
+    y: np.array,
+    y_cond: ConditionOrScalar,
+    x: np.array,
+    x_cond: ConditionOrScalar,
 ) -> Probability:
+    """Computes the probability of y given x"""
     y_cond = __ensure_is_condition(y_cond)
     x_cond = __ensure_is_condition(x_cond)
     x_is_x_value = x_cond(x)
@@ -69,4 +71,55 @@ def discrete_demographic_parities(x: np.array, y: np.array, y_cond: ConditionOrS
     return np.array(probabilities)
 
 
+def __compute_false_rates(x: np.array, y: np.array, y_pred: np.array, x_cond: ConditionOrScalar,
+                       y_cond: ConditionOrScalar) -> Probability:
+    # used to compute the differences contained in the array returned by the
+    # function discrete_equalised_odds (see its documentation)
+    x_cond = __ensure_is_condition(x_cond)
+    x_is_x_value = x_cond(x)
+    y_cond = __ensure_is_condition(y_cond)
+    y_is_not_y_value = np.bitwise_not(y_cond(y))
+
+    cond1 = y_cond(y_pred[y_is_not_y_value & x_is_x_value]).sum() / (x_is_x_value & y_cond(y)).sum()
+    cond2 = y_cond(y_pred[y_is_not_y_value]).sum() / (y_cond(y)).sum()
+    return abs(cond1 - cond2)
+
+
+def discrete_equalised_odds(x: np.array, y: np.array, y_pred: np.array) -> np.array:
+    """Computes the equalised odds for a given classifier h (represented by its predictions h(X)).
+        A classifier satisfies equalised odds if its predictions are independent of the protected
+        attribute given the labels. The following must hold for all unique values of Y and all the unique values of X. 
+
+    More formally:
+        :math:`eo_ij = \|P[h(X) \mid X = x_j, Y = y_i] - P[h(X) \mid Y = y_i]\|`
+
+    Also see:
+        * https://www.ijcai.org/proceedings/2020/0315.pdf, sec. 3, definition 2
+
+    :param x: (formally :math:`X`) vector of protected attribute (where each component gets values from a **discrete
+        distribution**, whose admissible values are :math:`{x_1, x_2, ..., x_n}`
+
+    :param y: (formally :math:`Y`) vector of ground truth values
+
+    :param y_pred: (formally :math:`h(X)`) vector of predicted values
+
+    :return: a math:`m x n` array where :math:`m` is the number of unique values of Y and :math:`n` is the number 
+        of unique values of X. Each element of the array :math:`eo` contains the previously defined difference. """
+
+    x_values = np.unique(x)
+    y_values = np.unique(y)
+
+    differences = []
+
+    for y_value in y_values:
+        differences_x = []
+        for x_value in x_values:
+            differences_x.append(__compute_false_rates(x, y, y_pred, x_value,
+                                                       y_value))
+        differences.append(differences_x)
+
+    differences = np.array(differences)
+    return differences
+
+
 aequitas.logger.debug("Module %s correctly loaded", __name__)

test/__init__.py

@@ -11,6 +11,33 @@ def uniform_binary_dataset(rows: int = 1000, columns: int = 2) -> np.array:
     return random.uniform(0, 1, size=(rows, columns)).round().astype(int)
 
 
+def uniform_binary_dataset_gt(rows: int = 1000, columns: int = 2) -> np.array:
+    xs = uniform_binary_dataset(rows, 1)
+    labels = uniform_binary_dataset(rows, 1) 
+    noise = random.choice([0, 1], p=[0.8, 0.2], size=(rows, 1))
+    preds = abs(labels-noise)
+    if columns > 2:
+        data = []
+        for _ in range(columns - 2):
+            data.append(uniform_binary_dataset(rows, 1))
+        data = np.concatenate(data, axis=1)
+        return np.concatenate((data, xs, labels, preds), axis=1) 
+    else:
+        return np.concatenate((xs, labels, preds), axis=1)
+
+def skewed_binary_dataset_gt(rows: int = 1000, columns: int = 2, p: float = 0.8) -> np.array:
+    xs = uniform_binary_dataset(rows, 1)
+    preds = np.array([bernoulli(p)[0] * x for x in xs])
+    labels = uniform_binary_dataset(rows, 1)
+    if columns > 2:
+        data = []
+        for _ in range(columns - 2):
+            data.append(uniform_binary_dataset(rows, 1))
+        data = np.concatenate(data, axis=1)
+        return np.concatenate((data, xs, labels, preds), axis=1) 
+    else:
+        return np.concatenate((xs, labels, preds), axis=1)
+
 def bernoulli(p: float, size: typing.Tuple[int, int] = (1,)) -> np.array:
     assert 0 <= p <= 1, "p must be in [0, 1]"
     return (random.uniform(0, 1, size=size) < p).astype(int)

test/core/test_metrics.py

@@ -1,20 +1,24 @@
-from test import uniform_binary_dataset, skewed_binary_dataset
+from test import uniform_binary_dataset, skewed_binary_dataset, uniform_binary_dataset_gt, skewed_binary_dataset_gt
 from aequitas.core.metrics import discrete_demographic_parities
+from aequitas.core.metrics import discrete_equalised_odds
 import unittest
+import numpy as np
 
 
 DATASET_SIZE = 10000
 
 
-class TestDemographicParity(unittest.TestCase):
-    def setUp(self) -> None:
-        self.fair_dataset = uniform_binary_dataset(rows=DATASET_SIZE)
-        self.unfair_dataset = skewed_binary_dataset(rows=DATASET_SIZE, p=0.9)
-
+class AbstractMetricTestCase(unittest.TestCase):
     def assertInRange(self, value, lower, upper):
         self.assertGreaterEqual(value, lower)
         self.assertLessEqual(value, upper)
 
+
+class TestDemographicParity(AbstractMetricTestCase):
+    def setUp(self) -> None:
+        self.fair_dataset = uniform_binary_dataset(rows=DATASET_SIZE)
+        self.unfair_dataset = skewed_binary_dataset(rows=DATASET_SIZE, p=0.9)
+
     def test_parity_on_fair_binary_case(self):
         x = self.fair_dataset[:, 0]
         y = self.fair_dataset[:, 1]
@@ -30,5 +34,39 @@ def test_parity_on_unfair_binary_case(self):
         self.assertInRange(parities[0], 0.4, 0.5)
 
 
+class TestEqualisedOdds(AbstractMetricTestCase):
+    def setUp(self) -> None:
+        self.fair_dataset = uniform_binary_dataset_gt(rows=DATASET_SIZE)
+        self.unfair_dataset = skewed_binary_dataset_gt(rows=DATASET_SIZE, p=0.9)
+
+    def test_equalised_odds_on_fair_binary_case(self):
+        x = self.fair_dataset[:, -3]
+        y = self.fair_dataset[:, -2]
+        y_pred = self.fair_dataset[:, -1]
+
+        y_values = np.unique(y)
+
+        differences = discrete_equalised_odds(x, y, y_pred)
+        for diff_row  in differences:
+            for diff in diff_row:            
+                self.assertInRange(diff, 0.0, 0.1)
+
+    def test_equalised_odds_on_unfair_binary_case(self):
+        x = self.unfair_dataset[:, -3]
+        y = self.unfair_dataset[:, -2]
+        y_pred = self.unfair_dataset[:, -1]
+
+        y_values = np.unique(y)
+
+        differences = discrete_equalised_odds(x, y, y_pred)
+        for diff_row  in differences:
+            for diff in diff_row:            
+                self.assertInRange(diff, 0.3, 1.0)
+
+
+# delete this abstract class, so that the included tests are not run
+del AbstractMetricTestCase
+
+
 if __name__ == '__main__':
     unittest.main()