Add TFTExplainer

CHANGELOG.md

@@ -7,6 +7,8 @@ but cannot always guarantee backwards compatibility. Changes that may **break co
 ## [Unreleased](https://github.com/unit8co/darts/tree/master)
 [Full Changelog](https://github.com/unit8co/darts/compare/0.23.1...master)
 
+- Added new `TFTExplainer` class to implement the Explainable AI part described in [the paper](https://arxiv.org/abs/1912.09363) of the `TFT` model. [#1392](https://github.com/unit8co/darts/pull/1392) by [Sebastian Cattes](https://github.com/cattes).
+
 ## [0.23.1](https://github.com/unit8co/darts/tree/0.23.1) (2023-01-12)
 Patch release
 
@@ -78,7 +80,6 @@ Patch release
   by [Antoine Madrona](https://github.com/madtoinou).
 
 
-
 **Fixed**
 - Fixed edge case in ShapExplainer for regression models where covariates series > target series
   [#1310](https://https://github.com/unit8co/darts/pull/1310) by [Rijk van der Meulen](https://github.com/rijkvandermeulen)

darts/explainability/__init__.py

@@ -5,3 +5,4 @@
 
 from darts.explainability.explainability_result import ExplainabilityResult
 from darts.explainability.shap_explainer import ShapExplainer
+from darts.explainability.tft_explainer import TFTExplainer

darts/explainability/explainability_result.py

@@ -11,7 +11,7 @@
 from numpy import integer
 
 from darts import TimeSeries
-from darts.logging import get_logger, raise_if, raise_if_not
+from darts.logging import get_logger, raise_if, raise_if_not, raise_log
 
 logger = get_logger(__name__)
 
@@ -27,21 +27,50 @@ def __init__(
         explained_forecasts: Union[
             Dict[integer, Dict[str, TimeSeries]],
             Sequence[Dict[integer, Dict[str, TimeSeries]]],
+            Dict[str, TimeSeries],
         ],
     ):
 
         self.explained_forecasts = explained_forecasts
         if isinstance(self.explained_forecasts, list):
+            raise_if_not(
+                isinstance(self.explained_forecasts[0], dict),
+                "The explained_forecasts Sequence must consist of dicts.",
+                logger,
+            )
+            raise_if_not(
+                all(isinstance(key, int) for key in self.explained_forecasts[0].keys()),
+                "The explained_forecasts dict Sequence must have all integer keys.",
+                logger,
+            )
             self.available_horizons = list(self.explained_forecasts[0].keys())
             h_0 = self.available_horizons[0]
             self.available_components = list(self.explained_forecasts[0][h_0].keys())
+        elif isinstance(self.explained_forecasts, dict):
+            if all(isinstance(key, int) for key in self.explained_forecasts.keys()):
+                self.available_horizons = list(self.explained_forecasts.keys())
+                h_0 = self.available_horizons[0]
+                self.available_components = list(self.explained_forecasts[h_0].keys())
+            elif all(isinstance(key, str) for key in self.explained_forecasts.keys()):
+                self.available_horizons = []
+                self.available_components = list(self.explained_forecasts.keys())
+            else:
+                raise_log(
+                    ValueError(
+                        "The explained_forecasts dictionary must have all integer or all string keys."
+                    ),
+                    logger,
+                )
         else:
-            self.available_horizons = list(self.explained_forecasts.keys())
-            h_0 = self.available_horizons[0]
-            self.available_components = list(self.explained_forecasts[h_0].keys())
+            raise_log(
+                ValueError(
+                    "The explained_forecasts must be a dictionary or a list of dictionaries."
+                ),
+                logger,
+            )
 
     def get_explanation(
-        self, horizon: int, component: Optional[str] = None
+        self, horizon: Optional[int] = None, component: Optional[str] = None
     ) -> Union[TimeSeries, Sequence[TimeSeries]]:
         """
         Returns one or several `TimeSeries` representing the explanations
@@ -56,35 +85,50 @@ def get_explanation(
             need to be specified for univariate series.
         """
 
+        # validate component argument
         raise_if(
             component is None and len(self.available_components) > 1,
-            ValueError(
-                "The component parameter is required when the model has more than one component."
-            ),
+            "The component parameter is required when the model has more than one component.",
             logger,
         )
 
         if component is None:
             component = self.available_components[0]
 
-        raise_if_not(
-            horizon in self.available_horizons,
-            "Horizon {} is not available. Available horizons are: {}".format(
-                horizon, self.available_horizons
-            ),
-        )
-
         raise_if_not(
             component in self.available_components,
             "Component {} is not available. Available components are: {}".format(
                 component, self.available_components
             ),
         )
 
+        # validate horizon argument
+        if horizon is not None:
+            raise_if(
+                len(self.available_horizons) == 0,
+                "The horizon parameter can not be used for a model where all time horizons are saved in the component.",
+            )
+
+            raise_if_not(
+                horizon in self.available_horizons,
+                "Horizon {} is not available. Available horizons are: {}".format(
+                    horizon, self.available_horizons
+                ),
+            )
+
         if isinstance(self.explained_forecasts, list):
             return [
                 self.explained_forecasts[i][horizon][component]
                 for i in range(len(self.explained_forecasts))
             ]
-        else:
+        elif all(isinstance(key, int) for key in self.explained_forecasts.keys()):
             return self.explained_forecasts[horizon][component]
+        elif all(isinstance(key, str) for key in self.explained_forecasts.keys()):
+            return self.explained_forecasts[component]
+        else:
+            raise_log(
+                ValueError(
+                    "Something went wrong. ExplainabilityResult got instantiated with an unexpected type."
+                ),
+                logger,
+            )