manager.models.forecaster_recursive_model

manager.models.forecaster_recursive_model

Recursive forecaster model wrappers for different estimators.

Classes

Name	Description
ForecasterRecursiveModel	Base wrapper around ForecasterRecursive to match application logic.

ForecasterRecursiveModel

manager.models.forecaster_recursive_model.ForecasterRecursiveModel(
    iteration,
    end_dev=None,
    train_size=None,
    periods=None,
    country_code='DE',
    random_state=123456789,
    predict_size=24,
    refit_size=7,
    name='base',
    **kwargs,
)

Base wrapper around ForecasterRecursive to match application logic.

This class manages the lifecycle of a recursive forecaster, including feature building, tuning (simulated), and packaging predictions for UI.

Attributes

Name	Type	Description
iteration	int	The current training iteration.
end_dev	pd.Timestamp	The end date of the development/training period.
train_size	Optional[pd.Timedelta]	Lookback window for training data.
preprocessor	ExogBuilder	Builder for exogenous features.
name	str	Label for the model type.
forecaster	Optional[`ForecasterRecursive`]	The underlying forecaster instance.
is_tuned	bool	Flag indicating if hyperparameter tuning has been performed.
predict_size	int	Prediction horizon in hours.
refit_size	int	Refit interval in days.
random_state	int	Seed for reproducibility.

Examples

>>> import pandas as pd
>>> from spotforecast2_safe.manager.models.forecaster_recursive_model import ForecasterRecursiveModel
>>> from spotforecast2_safe.forecaster.recursive import ForecasterRecursive
>>> from sklearn.linear_model import LinearRegression
>>>
>>> model = ForecasterRecursiveModel(iteration=0)
>>> model.forecaster = ForecasterRecursive(estimator=LinearRegression(), lags=1)
>>> model.name = "linear"
>>> model.tune()
>>> model.is_tuned
True

Methods

Name	Description
backtest	Back-test the forecaster on the test data.
fit	Fit the underlying forecaster.
fit_with_best	Fit the forecaster using the recorded best hyperparameters.
from_config	Create a model instance with defaults drawn from a config object.
get_error_forecast	Compute the error of the ENTSO-E benchmark forecast.
get_error_training	Compute in-sample error on the training data.
get_feature_importance	Return feature importances from the underlying estimator.
get_global_shap_feature_importance	Return global SHAP-based feature importances.
get_params	Get parameters for this forecaster model.
package_prediction	Package predictions, training errors, and benchmarks for the UI.
predict	Generate predictions and compute error metrics.
save_to_file	Serialize the model to disk via :func:`joblib.dump`.
set_params	Set the parameters of this forecaster model.
tune	Simulate hyperparameter tuning.

backtest

manager.models.forecaster_recursive_model.ForecasterRecursiveModel.backtest()

Back-test the forecaster on the test data.

Returns

Name	Type	Description
	pd.DataFrame	pd.DataFrame: Backtesting metric values.

Raises

Name	Type	Description
	ValueError	If the forecaster has not been initialized.

fit

manager.models.forecaster_recursive_model.ForecasterRecursiveModel.fit(
    y,
    exog=None,
)

Fit the underlying forecaster.

Parameters

Name	Type	Description	Default
y	pd.Series	Target time series.	required
exog	Optional[pd.DataFrame]	Optional exogenous features.	`None`

Raises

Name	Type	Description
	ValueError	If the forecaster has not been initialized.

Examples

