Skip to content

Python API

PSPSO 1.0 exposes one typed Python API. Use optimize() for concise notebook work and PSPSOOptimizer when you need to retain or inspect the optimizer instance.

Notebook Helper

pspso.optimize 

optimize(
    X: Any,
    y: Any,
    *,
    estimator: str
    | EstimatorConfig
    | Callable[..., Any] = "svm",
    search_space: SearchSpace
    | Mapping[str, ParameterSpec]
    | None = None,
    config: OptimizationConfig | None = None,
    X_validation: Any | None = None,
    y_validation: Any | None = None,
    progress_callback: ProgressCallback | None = None,
    should_cancel: Callable[[], bool] | None = None,
    tracking: TrackingConfig | None = None,
) -> OptimizationResult

Optimize an estimator directly from Python or a notebook.

Tracking is disabled unless tracking is supplied. Tracked calls write to the same versioned workspace used by the dashboard and CLI, but execute in the current Python process and do not require FastAPI.

Source code in pspso/api.py 
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def optimize(
    X: Any,
    y: Any,
    *,
    estimator: str | EstimatorConfig | Callable[..., Any] = "svm",
    search_space: SearchSpace | Mapping[str, ParameterSpec] | None = None,
    config: OptimizationConfig | None = None,
    X_validation: Any | None = None,
    y_validation: Any | None = None,
    progress_callback: ProgressCallback | None = None,
    should_cancel: Callable[[], bool] | None = None,
    tracking: TrackingConfig | None = None,
) -> OptimizationResult:
    """Optimize an estimator directly from Python or a notebook.

    Tracking is disabled unless ``tracking`` is supplied. Tracked calls write to
    the same versioned workspace used by the dashboard and CLI, but execute in
    the current Python process and do not require FastAPI.
    """

    optimizer = PSPSOOptimizer(estimator, search_space, config)
    if tracking is None:
        return optimizer.optimize(
            X,
            y,
            X_validation,
            y_validation,
            progress_callback,
            should_cancel=should_cancel,
        )

    repository = TrackingRepository(_tracking_database(tracking.workspace))
    experiment = _resolve_experiment(repository, tracking)
    spec = _python_run_spec(optimizer, X, y, X_validation, y_validation, tracking)
    snapshot = repository.create_run(spec, experiment_id=experiment["experiment_id"])
    run_id = snapshot["run_id"]
    attempt = repository.create_attempt(run_id)
    repository.set_attempt_running(attempt["attempt_id"], os.getpid())

    def tracked_callback(event: ProgressEvent) -> None:
        repository.record_event(run_id, event.to_dict())
        if progress_callback is not None:
            progress_callback(event)

    try:
        result = optimizer.optimize(
            X,
            y,
            X_validation,
            y_validation,
            tracked_callback,
            should_cancel=should_cancel,
            defer_terminal_event=True,
        )
        result.run_id = run_id
        result.experiment_id = experiment["experiment_id"]
        artifact_store = ArtifactStore(repository.workspace)
        artifacts = artifact_store.save_run(
            run_id,
            spec,
            result,
            provenance={
                "dataset_fingerprint": spec["dataset"]["fingerprint"],
                "random_seed": optimizer.config.random_state,
                "seeds": {"optimization": optimizer.config.random_state},
            },
        )
        for warning in artifact_store.last_warnings:
            tracked_callback(
                ProgressEvent(
                    "run_warning",
                    {
                        "reason": "artifact_serialization",
                        "message": warning,
                    },
                )
            )
        if tracking.snapshot_data:
            data_path = artifact_store.run_path(run_id) / "dataset.joblib"
            joblib.dump(
                {
                    "X": X,
                    "y": y,
                    "X_validation": X_validation,
                    "y_validation": y_validation,
                },
                data_path,
            )
            artifacts["dataset_snapshot"] = str(data_path)
            manifest = artifact_store.run_path(run_id) / "manifest.json"
            manifest.write_text(json.dumps(artifacts, indent=2), encoding="utf-8")
        result.artifacts = artifacts
        artifact_store.save_result(run_id, result)
        repository.save_result(run_id, result)
        repository.save_artifacts(run_id, artifacts)
        tracked_callback(optimizer.terminal_event)
        status = "completed" if optimizer.terminal_event.type == "run_completed" else "failed"
        repository.finish_attempt(attempt["attempt_id"], status)
        return result
    except (Exception, KeyboardInterrupt) as exc:
        repository.save_error(run_id, str(exc))
        status = (
            "cancelled" if isinstance(exc, (OptimizationCancelled, KeyboardInterrupt)) else "failed"
        )
        category = (
            "cancellation"
            if status == "cancelled"
            else "timeout"
            if isinstance(exc, OptimizationTimedOut)
            else optimizer._failure_category(exc)
            if isinstance(exc, Exception)
            else "cancellation"
        )
        repository.record_event(
            run_id,
            ProgressEvent(f"run_{status}", {"error": str(exc), "reason": category}).to_dict(),
        )
        repository.finish_attempt(attempt["attempt_id"], status)
        raise

