Add ReseampleAggregate transformer

Author: mazzma12

src/adtk/transformer/__init__.py

@@ -16,6 +16,7 @@
     DoubleRollingAggregate,
     Retrospect,
     RollingAggregate,
+    ResampleAggregate,
     StandardScale,
 )
 from ._transformer_hd import (
@@ -54,6 +55,7 @@ def print_all_models() -> None:
 
 __all__ = [
     "RollingAggregate",
+    "ResampleAggregate",
     "DoubleRollingAggregate",
     "ClassicSeasonalDecomposition",
     "Retrospect",

src/adtk/transformer/_transformer_1d.py

@@ -127,6 +127,143 @@ def _predict_core(self, s: pd.Series) -> pd.Series:
         return (s - mean) / std
 
 
+
+class ResampleAggregate(_NonTrainableUnivariateTransformer):
+    """Transformer that resample a time series to a specified frequency
+
+    Parameters
+    ----------
+    rule: str. The new frequency for resampling. Examples: '1D', '2H', '30min'.
+
+    agg: str or function
+        Aggregation method applied to series.
+        If str, must be one of supported built-in methods:
+
+        - 'mean': mean of all values in a rolling window.
+        - 'median': median of all values in a rolling window.
+        - 'sum': summation of all values in a rolling window.
+        - 'min': minimum of all values in a rolling window.
+        - 'max': maximum of all values in a rolling window.
+        - 'std': sample standard deviation of all values in a rolling window.
+        - 'var': sample variance of all values in a rolling window.
+        - 'skew': skewness of all values in a rolling window.
+        - 'kurt': kurtosis of all values in a rolling window.
+        - 'count': number of non-nan values in a rolling window.
+        - 'nnz': number of non-zero values in a rolling window.
+        - 'nunique': number of unique values in a rolling window.
+        - 'quantile': quantile of all values in a rolling window. Require
+          percentile parameter `q` in in parameter `agg_params`, which is a
+          float or a list of float between 0 and 1 inclusive.
+        - 'iqr': interquartile range, i.e. difference between 75% and 25%
+          quantiles.
+        - 'idr': interdecile range, i.e. difference between 90% and 10%
+          quantiles.
+
+        If function, it should accept a rolling window in form of a pandas
+        Series, and return either a scalar or a 1D numpy array. To specify
+        names of outputs, specify a list of strings as a parameter `names` in
+        parameter `agg_params`.
+
+        Default: 'mean'
+
+    agg_params: dict, optional
+        Parameters of aggregation function. Default: None.
+
+    """
+
+    def __init__(
+        self,
+        rule: str,
+        agg: Union[
+            str, Callable[[pd.Series], Union[float, np.ndarray]]
+        ] = "mean",
+        agg_params: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        super().__init__()
+        self.rule = rule
+        self.agg = agg
+        self.agg_params = agg_params
+
+    @property
+    def _param_names(self) -> Tuple[str, ...]:
+        return ("freq", "agg", "agg_params")
+
+    def _predict_core(self, s: pd.Series) -> Union[pd.Series, pd.DataFrame]:
+        if not (
+            s.index.is_monotonic_increasing or s.index.is_monotonic_decreasing
+        ):
+            raise ValueError("Time series must have a monotonic time index. ")
+
+        agg = self.agg
+        agg_params = self.agg_params if (self.agg_params is not None) else {}
+
+        resample = s.resample(
+            rule=self.rule
+        )  # type: Union[pd.Series, pd.DataFrame]
+
+        aggList = [
+            "mean",
+            "median",
+            "sum",
+            "min",
+            "max",
+            "quantile",
+            "iqr",
+            "idr",
+            "count",
+            "nnz",
+            "nunique",
+            "std",
+            "var",
+            "skew",
+            "kurt",
+            "hist",
+        ]
+        if agg in [
+            "mean",
+            "median",
+            "sum",
+            "min",
+            "max",
+            "count",
+            "std",
+            "var",
+            "skew",
+            "kurt",
+        ]:
+            s_resample = resample.agg(agg)
+        elif agg == "nunique":
+            s_resample = resample.agg(lambda x: len(np.unique(x.dropna())))
+        elif agg == "nnz":
+            s_resample = resample.agg(np.count_nonzero)
+        elif agg == "quantile":
+            if hasattr(agg_params["q"], "__iter__"):
+                s_resample = pd.concat(
+                    [
+                        resample.quantile(q).rename("q{}".format(q))
+                        for q in agg_params["q"]
+                    ],
+                    axis=1,
+                )
+            else:
+                s_resample = resample.quantile(agg_params["q"])
+        elif agg == "iqr":
+            s_resample = resample.quantile(0.75) - resample.quantile(0.25)
+        elif agg == "idr":
+            s_resample = resample.quantile(0.9) - resample.quantile(0.1)
+
+        elif callable(agg):
+            s_resample = resample.agg(agg)
+
+        else:
+            raise ValueError(f"Attribute agg must be one of {aggList}")
+
+        if isinstance(s_resample, pd.Series):
+            s_resample.name = s.name
+
+        return s_resample
+
+
 class RollingAggregate(_NonTrainableUnivariateTransformer):
     """Transformer that rolls a sliding window along a time series, and
     aggregates using a user-selected operation.