Spaces:
Sleeping
Sleeping
File size: 4,150 Bytes
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 90 91 92 93 94 95 96 97 98 | 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
|