multitask.guards.assert_no_leakage

multitask.guards.assert_no_leakage(mt, forbidden, *, task)

Raise LeakageError if any forbidden column reached the model.

Checks three surfaces independently:

Training frame — mt.data_with_exog.columns
Selected exogenous features — mt.exog_feature_names
Fitted model features — the fitted estimator’s input feature names (LightGBM feature_name_, CatBoost feature_names_, or the sklearn-standard feature_names_in_ / XGBoost booster) for every target in mt.run_state.targets

All three surfaces are read before anything is raised, so a single call reports every violation at once and the guard fails closed:

A forbidden column on any readable surface raises LeakageError, even when another surface could not be read.
Otherwise an unreadable surface raises RuntimeError rather than silently passing: fitted features that cannot be read (estimator missing, not fitted, etc.), an empty mt.run_state.targets, or mt.data_with_exog / mt.exog_feature_names that is missing or None. An empty list of exogenous features is a readable surface.

Parameters

Name	Type	Description	Default
mt	object	A duck-typed pipeline object exposing the protocol described in the module docstring. Any `BaseTask` subclass after training satisfies it.	required
forbidden	set[str]	Set of column names that must not appear in any model surface. The operator constructs this set based on the pipeline mode (e.g. `{"Forecasted Load", "Actual Load"}` for a combined variant, plus per-zone forecast columns for the four-zone variant).	required
task	str	Key identifying the task result to inspect inside `mt.results` (e.g. `"spotoptim"` or `"defaults"`).	required

Raises

Name	Type	Description
	LeakageError	When any forbidden column appears in the training frame, the selected exogenous feature names, or any fitted model’s feature list. The message names the surface and the offending columns.
	RuntimeError	When no readable surface leaks but a surface cannot be read: `mt.results[task][target]["forecaster"].estimator` is missing or exposes none of the supported fitted-feature-name attributes (see `_fitted_feature_names`), `mt.run_state.targets` is empty, or `mt.data_with_exog` / `mt.exog_feature_names` is missing or `None`. The message names every unreadable surface, fitted features first. This signals a verifiability invariant violation rather than a leakage violation.

Examples

from types import SimpleNamespace
from spotforecast2_safe.multitask.guards import assert_no_leakage
from spotforecast2_safe.exceptions import LeakageError
import pandas as pd

# Build a minimal duck-typed stub that passes the guard.
run_state = SimpleNamespace(targets=["load"])
estimator = SimpleNamespace(feature_name_=["lag_1", "lag_24", "hour_sin"])
forecaster = SimpleNamespace(estimator=estimator)
results = {"defaults": {"load": {"forecaster": forecaster}}}
idx = pd.date_range("2026-06-01", periods=3, freq="h", tz="UTC")
data_with_exog = pd.DataFrame(
    {"load": [1.0, 2.0, 3.0], "lag_1": [0.0, 1.0, 2.0]}, index=idx
)
mt = SimpleNamespace(
    run_state=run_state,
    results=results,
    data_with_exog=data_with_exog,
    exog_feature_names=["lag_1"],
)
assert_no_leakage(mt, forbidden={"Forecasted Load"}, task="defaults")
print("clean mt: leakage guard passed")

# Inject a forbidden column into the training frame -> raises.
mt_leak = SimpleNamespace(
    run_state=run_state,
    results=results,
    data_with_exog=pd.DataFrame(
        {"load": [1.0], "Forecasted Load": [50000.0]},
        index=idx[:1],
    ),
    exog_feature_names=["lag_1"],
)
try:
    assert_no_leakage(mt_leak, forbidden={"Forecasted Load"}, task="defaults")
except LeakageError as exc:
    print(exc)

clean mt: leakage guard passed
Leakage detected in task 'defaults': forbidden column(s) ['Forecasted Load'] found in training frame. Refusing to proceed (data-governance invariant).