Tensorflow/keras: Add support and tests for regression problems

ceml/backend/tensorflow/costfunctions/costfunctions.py

@@ -77,6 +77,23 @@ def score_impl(self, x):
         return lmad(x, self.x_orig, self.mad) 
 
 
+class SquaredError(CostFunctionDifferentiableTf):
+    """
+    Squared error cost function.
+    """
+    def __init__(self, input_to_output, y_target):
+        self.y_target = y_target
+        self.input_to_output = input_to_output
+
+        super(SquaredError, self).__init__()
+
+    def score_impl(self, x):
+        """
+        Computes the loss - squared error.
+        """
+        return l2(self.input_to_output(x), self.y_target)
+
+
 class NegLogLikelihoodCost(CostFunctionDifferentiableTf):
     """
     Negative-log-likelihood cost function.

ceml/tfkeras/counterfactual.py

@@ -79,7 +79,7 @@ def compute_counterfactual_ex(self, input_wrapper, x_orig, loss, loss_npy, loss_
         else:
             return x_cf, y_cf, delta
 
-    def compute_counterfactual(self, x, y_target, features_whitelist=None, regularization=None, C=1.0, optimizer="nelder-mead", optimizer_args=None, return_as_dict=True):
+    def compute_counterfactual(self, x, y_target, features_whitelist=None, regularization=None, C=1.0, optimizer="nelder-mead", optimizer_args=None, return_as_dict=True, done=None):
         """Computes a counterfactual of a given input `x`.
 
         Parameters
