diff --git a/CHANGELOG.md b/CHANGELOG.md index 8709c796d..47030003d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,7 @@ Changelogs for this project are recorded in this file since v0.2.0. ### Changed * Explicit exception when using Global Alignment Kernel with sigma close to zero ([#440](https://github.com/tslearn-team/tslearn/issues/440)) +* Move preprocessing classes into dedicated modules. * Fix shifting in K-shape shape extraction process ([#385](https://github.com/tslearn-team/tslearn/issues/385)) ### Removed @@ -24,6 +25,7 @@ Changelogs for this project are recorded in this file since v0.2.0. * `per_timeseries` and `per_feature` options for min-max and mean-variance scalers ([#536](https://github.com/tslearn-team/tslearn/issues/536)) * `TimeSeriesImputer`class: missing value imputer for time series ([#564](https://github.com/tslearn-team/tslearn/issues/564)) +* `mean`, `max` and `uniform` resampling methods to the `TimeSeriesResampler`class: missing value imputer for time series ([#537](https://github.com/tslearn-team/tslearn/issues/537)) * Frechet metrics and KNeighbors integration ([#402](https://github.com/tslearn-team/tslearn/issues/402) ### Changed diff --git a/tslearn/preprocessing/__init__.py b/tslearn/preprocessing/__init__.py index fd2407ddb..865ebdf8c 100644 --- a/tslearn/preprocessing/__init__.py +++ b/tslearn/preprocessing/__init__.py @@ -1,14 +1,12 @@ """ -The :mod:`tslearn.preprocessing` module gathers time series scalers and -resamplers. +The :mod:`tslearn.preprocessing` module gathers time series scalers, +resampler and imputer. """ -from .preprocessing import ( - TimeSeriesScalerMeanVariance, - TimeSeriesScalerMinMax, - TimeSeriesResampler, - TimeSeriesImputer -) +from .imputer import TimeSeriesImputer +from .resampler import TimeSeriesResampler +from .scaler import TimeSeriesScalerMeanVariance, TimeSeriesScalerMinMax + __all__ = [ "TimeSeriesResampler", diff --git a/tslearn/preprocessing/imputer.py b/tslearn/preprocessing/imputer.py new file mode 100644 index 000000000..5b95b3386 --- /dev/null +++ b/tslearn/preprocessing/imputer.py @@ -0,0 +1,244 @@ +"""Imputer for time series preprocessing""" +import typing + +import numpy + +from sklearn.base import TransformerMixin +from sklearn.utils.validation import check_is_fitted + +from tslearn.bases import TimeSeriesBaseEstimator +from tslearn.bases.bases import ALLOW_VARIABLE_LENGTH +from tslearn.utils import ( + check_variable_length_input, + to_time_series_dataset, + to_time_series, + ts_size, + check_dims +) + +__author__ = 'Romain Tavenard romain.tavenard[at]univ-rennes2.fr' + + +class TimeSeriesImputer(TransformerMixin, TimeSeriesBaseEstimator): + """Missing value imputer for time series. + + Missing values (nans) are replaced according to the chosen imputation + method. There might be cases where the computation of missing values is + impossible, in which case they are left unchanged + (ex: mean of all nans, ffill for the first value... ). + + The imputer can be configured so that trailing 'empty' samples (nans for + all features) are unprocessed by setting the `keep_trailing_nans` parameter + to `True`. This might be handy when dealing with variable length time + series datasets formatted with + :ref:`to_time_series_dataset `, + where time series are padded with 'empty' samples to match the length of the + longest time serie. This option aims at preserving the variable length + nature of the input dataset. + + Time series are processed sequentially by the :func:`~transform` and + :func:`~fit_transform` methods, and gathered using + :ref:`to_time_series_dataset `, + effectively padding if needed. + + Parameters + ---------- + method : {'mean', 'median', 'ffill', 'bfill', 'linear', 'constant', Callable}(default: 'mean') + The method used to compute missing values. + + When using linear imputation, starting nans will be replaced with first non-null value + and ending nans will be replaced with last non-null value ( + except for 'empty' samples when `keep_trailing_nans` set to `True`). + + When using a Callable, the function should take an array-like + representing a timeseries with missing values as input parameter and + should return the transformed timeseries. + value: float (default: nan) + The value to replace missing values with. Only used when method is + `constant`. + keep_trailing_nans: bool (default: False) + Whether trailing samples with nans on all dimensions should be considered + padding for variable length time series and kept unprocessed. When set to + `True`, trailing 'empty' samples will not be imputed. + + Notes + ----- + This method allows datasets of variable lenght time series. + While most missing values should be replaced, there might still be nan + values in the resulting dataset representing padding when used with + variable length time series, or uncomputable data. + + Examples + -------- + >>> TimeSeriesImputer().fit_transform([[0, numpy.nan, 6]]) + array([[[0.], + [3.], + [6.]]]) + >>> # Padding occurs after processing for variable length inputs + >>> TimeSeriesImputer().fit_transform([[numpy.nan, 3, 6], [numpy.nan, 3]]) + array([[[4.5], + [3. ], + [6. ]], + + [[3. ], + [3. ], + [nan]]]) + >>> # Trailing empty samples are preserved with `keep_trailing_nans` + >>> TimeSeriesImputer('ffill', keep_trailing_nans=True).fit_transform( + ... [[[1, 2], [2, numpy.nan]], [[3, 4], [numpy.nan, numpy.nan]]] + ... ) + array([[[ 1., 2.], + [ 2., 2.]], + + [[ 3., 4.], + [nan, nan]]]) + >>> # Uncomputable values are left unchanged + >>> TimeSeriesImputer('ffill').fit_transform([[numpy.nan, 3, 6]]) + array([[[nan], + [ 3.], + [ 6.]]]) + """ + def __init__(self, + method: typing.Union[str, typing.Callable] = "mean", + value: float = numpy.nan, + keep_trailing_nans: bool = False) -> None: + self.method = method + self.value = value + self.keep_trailing_nans = keep_trailing_nans + super().__init__() + + @property + def _imputer(self) -> typing.Union[typing.Callable, None]: + if callable(self.method): + return self.method + + if hasattr(self, "_{}_impute".format(self.method)): + return getattr(self, "_{}_impute".format(self.method)) + return None + + def _constant_impute(self, ts): + return numpy.where(numpy.isnan(ts), self.value, ts) + + @staticmethod + def _mean_impute(ts): + return numpy.where(numpy.isnan(ts), numpy.nanmean(ts, axis=0, keepdims=True), ts) + + @staticmethod + def _median_impute(ts): + return numpy.where(numpy.isnan(ts), numpy.nanmedian(ts, axis=0, keepdims=True), ts) + + @staticmethod + def _linear_impute(ts): + for di in range(ts.shape[-1]): + ts_di = ts[:, di] + mask = numpy.isnan(ts_di) + ts_di[mask] = numpy.interp( + numpy.nonzero(mask)[0], + numpy.nonzero(~mask)[0], + ts_di[~mask] + ) + return ts + + @staticmethod + def _ffill_impute(ts): + # Forward fill + mask = numpy.isnan(ts) + idx = numpy.where( + ~mask, + numpy.arange(ts.shape[0]).reshape(ts.shape[0], 1), + 0 + ) + numpy.maximum.accumulate(idx, axis=0, out=idx) + if ts.shape[-1] > 1: + # Multivariate + ts[mask] = ts[idx[mask], numpy.nonzero(mask)[1]] + else: + # Univariate + ts[mask] = ts[idx[mask]].flatten() + return ts + + @staticmethod + def _bfill_impute(ts): + # Backward fill + mask = numpy.isnan(ts) + idx = numpy.where( + ~mask, + numpy.arange(ts.shape[0]).reshape(ts.shape[0], 1), + ts.shape[0] -1 + ) + numpy.minimum.accumulate(numpy.flip(idx, axis=0), axis=0, out=idx) + idx = numpy.flip(idx, axis=0) + if ts.shape[-1] > 1: + # Multivariate + ts[mask] = ts[idx[mask], numpy.nonzero(mask)[1]] + else: + # Univariate + ts[mask] = ts[idx[mask]].flatten() + return ts + + def fit(self, X, y=None, **kwargs): + """A dummy method such that it complies to the sklearn requirements. + Since this method is completely stateless, it just returns itself. + + Parameters + ---------- + X + Ignored + + Returns + ------- + self + """ + X_ = check_variable_length_input(X) + self._X_fit_dims = X_.shape + return self + + def fit_transform(self, X, y=None, **kwargs): + """Fit to data, then transform it. + + Parameters + ---------- + X : array-like of shape (n_ts, sz, d) + Time series dataset to be imputed. + + Returns + ------- + numpy.ndarray + Imputed time series dataset. + """ + return self.fit(X).transform(X, kwargs) + + def transform(self, X, y=None, **kwargs): + """Fit to data, then transform it. + + Parameters + ---------- + X : array-like of shape (n_ts, sz, d) + Time series dataset to be imputed + + Returns + ------- + numpy.ndarray + Imputed time series dataset + """ + check_is_fitted(self, '_X_fit_dims') + + X_ = check_variable_length_input(X) + X_ = check_dims(X_, X_fit_dims=self._X_fit_dims, extend=False) + + imputer = self._imputer + if imputer is None: + raise ValueError("Imputer {} not implemented.".format(self.method)) + + for ts_index in range(X_.shape[0]): + ts = to_time_series(X[ts_index]) + stop_index = ts.shape[0] + if self.keep_trailing_nans: + stop_index = ts_size(ts) + X_[ts_index, :stop_index] = imputer(ts[:stop_index]) + return to_time_series_dataset(X_) + + def _more_tags(self) -> dict[str, typing.Any]: + more_tags = super()._more_tags() + more_tags.update({'allow_nan': True, ALLOW_VARIABLE_LENGTH: True}) + return more_tags diff --git a/tslearn/preprocessing/preprocessing.py b/tslearn/preprocessing/preprocessing.py deleted file mode 100644 index 7f0e31798..000000000 --- a/tslearn/preprocessing/preprocessing.py +++ /dev/null @@ -1,597 +0,0 @@ -from math import nan -from typing import Callable, Optional, Union - -import numpy - -from sklearn.base import TransformerMixin -from sklearn.utils.validation import check_is_fitted - -from tslearn.bases import TimeSeriesBaseEstimator -from tslearn.bases.bases import ALLOW_VARIABLE_LENGTH -from tslearn.utils import ( - check_variable_length_input, - to_time_series_dataset, - to_time_series, - check_equal_size, - ts_size, - check_array, - check_dims -) - -__author__ = 'Romain Tavenard romain.tavenard[at]univ-rennes2.fr' - - -class TimeSeriesResampler(TransformerMixin, TimeSeriesBaseEstimator): - """Resampler for time series. Resample time series so that they reach the - target size. - - Parameters - ---------- - sz : int (default: -1) - Size of the output time series. If not strictly positive, the size of - the longuest timeseries in the dataset is used. - - Examples - -------- - >>> TimeSeriesResampler(sz=5).fit_transform([[0, 3, 6]]) - array([[[0. ], - [1.5], - [3. ], - [4.5], - [6. ]]]) - """ - def __init__(self, sz: int=-1): - self.sz = sz - - def _get_resampling_size(self, X): - return self.sz if self.sz > 0 else X.shape[1] - - def fit(self, X, y=None, **kwargs): - """A dummy method such that it complies to the sklearn requirements. - Since this method is completely stateless, it just returns itself. - - Parameters - ---------- - X - Ignored - - Returns - ------- - self - """ - X_ = check_variable_length_input(X) - self._X_fit_dims = X_.shape - - return self - - def _transform_unit_sz(self, X): - n_ts, sz, d = X.shape - X_out = numpy.empty((n_ts, 1, d)) - for i in range(X.shape[0]): - X_out[i] = numpy.nanmean(X[i], axis=0, keepdims=True) - return X_out - - def fit_transform(self, X, y=None, **kwargs): - """Fit to data, then transform it. - - Parameters - ---------- - X : array-like of shape (n_ts, sz, d) - Time series dataset to be resampled. - - Returns - ------- - numpy.ndarray - Resampled time series dataset. - """ - return self.fit(X).transform(X) - - def transform(self, X, y=None, **kwargs): - """Fit to data, then transform it. - - Parameters - ---------- - X : array-like of shape (n_ts, sz, d) - Time series dataset to be resampled. - - Returns - ------- - numpy.ndarray - Resampled time series dataset. - """ - check_is_fitted(self, '_X_fit_dims') - - X_ = check_variable_length_input(X) - X_ = check_dims(X_, X_fit_dims=self._X_fit_dims, extend=False) - - target_sz = self._get_resampling_size(X_) - if target_sz == 1: - return self._transform_unit_sz(X_) - - n_ts, sz, d = X_.shape - equal_size = check_equal_size(X_) - X_out = numpy.empty((n_ts, target_sz, d)) - for i in range(X_.shape[0]): - if not equal_size: - sz = ts_size(X_[i]) - for di in range(d): - X_out[i, :, di] = numpy.interp( - numpy.linspace(0, 1, target_sz), - numpy.linspace(0, 1, sz), - X_[i, :sz, di] - ) - return X_out - - def _more_tags(self): - more_tags = super()._more_tags() - more_tags.update({'allow_nan': True, ALLOW_VARIABLE_LENGTH: True}) - return more_tags - - -class TimeSeriesScalerMinMax(TransformerMixin, TimeSeriesBaseEstimator): - """Scaler for time series datasets. Scales features values so that their span in given dimensions - is between ``min`` and ``max`` where ``value_range=(min, max)``. - - Parameters - ---------- - value_range : tuple (default: (0., 1.)) - The minimum and maximum value for the output time series. - per_timeseries: bool (default: True) - Wether the scaling should be performed per time series. - per_feature: bool (default: True) - Wether the scaling should be performed per feature. - Meaningless for univariate timeseries. - - Notes - ----- - This method requires a dataset of equal-sized time series. - - NaNs within a time series are ignored when calculating min and max. - - Examples - -------- - >>> TimeSeriesScalerMinMax(value_range=(1., 2.)).fit_transform([[0, 3, 6]]) - array([[[1. ], - [1.5], - [2. ]]]) - >>> TimeSeriesScalerMinMax(value_range=(1., 2.)).fit_transform( - ... [[numpy.nan, 3, 6]] - ... ) - array([[[nan], - [ 1.], - [ 2.]]]) - >>> TimeSeriesScalerMinMax(value_range=(1., 2.), per_timeseries=False, per_feature=False).fit_transform( - ... [[[1, 2], [2, 3]], - ... [[3, 4], [4, 5]]] - ... ) - array([[[1. , 1.25], - [1.25, 1.5 ]], - - [[1.5 , 1.75], - [1.75, 2. ]]]) - """ - 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): - """A dummy method such that it complies to the sklearn requirements. - Since this method is completely stateless, it just returns itself. - - Parameters - ---------- - X - Ignored - - Returns - ------- - self - """ - X_ = check_array(X, allow_nd=True, force_all_finite=False) - X_ = to_time_series_dataset(X_) - self._X_fit_dims = X_.shape - return self - - def fit_transform(self, X, y=None, **kwargs): - """Fit to data, then transform it. - - Parameters - ---------- - X : array-like of shape (n_ts, sz, d) - Time series dataset to be rescaled. - - Returns - ------- - numpy.ndarray - Resampled time series dataset. - """ - return self.fit(X).transform(X) - - def transform(self, X, y=None, **kwargs): - """Will normalize (min-max) each of the timeseries. IMPORTANT: this - transformation is completely stateless, and is applied to each of - the timeseries individually. - - Parameters - ---------- - X : array-like of shape (n_ts, sz, d) - Time series dataset to be rescaled. - - Returns - ------- - numpy.ndarray - Rescaled time series dataset. - """ - if self.value_range[0] >= self.value_range[1]: - raise ValueError("Minimum of desired range must be smaller" - " than maximum. Got %s." % str(self.value_range)) - - 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) - - axis = (1,) - if not self.per_feature: - axis += (2,) - if not self.per_timeseries: - axis += (0,) - - min_t = numpy.nanmin(X_, axis=axis, keepdims=True) - max_t = numpy.nanmax(X_, axis=axis, keepdims=True) - - range_t = max_t - min_t - range_t[range_t == 0.] = 1. - nomin = (X_ - min_t) * (self.value_range[1] - self.value_range[0]) - X_ = nomin / range_t + self.value_range[0] - return X_ - - def _more_tags(self): - more_tags = super()._more_tags() - more_tags.update({'allow_nan': True}) - return more_tags - - -class TimeSeriesScalerMeanVariance(TransformerMixin, TimeSeriesBaseEstimator): - """Scaler for time series datasets. Scales fetures values so that their mean (resp. - standard deviation) in given dimensions is mu (resp. std). - - Parameters - ---------- - mu : float (default: 0.) - Mean of the output time series. - std : float (default: 1.) - Standard deviation of the output time series. - per_timeseries: bool (default: True) - Whether the scaling should be performed per time series. - per_feature: bool (default: True) - Whether the scaling should be performed per feature. - Meaningless for univariate timeseries. - - Notes - ----- - This method requires a dataset of equal-sized time series. - - NaNs within a time series are ignored when calculating mu and std. - - Examples - -------- - >>> TimeSeriesScalerMeanVariance(mu=0., - ... std=1.).fit_transform([[0, 3, 6]]) - array([[[-1.22474487], - [ 0. ], - [ 1.22474487]]]) - >>> TimeSeriesScalerMeanVariance(mu=0., - ... std=1.).fit_transform([[numpy.nan, 3, 6]]) - array([[[nan], - [-1.], - [ 1.]]]) - >>> TimeSeriesScalerMeanVariance(per_timeseries=False, - ... per_feature=False - ... ).fit_transform([[[1, 2], [2, 3]], [[3, 4], [4, 5]]]) - array([[[-1.63299316, -0.81649658], - [-0.81649658, 0. ]], - - [[ 0. , 0.81649658], - [ 0.81649658, 1.63299316]]]) - """ - def __init__(self, mu=0., std=1., per_timeseries=True, per_feature=True): - self.mu = mu - self.std = std - self.per_timeseries = per_timeseries - self.per_feature = per_feature - - def fit(self, X, y=None, **kwargs): - """A dummy method such that it complies to the sklearn requirements. - Since this method is completely stateless, it just returns itself. - - Parameters - ---------- - X - Ignored - - Returns - ------- - self - """ - X_ = check_array(X, allow_nd=True, force_all_finite=False) - X_ = to_time_series_dataset(X_) - self._X_fit_dims = X_.shape - return self - - def fit_transform(self, X, y=None, **kwargs): - """Fit to data, then transform it. - - Parameters - ---------- - X : array-like of shape (n_ts, sz, d) - Time series dataset to be rescaled. - - Returns - ------- - numpy.ndarray - Resampled time series dataset. - """ - return self.fit(X).transform(X) - - def transform(self, X, y=None, **kwargs): - """Fit to data, then transform it. - - Parameters - ---------- - X : array-like of shape (n_ts, sz, d) - Time series dataset to be rescaled - - Returns - ------- - numpy.ndarray - Rescaled time series dataset - """ - 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) - - axis = (1,) - if not self.per_timeseries: - axis += (0,) - if not self.per_feature: - axis += (2,) - - mean_t = numpy.nanmean(X_, axis=axis, keepdims=True) - std_t = numpy.nanstd(X_, axis=axis, keepdims=True) - - std_t[std_t == 0.] = 1. - X_ = (X_ - mean_t) * self.std / std_t + self.mu - return X_ - - def _more_tags(self): - more_tags = super()._more_tags() - more_tags.update({'allow_nan': True}) - return more_tags - - -class TimeSeriesImputer(TransformerMixin, TimeSeriesBaseEstimator): - """Missing value imputer for time series. - - Missing values (nans) are replaced according to the chosen imputation - method. There might be cases where the computation of missing values is - impossible, in which case they are left unchanged - (ex: mean of all nans, ffill for the first value... ). - - The imputer can be configured so that trailing 'empty' samples (nans for - all features) are unprocessed by setting the `keep_trailing_nans` parameter - to `True`. This might be handy when dealing with variable length time - series datasets formatted with - :ref:`to_time_series_dataset `, - where time series are padded with 'empty' samples to match the length of the - longest time serie. This option aims at preserving the variable length - nature of the input dataset. - - Time series are processed sequentially by the :func:`~transform` and - :func:`~fit_transform` methods, and gathered using - :ref:`to_time_series_dataset `, - effectively padding if needed. - - Parameters - ---------- - method : {'mean', 'median', 'ffill', 'bfill', 'linear', 'constant', Callable}(default: 'mean') - The method used to compute missing values. - - When using linear imputation, starting nans will be replaced with first non-null value - and ending nans will be replaced with last non-null value ( - except for 'empty' samples when `keep_trailing_nans` set to `True`). - - When using a Callable, the function should take an array-like - representing a timeseries with missing values as input parameter and - should return the transformed timeseries. - value: float (default: nan) - The value to replace missing values with. Only used when method is - `constant`. - keep_trailing_nans: bool (default: False) - Whether trailing samples with nans on all dimensions should be considered - padding for variable length time series and kept unprocessed. When set to - `True`, trailing 'empty' samples will not be imputed. - - Notes - ----- - This method allows datasets of variable lenght time series. - While most missing values should be replaced, there might still be nan - values in the resulting dataset representing padding when used with - variable length time series, or uncomputable data. - - Examples - -------- - >>> TimeSeriesImputer().fit_transform([[0, numpy.nan, 6]]) - array([[[0.], - [3.], - [6.]]]) - >>> # Padding occurs after processing for variable length inputs - >>> TimeSeriesImputer().fit_transform([[numpy.nan, 3, 6], [numpy.nan, 3]]) - array([[[4.5], - [3. ], - [6. ]], - - [[3. ], - [3. ], - [nan]]]) - >>> # Trailing empty samples are preserved with `keep_trailing_nans` - >>> TimeSeriesImputer('ffill', keep_trailing_nans=True).fit_transform( - ... [[[1, 2], [2, numpy.nan]], [[3, 4], [numpy.nan, numpy.nan]]] - ... ) - array([[[ 1., 2.], - [ 2., 2.]], - - [[ 3., 4.], - [nan, nan]]]) - >>> # Uncomputable values are left unchanged - >>> TimeSeriesImputer('ffill').fit_transform([[numpy.nan, 3, 6]]) - array([[[nan], - [ 3.], - [ 6.]]]) - """ - def __init__(self, - method: Union[str, Callable]="mean", - value: Optional[float]=nan, - keep_trailing_nans: bool = False): - self.method = method - self.value = value - self.keep_trailing_nans = keep_trailing_nans - super().__init__() - - @property - def _imputer(self): - if callable(self.method): - return self.method - - if hasattr(self, "_{}_impute".format(self.method)): - return getattr(self, "_{}_impute".format(self.method)) - return None - - def _constant_impute(self, ts): - return numpy.where(numpy.isnan(ts), self.value, ts) - - @staticmethod - def _mean_impute(ts): - return numpy.where(numpy.isnan(ts), numpy.nanmean(ts, axis=0, keepdims=True), ts) - - @staticmethod - def _median_impute(ts): - return numpy.where(numpy.isnan(ts), numpy.nanmedian(ts, axis=0, keepdims=True), ts) - - @staticmethod - def _linear_impute(ts): - for di in range(ts.shape[-1]): - ts_di = ts[:, di] - mask = numpy.isnan(ts_di) - ts_di[mask] = numpy.interp( - numpy.nonzero(mask)[0], - numpy.nonzero(~mask)[0], - ts_di[~mask] - ) - return ts - - @staticmethod - def _ffill_impute(ts): - # Forward fill - mask = numpy.isnan(ts) - idx = numpy.where( - ~mask, - numpy.arange(ts.shape[0]).reshape(ts.shape[0], 1), - 0 - ) - numpy.maximum.accumulate(idx, axis=0, out=idx) - if ts.shape[-1] > 1: - # Multivariate - ts[mask] = ts[idx[mask], numpy.nonzero(mask)[1]] - else: - # Univariate - ts[mask] = ts[idx[mask]].flatten() - return ts - - @staticmethod - def _bfill_impute(ts): - # Backward fill - mask = numpy.isnan(ts) - idx = numpy.where( - ~mask, - numpy.arange(ts.shape[0]).reshape(ts.shape[0], 1), - ts.shape[0] -1 - ) - numpy.minimum.accumulate(numpy.flip(idx, axis=0), axis=0, out=idx) - idx = numpy.flip(idx, axis=0) - if ts.shape[-1] > 1: - # Multivariate - ts[mask] = ts[idx[mask], numpy.nonzero(mask)[1]] - else: - # Univariate - ts[mask] = ts[idx[mask]].flatten() - return ts - - def fit(self, X, y=None, **kwargs): - """A dummy method such that it complies to the sklearn requirements. - Since this method is completely stateless, it just returns itself. - - Parameters - ---------- - X - Ignored - - Returns - ------- - self - """ - X_ = check_variable_length_input(X) - self._X_fit_dims = X_.shape - return self - - def fit_transform(self, X, y=None, **kwargs): - """Fit to data, then transform it. - - Parameters - ---------- - X : array-like of shape (n_ts, sz, d) - Time series dataset to be imputed. - - Returns - ------- - numpy.ndarray - Imputed time series dataset. - """ - return self.fit(X).transform(X, kwargs) - - def transform(self, X, y=None, **kwargs): - """Fit to data, then transform it. - - Parameters - ---------- - X : array-like of shape (n_ts, sz, d) - Time series dataset to be imputed - - Returns - ------- - numpy.ndarray - Imputed time series dataset - """ - check_is_fitted(self, '_X_fit_dims') - - X_ = check_variable_length_input(X) - X_ = check_dims(X_, X_fit_dims=self._X_fit_dims, extend=False) - - imputer = self._imputer - if imputer is None: - raise ValueError("Imputer {} not implemented.".format(self.method)) - - for ts_index in range(X_.shape[0]): - ts = to_time_series(X[ts_index]) - stop_index = ts.shape[0] - if self.keep_trailing_nans: - stop_index = ts_size(ts) - X_[ts_index, :stop_index] = imputer(ts[:stop_index]) - return to_time_series_dataset(X_) - - def _more_tags(self): - more_tags = super()._more_tags() - more_tags.update({'allow_nan': True, ALLOW_VARIABLE_LENGTH: True}) - return more_tags diff --git a/tslearn/preprocessing/resampler.py b/tslearn/preprocessing/resampler.py new file mode 100644 index 000000000..5c07deff6 --- /dev/null +++ b/tslearn/preprocessing/resampler.py @@ -0,0 +1,218 @@ +"""Resampler for time series preprocessing""" +import math +import typing + +import numpy + +from sklearn.base import TransformerMixin +from sklearn.utils.validation import check_is_fitted + +from tslearn.bases import TimeSeriesBaseEstimator +from tslearn.bases.bases import ALLOW_VARIABLE_LENGTH +from tslearn.utils import ( + check_variable_length_input, + check_equal_size, + ts_size, + check_dims +) + +__author__ = 'Romain Tavenard romain.tavenard[at]univ-rennes2.fr' + + +class TimeSeriesResampler(TransformerMixin, TimeSeriesBaseEstimator): + """Resampler for time series. Resample time series so that they reach the + target size. + + Each time series of a dataset is processed independently. If the target size + is equal to the actual size of the sime series, the time series is left + unchanged. When computing the actual size of a time series, trailing empty + samples (nans on all dimensions) are not taken into account. + + Parameters + ---------- + sz : int (default: -1) + Size of the output time series. If not strictly positive, the size of + the longuest time series in the dataset is used. + method: {'linear', 'mean', 'max', 'uniform'} (default: linear) + The method used to resample. + - linear: linear interpolation for `sz` evenly spaced samples among + each time series. + - mean: mean computed for `sz` evenly spaced samples among each time + series within a `window_size` interval. + - max: mean computed for `sz` evenly spaced samples among each time + series within a `window_size` interval. + - uniform: select `sz` evenly spaced samples among each time series + + window_size: strictly positive int or None (default: None) + Custom window size. Used for `mean` and `max` resampling method. Ignored + otherwise. If set to `None`, the resampling factor is used. + + Examples + -------- + >>> # Linear upsampling + >>> TimeSeriesResampler(sz=5).fit_transform([[0, 3, 6]]) + array([[[0. ], + [1.5], + [3. ], + [4.5], + [6. ]]]) + >>> # Linear downsampling + >>> TimeSeriesResampler(sz=5).fit_transform([[0, 3, 6, 9, 12, 15]]) + array([[[ 0. ], + [ 3.75], + [ 7.5 ], + [11.25], + [15. ]]]) + >>> # Mean downsampling with custom window size + >>> TimeSeriesResampler(sz=5, method="mean", window_size=2).fit_transform([[0, 3, 6, 9, 12, 15]]) + array([[[ 1.5], + [ 4.5], + [ 7.5], + [10.5], + [13.5]]]) + """ + def __init__(self, + sz: int = -1, + method: str = 'linear', + window_size: typing.Optional[int] = None) -> None: + self.sz = sz + self.method = method + self.window_size = window_size + + @property + def _resampler(self) -> typing.Optional[typing.Callable]: + return getattr(self, "_{}_resample".format(self.method), None) + + def _get_resampling_size(self, X) -> int: + return self.sz if self.sz > 0 else X.shape[1] + + def fit(self, X, y=None, **kwargs): + """A dummy method such that it complies to the sklearn requirements. + Since this method is completely stateless, it just returns itself. + + Parameters + ---------- + X + Ignored + + Returns + ------- + self + """ + X_ = check_variable_length_input(X) + self._X_fit_dims = X_.shape + + return self + + def _linear_resample(self, X): + target_sz = self._get_resampling_size(X) + if target_sz == 1: + return numpy.nanmean(X, axis=1, keepdims=True) + + n_ts, sz, d = X.shape + equal_size = check_equal_size(X) + X_out = numpy.empty((n_ts, target_sz, d)) + for i in range(X.shape[0]): + if not equal_size: + sz = ts_size(X[i]) + for di in range(d): + X_out[i, :, di] = numpy.interp( + numpy.linspace(0, 1, target_sz), + numpy.linspace(0, 1, sz), + X[i, :sz, di] + ) + return X_out + + def _uniform_resample(self, X): + target_size = self._get_resampling_size(X) + n_ts, sz, d = X.shape + equal_size = check_equal_size(X) + X_out = numpy.empty((n_ts, target_size, d)) + for i in range(X.shape[0]): + if not equal_size: + sz = ts_size(X[i]) + indices = numpy.rint(numpy.linspace(0, sz -1, target_size)).astype("int64") + X_out[i] = X[i, indices] + return X_out + + def _max_resample(self, X): + return self._window_resample_generic(X, numpy.max) + + def _mean_resample(self, X): + return self._window_resample_generic(X, numpy.nanmean) + + def _window_resample_generic(self, X, method): + target_size = self._get_resampling_size(X) + if target_size == 1: + return method(X, axis=1, keepdims=True) + + n_ts, sz, d = X.shape + equal_size = check_equal_size(X) + X_out = numpy.zeros((n_ts, target_size, d)) + for i in range(X.shape[0]): + if not equal_size: + sz = ts_size(X[i]) + if sz == target_size: + X_out[i] = X[i, :sz] + else: + self._window_resample(X[i], X_out[i], method) + return X_out + + def _window_resample(self, timeseries, output, method) -> None: + original_size = ts_size(timeseries) + target_size = self._get_resampling_size(timeseries) + window_size = self.window_size or max(target_size / original_size, original_size / target_size) + indices = numpy.linspace(0, original_size - 1, target_size) + for output_index, input_index in enumerate(indices): + output[output_index] = self._compute_window(timeseries, input_index, method, window_size) + + @staticmethod + def _compute_window(timeseries, index, method, window_size): + return method( + timeseries[max(0, math.ceil(index - window_size/2)): math.floor(index + window_size/2) + 1], + axis=0, + keepdims=True) + + def fit_transform(self, X, y=None, **kwargs): + """Fit to data, then transform it. + + Parameters + ---------- + X : array-like of shape (n_ts, sz, d) + Time series dataset to be resampled. + + Returns + ------- + numpy.ndarray + Resampled time series dataset. + """ + return self.fit(X).transform(X) + + def transform(self, X, y=None, **kwargs): + """Fit to data, then transform it. + + Parameters + ---------- + X : array-like of shape (n_ts, sz, d) + Time series dataset to be resampled. + + Returns + ------- + numpy.ndarray + Resampled time series dataset. + """ + check_is_fitted(self, '_X_fit_dims') + + X_ = check_variable_length_input(X) + X_ = check_dims(X_, X_fit_dims=self._X_fit_dims, extend=False) + + resampler = self._resampler + if resampler is None: + raise ValueError("Resampler {} not implemented.".format(self.method)) + + return resampler(X_) + + def _more_tags(self) -> dict[str, typing.Any]: + more_tags = super()._more_tags() + more_tags.update({'allow_nan': True, ALLOW_VARIABLE_LENGTH: True}) + return more_tags diff --git a/tslearn/preprocessing/scaler.py b/tslearn/preprocessing/scaler.py new file mode 100644 index 000000000..6c24575f4 --- /dev/null +++ b/tslearn/preprocessing/scaler.py @@ -0,0 +1,223 @@ +"""Scalers for time series preprocessing""" +import typing + +import numpy + +from sklearn.base import TransformerMixin +from sklearn.utils.validation import check_is_fitted + +from tslearn.bases import TimeSeriesBaseEstimator +from tslearn.utils import ( + to_time_series_dataset, + check_array, + check_dims +) + +__author__ = 'Romain Tavenard romain.tavenard[at]univ-rennes2.fr' + + +class TimeSeriesScalerBase(TimeSeriesBaseEstimator): + """Base class for time series scalers""" + + def __init__(self, + per_timeseries: bool = True, + per_feature: bool = True) -> None: + + self.per_timeseries = per_timeseries + self.per_feature = per_feature + + def fit(self, X, y=None, **kwargs): + """A dummy method such that it complies to the sklearn requirements. + Since this method is completely stateless, it just returns itself. + + Parameters + ---------- + X + Ignored + + Returns + ------- + self + """ + X_ = check_array(X, allow_nd=True, force_all_finite=False) + X_ = to_time_series_dataset(X_) + self._X_fit_dims = X_.shape + return self + + def _more_tags(self) -> dict[str, typing.Any]: + more_tags = super()._more_tags() + more_tags.update({'allow_nan': True}) + return more_tags + + +class TimeSeriesScalerMeanVariance(TransformerMixin, TimeSeriesScalerBase): + """Scaler for time series datasets. Scales fetures values so that their mean (resp. + standard deviation) in given dimensions is mu (resp. std). + + Parameters + ---------- + mu : float (default: 0.) + Mean of the output time series. + std : float (default: 1.) + Standard deviation of the output time series. + per_timeseries: bool (default: True) + Whether the scaling should be performed per time series. + per_feature: bool (default: True) + Whether the scaling should be performed per feature. + Meaningless for univariate timeseries. + + Notes + ----- + This method requires a dataset of equal-sized time series. + + NaNs within a time series are ignored when calculating mu and std. + + Examples + -------- + >>> TimeSeriesScalerMeanVariance(mu=0., + ... std=1.).fit_transform([[0, 3, 6]]) + array([[[-1.22474487], + [ 0. ], + [ 1.22474487]]]) + >>> TimeSeriesScalerMeanVariance(mu=0., + ... std=1.).fit_transform([[numpy.nan, 3, 6]]) + array([[[nan], + [-1.], + [ 1.]]]) + >>> TimeSeriesScalerMeanVariance(per_timeseries=False, + ... per_feature=False + ... ).fit_transform([[[1, 2], [2, 3]], [[3, 4], [4, 5]]]) + array([[[-1.63299316, -0.81649658], + [-0.81649658, 0. ]], + + [[ 0. , 0.81649658], + [ 0.81649658, 1.63299316]]]) + """ + def __init__(self, + mu: float = 0., + std: float = 1., + per_timeseries: bool = True, + per_feature: bool = True) -> None: + self.mu = mu + self.std = std + super().__init__(per_timeseries=per_timeseries, per_feature=per_feature) + + def transform(self, X, y=None, **kwargs): + """Fit to data, then transform it. + + Parameters + ---------- + X : array-like of shape (n_ts, sz, d) + Time series dataset to be rescaled + + Returns + ------- + numpy.ndarray + Rescaled time series dataset + """ + 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) + + axis = (1,) + if not self.per_timeseries: + axis += (0,) + if not self.per_feature: + axis += (2,) + + mean_t = numpy.nanmean(X_, axis=axis, keepdims=True) + std_t = numpy.nanstd(X_, axis=axis, keepdims=True) + + std_t[numpy.isclose(std_t, 0.)] = 1. + X_ = (X_ - mean_t) * self.std / std_t + self.mu + return X_ + + +class TimeSeriesScalerMinMax(TransformerMixin, TimeSeriesScalerBase): + """Scaler for time series datasets. Scales features values so that their span in given dimensions + is between ``min`` and ``max`` where ``value_range=(min, max)``. + + Parameters + ---------- + value_range : tuple (default: (0., 1.)) + The minimum and maximum value for the output time series. + per_timeseries: bool (default: True) + Wether the scaling should be performed per time series. + per_feature: bool (default: True) + Wether the scaling should be performed per feature. + Meaningless for univariate timeseries. + + Notes + ----- + This method requires a dataset of equal-sized time series. + + NaNs within a time series are ignored when calculating min and max. + + Examples + -------- + >>> TimeSeriesScalerMinMax(value_range=(1., 2.)).fit_transform([[0, 3, 6]]) + array([[[1. ], + [1.5], + [2. ]]]) + >>> TimeSeriesScalerMinMax(value_range=(1., 2.)).fit_transform( + ... [[numpy.nan, 3, 6]] + ... ) + array([[[nan], + [ 1.], + [ 2.]]]) + >>> TimeSeriesScalerMinMax(value_range=(1., 2.), per_timeseries=False, per_feature=False).fit_transform( + ... [[[1, 2], [2, 3]], + ... [[3, 4], [4, 5]]] + ... ) + array([[[1. , 1.25], + [1.25, 1.5 ]], + + [[1.5 , 1.75], + [1.75, 2. ]]]) + """ + def __init__(self, + value_range: tuple[float, float] = (0., 1.), + per_timeseries: bool = True, + per_feature: bool = True) -> None: + self.value_range = value_range + super().__init__(per_timeseries=per_timeseries, per_feature=per_feature) + + def transform(self, X, y=None, **kwargs): + """Will normalize (min-max) each of the timeseries. IMPORTANT: this + transformation is completely stateless, and is applied to each of + the timeseries individually. + + Parameters + ---------- + X : array-like of shape (n_ts, sz, d) + Time series dataset to be rescaled. + + Returns + ------- + numpy.ndarray + Rescaled time series dataset. + """ + if self.value_range[0] >= self.value_range[1]: + raise ValueError("Minimum of desired range must be smaller" + " than maximum. Got %s." % str(self.value_range)) + + 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) + + axis = (1,) + if not self.per_feature: + axis += (2,) + if not self.per_timeseries: + axis += (0,) + + min_t = numpy.nanmin(X_, axis=axis, keepdims=True) + max_t = numpy.nanmax(X_, axis=axis, keepdims=True) + + range_t = max_t - min_t + range_t[numpy.isclose(range_t, 0.)] = 1. + nomin = (X_ - min_t) * (self.value_range[1] - self.value_range[0]) + X_ = nomin / range_t + self.value_range[0] + return X_ diff --git a/tslearn/tests/test_preprocessing.py b/tslearn/tests/test_preprocessing.py index ea642994e..9b0fad8bf 100644 --- a/tslearn/tests/test_preprocessing.py +++ b/tslearn/tests/test_preprocessing.py @@ -1,15 +1,186 @@ import numpy as np +from numpy.testing import assert_array_equal import pytest from tslearn.bases.bases import ALLOW_VARIABLE_LENGTH -from tslearn.preprocessing import (TimeSeriesScalerMeanVariance, - TimeSeriesScalerMinMax, - TimeSeriesImputer) +from tslearn.preprocessing import ( + TimeSeriesScalerMeanVariance, + TimeSeriesScalerMinMax, + TimeSeriesImputer, + TimeSeriesResampler +) from tslearn.utils import to_time_series_dataset, to_time_series +def test_resampler_invalid_method(): + with pytest.raises(ValueError): + TimeSeriesResampler(method="invalid").fit_transform([[1, 2, 3]]) + + +def test_resampler_uniform(): + + # Test downsampling + X = to_time_series_dataset([1, 2, 3, 4, 5, 6, 7]) + resampled = TimeSeriesResampler(sz=3, method="uniform").fit_transform(X) + expected = np.array([[[1.], [4.], [7.]]]) + assert_array_equal(resampled, expected) + + # Test downsampling variable length dataset + X = to_time_series_dataset([[1, 2, 3, 4, 5, 6, 7], [1, 2, 3, 4, 5]]) + resampled = TimeSeriesResampler(sz=3, method="uniform").fit_transform(X) + expected = np.array([ + [[1.], [4.], [7.]], + [[1.], [3.], [5.]]] + ) + assert_array_equal(resampled, expected) + + # Test downsampling multivariate + X = to_time_series_dataset([[[1, 2], [3, 4], [5, 6], [7, 8] , [1, 2], [3, 4]]]) + resampled = TimeSeriesResampler(sz=3, method="uniform").fit_transform(X) + expected = np.array([[[1., 2.], [5., 6.], [3., 4.]]]) + assert_array_equal(resampled, expected) + + # Test upsampling + X = to_time_series_dataset([1, 2, 3, 4, 5]) + resampled = TimeSeriesResampler(sz=7, method="uniform").fit_transform(X) + expected = np.array([[[1.], [2.], [2.], [3.], [4.], [4.], [5.]]]) + assert_array_equal(resampled, expected) + +def test_resampler_linear(): + + # Test downsampling target_size = 1 + X = to_time_series_dataset([1, 2, 3, 4, 5, 6, 7]) + resampled = TimeSeriesResampler(sz=1, method="linear").fit_transform(X) + expected = np.array([[[4.]]]) + assert_array_equal(resampled, expected) + + # Test downsampling + X = to_time_series_dataset([1, 2, 3, 4, 5, 6, 7]) + resampled = TimeSeriesResampler(sz=3, method="linear").fit_transform(X) + expected = np.array([[[1.], [4.], [7.]]]) + assert_array_equal(resampled, expected) + + # Test downsampling variable length dataset + X = to_time_series_dataset([[1, 2, 3, 4, 5, 6, 7], [1, 2, 3, 4, 5]]) + resampled = TimeSeriesResampler(sz=3, method="linear").fit_transform(X) + expected = np.array([ + [[1.], [4.], [7.]], + [[1.], [3.], [5.]]] + ) + assert_array_equal(resampled, expected) + + # Test downsampling multivariate + X = to_time_series_dataset([[[1, 2], [3, 4], [5, 6], [7, 8] , [1, 2], [3, 4]]]) + resampled = TimeSeriesResampler(sz=3, method="linear").fit_transform(X) + expected = np.array([[[1., 2.], [6., 7.], [3., 4.]]]) + np.testing.assert_array_almost_equal(resampled, expected) + + # Test upsampling + X = to_time_series_dataset([1, 2, 3, 4, 5]) + resampled = TimeSeriesResampler(sz=7, method="linear").fit_transform(X) + expected = np.array([[[1.], [5/3], [7/3], [3.], [11/3], [13/3], [5.]]]) + np.testing.assert_array_almost_equal(resampled, expected) + + +def test_resampler_mean(): + + # Test downsampling target_size = 1 + X = to_time_series_dataset([1, 2, 3, 4, 5, 6, 7]) + resampled = TimeSeriesResampler(sz=1, method="mean").fit_transform(X) + expected = np.array([[[4.]]]) + assert_array_equal(resampled, expected) + + # Test downsampling with integer size / target_size + X = to_time_series_dataset([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]) + resampled = TimeSeriesResampler(sz=5, method="mean").fit_transform(X) + expected = np.array([[[1.5], [3.5], [5.5], [7.5], [9.5]]]) + assert_array_equal(resampled, expected) + + # Test downsampling with float size / target_size + X = to_time_series_dataset([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]) + resampled = TimeSeriesResampler(sz=7, method="mean").fit_transform(X) + expected = np.array([[[1], [2.5], [4.], [5.5], [7.], [8.5], [10]]]) + assert_array_equal(resampled, expected) + + # Test downsampling variable length dataset + X = to_time_series_dataset([[1, 2, 3, 4, 5, 6, 7], [1, 2, 3, 4, 5, 6]]) + resampled = TimeSeriesResampler(sz=3, method="mean").fit_transform(X) + expected = np.array([ + [[1.5], [4.], [6.5]], + [[1.5], [3.5], [5.5]]] + ) + assert_array_equal(resampled, expected) + + # Test downsampling multivariate + X = to_time_series_dataset([[[1, 4], [3, 2], [5, 1], [7, 2], [1, 8], [3, 4]]]) + resampled = TimeSeriesResampler(sz=3, method="mean").fit_transform(X) + expected = np.array([[[2., 3.], [6., 1.5], [2., 6.]]]) + np.testing.assert_array_almost_equal(resampled, expected) + + # Test downsampling multivariate with nans + X = to_time_series_dataset([[[1, 4], [3, 2], [np.nan, 1], [7, 2], [1, 8], [3, 4], [7, 9]]]) + resampled = TimeSeriesResampler(sz=3, method="mean").fit_transform(X) + expected = np.array([[[2, 3], [4., 11/3], [5, 6.5]]]) + np.testing.assert_array_almost_equal(resampled, expected) + + # Test downsampling multivariate with window and nans + X = to_time_series_dataset([[[1, 4], [3, 2], [np.nan, 1], [7, 2], [1, 8], [3, 4], [7, 9]]]) + resampled = TimeSeriesResampler(sz=3, method="mean", window_size=4).fit_transform(X) + expected = np.array([[[2, 7/3], [3.5, 17/5], [11/3, 7]]]) + np.testing.assert_array_almost_equal(resampled, expected) + + # Test upsampling + X = to_time_series_dataset([1, 2, 3, 4, 5]) + resampled = TimeSeriesResampler(sz=7, method="mean").fit_transform(X) + expected = np.array([[[1.], [1.5], [2.5], [3], [3.5], [4.5], [5.]]]) + np.testing.assert_array_almost_equal(resampled, expected) + + # Test upsampling multivariate + X = to_time_series_dataset([[[1, 4], [3, 2], [5, 1], [7, 2], [1, 8], [3, 4]]]) + resampled = TimeSeriesResampler(sz=10, method="mean").fit_transform(X) + expected = np.array([[ + [1, 4], [2., 3.], [3., 2.], [4, 1.5], [6., 1.5], [6, 1.5], [4, 5], [1, 8], [2, 6], [3., 4.]]] + ) + np.testing.assert_array_almost_equal(resampled, expected) + + # Test upsampling with window + X = to_time_series_dataset([1, 2, 3, 4, 5]) + resampled = TimeSeriesResampler(sz=7, method="mean", window_size=3).fit_transform(X) + expected = np.array([[[1.5], [2], [2], [3], [4], [4], [4.5]]]) + np.testing.assert_array_almost_equal(resampled, expected) + + +def test_resampler_max(): + + # Test downsampling with integer size / target_size + X = to_time_series_dataset([1, 2, 3, 4, 5, 6]) + resampled = TimeSeriesResampler(sz=3, method="max").fit_transform(X) + expected = np.array([ + [[2], [4], [6]] + ]) + np.testing.assert_array_equal(resampled, expected) + + # Test downsampling with non integer size / target_size + X = to_time_series_dataset([1, 2, 3, 4, 5, 6, 7]) + resampled = TimeSeriesResampler(sz=3, method="max").fit_transform(X) + expected = np.array([ + [[2.], [5], [7]] + ]) + np.testing.assert_array_equal(resampled, expected) + + # Test variable length + X = to_time_series_dataset([[1, 2, 3, 4, 5, 6], [6, 8, 10, 12], [4, 5, 6, 7, 8, 9, 10, 11, 12]]) + resampled = TimeSeriesResampler(sz=6, method="max").fit_transform(X) + expected = np.array([ + [[1.], [2.], [3.], [4.], [5.], [6.]], + [[6], [8], [8], [10], [12], [12]], + [[4.], [6.], [7.], [9.], [11.], [12.]] + ]) + np.testing.assert_array_equal(resampled, expected) + + def test_single_value_ts_no_nan(): X = to_time_series_dataset([[1, 1, 1, 1]]) @@ -119,6 +290,11 @@ def test_min_max_scaler_modes(): ) +def test_min_max_scaler_invalid_range(): + with pytest.raises(ValueError): + TimeSeriesScalerMinMax((1,0)).fit_transform([[1, 2, 3]]) + + def test_mean_variance_scaler_modes(): univariate_dataset = [ [1, 2, 3],