| """ |
| AutoARIMA forecaster, via Nixtla's statsforecast (actively maintained, |
| confirmed compatible with this project's pandas 3.x / numpy 2.x — the |
| alternative, pmdarima, was not tried after statsforecast installed and |
| ran cleanly on the first attempt; no need to chase a second option once |
| one is verified working). |
| |
| Unlike Moirai-MoE and Chronos-2 (zero-shot pretrained foundation |
| models with a real load() step and a downloaded checkpoint), AutoARIMA |
| is a classical statistical method: it FITS a fresh ARIMA(p,d,q) model |
| to each context window at predict()-time. That is the normal, correct |
| way this kind of model works, not a shortcut or a simplification of |
| "real" loading — there is no pretrained checkpoint for it to load. |
| |
| API verified by REAL execution in this sandbox (network access to |
| PyPI, no real market data needed — synthetic data exercises the same |
| code path): `pip install statsforecast`, fit on a |
| (unique_id, ds, y)-columned frame, `sf.predict(h=..., level=[50, 80])` |
| returns point forecasts plus 50%/80% prediction intervals, which map |
| exactly onto this project's five quantile levels (0.1/0.25/0.5/0.75/0.9) |
| since a symmetric interval at level L brackets the (100-L)/2 and |
| 100-(100-L)/2 percentiles: level=80 -> P10/P90, level=50 -> P25/P75. |
| See tests/test_arima_model.py, which runs this for real (not stubbed) |
| since statsforecast has no heavy/network dependency to work around. |
| """ |
| from __future__ import annotations |
|
|
| import numpy as np |
| import pandas as pd |
|
|
| from forecast_types import ModelInferenceError, ModelInputError, ModelLoadError, QuantileForecast |
|
|
| MODEL_NAME = "AutoARIMA (statsforecast)" |
|
|
|
|
| class AutoArimaForecaster: |
| """No real load() step (see module docstring) — kept for interface |
| parity with MoiraiMoEForecaster/Chronos2Forecaster so app.py can |
| treat all three uniformly.""" |
|
|
| def __init__(self, season_length: int = 1, min_context: int = 10): |
| self.season_length = season_length |
| self.min_context = min_context |
| self._loaded = True |
|
|
| def load(self): |
| pass |
|
|
| def predict(self, context_df: pd.DataFrame, target_col: str, past_covariate_cols, |
| prediction_length: int, freq: str) -> QuantileForecast: |
| closes = context_df[target_col] |
| if closes.isna().any() or not np.isfinite(closes.to_numpy()).all(): |
| raise ModelInputError("context contains NaN/inf values in the target column") |
| if len(closes) < self.min_context: |
| raise ModelInputError( |
| f"context too short for AutoARIMA: {len(closes)} row(s), need at least {self.min_context}" |
| ) |
|
|
| try: |
| from statsforecast import StatsForecast |
| from statsforecast.models import AutoARIMA |
| except ImportError as e: |
| raise ModelLoadError(f"statsforecast not installed: {e}") from e |
|
|
| fit_df = pd.DataFrame({"unique_id": "series", "ds": context_df.index, "y": closes.to_numpy()}) |
| try: |
| sf = StatsForecast(models=[AutoARIMA(season_length=self.season_length)], freq=freq, n_jobs=1) |
| sf.fit(fit_df) |
| fc = sf.predict(h=prediction_length, level=[50, 80]) |
| except Exception as e: |
| raise ModelInferenceError(f"AutoARIMA fit/predict failed: {e}") from e |
|
|
| |
| |
| |
| quantiles = np.stack([ |
| fc["AutoARIMA-lo-80"].to_numpy(), |
| fc["AutoARIMA-lo-50"].to_numpy(), |
| fc["AutoARIMA"].to_numpy(), |
| fc["AutoARIMA-hi-50"].to_numpy(), |
| fc["AutoARIMA-hi-80"].to_numpy(), |
| ], axis=1) |
|
|
| return QuantileForecast( |
| horizon=prediction_length, quantile_levels=[0.1, 0.25, 0.5, 0.75, 0.9], |
| quantiles=quantiles, samples=None, |
| context_length_used=len(context_df), prediction_length=prediction_length, |
| ) |
|
|