pspso.TrackingConfig dataclass 

TrackingConfig(
    workspace: str | Path | None = None,
    experiment_name: str = "Notebook experiment",
    experiment_id: str | None = None,
    run_name: str | None = None,
    tags: tuple[str, ...] = tuple(),
    snapshot_data: bool = False,
)

Opt-in persistence settings for an in-process Python optimization.

Parameters:

Name Type Description Default
workspace str | Path | None

Directory containing the 1.0 tracking database and artifacts.

 None
experiment_name str

Name used when creating a new experiment.

 'Notebook experiment'
experiment_id str | None

Existing experiment to receive the run.

 None
run_name str | None

Optional human-readable label stored in the run specification.

 None
tags tuple[str, ...]

Tags applied when a new experiment is created.

 tuple()
snapshot_data bool

Save the supplied training and validation objects with the run.

 False

Optimizer

pspso.PSPSOOptimizer 

PSPSOOptimizer(
    estimator: str
    | EstimatorConfig
    | Callable[..., Any] = "svm",
    search_space: SearchSpace
    | Mapping[str, ParameterSpec]
    | None = None,
    config: OptimizationConfig | None = None,
)

Hyperparameter optimizer with PSO, grid, and random strategies.

Source code in pspso/optimizer.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def __init__(
    self,
    estimator: str | EstimatorConfig | Callable[..., Any] = "svm",
    search_space: SearchSpace | Mapping[str, ParameterSpec] | None = None,
    config: OptimizationConfig | None = None,
) -> None:
    self.config = config or OptimizationConfig()
    if self.config.random_state is None:
        self.config = replace(self.config, random_state=secrets.randbits(32))
    self.config.validate()
    if isinstance(estimator, EstimatorConfig):
        self.estimator_config = estimator
    elif isinstance(estimator, str):
        self.estimator_config = EstimatorConfig(
            name=estimator,
            fixed_params=default_fixed_params(estimator, self.config.task),
        )
    else:
        self.estimator_config = EstimatorConfig(factory=estimator)
    self.estimator_config.validate()
    if search_space is None:
        if self.estimator_config.name is None:
            raise ValueError("search_space is required for custom estimator factories.")
        search_space = default_search_space(self.estimator_config.name, self.config.task)
    self.search_space = (
        search_space if isinstance(search_space, SearchSpace) else SearchSpace(search_space)
    )
    self.result: OptimizationResult | None = None
    self.best_params: dict[str, Any] | None = None
    self.best_position: list[float] | None = None
    self.best_cost: float | None = None
    self.best_metric: float | None = None
    self.best_model: Any = None
    self.trials: list[TrialResult] = []
    self.failures: list[str] = []
    self.optimizer_state: dict[str, Any] = {}
    self._state_lock = Lock()
    self._next_trial_id = 1
    self._best_trial: TrialResult | None = None
    self.prepared_bundle: dict[str, Any] | None = None
    self.dataset_frame: Any = None
    self._worker_slots: SimpleQueue[int] = SimpleQueue()

