Repository: alan-turing-institute/sktime
Pull request #3274 (2022-08-14 19:46:08)

[ENH] inversion of transformer wrapper and dunder
This PR adds a transformer wrapper `InvertTransform` which switches `transform` and `inverse_transform`. This can be potentially useful in graph pipeline building, or pipeline blueprints where transformers "cancel each other out".

Also adds:
* a dunder shortcut `~` = `__invert__`
* a test that checks that the transformer indeed inverts


Patch:
diff --git a/docs/source/api_reference/transformations.rst b/docs/source/api_reference/transformations.rst
index 178623662..e0e3174d7 100644
--- a/docs/source/api_reference/transformations.rst
+++ b/docs/source/api_reference/transformations.rst
@@ -51,17 +51,12 @@ Pipeline building
 
     TransformerPipeline
     FeatureUnion
+    ColumnwiseTransformer
     FitInTransform
     MultiplexTransformer
-
-.. currentmodule:: sktime.transformations.series.compose
-
-.. autosummary::
-    :toctree: auto_generated/
-    :template: class.rst
-
     OptionalPassthrough
-    ColumnwiseTransformer
+    InvertTransform
+    Id
     YtoX
 
 .. currentmodule:: sktime.transformations.panel.compose
diff --git a/sktime/transformations/base.py b/sktime/transformations/base.py
index 93f7ebd05..7f3ae85af 100644
--- a/sktime/transformations/base.py
+++ b/sktime/transformations/base.py
@@ -278,6 +278,17 @@ class BaseTransformer(BaseEstimator):
         else:
             return NotImplemented
 
+    def __invert__(self):
+        """Magic unary ~ (inversion) method, return InvertTransform of self.
+
+        Returns
+        -------
+        `InvertTransform` object, containing `self`.
+        """
+        from sktime.transformations.compose import InvertTransform
+
+        return InvertTransform(self)
+
     def __neg__(self):
         """Magic unary - (negation) method, return OptionalPassthrough of self.
 
@@ -288,7 +299,7 @@ class BaseTransformer(BaseEstimator):
         `OptionalPassthrough` object, containing `self`, with `passthrough=False`.
             The `passthrough` parameter can be set via `set_params`.
         """
-        from sktime.transformations.series.compose import OptionalPassthrough
+        from sktime.transformations.compose import OptionalPassthrough
 
         return OptionalPassthrough(self, passthrough=False)
 
diff --git a/sktime/transformations/compose.py b/sktime/transformations/compose.py
index 102c21842..ff561bb67 100644
--- a/sktime/transformations/compose.py
+++ b/sktime/transformations/compose.py
@@ -2,6 +2,8 @@
 """Meta-transformers for building composite transformers."""
 # copyright: sktime developers, BSD-3-Clause License (see LICENSE file)
 
+from warnings import warn
+
 import pandas as pd
 from sklearn import clone
 from sklearn.utils.metaestimators import if_delegate_has_method
@@ -23,6 +25,7 @@ __all__ = [
     "FeatureUnion",
     "FitInTransform",
     "Id",
+    "InvertTransform",
     "MultiplexTransformer",
     "OptionalPassthrough",
     "TransformerPipeline",
@@ -1074,18 +1077,156 @@ class MultiplexTransformer(_DelegatedTransformer, _HeterogenousMetaEstimator):
         )
 
 
