-
Notifications
You must be signed in to change notification settings - Fork 89
/
theta.py
131 lines (102 loc) · 4.52 KB
/
theta.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
"""Implements Theta-lines transformation for use with Theta forecasting."""
__author__ = ["GuzalBulatova", "mloning"]
__all__ = ["ThetaLinesTransformer"]
import numpy as np
import pandas as pd
from aeon.forecasting.base import ForecastingHorizon
from aeon.forecasting.trend import PolynomialTrendForecaster
from aeon.transformations.base import BaseTransformer
class ThetaLinesTransformer(BaseTransformer):
"""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.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 = {
"input_data_type": "Series",
# what is the abstract type of X: Series, or Panel
"output_data_type": "Series",
# what abstract type is returned: Primitives, Series, Panel
"instancewise": True, # is this an instance-wise transform?
"X_inner_type": ["pd.DataFrame", "pd.Series"],
"y_inner_type": "None",
"transform-returns-same-time-index": True,
"univariate-only": True,
"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.Series or pd.DataFrame
Data to be transformed
y : ignored argument for interface compatibility
Additional data, e.g., labels for transformation
Returns
-------
theta_lines: pd.Series or pd.DataFrame
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
"""
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