Skip to content

Add inverse_transform to tslearn.preprocessing.TimeSeriesScalerMinMax #697

Description

@asemv2

I am working on a spatio-temporal prediction project, and my data have the shape (n_ts, sz, d), which matches the expected input format of your scaler.

It is necessary for me to reverse the normalization after prediction so that the model outputs can be converted back to their original physical values.

I discussed this with ChatGPT and asked it to add an inverse_transform() method to the TimeSeriesScalerMinMax class. It generated an implementation, and when I tested it on sample data, it successfully recovered the original values after scaling.

Here is the modified code:

from tslearn.preprocessing import TimeSeriesScalerMinMax
import numpy as np


import numpy
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.utils.validation import check_is_fitted

# tslearn imports
from tslearn.utils import (
    check_array,
    to_time_series_dataset,
    check_dims
)
from tslearn.bases import TimeSeriesMixin


class TimeSeriesScalerMinMax(TimeSeriesMixin, TransformerMixin, BaseEstimator):
    """
    Min-Max scaler for time series datasets with inverse_transform support.
    """

    def __init__(
        self,
        value_range=(0., 1.),
        per_timeseries=True,
        per_feature=True
    ):
        self.value_range = value_range
        self.per_timeseries = per_timeseries
        self.per_feature = per_feature


    def fit(self, X, y=None, **kwargs):
        """
        Check input dimensions.
        """

        X_ = check_array(
            X,
            allow_nd=True,
            force_all_finite=False
        )

        X_ = to_time_series_dataset(X_)

        self._X_fit_dims = X_.shape
        self.n_features_in_ = self._X_fit_dims[-1]

        return self


    def fit_transform(self, X, y=None, **kwargs):
        return self.fit(X).transform(X)


    def transform(self, X, y=None, **kwargs):
        """
        Apply min-max scaling and store parameters
        for inverse transformation.
        """

        if self.value_range[0] >= self.value_range[1]:
            raise ValueError(
                "Minimum of desired range must be smaller than maximum."
            )


        check_is_fitted(
            self,
            "_X_fit_dims"
        )


        X_ = check_array(
            X,
            allow_nd=True,
            force_all_finite=False
        )

        X_ = to_time_series_dataset(X_)


        X_ = check_dims(
            X_,
            X_fit_dims=self._X_fit_dims,
            extend=False
        )


        # Define reduction axes
        axis = (1,)

        if not self.per_feature:
            axis += (2,)

        if not self.per_timeseries:
            axis += (0,)


        # Save min and max values
        self.min_t_ = numpy.nanmin(
            X_,
            axis=axis,
            keepdims=True
        )

        self.max_t_ = numpy.nanmax(
            X_,
            axis=axis,
            keepdims=True
        )


        range_t = self.max_t_ - self.min_t_

        # avoid division by zero
        range_t[range_t == 0.] = 1.


        # Min-max normalization
        X_scaled = (
            (X_ - self.min_t_)
            *
            (self.value_range[1] - self.value_range[0])
            /
            range_t
            +
            self.value_range[0]
        )


        return X_scaled



    def inverse_transform(self, X, y=None, **kwargs):
        """
        Convert scaled data back to original scale.
        """


        check_is_fitted(
            self,
            [
                "min_t_",
                "max_t_"
            ]
        )


        X_ = check_array(
            X,
            allow_nd=True,
            force_all_finite=False
        )

        X_ = to_time_series_dataset(X_)


        range_t = self.max_t_ - self.min_t_

        range_t[range_t == 0.] = 1.


        # inverse min-max transformation
        X_original = (
            (X_ - self.value_range[0])
            *
            range_t
            /
            (self.value_range[1] - self.value_range[0])
            +
            self.min_t_
        )


        return X_original



    def _more_tags(self):
        tags = super()._more_tags()
        tags.update(
            {
                "allow_nan": True
            }
        )
        return tags


    def __sklearn_tags__(self):
        tags = super().__sklearn_tags__()

        tags.input_tags.allow_nan = True

        return tags
    
    


X = np.random.random((1,5,2))


scaler = TimeSeriesScalerMinMax()


X_scaled = scaler.fit_transform(X)

# print("Scaled:")
# print(X_scaled)


X_original = scaler.inverse_transform(X_scaled)

print("Recovered:")
print(X_original)
print(X)
print(np.equal(X_original,X))

Activity

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions