Merge branch 'main' into add-nbeats-v2

Author: harshsomankar123-tech

pytorch_forecasting/adapters/__init__.py

@@ -0,0 +1,5 @@
+"""Module for pytorch-forecasting adapters"""
+
+from pytorch_forecasting.adapters.scaler_adapters import ScalerAdapter
+
+__all__ = ["ScalerAdapter"]

pytorch_forecasting/adapters/scaler_adapters.py

@@ -0,0 +1,201 @@
+import pandas as pd
+import torch
+
+from pytorch_forecasting._registry import all_objects
+from pytorch_forecasting.adapters.scaler_strategy import (
+    ScalerStrategy,
+)
+from pytorch_forecasting.adapters.utils import (
+    ArrayLike,
+    _to_numpy,
+    _to_tensor,
+)
+from pytorch_forecasting.data.encoders import (
+    MultiNormalizer,
+)
+
+
+def get_scaler_strategy(scaler) -> ScalerStrategy:
+    """Single dispatch point: the only place that inspects scaler type."""
+    discovered = all_objects(
+        object_types="scaler_strategy",
+        return_names=False,
+    )
+    for strategy_cls in discovered:
+        if strategy_cls._is_applicable(scaler):
+            return strategy_cls()
+    return ScalerStrategy()
+
+
+class ScalerAdapter:
+    """
+    Unified array-in / tensor-out interface for single and multi-target scalers.
+
+    Accepts torch.Tensor, np.ndarray, or pd.Series as input. Output is always
+    a torch.Tensor.  Type-specific behavior (sklearn scalers, GroupNormalizer,
+    NaNLabelEncoder, EncoderNormalizer, ...) is delegated to a strategy chosen
+    once at construction time (see ``adapters/scaler_strategy.py``).
+
+
+    Parameters
+    ----------
+    scaler : object
+        The underlying scaling/encoding instance. Accepted types, their expected
+            origins, and assumed API contracts are:
+
+            * scikit-learn scalers (from ``sklearn.preprocessing``):
+                Implements ``.fit(X)` and ``.transform(X)``. Expects 2D
+                numpy arrays of shape ``(n_samples, 1)``. Outputs numpy arrays.
+
+            * ``TorchNormalizer``  (from ``pytorch_forecasting.data.encoders``):
+                Implements `.fit(data)` and `.transform(data)`. Expects 1D
+                tensors or numpy arrays. Output can be tensor or array.
+
+            *``EncoderNormalizer`` (from ``pytorch_forecasting.data.encoders``):
+                Implements `.fit(data)` and `.transform(data)`. Expects 1D
+                tensors or numpy arrays. Output can be tensor or array.
+                `EncoderNormalizer` signals that it must be fit per-sequence.
+
+            * ``NaNLabelEncoder`` (from `pytorch_forecasting.data.encoders`):
+                Implements ``.fit(data)`` and ``.transform(data)``. Expects a
+                1D ``pd.Series`` (or 1D array). Used for categorical encoding.
+
+            * ``GroupNormalizer`` (from ``pytorch_forecasting.data.encoders``):
+                Implements ``.fit(data, X)`` and ``.transform(data, X)``. Expects
+                `data` as a 1D ``pd.Series`` and `X` as a ``pd.DataFrame`` containing
+                required group columns to compute grouped statistics.
+
+            * ``MultiNormalizer`` (from ``pytorch_forecasting.data.encoders``):
+                Implements ``.fit(data, X)`` and ``.transform(data.T, X)``.
+                Expects 2D array-like inputs of shape ``(n_samples, n_targets)``.
+                Must expose a ``.normalizers`` attribute (iterable) containing the
+                individual sub-normalizers for each target.
+    """
+
+    def __init__(self, scaler):
+        self._scaler = scaler
+        self.is_multi = isinstance(scaler, MultiNormalizer)
+
+        if self.is_multi:
+            self._sub_adapters = [ScalerAdapter(norm) for norm in scaler.normalizers]
+            self._strategy = None
+            self.is_label_encoder = False
+            self.fit_per_sequence = any(a.fit_per_sequence for a in self._sub_adapters)
+        else:
+            self._strategy = get_scaler_strategy(scaler) if scaler is not None else None
+            self.is_label_encoder = (
+                self._strategy.get_tag("is_label_encoder", None)
+                if self._strategy
+                else False
+            )
+            self.fit_per_sequence = (
+                self._strategy.get_tag("fit_per_sequence", None)
+                if self._strategy
+                else False
+            )
+
+    @property
+    def label_encoder_mask(self) -> list[bool]:
+        """Per-target bool list indicating which sub-normalizers are label encoders."""
+        if self.is_multi:
+            return [sub.is_label_encoder for sub in self._sub_adapters]
+        return [self.is_label_encoder]
+
+    def _prepare_input(self, data: ArrayLike) -> ArrayLike:
+        """Coerce data to the type the underlying scaler expects."""
+        if self.is_multi:
+            arr = _to_numpy(data)
+            return arr if arr.ndim == 2 else arr[:, None]
+        return self._strategy.prepare_input(data)
+
+    def fit(self, data: ArrayLike, X: pd.DataFrame = None) -> "ScalerAdapter":
+        """Fit the scaler.
+
+        Parameters
+        ----------
+        data : tensor, ndarray, or Series
+            Shape ``(n_samples,)`` for single-target or
+            ``(n_samples, n_targets)`` for multi-target.
+        X : pd.DataFrame, optional
+            Group columns. Required when scaler is GroupNormalizer or
+            when MultiNormalizer contains GroupNormalizer sub-normalizers.
+        """
+        if self._scaler is None:
+            return self
+
+        prepared = self._prepare_input(data)
+        if self.is_multi:
+            self._scaler.fit(prepared, X)
+            return self
+
+        self._strategy.fit(self._scaler, prepared, X)
+        return self
+
+    def transform(self, data: ArrayLike, X: pd.DataFrame = None) -> torch.Tensor:
+        """Transform data, always returning a torch.Tensor.
+
+        Parameters
+        ----------
+        data : tensor, ndarray, or Series
+            Shape ``(n_samples,)`` for single-target or
+            ``(n_samples, n_targets)`` for multi-target.
+        X : pd.DataFrame, optional
+            Group columns. Required when scaler is GroupNormalizer or
+            when MultiNormalizer contains GroupNormalizer sub-normalizers.
+
+        Returns
+        -------
+        torch.Tensor
+            Same shape as input.
+        """
+        if self._scaler is None:
+            return _to_tensor(data)
+        prepared = self._prepare_input(data)
+
+        if self.is_multi:
+            results = self._scaler.transform(prepared.T, X)
+            return torch.stack([_to_tensor(r) for r in results], dim=-1)
+
+        return self._strategy.transform(self._scaler, prepared, data, X)
+
+    def fit_transform(self, data: ArrayLike, X: pd.DataFrame = None) -> torch.Tensor:
+        return self.fit(data, X).transform(data, X)
+
+    def fit_transform_sequence(
+        self, data: ArrayLike, X: pd.DataFrame = None
+    ) -> torch.Tensor:
+        """Fit-and-transform only per-sequence sub-normalizers; transform the rest.
+
+        Used at ``__getitem__`` time for encoder windows. Non-per-sequence
+        normalizers use their already-fitted global state.
+
+        For single-target adapters this collapses to fit_transform
+        (EncoderNormalizer) or transform (everything else).
+
+        Parameters
+        ----------
+        data : tensor, ndarray, or Series
+            Shape ``(enc_length,)`` or ``(enc_length, n_targets)``.
+
+        Returns
+        -------
+        torch.Tensor
+            Same shape as input.
+        """
+        if not self.is_multi:
+            return (
+                self.fit_transform(data, X)
+                if self.fit_per_sequence
+                else _to_tensor(data)
+            )
+
+        t = _to_tensor(data)
+        if t.ndim == 1:
+            t = t.unsqueeze(-1)
+
+        columns = []
+        for idx, sub in enumerate(self._sub_adapters):
+            col = t[:, idx]
+            col = sub.fit_transform(col, X) if sub.fit_per_sequence else col
+            columns.append(col.unsqueeze(-1))
+        return torch.cat(columns, dim=-1)

