Comte

ClassificationModels/models/ECG5000/classification_report.csv

@@ -1,5 +1,5 @@
 0,1,2,3,4,accuracy,macro avg,weighted avg
-0.9821088694328131,0.9282567652611705,0.6521739130434783,0.32051282051282054,0.5,0.9248888888888889,0.6766104736500564,0.928689990417054
-0.9821088694328131,0.9276729559748428,0.3488372093023256,0.42857142857142855,0.09090909090909091,0.9248888888888889,0.5556199108381001,0.9248888888888889
-0.9821088694328131,0.9279647687952186,0.4545454545454546,0.36674816625916873,0.15384615384615385,0.9248888888888889,0.5770426825757616,0.9249156524345059
-2627.0,1590.0,86.0,175.0,22.0,0.9248888888888889,4500.0,4500.0
+0.967861100849649,0.9443671766342142,0.6530612244897959,0.2962962962962963,0.5555555555555556,0.9117777777777778,0.6834282707651023,0.9254116138134725
+0.9973353635325466,0.8540880503144654,0.37209302325581395,0.5028571428571429,0.22727272727272727,0.9117777777777778,0.5907292614465391,0.9117777777777778
+0.9823772028496438,0.8969616908850727,0.47407407407407404,0.37288135593220334,0.3225806451612903,0.9117777777777778,0.6097749937804569,0.9155545293878521
+2627.0,1590.0,86.0,175.0,22.0,0.9117777777777778,4500.0,4500.0

README.md

@@ -29,11 +29,6 @@ pip install https://github.com/fzi-forschungszentrum-informatik/TSInterpret/arch
 ```
 
 
-Due to the <a href='https://github.com/scikit-learn/sklearn-pypi-package'>sklearn brownout</a> `pip install sklearn` is no longer available in third party dependencies. As the current release of <a href='https://github.com/gkhayes/mlrose'>mlrose</a> still relies on sklearn, we eliminated the dependency. If you still want to use COMTE (with dependecy to mlrose), it can be installed via : 
-```shell
-pip install https://github.com/gkhayes/mlrose/archive/refs/heads/master.zip
-```
-
 
 ## 🍫 Quickstart
 The following example creates a simple Neural Network based on tensorflow and interprets the Classfier with Integrated Gradients and Temporal Saliency Rescaling [1].

TSInterpret/InterpretabilityModels/counterfactual/CF.py

@@ -178,7 +178,7 @@ def plot_in_one(
         """
         if self.mode == "time":
             item = item.reshape(item.shape[-1], item.shape[-2])
-            exp = exp.reshape(item.shape[-1], item.shape[-2])
+            exp = exp.reshape(exp.shape[-1], exp.shape[-2])
         else:
             item = item.reshape(item.shape[-2], item.shape[-1])
             exp = exp.reshape(item.shape[-2], item.shape[-1])

TSInterpret/InterpretabilityModels/counterfactual/Ates.py → TSInterpret/InterpretabilityModels/counterfactual/COMTE/Optimization.py

@@ -1,23 +1,18 @@
 import logging
 import multiprocessing
 import numbers
-import sys
 from typing import Tuple
 import numpy as np
 import pandas as pd
-
-# Workaround for mlrose package
-import six
 from sklearn.neighbors import KDTree
 from skopt import gbrt_minimize, gp_minimize
-
-sys.modules["sklearn.externals.six"] = six
-import mlrose
-
-from TSInterpret.InterpretabilityModels.counterfactual.CF import CF
-from TSInterpret.Models.PyTorchModel import PyTorchModel
-from TSInterpret.Models.SklearnModel import SklearnModel
-from TSInterpret.Models.TensorflowModel import TensorFlowModel
+from TSInterpret.InterpretabilityModels.counterfactual.COMTE.Problem import (
+    LossDiscreteState,
+    Problem,
+)
+from TSInterpret.InterpretabilityModels.counterfactual.COMTE.Optmization_helpers import (
+    random_hill_climb,
+)
 
 
 class BaseExplanation:
@@ -91,9 +86,6 @@ def _get_distractors(self, x_test, to_maximize, n_distractors=2):
         if isinstance(to_maximize, numbers.Integral):
             to_maximize = np.unique(self.labels)[to_maximize]
         distractors = []
-        # print('to_maximize',to_maximize)
-        # print('Class Tree',self.per_class_trees)
-        # print('Class Tree with id',self.per_class_trees[to_maximize])
         for idx in (
             self.per_class_trees[to_maximize]
             .query(x_test.T.flatten().reshape(1, -1), k=n_distractors)[1]
@@ -332,58 +324,6 @@ def explain(self, x_test, to_maximize=None, num_features=10):
         return other, target
 
 
-class LossDiscreteState:
-    def __init__(
-        self,
-        label_idx,
-        clf,
-        x_test,
-        distractor,
-        cols_swap,
-        reg,
-        max_features=3,
-        maximize=True,
-    ):
-        self.target = label_idx
-        self.clf = clf
-        self.x_test = x_test
-        self.reg = reg
-        self.distractor = distractor
-        self.cols_swap = cols_swap  # Column names that we can swap
-        self.prob_type = "discrete"
-        self.max_features = 3 if max_features is None else max_features
-        self.maximize = maximize
-        self.window_size = x_test.shape[-1]
-        self.channels = x_test.shape[-2]
-
-    def __call__(self, feature_matrix):
-        return self.evaluate(feature_matrix)
-
-    def evaluate(self, feature_matrix):
-        new_case = self.x_test.copy()
-        assert len(self.cols_swap) == len(feature_matrix)
-
-        for col_replace, a in zip(self.cols_swap, feature_matrix):
-            if a == 1:
-                new_case[0][col_replace] = self.distractor[0][col_replace]
-
-        replaced_feature_count = np.sum(feature_matrix)
-
-        input_ = new_case.reshape(1, self.channels, self.window_size)
-        result = self.clf(input_)[0][self.target]
-        feature_loss = self.reg * np.maximum(
-            0, replaced_feature_count - self.max_features
-        )
-        loss_pred = np.square(np.maximum(0, 0.95 - result))
-
-        loss_pred = loss_pred + feature_loss
-        # print(loss_pred)
-        return -loss_pred if self.maximize else loss_pred
-
-    def get_prob_type(self):
-        return self.prob_type
-
-
 class OptimizedSearch(BaseExplanation):
     def __init__(
         self,
@@ -414,10 +354,8 @@ def opt_Discrete(self, to_maximize, x_test, dist, columns, init, num_features=No
             max_features=num_features,
             maximize=False,
         )
-        problem = mlrose.DiscreteOpt(
-            length=len(columns), fitness_fn=fitness_fn, maximize=False, max_val=2
-        )
-        best_state, best_fitness = mlrose.random_hill_climb(
+        problem = Problem(length=len(columns), loss=fitness_fn, max_val=2)
+        best_state, best_fitness = random_hill_climb(
             problem,
             max_attempts=self.max_attemps,
             max_iters=self.maxiter,
@@ -469,9 +407,6 @@ def explain(
         if to_maximize is None:
             to_maximize = np.argsort(orig_preds)[0][-2:-1][0]
 
-        # print('Current may',np.argmax(orig_preds))
-        # print(to_maximize)
-
         if orig_label == to_maximize:
             print("Original and Target Label are identical !")
             return None, None
@@ -494,6 +429,8 @@ def _get_explanation(self, x_test, to_maximize, num_features):
         distractors = self._get_distractors(
             x_test, to_maximize, n_distractors=self.num_distractors
         )
+        # print('distracotr shape',np.array(distractors).shape)
+        # print('distracotr classification',np.argmax(self.clf(np.array(distractors).reshape(2,6,100)), axis=1))
 
         # Avoid constructing KDtrees twice
         self.backup.per_class_trees = self.per_class_trees
@@ -537,7 +474,6 @@ def _get_explanation(self, x_test, to_maximize, num_features):
 
             if not self.silent:
                 logging.info("Current probas: %s", probas)
-
             if np.argmax(probas) == to_maximize:
                 current_best = np.max(probas)
                 if current_best > best_explanation_score:
@@ -549,100 +485,3 @@ def _get_explanation(self, x_test, to_maximize, num_features):
             return None, None
 
         return best_modified, best_explanation
-
-
-class AtesCF(CF):
-    """Calculates and Visualizes Counterfactuals for Multivariate Time Series in accordance to the paper [1].
-
-    References
-    ----------
-     [1] Ates, Emre, et al.
-     "Counterfactual Explanations for Multivariate Time Series."
-     2021 International Conference on Applied Artificial Intelligence (ICAPAI). IEEE, 2021.
-    ----------
-    """
-
-    def __init__(
-        self,
-        model,
-        data,
-        backend,
-        mode,
-        method="opt",
-        number_distractors=2,
-        max_attempts=1000,
-        max_iter=1000,
-        silent=False,
-    ) -> None:
-        """
-        Arguments:
-            model [torch.nn.Module, Callable, tf.keras.model]: Model to be interpreted.
-            ref Tuple: Reference Dataset as Tuple (x,y).
-            backend str: desired Model Backend ('PYT', 'TF', 'SK').
-            mode str: Name of second dimension: `time` -> `(-1, time, feature)` or `feat` -> `(-1, feature, time)`
-            method str : 'opt' if optimized calculation, 'brute' for Brute Force
-            number_distractors int: number of distractore to be used
-            silent bool: logging.
-
-        """
-        super().__init__(model, mode)
-        self.backend = backend
-        test_x, test_y = data
-        shape = test_x.shape
-        if mode == "time":
-            # Parse test data into (1, feat, time):
-            change = True
-            self.ts_length = shape[-2]
-            test_x = test_x.reshape(test_x.shape[0], test_x.shape[2], test_x.shape[1])
-        elif mode == "feat":
-            change = False
-            self.ts_length = shape[-1]
-
-        if backend == "PYT":
-            self.predict = PyTorchModel(model, change).predict
-        elif backend == "TF":
-            self.predict = TensorFlowModel(model, change).predict
-
-        elif backend == "SK":
-            self.predict = SklearnModel(model, change).predict
-        self.referenceset = (test_x, test_y)
-        self.method = method
-        self.silent = silent
-        self.number_distractors = number_distractors
-        self.max_attemps = max_attempts
-        self.max_iter = max_iter
-
-    def explain(
-        self, x: np.ndarray, orig_class: int = None, target: int = None
-    ) -> Tuple[np.ndarray, int]:
-        """
-        Calculates the Counterfactual according to Ates.
-        Arguments:
-            x (np.array): The instance to explain. Shape : `mode = time` -> `(1,time, feat)` or `mode = time` -> `(1,feat, time)`
-            target int: target class. If no target class is given, the class with the secon heighest classification probability is selected.
-
-        Returns:
-            ([np.array], int): Tuple of Counterfactual and Label. Shape of CF : `mode = time` -> `(time, feat)` or `mode = time` -> `(feat, time)`
-
-        """
-
-        if self.mode != "feat":
-            x = x.reshape(-1, x.shape[-1], x.shape[-2])
-        train_x, train_y = self.referenceset
-        if len(train_y.shape) > 1:
-            train_y = np.argmax(train_y, axis=1)
-        if self.method == "opt":
-            opt = OptimizedSearch(
-                self.predict,
-                train_x,
-                train_y,
-                silent=self.silent,
-                threads=1,
-                num_distractors=self.number_distractors,
-                max_attempts=self.max_attemps,
-                maxiter=self.max_iter,
-            )
-            return opt.explain(x, to_maximize=target)
-        elif self.method == "brute":
-            opt = BruteForceSearch(self.predict, train_x, train_y, threads=1)
-            return opt.explain(x, to_maximize=target)

TSInterpret/InterpretabilityModels/counterfactual/COMTE/Optimization_helpers.py

@@ -0,0 +1,63 @@
+import numpy as np
+
+
+def random_hill_climb(
+    problem,
+    max_attempts=10,
+    max_iters=np.inf,
+    restarts=0,
+    init_state=None,
+    curve=False,
+    random_state=None,
+):
+    # Set random seed
+    if isinstance(random_state, int) and random_state > 0:
+        np.random.seed(random_state)
+
+    best_fitness = np.inf
+    best_state = None
+
+    if curve:
+        fitness_values = []
+
+    for _ in range(restarts + 1):
+        # Initialize optimization problem and attempts counter
+        if init_state is None:
+            problem.reset()
+        else:
+            problem.set_state(init_state)
+
+        attempts = 0
+        iters = 0
+
+        while (attempts < max_attempts) and (iters < max_iters):
+            iters += 1
+
+            # Find random neighbor and evaluate fitness
+            next_state = problem.random_neighbor()
+            next_fitness = problem.eval_fitness(next_state)
+
+            if next_fitness < problem.get_fitness():
+                problem.set_state(next_state)
+                attempts = 0
+
+            else:
+                attempts += 1
+
+            if curve:
+                fitness_values.append(problem.get_fitness())
+
+        # Update best state and best fitness
+        # print('best_fitness',best_fitness)
+        if problem.get_fitness() < best_fitness:
+            best_fitness = problem.get_fitness()
+            best_state = problem.get_state()
+            # print('bestfitness after', best_fitness)
+
+    if curve:
+        import matplotlib.pyplot as plt
+
+        plt.plot(np.asarray(fitness_values))
+        plt.show()
+
+    return best_state, best_fitness