optimize 

optimize(
    X: Any,
    y: Any,
    X_val: Any | None = None,
    y_val: Any | None = None,
    progress_callback: ProgressCallback | None = None,
    should_cancel: Callable[[], bool] | None = None,
    defer_terminal_event: bool = False,
) -> OptimizationResult

Run the configured hyperparameter optimization.

Source code in pspso/optimizer.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def optimize(
    self,
    X: Any,
    y: Any,
    X_val: Any | None = None,
    y_val: Any | None = None,
    progress_callback: ProgressCallback | None = None,
    should_cancel: Callable[[], bool] | None = None,
    defer_terminal_event: bool = False,
) -> OptimizationResult:
    """Run the configured hyperparameter optimization."""

    self._reset()
    X_train, X_validation, y_train, y_validation = self._prepare_data(X, y, X_val, y_val)
    start = time.time()
    self._started_at = start
    self._should_cancel = should_cancel
    self._best_trial_index = 0
    planned_trials = self._planned_trials()
    candidate_fits = (planned_trials or 0) * (
        self.config.cv_folds if self.config.evaluation_protocol == "cross_validation" else 1
    )
    self.optimizer_state["planned_fits"] = candidate_fits + int(
        self.config.refit_best
        and (
            self.config.evaluation_protocol == "cross_validation"
            or self.prepared_bundle is not None
        )
    )
    self._emit(
        progress_callback,
        "run_started",
        {
            "strategy": self.config.strategy,
            "task": self.config.task,
            "metric": self.config.normalized_metric(),
            "dimensions": self.search_space.dimensions,
            "planned_trials": planned_trials,
            "planned_fits": self.optimizer_state["planned_fits"],
            "planned_candidate_fits": candidate_fits,
            "evaluation_protocol": self.config.evaluation_protocol,
            "cv_folds": self.config.cv_folds,
            "trial_workers": self.config.trial_workers,
            "random_seed": self.config.random_state,
            "positive_label": self.config.positive_label,
            "decision_threshold": self.config.decision_threshold,
            "execution_mode": ("parallel" if self.config.trial_workers > 1 else "sequential"),
        },
    )
    if self.config.strategy == "grid":
        self._run_grid(X_train, y_train, X_validation, y_validation, progress_callback)
    elif self.config.strategy == "random":
        self._run_random(X_train, y_train, X_validation, y_validation, progress_callback)
    elif self.config.strategy == "pso":
        self._run_pso(X_train, y_train, X_validation, y_validation, progress_callback)
    else:
        raise ValueError(f"Unsupported strategy: {self.config.strategy!r}")
    refit_error = None
    if (
        self.best_params is not None
        and self.config.evaluation_protocol == "cross_validation"
        and X_validation is None
        and self.config.refit_best
    ):
        self._emit(progress_callback, "final_refit_started", {"best_params": self.best_params})
        try:
            self._check_runtime()
            self.best_model = create_estimator(
                self.estimator_config, self.config.task, self.best_params
            )
            self._fit_model(
                self.best_model,
                X_train,
                y_train,
                progress_callback,
                phase="final_refit",
            )
            self._emit(
                progress_callback,
                "final_refit_completed",
                {"rows": int(len(y_train)), "best_params": self.best_params},
            )
        except (OptimizationCancelled, OptimizationTimedOut):
            raise
        except Exception as exc:
            refit_error = str(exc)
            self.failures.append(refit_error)
            self.best_model = None
            self._emit(
                progress_callback,
                "run_warning",
                {"reason": "final_refit_failed", "error": refit_error},
            )
    duration = time.time() - start
    self.optimizer_state.update(
        {
            "evaluation_protocol": self.config.evaluation_protocol,
            "cv_folds": (
                self.config.cv_folds
                if self.config.evaluation_protocol == "cross_validation"
                else 1
            ),
            "trial_workers": self.config.trial_workers,
        }
    )
    self.optimizer_state.update(
        {
            "random_state": self.config.random_state,
            "positive_label": self.config.positive_label,
            "decision_threshold": self.config.decision_threshold,
        }
    )
    self.result = OptimizationResult(
        best_params=self.best_params,
        best_cost=self.best_cost,
        best_metric=self.best_metric,
        model=self.best_model,
        duration=duration,
        trials=list(self.trials),
        strategy=self.config.strategy,
        task=self.config.task,
        metric_name=self.config.normalized_metric(),
        failures=list(self.failures),
        best_position=self.best_position,
        optimizer_state=dict(self.optimizer_state),
    )
    event_type = (
        "run_completed"
        if self.best_params is not None
        and (not self.config.refit_best or self.best_model is not None)
        else "run_failed"
    )
    terminal_payload = {
        "duration": duration,
        "best_cost": self.best_cost,
        "best_metric": self.best_metric,
        "best_params": self.best_params,
        "n_trials": len(self.trials),
        "n_failures": len(self.failures),
        "refit_error": refit_error,
    }
    self.terminal_event = ProgressEvent(event_type, terminal_payload)
    if not defer_terminal_event:
        self._emit(progress_callback, event_type, terminal_payload)
    return self.result

