ForecasterRnn

skforecast.deep_learning._forecaster_rnn.ForecasterRnn

ForecasterRnn(
    regressor,
    levels,
    lags="auto",
    steps="auto",
    transformer_series=MinMaxScaler(feature_range=(0, 1)),
    weight_func=None,
    fit_kwargs={},
    forecaster_id=None,
    n_jobs=None,
    transformer_exog=None,
)

Bases: ForecasterBase

This class turns any regressor compatible with the Keras API into a Keras RNN multi-serie multi-step forecaster. A unique model is created to forecast all time steps and series. Keras enables workflows on top of either JAX, TensorFlow, or PyTorch. See documentation for more details.

Parameters:

Name	Type	Description	Default
regressor	regressor or pipeline compatible with the Keras API	An instance of a regressor or pipeline compatible with the Keras API.	required
levels	(str, list)	Name of one or more time series to be predicted. This determine the series the forecaster will be handling. If None, all series used during training will be available for prediction.	required
lags	(int, list, str)	Lags used as predictors. If 'auto', lags used are from 1 to N, where N is extracted from the input layer self.regressor.layers[0].input_shape[0][1].	`'auto'`
transformer_series	(object, dict)	An instance of a transformer (preprocessor) compatible with the scikit-learn preprocessing API with methods: fit, transform, fit_transform and inverse_transform. Transformation is applied to each series before training the forecaster. ColumnTransformers are not allowed since they do not have inverse_transform method. If single transformer: it is cloned and applied to all series. If dict of transformers: a different transformer can be used for each series.	`sklearn.preprocessing.MinMaxScaler(feature_range=(0, 1))`
fit_kwargs	dict	Additional arguments to be passed to the fit method of the regressor.	`None`
forecaster_id	(str, int)	Name used as an identifier of the forecaster.	`None`
steps	(int, list, str)	Steps to be predicted. If 'auto', steps used are from 1 to N, where N is extracted from the output layer self.regressor.layers[-1].output_shape[1].	`'auto'`
lags	Optional[Union[int, list, str]]	Not used, present here for API consistency by convention.	'auto'
transformer_exog	Ignored	Not used, present here for API consistency by convention.	None
weight_func	Ignored	Not used, present here for API consistency by convention.	None
n_jobs	Ignored	Not used, present here for API consistency by convention.	None

Attributes:

Name	Type	Description
regressor	regressor or pipeline compatible with the Keras API	An instance of a regressor or pipeline compatible with the Keras API. An instance of this regressor is trained for each step. All of them are stored in self.regressors_.
levels	(str, list)	Name of one or more time series to be predicted. This determine the series the forecaster will be handling. If None, all series used during training will be available for prediction.
steps	numpy ndarray	Number of future steps the forecaster will predict when using method predict(). Since a different model is created for each step, this value should be defined before training.
lags	numpy ndarray	Lags used as predictors.
transformer_series	(object, dict)	An instance of a transformer (preprocessor) compatible with the scikit-learn preprocessing API with methods: fit, transform, fit_transform and inverse_transform. Transformation is applied to each series before training the forecaster. ColumnTransformers are not allowed since they do not have inverse_transform method.
transformer_series_	dict	Dictionary with the transformer for each series. It is created cloning the objects in transformer_series and is used internally to avoid overwriting.
transformer_exog	Ignored	Not used, present here for API consistency by convention.
max_lag	int	Maximum lag included in lags.
window_size	int	Size of the window needed to create the predictors.
last_window_	pandas Series	Last window seen by the forecaster during training. It stores the values needed to predict the next step immediately after the training data.
index_type_	type	Type of index of the input used in training.
index_freq_	str	Frequency of Index of the input used in training.
training_range_	pandas Index	First and last values of index of the data used during training.
series_names_in_	list	Names of the series used during training.
exog_in_	bool	If the forecaster has been trained using exogenous variable/s.
exog_names_in_	list	Names of the exogenous variables used during training.
exog_type_in_	type	Type of exogenous variable/s used in training.
exog_dtypes_in_	dict	Type of each exogenous variable/s used in training. If transformer_exog is used, the dtypes are calculated after the transformation.
X_train_dim_names_	dict	Labels for the multi-dimensional arrays created internally for training.
y_train_dim_names_	dict	Labels for the multi-dimensional arrays created internally for training.
fit_kwargs	dict	Additional arguments to be passed to the fit method of the regressor.
in_sample_residuals_	dict	Residuals of the models when predicting training data. Only stored up to 1000 values per model in the form {step: residuals}. If transformer_series is not None, residuals are stored in the transformed scale.
out_sample_residuals	dict	Residuals of the models when predicting non training data. Only stored up to 1000 values per model in the form {step: residuals}. If transformer_series is not None, residuals are assumed to be in the transformed scale. Use set_out_sample_residuals() method to set values.
creation_date	str	Date of creation.
is_fitted	bool	Tag to identify if the regressor has been fitted (trained).
fit_date	str	Date of last fit.
skforcast_version	str	Version of skforecast library used to create the forecaster.
python_version	str	Version of python used to create the forecaster.
forecaster_id	(str, int)	Name used as an identifier of the forecaster.
history	dict	Dictionary with the history of the training of each step. It is created internally to avoid overwriting.
_probabilistic_mode	(str, bool)	Private attribute used to indicate whether the forecaster should perform some calculations during backtesting.
dropna_from_series	Ignored	Not used, present here for API consistency by convention.
encoding	Ignored	Not used, present here for API consistency by convention.
differentiation	Ignored	Not used, present here for API consistency by convention.
differentiation_max	Ignored	Not used, present here for API consistency by convention.
differentiator	Ignored	Not used, present here for API consistency by convention.
differentiator_	Ignored	Not used, present here for API consistency by convention.

Source code in skforecast\deep_learning\_forecaster_rnn.py

def __init__(
    self,
    regressor: object,
    levels: Union[str, list],
    lags: Optional[Union[int, list, str]] = "auto",
    steps: Optional[Union[int, list, str]] = "auto",
    transformer_series: Optional[Union[object, dict]] = MinMaxScaler(
        feature_range=(0, 1)
    ),
    weight_func: Optional[Callable] = None,
    fit_kwargs: Optional[dict] = {},
    forecaster_id: Optional[Union[str, int]] = None,
    n_jobs: Any = None,
    transformer_exog: Any = None,
) -> None:
    self.levels = None
    self.transformer_series = transformer_series
    self.transformer_series_ = None
    self.transformer_exog = None
    self.weight_func = weight_func
    self.source_code_weight_func = None
    self.max_lag = None
    self.window_size = None
    self.last_window_ = None
    self.index_type_ = None
    self.index_freq_ = None
    self.training_range_ = None
    self.exog_in_ = False
    self.exog_type_in_ = None
    self.exog_dtypes_in_ = None
    self.exog_names_in_ = None
    self.series_names_in_ = None
    self.X_train_dim_names_ = None
    self.y_train_dim_names_ = None
    self.is_fitted = False
    self.creation_date = pd.Timestamp.today().strftime("%Y-%m-%d %H:%M:%S")
    self.fit_date = None
    self.skforecast_version = skforecast.__version__
    self.python_version = sys.version.split(" ")[0]
    self.forecaster_id = forecaster_id
    self._probabilistic_mode = "no_binned"
    self.history = None  # TODO: Change to history_ as come from fit method?
    self.dropna_from_series = False  # Ignored in this forecaster
    self.encoding = None   # Ignored in this forecaster
    self.differentiation = None   # Ignored in this forecaster
    self.differentiation_max = None   # Ignored in this forecaster
    self.differentiator = None   # Ignored in this forecaster
    self.differentiator_ = None   # Ignored in this forecaster

    # Infer parameters from the model
    self.regressor = regressor  # TODO: Create copy of regressor copy(regressor)
    layer_init = self.regressor.layers[0]

    if lags == "auto":
        if keras.__version__ < "3.0":
            self.lags = np.arange(layer_init.input_shape[0][1]) + 1
        else:
            self.lags = np.arange(layer_init.output.shape[1]) + 1

        warnings.warn(
            "Setting `lags` = 'auto'. `lags` are inferred from the regressor "
            "architecture. Avoid the warning with lags=lags."
        )
    elif isinstance(lags, int):
        self.lags = np.arange(lags) + 1
    elif isinstance(lags, list):
        self.lags = np.array(lags)
    else:
        raise TypeError(
            f"`lags` argument must be an int, list or 'auto'. Got {type(lags)}."
        )

    self.max_lag = np.max(self.lags)
    self.window_size = self.max_lag

    layer_end = self.regressor.layers[-1]

    try:
        if keras.__version__ < "3.0":
            self.series = layer_end.output_shape[-1]
        else:
            self.series = layer_end.output.shape[-1]
    # if does not work, break the and raise an error the input shape should
    # be shape=(lags, n_series))
    except:
        raise TypeError(
            "Input shape of the regressor should be Input(shape=(lags, n_series))."
        )

    if steps == "auto":
        if keras.__version__ < "3.0":
            self.steps = np.arange(layer_end.output_shape[1]) + 1
        else:
            self.steps = np.arange(layer_end.output.shape[1]) + 1
        warnings.warn(
            "`steps` default value = 'auto'. `steps` inferred from regressor "
            "architecture. Avoid the warning with steps=steps."
        )
    elif isinstance(steps, int):
        self.steps = np.arange(steps) + 1
    elif isinstance(steps, list):
        self.steps = np.array(steps)
    else:
        raise TypeError(
            f"`steps` argument must be an int, list or 'auto'. Got {type(steps)}."
        )

    self.max_step = np.max(self.steps)
    if keras.__version__ < "3.0":
        self.outputs = layer_end.output_shape[-1]
    else:
        self.outputs = layer_end.output.shape[-1]

    if not isinstance(levels, (list, str, type(None))):
        raise TypeError(
            f"`levels` argument must be a string, list or. Got {type(levels)}."
        )

    if isinstance(levels, str):
        self.levels = [levels]
    elif isinstance(levels, list):
        self.levels = levels
    else:
        raise TypeError(
            f"`levels` argument must be a string or a list. Got {type(levels)}."
        )

    self.in_sample_residuals_ = {step: None for step in self.steps}
    self.out_sample_residuals_ = None

    self.series_val = fit_kwargs.pop("series_val", None)
    self.fit_kwargs = check_select_fit_kwargs(
        regressor=self.regressor, fit_kwargs=fit_kwargs
    )