@@ -135,6 +135,16 @@ def compute_counterfactual(self, x, y_target, features_whitelist=None, regulariz
             If False, the results are returned as a triple.
 
             The default is True.
+        done : `callable`, optional
+            A callable that returns `True` if a counterfactual with a given output/prediction is accepted and `False` otherwise.
+
+            If `done` is None, the output/prediction of the counterfactual must match `y_target` exactly.
+
+            The default is None.
+
+            Note
+            ----
+            In case of a regression it might not always be possible to achieve a given output/prediction exactly.
 
         Returns
         -------
@@ -152,7 +162,7 @@ def compute_counterfactual(self, x, y_target, features_whitelist=None, regulariz
         input_wrapper, x_orig, _, grad_mask = self.wrap_input(features_whitelist, x, optimizer)
 
         # Check if the prediction of the given input is already consistent with y_target
-        done = y_target if callable(y_target) else lambda y: y == y_target
+        done = done = done if done is not None else y_target if callable(y_target) else lambda y: y == y_target
         self.warn_if_already_done(x, done)
 
         # Repeat for all C
@@ -180,7 +190,7 @@ def compute_counterfactual(self, x, y_target, features_whitelist=None, regulariz
         raise Exception("No counterfactual found - Consider changing parameters 'C', 'regularization', 'features_whitelist', 'optimizer' and try again")
 
 
-def generate_counterfactual(model, x, y_target, features_whitelist=None, regularization=None, C=1.0, optimizer="nelder-mead", optimizer_args=None, return_as_dict=True):
+def generate_counterfactual(model, x, y_target, features_whitelist=None, regularization=None, C=1.0, optimizer="nelder-mead", optimizer_args=None, return_as_dict=True, done=None):
     """Computes a counterfactual of a given input `x`.
 
     Parameters
@@ -238,6 +248,16 @@ def generate_counterfactual(model, x, y_target, features_whitelist=None, regular
         If False, the results are returned as a triple.
 
         The default is True.
+    done : `callable`, optional
+        A callable that returns `True` if a counterfactual with a given output/prediction is accepted and `False` otherwise.
+
+        If `done` is None, the output/prediction of the counterfactual must match `y_target` exactly.
+
+        The default is None.
+
+        Note
+        ----
+        In case of a regression it might not always be possible to achieve a given output/prediction exactly.
 
     Returns
     -------
@@ -248,4 +268,4 @@ def generate_counterfactual(model, x, y_target, features_whitelist=None, regular
     """
     cf = TfCounterfactual(model)
 
-    return cf.compute_counterfactual(x, y_target, features_whitelist, regularization, C, optimizer, optimizer_args, return_as_dict)
+    return cf.compute_counterfactual(x, y_target, features_whitelist, regularization, C, optimizer, optimizer_args, return_as_dict, done)

tests/tfkeras/test_tfkeras_linearregression.py

@@ -0,0 +1,82 @@
+# -*- coding: utf-8 -*-
+import sys
+sys.path.insert(0,'..')
+
+import tensorflow as tf
+tf.compat.v1.enable_eager_execution()
+tf.random.set_random_seed(42)
+
+import numpy as np
+from sklearn.datasets import load_boston
+from sklearn.model_selection import train_test_split
+from sklearn.metrics import r2_score
+
+from ceml.tfkeras import generate_counterfactual
+from ceml.backend.tensorflow.costfunctions import SquaredError
+from ceml.model import ModelWithLoss
+
+
+def test_linearregression():
+    # Neural network - Linear regression
+    class Model(ModelWithLoss):
+        def __init__(self, input_size):
+            super(Model, self).__init__()
+
+            self.model = tf.keras.models.Sequential([
+                tf.keras.layers.Dense(1, input_shape=(input_size,), kernel_regularizer=tf.keras.regularizers.l2(0.0001))
+            ])
+
+        def fit(self, x_train, y_train, num_epochs=800):
+            self.model.compile(optimizer='adam', loss='mse')
+
+            self.model.fit(X_train, y_train, epochs=num_epochs, verbose=False)
+
+        def predict(self, x):
+            return self.model(x)
+
+        def __call__(self, x):
+            return self.predict(x)
+
+        def get_loss(self, y_target, pred=None):
+            return SquaredError(self.model.predict, y_target)
+
+    # Load data
+    X, y = load_boston(True)
+
+    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.33, random_state=1)
+
+    # Create and fit model
+    model = Model(X.shape[1])
+    model.fit(X_train, y_train)
+
+    # Evaluation
+    y_pred = model.predict(X_test)
+    assert r2_score(y_test, y_pred) >= 0.6
+
+    # Select data point for explaining its prediction
+    x_orig = X_test[3,:]
+    y_orig_pred = model.predict(np.array([x_orig]))
+    assert y_orig_pred >= 16. and y_orig_pred < 22.
+
+    # Compute counterfactual
+    features_whitelist = None
+    y_target = 30.
+    y_target_done = lambda z: np.abs(z - y_target) < 6.
+
+    optimizer = tf.compat.v1.train.GradientDescentOptimizer(learning_rate=0.001)
+    optimizer_args = {"max_iter": 1000}
+    x_cf, y_cf, delta = generate_counterfactual(model, x_orig, y_target=y_target, features_whitelist=features_whitelist, regularization="l2", C=10., optimizer=optimizer, optimizer_args=optimizer_args, return_as_dict=False, done=y_target_done)
+    assert y_target_done(y_cf)
+    assert y_target_done(model.predict(np.array([x_cf])))
+
+    optimizer = "bfgs"
+    optimizer_args = {"max_iter": 1000}
+    x_cf, y_cf, delta = generate_counterfactual(model, x_orig, y_target=y_target, features_whitelist=features_whitelist, regularization="l2", C=10., optimizer=optimizer, optimizer_args=optimizer_args, return_as_dict=False, done=y_target_done)
+    assert y_target_done(y_cf)
+    assert y_target_done(model.predict(np.array([x_cf])))
+
+    optimizer = "nelder-mead"
+    optimizer_args = {"max_iter": 1000}
+    x_cf, y_cf, delta = generate_counterfactual(model, x_orig, y_target=y_target, features_whitelist=features_whitelist, regularization="l2", C=[0.1, 1.0, 10., 20.], optimizer=optimizer, optimizer_args=optimizer_args, return_as_dict=False, done=y_target_done)
+    assert y_target_done(y_cf)
+    assert y_target_done(model.predict(np.array([x_cf])))