[ENH] Convert Theta transform to series transformer

aeon/transformations/series/__init__.py

@@ -3,10 +3,11 @@
 __all__ = [
     "AutoCorrelationSeriesTransformer",
     "BaseSeriesTransformer",
+    "Dobin",
     "MatrixProfileSeriesTransformer",
     "StatsModelsACF",
     "StatsModelsPACF",
-    "Dobin",
+    "ThetaTransformer",
 ]
 
 from aeon.transformations.series._acf import (
@@ -16,4 +17,5 @@
 )
 from aeon.transformations.series._dobin import Dobin
 from aeon.transformations.series._matrix_profile import MatrixProfileSeriesTransformer
+from aeon.transformations.series._theta import ThetaTransformer
 from aeon.transformations.series.base import BaseSeriesTransformer

aeon/transformations/series/_theta.py

@@ -0,0 +1,123 @@
+__maintainer__ = []
+__all__ = ["ThetaTransformer"]
+
+import numpy as np
+import pandas as pd
+
+from aeon.forecasting.base import ForecastingHorizon
+from aeon.forecasting.trend import PolynomialTrendForecaster
+from aeon.transformations.series.base import BaseSeriesTransformer
+
+
+class ThetaTransformer(BaseSeriesTransformer):
+    """Decompose the original data into two or more Theta-lines.
+
+    Implementation of decomposition for Theta-method [1]_ as described in [2]_.
+
+    Overview: Input :term:`univariate series <Univariate time series>` of length
+    "n" and ThetaLinesTransformer modifies the local curvature of the time series
+    using Theta-coefficient values passed through the parameter `theta`.
+
+    Each Theta-coefficient is applied directly to the second differences of the input
+    series. The resulting transformed series (Theta-lines) are returned as a
+    pd.DataFrame of shape `len(input series) * len(theta)`.
+
+    Parameters
+    ----------
+    theta : sequence of float, default=(0,2)
+        Theta-coefficients to use in transformation.
+
+    Notes
+    -----
+    Depending on the value of the Theta-coefficient, Theta-lines either augment the
+    long-term trend (0 < Theta < 1) or the the short-term behaviour (Theta > 1).
+
+    Special cases:
+        - Theta == 0 : deflates input data to linear trend
+        - Theta == 1 : returns data unchanged
+        - Theta < 0 : transforms time series and mirrors it along the linear trend.
+
+    References
+    ----------
+    .. [1] V.Assimakopoulos et al., "The theta model: a decomposition approach
+       to forecasting", International Journal of Forecasting, vol. 16, pp. 521-530,
+       2000.
+    .. [2] E.Spiliotis et al., "Generalizing the Theta method for
+       automatic forecasting ", European Journal of Operational
+       Research, vol. 284, pp. 550-558, 2020.
+
+    Examples
+    --------
+    >>> from aeon.transformations.series._theta import ThetaTransformer
+    >>> from aeon.datasets import load_airline
+    >>> y = load_airline()
+    >>> transformer = ThetaTransformer([0, 0.25, 0.5, 0.75])
+    >>> y_thetas = transformer.fit_transform(y)
+    """
+
+    _tags = {
+        "X_inner_type": ["pd.DataFrame"],
+        "capability:multivariate": False,
+        "fit_is_empty": True,
+    }
+
+    def __init__(self, theta=(0, 2)):
+        self.theta = theta
+        super().__init__()
+
+    def _transform(self, X, y=None):
+        """Transform X and return a transformed version.
+
+        private _transform containing the core logic, called from transform
+
+        Parameters
+        ----------
+        X : pd.DataFrame
+            Data to be transformed
+        y : ignored argument for interface compatibility
+            Additional data, e.g., labels for transformation
+
+        Returns
+        -------
+        theta_lines: pd.Series
+            Transformed series
+            pd.Series, with single Theta-line, if self.theta is float
+            pd.DataFrame of shape: [len(X), len(self.theta)], if self.theta is tuple
+        """
+        X = X.squeeze()
+        theta = _check_theta(self.theta)
+
+        forecaster = PolynomialTrendForecaster()
+        forecaster.fit(y=X)
+        fh = ForecastingHorizon(X.index, is_relative=False)
+        trend = forecaster.predict(fh=fh)
+
+        theta_lines = np.zeros((X.shape[0], len(theta)))
+        for i, theta_i in enumerate(theta):
+            theta_lines[:, i] = _theta_transform(X, trend, theta_i)
+        if isinstance(self.theta, (float, int)):
+            return pd.Series(theta_lines.flatten(), index=X.index)
+        else:
+            return pd.DataFrame(theta_lines, columns=self.theta, index=X.index)
+
+
+def _theta_transform(Z, trend, theta):
+    # obtain one Theta-line
+    theta_line = Z * theta + (1 - theta) * trend
+    theta_line = theta_line.values.flatten()
+    return theta_line
+
+
+def _check_theta(theta):
+    valid_theta_types = (list, int, float, tuple)
+
+    if not isinstance(theta, valid_theta_types):
+        raise ValueError(f"invalid input, please use one of {valid_theta_types}")
+
+    if isinstance(theta, (int, float)):
+        theta = [theta]
+
+    if isinstance(theta, tuple):
+        theta = list(theta)
+
+    return theta

