manager.persistence

manager.persistence

Functions

Name	Description
ensure_model_dir	Ensure model directory exists.
get_model_filepath	Get filepath for a single model.
load_forecasters	Load trained forecasters from disk using joblib.
model_directory_exists	Check if model directory exists.
save_forecaster	Save a single trained forecaster to disk using joblib.
save_forecasters	Save trained forecasters to disk using joblib.

ensure_model_dir

manager.persistence.ensure_model_dir(model_dir)

Ensure model directory exists.

Parameters

Name	Type	Description	Default
model_dir	Union[str, Path]	Directory path for model storage.	required

Returns

Name	Type	Description
Path	Path	Validated Path object.

Raises

Name	Type	Description
	OSError	If directory cannot be created.

Examples

import tempfile
from pathlib import Path
from spotforecast2_safe.manager.persistence import ensure_model_dir

with tempfile.TemporaryDirectory() as tmpdir:
    model_path = ensure_model_dir(Path(tmpdir) / "models")
    assert model_path.exists()
    print(model_path.exists())

with tempfile.TemporaryDirectory() as tmpdir:
    nested = ensure_model_dir(Path(tmpdir) / "a" / "b" / "c")
    assert nested.is_dir()
    print(nested.is_dir())

True
True

get_model_filepath

manager.persistence.get_model_filepath(model_dir, target)

Get filepath for a single model.

Parameters

Name	Type	Description	Default
model_dir	Path	Directory containing models.	required
target	str	Target variable name.	required

Returns

Name	Type	Description
Path	Path	Full filepath for the model.

Examples

from pathlib import Path
from spotforecast2_safe.manager.persistence import get_model_filepath

path = get_model_filepath(Path("/models"), "power")
assert path.name == "forecaster_power.joblib"
assert path.suffix == ".joblib"
print(path.name)
print(path.suffix)

path2 = get_model_filepath(Path("/models"), "energy")
assert path2.name == "forecaster_energy.joblib"
print(path2.name)

forecaster_power.joblib
.joblib
forecaster_energy.joblib

load_forecasters

manager.persistence.load_forecasters(target_columns, model_dir, verbose=False)

Load trained forecasters from disk using joblib.

Attempts to load all forecasters for given targets. Missing models are indicated in the return value for selective retraining.

Parameters

Name	Type	Description	Default
target_columns	List[str]	List of target variable names to load.	required
model_dir	Union[str, Path]	Directory containing saved models.	required
verbose	bool	Print progress messages. Default: False.	`False`

Returns

Name	Type	Description
	Dict[str, object]	Tuple[Dict[str, object], List[str]]:
	List[str]	- forecasters: Dictionary of successfully loaded forecaster objects.
	Tuple[Dict[str, object], List[str]]	- missing_targets: List of target names with no saved model file on disk (eligible for first-time training). A model whose file exists but fails to deserialise raises `OSError` instead.

Raises

Name	Type	Description
	OSError	If a model file exists on disk but cannot be deserialised (corrupt joblib, version mismatch, etc.). The caller cannot silently retrain over a corrupt model — they must observe the failure and decide.

Examples

import tempfile
import warnings
import numpy as np
import pandas as pd
from lightgbm import LGBMRegressor
from spotforecast2_safe.forecaster.recursive._forecaster_recursive import (
    ForecasterRecursive,
)
from spotforecast2_safe.manager.persistence import load_forecasters, save_forecasters

rng = np.random.default_rng(0)
y = pd.Series(rng.normal(size=50), name="target")
with warnings.catch_warnings():
    warnings.simplefilter("ignore")
    fc = ForecasterRecursive(
        estimator=LGBMRegressor(
            n_estimators=10, random_state=1234, verbose=-1
        ),
        lags=3,
    )
    fc.fit(y)

with tempfile.TemporaryDirectory() as tmpdir:
    _ = save_forecasters({"power": fc}, tmpdir)
    forecasters, missing = load_forecasters(
        ["power", "energy"], tmpdir, verbose=False
    )
    assert "power" in forecasters
    assert "energy" in missing
    assert isinstance(forecasters["power"], ForecasterRecursive)
    print("power" in forecasters)
    print("energy" in missing)
    print(type(forecasters["power"]).__name__)

