diff --git a/aeon/forecasting/stats/__init__.py b/aeon/forecasting/stats/__init__.py index a86d5346a0..bb4b54cde3 100644 --- a/aeon/forecasting/stats/__init__.py +++ b/aeon/forecasting/stats/__init__.py @@ -1,6 +1,7 @@ """Stats based forecasters.""" __all__ = [ + "AutoETS", "ARIMA", "AutoARIMA", "AutoTAR", @@ -11,7 +12,7 @@ ] from aeon.forecasting.stats._arima import ARIMA, AutoARIMA -from aeon.forecasting.stats._ets import ETS +from aeon.forecasting.stats._ets import ETS, AutoETS from aeon.forecasting.stats._tar import TAR, AutoTAR from aeon.forecasting.stats._theta import Theta from aeon.forecasting.stats._tvp import TVP diff --git a/aeon/forecasting/stats/_ets.py b/aeon/forecasting/stats/_ets.py index 542ecd40bb..3459fe6624 100644 --- a/aeon/forecasting/stats/_ets.py +++ b/aeon/forecasting/stats/_ets.py @@ -1,4 +1,4 @@ -"""ETS class. +"""ETS and AutoETS class. An implementation of the exponential smoothing statistics forecasting algorithm. Implements additive and multiplicative error models. We recommend using the AutoETS @@ -6,7 +6,7 @@ """ __maintainer__ = [] -__all__ = ["ETS"] +__all__ = ["ETS", "AutoETS"] import numpy as np @@ -20,6 +20,7 @@ _ets_predict_value, ) from aeon.forecasting.utils._nelder_mead import nelder_mead +from aeon.forecasting.utils._seasonality import calc_seasonal_period ADDITIVE = "additive" MULTIPLICATIVE = "multiplicative" @@ -271,6 +272,111 @@ def iterative_forecast(self, y, prediction_horizon): return preds +class AutoETS(BaseForecaster): + """Automatic Exponential Smoothing forecaster. + + An implementation of the exponential smoothing statistics forecasting algorithm. + Chooses betweek additive and multiplicative error models, + None, additive and multiplicative (including damped) trend and + None, additive and multiplicative seasonality[1]_. + + Parameters + ---------- + horizon : int, default = 1 + The horizon to forecast to. + + References + ---------- + .. [1] R. J. Hyndman and G. Athanasopoulos, + Forecasting: Principles and Practice. Melbourne, Australia: OTexts, 2014. + + Examples + -------- + >>> from aeon.forecasting.stats import AutoETS + >>> from aeon.datasets import load_airline + >>> y = load_airline() + >>> forecaster = AutoETS() + >>> forecaster.forecast(y) + 435.9312382780535 + """ + + _tags = { + "capability:horizon": False, + } + + def __init__(self): + self.error_type_ = 0 + self.trend_type_ = 0 + self.seasonality_type_ = 0 + self.seasonal_period_ = 0 + self.wrapped_model_ = None + super().__init__(horizon=1, axis=1) + + def _fit(self, y, exog=None): + """Fit Auto Exponential Smoothing forecaster to series y. + + Fit a forecaster to predict self.horizon steps ahead using y. + + Parameters + ---------- + y : np.ndarray + A time series on which to learn a forecaster to predict horizon ahead + exog : np.ndarray, default =None + Optional exogenous time series data assumed to be aligned with y + + Returns + ------- + self + Fitted AutoETS. + """ + data = y.squeeze() + ( + self.error_type_, + self.trend_type_, + self.seasonality_type_, + self.seasonal_period_, + ) = auto_ets(data) + self.wrapped_model_ = ETS( + self.error_type_, + self.trend_type_, + self.seasonality_type_, + self.seasonal_period_, + ) + self.wrapped_model_.fit(y, exog) + return self + + def _predict(self, y=None, exog=None): + """ + Predict the next horizon steps ahead. + + Parameters + ---------- + y : np.ndarray, default = None + A time series to predict the next horizon value for. If None, + predict the next horizon value after series seen in fit. + exog : np.ndarray, default =None + Optional exogenous time series data assumed to be aligned with y + + Returns + ------- + float + single prediction self.horizon steps ahead of y. + """ + return self.wrapped_model_.predict(y, exog) + + def _forecast(self, y, exog=None, axis=1): + self.fit(y, exog=exog) + return float(self.wrapped_model_.forecast_) + + def iterative_forecast(self, y, prediction_horizon): + """Forecast with ETS specific iterative method. + + Overrides the base class iterative_forecast to avoid refitting on each step. + This simply rolls the ETS model forward + """ + return self.wrapped_model_.iterative_forecast(y, prediction_horizon) + + @njit(fastmath=True, cache=True) def _numba_predict( trend_type, @@ -320,3 +426,43 @@ def _validate_parameter(var, can_be_none): f"variable must be either string or integer with values" f" {valid_str} or {valid_int} but saw {var}" ) + + +def auto_ets(data): + """Calculate model parameters based on the internal nelder-mead implementation.""" + seasonal_period = calc_seasonal_period(data) + lowest_aic = -1 + best_model = None + for error_type in range(1, 3): + for trend_type in range(0, 3): + for seasonality_type in range(0, 2 * (seasonal_period != 1) + 1): + model_seasonal_period = seasonal_period + if seasonal_period < 1 or seasonality_type == 0: + model_seasonal_period = 1 + model = np.array( + [ + error_type, + trend_type, + seasonality_type, + model_seasonal_period, + ], + dtype=np.int32, + ) + try: + (_, aic) = nelder_mead( + 1, + 1 + 2 * (trend_type != 0) + (seasonality_type != 0), + data, + model, + ) + except ZeroDivisionError: + continue + if lowest_aic == -1 or lowest_aic > aic: + lowest_aic = aic + best_model = ( + error_type, + trend_type, + seasonality_type, + model_seasonal_period, + ) + return best_model diff --git a/aeon/forecasting/stats/tests/test_ets.py b/aeon/forecasting/stats/tests/test_ets.py index ab0f078e17..184a12b28e 100644 --- a/aeon/forecasting/stats/tests/test_ets.py +++ b/aeon/forecasting/stats/tests/test_ets.py @@ -3,7 +3,7 @@ import numpy as np import pytest -from aeon.forecasting.stats._ets import ETS, _validate_parameter +from aeon.forecasting.stats._ets import ETS, AutoETS, _validate_parameter @pytest.mark.parametrize( @@ -105,3 +105,115 @@ def test_ets_iterative_forecast(): forecaster = ETS(trend_type=None) forecaster._fit(y) assert forecaster._trend_type == 0 + + +# small seasonal-ish series (same as in ETS tests) +Y_SEASONAL = np.array( + [3, 10, 12, 13, 12, 10, 12, 3, 10, 12, 13, 12, 10, 12], dtype=float +) +# another shortish series for basic sanity checks +Y_SHORT = np.array([10, 12, 14, 13, 15, 16, 18, 19, 20, 21, 22, 23], dtype=float) + + +def test_autoets_fit_sets_attributes_and_wraps(): + """Fit should set type/period attributes and wrap an ETS instance.""" + forecaster = AutoETS() + forecaster.fit(Y_SEASONAL) + + # wrapped model exists and is ETS + assert forecaster.wrapped_model_ is not None + assert isinstance(forecaster.wrapped_model_, ETS) + + # discovered structure attributes should exist and be integers >= 0 + for attr in ("error_type_", "trend_type_", "seasonality_type_", "seasonal_period_"): + val = getattr(forecaster, attr) + assert isinstance(val, (int, np.integer)) + assert val >= 0 + + # wrapped model should have been fitted and expose a finite forecast_ + assert hasattr(forecaster.wrapped_model_, "forecast_") + assert np.isfinite(forecaster.wrapped_model_.forecast_) + + +def test_autoets_predict_returns_finite_float(): + """_predict should return a finite float once fitted.""" + forecaster = AutoETS() + forecaster.fit(Y_SHORT) + pred = forecaster._predict(Y_SHORT) + assert isinstance(pred, float) + assert np.isfinite(pred) + + +def test_autoets_forecast_sets_wrapped_and_returns_forecast_float(): + """_forecast should fit internally, set wrapped forecast_, and return that value.""" + forecaster = AutoETS() + f = forecaster._forecast(Y_SEASONAL) + assert isinstance(f, float) + assert np.isfinite(f) + assert forecaster.wrapped_model_ is not None + assert hasattr(forecaster.wrapped_model_, "forecast_") + assert np.isclose(f, float(forecaster.wrapped_model_.forecast_)) + + +def test_autoets_iterative_forecast_shape_and_validity(): + """iterative_forecast should delegate to wrapped ETS and return valid outputs.""" + h = 5 + forecaster = AutoETS() + forecaster.fit(Y_SHORT) + preds = forecaster.iterative_forecast(Y_SHORT, prediction_horizon=h) + + assert isinstance(preds, np.ndarray) + assert preds.shape == (h,) + assert np.all(np.isfinite(preds)) + + # Optional: first iterative step should match one-step-ahead forecast after fit + assert np.isclose(preds[0], forecaster.wrapped_model_.forecast_, atol=1e-6) + + +def test_autoets_horizon_greater_than_one_raises(): + """ + AutoETS.fit should raise ValueError. + + when horizon > 1 (ETS only supports 1-step fit). + """ + forecaster = AutoETS() + forecaster.horizon = 2 + with pytest.raises(ValueError, match="Horizon is set >1"): + forecaster.fit(Y_SEASONAL) + + +def test_autoets_predict_matches_wrapped_predict(): + """_predict should match the wrapped ETS model's predict.""" + forecaster = AutoETS() + forecaster.fit(Y_SEASONAL) + a = forecaster._predict(Y_SEASONAL) + b = forecaster.wrapped_model_.predict(Y_SEASONAL) + assert isinstance(a, float) and isinstance(b, float) + assert np.isfinite(a) and np.isfinite(b) + assert np.isclose(a, b) + + +def test_autoets_forecast_is_consistent_with_wrapped(): + """_forecast should equal the wrapped model's forecast after internal fit.""" + forecaster = AutoETS() + val = forecaster._forecast(Y_SHORT) + assert np.isclose(val, float(forecaster.wrapped_model_.forecast_)) + + +def test_autoets_exog_raises(): + """AutoETS.fit should raise ValueError when exog passed.""" + forecaster = AutoETS() + exog = np.arange(len(Y_SEASONAL), dtype=float) # simple aligned exogenous regressor + with pytest.raises( + ValueError, + match="AutoETS cannot handle exogenous variables", + ): + forecaster.fit(Y_SEASONAL, exog=exog) + + +def test_autoets_repeatability_on_same_input(): + """Forecasting twice on the same series should be deterministic.""" + forecaster = AutoETS() + f1 = forecaster._forecast(Y_SEASONAL) + f2 = forecaster._forecast(Y_SEASONAL) + assert np.isclose(f1, f2) diff --git a/aeon/forecasting/utils/_seasonality.py b/aeon/forecasting/utils/_seasonality.py index 356b1a40d2..6b1a3411b6 100644 --- a/aeon/forecasting/utils/_seasonality.py +++ b/aeon/forecasting/utils/_seasonality.py @@ -88,7 +88,7 @@ def calc_seasonal_period(data): The estimated seasonal period (lag) of the series. Returns 1 if no significant peak is detected in the autocorrelation. """ - lags = acf(data, 24) + lags = acf(data, min(24, len(data) - 1)) lags = np.concatenate((np.array([1.0]), lags)) peaks = [] mean_lags = np.mean(lags)