Spaces:
Sleeping
Sleeping
| import threading | |
| import numpy as np | |
| import pandas as pd | |
| from models.base_model import BaseForecastModel, causal_exogenous, validate_history_and_horizon | |
| from models.dependencies import fit_arima, import_arima, import_kpss, quiet_fit | |
| from features.feature_pipeline import validate_features | |
| class AutoArimaModel(BaseForecastModel): | |
| name = 'Auto-ARIMA' | |
| def __init__(self, max_p=5, max_q=5, max_d=2, order=None, refit_order_every=None): | |
| self.max_p = max_p | |
| self.max_q = max_q | |
| self.max_d = max_d | |
| self._cached_order = order | |
| self.refit_order_every = refit_order_every | |
| self._calls_since_search = 0 | |
| self._lock = threading.Lock() | |
| self.last_diagnostics = {} | |
| def _find_d(self, series: pd.Series) -> int: | |
| kpss = import_kpss(self.name) | |
| s = series.copy() | |
| d = 0 | |
| if float(s.std()) < 1e-12: | |
| return 0 | |
| with quiet_fit(): | |
| while d < self.max_d: | |
| try: | |
| _, p_value, _, _ = kpss(s, regression='c', nlags='auto') | |
| except Exception as error: | |
| raise ValueError(f'Auto-ARIMA stationarity test failed ({type(error).__name__}): {error}') from error | |
| if p_value >= 0.05: | |
| break | |
| s = s.diff().dropna() | |
| d += 1 | |
| return d | |
| def _aic(self, series: pd.Series, order, exog: pd.DataFrame=None) -> float: | |
| arima = import_arima(self.name) | |
| try: | |
| return fit_arima(arima(series, exog=exog, order=order)).aic | |
| except Exception as error: | |
| self.last_diagnostics.setdefault('candidate_errors', []).append( | |
| {'order': order, 'error': f'{type(error).__name__}: {error}'}) | |
| return np.inf | |
| def _stepwise_search(self, series: pd.Series, d: int, exog: pd.DataFrame=None): | |
| # A compact Hyndman-style candidate set is more stable for noisy | |
| # intraday data than repeatedly walking a large grid at every origin. | |
| candidates = [(0, 0), (1, 0), (0, 1), (1, 1), (2, 0), (0, 2), (2, 1), (1, 2)] | |
| candidates = [(p, q) for p, q in candidates if p <= self.max_p and q <= self.max_q] | |
| best_order, best_aic = (None, np.inf) | |
| with quiet_fit(): | |
| for p, q in candidates: | |
| aic = self._aic(series, (p, d, q), exog=exog) | |
| if aic < best_aic: | |
| best_aic, best_order = (aic, (p, q)) | |
| if best_order is None: | |
| raise ValueError(f'No Auto-ARIMA candidate converged on this training window: {self.last_diagnostics}') | |
| p, q = best_order | |
| return (p, d, q) | |
| def predict(self, history: pd.Series, horizon: int=1, features: pd.DataFrame=None) -> list: | |
| with self._lock: | |
| self.last_diagnostics = {} | |
| validate_history_and_horizon(history, horizon, self.name) | |
| series = pd.Series(history).astype(float).reset_index(drop=True) | |
| validate_history_and_horizon(series, horizon, self.name) | |
| if len(series) < 20: | |
| raise ValueError('Auto-ARIMA needs at least 20 candles of history.') | |
| arima = import_arima(self.name) | |
| if (series <= 0).any(): | |
| raise ValueError('Auto-ARIMA needs strictly positive prices for its log-price model.') | |
| exog = future_exog = None | |
| if features is not None: | |
| validate_features(features, expected_len=len(series), model_name='Auto-ARIMA') | |
| exog, future_exog = causal_exogenous(features, horizon, self.name) | |
| if exog is not None: | |
| series = series.iloc[1:].reset_index(drop=True) | |
| series = np.log(series) | |
| need_search = self._cached_order is None | |
| if self.refit_order_every and self._calls_since_search >= self.refit_order_every: | |
| need_search = True | |
| if need_search: | |
| d = self._find_d(series) | |
| self._cached_order = self._stepwise_search(series, d, exog=exog) | |
| self._calls_since_search = 0 | |
| self._calls_since_search += 1 | |
| from models.arima_model import ArimaModel | |
| forecast = ArimaModel._fit_forecast(self, arima, series, self._cached_order, horizon, exog, future_exog) | |
| return forecast.tolist() | |