Repository: aeon-toolkit/aeon
Pull request #2019 (2024-08-31 19:28:08)

[ENH] add BaseSeriesTransformer - ExpSmoothingSeriesTransformer
#### Reference Issues/PRs
- Part of #1920

#### What does this implement/fix? Explain your changes.
- Add smoothing filter ExpSmoothingSeriesTransformer (note abbreviated Exp)
- Tests check for different alpha, and window_size values

#### Does your contribution introduce a new dependency? If yes, which one?
- No new dependencies. Imports libraries: numpy, typing

#### Any other comments?
- [Arxiv paper reference](https://arxiv.org/abs/1811.00894) has an error in the exponential smoothing equation. The decaying term should be based on s_j-1, and not t_j-1 . I'm not sure what the [Springer reference](https://doi.org/10.1007/978-3-030-29859-3_5) has. I do not have access
Paper:  s_0 = t_0 and s_j = alpha * t_j + (1 - alpha) * t_j-1
Corrected: s_0 = t_0 and s_j = alpha * t_j + (1 - alpha) * s_j-1
- Allows user to use "alpha" or "window_size". If "window_size" is specified, alpha is overwritten with 2. / (window_size + 1)
- For the reference paper link, I added an arxiv link to the paper for folks without access to Springer

### PR checklist

<!--
Please go through the checklist below. Please feel free to remove points if they are not applicable.
-->

##### For new estimators and functions
- [ ] I've added the estimator to the online [API documentation](https://www.aeon-toolkit.org/en/latest/api_reference.html).
- [X] (OPTIONAL) I've added myself as a `__maintainer__` at the top of relevant files and want to be contacted regarding its maintenance. Unmaintained files may be removed. This is for the full file, and you should not add yourself if you are just making minor changes or do not want to help maintain its contents.

##### For developers with write access
- [ ] (OPTIONAL) I've updated aeon's [CODEOWNERS](https://github.com/aeon-toolkit/aeon/blob/main/CODEOWNERS) to receive notifications about future changes to these files.


<!--
Thanks for contributing!
-->


aeon-actions-bot[bot]: 
## Thank you for contributing to `aeon`

I have added the following labels to this PR based on the title: **[ $\\color{#FEF1BE}{\\textsf{enhancement}}$ ]**.
I have added the following labels to this PR based on the changes made: **[ $\\color{#41A8F6}{\\textsf{transformations}}$ ]**. Feel free to change these if they do not properly represent the PR.

The [Checks](https://github.com/aeon-toolkit/aeon/pull/2019/checks) tab will show the status of our automated tests. You can click on individual test runs in the tab or "Details" in the panel below to see more information if there is a failure.

If our `pre-commit` code quality check fails, any trivial fixes will automatically be pushed to your PR unless it is a draft.

Don't hesitate to ask questions on the `aeon` [Slack](https://join.slack.com/t/aeon-toolkit/shared_invite/zt-22vwvut29-HDpCu~7VBUozyfL_8j3dLA) channel if you have any.

### PR CI actions

These checkboxes will add labels to enable/disable CI functionality for this PR. This may not take effect immediately, and a new commit may be required to run the new configuration.

- [ ] Run `pre-commit` checks for all files
- [ ] Run all `pytest` tests and configurations
- [ ] Run all notebook example tests
- [ ] Run numba-disabled `codecov` tests
- [ ] Stop automatic `pre-commit` fixes (always disabled for drafts)
- [ ] Push an empty commit to re-run CI checks
    

Patch:
diff --git a/aeon/transformations/series/_exp_smoothing.py b/aeon/transformations/series/_exp_smoothing.py
new file mode 100644
index 00000000..55667694
--- /dev/null
+++ b/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


Test patch:
diff --git a/aeon/transformations/series/tests/test_exp_smoothing.py b/aeon/transformations/series/tests/test_exp_smoothing.py
new file mode 100644
index 00000000..cb83a4e7
--- /dev/null
+++ b/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)