_create_lags

_create_lags(y)

Transforms a 1d array into a 3d array (X) and a 3d array (y). Each row in X is associated with a value of y and it represents the lags that precede it.

Notice that, the returned matrix X_data, contains the lag 1 in the first column, the lag 2 in the second column and so on.

Parameters:

Name	Type	Description	Default
y	numpy ndarray	1d numpy ndarray Training time series.	required

Returns:

Name	Type	Description
X_data	numpy ndarray	3d numpy ndarray with the lagged values (predictors). Shape: (samples - max(lags), len(lags))
y_data	numpy ndarray	3d numpy ndarray with the values of the time series related to each row of X_data for each step. Shape: (len(max_step), samples - max(lags))

Source code in skforecast\deep_learning\_forecaster_rnn.py

def _create_lags(self, y: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
    """
    Transforms a 1d array into a 3d array (X) and a 3d array (y). Each row
    in X is associated with a value of y and it represents the lags that
    precede it.

    Notice that, the returned matrix X_data, contains the lag 1 in the first
    column, the lag 2 in the second column and so on.

    Parameters
    ----------
    y : numpy ndarray
        1d numpy ndarray Training time series.

    Returns
    -------
    X_data : numpy ndarray
        3d numpy ndarray with the lagged values (predictors).
        Shape: (samples - max(lags), len(lags))
    y_data : numpy ndarray
        3d numpy ndarray with the values of the time series related to each
        row of `X_data` for each step.
        Shape: (len(max_step), samples - max(lags))

    """

    n_splits = len(y) - self.max_lag - self.max_step + 1  # rows of y_data
    if n_splits <= 0:
        raise ValueError(
            (
                f"The maximum lag ({self.max_lag}) must be less than the length "
                f"of the series minus the maximum of steps ({len(y) - self.max_step})."
            )
        )

    X_data = np.full(
        shape=(n_splits, (self.max_lag)), fill_value=np.nan, order="F", dtype=float
    )
    for i, lag in enumerate(range(self.max_lag - 1, -1, -1)):
        X_data[:, i] = y[self.max_lag - lag - 1 : -(lag + self.max_step)]

    y_data = np.full(
        shape=(n_splits, self.max_step), fill_value=np.nan, order="F", dtype=float
    )
    for step in range(self.max_step):
        y_data[:, step] = y[self.max_lag + step : self.max_lag + step + n_splits]

    # Get lags index
    X_data = X_data[:, self.lags - 1]

    # Get steps index
    y_data = y_data[:, self.steps - 1]

    return X_data, y_data

create_train_X_y

create_train_X_y(series, exog=None)

Create training matrices. The resulting multi-dimensional matrices contain the target variable and predictors needed to train the model.

Parameters:

Name	Type	Description	Default
series	pandas DataFrame	Training time series.	required
exog	Ignored	Not used, present here for API consistency by convention. This type of forecaster does not allow exogenous variables.	None

Returns:

Name	Type	Description
X_train	ndarray	Training values (predictors) for each step. The resulting array has 3 dimensions: (time_points, n_lags, n_series)
y_train	ndarray	Values (target) of the time series related to each row of X_train. The resulting array has 3 dimensions: (time_points, n_steps, n_levels)
dimension_names	dict	Labels for the multi-dimensional arrays created internally for training.

Source code in skforecast\deep_learning\_forecaster_rnn.py

def create_train_X_y(
    self, series: pd.DataFrame, exog: Any = None
) -> Tuple[np.ndarray, np.ndarray, dict]:
    """
    Create training matrices. The resulting multi-dimensional matrices contain
    the target variable and predictors needed to train the model.

    Parameters
    ----------
    series : pandas DataFrame
        Training time series.
    exog : Ignored
        Not used, present here for API consistency by convention. This type of
        forecaster does not allow exogenous variables.

    Returns
    -------
    X_train : np.ndarray
        Training values (predictors) for each step. The resulting array has
        3 dimensions: (time_points, n_lags, n_series)
    y_train : np.ndarray
        Values (target) of the time series related to each row of `X_train`.
        The resulting array has 3 dimensions: (time_points, n_steps, n_levels)
    dimension_names : dict
        Labels for the multi-dimensional arrays created internally for training.

    """

    if not isinstance(series, pd.DataFrame):
        raise TypeError(f"`series` must be a pandas DataFrame. Got {type(series)}.")

    series_names_in_ = list(series.columns)

    if not set(self.levels).issubset(set(series.columns)):
        raise ValueError(
            (
                f"`levels` defined when initializing the forecaster must be included "
                f"in `series` used for trainng. {set(self.levels) - set(series.columns)} "
                f"not found."
            )
        )

    if len(series) < self.max_lag + self.max_step:
        raise ValueError(
            (
                f"Minimum length of `series` for training this forecaster is "
                f"{self.max_lag + self.max_step}. Got {len(series)}. Reduce the "
                f"number of predicted steps, {self.max_step}, or the maximum "
                f"lag, {self.max_lag}, if no more data is available."
            )
        )

    if self.transformer_series is None:
        self.transformer_series_ = {serie: None for serie in series_names_in_}
    elif not isinstance(self.transformer_series, dict):
        self.transformer_series_ = {
            serie: clone(self.transformer_series) for serie in series_names_in_
        }
    else:
        self.transformer_series_ = {serie: None for serie in series_names_in_}
        # Only elements already present in transformer_series_ are updated
        self.transformer_series_.update(
            (k, v)
            for k, v in deepcopy(self.transformer_series).items()
            if k in self.transformer_series_
        )
        series_not_in_transformer_series = set(series.columns) - set(
            self.transformer_series.keys()
        )
        if series_not_in_transformer_series:
            warnings.warn(
                (
                    f"{series_not_in_transformer_series} not present in "
                    f"`transformer_series`. No transformation is applied to "
                    f"these series."
                ),
                IgnoredArgumentWarning,
            )

    # Step 1: Create lags for all columns
    X_train = []
    y_train = []

    for i, serie in enumerate(series.columns):
        x = series[serie]
        check_y(y=x)
        x = transform_series(
            series=x,
            transformer=self.transformer_series_[serie],
            fit=True,
            inverse_transform=False,
        )
        X, _ = self._create_lags(x)
        X_train.append(X)

    for i, serie in enumerate(self.levels):
        y = series[serie]
        check_y(y=y)
        y = transform_series(
            series=y,
            transformer=self.transformer_series_[serie],
            fit=True,
            inverse_transform=False,
        )

        _, y = self._create_lags(y)
        y_train.append(y)

    X_train = np.stack(X_train, axis=2)
    y_train = np.stack(y_train, axis=2)

    train_index = series.index.to_list()[
        self.max_lag : (len(series.index.to_list()) - self.max_step + 1)
    ]
    dimension_names = {
        "X_train": {
            0: train_index,
            1: ["lag_" + str(l) for l in self.lags],
            2: series.columns.to_list(),
        },
        "y_train": {
            0: train_index,
            1: ["step_" + str(l) for l in self.steps],
            2: self.levels,
        },
    }

    return X_train, y_train, dimension_names

fit

fit(
    series,
    store_in_sample_residuals=True,
    exog=None,
    suppress_warnings=False,
    store_last_window="Ignored",
)

Training Forecaster.

Additional arguments to be passed to the fit method of the regressor can be added with the fit_kwargs argument when initializing the forecaster.

Parameters:

Name	Type	Description	Default
series	pandas DataFrame	Training time series.	required
store_in_sample_residuals	bool	If True, in-sample residuals will be stored in the forecaster object after fitting (in_sample_residuals_ attribute).	`True`
exog	Ignored	Not used, present here for API consistency by convention.	None
suppress_warnings	bool	If True, skforecast warnings will be suppressed during the prediction process. See skforecast.exceptions.warn_skforecast_categories for more information.	`False`
store_last_window	Ignored	Not used, present here for API consistency by convention.	'Ignored'

Returns:

Type	Description
None

Source code in skforecast\deep_learning\_forecaster_rnn.py

def fit(
    self,
    series: pd.DataFrame,
    store_in_sample_residuals: bool = True,
    exog: Any = None,
    suppress_warnings: bool = False,
    store_last_window: str = "Ignored",
) -> None:
    """
    Training Forecaster.

    Additional arguments to be passed to the `fit` method of the regressor
    can be added with the `fit_kwargs` argument when initializing the forecaster.

    Parameters
    ----------
    series : pandas DataFrame
        Training time series.
    store_in_sample_residuals : bool, default `True`
        If `True`, in-sample residuals will be stored in the forecaster object
        after fitting (`in_sample_residuals_` attribute).
    exog : Ignored
        Not used, present here for API consistency by convention.
    suppress_warnings : bool, default `False`
        If `True`, skforecast warnings will be suppressed during the prediction
        process. See skforecast.exceptions.warn_skforecast_categories for more
        information.
    store_last_window : Ignored
        Not used, present here for API consistency by convention.

    Returns
    -------
    None

    """

    set_skforecast_warnings(suppress_warnings, action="ignore")

    # Reset values in case the forecaster has already been fitted.
    self.index_type_ = None
    self.index_freq_ = None
    self.last_window_ = None
    self.exog_in_ = None
    self.exog_type_in_ = None
    self.exog_dtypes_in_ = None
    self.exog_names_in_ = None
    self.series_names_in_ = None
    self.X_train_dim_names_ = None
    self.y_train_dim_names_ = None
    self.in_sample_residuals_ = None
    self.is_fitted = False
    self.training_range_ = None

    self.series_names_in_ = list(series.columns)

    X_train, y_train, X_train_dim_names_ = self.create_train_X_y(series=series)
    self.X_train_dim_names_ = X_train_dim_names_["X_train"]
    self.y_train_dim_names_ = X_train_dim_names_["y_train"]
    if keras.__version__ > "3.0" and keras.backend.backend() == "torch":
        import torch

        device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        torch_device = torch.device(device)

        print(f"Using device: {device}")
        X_train = torch.tensor(X_train).to(torch_device)
        y_train = torch.tensor(y_train).to(torch_device)

    if self.series_val is not None:
        X_val, y_val, _ = self.create_train_X_y(series=self.series_val)
        if keras.__version__ > "3.0" and keras.backend.backend() == "torch":
            X_val = torch.tensor(X_val).to(torch_device)
            y_val = torch.tensor(y_val).to(torch_device)
            history = self.regressor.fit(
                x=X_train,
                y=y_train,
                validation_data=(X_val, y_val),
                **self.fit_kwargs,
            )
        else:
            history = self.regressor.fit(
                x=X_train,
                y=y_train,
                validation_data=(X_val, y_val),
                **self.fit_kwargs,
            )
    else:
        history = self.regressor.fit(
            x=X_train,
            y=y_train,
            **self.fit_kwargs,
        )

    self.history = history.history
    self.is_fitted = True
    self.fit_date = pd.Timestamp.today().strftime("%Y-%m-%d %H:%M:%S")
    _, y_index = preprocess_y(y=series[self.levels], return_values=False)
    self.training_range_ = y_index[[0, -1]]
    self.index_type_ = type(y_index)
    if isinstance(y_index, pd.DatetimeIndex):
        self.index_freq_ = y_index.freqstr
    else:
        self.index_freq_ = y_index.step

    self.last_window_ = series.iloc[-self.max_lag :].copy()

    set_skforecast_warnings(suppress_warnings, action="default")

    if store_in_sample_residuals:
        residuals = y_train - self.regressor.predict(x=X_train, verbose=0)
        self.in_sample_residuals_ = {step: residuals[:, i, :] for i, step in enumerate(self.steps)}

predict

predict(
    steps=None,
    levels=None,
    last_window=None,
    exog=None,
    suppress_warnings=False,
    check_inputs=None,
)

Predict n steps ahead

Parameters:

Name	Type	Description	Default
steps	(int, list, None)	Predict n steps. The value of steps must be less than or equal to the value of steps defined when initializing the forecaster. Starts at 1. If int: Only steps within the range of 1 to int are predicted. If list: List of ints. Only the steps contained in the list are predicted. If None: As many steps are predicted as were defined at initialization.	`None`
levels	(str, list)	Name of one or more time series to be predicted. It must be included in levels defined when initializing the forecaster. If None, all all series used during training will be available for prediction.	`None`
last_window	pandas DataFrame	Series values used to create the predictors (lags) needed in the first iteration of the prediction (t + 1). If last_window = None, the values stored in self.last_window_ are used to calculate the initial predictors, and the predictions start right after training data.	`None`
exog	Ignored	Not used, present here for API consistency by convention.	None
suppress_warnings	bool	If True, skforecast warnings will be suppressed during the fitting process. See skforecast.exceptions.warn_skforecast_categories for more information.	`False`
check_inputs	Ignored	Not used, present here for API consistency by convention.	None

Returns:

Name	Type	Description
predictions	pandas DataFrame	Predicted values.

Source code in skforecast\deep_learning\_forecaster_rnn.py

def predict(
    self,
    steps: Optional[Union[int, list]] = None,
    levels: Optional[Union[str, list]] = None,
    last_window: Optional[pd.DataFrame] = None,
    exog: Any = None,
    suppress_warnings: bool = False,
    check_inputs: Any = None
) -> pd.DataFrame:
    """
    Predict n steps ahead

    Parameters
    ----------
    steps : int, list, None, default `None`
        Predict n steps. The value of `steps` must be less than or equal to the
        value of steps defined when initializing the forecaster. Starts at 1.

        - If `int`: Only steps within the range of 1 to int are predicted.
        - If `list`: List of ints. Only the steps contained in the list
        are predicted.
        - If `None`: As many steps are predicted as were defined at
        initialization.
    levels : str, list, default `None`
        Name of one or more time series to be predicted. It must be included
        in `levels` defined when initializing the forecaster. If `None`, all
        all series used during training will be available for prediction.
    last_window : pandas DataFrame, default `None`
        Series values used to create the predictors (lags) needed in the
        first iteration of the prediction (t + 1).
        If `last_window = None`, the values stored in `self.last_window_` are
        used to calculate the initial predictors, and the predictions start
        right after training data.
    exog : Ignored
        Not used, present here for API consistency by convention.
    suppress_warnings : bool, default `False`
        If `True`, skforecast warnings will be suppressed during the fitting
        process. See skforecast.exceptions.warn_skforecast_categories for more
        information.
    check_inputs : Ignored
        Not used, present here for API consistency by convention.

    Returns
    -------
    predictions : pandas DataFrame
        Predicted values.

    """

    set_skforecast_warnings(suppress_warnings, action="ignore")

    if levels is None:
        levels = self.levels
    elif isinstance(levels, str):
        levels = [levels]
    if isinstance(steps, int):
        steps = list(np.arange(steps) + 1)
    elif steps is None:
        if isinstance(self.steps, int):
            steps = list(np.arange(self.steps) + 1)
        elif isinstance(self.steps, (list, np.ndarray)):
            steps = list(np.array(self.steps))
    elif isinstance(steps, list):
        steps = list(np.array(steps))

    for step in steps:
        if not isinstance(step, (int, np.int64, np.int32)):
            raise TypeError(
                (
                    f"`steps` argument must be an int, a list of ints or `None`. "
                    f"Got {type(steps)}."
                )
            )

    if last_window is None:
        last_window = self.last_window_

    check_predict_input(
        forecaster_name=type(self).__name__,
        steps=steps,
        is_fitted=self.is_fitted,
        exog_in_=self.exog_in_,
        index_type_=self.index_type_,
        index_freq_=self.index_freq_,
        window_size=self.window_size,
        last_window=last_window,
        exog=None,
        exog_type_in_=None,
        exog_names_in_=None,
        interval=None,
        max_steps=self.max_step,
        levels=levels,
        levels_forecaster=self.levels,
        series_names_in_=self.series_names_in_,
    )

    last_window = last_window.iloc[-self.window_size :,].copy()

    for serie_name in self.series_names_in_:
        last_window_serie = transform_series(
            series=last_window[serie_name],
            transformer=self.transformer_series_[serie_name],
            fit=False,
            inverse_transform=False,
        )
        last_window_values, last_window_index = preprocess_last_window(
            last_window=last_window_serie
        )
        last_window.loc[:, serie_name] = last_window_values

    X = np.reshape(last_window.to_numpy(), (1, self.max_lag, last_window.shape[1]))
    predictions = self.regressor.predict(X, verbose=0)
    predictions_reshaped = np.reshape(
        predictions, (predictions.shape[1], predictions.shape[2])
    )

    # if len(self.levels) == 1:
    #     predictions_reshaped = np.reshape(predictions, (predictions.shape[1], 1))
    # else:
    #     predictions_reshaped = np.reshape(
    #         predictions, (predictions.shape[1], predictions.shape[2])
    #     )
    idx = expand_index(index=last_window_index, steps=max(steps))

    predictions = pd.DataFrame(
        data=predictions_reshaped[np.array(steps) - 1],
        columns=self.levels,
        index=idx[np.array(steps) - 1],
    )
    predictions = predictions[levels]

    for serie in levels:
        x = predictions[serie]
        check_y(y=x)
        x = transform_series(
            series=x,
            transformer=self.transformer_series_[serie],
            fit=False,
            inverse_transform=True,
        )
        predictions.loc[:, serie] = x

    # Temporal standardization. Pending full refactoring
    predictions = (
        predictions.melt(var_name="level", value_name="pred", ignore_index=False)
        .reset_index()
        .sort_values(by=["index", "level"])
        .set_index("index")
        .rename_axis(None, axis=0)
    )

    set_skforecast_warnings(suppress_warnings, action="default")

    return predictions

predict_bootstrapping

predict_bootstrapping(
    steps=None,
    last_window=None,
    exog=None,
    n_boot=250,
    random_state=123,
    use_in_sample_residuals=True,
    suppress_warnings=False,
    levels=None,
)

Generate multiple forecasting predictions using a bootstrapping process. By sampling from a collection of past observed errors (the residuals), each iteration of bootstrapping generates a different set of predictions. Only levels whose last window ends at the same datetime index can be predicted together. See the Notes section for more information.

Parameters:

Name	Type	Description	Default
steps	int	Number of steps to predict.	None
levels	(str, list)	Time series to be predicted. If None all levels whose last window ends at the same datetime index will be predicted together.	None
last_window	pandas DataFrame	Series values used to create the predictors (lags) needed in the first iteration of the prediction (t + 1). If last_window = None, the values stored in self.last_window_ are used to calculate the initial predictors, and the predictions start right after training data.	None
exog	pandas Series, pandas DataFrame, dict	Exogenous variable/s included as predictor/s.	None
n_boot	int	Number of bootstrapping iterations used to estimate predictions.	250
random_state	int	Sets a seed to the random generator, so that boot predictions are always deterministic.	123
use_in_sample_residuals	bool	If True, residuals from the training data are used as proxy of prediction error to create predictions. If False, out of sample residuals are used. In the latter case, the user should have calculated and stored the residuals within the forecaster (see set_out_sample_residuals()).	True
suppress_warnings	bool	If True, skforecast warnings will be suppressed during the prediction process. See skforecast.exceptions.warn_skforecast_categories for more information.	False

Returns:

Name	Type	Description
boot_predictions	dict	Predictions generated by bootstrapping for each level.

Notes

More information about prediction intervals in forecasting: https://otexts.com/fpp3/prediction-intervals.html#prediction-intervals-from-bootstrapped-residuals Forecasting: Principles and Practice (3^rd ed) Rob J Hyndman and George Athanasopoulos.

Source code in skforecast\deep_learning\_forecaster_rnn.py

def predict_bootstrapping(
        self,
        steps: Optional[Union[int, list]] = None,
        last_window: Optional[pd.DataFrame] = None,
        exog: Optional[Union[pd.Series, pd.DataFrame]] = None,
        n_boot: int = 250,
        random_state: int = 123,
        use_in_sample_residuals: bool = True,
        suppress_warnings: bool = False,
        levels: Any = None
) -> dict:
    """
    Generate multiple forecasting predictions using a bootstrapping process.
    By sampling from a collection of past observed errors (the residuals),
    each iteration of bootstrapping generates a different set of predictions.
    Only levels whose last window ends at the same datetime index can be
    predicted together. See the Notes section for more information.

    Parameters
    ----------
    steps : int
        Number of steps to predict. 
    levels : str, list, default None
        Time series to be predicted. If `None` all levels whose last window
        ends at the same datetime index will be predicted together.
    last_window : pandas DataFrame, default None
        Series values used to create the predictors (lags) needed in the
        first iteration of the prediction (t + 1).
        If `last_window = None`, the values stored in `self.last_window_` are
        used to calculate the initial predictors, and the predictions start
        right after training data.
    exog : pandas Series, pandas DataFrame, dict, default None
        Exogenous variable/s included as predictor/s.
    n_boot : int, default 250
        Number of bootstrapping iterations used to estimate predictions.
    random_state : int, default 123
        Sets a seed to the random generator, so that boot predictions are always
        deterministic.
    use_in_sample_residuals : bool, default True
        If `True`, residuals from the training data are used as proxy of
        prediction error to create predictions. If `False`, out of sample
        residuals are used. In the latter case, the user should have
        calculated and stored the residuals within the forecaster (see
        `set_out_sample_residuals()`).
    suppress_warnings : bool, default False
        If `True`, skforecast warnings will be suppressed during the prediction
        process. See skforecast.exceptions.warn_skforecast_categories for more
        information.

    Returns
    -------
    boot_predictions : dict
        Predictions generated by bootstrapping for each level.
        {level: pandas DataFrame, shape (steps, n_boot)}

    Notes
    -----
    More information about prediction intervals in forecasting:
    https://otexts.com/fpp3/prediction-intervals.html#prediction-intervals-from-bootstrapped-residuals
    Forecasting: Principles and Practice (3rd ed) Rob J Hyndman and George Athanasopoulos.

    """

    set_skforecast_warnings(suppress_warnings, action="ignore")

    if levels is None:
        levels = self.levels
    elif isinstance(levels, str):
        levels = [levels]

    if isinstance(steps, int):
        steps = list(np.arange(steps) + 1)
    elif steps is None:
        if isinstance(self.steps, int):
            steps = list(np.arange(self.steps) + 1)
        elif isinstance(self.steps, (list, np.ndarray)):
            steps = list(np.array(self.steps))
    elif isinstance(steps, list):
        steps = list(np.array(steps))

    if use_in_sample_residuals:
        if not set(steps).issubset(set(self.in_sample_residuals_.keys())):
            raise ValueError(
                f"Not `forecaster.in_sample_residuals_` for steps: "
                f"{set(steps) - set(self.in_sample_residuals_.keys())}."
            )
        residuals = self.in_sample_residuals_
    else:
        if self.out_sample_residuals_ is None:
            raise ValueError(
                "`forecaster.out_sample_residuals_` is `None`. Use "
                "`use_in_sample_residuals=True` or the "
                "`set_out_sample_residuals()` method before predicting."
            )
        else:
            if not set(steps).issubset(set(self.out_sample_residuals_.keys())):
                raise ValueError(
                    f"Not `forecaster.out_sample_residuals_` for steps: "
                    f"{set(steps) - set(self.out_sample_residuals_.keys())}. "
                    f"Use method `set_out_sample_residuals()`."
                )
        residuals = self.out_sample_residuals_

    check_residuals = (
        'forecaster.in_sample_residuals_' if use_in_sample_residuals
        else 'forecaster.out_sample_residuals_'
    )
    for step in steps:
        if residuals[step] is None:
            raise ValueError(
                f"forecaster residuals for step {step} are `None`. "
                f"Check {check_residuals}."
            )
        elif (any(element is None for element in residuals[step]) or
              np.any(np.isnan(residuals[step]))):
            raise ValueError(
                f"forecaster residuals for step {step} contains `None` "
                f"or `NaNs` values. Check {check_residuals}."
            )

    if last_window is None:
        last_window = self.last_window_

    check_predict_input(
        forecaster_name=type(self).__name__,
        steps=steps,
        is_fitted=self.is_fitted,
        exog_in_=self.exog_in_,
        index_type_=self.index_type_,
        index_freq_=self.index_freq_,
        window_size=self.window_size,
        last_window=last_window,
        exog=None,
        exog_type_in_=None,
        exog_names_in_=None,
        interval=None,
        max_steps=self.max_step,
        levels=levels,
        levels_forecaster=self.levels,
        series_names_in_=self.series_names_in_,
    )

    last_window = last_window.iloc[-self.window_size:, ].copy()

    for serie_name in self.series_names_in_:
        last_window_serie = transform_series(
            series=last_window[serie_name],
            transformer=self.transformer_series_[serie_name],
            fit=False,
            inverse_transform=False,
        )
        last_window_values, last_window_index = preprocess_last_window(
            last_window=last_window_serie
        )
        last_window.loc[:, serie_name] = last_window_values

    X = np.reshape(last_window.to_numpy(), (1, self.max_lag, last_window.shape[1]))
    prediction_index = expand_index(index=last_window_index, steps=max(steps))

    predictions = self.regressor.predict(X, verbose=0)
    predictions = np.squeeze(predictions, axis=0)

    boot_predictions = {}
    boot_columns = [f"pred_boot_{i}" for i in range(n_boot)]
    rng = np.random.default_rng(seed=random_state)

    for j, level in enumerate(levels):
        boot_level = np.tile(predictions[:, j], (n_boot, 1)).T

        for i, step in enumerate(steps):
            sampled_residuals = residuals[step][rng.integers(low=0, high=len(residuals[step]), size=n_boot), j]
            boot_level[i, :] += sampled_residuals

        if self.transformer_series_[level]:
            boot_level = np.apply_along_axis(
                func1d=transform_numpy,
                axis=0,
                arr=boot_level,
                transformer=self.transformer_series_[level],
                fit=False,
                inverse_transform=True
            )

        boot_level = pd.DataFrame(
            data=boot_level[np.array(steps) - 1],
            index=prediction_index,
            columns=boot_columns
        )

        boot_predictions[level] = boot_level

    # Temporal standardization. Pending full code refactoring:
    boot_predictions = (
        pd.concat([value.assign(level=key) for key, value in boot_predictions.items()])
        .reset_index()
        .sort_values(by=["index", "level"])
        .set_index("index")
        .rename_axis(None, axis=0)
    )
    boot_predictions = boot_predictions[
        ["level"] + [col for col in boot_predictions.columns if col not in ["level", "index"]]
        ]
    if isinstance(boot_predictions.index, pd.DatetimeIndex) and boot_predictions.index.freq is not None:
        boot_predictions.index.freq = None

    set_skforecast_warnings(suppress_warnings, action='default')

    return boot_predictions

predict_interval

predict_interval(
    steps,
    levels=None,
    last_window=None,
    exog=None,
    interval=[5, 95],
    n_boot=250,
    random_state=123,
    use_in_sample_residuals=True,
    suppress_warnings=False,
)

Iterative process in which, each prediction, is used as a predictor for the next step and bootstrapping is used to estimate prediction intervals. Both predictions and intervals are returned.

Parameters:

Name	Type	Description	Default
steps	int	Number of steps to predict.	required
levels	(str, list)	Time series to be predicted. If None all levels whose last window ends at the same datetime index will be predicted together.	None
last_window	pandas DataFrame	Series values used to create the predictors (lags) needed in the first iteration of the prediction (t + 1). If last_window = None, the values stored in self.last_window_ are used to calculate the initial predictors, and the predictions start right after training data.	None
exog	pandas Series, pandas DataFrame, dict	Exogenous variable/s included as predictor/s.	None
interval	(list, tuple)	Confidence of the prediction interval estimated. Sequence of percentiles to compute, which must be between 0 and 100 inclusive. For example, interval of 95% should be as interval = [2.5, 97.5].	`[5, 95]`
n_boot	int	Number of bootstrapping iterations used to estimate prediction intervals.	250
random_state	int	Sets a seed to the random generator, so that boot predictions are always deterministic.	123
use_in_sample_residuals	bool	If True, residuals from the training data are used as proxy of prediction error to create predictions. If False, out of sample residuals are used. In the latter case, the user should have calculated and stored the residuals within the forecaster (see set_out_sample_residuals()).	True
suppress_warnings	bool	If True, skforecast warnings will be suppressed during the prediction process. See skforecast.exceptions.warn_skforecast_categories for more information.	False

Returns:

Name	Type	Description
predictions	pandas DataFrame	Long-format DataFrame with the predictions and the lower and upper bounds of the estimated interval. The columns are level, pred, lower_bound, upper_bound.

Notes

More information about prediction intervals in forecasting: https://otexts.com/fpp3/prediction-intervals.html Forecasting: Principles and Practice (3^rd ed) Rob J Hyndman and George Athanasopoulos.

Source code in skforecast\deep_learning\_forecaster_rnn.py

def predict_interval(
        self,
        steps: int,
        levels: str | list[str] | None = None,
        last_window: pd.DataFrame | None = None,
        exog: pd.Series | pd.DataFrame | dict[str, pd.Series | pd.DataFrame] | None = None,
        interval: list[float] | tuple[float] = [5, 95],
        n_boot: int = 250,
        random_state: int = 123,
        use_in_sample_residuals: bool = True,
        suppress_warnings: bool = False
) -> pd.DataFrame:
    """
    Iterative process in which, each prediction, is used as a predictor
    for the next step and bootstrapping is used to estimate prediction
    intervals. Both predictions and intervals are returned.

    Parameters
    ----------
    steps : int
        Number of steps to predict. 
    levels : str, list, default None
        Time series to be predicted. If `None` all levels whose last window
        ends at the same datetime index will be predicted together.
    last_window : pandas DataFrame, default None
        Series values used to create the predictors (lags) needed in the
        first iteration of the prediction (t + 1).
        If `last_window = None`, the values stored in `self.last_window_` are
        used to calculate the initial predictors, and the predictions start
        right after training data.
    exog : pandas Series, pandas DataFrame, dict, default None
        Exogenous variable/s included as predictor/s.
    interval : list, tuple, default `[5, 95]`
        Confidence of the prediction interval estimated. Sequence of
        percentiles to compute, which must be between 0 and 100 inclusive.
        For example, interval of 95% should be as `interval = [2.5, 97.5]`.
    n_boot : int, default 250
        Number of bootstrapping iterations used to estimate prediction
        intervals.
    random_state : int, default 123
        Sets a seed to the random generator, so that boot predictions are always
        deterministic.
    use_in_sample_residuals : bool, default True
        If `True`, residuals from the training data are used as proxy of
        prediction error to create predictions. If `False`, out of sample
        residuals are used. In the latter case, the user should have
        calculated and stored the residuals within the forecaster (see
        `set_out_sample_residuals()`).
    suppress_warnings : bool, default False
        If `True`, skforecast warnings will be suppressed during the prediction
        process. See skforecast.exceptions.warn_skforecast_categories for more
        information.

    Returns
    -------
    predictions : pandas DataFrame
        Long-format DataFrame with the predictions and the lower and upper
        bounds of the estimated interval. The columns are `level`, `pred`,
        `lower_bound`, `upper_bound`.

    Notes
    -----
    More information about prediction intervals in forecasting:
    https://otexts.com/fpp3/prediction-intervals.html
    Forecasting: Principles and Practice (3rd ed) Rob J Hyndman and
    George Athanasopoulos.

    """

    set_skforecast_warnings(suppress_warnings, action='ignore')

    check_interval(interval=interval)

    boot_predictions = self.predict_bootstrapping(
        steps=steps,
        levels=levels,
        last_window=last_window,
        exog=exog,
        n_boot=n_boot,
        random_state=random_state,
        use_in_sample_residuals=use_in_sample_residuals,
        suppress_warnings=suppress_warnings
    )

    predictions = self.predict(
        steps=steps,
        levels=levels,
        last_window=last_window,
        exog=exog,
        suppress_warnings=suppress_warnings
        #check_inputs=False
    )

    interval = np.array(interval) / 100
    boot_predictions[['lower_bound', 'upper_bound']] = (
        boot_predictions.iloc[:, 1:].quantile(q=interval, axis=1).transpose()
    )

    predictions = pd.concat([
        predictions, boot_predictions[['lower_bound', 'upper_bound']]
    ], axis=1)

    set_skforecast_warnings(suppress_warnings, action='default')

    return predictions

predict_quantiles

predict_quantiles(
    steps,
    levels=None,
    last_window=None,
    exog=None,
    quantiles=[0.05, 0.5, 0.95],
    n_boot=250,
    random_state=123,
    use_in_sample_residuals=True,
    suppress_warnings=False,
)

Calculate the specified quantiles for each step. After generating multiple forecasting predictions through a bootstrapping process, each quantile is calculated for each step.

Parameters:

Name	Type	Description	Default
steps	int	Number of steps to predict.	required
levels	(str, list)	Time series to be predicted. If None all levels whose last window ends at the same datetime index will be predicted together.	None
last_window	pandas DataFrame	Series values used to create the predictors (lags) needed in the first iteration of the prediction (t + 1). If last_window = None, the values stored in self.last_window_ are used to calculate the initial predictors, and the predictions start right after training data.	None
exog	pandas Series, pandas DataFrame, dict	Exogenous variable/s included as predictor/s.	None
quantiles	(list, tuple)	Sequence of quantiles to compute, which must be between 0 and 1 inclusive. For example, quantiles of 0.05, 0.5 and 0.95 should be as quantiles = [0.05, 0.5, 0.95].	[0.05, 0.5, 0.95]
n_boot	int	Number of bootstrapping iterations used to estimate quantiles.	250
random_state	int	Sets a seed to the random generator, so that boot quantiles are always deterministic.	123
use_in_sample_residuals	bool	If True, residuals from the training data are used as proxy of prediction error to create quantiles. If False, out of sample residuals are used. In the latter case, the user should have calculated and stored the residuals within the forecaster (see set_out_sample_residuals()).	True
suppress_warnings	bool	If True, skforecast warnings will be suppressed during the prediction process. See skforecast.exceptions.warn_skforecast_categories for more information.	False

Returns:

Name	Type	Description
predictions	pandas DataFrame	Long-format DataFrame with the quantiles predicted by the forecaster. For example, if quantiles = [0.05, 0.5, 0.95], the columns are level, q_0.05, q_0.5, q_0.95.

Notes

More information about prediction intervals in forecasting: https://otexts.com/fpp3/prediction-intervals.html Forecasting: Principles and Practice (3^rd ed) Rob J Hyndman and George Athanasopoulos.

Source code in skforecast\deep_learning\_forecaster_rnn.py

def predict_quantiles(
        self,
        steps: int,
        levels: str | list[str] | None = None,
        last_window: pd.DataFrame | None = None,
        exog: pd.Series | pd.DataFrame | dict[str, pd.Series | pd.DataFrame] | None = None,
        quantiles: list[float] | tuple[float] = [0.05, 0.5, 0.95],
        n_boot: int = 250,
        random_state: int = 123,
        use_in_sample_residuals: bool = True,
        suppress_warnings: bool = False
) -> pd.DataFrame:
    """
    Calculate the specified quantiles for each step. After generating
    multiple forecasting predictions through a bootstrapping process, each
    quantile is calculated for each step.

    Parameters
    ----------
    steps : int
        Number of steps to predict. 
    levels : str, list, default None
        Time series to be predicted. If `None` all levels whose last window
        ends at the same datetime index will be predicted together.
    last_window : pandas DataFrame, default None
        Series values used to create the predictors (lags) needed in the
        first iteration of the prediction (t + 1).
        If `last_window = None`, the values stored in `self.last_window_` are
        used to calculate the initial predictors, and the predictions start
        right after training data.
    exog : pandas Series, pandas DataFrame, dict, default None
        Exogenous variable/s included as predictor/s.
    quantiles : list, tuple, default [0.05, 0.5, 0.95]
        Sequence of quantiles to compute, which must be between 0 and 1
        inclusive. For example, quantiles of 0.05, 0.5 and 0.95 should be as
        `quantiles = [0.05, 0.5, 0.95]`.
    n_boot : int, default 250
        Number of bootstrapping iterations used to estimate quantiles.
    random_state : int, default 123
        Sets a seed to the random generator, so that boot quantiles are always
        deterministic.
    use_in_sample_residuals : bool, default True
        If `True`, residuals from the training data are used as proxy of
        prediction error to create quantiles. If `False`, out of sample
        residuals are used. In the latter case, the user should have
        calculated and stored the residuals within the forecaster (see
        `set_out_sample_residuals()`).
    suppress_warnings : bool, default False
        If `True`, skforecast warnings will be suppressed during the prediction
        process. See skforecast.exceptions.warn_skforecast_categories for more
        information.

    Returns
    -------
    predictions : pandas DataFrame
        Long-format DataFrame with the quantiles predicted by the forecaster.
        For example, if `quantiles = [0.05, 0.5, 0.95]`, the columns are
        `level`, `q_0.05`, `q_0.5`, `q_0.95`.

    Notes
    -----
    More information about prediction intervals in forecasting:
    https://otexts.com/fpp3/prediction-intervals.html
    Forecasting: Principles and Practice (3rd ed) Rob J Hyndman and
    George Athanasopoulos.

    """

    set_skforecast_warnings(suppress_warnings, action='ignore')

    check_interval(quantiles=quantiles)

    predictions = self.predict_bootstrapping(
        steps=steps,
        levels=levels,
        last_window=last_window,
        exog=exog,
        n_boot=n_boot,
        random_state=random_state,
        use_in_sample_residuals=use_in_sample_residuals,
        suppress_warnings=suppress_warnings
    )

    quantiles_cols = [f'q_{q}' for q in quantiles]
    predictions[quantiles_cols] = (
        predictions.iloc[:, 1:].quantile(q=quantiles, axis=1).transpose()
    )
    predictions = predictions[['level'] + quantiles_cols]

    set_skforecast_warnings(suppress_warnings, action='default')

    return predictions

predict_dist

predict_dist(
    steps,
    distribution,
    levels=None,
    last_window=None,
    exog=None,
    n_boot=250,
    random_state=123,
    use_in_sample_residuals=True,
    suppress_warnings=False,
)

Fit a given probability distribution for each step. After generating multiple forecasting predictions through a bootstrapping process, each step is fitted to the given distribution.

Parameters:

Name	Type	Description	Default
steps	int	Number of steps to predict.	required
distribution	object	A distribution object from scipy.stats with methods _pdf and fit. For example scipy.stats.norm.	required
levels	(str, list)	Time series to be predicted. If None all levels whose last window ends at the same datetime index will be predicted together.	None
last_window	pandas DataFrame	Series values used to create the predictors (lags) needed in the first iteration of the prediction (t + 1). If last_window = None, the values stored in self.last_window_ are used to calculate the initial predictors, and the predictions start right after training data.	None
exog	pandas Series, pandas DataFrame, dict	Exogenous variable/s included as predictor/s.	None
n_boot	int	Number of bootstrapping iterations used to estimate predictions.	250
random_state	int	Sets a seed to the random generator, so that boot predictions are always deterministic.	123
use_in_sample_residuals	bool	If True, residuals from the training data are used as proxy of prediction error to create predictions. If False, out of sample residuals are used. In the latter case, the user should have calculated and stored the residuals within the forecaster (see set_out_sample_residuals()).	True
suppress_warnings	bool	If True, skforecast warnings will be suppressed during the prediction process. See skforecast.exceptions.warn_skforecast_categories for more information.	False

Returns:

Name	Type	Description
predictions	pandas DataFrame	Long-format DataFrame with the parameters of the fitted distribution for each step. The columns are level, param_0, param_1, ..., param_n, where param_i are the parameters of the distribution.

Source code in skforecast\deep_learning\_forecaster_rnn.py

def predict_dist(
        self,
        steps: int,
        distribution: object,
        levels: str | list[str] | None = None,
        last_window: pd.DataFrame | None = None,
        exog: pd.Series | pd.DataFrame | dict[str, pd.Series | pd.DataFrame] | None = None,
        n_boot: int = 250,
        random_state: int = 123,
        use_in_sample_residuals: bool = True,
        suppress_warnings: bool = False
) -> pd.DataFrame:
    """
    Fit a given probability distribution for each step. After generating
    multiple forecasting predictions through a bootstrapping process, each
    step is fitted to the given distribution.

    Parameters
    ----------
    steps : int
        Number of steps to predict. 
    distribution : object
        A distribution object from scipy.stats with methods `_pdf` and `fit`.
        For example scipy.stats.norm.
    levels : str, list, default None
        Time series to be predicted. If `None` all levels whose last window
        ends at the same datetime index will be predicted together.
    last_window : pandas DataFrame, default None
        Series values used to create the predictors (lags) needed in the
        first iteration of the prediction (t + 1).
        If `last_window = None`, the values stored in `self.last_window_` are
        used to calculate the initial predictors, and the predictions start
        right after training data.
    exog : pandas Series, pandas DataFrame, dict, default None
        Exogenous variable/s included as predictor/s.
    n_boot : int, default 250
        Number of bootstrapping iterations used to estimate predictions.
    random_state : int, default 123
        Sets a seed to the random generator, so that boot predictions are always
        deterministic.
    use_in_sample_residuals : bool, default True
        If `True`, residuals from the training data are used as proxy of
        prediction error to create predictions. If `False`, out of sample
        residuals are used. In the latter case, the user should have
        calculated and stored the residuals within the forecaster (see
        `set_out_sample_residuals()`).
    suppress_warnings : bool, default False
        If `True`, skforecast warnings will be suppressed during the prediction
        process. See skforecast.exceptions.warn_skforecast_categories for more
        information.

    Returns
    -------
    predictions : pandas DataFrame
        Long-format DataFrame with the parameters of the fitted distribution
        for each step. The columns are `level`, `param_0`, `param_1`, ...,
        `param_n`, where `param_i` are the parameters of the distribution.

    """

    if not hasattr(distribution, "_pdf") or not callable(getattr(distribution, "fit", None)):
        raise TypeError(
            "`distribution` must be a valid probability distribution object "
            "from scipy.stats, with methods `_pdf` and `fit`."
        )

    set_skforecast_warnings(suppress_warnings, action='ignore')

    predictions = self.predict_bootstrapping(
        steps=steps,
        levels=levels,
        last_window=last_window,
        exog=exog,
        n_boot=n_boot,
        random_state=random_state,
        use_in_sample_residuals=use_in_sample_residuals,
        suppress_warnings=suppress_warnings
    )

    param_names = [
                      p for p in inspect.signature(distribution._pdf).parameters if not p == "x"
                  ] + ["loc", "scale"]

    predictions[param_names] = (
        predictions.iloc[:, 1:].apply(
            lambda x: distribution.fit(x), axis=1, result_type='expand'
        )
    )
    predictions = predictions[['level'] + param_names]

    set_skforecast_warnings(suppress_warnings, action='default')

    return predictions

plot_history

plot_history(ax=None, **fig_kw)

Plots the training and validation loss curves from the given history object stores in the ForecasterRnn.

Parameters:

Name	Type	Description	Default
ax	Axes	Pre-existing ax for the plot. Otherwise, call matplotlib.pyplot.subplots() internally.	`None`
fig_kw	dict	Other keyword arguments are passed to matplotlib.pyplot.subplots().	{}

Returns:

Name	Type	Description
fig	Figure	Matplotlib Figure.

Source code in skforecast\deep_learning\_forecaster_rnn.py

def plot_history(
    self, ax: matplotlib.axes.Axes = None, **fig_kw
) -> matplotlib.figure.Figure:
    """
    Plots the training and validation loss curves from the given history object stores
    in the ForecasterRnn.

    Parameters
    ----------
    ax : matplotlib.axes.Axes, default `None`
        Pre-existing ax for the plot. Otherwise, call matplotlib.pyplot.subplots()
        internally.
    fig_kw : dict
        Other keyword arguments are passed to matplotlib.pyplot.subplots().

    Returns
    -------
    fig: matplotlib.figure.Figure
        Matplotlib Figure.

    """

    if ax is None:
        fig, ax = plt.subplots(1, 1, **fig_kw)
    else:
        fig = ax.get_figure()

    # Setting up the plot style

    if self.history is None:
        raise ValueError("ForecasterRnn has not been fitted yet.")

    # Plotting training loss
    ax.plot(
        range(1, len(self.history["loss"]) + 1),
        self.history["loss"],
        color="b",
        label="Training Loss",
    )

    # Plotting validation loss
    if "val_loss" in self.history:
        ax.plot(
            range(1, len(self.history["val_loss"]) + 1),
            self.history["val_loss"],
            color="r",
            label="Validation Loss",
        )

    # Labeling the axes and adding a title
    ax.set_xlabel("Epochs")
    ax.set_ylabel("Loss")
    ax.set_title("Training and Validation Loss")

    # Adding a legend
    ax.legend()

    # Displaying grid for better readability
    ax.grid(True, linestyle="--", alpha=0.7)

    # Setting x-axis ticks to integers only
    ax.set_xticks(range(1, len(self.history["loss"]) + 1))

set_params

set_params(params)

Set new values to the parameters of the scikit-learn model stored in the forecaster. It is important to note that all models share the same configuration of parameters and hyperparameters.

Parameters:

Name	Type	Description	Default
params	dict	Parameters values.	required

Returns:

Type	Description
None

Source code in skforecast\deep_learning\_forecaster_rnn.py

def set_params(self, params: dict) -> None:  # TODO testear
    """
    Set new values to the parameters of the scikit-learn model stored in the
    forecaster. It is important to note that all models share the same
    configuration of parameters and hyperparameters.

    Parameters
    ----------
    params : dict
        Parameters values.

    Returns
    -------
    None

    """

    self.regressor = clone(self.regressor)
    self.regressor.reset_states()
    self.regressor.compile(**params)

set_fit_kwargs

set_fit_kwargs(fit_kwargs)

Set new values for the additional keyword arguments passed to the fit method of the regressor.

Parameters:

Name	Type	Description	Default
fit_kwargs	dict	Dict of the form {"argument": new_value}.	required

Returns:

Type	Description
None

Source code in skforecast\deep_learning\_forecaster_rnn.py

def set_fit_kwargs(self, fit_kwargs: dict) -> None:
    """
    Set new values for the additional keyword arguments passed to the `fit`
    method of the regressor.

    Parameters
    ----------
    fit_kwargs : dict
        Dict of the form {"argument": new_value}.

    Returns
    -------
    None

    """

    self.fit_kwargs = check_select_fit_kwargs(self.regressor, fit_kwargs=fit_kwargs)

set_lags

set_lags(lags)

Not used, present here for API consistency by convention.

Returns:

Type	Description
None

Source code in skforecast\deep_learning\_forecaster_rnn.py

def set_lags(self, lags: Any) -> None:
    """
    Not used, present here for API consistency by convention.

    Returns
    -------
    None

    """

    pass

skforecast.deep_learning.utils.create_and_compile_model

create_and_compile_model(
    series,
    lags,
    steps,
    levels=None,
    recurrent_layer="LSTM",
    recurrent_units=100,
    dense_units=64,
    activation="relu",
    optimizer=Adam(learning_rate=0.01),
    loss=MeanSquaredError(),
    compile_kwargs={},
)

Creates a neural network model for time series prediction with flexible recurrent layers.

Parameters:

Name	Type	Description	Default
series	pandas DataFrame	Input time series.	required
lags	(int, list)	Number of lagged time steps to consider in the input, or a list of specific lag indices.	required
steps	(int, list)	Number of steps to predict into the future, or a list of specific step indices.	required
levels	(str, int, list)	Number of output levels (features) to predict, or a list of specific level indices. If None, defaults to the number of input series.	`None`
recurrent_layer	str	Type of recurrent layer to be used ('LSTM' or 'RNN').	`'LSTM'`
recurrent_units	(int, list)	Number of units in the recurrent layer(s). Can be an integer or a list of integers for multiple layers.	`100`
dense_units	(int, list)	List of integers representing the number of units in each dense layer.	`64`
activation	(str, dict)	Activation function for the recurrent and dense layers. Can be a single string for all layers or a dictionary specifying different activations for 'recurrent_units' and 'dense_units'.	`'relu'`
optimizer	object	Optimization algorithm and learning rate.	`Adam(learning_rate=0.01)`
loss	object	Loss function for model training.	`MeanSquaredError()`
compile_kwargs	dict	Additional arguments for model compilation.	`{}`

Returns:

Name	Type	Description
model	Model	Compiled neural network model.

Raises:

Type	Description
TypeError	If any of the input arguments are of incorrect type.
ValueError	If the activation dictionary does not have the required keys or if the lengths of the lists in the activation dictionary do not match the corresponding parameters.

Source code in skforecast\deep_learning\utils.py

def create_and_compile_model(
    series: pd.DataFrame,
    lags: Union[int, list],
    steps: Union[int, list],
    levels: Optional[Union[str, int, list]] = None,
    recurrent_layer: str = "LSTM",
    recurrent_units: Union[int, list] = 100,
    dense_units: Union[int, list] = 64,
    activation: Union[str, dict] = "relu",
    optimizer: object = Adam(learning_rate=0.01),
    loss: object = MeanSquaredError(),
    compile_kwargs: dict = {},
) -> keras.models.Model:
    """
    Creates a neural network model for time series prediction with flexible recurrent layers.

    Parameters
    ----------
    series : pandas DataFrame
        Input time series.
    lags : int, list
        Number of lagged time steps to consider in the input, or a list of 
        specific lag indices.
    steps : int, list
        Number of steps to predict into the future, or a list of specific step 
        indices.
    levels : str, int, list, default `None`
        Number of output levels (features) to predict, or a list of specific 
        level indices. If None, defaults to the number of input series.
    recurrent_layer : str, default `'LSTM'`
        Type of recurrent layer to be used ('LSTM' or 'RNN').
    recurrent_units : int, list, default `100`
        Number of units in the recurrent layer(s). Can be an integer or a 
        list of integers for multiple layers.
    dense_units : int, list, default `64`
        List of integers representing the number of units in each dense layer.
    activation : str, dict, default `'relu'`
        Activation function for the recurrent and dense layers. Can be a single 
        string for all layers or a dictionary specifying different activations 
        for 'recurrent_units' and 'dense_units'.
    optimizer : object, default `Adam(learning_rate=0.01)`
        Optimization algorithm and learning rate.
    loss : object, default `MeanSquaredError()`
        Loss function for model training.
    compile_kwargs : dict, default `{}` 
        Additional arguments for model compilation.

    Returns
    -------
    model : keras.models.Model
        Compiled neural network model.

    Raises
    ------
    TypeError
        If any of the input arguments are of incorrect type.
    ValueError
        If the activation dictionary does not have the required keys or if the 
        lengths of the lists in the activation dictionary do not match the 
        corresponding parameters.

    """

    if keras.__version__ > "3":
        print(f"keras version: {keras.__version__}")
        print(f"Using backend: {keras.backend.backend()}")
        if keras.backend.backend() == "tensorflow":
            import tensorflow
            print(f"tensorflow version: {tensorflow.__version__}")
        elif keras.backend.backend() == "torch":
            import torch
            print(f"torch version: {torch.__version__}")
        elif keras.backend.backend() == "jax":
            import jax
            print(f"jax version: {jax.__version__}")
        else:
            print("Backend not recognized")

    err_msg = f"`series` must be a pandas DataFrame. Got {type(series)}."

    if not isinstance(series, pd.DataFrame):
        raise TypeError(err_msg)

    n_series = series.shape[1]

    # Dense units must be a list, None or int
    if not isinstance(dense_units, (list, int, type(None))):
        raise TypeError(
            f"`dense_units` argument must be a list or int. Got {type(dense_units)}."
        )
    if isinstance(dense_units, int):
        dense_units = [dense_units]

    # Recurrent units must be a list or int
    if not isinstance(recurrent_units, (list, int)):
        raise TypeError(
            f"`recurrent_units` argument must be a list or int. Got {type(recurrent_units)}."
        )
    if isinstance(recurrent_units, int):
        recurrent_units = [recurrent_units]

    # Lags, steps and levels must be int or list
    if not isinstance(lags, (int, list)):
        raise TypeError(f"`lags` argument must be a list or int. Got {type(lags)}.")
    if not isinstance(steps, (int, list)):
        raise TypeError(f"`steps` argument must be a list or int. Got {type(steps)}.")
    if not isinstance(levels, (str, int, list, type(None))):
        raise TypeError(
            f"`levels` argument must be a string, list or int. Got {type(levels)}."
        )

    if isinstance(lags, list):
        lags = len(lags)
    if isinstance(steps, list):
        steps = len(steps)
    if isinstance(levels, list):
        levels = len(levels)
    elif isinstance(levels, (str)):
        levels = 1
    elif isinstance(levels, type(None)):
        levels = series.shape[1]
    elif isinstance(levels, int):
        pass
    else:
        raise TypeError(
            f"`levels` argument must be a string, list or int. Got {type(levels)}."
        )

    if isinstance(activation, str):
        if dense_units is not None:
            activation = {
                "recurrent_units": [activation] * len(recurrent_units),
                "dense_units": [activation] * len(dense_units),
            }
        else:
            activation = {
                "recurrent_units": [activation] * len(recurrent_units)
            }
    elif isinstance(activation, dict):
        # Check if the dictionary has the required keys
        if "recurrent_units" not in activation.keys():
            raise ValueError("The activation dictionary must have a 'recurrent_units' key.")
        if dense_units is not None and "dense_units" not in activation.keys():
            raise ValueError("The activation dictionary must have a 'dense_units' key if dense_units is not None.")
        # Check if the values are lists
        if not isinstance(activation["recurrent_units"], list):
            raise TypeError("The 'recurrent_units' value in the activation dictionary must be a list.")
        if dense_units is not None and not isinstance(activation["dense_units"], list):
            raise TypeError("The 'dense_units' value in the activation dictionary must be a list if dense_units is not None.")
        # Check if the lists have the same length as the corresponding parameters
        if len(activation["recurrent_units"]) != len(recurrent_units):
            raise ValueError("The 'recurrent_units' list in the activation dictionary must have the same length as the recurrent_units parameter.")
        if dense_units is not None and len(activation["dense_units"]) != len(dense_units):
            raise ValueError("The 'dense_units' list in the activation dictionary must have the same length as the dense_units parameter.")
    else:
        raise TypeError(f"`activation` argument must be a string or dict. Got {type(activation)}.")

    input_layer = Input(shape=(lags, n_series))
    x = input_layer

    # Dynamically create multiple recurrent layers if recurrent_units is a list
    if isinstance(recurrent_units, list):
        for i, units in enumerate(recurrent_units[:-1]):  # All layers except the last one
            if recurrent_layer == "LSTM":
                x = LSTM(units, activation=activation["recurrent_units"][i], return_sequences=True)(x)
            elif recurrent_layer == "RNN":
                x = SimpleRNN(units, activation=activation["recurrent_units"][i], return_sequences=True)(x)
            else:
                raise ValueError(f"Invalid recurrent layer: {recurrent_layer}")
        # Last layer without return_sequences
        if recurrent_layer == "LSTM":
            x = LSTM(recurrent_units[-1], activation=activation["recurrent_units"][-1])(x)
        elif recurrent_layer == "RNN":
            x = SimpleRNN(recurrent_units[-1], activation=activation["recurrent_units"][-1])(x)
        else:
            raise ValueError(f"Invalid recurrent layer: {recurrent_layer}")
    else:
        # Single recurrent layer
        if recurrent_layer == "LSTM":
            x = LSTM(recurrent_units, activation=activation["recurrent_units"][0])(x)
        elif recurrent_layer == "RNN":
            x = SimpleRNN(recurrent_units, activation=activation["recurrent_units"][0])(x)
        else:
            raise ValueError(f"Invalid recurrent layer: {recurrent_layer}")

    # Dense layers
    if dense_units is not None:
        for i, nn in enumerate(dense_units):
            x = Dense(nn, activation=activation["dense_units"][i])(x)

    # Output layer
    x = Dense(levels * steps, activation="linear")(x)
    # model = Model(inputs=input_layer, outputs=x)
    output_layer = keras.layers.Reshape((steps, levels))(x)
    model = Model(inputs=input_layer, outputs=output_layer)

    # Compile the model if optimizer, loss or compile_kwargs are passed
    if optimizer is not None or loss is not None or compile_kwargs:
        # give more priority to the parameters passed in the function check if the
        # parameters passes in compile_kwargs include optimizer and loss if so,
        # delete them from compile_kwargs and raise a warning
        if "optimizer" in compile_kwargs.keys():
            compile_kwargs.pop("optimizer")
            warnings.warn("`optimizer` passed in `compile_kwargs`. Ignoring it.")
        if "loss" in compile_kwargs.keys():
            compile_kwargs.pop("loss")
            warnings.warn("`loss` passed in `compile_kwargs`. Ignoring it.")

        model.compile(optimizer=optimizer, loss=loss, **compile_kwargs)

    return model