[ENH] add BaseSeriesTransformer - ExpSmoothingSeriesTransformer

aeon/transformations/series/_exp_smoothing.py

@@ -0,0 +1,95 @@
+"""Exponential smoothing transformation."""
+
+__maintainer__ = ["Datadote"]
+__all__ = ["ExpSmoothingSeriesTransformer"]
+
+from typing import Union
+
+import numpy as np
+
+from aeon.transformations.series.base import BaseSeriesTransformer
+
+
+class ExpSmoothingSeriesTransformer(BaseSeriesTransformer):
+    """Filter a time series using exponential smoothing.
+
+    - Exponential smoothing (EXP) is a generalisaton of moving average smoothing that
+    assigns a decaying weight to each element rather than averaging over a window.
+    - Assume time series T = [t_0, ..., t_j], and smoothed values S = [s_0, ..., s_j]
+    - Then, s_0 = t_0 and s_j = alpha * t_j + (1 - alpha) * s_j-1
+    where 0 ≤ alpha ≤ 1. If window_size is given, alpha is overwritten, and set as
+    alpha = 2. / (window_size + 1)
+
+    Parameters
+    ----------
+    alpha: float, default=0.2
+        decaying weight. Range [0, 1]. Overwritten by window_size if window_size exists
+    window_size: int or float or None, default=None
+        If window_size is specified, alpha is set to 2. / (window_size + 1)
+
+    References
+    ----------
+    Large, J., Southam, P., Bagnall, A. (2019).
+        Can Automated Smoothing Significantly Improve Benchmark Time Series
+        Classification Algorithms?. In: Pérez García, H., Sánchez González,
+        L., CastejónLimas, M., Quintián Pardo, H., Corchado Rodríguez, E. (eds) Hybrid
+        Artificial Intelligent Systems. HAIS 2019. Lecture Notes in Computer Science(),
+        vol 11734. Springer, Cham. https://doi.org/10.1007/978-3-030-29859-3_5
+        https://arxiv.org/abs/1811.00894
+
+    Examples
+    --------
+    >>> import numpy as np
+    >>> from aeon.transformations.series._exp_smoothing import \
+        ExpSmoothingSeriesTransformer
+    >>> X = np.array([-2, -1,  0,  1,  2])
+    >>> transformer = ExpSmoothingSeriesTransformer(0.5)
+    >>> Xt = transformer.fit_transform(X)
+    >>> print(Xt)
+    [[-2.     -1.5    -0.75    0.125   1.0625]]
+    >>> X = np.array([[1, 2, 3, 4], [10, 9, 8, 7]])
+    >>> Xt = transformer.fit_transform(X)
+    >>> print(Xt)
+    [[ 1.     1.5    2.25   3.125]
+     [10.     9.5    8.75   7.875]]
+    """
+
+    _tags = {
+        "capability:multivariate": True,
+        "X_inner_type": "np.ndarray",
+        "fit_is_empty": True,
+    }
+
+    def __init__(
+        self, alpha: float = 0.2, window_size: Union[int, float, None] = None
+    ) -> None:
+        if not 0 <= alpha <= 1:
+            raise ValueError(f"alpha must be in range [0, 1], got {alpha}")
+        if window_size is not None and window_size <= 0:
+            raise ValueError(f"window_size must be > 0, got {window_size}")
+        super().__init__(axis=1)
+        self.alpha = alpha if window_size is None else 2.0 / (window_size + 1)
+        self.window_size = window_size
+
+    def _transform(self, X, y=None):
+        """Transform X and return a transformed version.
+
+        private _transform containing core logic, called from transform
+
+        Parameters
+        ----------
+        X : np.ndarray
+            Data to be transformed
+        y : ignored argument for interface compatibility
+            Additional data, e.g., labels for transformation
+
+        Returns
+        -------
+        Xt: 2D np.ndarray
+            transformed version of X
+        """
+        Xt = np.zeros_like(X, dtype="float")
+        Xt[:, 0] = X[:, 0]
+        for i in range(1, Xt.shape[1]):
+            Xt[:, i] = self.alpha * X[:, i] + (1 - self.alpha) * Xt[:, i - 1]
+        return Xt

aeon/transformations/series/tests/test_exp_smoothing.py

@@ -0,0 +1,60 @@
+"""Tests for ExpSmoothingSeriesTransformer."""
+
+__maintainer__ = ["Datadote"]
+
+import numpy as np
+import pytest
+
+from aeon.transformations.series._exp_smoothing import ExpSmoothingSeriesTransformer
+
+TEST_DATA = [np.array([-2, -1, 0, 1, 2]), np.array([[1, 2, 3, 4], [10, 9, 8, 7]])]
+EXPECTED_RESULTS = [
+    np.array([[-2.0, -1.5, -0.75, 0.125, 1.0625]]),
+    np.array([[1.0, 1.5, 2.25, 3.125], [10.0, 9.5, 8.75, 7.875]]),
+]
+
+
+def test_input_1d_array():
+    """Test inputs of dimension 1."""
+    transformer = ExpSmoothingSeriesTransformer(0.5)
+    idx_data = 0
+    Xt = transformer.fit_transform(TEST_DATA[idx_data])
+    np.testing.assert_almost_equal(Xt, EXPECTED_RESULTS[idx_data], decimal=5)
+
+
+def test_input_2d_array():
+    """Test inputs of dimension 2."""
+    transformer = ExpSmoothingSeriesTransformer(0.5)
+    idx_data = 1
+    Xt = transformer.fit_transform(TEST_DATA[idx_data])
+    np.testing.assert_almost_equal(Xt, EXPECTED_RESULTS[idx_data], decimal=5)
+
+
+@pytest.mark.parametrize("alpha_window", [(0.2, 9), (0.5, 3), (1, 1)])
+def test_window_size_matches_alpha(alpha_window):
+    """Check same output results using equivalent alpha and window_size."""
+    alpha, window_size = alpha_window
+    transformer1 = ExpSmoothingSeriesTransformer(alpha=alpha)
+    transformer2 = ExpSmoothingSeriesTransformer(window_size=window_size)
+    for i in range(len(TEST_DATA)):
+        Xt1 = transformer1.fit_transform(TEST_DATA[i])
+        Xt2 = transformer2.fit_transform(TEST_DATA[i])
+        np.testing.assert_array_almost_equal(Xt1, Xt2, decimal=5)
+
+
+def test_alpha_less_than_zero():
+    """Test alpha less than zero."""
+    with pytest.raises(ValueError):
+        ExpSmoothingSeriesTransformer(-0.5)
+
+
+def test_alpha_greater_than_one():
+    """Test alpha greater than one."""
+    with pytest.raises(ValueError):
+        ExpSmoothingSeriesTransformer(2.0)
+
+
+def test_window_size_than_one():
+    """Test window_size < 0."""
+    with pytest.raises(ValueError):
+        ExpSmoothingSeriesTransformer(window_size=0)