aeon/transformations/tests/test_theta.py → aeon/transformations/series/tests/test_theta.py

@@ -8,13 +8,13 @@
 from scipy.stats import linregress
 
 from aeon.datasets import load_airline
-from aeon.transformations.theta import ThetaLinesTransformer
+from aeon.transformations.series._theta import ThetaTransformer
 
 
 def test_theta_0():
     # with theta = 0
     y = load_airline()
-    t = ThetaLinesTransformer(0)
+    t = ThetaTransformer(0)
     t.fit(y)
     actual = t.transform(y)
     x = np.arange(y.size) + 1
@@ -27,16 +27,16 @@ def test_theta_0():
 def test_theta_1():
     # with theta = 1 Theta-line is equal to the original time-series
     y = load_airline()
-    t = ThetaLinesTransformer(1)
+    t = ThetaTransformer(1)
     t.fit(y)
     actual = t.transform(y)
     np.testing.assert_array_equal(actual, y)
 
 
 @pytest.mark.parametrize("theta", [(1, 1.5), (0, 1, 2), (0.25, 0.5, 0.75, 1, 2)])
-def test_thetalines_shape(theta):
+def test_theta_shape(theta):
     y = load_airline()
-    t = ThetaLinesTransformer(theta)
+    t = ThetaTransformer(theta)
     t.fit(y)
     actual = t.transform(y)
     assert actual.shape == (y.shape[0], len(theta))

aeon/transformations/theta.py

@@ -5,12 +5,20 @@
 
 import numpy as np
 import pandas as pd
+from deprecated.sphinx import deprecated
 
 from aeon.forecasting.base import ForecastingHorizon
 from aeon.forecasting.trend import PolynomialTrendForecaster
 from aeon.transformations.base import BaseTransformer
 
 
+# TODO: remove in v0.10.0
+@deprecated(
+    version="0.9.0",
+    reason="ThetaLinesTransformer will be removed in version 0.10 and replaced with a "
+    "BaseSeriesTransformer version in the transformations.series module.",
+    category=FutureWarning,
+)
 class ThetaLinesTransformer(BaseTransformer):
     """Decompose the original data into two or more Theta-lines.
 
@@ -47,14 +55,6 @@ class ThetaLinesTransformer(BaseTransformer):
     .. [2] E.Spiliotis et al., "Generalizing the Theta method for
        automatic forecasting ", European Journal of Operational
        Research, vol. 284, pp. 550-558, 2020.
-
-    Examples
-    --------
-    >>> from aeon.transformations.theta import ThetaLinesTransformer
-    >>> from aeon.datasets import load_airline
-    >>> y = load_airline()
-    >>> transformer = ThetaLinesTransformer([0, 0.25, 0.5, 0.75])
-    >>> y_thetas = transformer.fit_transform(y)
     """
 
     _tags = {