|
| 1 | +"""ARIMA(p,d,q) / SARIMAX wrapper with automatic order selection. |
| 2 | +
|
| 3 | +Wraps ``statsmodels.tsa.statespace.SARIMAX`` to provide a StatsPAI-style |
| 4 | +interface: formula-like specification, ``.summary()``, ``.forecast()``, |
| 5 | +``.plot()``, and automatic (p,d,q) selection via AICc grid search. |
| 6 | +
|
| 7 | +Examples |
| 8 | +-------- |
| 9 | +>>> import statspai as sp |
| 10 | +>>> result = sp.arima(df["gdp"], order=(1, 1, 1)) |
| 11 | +>>> fc = result.forecast(horizon=12) |
| 12 | +>>> result.plot() |
| 13 | +""" |
| 14 | +from __future__ import annotations |
| 15 | + |
| 16 | +from dataclasses import dataclass |
| 17 | +from typing import Optional, Tuple |
| 18 | + |
| 19 | +import numpy as np |
| 20 | +import pandas as pd |
| 21 | + |
| 22 | + |
| 23 | +@dataclass |
| 24 | +class ARIMAResult: |
| 25 | + order: Tuple[int, int, int] |
| 26 | + seasonal_order: Optional[Tuple[int, int, int, int]] |
| 27 | + params: pd.Series |
| 28 | + aic: float |
| 29 | + bic: float |
| 30 | + aicc: float |
| 31 | + log_likelihood: float |
| 32 | + residuals: np.ndarray |
| 33 | + fitted_values: np.ndarray |
| 34 | + n: int |
| 35 | + _model: object # statsmodels result (opaque) |
| 36 | + |
| 37 | + def forecast(self, horizon: int = 10, alpha: float = 0.05) -> pd.DataFrame: |
| 38 | + fc = self._model.get_forecast(steps=horizon) |
| 39 | + pred = np.asarray(fc.predicted_mean).ravel() |
| 40 | + ci = fc.conf_int(alpha=alpha) |
| 41 | + ci = np.asarray(ci) |
| 42 | + return pd.DataFrame({ |
| 43 | + "forecast": pred, |
| 44 | + "lower": ci[:, 0] if ci.ndim == 2 else ci, |
| 45 | + "upper": ci[:, 1] if ci.ndim == 2 else ci, |
| 46 | + }) |
| 47 | + |
| 48 | + def plot(self, horizon: int = 20, alpha: float = 0.05, ax=None): |
| 49 | + import matplotlib.pyplot as plt |
| 50 | + if ax is None: |
| 51 | + _, ax = plt.subplots(figsize=(10, 4)) |
| 52 | + T = np.arange(self.n) |
| 53 | + ax.plot(T, self.fitted_values, color="C0", linewidth=0.8, label="fitted") |
| 54 | + ax.plot(T, self.fitted_values + self.residuals, ".", color="grey", |
| 55 | + markersize=2, alpha=0.5, label="observed") |
| 56 | + fc = self.forecast(horizon, alpha) |
| 57 | + T_fc = np.arange(self.n, self.n + horizon) |
| 58 | + ax.plot(T_fc, fc["forecast"], "-", color="C3", label="forecast") |
| 59 | + ax.fill_between(T_fc, fc["lower"], fc["upper"], color="C3", alpha=0.2) |
| 60 | + ax.legend(); ax.set_xlabel("t"); ax.set_ylabel("y") |
| 61 | + return ax |
| 62 | + |
| 63 | + def summary(self) -> str: |
| 64 | + lines = [ |
| 65 | + f"ARIMA{self.order}" |
| 66 | + + (f" x {self.seasonal_order}" if self.seasonal_order else ""), |
| 67 | + "-" * 40, |
| 68 | + f"n : {self.n}", |
| 69 | + f"AIC : {self.aic:.2f}", |
| 70 | + f"BIC : {self.bic:.2f}", |
| 71 | + f"AICc : {self.aicc:.2f}", |
| 72 | + f"Log-Lik : {self.log_likelihood:.2f}", |
| 73 | + "", |
| 74 | + "Parameters:", |
| 75 | + ] |
| 76 | + for nm, val in self.params.items(): |
| 77 | + lines.append(f" {nm:<15s} {val: .4f}") |
| 78 | + return "\n".join(lines) |
| 79 | + |
| 80 | + def __repr__(self) -> str: |
| 81 | + return self.summary() |
| 82 | + |
| 83 | + |
| 84 | +def arima( |
| 85 | + y, |
| 86 | + order: Tuple[int, int, int] = (1, 0, 0), |
| 87 | + seasonal_order: Optional[Tuple[int, int, int, int]] = None, |
| 88 | + exog=None, |
| 89 | + auto: bool = False, |
| 90 | + max_p: int = 5, |
| 91 | + max_q: int = 5, |
| 92 | + max_d: int = 2, |
| 93 | +) -> ARIMAResult: |
| 94 | + """Fit ARIMA(p,d,q) or SARIMAX. |
| 95 | +
|
| 96 | + Parameters |
| 97 | + ---------- |
| 98 | + y : array-like or pd.Series |
| 99 | + order : (p, d, q) |
| 100 | + seasonal_order : (P, D, Q, s), optional |
| 101 | + exog : array-like, optional |
| 102 | + Exogenous regressors (ARIMAX). |
| 103 | + auto : bool, default False |
| 104 | + If True, select (p, d, q) by AICc grid search (ignores ``order``). |
| 105 | + max_p, max_q, max_d : int |
| 106 | + Bounds for the auto search. |
| 107 | + """ |
| 108 | + try: |
| 109 | + from statsmodels.tsa.statespace.sarimax import SARIMAX |
| 110 | + except ImportError as e: |
| 111 | + raise ImportError( |
| 112 | + "statsmodels is required for arima(). " |
| 113 | + "Install with `pip install statsmodels`." |
| 114 | + ) from e |
| 115 | + |
| 116 | + y = np.asarray(y, dtype=float).ravel() |
| 117 | + n = len(y) |
| 118 | + |
| 119 | + if auto: |
| 120 | + best_aicc = np.inf |
| 121 | + best_order = order |
| 122 | + for d in range(max_d + 1): |
| 123 | + for p in range(max_p + 1): |
| 124 | + for q in range(max_q + 1): |
| 125 | + if p == 0 and q == 0: |
| 126 | + continue |
| 127 | + try: |
| 128 | + m = SARIMAX(y, order=(p, d, q), |
| 129 | + seasonal_order=seasonal_order or (0, 0, 0, 0), |
| 130 | + exog=exog, enforce_stationarity=False, |
| 131 | + enforce_invertibility=False) |
| 132 | + res = m.fit(disp=False, maxiter=50) |
| 133 | + k = p + q + 1 + d |
| 134 | + aicc = res.aic + 2 * k * (k + 1) / max(n - k - 1, 1) |
| 135 | + if aicc < best_aicc: |
| 136 | + best_aicc = aicc |
| 137 | + best_order = (p, d, q) |
| 138 | + except Exception: |
| 139 | + continue |
| 140 | + order = best_order |
| 141 | + |
| 142 | + model = SARIMAX(y, order=order, |
| 143 | + seasonal_order=seasonal_order or (0, 0, 0, 0), |
| 144 | + exog=exog, enforce_stationarity=False, |
| 145 | + enforce_invertibility=False) |
| 146 | + res = model.fit(disp=False) |
| 147 | + |
| 148 | + k = sum(order) + 1 |
| 149 | + aicc = res.aic + 2 * k * (k + 1) / max(n - k - 1, 1) |
| 150 | + |
| 151 | + return ARIMAResult( |
| 152 | + order=order, |
| 153 | + seasonal_order=seasonal_order, |
| 154 | + params=pd.Series(res.params, index=res.param_names) if hasattr(res, "param_names") else pd.Series(res.params), |
| 155 | + aic=float(res.aic), |
| 156 | + bic=float(res.bic), |
| 157 | + aicc=float(aicc), |
| 158 | + log_likelihood=float(res.llf), |
| 159 | + residuals=np.asarray(res.resid), |
| 160 | + fitted_values=np.asarray(res.fittedvalues), |
| 161 | + n=n, |
| 162 | + _model=res, |
| 163 | + ) |
0 commit comments