>>> import pandas as pd
>>> import numpy as np
>>> from spotforecast2_safe.manager.models.forecaster_recursive_model import ForecasterRecursiveModel
>>> from spotforecast2_safe.forecaster.recursive import ForecasterRecursive
>>> from sklearn.linear_model import LinearRegression
>>>
>>> # Example 1: Basic usage with pd.Series
>>> model = ForecasterRecursiveModel(iteration=0)
>>> model.forecaster = ForecasterRecursive(estimator=LinearRegression(), lags=3)
>>> y = pd.Series(
...     np.random.rand(10),
...     index=pd.date_range("2023-01-01", periods=10, freq="h")
... )
>>> model.fit(y=y)
>>> model.forecaster.is_fitted
True
>>>
>>> # Example 2: Usage with exogenous variables
>>> model_exog = ForecasterRecursiveModel(iteration=0)
>>> model_exog.forecaster = ForecasterRecursive(estimator=LinearRegression(), lags=3)
>>> exog = pd.DataFrame(
...     np.random.rand(10, 2),
...     index=y.index,
...     columns=["exog_1", "exog_2"]
... )
>>> model_exog.fit(y=y, exog=exog)
>>> model_exog.forecaster.is_fitted
True

fit_with_best

manager.models.forecaster_recursive_model.ForecasterRecursiveModel.fit_with_best(
)

Fit the forecaster using the recorded best hyperparameters.

After tuning (or manually setting best_params and best_lags), this method loads the data, sets the optimal parameters/lags, and fits the forecaster on the full training + dev set up to end_dev.

Raises

Name	Type	Description
	ValueError	If the forecaster has not been initialized.

from_config

manager.models.forecaster_recursive_model.ForecasterRecursiveModel.from_config(
    iteration,
    config,
    **overrides,
)

Create a model instance with defaults drawn from a config object.

Extracts every __init__ parameter that exists as a config attribute, translating the two known name mismatches (API_COUNTRY_CODE → country_code, end_train_default → end_dev). Caller-supplied overrides take precedence over config values.

Parameters

Name	Type	Description	Default
iteration	int	Current iteration index.	required
config	Any	A config object (e.g. `ConfigMulti` or `ConfigEntsoe`) whose attributes overlap with the model’s `__init__` kwargs.	required
**overrides	Any	Explicit keyword arguments that override config values.	`{}`

Returns

Name	Type	Description
	ForecasterRecursiveModel	An instance of cls (or the calling subclass).

Examples

>>> from spotforecast2_safe.manager.configurator.config_multi import ConfigMulti
>>> from spotforecast2_safe.manager.models.forecaster_recursive_model import (
...     ForecasterRecursiveModel,
... )
>>> cfg = ConfigMulti(country_code="FR", predict_size=48)
>>> model = ForecasterRecursiveModel.from_config(iteration=1, config=cfg)
>>> model.predict_size
48

get_error_forecast

manager.models.forecaster_recursive_model.ForecasterRecursiveModel.get_error_forecast(
    delta_predict=None,
)

Compute the error of the ENTSO-E benchmark forecast.

Parameters

Name	Type	Description	Default
delta_predict	Optional[pd.Timedelta]	Optional prediction horizon.	`None`

Returns

Name	Type	Description
	Tuple[dict, Tuple[pd.Series, pd.Series]]	Tuple[dict, Tuple[pd.Series, pd.Series]]: `(metrics, (y_actual, y_forecast))`.

get_error_training

manager.models.forecaster_recursive_model.ForecasterRecursiveModel.get_error_training(
)

Compute in-sample error on the training data.

Returns

Name	Type	Description
	Tuple[dict, Tuple[pd.Series, pd.Series]]	Tuple[dict, Tuple[pd.Series, pd.Series]]: `(metrics, (y_train, y_train_pred))`.

get_feature_importance

manager.models.forecaster_recursive_model.ForecasterRecursiveModel.get_feature_importance(
)

Return feature importances from the underlying estimator.

Only supported for tree-based models (xgb, lgbm).

Returns

Name	Type	Description
	Optional[pd.DataFrame]	pd.DataFrame or None: Feature importances, or None if the model does not support this operation.

get_global_shap_feature_importance

manager.models.forecaster_recursive_model.ForecasterRecursiveModel.get_global_shap_feature_importance(
    frac=0.1,
)

Return global SHAP-based feature importances.

.. note::

This is a stub.  The full implementation using
``shap.TreeExplainer`` will be provided in the
``spotforecast2`` package.

#TODO: Implement shap feature importance in spotforecast2