pytorch_forecasting/adapters/scaler_strategy.py

@@ -0,0 +1,122 @@
+from abc import abstractmethod
+
+import pandas as pd
+import torch
+
+from pytorch_forecasting import EncoderNormalizer, GroupNormalizer, NaNLabelEncoder
+from pytorch_forecasting.adapters.utils import (
+    ArrayLike,
+    _is_sklearn_transformer,
+    _series_from,
+    _to_numpy,
+    _to_tensor,
+    _was_2d_singleton,
+)
+from pytorch_forecasting.base._base_object import _BaseObject
+
+
+class ScalerStrategy(_BaseObject):
+    """Default behavior for scalers."""
+
+    _tags = {
+        "object_type": "scaler_strategy",
+        "fit_per_sequence": False,
+        "is_label_encoder": False,
+    }
+
+    @staticmethod
+    @abstractmethod
+    def _is_applicable(scaler) -> bool:
+        """Whether the scaler follows the given strategy."""
+
+    def prepare_input(self, data: ArrayLike) -> ArrayLike:
+        t = _to_tensor(data)
+        return t.squeeze(-1) if (t.ndim == 2 and t.shape[1] == 1) else t
+
+    def fit(self, scaler, prepared: ArrayLike, X: pd.DataFrame = None) -> None:
+        scaler.fit(prepared)
+
+    def transform(
+        self, scaler, prepared: ArrayLike, data: ArrayLike, X: pd.DataFrame = None
+    ) -> torch.Tensor:
+        result = _to_tensor(scaler.transform(prepared))
+        return result.unsqueeze(-1) if _was_2d_singleton(data) else result
+
+
+class EncoderNormalizerStrategy(ScalerStrategy):
+    """EncoderNormalizer must be re-fit per encoder window."""
+
+    _tags = {
+        "fit_per_sequence": True,
+    }
+
+    @staticmethod
+    def _is_applicable(scaler) -> bool:
+        return isinstance(scaler, EncoderNormalizer)
+
+
+class SklearnStrategy(ScalerStrategy):
+    """sklearn scalers expect/return 2D numpy arrays."""
+
+    @staticmethod
+    def _is_applicable(scaler) -> bool:
+        return _is_sklearn_transformer(scaler)
+
+    def prepare_input(self, data: ArrayLike) -> ArrayLike:
+        return _to_numpy(data).reshape(-1, 1)
+
+    def fit(self, scaler, prepared: ArrayLike, X: pd.DataFrame = None) -> None:
+        scaler.fit(prepared)
+
+    def transform(
+        self, scaler, prepared: ArrayLike, data: ArrayLike, X: pd.DataFrame = None
+    ) -> torch.Tensor:
+        original_shape = _to_numpy(data).shape
+        result = scaler.transform(prepared).reshape(original_shape)
+        return torch.tensor(result, dtype=torch.float32)
+
+
+class LabelEncoderStrategy(ScalerStrategy):
+    _tags = {
+        "is_label_encoder": True,
+    }
+
+    @staticmethod
+    def _is_applicable(scaler) -> bool:
+        return isinstance(scaler, NaNLabelEncoder)
+
+    def prepare_input(self, data: ArrayLike) -> ArrayLike:
+        return _series_from(data)
+
+    def transform(
+        self, scaler, prepared: ArrayLike, data: ArrayLike, X: pd.DataFrame = None
+    ) -> torch.Tensor:
+        result = _to_tensor(scaler.transform(prepared))
+        return result.unsqueeze(-1) if _was_2d_singleton(data) else result
+
+
+class GroupNormalizerStrategy(ScalerStrategy):
+    @staticmethod
+    def _is_applicable(scaler) -> bool:
+        return isinstance(scaler, GroupNormalizer)
+
+    def prepare_input(self, data: ArrayLike) -> ArrayLike:
+        return _series_from(data)
+
+    def fit(self, scaler, prepared: ArrayLike, X: pd.DataFrame = None) -> None:
+        assert X is not None, (
+            "GroupNormalizer requires X (DataFrame with group columns) "
+            "to be passed to fit()."
+        )
+        scaler.fit(prepared, X)
+
+    def transform(
+        self, scaler, prepared: ArrayLike, data: ArrayLike, X: pd.DataFrame = None
+    ) -> torch.Tensor:
+        assert X is not None, (
+            "GroupNormalizer requires X (DataFrame with group columns) "
+            "to be passed to transform()."
+        )
+        input_was_2d = isinstance(data, torch.Tensor) and data.ndim == 2
+        result = _to_tensor(scaler.transform(prepared, X))
+        return result.unsqueeze(-1) if input_was_2d else result

