Spaces:
Running
Running
| 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 | |
| # 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: | |
| continue | |
| if fit.aic < best_aic: | |
| best_aic = fit.aic | |
| best_order = order | |
| if best_order is None: | |
| raise ValueError('No ARIMA candidate converged on this training window.') | |
| return best_order | |
| def _fit_forecast(self, arima, series, order, horizon, exog, future_exog): | |
| """Fit a positive-price ARIMA with deterministic, real fallbacks.""" | |
| candidates = [order, (0, 1, 0), (1, 1, 0), (0, 1, 1), (1, 1, 1)] | |
| seen = set() | |
| last_error = None | |
| for candidate in candidates: | |
| if candidate in seen: | |
| continue | |
| seen.add(candidate) | |
| try: | |
| with quiet_fit(): | |
| fit = fit_arima(arima(series, exog=exog, order=candidate)) | |
| 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,) and np.isfinite(values).all() and (values > 0).all(): | |
| self._cached_order = candidate | |
| self.last_diagnostics = {'converged': True, 'selected_order': order, 'executed_order': candidate, 'order_fallback': candidate != order, 'feature_policy': 'lagged one candle; unknown future covariates persist'} | |
| return values | |
| except Exception as error: | |
| last_error = error | |
| raise ValueError(f'ARIMA could not fit a stable positive-price model: {last_error}') from None | |
| def predict(self, history: pd.Series, horizon: int=1, features: pd.DataFrame=None) -> list: | |
| with self._lock: | |
| validate_history_and_horizon(history, horizon, self.name) | |
| self.last_diagnostics = {} | |
| 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] | |