def check_search_error_score_handling(
search_cv,
y: pl.DataFrame,
X_actual: pl.DataFrame | None = None,
forecasting_horizon: int = 3,
X_future: pl.DataFrame | None = None,
X_forecast: pl.DataFrame | None = None,
) -> None:
"""Check error_score parameter handles failing fits correctly.
Parameters
----------
search_cv : BaseSearchCV
Unfitted search CV instance with error_score=np.nan
y : pl.DataFrame
Training target data
X_actual : pl.DataFrame, optional
Training features
forecasting_horizon : int, default=3
Number of steps ahead to forecast
Raises
------
AssertionError
If error scores are not handled correctly
"""
search_cv_clone = clone(search_cv)
# Inject one guaranteed-failing candidate alongside the existing valid ones.
# The sentinel is an out-of-domain value that the forecaster's parameter
# constraints reject during _validate_params (raised inside the candidate
# fit, which is exactly what error_score absorbs). set_params accepts any
# value, so the failure surfaces at fit time rather than before it. Keeping
# the valid candidates avoids the all-candidates-failed path, which raises
# regardless of error_score.
_invalid = "__yohou_invalid_param_value__"
if hasattr(search_cv_clone, "param_grid"):
grid = search_cv_clone.param_grid
valid_dicts = [dict(d) for d in grid] if isinstance(grid, list) else [dict(grid)]
failing_key = next(iter(valid_dicts[0].keys()))
failing_grid = [*valid_dicts, {failing_key: [_invalid]}]
search_cv_clone.set_params(param_grid=failing_grid)
elif hasattr(search_cv_clone, "param_distributions"):
dist = search_cv_clone.param_distributions
valid_dists = [dict(d) for d in dist] if isinstance(dist, list) else [dict(dist)]
failing_key = next(iter(valid_dists[0].keys()))
# Materialize a fixed set of concrete valid candidates from the original
# distributions, then append the failing one and enumerate the whole
# finite grid. Sampling from the live distributions could otherwise miss
# the single failing candidate entirely.
n_valid = max(1, getattr(search_cv_clone, "n_iter", 1))
concrete_valid = list(
ParameterSampler(valid_dists, n_iter=n_valid, random_state=getattr(search_cv_clone, "random_state", None))
)
# Each sampled candidate becomes a one-value finite grid so the whole
# space (valid samples + failing candidate) can be enumerated exactly.
finite_grids = [{key: [value] for key, value in candidate.items()} for candidate in concrete_valid]
failing_dist = [*finite_grids, {failing_key: [_invalid]}]
search_cv_clone.set_params(param_distributions=failing_dist, n_iter=len(failing_dist))
else: # pragma: no cover - the check suite only runs on searches that expose a tunable space
# No tunable search space to corrupt; nothing to assert.
return
# Set error_score to np.nan (don't raise)
search_cv_clone.set_params(error_score=np.nan)
search_cv_clone.fit(y, X_actual, forecasting_horizon=forecasting_horizon, X_future=X_future, X_forecast=X_forecast)
# Check that fit completed without raising despite the failing candidate.
assert hasattr(search_cv_clone, "cv_results_"), (
"fit() should complete with error_score=np.nan even when some fits fail"
)
# The failing candidate must be recorded as np.nan, not silently dropped or
# scored as a real value. Single-metric searches expose 'mean_test_score';
# multimetric searches expose one 'mean_test_<name>' key per scorer.
cv_results = search_cv_clone.cv_results_
mean_test_keys = [k for k in cv_results if k.startswith("mean_test_")]
assert mean_test_keys, f"cv_results_ must expose at least one mean_test_* key, got {list(cv_results)}"
for key in mean_test_keys:
mean_test_score = np.asarray(cv_results[key], dtype=float)
assert np.isnan(mean_test_score).any(), (
f"error_score=np.nan must record np.nan for the failing candidate in {key!r}, got {mean_test_score}"
)