Search Space

pspso.SearchSpace 

SearchSpace(params: Mapping[str, ParameterSpec])

Ordered collection of tunable hyperparameters.

Source code in pspso/search_space.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def __init__(self, params: Mapping[str, ParameterSpec]):
    if not params:
        raise ValueError("SearchSpace requires at least one parameter.")
    invalid = [
        name
        for name, spec in params.items()
        if not isinstance(spec, (Choice, FloatRange, LogFloatRange, IntRange))
    ]
    if invalid:
        raise TypeError(
            "SearchSpace values must be Choice, IntRange, FloatRange, or LogFloatRange; "
            f"invalid parameters: {', '.join(invalid)}."
        )
    self.params = dict(params)
    self.names = list(self.params)

grid_size property 

grid_size: int

Return the exact number of combinations produced by :meth:iter_grid.

to_schema 

to_schema() -> dict[str, dict[str, Any]]

Return the canonical JSON representation used by API v1.

Source code in pspso/search_space.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
def to_schema(self) -> dict[str, dict[str, Any]]:
    """Return the canonical JSON representation used by API v1."""

    schema: dict[str, dict[str, Any]] = {}
    for name, spec in self.params.items():
        if isinstance(spec, Choice):
            schema[name] = {"type": "choice", "values": list(spec.values)}
        elif isinstance(spec, IntRange):
            schema[name] = {"type": "int", "low": spec.low, "high": spec.high}
        elif isinstance(spec, LogFloatRange):
            schema[name] = {
                "type": "log_float",
                "low": spec.low,
                "high": spec.high,
                "precision": spec.precision,
            }
        else:
            schema[name] = {
                "type": "float",
                "low": spec.low,
                "high": spec.high,
                "precision": spec.precision,
            }
    return schema

pspso.Choice dataclass 

Choice(values: Sequence[Any])

A categorical hyperparameter.

Source code in pspso/search_space.py 
19
20
21
22
def __init__(self, values: Sequence[Any]):
    if not values:
        raise ValueError("Choice requires at least one value.")
    object.__setattr__(self, "values", tuple(values))

pspso.IntRange dataclass 

IntRange(low: int, high: int)

An integer hyperparameter.

pspso.FloatRange dataclass 

FloatRange(low: float, high: float, precision: int = 2)

A continuous numeric hyperparameter rounded to a fixed precision.

pspso.LogFloatRange dataclass 

LogFloatRange(low: float, high: float, precision: int = 2)

Bases: FloatRange

Positive float range sampled uniformly in log space.

Config And Results

pspso.OptimizationConfig dataclass 

