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.
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.
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.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.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.series_val = None
    if "series_val" in fit_kwargs:
        self.series_val = fit_kwargs["series_val"]
        fit_kwargs.pop("series_val")

    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 self.series_val is not None:
        X_val, y_val, _ = self.create_train_X_y(series=self.series_val)
        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')

predict

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

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`

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
) -> 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.

    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

    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