import contextlib import inspect import warnings def _missing(package: str, model: str, error: Exception) -> ImportError: return ImportError(f"{model} needs the '{package}' package. Install with:\n pip install {package}\n(original error: {error})") def _patch_statsmodels_pandas_compat() -> None: """Bridge the legacy decorator call used by statsmodels 0.14.x.""" try: import pandas.util._decorators as pandas_decorators pandas_original = pandas_decorators.deprecate_kwarg pandas_parameters = list(inspect.signature(pandas_original).parameters.values()) if ( pandas_parameters and pandas_parameters[0].name == 'klass' and not getattr(pandas_original, "_rinking_compat", False) ): def pandas_compatible_deprecate_kwarg(*args, **kwargs): if len(args) >= 2 and all(isinstance(value, str) for value in args[:2]): return pandas_original(FutureWarning, args[0], args[1], *args[2:], **kwargs) return pandas_original(*args, **kwargs) pandas_compatible_deprecate_kwarg._rinking_compat = True pandas_decorators.deprecate_kwarg = pandas_compatible_deprecate_kwarg except ImportError: pass try: import statsmodels.compat.pandas as compat except ImportError: return signature = inspect.signature(compat.deprecate_kwarg) if len(signature.parameters) != 5: return original = compat.deprecate_kwarg if getattr(original, "_rinking_compat", False): return def compatible_deprecate_kwarg(*args, **kwargs): if len(args) >= 2 and all(isinstance(value, str) for value in args[:2]): return original(FutureWarning, args[0], args[1], *args[2:], **kwargs) return original(*args, **kwargs) compatible_deprecate_kwarg._rinking_compat = True compat.deprecate_kwarg = compatible_deprecate_kwarg def import_arima(model_name: str): try: _patch_statsmodels_pandas_compat() from statsmodels.tsa.arima.model import ARIMA except ImportError as e: raise _missing('statsmodels', model_name, e) from e return ARIMA def import_kpss(model_name: str): try: _patch_statsmodels_pandas_compat() from statsmodels.tsa.stattools import kpss except ImportError as e: raise _missing('statsmodels', model_name, e) from e return kpss def import_arch_model(model_name: str): try: _patch_statsmodels_pandas_compat() from arch import arch_model except ImportError as e: raise _missing('arch', model_name, e) from e return arch_model def fit_arima(model): """Fit statsmodels ARIMA with a bounded optimizer budget. Walk-forward evaluation can fit hundreds of models. An unbounded BFGS call makes a free Space look hung while adding little value to a short intraday window; the selected order is still re-estimated from each training window. """ fit = model.fit(method_kwargs={'maxiter': 120, 'disp': False}) if not getattr(fit, 'mle_retvals', {}).get('converged', False): raise ValueError('ARIMA optimizer did not converge within the bounded iteration budget.') if not __import__('numpy').isfinite(fit.params).all(): raise ValueError('ARIMA fit produced non-finite parameters.') if not __import__('numpy').isfinite(fit.aic): raise ValueError('ARIMA fit produced non-finite AIC.') return fit @contextlib.contextmanager def quiet_fit(): with warnings.catch_warnings(): # Expected candidate-selection warnings only. Convergence is checked by # fit_arima; unrelated warnings are not blanket-suppressed. warnings.filterwarnings('ignore', message='Non-stationary starting autoregressive parameters.*') warnings.filterwarnings('ignore', message='Non-invertible starting MA parameters.*') warnings.filterwarnings('ignore', message='Maximum Likelihood optimization failed to converge.*') warnings.filterwarnings('ignore', message='The test statistic is outside of the range.*') yield