OptimizationConfig(
    task: TaskName = "regression",
    metric: MetricName = "rmse",
    strategy: StrategyName = "pso",
    validation_size: float = 0.2,
    test_size: float = 0.0,
    random_state: int | None = None,
    n_particles: int = 5,
    n_iterations: int = 10,
    pso_options: Mapping[str, float] = (
        lambda: {"c1": 1.49618, "c2": 1.49618, "w": 0.7298}
    )(),
    pso_topology: Literal["global", "local"] = "global",
    max_trials: int | None = None,
    timeout_seconds: float | None = None,
    early_stopping_rounds: int | None = None,
    evaluation_protocol: EvaluationProtocol = "holdout",
    cv_folds: int = 5,
    shuffle_folds: bool = True,
    time_series: bool = False,
    time_gap: int = 0,
    stratify: bool = True,
    positive_label: str | int | float | bool | None = None,
    decision_threshold: float = 0.5,
    refit_best: bool = True,
    trial_workers: int = 1,
    verbose: int = 0,
)

Runtime settings for an optimization run.

pspso.EstimatorConfig dataclass 

EstimatorConfig(
    name: str | None = None,
    factory: Callable[..., Any] | None = None,
    fixed_params: Mapping[str, Any] = dict(),
)

Estimator name/factory plus parameters fixed across all trials.

pspso.OptimizationResult dataclass 

OptimizationResult(
    best_params: Mapping[str, Any] | None,
    best_cost: float | None,
    best_metric: float | None,
    model: Any,
    duration: float,
    trials: list[TrialResult],
    strategy: str,
    task: str,
    metric_name: str,
    failures: list[str] = list(),
    best_position: list[float] | None = None,
    optimizer_state: dict[str, Any] = dict(),
    run_id: str | None = None,
    experiment_id: str | None = None,
    artifacts: Mapping[str, str] = dict(),
)

Final result returned by the modern optimizer.

evaluate 

evaluate(X: Any, y: Any) -> dict[str, Any]

Evaluate the fitted best model with the run's task and metric.

Source code in pspso/config.py 
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
def evaluate(self, X: Any, y: Any) -> dict[str, Any]:
    """Evaluate the fitted best model with the run's task and metric."""

    if self.model is None:
        raise RuntimeError("No successful fitted model is available.")
    from .metrics import evaluation_report

    return evaluation_report(
        self.model,
        self.task,
        self.metric_name,
        X,
        y,
        positive_label=self.optimizer_state.get("positive_label"),
        decision_threshold=self.optimizer_state.get("decision_threshold", 0.5),
    )

predict 

predict(X: Any) -> Any

Predict with the fitted best model.

Source code in pspso/config.py 
232
233
234
235
236
237
def predict(self, X: Any) -> Any:
    """Predict with the fitted best model."""

    if self.model is None:
        raise RuntimeError("No successful fitted model is available.")
    return self.model.predict(X)

predict_proba 

predict_proba(X: Any) -> Any

Return probabilities when the fitted best model supports them.

Source code in pspso/config.py 
239
240
241
242
243
244
245
246
def predict_proba(self, X: Any) -> Any:
    """Return probabilities when the fitted best model supports them."""

    if self.model is None:
        raise RuntimeError("No successful fitted model is available.")
    if not hasattr(self.model, "predict_proba"):
        raise AttributeError("The fitted best model does not support predict_proba().")
    return self.model.predict_proba(X)

summary 

summary() -> str

Return a compact human-readable summary of the optimization.

Source code in pspso/config.py 
185
186
187
188
189
190
191
192
193
194
def summary(self) -> str:
    """Return a compact human-readable summary of the optimization."""

    metric = "n/a" if self.best_metric is None else f"{self.best_metric:.6g}"
    cost = "n/a" if self.best_cost is None else f"{self.best_cost:.6g}"
    return (
        f"{self.strategy.upper()} {self.task}: {self.metric_name}={metric}, "
        f"cost={cost}, trials={len(self.trials)}, failures={len(self.failures)}, "
        f"duration={self.duration:.3f}s, best_params={dict(self.best_params or {})}"
    )

trials_as_rows 

trials_as_rows() -> list[dict[str, Any]]

Return JSON-serializable trial dictionaries.

Source code in pspso/config.py 
286
287
288
289
def trials_as_rows(self) -> list[dict[str, Any]]:
    """Return JSON-serializable trial dictionaries."""

    return [trial.to_dict() for trial in self.trials]

