forecaster.wrappers.model
Recursive forecaster model wrappers for different estimators.
Classes
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 ,
include_covid_infection_rate= False ,
include_entsoe_forecast_load= False ,
include_entsoe_renewable_forecast= False ,
include_entsoe_net_load= False ,
include_entsoe_day_ahead_price= False ,
include_football_match_window= False ,
include_energy_saving_window= False ,
on_exog_provider_failure= 'raise' ,
exog_max_gap_hours= 0 ,
exog_max_tail_gap_hours= 0 ,
exog_provider_window= 'full' ,
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
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)
Methods
backtest
forecaster.wrappers.model.ForecasterRecursiveModel.backtest()
Back-test the forecaster on the test data.
Raises
ValueError
If the forecaster has not been initialized.
Examples
import os
import shutil
import tempfile
from pathlib import Path
import pandas as pd
from sklearn.linear_model import Ridge
from spotforecast2_safe.data.fetch_data import get_package_data_home
from spotforecast2_safe.forecaster.recursive import ForecasterRecursive
from spotforecast2_safe.forecaster.wrappers import ForecasterRecursiveModel
tmp = tempfile.mkdtemp()
os.environ["SPOTFORECAST2_DATA" ] = tmp
interim = Path(tmp) / "interim"
interim.mkdir(parents= True )
demo = get_package_data_home() / "demo01.csv"
df = pd.read_csv(demo).rename(
columns= {
"Time" : "Time (UTC)" ,
"Actual" : "Actual Load" ,
"Forecast" : "Forecasted Load" ,
}
)
df["Time (UTC)" ] = pd.to_datetime(df["Time (UTC)" ], utc= True )
df[df["Time (UTC)" ] <= "2022-01-15 23:00:00+00:00" ].to_csv(
interim / "energy_load.csv" , index= False
)
model = ForecasterRecursiveModel(
iteration= 0 ,
end_dev= "2022-01-10 00:00+00:00" ,
predict_size= 4 ,
refit_size= 1 ,
)
model.forecaster = ForecasterRecursive(estimator= Ridge(), lags= 4 )
model.best_params = {}
model.best_lags = list (range (1 , 5 ))
metrics = model.backtest()
assert "mean_absolute_error" in metrics.columns
print (metrics[["mean_absolute_error" , "mean_absolute_percentage_error" ]])
shutil.rmtree(tmp)
del os.environ["SPOTFORECAST2_DATA" ]
mean_absolute_error mean_absolute_percentage_error
0 0.00461 0.000317
fit
forecaster.wrappers.model.ForecasterRecursiveModel.fit(y, exog= None )
Fit the underlying forecaster.
Raises
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)
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)
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
ValueError
If the forecaster has not been initialized.
Examples
import os
import shutil
import tempfile
from pathlib import Path
import pandas as pd
from sklearn.linear_model import Ridge
from spotforecast2_safe.data.fetch_data import get_package_data_home
from spotforecast2_safe.forecaster.recursive import ForecasterRecursive
from spotforecast2_safe.forecaster.wrappers import ForecasterRecursiveModel
tmp = tempfile.mkdtemp()
os.environ["SPOTFORECAST2_DATA" ] = tmp
interim = Path(tmp) / "interim"
interim.mkdir(parents= True )
demo = get_package_data_home() / "demo01.csv"
df = pd.read_csv(demo).rename(
columns= {
"Time" : "Time (UTC)" ,
"Actual" : "Actual Load" ,
"Forecast" : "Forecasted Load" ,
}
)
df["Time (UTC)" ] = pd.to_datetime(df["Time (UTC)" ], utc= True )
df[df["Time (UTC)" ] <= "2022-01-15 23:00:00+00:00" ].to_csv(
interim / "energy_load.csv" , index= False
)
model = ForecasterRecursiveModel(
iteration= 0 ,
end_dev= "2022-01-10 00:00+00:00" ,
predict_size= 4 ,
refit_size= 1 ,
)
model.forecaster = ForecasterRecursive(estimator= Ridge(), lags= 4 )
model.best_params = {}
model.best_lags = list (range (1 , 5 ))
model.fit_with_best()
assert model.forecaster.is_fitted
print (model.forecaster.is_fitted)
shutil.rmtree(tmp)
del os.environ["SPOTFORECAST2_DATA" ]
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 one known name mismatch (end_train_default → end_dev). Caller-supplied overrides take precedence over config values.
Parameters
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.
{}
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)
get_error_forecast
forecaster.wrappers.model.ForecasterRecursiveModel.get_error_forecast(
delta_predict= None ,
)
Compute the error of the ENTSO-E benchmark forecast.
Examples
import os
import shutil
import tempfile
from pathlib import Path
import pandas as pd
from spotforecast2_safe.data.fetch_data import get_package_data_home
from spotforecast2_safe.forecaster.wrappers import ForecasterRecursiveModel
tmp = tempfile.mkdtemp()
os.environ["SPOTFORECAST2_DATA" ] = tmp
interim = Path(tmp) / "interim"
interim.mkdir(parents= True )
demo = get_package_data_home() / "demo01.csv"
df = pd.read_csv(demo).rename(
columns= {
"Time" : "Time (UTC)" ,
"Actual" : "Actual Load" ,
"Forecast" : "Forecasted Load" ,
}
)
df["Time (UTC)" ] = pd.to_datetime(df["Time (UTC)" ], utc= True )
df[df["Time (UTC)" ] <= "2022-01-15 23:00:00+00:00" ].to_csv(
interim / "energy_load.csv" , index= False
)
model = ForecasterRecursiveModel(
iteration= 0 ,
end_dev= "2022-01-10 00:00+00:00" ,
)
metrics, (y_actual, y_forecast) = model.get_error_forecast(
delta_predict= pd.Timedelta(hours= 4 )
)
assert {"mae" , "mape" } == set (metrics.keys())
assert len (y_actual) == len (y_forecast)
print (f"Benchmark MAE: { metrics['mae' ]:.4f} , steps: { len (y_actual)} " )
shutil.rmtree(tmp)
del os.environ["SPOTFORECAST2_DATA" ]
Benchmark MAE: 1.5931, steps: 5
get_error_training
forecaster.wrappers.model.ForecasterRecursiveModel.get_error_training()
Compute in-sample error on the training data.
Examples
import os
import shutil
import tempfile
from pathlib import Path
import pandas as pd
from sklearn.linear_model import Ridge
from spotforecast2_safe.data.fetch_data import get_package_data_home, load_timeseries
from spotforecast2_safe.forecaster.recursive import ForecasterRecursive
from spotforecast2_safe.forecaster.wrappers import ForecasterRecursiveModel
from spotforecast2_safe.preprocessing import LinearlyInterpolateTS
tmp = tempfile.mkdtemp()
os.environ["SPOTFORECAST2_DATA" ] = tmp
interim = Path(tmp) / "interim"
interim.mkdir(parents= True )
demo = get_package_data_home() / "demo01.csv"
df = pd.read_csv(demo).rename(
columns= {
"Time" : "Time (UTC)" ,
"Actual" : "Actual Load" ,
"Forecast" : "Forecasted Load" ,
}
)
df["Time (UTC)" ] = pd.to_datetime(df["Time (UTC)" ], utc= True )
df[df["Time (UTC)" ] <= "2022-01-15 23:00:00+00:00" ].to_csv(
interim / "energy_load.csv" , index= False
)
model = ForecasterRecursiveModel(
iteration= 0 ,
end_dev= "2022-01-10 00:00+00:00" ,
predict_size= 4 ,
refit_size= 1 ,
)
model.forecaster = ForecasterRecursive(estimator= Ridge(), lags= 4 )
model.tune()
y = load_timeseries(on_missing= "passthrough" )
y = LinearlyInterpolateTS(on_missing= "ffill_bfill" ).fit_transform(y)
X = model.preprocessor.build(
start_date= y.index.min (), end_date= model.end_dev
)
model.forecaster.fit(
y.loc[: model.end_dev], exog= X.loc[: model.end_dev]
)
metrics, (y_train, y_train_pred) = model.get_error_training()
assert {"mae" , "mape" } == set (metrics.keys())
assert len (y_train) == len (y_train_pred)
print (f"Train MAE: { metrics['mae' ]:.4f} " )
shutil.rmtree(tmp)
del os.environ["SPOTFORECAST2_DATA" ]
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
Optional [pd .DataFrame ]
pd.DataFrame or None: Feature importances, or None if the model does not support this operation.
Examples
import numpy as np
import pandas as pd
from lightgbm import LGBMRegressor
from spotforecast2_safe.forecaster.recursive import ForecasterRecursive
from spotforecast2_safe.forecaster.wrappers import ForecasterRecursiveModel
rng = np.random.default_rng(0 )
y = pd.Series(
rng.random(50 ),
index= pd.date_range("2023-01-01" , periods= 50 , freq= "h" ),
)
model = ForecasterRecursiveModel(iteration= 0 )
model.name = "lgbm"
model.forecaster = ForecasterRecursive(
estimator= LGBMRegressor(n_jobs= 1 , verbose=- 1 , random_state= 0 ),
lags= 4 ,
)
model.forecaster.fit(y= y)
fi = model.get_feature_importance()
assert fi is not None
assert "feature" in fi.columns and "importance" in fi.columns
print (fi.head(3 ).to_string(index= False ))
feature importance
lag_3 85
lag_4 7
lag_2 4
from spotforecast2_safe.forecaster.wrappers import ForecasterRecursiveModel
model_no_tree = ForecasterRecursiveModel(iteration= 0 )
model_no_tree.name = "ridge"
fi = model_no_tree.get_feature_importance()
assert fi is None
print (fi)
Regressor does not support feature importance!
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 real implementation (using
``shap.TreeExplainer``) belongs in the sibling package
``spotforecast2``; ``shap`` is not on sf2-safe's allowed
dependency list per ``MODEL_CARD.md``.
Parameters
frac
float
Fraction of training data to use for SHAP values.
0.1
Returns
pd .Series
pd.Series: Empty series (stub).
Examples
from spotforecast2_safe.forecaster.wrappers import ForecasterRecursiveModel
model = ForecasterRecursiveModel(iteration= 0 , name= "lgbm" )
shap_vals = model.get_global_shap_feature_importance(frac= 0.1 )
assert shap_vals.empty
print (type (shap_vals), shap_vals.empty)
get_global_shap_feature_importance is a stub in spotforecast2-safe. Use spotforecast2 for the full implementation.
<class 'pandas.Series'> True
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
deep
bool
If True, will return the parameters for this forecaster model and contained sub-objects that are estimators.
True
Returns
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)
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)
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
predict_size
Optional [int ]
Optional override for the prediction horizon.
None
Returns
Dict [str , Any ]
Dict[str, Any]: A result package containing actual values, predictions, and calculated metrics (MAE, MAPE).
Raises
NotFittedError
If the wrapper has no fitted forecaster (i.e. fit was never called and no forecaster was assigned).
PredictionPackageError
If the underlying prediction pipeline fails for any other reason. The original exception is chained via __cause__.
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. │
│ │
│ Category : spotforecast2.exceptions.IgnoredArgumentWarning │
│ Location : │
│ /home/runner/work/spotforecast2-safe/spotforecast2-safe/src/spotforecast2_safe/prepr │
│ ocessing/_binner.py:259 │
│ Suppress : warnings.simplefilter('ignore', category=IgnoredArgumentWarning) │
╰──────────────────────────────────────────────────────────────────────────────────────╯
predict
forecaster.wrappers.model.ForecasterRecursiveModel.predict(delta_predict= None )
Generate predictions and compute error metrics.
Parameters
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
Raises
ValueError
If the forecaster has not been initialized.
Examples
import os
import shutil
import tempfile
from pathlib import Path
import pandas as pd
from sklearn.linear_model import Ridge
from spotforecast2_safe.data.fetch_data import get_package_data_home, load_timeseries
from spotforecast2_safe.forecaster.recursive import ForecasterRecursive
from spotforecast2_safe.forecaster.wrappers import ForecasterRecursiveModel
from spotforecast2_safe.preprocessing import LinearlyInterpolateTS
tmp = tempfile.mkdtemp()
os.environ["SPOTFORECAST2_DATA" ] = tmp
interim = Path(tmp) / "interim"
interim.mkdir(parents= True )
demo = get_package_data_home() / "demo01.csv"
df = pd.read_csv(demo).rename(
columns= {
"Time" : "Time (UTC)" ,
"Actual" : "Actual Load" ,
"Forecast" : "Forecasted Load" ,
}
)
df["Time (UTC)" ] = pd.to_datetime(df["Time (UTC)" ], utc= True )
df[df["Time (UTC)" ] <= "2022-01-15 23:00:00+00:00" ].to_csv(
interim / "energy_load.csv" , index= False
)
model = ForecasterRecursiveModel(
iteration= 0 ,
end_dev= "2022-01-10 00:00+00:00" ,
predict_size= 4 ,
refit_size= 1 ,
)
model.forecaster = ForecasterRecursive(estimator= Ridge(), lags= 4 )
model.tune()
y = load_timeseries(on_missing= "passthrough" )
y = LinearlyInterpolateTS(on_missing= "ffill_bfill" ).fit_transform(y)
X = model.preprocessor.build(
start_date= y.index.min (), end_date= y.index.max ()
)
model.forecaster.fit(
y.loc[: model.end_dev], exog= X.loc[: model.end_dev]
)
metrics, (y_actual, y_pred) = model.predict(
delta_predict= pd.Timedelta(hours= 4 )
)
assert {"mae" , "mape" } == set (metrics.keys())
print (f"MAE: { metrics['mae' ]:.4f} , steps predicted: { len (y_pred)} " )
shutil.rmtree(tmp)
del os.environ["SPOTFORECAST2_DATA" ]
MAE: 0.0011, steps predicted: 5
save_to_file
forecaster.wrappers.model.ForecasterRecursiveModel.save_to_file(model_dir= None )
Serialize the model to disk via joblib.dump().
Parameters
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)))
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
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
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)
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)
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. The real implementation (using bayesian_search_forecaster) belongs in the sibling package spotforecast2; sf2-safe deliberately excludes auto-tuning per MODEL_CARD.md.
Examples
from spotforecast2_safe.forecaster.wrappers import ForecasterRecursiveModel
model = ForecasterRecursiveModel(iteration= 0 , name= "ridge" )
assert not model.is_tuned
model.tune()
assert model.is_tuned
print (model.is_tuned)