Spaces:
Sleeping
Sleeping
File size: 4,370 Bytes
9831ced a9e6751 9831ced a9e6751 9831ced a9e6751 9831ced a9e6751 9831ced a9e6751 | 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 | 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()
|