trials_frame 

trials_frame()

Return trial results as a pandas DataFrame.

Source code in pspso/config.py 
219
220
221
222
223
224
225
226
227
228
229
230
def trials_frame(self):
    """Return trial results as a pandas DataFrame."""

    import pandas as pd

    rows = []
    for trial in self.trials:
        row = trial.to_dict()
        params = row.pop("params")
        row.update({f"param_{name}": value for name, value in params.items()})
        rows.append(row)
    return pd.DataFrame(rows)

pspso.TrialResult dataclass 

TrialResult(
    trial_id: int,
    params: Mapping[str, Any],
    cost: float | None,
    metric: float | None,
    train_metric: float | None,
    status: Literal["completed", "failed"],
    duration: float,
    error: str | None = None,
    iteration: int | None = None,
    particle_index: int | None = None,
    strategy_slot: int | None = None,
    worker_slot: int | None = None,
    validation_metrics: Mapping[str, float | None] = dict(),
    train_metrics: Mapping[str, float | None] = dict(),
    metric_std: float | None = None,
    fold_metrics: list[dict[str, Any]] = list(),
    failure_category: str | None = None,
    model: Any = None,
)

A single evaluated parameter set.

Model Discovery

pspso.list_estimators 

list_estimators(task: str | None = None) -> list[str]

List canonical estimator IDs, optionally filtered by task.

Source code in pspso/estimators.py 
262
263
264
265
266
267
268
def list_estimators(task: str | None = None) -> list[str]:
    """List canonical estimator IDs, optionally filtered by task."""

    presets = estimator_presets()
    return sorted(
        name for name, preset in presets.items() if task is None or task in preset["tasks"]
    )

pspso.get_estimator_info 

get_estimator_info(name: str) -> dict[str, Any]

Return capabilities, dependency status, and typed defaults for an estimator.

Source code in pspso/estimators.py 
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
def get_estimator_info(name: str) -> dict[str, Any]:
    """Return capabilities, dependency status, and typed defaults for an estimator."""

    canonical = canonical_estimator_name(name)
    try:
        preset = dict(estimator_presets()[canonical])
    except KeyError as exc:
        raise ValueError(f"Unknown estimator preset: {name!r}.") from exc
    preset["id"] = canonical
    preset["dependency"] = optional_dependency_status(preset.get("optional_dependency"))
    preset["defaults"] = {
        task: {
            "fixed_params": default_fixed_params(canonical, task),
            "search_space": default_search_space(canonical, task),
            "allowed_params": sorted(allowed_estimator_params(canonical, task)),
        }
        for task in preset["tasks"]
    }
    return preset

Recipe Registration

pspso.register_recipe 

register_recipe(
    name: str,
    factory: Callable[..., Any],
    *,
    tasks: list[str],
    description: str,
    fixed_params: Mapping[str, Mapping[str, Any]]
    | None = None,
    search_spaces: Mapping[
        str, SearchSpace | Mapping[str, ParameterSpec]
    ]
    | None = None,
) -> None

Register a trusted recipe with typed task-specific search spaces.

Source code in pspso/recipes.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def register_recipe(
    name: str,
    factory: Callable[..., Any],
    *,
    tasks: list[str],
    description: str,
    fixed_params: Mapping[str, Mapping[str, Any]] | None = None,
    search_spaces: Mapping[str, SearchSpace | Mapping[str, ParameterSpec]] | None = None,
) -> None:
    """Register a trusted recipe with typed task-specific search spaces."""

    if not name.replace("_", "").replace("-", "").isalnum():
        raise ValueError(
            "Recipe names may contain only letters, numbers, hyphens, and underscores."
        )
    _RECIPES[name.lower()] = Recipe(
        name=name.lower(),
        factory=factory,
        tasks=tuple(tasks),
        description=description,
        fixed_params=fixed_params or {},
        search_spaces={
            task: space if isinstance(space, SearchSpace) else SearchSpace(space)
            for task, space in (search_spaces or {}).items()
        },
    )