Source code for evalml.pipelines.components.estimators.regressors.arima_regressor

"""Autoregressive Integrated Moving Average Model. The three parameters (p, d, q) are the AR order, the degree of differencing, and the MA order. More information here:"""
from typing import Dict, Hashable, List, Optional, Tuple, Union

import numpy as np
import pandas as pd
from pandas.api.types import is_integer_dtype
from import Integer
from sktime.forecasting.base import ForecastingHorizon

from evalml.model_family import ModelFamily
from evalml.pipelines.components.estimators import Estimator
from evalml.pipelines.components.utils import convert_bool_to_double, match_indices
from evalml.problem_types import ProblemTypes
from evalml.utils import (

[docs]class ARIMARegressor(Estimator): """Autoregressive Integrated Moving Average Model. The three parameters (p, d, q) are the AR order, the degree of differencing, and the MA order. More information here: Currently ARIMARegressor isn't supported via conda install. It's recommended that it be installed via PyPI. Args: time_index (str): Specifies the name of the column in X that provides the datetime objects. Defaults to None. trend (str): Controls the deterministic trend. Options are ['n', 'c', 't', 'ct'] where 'c' is a constant term, 't' indicates a linear trend, and 'ct' is both. Can also be an iterable when defining a polynomial, such as [1, 1, 0, 1]. start_p (int): Minimum Autoregressive order. Defaults to 2. d (int): Minimum Differencing degree. Defaults to 0. start_q (int): Minimum Moving Average order. Defaults to 2. max_p (int): Maximum Autoregressive order. Defaults to 5. max_d (int): Maximum Differencing degree. Defaults to 2. max_q (int): Maximum Moving Average order. Defaults to 5. seasonal (boolean): Whether to fit a seasonal model to ARIMA. Defaults to True. sp (int or str): Period for seasonal differencing, specifically the number of periods in each season. If "detect", this model will automatically detect this parameter (given the time series is a standard frequency) and will fall back to 1 (no seasonality) if it cannot be detected. Defaults to 1. n_jobs (int or None): Non-negative integer describing level of parallelism used for pipelines. Defaults to -1. random_seed (int): Seed for the random number generator. Defaults to 0. """ name = "ARIMA Regressor" hyperparameter_ranges = { "start_p": Integer(1, 3), "d": Integer(0, 2), "start_q": Integer(1, 3), "max_p": Integer(3, 10), "max_d": Integer(2, 5), "max_q": Integer(3, 10), "seasonal": [True, False], } """{ "start_p": Integer(1, 3), "d": Integer(0, 2), "start_q": Integer(1, 3), "max_p": Integer(3, 10), "max_d": Integer(2, 5), "max_q": Integer(3, 10), "seasonal": [True, False], }""" model_family = ModelFamily.ARIMA """ModelFamily.ARIMA""" supported_problem_types = [ProblemTypes.TIME_SERIES_REGRESSION] """[ProblemTypes.TIME_SERIES_REGRESSION]""" max_rows = 1000 max_cols = 7 def __init__( self, time_index: Optional[Hashable] = None, trend: Optional[str] = None, start_p: int = 2, d: int = 0, start_q: int = 2, max_p: int = 5, max_d: int = 2, max_q: int = 5, seasonal: bool = True, sp: int = 1, n_jobs: int = -1, random_seed: Union[int, float] = 0, maxiter: int = 10, use_covariates: bool = True, **kwargs, ): self.preds_95_upper = None self.preds_95_lower = None parameters = { "trend": trend, "start_p": start_p, "d": d, "start_q": start_q, "max_p": max_p, "max_d": max_d, "max_q": max_q, "seasonal": seasonal, "maxiter": maxiter, "n_jobs": n_jobs, } parameters.update(kwargs) arima_model_msg = ( "sktime is not installed. Please install using `pip install sktime.`" ) sktime_arima = import_or_raise( "sktime.forecasting.arima", error_msg=arima_model_msg, ) arima_model = sktime_arima.AutoARIMA(**parameters) parameters["use_covariates"] = use_covariates parameters["time_index"] = time_index self.sp = sp self.use_covariates = use_covariates super().__init__( parameters=parameters, component_obj=arima_model, random_seed=random_seed, ) def _remove_datetime( self, data: pd.DataFrame, features: bool = False, ) -> pd.DataFrame: if data is None: return None data_no_dt = data.ww.copy() if isinstance( data_no_dt.index, (pd.DatetimeIndex, pd.PeriodIndex, pd.IntervalIndex), ): data_no_dt = data_no_dt.ww.reset_index(drop=True) if features: data_no_dt =["Datetime"]) return data_no_dt def _set_forecast(self, X: pd.DataFrame): # we can only calculate the difference if the indices are of the same type units_diff = 1 if isinstance(X.index[0], type(self.last_X_index)) and isinstance( X.index, pd.DatetimeIndex, ): dates_diff = pd.date_range( start=self.last_X_index, end=X.index[0], freq=X.index.freq, ) units_diff = len(dates_diff) - 1 elif is_integer_dtype(type(X.index[0])) and is_integer_dtype( type(self.last_X_index), ): units_diff = X.index[0] - self.last_X_index fh_ = ForecastingHorizon( [units_diff + i for i in range(len(X))], is_relative=True, ) return fh_ def _get_sp(self, X: pd.DataFrame) -> int: if X is None: return 1 freq_mappings = { "D": 7, "M": 12, "Q": 4, } time_index = self._parameters.get("time_index", None) sp = self.sp if sp == "detect": inferred_freqs = X.ww.infer_temporal_frequencies() freq = inferred_freqs.get(time_index, None) sp = 1 if freq is not None: sp = freq_mappings.get(freq[:1], 1) return sp
[docs] def fit(self, X: pd.DataFrame, y: Optional[pd.Series] = None): """Fits ARIMA regressor to data. Args: X (pd.DataFrame): The input training data of shape [n_samples, n_features]. y (pd.Series): The target training data of length [n_samples]. Returns: self Raises: ValueError: If y was not passed in. """ X, y = self._manage_woodwork(X, y) if X is not None: X = X.ww.fillna(X.mean()) if y is None: raise ValueError("ARIMA Regressor requires y as input.") sp = self._get_sp(X) self._component_obj.sp = sp self.last_X_index = X.index[-1] if X is not None else y.index[-1] X = self._remove_datetime(X, features=True) if X is not None: X = convert_bool_to_double(X) y = self._remove_datetime(y) X, y = match_indices(X, y) if X is not None and not X.empty and self.use_covariates:, X=X) else: return self
def _manage_types_and_forecast(self, X: pd.DataFrame) -> tuple: fh_ = self._set_forecast(X) X =["Datetime"]) X = convert_bool_to_double(X) return X, fh_ @staticmethod def _parse_prediction_intervals( y_pred_intervals: pd.DataFrame, conf_int: float, ) -> Tuple[pd.Series, pd.Series]: coverage_name = y_pred_intervals.columns[0][0] preds_lower = y_pred_intervals.loc(axis=1)[(coverage_name, conf_int, "lower")] preds_upper = y_pred_intervals.loc(axis=1)[(coverage_name, conf_int, "upper")] = None = None return preds_lower, preds_upper
[docs] def predict(self, X: pd.DataFrame, y: Optional[pd.Series] = None) -> pd.Series: """Make predictions using fitted ARIMA regressor. Args: X (pd.DataFrame): Data of shape [n_samples, n_features]. y (pd.Series): Target data. Returns: pd.Series: Predicted values. Raises: ValueError: If X was passed to `fit` but not passed in `predict`. """ X, y = self._manage_woodwork(X, y) X, fh_ = self._manage_types_and_forecast(X=X) if not X.empty and self.use_covariates: if fh_[0] != 1: # pmdarima (which sktime uses under the hood) only forecasts off the training data # but sktime circumvents this by predicting everything from the end of training data to the date / periods requested # and only returning the values for dates / periods given to sktime. Because of this, # pmdarima requires the number of covariate rows to equal the length of the total number of periods (X.shape[0] == fh_[-1]) if covariates are used. # We circument this by adding arbitrary rows to the start of X since sktime discards these values when predicting. num_rows_diff = fh_[-1] - X.shape[0] filler = pd.DataFrame( columns=X.columns, index=range(num_rows_diff), ).fillna(0) X_ = pd.concat([filler, X], ignore_index=True) X_.ww.init(schema=X.ww.schema) else: X_ = X y_pred_intervals = self._component_obj.predict_interval( fh=fh_, X=X_, coverage=[0.95], ) else: y_pred_intervals = self._component_obj.predict_interval( fh=fh_, coverage=[0.95], ) y_pred_intervals.index = X.index ( self.preds_95_lower, self.preds_95_upper, ) = ARIMARegressor._parse_prediction_intervals(y_pred_intervals, 0.95) y_pred = pd.concat((self.preds_95_lower, self.preds_95_upper), axis=1).mean( axis=1, ) return infer_feature_types(y_pred)
[docs] def get_prediction_intervals( self, X: pd.DataFrame, y: pd.Series = None, coverage: List[float] = None, predictions: pd.Series = None, ) -> Dict[str, pd.Series]: """Find the prediction intervals using the fitted ARIMARegressor. Args: X (pd.DataFrame): Data of shape [n_samples, n_features]. y (pd.Series): Target data. Optional. coverage (list[float]): A list of floats between the values 0 and 1 that the upper and lower bounds of the prediction interval should be calculated for. predictions (pd.Series): Not used for ARIMA regressor. Returns: dict: Prediction intervals, keys are in the format {coverage}_lower or {coverage}_upper. """ if coverage is None: coverage = [0.95] X, y = self._manage_woodwork(X, y) X, fh_ = self._manage_types_and_forecast(X=X) prediction_interval_result = {} if not X.empty and self.use_covariates: y_pred_intervals = self._component_obj.predict_interval( fh=fh_, X=X, coverage=coverage, ) else: y_pred_intervals = self._component_obj.predict_interval( fh=fh_, coverage=coverage, ) y_pred_intervals.index = X.index for conf_int in coverage: if ( conf_int == 0.95 and self.preds_95_lower is not None and self.preds_95_upper is not None ): prediction_interval_result[f"{conf_int}_lower"] = self.preds_95_lower prediction_interval_result[f"{conf_int}_upper"] = self.preds_95_upper continue preds_lower, preds_upper = ARIMARegressor._parse_prediction_intervals( y_pred_intervals, conf_int, ) prediction_interval_result[f"{conf_int}_lower"] = preds_lower prediction_interval_result[f"{conf_int}_upper"] = preds_upper return prediction_interval_result
@property def feature_importance(self) -> np.ndarray: """Returns array of 0's with a length of 1 as feature_importance is not defined for ARIMA regressor.""" return np.zeros(1)