pytorch_forecasting/adapters/utils.py

@@ -0,0 +1,46 @@
+import numpy as np
+import pandas as pd
+from sklearn.base import TransformerMixin
+import torch
+
+from pytorch_forecasting.data.encoders import TransformMixIn
+
+ArrayLike = torch.Tensor | np.ndarray | pd.Series
+
+
+def _to_numpy(data: ArrayLike) -> np.ndarray:
+    """Convert any array-like to numpy."""
+    if isinstance(data, torch.Tensor):
+        return data.detach().numpy()
+    elif isinstance(data, pd.Series):
+        return data.to_numpy()
+    return np.asarray(data)
+
+
+def _to_tensor(data: ArrayLike, dtype=torch.float32) -> torch.Tensor:
+    """Convert any array-like to a float32 tensor."""
+    if isinstance(data, torch.Tensor):
+        return data.to(dtype)
+    elif isinstance(data, pd.Series):
+        return torch.tensor(data.to_numpy(), dtype=dtype)
+    return torch.tensor(np.asarray(data), dtype=dtype)
+
+
+def _is_sklearn_transformer(scaler):
+    is_sklearn_transform = isinstance(scaler, TransformerMixin)
+    is_ptf_transform = isinstance(scaler, TransformMixIn)
+
+    return is_sklearn_transform and not is_ptf_transform
+
+
+def _series_from(data: ArrayLike) -> pd.Series:
+    """Prep for scalers that want a pd.Series (label encoder, group normalizer)."""
+    if isinstance(data, pd.Series):
+        return data
+    np_data = _to_numpy(data)
+    return pd.Series(np_data.squeeze() if np_data.ndim == 2 else np_data)
+
+
+def _was_2d_singleton(data: ArrayLike) -> bool:
+    """True if `data` is a torch.Tensor of shape (n, 1)."""
+    return isinstance(data, torch.Tensor) and data.ndim == 2 and data.shape[1] == 1