-# mtypes for Series, Panel, Hierarchical,
-# with exception of some ambiguous and discouraged mtypes
-CORE_MTYPES = [
-    "pd.DataFrame",
-    "np.ndarray",
-    "pd.Series",
-    "pd-multiindex",
-    "df-list",
-    "nested_univ",
-    "numpy3D",
-    "pd_multiindex_hier",
-]
+class InvertTransform(_DelegatedTransformer):
+    """Invert a series-to-series transformation.
+
+    Switches `transform` and `inverse_transform`, leaves `fit` and `update` the same.
+
+    Parameters
+    ----------
+    transformer : sktime transformer, must transform Series input to Series output
+        this is a "blueprint" transformer, state does not change when `fit` is called
+
+    Attributes
+    ----------
+    transformer_: transformer,
+        this clone is fitted when `fit` is called and provides `transform` and inverse
+
+    Examples
+    --------
+    >>> from sktime.datasets import load_airline
+    >>> from sktime.transformations.compose import InvertTransform
+    >>> from sktime.transformations.series.exponent import ExponentTransformer
+    >>>
+    >>> inverse_exponent = InvertTransform(ExponentTransformer(power=3))
+    >>> X = load_airline()
+    >>> Xt = inverse_exponent.fit_transform(X)  # computes 3rd square root
+    """
+
+    _tags = {
+        "scitype:transform-input": "Series",
+        # what is the scitype of X: Series, or Panel
+        "scitype:transform-output": "Series",
+        # what scitype is returned: Primitives, Series, Panel
+        "scitype:instancewise": True,  # is this an instance-wise transform?
+        "X_inner_mtype": ["pd.DataFrame", "pd.Series"],
+        # which mtypes do _fit/_predict support for X?
+        "y_inner_mtype": "None",  # which mtypes do _fit/_predict support for y?
+        "univariate-only": False,
+        "fit_is_empty": False,
+        "capability:inverse_transform": True,
+    }
+
+    def __init__(self, transformer):
+        self.transformer = transformer
+
+        super(InvertTransform, self).__init__()
+
+        self.transformer_ = transformer.clone()
+
+        # should be all tags, but not fit_is_empty
+        #   (_fit should not be skipped)
+        tags_to_clone = [
+            "scitype:transform-input",
+            "scitype:transform-output",
+            "scitype:instancewise",
+            "X_inner_mtype",
+            "y_inner_mtype",
+            "handles-missing-data",
+            "X-y-must-have-same-index",
+            "transform-returns-same-time-index",
+            "skip-inverse-transform",
+        ]
+        self.clone_tags(transformer, tag_names=tags_to_clone)
+
+        if not transformer.get_tag("capability:inverse_transform", False):
+            warn(
+                "transformer does not have capability to inverse transform, "
+                "according to capability:inverse_transform tag. "
+                "If the tag was correctly set, this transformer will likely crash"
+            )
+        inner_output = transformer.get_tag("scitype:transform-output")
+        if not transformer.get_tag("scitype:transform-output") != "Series":
+            warn(
+                f"transformer output is not Series but {inner_output}, "
+                "according to scitype:transform-output tag. "
+                "The InvertTransform wrapper supports only Series output, therefore"
+                " this transformer will likely crash on input."
+            )
+
+    _delegate_name = "transformer_"
+
+    def _transform(self, X, y=None):
+        """Transform X and return a transformed version.
+
+        private _transform containing the core logic, called from transform
+
+        Returns a transformed version of X by iterating over specified
+        columns and applying the wrapped transformer to them.
+
+        Parameters
+        ----------
+        X : sktime compatible time series container
+            Data to be transformed
+        y : Series or Panel, default=None
+            Additional data, e.g., labels for transformation
+
+        Returns
+        -------
+        Xt : sktime compatible time series container
+            transformed version of X
+        """
+        return self.transformer_.inverse_transform(X=X, y=y)
+
+    def _inverse_transform(self, X, y=None):
+        """Logic used by `inverse_transform` to reverse transformation on `X`.
+
+        Returns an inverse-transformed version of X by iterating over specified
+        columns and applying the univariate series transformer to them.
+
+        Only works if `self.transformer` has an `inverse_transform` method.
+
+        Parameters
+        ----------
+        X : sktime compatible time series container
+            Data to be inverse transformed
+        y : Series or Panel, default=None
+            Additional data, e.g., labels for transformation
+
+        Returns
+        -------
+        Xt : sktime compatible time series container
+            inverse transformed version of X
+        """
+        return self.transformer_.transform(X=X, y=y)
+
+    @classmethod
+    def get_test_params(cls, parameter_set="default"):
+        """Return testing parameter settings for the estimator.
+
+        Parameters
+        ----------
+        parameter_set : str, default="default"
+            Name of the set of test parameters to return, for use in tests. If no
+            special parameters are defined for a value, will return `"default"` set.
+
+        Returns
+        -------
+        params : dict or list of dict, default = {}
+            Parameters to create testing instances of the class
+            Each dict are parameters to construct an "interesting" test instance, i.e.,
+            `MyClass(**params)` or `MyClass(**params[i])` creates a valid test instance.
+            `create_test_instance` uses the first (or only) dictionary in `params`
+        """
+        from sktime.transformations.series.boxcox import BoxCoxTransformer
+        from sktime.transformations.series.exponent import ExponentTransformer
+
+        # ExponentTransformer skips fit
+        params1 = {"transformer": ExponentTransformer()}
+        # BoxCoxTransformer has fit
+        params2 = {"transformer": BoxCoxTransformer()}
+
+        return [params1, params2]
 
 
 class Id(_DelegatedTransformer):


Test patch:
diff --git a/sktime/transformations/tests/test_compose.py b/sktime/transformations/tests/test_compose.py
index 7acd7e8c0..1d578ea71 100644
--- a/sktime/transformations/tests/test_compose.py
+++ b/sktime/transformations/tests/test_compose.py
@@ -10,7 +10,11 @@ from sklearn.preprocessing import StandardScaler
 
 from sktime.datasets import load_airline
 from sktime.datatypes import get_examples
-from sktime.transformations.compose import FeatureUnion, TransformerPipeline
+from sktime.transformations.compose import (
+    FeatureUnion,
+    InvertTransform,
+    TransformerPipeline,
+)
 from sktime.transformations.panel.padder import PaddingTransformer
 from sktime.transformations.series.boxcox import LogTransformer
 from sktime.transformations.series.compose import OptionalPassthrough
@@ -227,6 +231,22 @@ def test_subset_getitem():
     _assert_array_almost_equal(t_none.fit_transform(X), X_theta)
 
 
+def test_dunder_invert():
+    """Test the invert dunder method, for wrapping in OptionalPassthrough."""
+    X = pd.DataFrame({"a": [1, 2], "b": [3, 4]})
+
+    t = ExponentTransformer(power=3)
+
+    t_inv = ~t
+
+    assert isinstance(t_inv, InvertTransform)
+    assert isinstance(t_inv.get_params()["transformer"], ExponentTransformer)
+
+    _assert_array_almost_equal(
+        t_inv.fit_transform(X), ExponentTransformer(1 / 3).fit_transform(X)
+    )
+
+
 def test_dunder_neg():
     """Test the neg dunder method, for wrapping in OptionalPassthrough."""
     X = pd.DataFrame({"a": [1, 2], "b": [3, 4]})