forecaster.wrappers.model

forecaster.wrappers.model

Recursive forecaster model wrappers for different estimators.

Classes

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

ForecasterRecursiveModel

forecaster.wrappers.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 sklearn.linear_model import LinearRegression

from spotforecast2_safe.forecaster.recursive import ForecasterRecursive
from spotforecast2_safe.forecaster.wrappers import ForecasterRecursiveModel

model = ForecasterRecursiveModel(iteration=0)
model.forecaster = ForecasterRecursive(estimator=LinearRegression(), lags=1)
model.name = "linear"
model.tune()
print(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 joblib.dump().
set_params	Set the parameters of this forecaster model.
tune	Simulate hyperparameter tuning.

backtest

forecaster.wrappers.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

forecaster.wrappers.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 numpy as np
import pandas as pd
from sklearn.linear_model import LinearRegression
from spotforecast2_safe.forecaster.recursive import ForecasterRecursive
from spotforecast2_safe.forecaster.wrappers import ForecasterRecursiveModel

rng = np.random.default_rng(0)
model = ForecasterRecursiveModel(iteration=0)
model.forecaster = ForecasterRecursive(estimator=LinearRegression(), lags=3)
y = pd.Series(
    rng.random(10),
    index=pd.date_range("2023-01-01", periods=10, freq="h"),
)
model.fit(y=y)
print(model.forecaster.is_fitted)

True

import numpy as np
import pandas as pd
from sklearn.linear_model import LinearRegression
from spotforecast2_safe.forecaster.recursive import ForecasterRecursive
from spotforecast2_safe.forecaster.wrappers import ForecasterRecursiveModel

rng = np.random.default_rng(1)
y = pd.Series(
    rng.random(10),
    index=pd.date_range("2023-01-01", periods=10, freq="h"),
)
model_exog = ForecasterRecursiveModel(iteration=0)
model_exog.forecaster = ForecasterRecursive(estimator=LinearRegression(), lags=3)
exog = pd.DataFrame(
    rng.random((10, 2)),
    index=y.index,
    columns=["exog_1", "exog_2"],
)
model_exog.fit(y=y, exog=exog)
print(model_exog.forecaster.is_fitted)

True

fit_with_best

forecaster.wrappers.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

forecaster.wrappers.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.forecaster.wrappers import ForecasterRecursiveModel
from spotforecast2_safe.configurator.config_multi import ConfigMulti

cfg = ConfigMulti(country_code="FR", predict_size=48)
model = ForecasterRecursiveModel.from_config(iteration=1, config=cfg)
print(model.predict_size)

48

get_error_forecast

forecaster.wrappers.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

forecaster.wrappers.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

forecaster.wrappers.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

forecaster.wrappers.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

forecaster.wrappers.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 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.forecaster.wrappers import ForecasterRecursiveModel

model = ForecasterRecursiveModel(iteration=0)
p = model.get_params(deep=False)
print(p["iteration"])
print(p["name"])
print(p["predict_size"])
print("forecaster" not in p)

0
base
24
True

from sklearn.linear_model import LinearRegression
from spotforecast2_safe.forecaster.recursive import ForecasterRecursive
from spotforecast2_safe.forecaster.wrappers import ForecasterRecursiveModel

model2 = ForecasterRecursiveModel(iteration=1)
model2.forecaster = ForecasterRecursive(estimator=LinearRegression(), lags=3)
p2 = model2.get_params(deep=False)
print(len(p2["forecaster__lags"]))
print(isinstance(p2["forecaster__estimator"], LinearRegression))

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

3
True
True

package_prediction

forecaster.wrappers.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 predict(), get_error_training(), and 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 shutil
import tempfile
from pathlib import Path

import pandas as pd
from lightgbm import LGBMRegressor

from spotforecast2_safe.data.fetch_data import get_package_data_home
from spotforecast2_safe.forecaster.recursive import ForecasterRecursive
from spotforecast2_safe.forecaster.wrappers import ForecasterRecursiveLGBM

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

print("train_actual" in result and "future_pred" in result)

# Cleanup
shutil.rmtree(tmp_dir)
del os.environ["SPOTFORECAST2_DATA"]

╭─────────────────────────────── IgnoredArgumentWarning ───────────────────────────────╮
│ The number of bins has been reduced from 10 to 6 due to duplicated edges caused by   │
│ repeated predicted values.\n\nCategory :                                             │
│ spotforecast2.exceptions.IgnoredArgumentWarning\nLocation :                          │
│ /home/runner/work/spotforecast2-safe/spotforecast2-safe/src/spotforecast2_safe/prepr │
│ ocessing/_binner.py:233\nSuppress : warnings.simplefilter('ignore',                  │
│ category=IgnoredArgumentWarning)                                                     │
╰──────────────────────────────────────────────────────────────────────────────────────╯

True

predict

forecaster.wrappers.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

forecaster.wrappers.model.ForecasterRecursiveModel.save_to_file(model_dir=None)

Serialize the model to disk via joblib.dump().

Parameters

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

Examples

import os
import tempfile
from spotforecast2_safe.forecaster.wrappers import ForecasterRecursiveModel

model = ForecasterRecursiveModel(iteration=0, name="test")
with tempfile.TemporaryDirectory() as tmpdir:
    model.save_to_file(model_dir=tmpdir)
    print(any("test_forecaster_0" in f for f in os.listdir(tmpdir)))

True

set_params

forecaster.wrappers.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 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

from spotforecast2_safe.forecaster.wrappers import ForecasterRecursiveModel

model = ForecasterRecursiveModel(iteration=0)
_ = model.set_params(name="updated", predict_size=48)
print(model.name)
print(model.predict_size)

updated
48

from sklearn.linear_model import LinearRegression
from spotforecast2_safe.forecaster.recursive import ForecasterRecursive
from spotforecast2_safe.forecaster.wrappers import ForecasterRecursiveModel

model2 = ForecasterRecursiveModel(iteration=1)
model2.forecaster = ForecasterRecursive(estimator=LinearRegression(), lags=3)
_ = model2.set_params(params={"forecaster__estimator__fit_intercept": False})
print(model2.forecaster.estimator.fit_intercept)

False

tune

forecaster.wrappers.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