with tempfile.TemporaryDirectory() as tmpdir:
    forecasters, missing = load_forecasters(["nonexistent"], tmpdir)
    assert len(forecasters) == 0
    assert len(missing) == 1
    print(len(forecasters), len(missing))

True
True
ForecasterRecursive
0 1

model_directory_exists

manager.persistence.model_directory_exists(model_dir)

Check if model directory exists.

Parameters

Name	Type	Description	Default
model_dir	Union[str, Path]	Directory path to check.	required

Returns

Name	Type	Description
bool	bool	True if directory exists, False otherwise.

Examples

import tempfile
from pathlib import Path
from spotforecast2_safe.manager.persistence import model_directory_exists

with tempfile.TemporaryDirectory() as tmpdir:
    assert model_directory_exists(tmpdir)
    print(model_directory_exists(tmpdir))

assert not model_directory_exists("/nonexistent/path/to/models")
print(model_directory_exists("/nonexistent/path/to/models"))

True
False

save_forecaster

manager.persistence.save_forecaster(
    forecaster,
    model_dir,
    target,
    task_name='',
    verbose=False,
)

Save a single trained forecaster to disk using joblib.

Public single-model counterpart to save_forecasters(). When task_name is provided the file is named {task_name}_{target}.joblib; otherwise the standard convention forecaster_{target}.joblib (identical to get_model_filepath()) is used.

Parameters

Name	Type	Description	Default
forecaster	object	Trained forecaster object (any joblib-serialisable model).	required
model_dir	Union[str, Path]	Directory to save the model. Created if it doesn’t exist.	required
target	str	Target variable name used in the filename.	required
task_name	str	Optional task identifier prepended to the filename (e.g. `"task_1_lazy"`). Defaults to `""` (standard naming).	`''`
verbose	bool	Print a confirmation message. Default: `False`.	`False`

Returns

Name	Type	Description
Path	Path	Full filepath of the saved model.

Raises

Name	Type	Description
	OSError	If the model cannot be written to disk.

Examples

import tempfile
from pathlib import Path
from sklearn.linear_model import LinearRegression
from spotforecast2_safe.manager.persistence import save_forecaster

model = LinearRegression()

with tempfile.TemporaryDirectory() as tmpdir:
    path = save_forecaster(model, tmpdir, "power")
    assert path.name == "forecaster_power.joblib"
    print(path.name)

with tempfile.TemporaryDirectory() as tmpdir:
    path = save_forecaster(model, tmpdir, "power", task_name="task_1")
    assert path.name == "task_1_power.joblib"
    print(path.name)

forecaster_power.joblib
task_1_power.joblib

save_forecasters

manager.persistence.save_forecasters(forecasters, model_dir, verbose=False)

Save trained forecasters to disk using joblib.

Follows scikit-learn persistence conventions using joblib for efficient serialization of sklearn-compatible estimators.

Parameters

Name	Type	Description	Default
forecasters	Dict[str, object]	Dictionary mapping target names to trained ForecasterEquivalentDate objects.	required
model_dir	Union[str, Path]	Directory to save models. Created if it doesn’t exist.	required
verbose	bool	Print progress messages. Default: False.	`False`

Returns

Name	Type	Description
	Dict[str, Path]	Dict[str, Path]: Dictionary mapping target names to saved model filepaths.

Raises

Name	Type	Description
	OSError	If models cannot be written to disk.
	TypeError	If forecasters contain non-serializable objects.

Examples

import tempfile
import warnings
import numpy as np
import pandas as pd
from lightgbm import LGBMRegressor
from spotforecast2_safe.forecaster.recursive._forecaster_recursive import (
    ForecasterRecursive,
)
from spotforecast2_safe.manager.persistence import save_forecasters

rng = np.random.default_rng(0)
y = pd.Series(rng.normal(size=50), name="target")
with warnings.catch_warnings():
    warnings.simplefilter("ignore")
    fc = ForecasterRecursive(
        estimator=LGBMRegressor(
            n_estimators=10, random_state=1234, verbose=-1
        ),
        lags=3,
    )
    fc.fit(y)

with tempfile.TemporaryDirectory() as tmpdir:
    paths = save_forecasters({"power": fc, "energy": fc}, tmpdir, verbose=False)
    assert "power" in paths
    assert paths["power"].exists()
    assert paths["power"].suffix == ".joblib"
    print("power" in paths)
    print(paths["power"].exists())
    print(paths["power"].suffix)

True
True
.joblib