Spaces:
Running
Running
File size: 4,588 Bytes
9831ced a9e6751 9831ced a9e6751 9831ced a9e6751 9831ced a9e6751 9831ced a9e6751 9831ced | 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 | 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, quiet_fit
from features.feature_pipeline import validate_features
class ArimaModel(BaseForecastModel):
name = 'ARIMA'
def __init__(self, max_p=3, max_d=2, max_q=3, order=None, refit_order_every=None):
self.max_p = max_p
self.max_d = max_d
self.max_q = max_q
self._cached_order = order
self.refit_order_every = refit_order_every
self._calls_since_search = 0
self._lock = threading.Lock()
self.last_diagnostics = {}
def _best_order(self, series: pd.Series, exog: pd.DataFrame=None):
arima = import_arima(self.name)
best_aic = np.inf
best_order = None
failures = []
# AIC values from different integration orders have different likelihood
# bases. Select d on the training sample, then compare only p/q at that d.
from models.auto_arima_model import AutoArimaModel
selected_d = AutoArimaModel(max_d=self.max_d)._find_d(series)
with quiet_fit():
candidates = [(p, d, q)
for d in [selected_d]
for p in range(self.max_p + 1)
for q in range(self.max_q + 1)
if p + q <= 2]
for order in candidates:
try:
fit = fit_arima(arima(series, exog=exog, order=order))
except Exception as error:
failures.append({'order': order, 'error': f'{type(error).__name__}: {error}'})
continue
if fit.aic < best_aic:
best_aic = fit.aic
best_order = order
if best_order is None:
self.last_diagnostics = {'converged': False, 'candidate_errors': failures}
raise ValueError(f'No ARIMA candidate converged on this training window: {failures}')
return best_order
def _fit_forecast(self, arima, series, order, horizon, exog, future_exog):
"""Execute the selected ARIMA order; a failed fit is a failed evaluation."""
with quiet_fit():
fit = fit_arima(arima(series, exog=exog, order=order))
forecast = fit.forecast(steps=horizon, exog=future_exog) if exog is not None else fit.forecast(steps=horizon)
values = np.exp(np.asarray(forecast, dtype=float))
if values.shape != (horizon,) or not np.isfinite(values).all() or (values <= 0).any():
raise ValueError(f'{self.name} returned an invalid forecast for selected order {order}.')
self.last_diagnostics = {'converged': True, 'selected_order': order, 'executed_order': order,
'order_fallback': False, 'feature_policy': 'lagged one candle; unknown future covariates persist'}
return values
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('ARIMA needs at least 20 candles of history.')
arima = import_arima(self.name)
exog = future_exog = None
if features is not None:
validate_features(features, expected_len=len(series), model_name='ARIMA')
exog, future_exog = causal_exogenous(features, horizon, self.name)
if exog is not None:
series = series.iloc[1:].reset_index(drop=True)
if (series <= 0).any():
raise ValueError('ARIMA needs strictly positive prices for its log-price model.')
log_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:
self._cached_order = self._best_order(log_series, exog=exog)
self._calls_since_search = 0
self._calls_since_search += 1
forecast = self._fit_forecast(arima, log_series, self._cached_order, horizon, exog, future_exog)
return [float(x) for x in forecast]
|