Parameters

Name	Type	Description	Default
frac	float	Fraction of training data to use for SHAP values.	`0.1`

Returns

Name	Type	Description
	pd.Series	pd.Series: Empty series (stub).

get_params

manager.models.forecaster_recursive_model.ForecasterRecursiveModel.get_params(
    deep=True,
)

Get parameters for this forecaster model.

Collects wrapper-level parameters (iteration, end_dev, train_size, random_state, predict_size, refit_size, name) and, when a forecaster is attached, delegates to :meth:ForecasterRecursive.get_params for forecaster-level parameters (estimator, lags, window_features, etc.).

Parameters

Name	Type	Description	Default
deep	bool	If True, will return the parameters for this forecaster model and contained sub-objects that are estimators.	`True`

Returns

Name	Type	Description
params	Dict[str, object]	Dictionary of parameter names mapped to their values.

Examples

>>> from spotforecast2_safe.manager.models.forecaster_recursive_model import (
...     ForecasterRecursiveModel,
... )
>>> model = ForecasterRecursiveModel(iteration=0)
>>> p = model.get_params(deep=False)
>>> p["iteration"]
0
>>> p["name"]
'base'
>>> p["predict_size"]
24
>>> "forecaster" not in p  # forecaster is None
True

>>> from sklearn.linear_model import LinearRegression
>>> from spotforecast2_safe.forecaster.recursive import ForecasterRecursive
>>> model2 = ForecasterRecursiveModel(iteration=1)
>>> model2.forecaster = ForecasterRecursive(
...     estimator=LinearRegression(), lags=3
... )
>>> p2 = model2.get_params(deep=False)
>>> len(p2["forecaster__lags"])
3
>>> isinstance(p2["forecaster__estimator"], LinearRegression)
True

>>> p3 = model2.get_params(deep=True)
>>> "forecaster__estimator__fit_intercept" in p3
True

package_prediction

manager.models.forecaster_recursive_model.ForecasterRecursiveModel.package_prediction(
    predict_size=None,
)

Package predictions, training errors, and benchmarks for the UI.

This is the main entry-point used by the application layer. It delegates to :meth:predict, :meth:get_error_training, and :meth:get_error_forecast.

Parameters

Name	Type	Description	Default
predict_size	Optional[int]	Optional override for the prediction horizon.	`None`

Returns

Name	Type	Description
	Dict[str, Any]	Dict[str, Any]: A result package containing actual values, predictions, and calculated metrics (MAE, MAPE).

Examples

>>> import os
>>> import tempfile
>>> import pandas as pd
>>> from pathlib import Path
>>> from spotforecast2_safe.manager.models.forecaster_recursive_lgbm import ForecasterRecursiveLGBM
>>> from spotforecast2_safe.data.fetch_data import get_package_data_home
>>>
>>> # Setup temporary data environment
>>> tmp_dir = tempfile.mkdtemp()
>>> os.environ["SPOTFORECAST2_DATA"] = tmp_dir
>>> data_path = Path(tmp_dir) / "interim"
>>> data_path.mkdir(parents=True)
>>>
>>> # Load demo data and rename columns to match expectations
>>> demo_path = get_package_data_home() / "demo01.csv"
>>> df = pd.read_csv(demo_path)
>>> df = df.rename(columns={
...     "Time": "Time (UTC)",
...     "Actual": "Actual Load",
...     "Forecast": "Forecasted Load"
... })
>>> df.to_csv(data_path / "energy_load.csv", index=False)
>>>
>>> # Initialize model — override forecaster for small demo data
>>> from spotforecast2_safe.forecaster.recursive import ForecasterRecursive
>>> from lightgbm import LGBMRegressor
>>> model = ForecasterRecursiveLGBM(iteration=0, end_dev="2022-01-05 00:00+00:00")
>>> model.forecaster = ForecasterRecursive(
...     estimator=LGBMRegressor(n_jobs=-1, verbose=-1, random_state=123456789),
...     lags=12,
... )
>>> result = model.package_prediction(predict_size=24)
>>>
>>> # Validate output
>>> "train_actual" in result and "future_pred" in result
True
>>>
>>> # Cleanup
>>> import shutil
>>> shutil.rmtree(tmp_dir)
>>> del os.environ["SPOTFORECAST2_DATA"]

