Spaces:
Running
Running
| 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', True): | |
| 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.') | |
| return fit | |
| 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 | |