predict

manager.models.forecaster_recursive_model.ForecasterRecursiveModel.predict(
    delta_predict=None,
)

Generate predictions and compute error metrics.

Parameters

Name	Type	Description	Default
delta_predict	Optional[pd.Timedelta]	Optional time horizon to predict. If None or if it exceeds the available data, predicts to the end of the dataset.	`None`

Returns

Name	Type	Description
	Tuple[dict, Tuple[pd.Series, pd.Series]]	Tuple[dict, Tuple[pd.Series, pd.Series]]: `(metrics, (y_actual, y_predicted))`.

Raises

Name	Type	Description
	ValueError	If the forecaster has not been initialized.

save_to_file

manager.models.forecaster_recursive_model.ForecasterRecursiveModel.save_to_file(
    model_dir=None,
)

Serialize the model to disk via :func:joblib.dump.

Parameters

Name	Type	Description	Default
model_dir	Optional[Union[str, Path]]	Directory for the model file. If None, defaults to :func:`get_cache_home`.	`None`

Examples

>>> import tempfile
>>> from spotforecast2_safe.manager.models.forecaster_recursive_model import (
...     ForecasterRecursiveModel,
... )
>>> model = ForecasterRecursiveModel(iteration=0, name="test")
>>> with tempfile.TemporaryDirectory() as tmpdir:
...     model.save_to_file(model_dir=tmpdir)
...     import os; any("test_forecaster_0" in f for f in os.listdir(tmpdir))
True

set_params

manager.models.forecaster_recursive_model.ForecasterRecursiveModel.set_params(
    params=None,
    **kwargs,
)

Set the parameters of this forecaster model.

Wrapper-level keys (iteration, name, predict_size, …) are set directly on the model. Keys prefixed with forecaster__ are forwarded to :meth:ForecasterRecursive.set_params.

Parameters

Name	Type	Description	Default
params	Dict[str, object]	Optional dictionary of parameter names mapped to their new values. If provided, these parameters are set first.	`None`
**kwargs	object	Additional parameter names mapped to their new values. Parameters can target the wrapper (e.g. `name="new"`), the forecaster (e.g. `forecaster__lags=5`), or the estimator inside the forecaster (e.g. `forecaster__estimator__fit_intercept=False`).	`{}`

Returns

Name	Type	Description
ForecasterRecursiveModel	ForecasterRecursiveModel	The model instance with updated parameters (supports method chaining).

Examples

Setting wrapper-level parameters:

>>> from spotforecast2_safe.manager.models.forecaster_recursive_model import (
...     ForecasterRecursiveModel,
... )
>>> model = ForecasterRecursiveModel(iteration=0)
>>> _ = model.set_params(name="updated", predict_size=48)
>>> model.name
'updated'
>>> model.predict_size
48

Setting forecaster-level parameters on an attached forecaster:

>>> from sklearn.linear_model import LinearRegression
>>> from spotforecast2_safe.forecaster.recursive import ForecasterRecursive
>>> model2 = ForecasterRecursiveModel(iteration=1)
>>> model2.forecaster = ForecasterRecursive(
...     estimator=LinearRegression(), lags=3
... )
>>> _ = model2.set_params(
...     params={"forecaster__estimator__fit_intercept": False}
... )
>>> model2.forecaster.estimator.fit_intercept
False

tune

manager.models.forecaster_recursive_model.ForecasterRecursiveModel.tune()

Simulate hyperparameter tuning.

In spotforecast2-safe this is a simulated stub that marks the model as tuned without performing an actual Bayesian search. A full implementation using bayesian_search_forecaster will be provided in the spotforecast2 package.

#TODO: Implement real Bayesian search in spotforecast2