API Reference

This reference is automatically generated from the docstrings in the source code.

Only the symbols listed in Stability and versioning and frozen under tests/compatibility/v1_0_0/ are stable for the 1.x series. Other importable or documented modules are experimental or internal as labelled.

Unified API

The primary user-facing entry points. Most users only need these.

vamos.experiment.unified

Unified API for VAMOS optimization.

This module provides a single, powerful entry point that consolidates problem-based runs, study-style configuration, and auto-parameter defaults into one flexible function.

optimize(problem, *, algorithm='auto', max_evaluations=None, termination=None, pop_size=None, engine=None, seed=DEFAULT_SEED, verbose=False, n_var=None, n_obj=None, problem_kwargs=None, algorithm_config=None, eval_strategy=None, live_viz=None, checkpoint=None)

optimize(problem: str | ProblemProtocol, *, algorithm: AlgorithmName | str = 'auto', max_evaluations: int | None = None, termination: TerminationSpec | None = None, pop_size: int | None = None, engine: EngineName | str | None = None, seed: int | None = 42, verbose: bool = False, n_var: int | None = None, n_obj: int | None = None, problem_kwargs: Mapping[str, object] | None = None, algorithm_config: AlgorithmConfigProtocol | None = None, eval_strategy: EvaluationBackend | str | None = None, live_viz: LiveVisualization | None = None, checkpoint: CheckpointPayload | None = None) -> OptimizationResult

optimize(problem: str | ProblemProtocol, *, algorithm: AlgorithmName | str = 'auto', max_evaluations: int | None = None, termination: TerminationSpec | None = None, pop_size: int | None = None, engine: EngineName | str | None = None, seed: list[int] | tuple[int, ...], verbose: bool = False, n_var: int | None = None, n_obj: int | None = None, problem_kwargs: Mapping[str, object] | None = None, algorithm_config: AlgorithmConfigProtocol | None = None, eval_strategy: EvaluationBackend | str | None = None, live_viz: LiveVisualization | None = None, checkpoint: CheckpointPayload | None = None) -> StudyResult

Unified entry point for VAMOS optimization.

This function consolidates multiple APIs into a single powerful interface: - Accepts problem names (strings) or instances - Supports AutoML with algorithm="auto" - Handles multi-run studies with seed=[0,1,2,...] - Prefer optimize(...) for all runs (explicit options are available).

Parameters:

Name	Type	Description	Default
problem	str | ProblemProtocol	Problem name (for registered problems) or a problem instance.	required
algorithm	AlgorithmName | str	Algorithm name or "auto" for automatic selection.	"auto"
max_evaluations	int | None	Maximum function evaluations. Auto-determined when omitted.	None
termination	TerminationSpec | None	Explicit termination pair for advanced runs that also pass algorithm_config. For example ("max_evaluations", 10000).	None
pop_size	int | None	Population size. Auto-determined when omitted.	None
engine	EngineName | str | None	Backend engine (for example "numpy", "numba", "moocore", or "auto").	None
seed	int | None | list[int] | tuple[int, ...]	Random seed for one run, None to generate and record a seed before execution, or a sequence of explicit seeds for multi-run studies.	``42``
verbose	bool	Enable VAMOS logging for the run.	``False``
n_var	int | None	Override problem dimensions when using a registered string problem key.	None
n_obj	int | None	Override problem dimensions when using a registered string problem key.	None
problem_kwargs	Mapping[str, object] | None	Extra keyword arguments forwarded to problem instantiation.	None
algorithm_config	AlgorithmConfigProtocol | None	Explicit algorithm config object.	None
eval_strategy	EvaluationBackend | str | None	Evaluation backend name or backend instance.	None
live_viz	LiveVisualization | None	Live visualization callback.	None
checkpoint	CheckpointPayload | None	Warm-start checkpoint for compatible algorithms. Multi-seed runs do not accept checkpoints.	None

Returns:

Type	Description
OptimizationResult | StudyResult	A single-run result for scalar seed input, or a sequence-compatible StudyResult when seed is a list/tuple.

Raises:

Type	Description
ConfigurationError	If inputs are invalid or the algorithm/engine combination is not supported.
Examples:	AutoML mode - zero config result = vamos.optimize("zdt1") Specify algorithm result = vamos.optimize("zdt1", algorithm="moead", max_evaluations=5000) Multi-seed study study = vamos.optimize("zdt1", seed=[0, 1, 2, 3, 4]) study.mean("evaluations")

Source code in src/vamos/experiment/unified.py

def optimize(
    problem: str | ProblemProtocol,
    *,
    algorithm: AlgorithmName | str = "auto",
    max_evaluations: int | None = None,
    termination: TerminationSpec | None = None,
    pop_size: int | None = None,
    engine: EngineName | str | None = None,
    seed: int | None | list[int] | tuple[int, ...] = DEFAULT_SEED,
    verbose: bool = False,
    n_var: int | None = None,
    n_obj: int | None = None,
    problem_kwargs: Mapping[str, object] | None = None,
    algorithm_config: AlgorithmConfigProtocol | None = None,
    eval_strategy: EvaluationBackend | str | None = None,
    live_viz: LiveVisualization | None = None,
    checkpoint: CheckpointPayload | None = None,
) -> OptimizationResult | StudyResult:
    """
    Unified entry point for VAMOS optimization.

    This function consolidates multiple APIs into a single powerful interface:
    - Accepts problem names (strings) or instances
    - Supports AutoML with algorithm="auto"
    - Handles multi-run studies with seed=[0,1,2,...]
    - Prefer optimize(...) for all runs (explicit options are available).

    Parameters
    ----------
    problem : str | ProblemProtocol
        Problem name (for registered problems) or a problem instance.
    algorithm : AlgorithmName | str, default "auto"
        Algorithm name or ``"auto"`` for automatic selection.
    max_evaluations : int | None, optional
        Maximum function evaluations. Auto-determined when omitted.
    termination : TerminationSpec | None, optional
        Explicit termination pair for advanced runs that also pass
        ``algorithm_config``. For example ``("max_evaluations", 10000)``.
    pop_size : int | None, optional
        Population size. Auto-determined when omitted.
    engine : EngineName | str | None, optional
        Backend engine (for example ``"numpy"``, ``"numba"``, ``"moocore"``,
        or ``"auto"``).
    seed : int | None | list[int] | tuple[int, ...], default ``42``
        Random seed for one run, ``None`` to generate and record a seed before
        execution, or a sequence of explicit seeds for multi-run studies.
    verbose : bool, default ``False``
        Enable VAMOS logging for the run.
    n_var, n_obj : int | None, optional
        Override problem dimensions when using a registered string problem key.
    problem_kwargs : Mapping[str, object] | None, optional
        Extra keyword arguments forwarded to problem instantiation.
    algorithm_config : AlgorithmConfigProtocol | None, optional
        Explicit algorithm config object.
    eval_strategy : EvaluationBackend | str | None, optional
        Evaluation backend name or backend instance.
    live_viz : LiveVisualization | None, optional
        Live visualization callback.
    checkpoint : CheckpointPayload | None, optional
        Warm-start checkpoint for compatible algorithms. Multi-seed runs do not
        accept checkpoints.

    Returns
    -------
    OptimizationResult | StudyResult
        A single-run result for scalar ``seed`` input, or a sequence-compatible
        ``StudyResult`` when ``seed`` is a list/tuple.

    Raises
    ------
    ConfigurationError
        If inputs are invalid or the algorithm/engine combination is not
        supported.

    Examples:
        # AutoML mode - zero config
        >>> result = vamos.optimize("zdt1")

        # Specify algorithm
        >>> result = vamos.optimize("zdt1", algorithm="moead", max_evaluations=5000)

        # Multi-seed study
        >>> study = vamos.optimize("zdt1", seed=[0, 1, 2, 3, 4])
        >>> study.mean("evaluations")
    """
    if isinstance(seed, (list, tuple)):
        if checkpoint is not None:
            raise ConfigurationError("checkpoint is only supported for single-seed runs.")
        if any(isinstance(value, bool) or not isinstance(value, numbers.Integral) for value in seed):
            raise ConfigurationError("seed sequences must contain only integers.")
        return StudyResult(
            [
                _run_single(
                    problem,
                    algorithm,
                    max_evaluations,
                    termination,
                    pop_size,
                    engine,
                    int(single_seed),
                    verbose,
                    n_var,
                    n_obj,
                    problem_kwargs,
                    algorithm_config,
                    eval_strategy,
                    live_viz,
                    checkpoint,
                )
                for single_seed in seed
            ]
        )

    # Single run
    return _run_single(
        problem,
        algorithm,
        max_evaluations,
        termination,
        pop_size,
        engine,
        seed,
        verbose,
        n_var,
        n_obj,
        problem_kwargs,
        algorithm_config,
        eval_strategy,
        live_viz,
        checkpoint,
    )

Problem Definition

vamos.foundation.problem.base

Base class for class-based custom optimization problems.

Problem

Base class for class-based custom optimization problems.

Subclass this when your problem needs state — a dataset, distance matrix, simulator, or any data set up in __init__.

Required: set n_var, n_obj, xl, xu in __init__. Optional: override encoding and n_constraints as class-level attributes (not in __init__).

Example — unconstrained::

import numpy as np
from vamos import Problem, optimize

class MyProblem(Problem):
    def __init__(self):
        self.n_var = 3
        self.n_obj = 2
        self.xl = np.zeros(3)
        self.xu = np.ones(3)

    def objectives(self, X: np.ndarray) -> np.ndarray:
        # X: (N, n_var) batch of candidate solutions
        f1 = np.sum(X ** 2, axis=1)
        f2 = np.sum((X - 1) ** 2, axis=1)
        return np.column_stack([f1, f2])

result = optimize(MyProblem(), algorithm="nsgaii", max_evaluations=5000)

Example — constrained::

class MyConstrainedProblem(Problem):
    n_constraints = 1          # declare at class level

    def __init__(self):
        self.n_var = 3
        self.n_obj = 2
        self.xl = np.zeros(3)
        self.xu = np.ones(3)

    def objectives(self, X):
        f1 = np.sum(X ** 2, axis=1)
        f2 = np.sum((X - 1) ** 2, axis=1)
        return np.column_stack([f1, f2])

    def constraints(self, X):
        # Sign convention: g(x) <= 0 means feasible.
        g = np.sum(X, axis=1) - 2.0   # sum(x) <= 2
        return g.reshape(-1, 1)

Source code in src/vamos/foundation/problem/base.py

class Problem:
    """Base class for class-based custom optimization problems.

    Subclass this when your problem needs state — a dataset, distance matrix,
    simulator, or any data set up in ``__init__``.

    **Required:** set ``n_var``, ``n_obj``, ``xl``, ``xu`` in ``__init__``.
    **Optional:** override ``encoding`` and ``n_constraints`` as class-level
    attributes (not in ``__init__``).

    Example — unconstrained::

        import numpy as np
        from vamos import Problem, optimize

        class MyProblem(Problem):
            def __init__(self):
                self.n_var = 3
                self.n_obj = 2
                self.xl = np.zeros(3)
                self.xu = np.ones(3)

            def objectives(self, X: np.ndarray) -> np.ndarray:
                # X: (N, n_var) batch of candidate solutions
                f1 = np.sum(X ** 2, axis=1)
                f2 = np.sum((X - 1) ** 2, axis=1)
                return np.column_stack([f1, f2])

        result = optimize(MyProblem(), algorithm="nsgaii", max_evaluations=5000)

    Example — constrained::

        class MyConstrainedProblem(Problem):
            n_constraints = 1          # declare at class level

            def __init__(self):
                self.n_var = 3
                self.n_obj = 2
                self.xl = np.zeros(3)
                self.xu = np.ones(3)

            def objectives(self, X):
                f1 = np.sum(X ** 2, axis=1)
                f2 = np.sum((X - 1) ** 2, axis=1)
                return np.column_stack([f1, f2])

            def constraints(self, X):
                # Sign convention: g(x) <= 0 means feasible.
                g = np.sum(X, axis=1) - 2.0   # sum(x) <= 2
                return g.reshape(-1, 1)
    """

    # ------------------------------------------------------------------
    # Required instance attributes — must be set in subclass __init__
    # ------------------------------------------------------------------

    n_var: int
    n_obj: int
    xl: float | int | np.ndarray
    xu: float | int | np.ndarray

    # ------------------------------------------------------------------
    # Class-level defaults — override at class body level, not in __init__
    # ------------------------------------------------------------------

    encoding: EncodingLike = "real"
    """Variable encoding.  Supported values: ``"real"``, ``"integer"``,
    ``"binary"``, ``"permutation"``, ``"mixed"``.  Default: ``"real"``."""

    n_constraints: int = 0
    """Number of inequality constraints.  Default: ``0`` (unconstrained)."""

    # ------------------------------------------------------------------
    # User-overridable interface
    # ------------------------------------------------------------------

    def objectives(self, X: np.ndarray) -> np.ndarray:
        """Compute objective values for a batch of solutions.

        Override this method in your subclass.

        Parameters
        ----------
        X : np.ndarray
            Decision matrix of shape ``(N, n_var)``.

        Returns
        -------
        np.ndarray
            Objective array of shape ``(N, n_obj)`` to minimize. A
            single-objective problem may return a 1-D array of length ``N``.
        """
        raise NotImplementedError(f"{type(self).__name__} must implement objectives(self, X).")

    def constraints(self, X: np.ndarray) -> np.ndarray | None:
        """Compute constraint violations for a batch of solutions.

        Override this method when your problem has inequality constraints.

        Parameters
        ----------
        X : np.ndarray
            Decision matrix of shape ``(N, n_var)``.

        Returns
        -------
        np.ndarray | None
            Constraint array of shape ``(N, n_constraints)`` where negative
            values indicate feasibility, or ``None`` for unconstrained problems.
        """
        return None

    # ------------------------------------------------------------------
    # Framework entry point — do not override
    # ------------------------------------------------------------------

    def evaluate(self, X: np.ndarray, out: dict[str, np.ndarray]) -> None:
        """Framework evaluation entry point.  Override :meth:`objectives`
        (and optionally :meth:`constraints`) instead of this method."""
        X = _coerce_decision_matrix(X, self.encoding)
        if X.ndim != 2 or X.shape[1] != self.n_var:
            raise ValueError(f"Expected decision matrix of shape (N, {self.n_var}), got {X.shape}.")

        # --- objectives ---
        N = X.shape[0]
        F_computed = np.asarray(self.objectives(X), dtype=float)
        if F_computed.ndim == 0:
            F_computed = F_computed.reshape(1, 1)
        elif F_computed.ndim == 1:
            F_computed = F_computed.reshape(-1, self.n_obj)
        if F_computed.shape != (N, self.n_obj):
            raise ValueError(f"{type(self).__name__}.objectives() returned shape {F_computed.shape}, expected ({N}, {self.n_obj}).")
        F_buf = out.get("F")
        if F_buf is not None and F_buf.shape == F_computed.shape:
            F_buf[:] = F_computed
        else:
            out["F"] = F_computed

        # --- constraints ---
        if self.n_constraints > 0:
            G_computed = self.constraints(X)
            if G_computed is None:
                raise ValueError(
                    f"{type(self).__name__}.constraints() returned None but "
                    f"n_constraints={self.n_constraints}. Override constraints() "
                    "to return an array of shape (N, n_constraints)."
                )
            G_computed = np.asarray(G_computed, dtype=float)
            if G_computed.ndim == 0:
                G_computed = G_computed.reshape(1, 1)
            elif G_computed.ndim == 1:
                G_computed = G_computed.reshape(-1, self.n_constraints)
            if G_computed.shape != (N, self.n_constraints):
                raise ValueError(
                    f"{type(self).__name__}.constraints() returned shape {G_computed.shape}, expected ({N}, {self.n_constraints})."
                )
            G_buf = out.get("G")
            if G_buf is not None and G_buf.shape == G_computed.shape:
                G_buf[:] = G_computed
            else:
                out["G"] = G_computed

encoding = 'real' class-attribute instance-attribute

Variable encoding. Supported values: "real", "integer", "binary", "permutation", "mixed". Default: "real".

n_constraints = 0 class-attribute instance-attribute

Number of inequality constraints. Default: 0 (unconstrained).

constraints(X)

Compute constraint violations for a batch of solutions.

Override this method when your problem has inequality constraints.

Parameters:

Name	Type	Description	Default
X	ndarray	Decision matrix of shape (N, n_var).	required

Returns:

Type	Description
ndarray | None	Constraint array of shape (N, n_constraints) where negative values indicate feasibility, or None for unconstrained problems.

Source code in src/vamos/foundation/problem/base.py

def constraints(self, X: np.ndarray) -> np.ndarray | None:
    """Compute constraint violations for a batch of solutions.

    Override this method when your problem has inequality constraints.

    Parameters
    ----------
    X : np.ndarray
        Decision matrix of shape ``(N, n_var)``.

    Returns
    -------
    np.ndarray | None
        Constraint array of shape ``(N, n_constraints)`` where negative
        values indicate feasibility, or ``None`` for unconstrained problems.
    """
    return None

evaluate(X, out)

Framework evaluation entry point. Override :meth:objectives (and optionally :meth:constraints) instead of this method.

Source code in src/vamos/foundation/problem/base.py

def evaluate(self, X: np.ndarray, out: dict[str, np.ndarray]) -> None:
    """Framework evaluation entry point.  Override :meth:`objectives`
    (and optionally :meth:`constraints`) instead of this method."""
    X = _coerce_decision_matrix(X, self.encoding)
    if X.ndim != 2 or X.shape[1] != self.n_var:
        raise ValueError(f"Expected decision matrix of shape (N, {self.n_var}), got {X.shape}.")

    # --- objectives ---
    N = X.shape[0]
    F_computed = np.asarray(self.objectives(X), dtype=float)
    if F_computed.ndim == 0:
        F_computed = F_computed.reshape(1, 1)
    elif F_computed.ndim == 1:
        F_computed = F_computed.reshape(-1, self.n_obj)
    if F_computed.shape != (N, self.n_obj):
        raise ValueError(f"{type(self).__name__}.objectives() returned shape {F_computed.shape}, expected ({N}, {self.n_obj}).")
    F_buf = out.get("F")
    if F_buf is not None and F_buf.shape == F_computed.shape:
        F_buf[:] = F_computed
    else:
        out["F"] = F_computed

    # --- constraints ---
    if self.n_constraints > 0:
        G_computed = self.constraints(X)
        if G_computed is None:
            raise ValueError(
                f"{type(self).__name__}.constraints() returned None but "
                f"n_constraints={self.n_constraints}. Override constraints() "
                "to return an array of shape (N, n_constraints)."
            )
        G_computed = np.asarray(G_computed, dtype=float)
        if G_computed.ndim == 0:
            G_computed = G_computed.reshape(1, 1)
        elif G_computed.ndim == 1:
            G_computed = G_computed.reshape(-1, self.n_constraints)
        if G_computed.shape != (N, self.n_constraints):
            raise ValueError(
                f"{type(self).__name__}.constraints() returned shape {G_computed.shape}, expected ({N}, {self.n_constraints})."
            )
        G_buf = out.get("G")
        if G_buf is not None and G_buf.shape == G_computed.shape:
            G_buf[:] = G_computed
        else:
            out["G"] = G_computed

objectives(X)

Compute objective values for a batch of solutions.

Override this method in your subclass.

Parameters:

Name	Type	Description	Default
X	ndarray	Decision matrix of shape (N, n_var).	required

Returns:

Type	Description
ndarray	Objective array of shape (N, n_obj) to minimize. A single-objective problem may return a 1-D array of length N.

Source code in src/vamos/foundation/problem/base.py

def objectives(self, X: np.ndarray) -> np.ndarray:
    """Compute objective values for a batch of solutions.

    Override this method in your subclass.

    Parameters
    ----------
    X : np.ndarray
        Decision matrix of shape ``(N, n_var)``.

    Returns
    -------
    np.ndarray
        Objective array of shape ``(N, n_obj)`` to minimize. A
        single-objective problem may return a 1-D array of length ``N``.
    """
    raise NotImplementedError(f"{type(self).__name__} must implement objectives(self, X).")

vamos.foundation.problem.builder

Friendly problem builder for VAMOS.

Provides make_problem() -- the simplest way to turn a plain Python function into a fully compatible VAMOS problem, ready for optimize().

Example

from vamos import make_problem, optimize problem = make_problem( ... lambda x: [x[0], 1 - x[0] ** 0.5], ... n_var=2, n_obj=2, ... bounds=[(0, 1), (0, 1)], ... encoding="real", ... ) result = optimize(problem, algorithm="nsgaii", max_evaluations=2000)

FunctionalProblem

Bases: Problem

Problem wrapper that adapts a user function to the VAMOS ProblemProtocol.

Created via :func:make_problem -- users should not instantiate this directly.

Source code in src/vamos/foundation/problem/builder.py

class FunctionalProblem(Problem):
    """Problem wrapper that adapts a user function to the VAMOS ProblemProtocol.

    Created via :func:`make_problem` -- users should not instantiate this
    directly.
    """

    def __init__(
        self,
        fn: Callable[..., object],
        *,
        n_var: int,
        n_obj: int,
        xl: np.ndarray,
        xu: np.ndarray,
        encoding: EncodingLike,
        vectorized: bool,
        name: str,
        constraints_fn: Callable[..., object] | None,
        n_constraints: int,
    ) -> None:
        self.n_var = n_var
        self.n_obj = n_obj
        self.xl = xl
        self.xu = xu
        self.encoding = encoding
        self.name = name
        self._fn = fn
        self._vectorized = vectorized
        self._constraints_fn = constraints_fn
        self.n_constraints = n_constraints

    # ------------------------------------------------------------------
    # ProblemProtocol.evaluate
    # ------------------------------------------------------------------

    def evaluate(self, X: np.ndarray, out: dict[str, np.ndarray]) -> None:
        """Evaluate the objective (and optional constraint) functions."""
        X = _coerce_decision_matrix(X, self.encoding)
        if X.ndim != 2 or X.shape[1] != self.n_var:
            raise ValueError(f"Expected decision matrix of shape (N, {self.n_var}), got {X.shape}.")

        if self._vectorized:
            F_result = np.asarray(self._fn(X), dtype=float)
        else:
            # Elementwise adaptation: evaluate one solution at a time.
            results = [self._fn(X[i]) for i in range(X.shape[0])]
            F_result = np.asarray(results, dtype=float)

        N = X.shape[0]
        if F_result.ndim == 0:
            F_result = F_result.reshape(1, 1)
        elif F_result.ndim == 1:
            F_result = F_result.reshape(-1, self.n_obj)
        if F_result.shape != (N, self.n_obj):
            raise ValueError(f"make_problem fn returned shape {F_result.shape}, expected ({N}, {self.n_obj}).")

        # Write into pre-allocated buffer when available, else assign.
        F = out.get("F")
        if F is not None and F.shape == F_result.shape:
            F[:] = F_result
        else:
            out["F"] = F_result

        # ---- constraints ----
        if self._constraints_fn is not None:
            if self._vectorized:
                G_result = np.asarray(self._constraints_fn(X), dtype=float)
            else:
                g_results = [self._constraints_fn(X[i]) for i in range(X.shape[0])]
                G_result = np.asarray(g_results, dtype=float)

            if G_result.ndim == 0:
                G_result = G_result.reshape(1, 1)
            elif G_result.ndim == 1:
                G_result = G_result.reshape(-1, self.n_constraints)
            if G_result.shape != (N, self.n_constraints):
                raise ValueError(f"make_problem constraints fn returned shape {G_result.shape}, expected ({N}, {self.n_constraints}).")

            G = out.get("G")
            if G is not None and G.shape == G_result.shape:
                G[:] = G_result
            else:
                out["G"] = G_result

    def __repr__(self) -> str:
        return f"FunctionalProblem(name={self.name!r}, n_var={self.n_var}, n_obj={self.n_obj})"

evaluate(X, out)

Evaluate the objective (and optional constraint) functions.

Source code in src/vamos/foundation/problem/builder.py

def evaluate(self, X: np.ndarray, out: dict[str, np.ndarray]) -> None:
    """Evaluate the objective (and optional constraint) functions."""
    X = _coerce_decision_matrix(X, self.encoding)
    if X.ndim != 2 or X.shape[1] != self.n_var:
        raise ValueError(f"Expected decision matrix of shape (N, {self.n_var}), got {X.shape}.")

    if self._vectorized:
        F_result = np.asarray(self._fn(X), dtype=float)
    else:
        # Elementwise adaptation: evaluate one solution at a time.
        results = [self._fn(X[i]) for i in range(X.shape[0])]
        F_result = np.asarray(results, dtype=float)

    N = X.shape[0]
    if F_result.ndim == 0:
        F_result = F_result.reshape(1, 1)
    elif F_result.ndim == 1:
        F_result = F_result.reshape(-1, self.n_obj)
    if F_result.shape != (N, self.n_obj):
        raise ValueError(f"make_problem fn returned shape {F_result.shape}, expected ({N}, {self.n_obj}).")

    # Write into pre-allocated buffer when available, else assign.
    F = out.get("F")
    if F is not None and F.shape == F_result.shape:
        F[:] = F_result
    else:
        out["F"] = F_result

    # ---- constraints ----
    if self._constraints_fn is not None:
        if self._vectorized:
            G_result = np.asarray(self._constraints_fn(X), dtype=float)
        else:
            g_results = [self._constraints_fn(X[i]) for i in range(X.shape[0])]
            G_result = np.asarray(g_results, dtype=float)

        if G_result.ndim == 0:
            G_result = G_result.reshape(1, 1)
        elif G_result.ndim == 1:
            G_result = G_result.reshape(-1, self.n_constraints)
        if G_result.shape != (N, self.n_constraints):
            raise ValueError(f"make_problem constraints fn returned shape {G_result.shape}, expected ({N}, {self.n_constraints}).")

        G = out.get("G")
        if G is not None and G.shape == G_result.shape:
            G[:] = G_result
        else:
            out["G"] = G_result

make_problem(fn, *, n_var, n_obj, bounds=None, xl=None, xu=None, vectorized=False, encoding, name=None, constraints=None, n_constraints=0)

Create a VAMOS-compatible problem from a plain Python function.

This is the friendliest way to define a custom optimization problem. Your function receives decision variables and returns objective values -- VAMOS handles bounds, protocol adaptation, and optional batched execution.

Parameters:

Name	Type	Description	Default
fn	callable	Objective function. Scalar mode (default, vectorized=False): receives a 1-D array of shape (n_var,) and returns a list or array of n_obj objective values. Vectorized mode (vectorized=True): receives a 2-D array of shape (N, n_var) and returns an array of shape (N, n_obj).	required
n_var	int	Number of decision variables.	required
n_obj	int	Number of objectives to minimize.	required
bounds	sequence of (lower, upper) tuples	Per-variable bounds, e.g. [(0, 1), (0, 5)]. Must have length n_var. Mutually exclusive with xl / xu.	None
xl	float or array - like	Lower / upper bounds for all variables. A scalar applies the same bound to every variable. Mutually exclusive with bounds.	None
xu	float or array - like	Lower / upper bounds for all variables. A scalar applies the same bound to every variable. Mutually exclusive with bounds.	None
vectorized	bool	If False VAMOS evaluates your scalar function one row at a time for compatibility. Set True only when your function already handles batches directly for real vectorized performance.	False
encoding	str	Variable encoding: "real", "binary", "integer", "permutation", or "mixed".	required
name	str	Human-readable name shown in logs and results. Defaults to the function name.	None
constraints	callable	Constraint function following the same signature convention as fn. Must return n_constraints values where g(x) <= 0 is feasible.	None
n_constraints	int	Number of constraint values. Required when constraints is provided.	0

Returns:

Type	Description
FunctionalProblem	A problem object ready to pass to vamos.optimize().

Raises:

Type	Description
TypeError	If fn is not callable.
ValueError	If bounds and xl/xu are both provided, if bounds length does not match n_var, or if n_constraints > 0 but no constraints callable is given.

Examples:

Minimal two-objective problem::

from vamos import make_problem, optimize

problem = make_problem(
    lambda x: [x[0], 1 - x[0] ** 0.5],
    n_var=2, n_obj=2,
    bounds=[(0, 1), (0, 1)],
    encoding="real",
)
result = optimize(problem, algorithm="nsgaii", max_evaluations=2000)

Vectorized for better performance::

import numpy as np

def my_objectives(X):
    f1 = X[:, 0]
    f2 = 1 - np.sqrt(X[:, 0])
    return np.column_stack([f1, f2])

problem = make_problem(
    my_objectives,
    n_var=2, n_obj=2,
    bounds=[(0, 1), (0, 1)],
    vectorized=True,
    encoding="real",
)

With constraints (g(x) <= 0 is feasible)::

problem = make_problem(
    lambda x: [x[0] + x[1], x[0] * x[1]],
    n_var=2, n_obj=2,
    bounds=[(0, 5), (0, 5)],
    encoding="real",
    constraints=lambda x: [x[0] + x[1] - 4],
    n_constraints=1,
)

Source code in src/vamos/foundation/problem/builder.py

def make_problem(
    fn: Callable[..., object],
    *,
    n_var: int,
    n_obj: int,
    bounds: Sequence[tuple[float, float]] | None = None,
    xl: float | Sequence[float] | np.ndarray | None = None,
    xu: float | Sequence[float] | np.ndarray | None = None,
    vectorized: bool = False,
    encoding: str,
    name: str | None = None,
    constraints: Callable[..., object] | None = None,
    n_constraints: int = 0,
) -> FunctionalProblem:
    """Create a VAMOS-compatible problem from a plain Python function.

    This is the friendliest way to define a custom optimization problem.
    Your function receives decision variables and returns objective values
    -- VAMOS handles bounds, protocol adaptation, and optional batched execution.

    Parameters
    ----------
    fn : callable
        Objective function.

        * **Scalar mode** (default, ``vectorized=False``): receives a 1-D
          array of shape ``(n_var,)`` and returns a list or array of
          ``n_obj`` objective values.
        * **Vectorized mode** (``vectorized=True``): receives a 2-D array
          of shape ``(N, n_var)`` and returns an array of shape
          ``(N, n_obj)``.

    n_var : int
        Number of decision variables.

    n_obj : int
        Number of objectives to minimize.

    bounds : sequence of (lower, upper) tuples, optional
        Per-variable bounds, e.g. ``[(0, 1), (0, 5)]``.  Must have
        length ``n_var``.  Mutually exclusive with *xl* / *xu*.

    xl, xu : float or array-like, optional
        Lower / upper bounds for all variables.  A scalar applies the
        same bound to every variable.  Mutually exclusive with *bounds*.

    vectorized : bool, default False
        If ``False`` VAMOS evaluates your scalar function one row at a time
        for compatibility. Set ``True`` only when your function already
        handles batches directly for real vectorized performance.

    encoding : str
        Variable encoding: ``"real"``, ``"binary"``, ``"integer"``,
        ``"permutation"``, or ``"mixed"``.

    name : str, optional
        Human-readable name shown in logs and results.  Defaults to the
        function name.

    constraints : callable, optional
        Constraint function following the same signature convention as
        *fn*.  Must return ``n_constraints`` values where
        ``g(x) <= 0`` is feasible.

    n_constraints : int, default 0
        Number of constraint values.  Required when *constraints* is
        provided.

    Returns
    -------
    FunctionalProblem
        A problem object ready to pass to ``vamos.optimize()``.

    Raises
    ------
    TypeError
        If *fn* is not callable.
    ValueError
        If *bounds* and *xl*/*xu* are both provided, if *bounds* length
        does not match *n_var*, or if *n_constraints* > 0 but no
        *constraints* callable is given.

    Examples
    --------
    Minimal two-objective problem::

        from vamos import make_problem, optimize

        problem = make_problem(
            lambda x: [x[0], 1 - x[0] ** 0.5],
            n_var=2, n_obj=2,
            bounds=[(0, 1), (0, 1)],
            encoding="real",
        )
        result = optimize(problem, algorithm="nsgaii", max_evaluations=2000)

    Vectorized for better performance::

        import numpy as np

        def my_objectives(X):
            f1 = X[:, 0]
            f2 = 1 - np.sqrt(X[:, 0])
            return np.column_stack([f1, f2])

        problem = make_problem(
            my_objectives,
            n_var=2, n_obj=2,
            bounds=[(0, 1), (0, 1)],
            vectorized=True,
            encoding="real",
        )

    With constraints (``g(x) <= 0`` is feasible)::

        problem = make_problem(
            lambda x: [x[0] + x[1], x[0] * x[1]],
            n_var=2, n_obj=2,
            bounds=[(0, 5), (0, 5)],
            encoding="real",
            constraints=lambda x: [x[0] + x[1] - 4],
            n_constraints=1,
        )
    """
    # ---- validate callable ----
    if not callable(fn):
        raise ConfigurationError(
            f"First argument must be a callable, got {type(fn).__name__}."
            "\n\nHint: make_problem(my_function, ...) where my_function(x) "
            "returns a list of objective values."
        )

    # ---- validate dimensions ----
    if not isinstance(n_var, int) or n_var < 1:
        raise ProblemDimensionError("n_var must be a positive integer.", n_var=n_var)
    if not isinstance(n_obj, int) or n_obj < 1:
        raise ProblemDimensionError("n_obj must be a positive integer.", n_obj=n_obj)

    # ---- resolve bounds ----
    if bounds is not None and (xl is not None or xu is not None):
        raise BoundsError("Use either 'bounds' or 'xl'/'xu', not both.\n\nHint: bounds=[(0, 1), (0, 1)] is equivalent to xl=0.0, xu=1.0")

    if bounds is not None:
        if len(bounds) != n_var:
            raise BoundsError(
                f"bounds has {len(bounds)} entries but n_var={n_var}.\n\nHint: provide exactly one (lower, upper) pair per variable."
            )
        for i, b in enumerate(bounds):
            if not (isinstance(b, (tuple, list)) and len(b) == 2):
                raise BoundsError(f"bounds[{i}] must be a (lower, upper) pair, got {b!r}.")
            if b[0] > b[1]:
                raise BoundsError(f"bounds[{i}]: lower bound ({b[0]}) > upper bound ({b[1]}).")
        xl_arr = np.array([b[0] for b in bounds], dtype=float)
        xu_arr = np.array([b[1] for b in bounds], dtype=float)
    else:
        if xl is None:
            xl = 0.0
        if xu is None:
            xu = 1.0
        try:
            xl_arr = np.broadcast_to(np.asarray(xl, dtype=float), (n_var,)).copy()
            xu_arr = np.broadcast_to(np.asarray(xu, dtype=float), (n_var,)).copy()
        except ValueError as exc:
            raise BoundsError(
                f"Could not broadcast xl/xu to shape ({n_var},)."
                "\n\nHint: pass scalars or one value per decision variable."
            ) from exc
        if np.any(xl_arr > xu_arr):
            raise BoundsError("Lower bounds must not exceed upper bounds.")

    # ---- validate constraints ----
    if constraints is not None:
        if not callable(constraints):
            raise ConfigurationError("'constraints' must be a callable.")
        if n_constraints < 1:
            raise ProblemDimensionError(
                "n_constraints must be >= 1 when a constraints function is provided."
                "\n\nHint: n_constraints is the number of constraint values "
                "your function returns."
            )
    if n_constraints > 0 and constraints is None:
        raise ProblemDimensionError(
            f"n_constraints={n_constraints} but no constraints function was provided."
            "\n\nHint: pass constraints=your_function where your_function(x) "
            "returns a list of n_constraints values (g(x) <= 0 is feasible)."
        )

    # ---- resolve name ----
    if name is None:
        name = getattr(fn, "__name__", None) or "custom_problem"
        if name == "<lambda>":
            name = "custom_problem"

    # ---- normalize encoding ----
    try:
        enc = normalize_encoding(encoding)
    except ValueError as exc:
        raise ConfigurationError(str(exc)) from exc

    return FunctionalProblem(
        fn,
        n_var=n_var,
        n_obj=n_obj,
        xl=xl_arr,
        xu=xu_arr,
        encoding=enc,
        vectorized=vectorized,
        name=name,
        constraints_fn=constraints,
        n_constraints=n_constraints,
    )

vamos.foundation.problem.types

Results

vamos.experiment.optimization_result

Public optimization result types and helpers.

OptimizationResult

Container returned by optimize() with Pareto front data and selection helpers.

Use vamos.ux.api for summaries, plotting, and export helpers.

Source code in src/vamos/experiment/optimization_result/model.py

class OptimizationResult:
    """
    Container returned by optimize() with Pareto front data and selection helpers.

    Use `vamos.ux.api` for summaries, plotting, and export helpers.
    """

    F: NDArray[np.float64] | None
    X: NDArray[np.generic] | None
    data: dict[str, Any]
    meta: dict[str, Any]

    def __init__(
        self,
        payload: Mapping[str, Any],
        *,
        meta: Mapping[str, Any] | None = None,
        manifest: RunManifest | None = None,
    ):
        self.F = payload.get("F")
        self.X = payload.get("X")
        self.data = dict(payload)
        self.meta = dict(meta or {})
        self._manifest = manifest

    def __len__(self) -> int:
        return len(self.F) if self.F is not None else 0

    def __repr__(self) -> str:
        n_sol = len(self)
        n_obj = self.F.shape[1] if self.F is not None and len(self.F) > 0 else 0
        return f"OptimizationResult({n_sol} solutions, {n_obj} objectives)"

    @property
    def n_objectives(self) -> int:
        return self.F.shape[1] if self.F is not None and len(self.F) > 0 else 0

    @property
    def manifest(self) -> RunManifest | None:
        """Immutable source manifest when this result was loaded from a run."""
        return self._manifest

    @overload
    def front(self, *, return_indices: Literal[False] = False) -> NDArray[np.float64] | None: ...

    @overload
    def front(self, *, return_indices: Literal[True]) -> tuple[NDArray[np.float64], NDArray[np.int_]]: ...

    def front(
        self,
        *,
        return_indices: bool = False,
    ) -> NDArray[np.float64] | tuple[NDArray[np.float64], NDArray[np.int_]] | None:
        if return_indices:
            return pareto_filter(self.F, return_indices=True)
        return pareto_filter(self.F, return_indices=False)

    def best(self, method: BestMethod = "balanced_sum") -> BestResult:
        if self.F is None or len(self.F) == 0:
            raise NoSolutionsError("No solutions available.")

        front = self.front(return_indices=True)
        if front is None:
            raise NoSolutionsError("No solutions available.")
        front_F, front_idx = front
        if len(front_F) == 0:
            raise NoSolutionsError("No solutions available.")

        resolved_method = normalize_best_method(method)
        if resolved_method in {"balanced_sum", "knee"}:
            F_norm = (front_F - front_F.min(axis=0)) / (np.ptp(front_F, axis=0) + 1e-12)
            front_pos = int(np.argmin(F_norm.sum(axis=1)))
        elif resolved_method == "min_f1":
            front_pos = int(np.argmin(front_F[:, 0]))
        elif resolved_method == "min_f2":
            if front_F.shape[1] < 2:
                raise ResultSelectionError(f"'min_f2' requires at least 2 objectives, but this result has {front_F.shape[1]}.")
            front_pos = int(np.argmin(front_F[:, 1]))
        elif resolved_method == "balanced":
            F_norm = (front_F - front_F.min(axis=0)) / (np.ptp(front_F, axis=0) + 1e-12)
            front_pos = int(np.argmin(F_norm.max(axis=1)))
        else:
            raise AssertionError(f"Unhandled best() method '{resolved_method}'.")

        idx = int(front_idx[front_pos])
        return {
            "X": self.X[idx] if self.X is not None else None,
            "F": self.F[idx],
            "index": idx,
            "front_index": front_pos,
        }

    def top_k(
        self,
        k: int = 100,
        *,
        source: RankingSource = "archive",
        method: RankingMethod = "balanced_sum",
        nondominated_only: bool = True,
        weights: NDArray[np.float64] | None = None,
    ) -> TopKResult:
        return cast(
            TopKResult,
            rank_top_k(
                self,
                k=k,
                source=source,
                method=method,
                nondominated_only=nondominated_only,
                weights=weights,
            ),
        )

    def top_k_report(
        self,
        k: int = 100,
        *,
        source: RankingSource = "archive",
        method: RankingMethod = "balanced_sum",
        nondominated_only: bool = True,
        weights: NDArray[np.float64] | None = None,
    ) -> list[dict[str, Any]]:
        return build_top_k_report(
            result=self,
            k=k,
            source=source,
            method=method,
            nondominated_only=nondominated_only,
            weights=weights,
        )

    def explain_defaults(self) -> dict[str, object]:
        explained: dict[str, object] = {}
        resolved = self.meta.get("run_artifact_resolved_spec")
        sources = self.meta.get("default_sources")
        if resolved is not None:
            explained["resolved_spec"] = resolved
        if sources is not None:
            explained["default_sources"] = sources
        return explained

manifest property

Immutable source manifest when this result was loaded from a run.

StudyResult

Bases: Sequence[OptimizationResult]

Container returned by optimize(..., seed=[...]).

The object behaves like a read-only sequence of :class:OptimizationResult while also exposing light-weight aggregation helpers for numeric metrics.

Source code in src/vamos/experiment/optimization_result/study.py

class StudyResult(Sequence[OptimizationResult]):
    """Container returned by ``optimize(..., seed=[...])``.

    The object behaves like a read-only sequence of :class:`OptimizationResult`
    while also exposing light-weight aggregation helpers for numeric metrics.
    """

    def __init__(self, runs: Sequence[OptimizationResult] | Iterator[OptimizationResult]):
        self._runs = tuple(runs)

    @property
    def runs(self) -> tuple[OptimizationResult, ...]:
        """Return the underlying immutable run collection."""
        return self._runs

    def __len__(self) -> int:
        return len(self._runs)

    def __iter__(self) -> Iterator[OptimizationResult]:
        return iter(self._runs)

    @overload
    def __getitem__(self, index: int) -> OptimizationResult: ...

    @overload
    def __getitem__(self, index: slice) -> StudyResult: ...

    def __getitem__(self, index: int | slice) -> OptimizationResult | StudyResult:
        if isinstance(index, slice):
            return StudyResult(self._runs[index])
        return self._runs[index]

    def __repr__(self) -> str:
        return f"StudyResult({len(self._runs)} runs)"

    def metric_values(self, name: str) -> NDArray[np.float64]:
        """Return a float array with the named metric extracted from each run."""
        if not self._runs:
            return np.empty(0, dtype=float)
        return np.asarray([_lookup_metric_value(run, name) for run in self._runs], dtype=float)

    def mean(self, name: str) -> float:
        """Return the mean of a numeric metric across runs."""
        values = self.metric_values(name)
        if values.size == 0:
            raise NoSolutionsError("StudyResult contains no runs.")
        return float(np.mean(values))

    def std(self, name: str) -> float:
        """Return the population standard deviation of a numeric metric across runs."""
        values = self.metric_values(name)
        if values.size == 0:
            raise NoSolutionsError("StudyResult contains no runs.")
        return float(np.std(values))

    def best_run(self, name: str, maximize: bool = True) -> OptimizationResult:
        """Return the run with the best value for the named metric."""
        values = self.metric_values(name)
        if values.size == 0:
            raise NoSolutionsError("StudyResult contains no runs.")
        best_index = int(np.argmax(values) if maximize else np.argmin(values))
        return self._runs[best_index]

runs property

Return the underlying immutable run collection.

best_run(name, maximize=True)

Return the run with the best value for the named metric.

Source code in src/vamos/experiment/optimization_result/study.py

def best_run(self, name: str, maximize: bool = True) -> OptimizationResult:
    """Return the run with the best value for the named metric."""
    values = self.metric_values(name)
    if values.size == 0:
        raise NoSolutionsError("StudyResult contains no runs.")
    best_index = int(np.argmax(values) if maximize else np.argmin(values))
    return self._runs[best_index]

mean(name)

Return the mean of a numeric metric across runs.

Source code in src/vamos/experiment/optimization_result/study.py

def mean(self, name: str) -> float:
    """Return the mean of a numeric metric across runs."""
    values = self.metric_values(name)
    if values.size == 0:
        raise NoSolutionsError("StudyResult contains no runs.")
    return float(np.mean(values))

metric_values(name)

Return a float array with the named metric extracted from each run.

Source code in src/vamos/experiment/optimization_result/study.py

def metric_values(self, name: str) -> NDArray[np.float64]:
    """Return a float array with the named metric extracted from each run."""
    if not self._runs:
        return np.empty(0, dtype=float)
    return np.asarray([_lookup_metric_value(run, name) for run in self._runs], dtype=float)

std(name)

Return the population standard deviation of a numeric metric across runs.

Source code in src/vamos/experiment/optimization_result/study.py

def std(self, name: str) -> float:
    """Return the population standard deviation of a numeric metric across runs."""
    values = self.metric_values(name)
    if values.size == 0:
        raise NoSolutionsError("StudyResult contains no runs.")
    return float(np.std(values))

Run artifacts

The top-level persistence, verification, and exact built-in replay functions delegate to this facade. Loading and verification are data-only; replay is the separate explicit executable operation.

vamos.run_artifacts

Layer-neutral public bridge for canonical run-artifact operations.

StoredRun dataclass

Immutable data-only handle with lazy, side-effect-free artifact access.

Source code in src/vamos/experiment/artifacts/models.py

@dataclass(frozen=True)
class StoredRun:
    """Immutable data-only handle with lazy, side-effect-free artifact access."""

    root: Path
    manifest: RunManifest
    _result_loader: Callable[[], OptimizationResult] = field(repr=False, compare=False)
    _environment_loader: Callable[[], Mapping[str, Any]] = field(repr=False, compare=False)

    @property
    def status(self) -> str:
        return self.manifest.status

    @cached_property
    def result(self) -> OptimizationResult:
        """Load and cache the canonical result bundle without executing code."""
        return self._result_loader()

    @cached_property
    def environment(self) -> Mapping[str, Any]:
        """Load and cache the immutable captured environment document."""
        return self._environment_loader()

environment cached property

Load and cache the immutable captured environment document.

result cached property

Load and cache the canonical result bundle without executing code.

RunManifest dataclass

Bases: Mapping[str, Any]

Immutable validated envelope for one v1 execution attempt.

Source code in src/vamos/experiment/artifacts/models.py

@dataclass(frozen=True, slots=True)
class RunManifest(Mapping[str, Any]):
    """Immutable validated envelope for one v1 execution attempt."""

    _data: Mapping[str, Any]
    resolved_spec: ResolvedRunSpec
    artifacts: tuple[ArtifactDescriptor, ...]

    def __getitem__(self, key: str) -> Any:
        return self._data[key]

    def __iter__(self) -> Iterator[str]:
        return iter(self._data)

    def __len__(self) -> int:
        return len(self._data)

    @property
    def run_id(self) -> str:
        return str(self._data["run_id"])

    @property
    def task_id(self) -> str:
        return str(self._data["task_id"])

    @property
    def status(self) -> str:
        return str(self._data["status"])

    @property
    def requested_spec(self) -> Mapping[str, Any]:
        value = self._data["requested_spec"]
        if not isinstance(value, Mapping):
            raise ManifestValidationError(
                operation="access manifest",
                field="$.requested_spec",
                reason="is not an object",
                expected="JSON object",
                actual=type(value).__name__,
                action="Restore a valid v1 manifest.",
            )
        return value

    def artifact(self, role: str) -> ArtifactDescriptor | None:
        """Return the singleton descriptor for ``role`` when present."""
        for descriptor in self.artifacts:
            if descriptor.role == role:
                return descriptor
        return None

    def as_dict(self) -> dict[str, Any]:
        thawed = deep_thaw(self._data)
        if not isinstance(thawed, dict):
            raise AssertionError("deep_thaw returned a non-object manifest")
        return thawed

artifact(role)

Return the singleton descriptor for role when present.

Source code in src/vamos/experiment/artifacts/models.py

def artifact(self, role: str) -> ArtifactDescriptor | None:
    """Return the singleton descriptor for ``role`` when present."""
    for descriptor in self.artifacts:
        if descriptor.role == role:
            return descriptor
    return None

LoadLimits dataclass

Finite defensive limits used by trusted v1 readers.

Callers may pass a different instance explicitly when they trust a larger artifact. Normal loading never increases these limits automatically.

Source code in src/vamos/experiment/artifacts/models.py

@dataclass(frozen=True, slots=True)
class LoadLimits:
    """Finite defensive limits used by trusted v1 readers.

    Callers may pass a different instance explicitly when they trust a larger
    artifact. Normal loading never increases these limits automatically.
    """

    max_manifest_bytes: int = 8 * 1024 * 1024
    max_environment_bytes: int = 16 * 1024 * 1024
    max_artifact_bytes: int = 512 * 1024 * 1024
    max_artifacts: int = 128
    max_json_depth: int = 64
    max_zip_members: int = 128
    max_arrays: int = 64
    max_total_uncompressed_bytes: int = 1024 * 1024 * 1024
    max_array_bytes: int = 512 * 1024 * 1024
    max_total_elements: int = 100_000_000
    max_array_dimensions: int = 8
    max_npy_header_bytes: int = 64 * 1024
    max_compression_ratio: float = 1000.0

    def __post_init__(self) -> None:
        for item in fields(self):
            value = getattr(self, item.name)
            if item.name == "max_compression_ratio":
                valid = not isinstance(value, bool) and isinstance(value, (int, float)) and value > 0
            else:
                valid = not isinstance(value, bool) and isinstance(value, int) and value > 0
            if not valid:
                raise ValueError(f"LoadLimits.{item.name} must be a positive number.")

CompatibilityReport dataclass

Material compatibility of a stored run with the current runtime.

Source code in src/vamos/experiment/artifacts/reports.py

@dataclass(frozen=True, slots=True)
class CompatibilityReport:
    """Material compatibility of a stored run with the current runtime."""

    level: CompatibilityLevel
    findings: tuple[CompatibilityFinding, ...]

    @property
    def exact(self) -> bool:
        return self.level == "exact"

    def as_dict(self) -> dict[str, Any]:
        return {"level": self.level, "findings": [item.as_dict() for item in self.findings]}

VerificationReport dataclass

Independent verification dimensions for one canonical run.

Source code in src/vamos/experiment/artifacts/reports.py

@dataclass(frozen=True, slots=True)
class VerificationReport:
    """Independent verification dimensions for one canonical run."""

    root: Path
    run_id: str
    task_id: str
    status: str
    schema: str
    artifact_integrity: IntegrityStatus
    path_safety: IntegrityStatus
    numerical_bundle_safety: IntegrityStatus
    environment: CompatibilityReport
    component_reconstructability: ComponentStatus
    effective_replayability: ReplayabilityLevel
    reasons: tuple[VerificationReason, ...]
    optimization_executed: bool = False

    def as_dict(self) -> dict[str, Any]:
        return {
            "document_type": "vamos.verification-report",
            "version": "1",
            "root": str(self.root),
            "run_id": self.run_id,
            "task_id": self.task_id,
            "status": self.status,
            "schema": self.schema,
            "artifact_integrity": self.artifact_integrity,
            "path_safety": self.path_safety,
            "numerical_bundle_safety": self.numerical_bundle_safety,
            "environment_compatibility": self.environment.as_dict(),
            "component_reconstructability": self.component_reconstructability,
            "effective_replayability": self.effective_replayability,
            "reasons": [reason.as_dict() for reason in self.reasons],
            "optimization_executed": self.optimization_executed,
        }

ReplayReport dataclass

Outcome and evidence for a newly stored replay attempt.

Source code in src/vamos/experiment/artifacts/reports.py

@dataclass(frozen=True, slots=True)
class ReplayReport:
    """Outcome and evidence for a newly stored replay attempt."""

    source_root: Path
    output_root: Path
    source_run_id: str
    replay_run_id: str
    task_id: str
    source_manifest_sha256: str
    replay_plan_sha256: str
    exact: bool
    comparisons: tuple[ArrayComparison, ...]
    verification: VerificationReport
    optimization_executed: bool = True

    def as_dict(self) -> dict[str, Any]:
        return {
            "document_type": "vamos.replay-report",
            "version": "1",
            "source_root": str(self.source_root),
            "output_root": str(self.output_root),
            "source_run_id": self.source_run_id,
            "replay_run_id": self.replay_run_id,
            "task_id": self.task_id,
            "source_manifest_sha256": self.source_manifest_sha256,
            "replay_plan_sha256": self.replay_plan_sha256,
            "exact": self.exact,
            "comparisons": [item.as_dict() for item in self.comparisons],
            "verification": self.verification.as_dict(),
            "optimization_executed": self.optimization_executed,
        }

save_result(result, path, *, requested_spec=None, resolved_spec=None, labels=None, limits=None)

Persist a result as one immutable, relocatable v1 run directory.

Source code in src/vamos/experiment/artifacts/persistence.py

def save_result(
    result: ResultLike,
    path: str | Path,
    *,
    requested_spec: Mapping[str, Any] | None = None,
    resolved_spec: Mapping[str, Any] | None = None,
    labels: Mapping[str, str] | None = None,
    limits: LoadLimits | None = None,
) -> StoredRun:
    """Persist a result as one immutable, relocatable v1 run directory."""
    active_limits = limits if limits is not None else LoadLimits()
    arrays = snapshot_result_arrays(result, limits=active_limits)
    meta = _result_meta_mapping(result)
    recorded_requested, resolved, caller_supplied = _result_specs(
        result,
        requested_spec=requested_spec,
        resolved_spec=resolved_spec,
    )
    timestamps, runtime_ms = _timestamps(meta)
    backend = _kernel_name(resolved)
    entry_point = meta.get("run_artifact_entry_point")
    provenance, environment = capture_provenance(
        backend=backend,
        timestamps=timestamps,
        entry_point=entry_point if isinstance(entry_point, Mapping) else None,
    )
    deterministic = _declared_deterministic(resolved)
    replayability = replayability_from_provenance(provenance, deterministic=deterministic)
    unavailable_reason = _find_unavailable_reason(resolved)
    if unavailable_reason is not None:
        replayability = {
            "declared_level": "manual",
            "deterministic": deterministic,
            "exact_requirements": [],
            "reasons": [
                {
                    "code": "automatic_component_reconstruction_unavailable",
                    "message": unavailable_reason,
                }
            ],
        }
    if caller_supplied:
        provenance["entry_point"] = {
            "kind": "python_api",
            "python": {"callable": "vamos.save_result", "arguments_source": "caller_supplied_run_context"},
        }
        replayability = {
            "declared_level": "manual",
            "deterministic": deterministic,
            "exact_requirements": [],
            "reasons": [
                {
                    "code": "caller_supplied_execution_context",
                    "message": "The execution specification was supplied by the save_result caller.",
                }
            ],
        }
    replayability_override = meta.get("run_artifact_replayability")
    if isinstance(replayability_override, Mapping):
        normalized_replayability = normalize_json(replayability_override, field="$.replayability")
        if isinstance(normalized_replayability, dict):
            replayability = normalized_replayability
    task_id = "sha256:" + sha256_bytes(canonical_json_bytes(resolved))
    selected_run_id = meta.get("run_artifact_run_id")
    run_id = selected_run_id if isinstance(selected_run_id, str) else str(uuid.uuid4())
    manifest: dict[str, Any] = {
        "document_type": DOCUMENT_TYPE,
        "schema_version": SCHEMA_VERSION,
        "run_id": run_id,
        "task_id": task_id,
        "status": "succeeded",
        "timestamps": timestamps,
        "requested_spec": recorded_requested,
        "resolved_spec": resolved,
        "provenance": provenance,
        "replayability": replayability,
        "outcome": _outcome(result, arrays, resolved=resolved, runtime_ms=runtime_ms),
        "artifacts": [],
    }
    if labels is not None:
        manifest["labels"] = dict(labels)
    lineage = meta.get("run_artifact_lineage")
    if isinstance(lineage, Mapping):
        manifest["lineage"] = dict(lineage)
    store_succeeded_run(
        Path(path),
        arrays=arrays,
        environment=environment,
        manifest_base=manifest,
        limits=active_limits,
    )
    return read_run(Path(path), verify="required", limits=active_limits)

load_run(path, *, verify='required', limits=None)

Load immutable manifest access without resolving or executing code.

Source code in src/vamos/experiment/artifacts/persistence.py

def load_run(
    path: str | Path,
    *,
    verify: VerifyMode = "required",
    limits: LoadLimits | None = None,
) -> StoredRun:
    """Load immutable manifest access without resolving or executing code."""
    active_limits = limits if limits is not None else LoadLimits()
    return read_run(path, verify=verify, limits=active_limits)

load_result(path, *, verify='required', limits=None)

Load the canonical numerical result; this never reruns optimization.

Source code in src/vamos/experiment/artifacts/persistence.py

def load_result(
    path: str | Path,
    *,
    verify: VerifyMode = "required",
    limits: LoadLimits | None = None,
) -> OptimizationResult:
    """Load the canonical numerical result; this never reruns optimization."""
    return load_run(path, verify=verify, limits=limits).result

verify_run(path, *, require_level=None, limits=None)

Fully verify a canonical run without executing or resolving code.

Source code in src/vamos/experiment/artifacts/verification.py

def verify_run(
    path: str | Path,
    *,
    require_level: RequiredLevel | None = None,
    limits: LoadLimits | None = None,
) -> VerificationReport:
    """Fully verify a canonical run without executing or resolving code."""
    active_limits = limits if limits is not None else LoadLimits()
    stored = read_run(path, verify="all", limits=active_limits)
    _verify_every_artifact(stored, active_limits)
    bundle_status = _inspect_numerical_bundle(stored, active_limits)
    environment = compare_current_environment(stored.manifest, stored.environment)
    component_status, component_reasons = component_reconstructability(stored.manifest)
    reasons = (*component_reasons, *_effective_reasons(stored, environment.level, bundle_status))
    effective = _effective_level(stored, environment.level, component_status, bundle_status)
    report = VerificationReport(
        root=stored.root,
        run_id=stored.manifest.run_id,
        task_id=stored.manifest.task_id,
        status=stored.status,
        schema=str(stored.manifest["schema_version"]),
        artifact_integrity="valid",
        path_safety="valid",
        numerical_bundle_safety=bundle_status,
        environment=environment,
        component_reconstructability=component_status,
        effective_replayability=effective,
        reasons=reasons,
    )
    _enforce_requirement(report, require_level)
    return report

reproduce(path, *, output=None, limits=None)

Verify, exactly execute, compare, and store a new built-in replay.

Source code in src/vamos/experiment/artifacts/replay.py

def reproduce(
    path: str | Path,
    *,
    output: str | Path | None = None,
    limits: LoadLimits | None = None,
) -> ReplayReport:
    """Verify, exactly execute, compare, and store a new built-in replay."""
    active_limits = limits if limits is not None else LoadLimits()
    verification = verify_run(path, limits=active_limits)
    if verification.environment.level != "exact":
        raise EnvironmentIncompatibilityError(
            operation="reproduce run",
            field="$.environment",
            path=verification.root,
            reason="does not satisfy exact replay compatibility",
            expected="exact",
            actual=verification.environment.level,
            action="Use the same implementation, Python, dependencies, backend, BLAS, and material thread settings.",
            optimization_executed=False,
        )
    source = load_run(path, verify="all", limits=active_limits)
    plan = build_replay_plan(source, verification, output)
    source_arrays = snapshot_result_arrays(source.result, limits=active_limits)
    problem = _instantiate_problem(plan)
    started_at = _now()
    started_monotonic = time.perf_counter()
    try:
        result = _run_config(
            _OptimizeConfig(
                problem=problem,
                algorithm=plan.algorithm,
                algorithm_config=plan.algorithm_config,
                termination=plan.termination,
                seed=plan.seed,
                engine=plan.engine,
                eval_strategy=plan.eval_strategy,
            ),
            built_in_only=True,
        )
    except Exception as exc:
        completed_at = _now()
        runtime_ms = (time.perf_counter() - started_monotonic) * 1000.0
        _store_failed_attempt(plan, exc, started_at, completed_at, runtime_ms, active_limits)
        raise ReplayExecutionError(
            operation="reproduce run",
            field="$.lineage.comparison",
            path=plan.output_root,
            reason="optimization execution failed",
            expected="completed deterministic built-in execution",
            actual={"exception_type": type(exc).__name__, "message": _sanitized_message(exc)},
            action=f"Inspect the failed canonical attempt at {plan.output_root} and correct the built-in execution failure.",
            optimization_executed=True,
        ) from exc
    completed_at = _now()
    runtime_ms = (time.perf_counter() - started_monotonic) * 1000.0
    replay_arrays = snapshot_result_arrays(result, limits=active_limits)
    comparisons = compare_array_collections(source_arrays, replay_arrays)
    exact = comparisons_are_exact(comparisons)
    lineage = _lineage(plan, comparisons, status="exact_match" if exact else "mismatch")
    _attach_replay_metadata(result, plan, lineage, started_at, completed_at, runtime_ms, exact)
    stored = save_result(result, plan.output_root, limits=active_limits)
    report = ReplayReport(
        source_root=source.root,
        output_root=stored.root,
        source_run_id=plan.source_run_id,
        replay_run_id=stored.manifest.run_id,
        task_id=stored.manifest.task_id,
        source_manifest_sha256=plan.source_manifest_sha256,
        replay_plan_sha256=plan.replay_plan_sha256,
        exact=exact,
        comparisons=comparisons,
        verification=verification,
    )
    if not exact:
        raise ReplayResultMismatchError(
            operation="reproduce run",
            field="$.lineage.comparison",
            path=stored.root,
            reason="completed without bitwise equality",
            expected="exact F, X, and auxiliary deterministic arrays",
            actual={"exact": False, "comparisons": [item.as_dict() for item in comparisons if not item.exact]},
            action=f"Inspect the stored mismatch attempt at {stored.root}; do not treat it as an exact replay.",
            optimization_executed=True,
        )
    return report

Durable studies

The top-level StudySpec, plan_study, create_study, and load_study entry points resolve, publish, and verify durable studies. Creation performs no optimization, loading is data-only, and report/summary models remain available through the focused vamos.study_artifacts facade.

vamos.study_artifacts

Layer-neutral public bridge for canonical StudyManifest operations.

StudySpec dataclass

Validated, immutable user intent for a deterministic study plan.

Seeds are explicit. All defaults selected during creation are frozen into each task's resolved run specification before publication.

Source code in src/vamos/experiment/study/models.py

@dataclass(frozen=True, slots=True)
class StudySpec:
    """Validated, immutable user intent for a deterministic study plan.

    Seeds are explicit. All defaults selected during creation are frozen into
    each task's resolved run specification before publication.
    """

    problems: Sequence[str]
    algorithms: Sequence[str]
    seeds: Sequence[int]
    max_evaluations: int | None = None
    pop_size: int | None = None
    engine: str | None = None
    eval_strategy: str = "serial"
    n_var: int | None = None
    n_obj: int | None = None
    problem_kwargs: Mapping[str, object] | None = None
    algorithm_configs: Mapping[str, object] | None = None
    on_error: OnErrorPolicy = "fail_fast"
    max_attempts_per_task: int = 3
    labels: Mapping[str, object] | None = None
    metadata: Mapping[str, object] | None = None
    study_id: str | None = field(default=None, init=False, repr=False, compare=False)

    def __post_init__(self) -> None:
        object.__setattr__(self, "problems", _names("problems", self.problems))
        object.__setattr__(self, "algorithms", _names("algorithms", self.algorithms))
        object.__setattr__(self, "seeds", _seeds(self.seeds))
        object.__setattr__(self, "max_evaluations", _positive_optional("max_evaluations", self.max_evaluations))
        object.__setattr__(self, "pop_size", _positive_optional("pop_size", self.pop_size))
        object.__setattr__(self, "n_var", _positive_optional("n_var", self.n_var))
        object.__setattr__(self, "n_obj", _positive_optional("n_obj", self.n_obj))
        if self.engine is not None and (not isinstance(self.engine, str) or not self.engine.strip()):
            _invalid("engine", "non-empty string or None", self.engine)
        if self.eval_strategy not in {"serial", "multiprocessing", "dask"}:
            _invalid("eval_strategy", "'serial', 'multiprocessing', or 'dask'", self.eval_strategy)
        if self.on_error not in ("fail_fast", "continue"):
            _invalid("on_error", "'fail_fast' or 'continue'", self.on_error)
        attempts = _positive_optional("max_attempts_per_task", self.max_attempts_per_task)
        if attempts is None:
            raise AssertionError("max_attempts_per_task cannot be None")
        object.__setattr__(self, "max_attempts_per_task", attempts)
        object.__setattr__(self, "problem_kwargs", _frozen_json_mapping("problem_kwargs", self.problem_kwargs, max_bytes=256 * 1024))
        object.__setattr__(
            self, "algorithm_configs", _frozen_json_mapping("algorithm_configs", self.algorithm_configs, max_bytes=1024 * 1024)
        )
        object.__setattr__(self, "labels", _frozen_json_mapping("labels", self.labels, max_bytes=256 * 1024))
        object.__setattr__(self, "metadata", _frozen_json_mapping("metadata", self.metadata, max_bytes=1024 * 1024))

    def as_intent_dict(self) -> dict[str, Any]:
        """Return a detached JSON representation without document identity."""
        return {
            "matrix": {
                "problems": list(self.problems),
                "algorithms": list(self.algorithms),
                "seeds": list(self.seeds),
            },
            "run_defaults": {
                "max_evaluations": self.max_evaluations,
                "pop_size": self.pop_size,
                "engine": self.engine,
                "eval_strategy": self.eval_strategy,
                "n_var": self.n_var,
                "n_obj": self.n_obj,
                "problem_kwargs": deep_thaw(self.problem_kwargs),
                "algorithm_configs": deep_thaw(self.algorithm_configs),
            },
            "policy": {
                "on_error": self.on_error,
                "max_attempts_per_task": self.max_attempts_per_task,
            },
            "labels": deep_thaw(self.labels),
            "metadata": deep_thaw(self.metadata),
        }

as_intent_dict()

Return a detached JSON representation without document identity.

Source code in src/vamos/experiment/study/models.py

def as_intent_dict(self) -> dict[str, Any]:
    """Return a detached JSON representation without document identity."""
    return {
        "matrix": {
            "problems": list(self.problems),
            "algorithms": list(self.algorithms),
            "seeds": list(self.seeds),
        },
        "run_defaults": {
            "max_evaluations": self.max_evaluations,
            "pop_size": self.pop_size,
            "engine": self.engine,
            "eval_strategy": self.eval_strategy,
            "n_var": self.n_var,
            "n_obj": self.n_obj,
            "problem_kwargs": deep_thaw(self.problem_kwargs),
            "algorithm_configs": deep_thaw(self.algorithm_configs),
        },
        "policy": {
            "on_error": self.on_error,
            "max_attempts_per_task": self.max_attempts_per_task,
        },
        "labels": deep_thaw(self.labels),
        "metadata": deep_thaw(self.metadata),
    }

StudyPlanReport dataclass

Immutable explanation of a resolved study with no published state.

Source code in src/vamos/experiment/study/preflight.py

@dataclass(frozen=True, slots=True)
class StudyPlanReport:
    """Immutable explanation of a resolved study with no published state."""

    plan: ResolvedStudyPlan
    status: PlanStatus
    valid: bool
    total_evaluation_budget: int
    problem_ids: tuple[str, ...]
    algorithm_ids: tuple[str, ...]
    operator_ids: tuple[str, ...]
    backend_ids: tuple[str, ...]
    seeds: tuple[int, ...]
    population_sizes: tuple[int, ...]
    termination_categories: tuple[str, ...]
    failure_policy: OnErrorPolicy
    reconstructable: bool
    duplicate_tasks: bool
    output: StudyPlanOutput
    warnings: tuple[str, ...]
    errors: tuple[StudyPlanDiagnostic, ...]
    next_actions: tuple[str, ...]

    @property
    def plan_id(self) -> str:
        return self.plan.plan_id

    @property
    def task_ids(self) -> tuple[str, ...]:
        return tuple(task.task_id for task in self.plan.tasks)

    @property
    def task_count(self) -> int:
        return self.plan.task_count

    def as_dict(self) -> dict[str, object]:
        """Return a detached semantic payload for Python and CLI consumers."""
        return {
            "status": self.status,
            "valid": self.valid,
            "execution_occurred": False,
            "filesystem_write_occurred": False,
            "plan_id": self.plan_id,
            "task_ids": list(self.task_ids),
            "task_count": self.task_count,
            "total_evaluation_budget": self.total_evaluation_budget,
            "components": {
                "problems": list(self.problem_ids),
                "algorithms": list(self.algorithm_ids),
                "operators": list(self.operator_ids),
                "backends": list(self.backend_ids),
            },
            "seeds": list(self.seeds),
            "population_sizes": list(self.population_sizes),
            "termination_categories": list(self.termination_categories),
            "failure_policy": self.failure_policy,
            "reconstructable": self.reconstructable,
            "duplicate_tasks": self.duplicate_tasks,
            "output": self.output.as_dict(),
            "warnings": list(self.warnings),
            "errors": [error.as_dict() for error in self.errors],
            "next_actions": list(self.next_actions),
        }

as_dict()

Return a detached semantic payload for Python and CLI consumers.

Source code in src/vamos/experiment/study/preflight.py

def as_dict(self) -> dict[str, object]:
    """Return a detached semantic payload for Python and CLI consumers."""
    return {
        "status": self.status,
        "valid": self.valid,
        "execution_occurred": False,
        "filesystem_write_occurred": False,
        "plan_id": self.plan_id,
        "task_ids": list(self.task_ids),
        "task_count": self.task_count,
        "total_evaluation_budget": self.total_evaluation_budget,
        "components": {
            "problems": list(self.problem_ids),
            "algorithms": list(self.algorithm_ids),
            "operators": list(self.operator_ids),
            "backends": list(self.backend_ids),
        },
        "seeds": list(self.seeds),
        "population_sizes": list(self.population_sizes),
        "termination_categories": list(self.termination_categories),
        "failure_policy": self.failure_policy,
        "reconstructable": self.reconstructable,
        "duplicate_tasks": self.duplicate_tasks,
        "output": self.output.as_dict(),
        "warnings": list(self.warnings),
        "errors": [error.as_dict() for error in self.errors],
        "next_actions": list(self.next_actions),
    }

Study dataclass

Immutable, data-only handle for a verified persisted study.

Source code in src/vamos/experiment/study/models.py

@dataclass(frozen=True, slots=True)
class Study:
    """Immutable, data-only handle for a verified persisted study."""

    root: Path
    manifest: StudyManifest
    spec: StudySpec
    plan: ResolvedStudyPlan
    tasks: tuple[TaskRecord, ...]
    attempts: tuple[AttemptRecord, ...] = field(repr=False)
    events: tuple[StudyEvent, ...] = field(repr=False)
    stored_checkpoint_sequence: int = field(repr=False)
    stored_checkpoint_event_sha256: str = field(repr=False)
    reconciliation_required: bool = field(repr=False)

    @property
    def study_id(self) -> str:
        return self.manifest.study_id

    @property
    def plan_id(self) -> str:
        return self.manifest.plan_id

    @property
    def status(self) -> StudyState:
        return self.manifest.state

    def run(self) -> Study:
        """Execute this newly created durable study sequentially."""
        from .execution import execute_study

        return execute_study(self)

    def cancel(self) -> Study:
        """Cancel this study or request cancellation from its local runner."""
        from .execution import cancel_study

        return cancel_study(self)

    def resume(self, *, retry_failed: bool = False) -> Study:
        """Reconcile this study and execute its eligible unfinished tasks."""
        from .recovery import resume_study

        return resume_study(self, retry_failed=retry_failed)

    def retry(self, *, failed_only: bool = True) -> Study:
        """Explicitly retry eligible terminal attempts without changing the plan."""
        from .recovery import retry_study

        return retry_study(self, failed_only=failed_only)

    def inspect(self) -> StudyReport:
        """Reload and project current durable state without modifying it."""
        from .projection import project_study

        return project_study(self).report

    def summarize(self) -> StudySummary:
        """Return a deterministic in-memory summary without writing files."""
        from .projection import project_study

        return project_study(self).summary

cancel()

Cancel this study or request cancellation from its local runner.

Source code in src/vamos/experiment/study/models.py

def cancel(self) -> Study:
    """Cancel this study or request cancellation from its local runner."""
    from .execution import cancel_study

    return cancel_study(self)

inspect()

Reload and project current durable state without modifying it.

Source code in src/vamos/experiment/study/models.py

def inspect(self) -> StudyReport:
    """Reload and project current durable state without modifying it."""
    from .projection import project_study

    return project_study(self).report

resume(*, retry_failed=False)

Reconcile this study and execute its eligible unfinished tasks.

Source code in src/vamos/experiment/study/models.py

def resume(self, *, retry_failed: bool = False) -> Study:
    """Reconcile this study and execute its eligible unfinished tasks."""
    from .recovery import resume_study

    return resume_study(self, retry_failed=retry_failed)

retry(*, failed_only=True)

Explicitly retry eligible terminal attempts without changing the plan.

Source code in src/vamos/experiment/study/models.py

def retry(self, *, failed_only: bool = True) -> Study:
    """Explicitly retry eligible terminal attempts without changing the plan."""
    from .recovery import retry_study

    return retry_study(self, failed_only=failed_only)

run()

Execute this newly created durable study sequentially.

Source code in src/vamos/experiment/study/models.py

def run(self) -> Study:
    """Execute this newly created durable study sequentially."""
    from .execution import execute_study

    return execute_study(self)

summarize()

Return a deterministic in-memory summary without writing files.

Source code in src/vamos/experiment/study/models.py

def summarize(self) -> StudySummary:
    """Return a deterministic in-memory summary without writing files."""
    from .projection import project_study

    return project_study(self).summary

StudyLoadLimits dataclass

Source code in src/vamos/experiment/study/limits.py

@dataclass(frozen=True, slots=True)
class StudyLoadLimits:
    max_manifest_bytes: int = 8 * 1024 * 1024
    max_spec_bytes: int = 8 * 1024 * 1024
    max_plan_bytes: int = 64 * 1024 * 1024
    max_task_bytes: int = 1024 * 1024
    max_attempt_bytes: int = 2 * 1024 * 1024
    max_event_bytes: int = 2 * 1024 * 1024
    max_tasks: int = 100_000
    max_documents: int = 300_000
    max_total_bytes: int = 512 * 1024 * 1024
    max_json_depth: int = 64
    max_string_bytes: int = 64 * 1024

    def __post_init__(self) -> None:
        for item in fields(self):
            value = getattr(self, item.name)
            if isinstance(value, bool) or not isinstance(value, int) or value < 1:
                raise ValueError(f"StudyLoadLimits.{item.name} must be a positive integer.")

StudyReport dataclass

Deterministic current-state report for one canonical study.

Source code in src/vamos/experiment/study/report_models.py

@dataclass(frozen=True, slots=True)
class StudyReport:
    """Deterministic current-state report for one canonical study."""

    study_id: str
    plan_id: str
    state: str
    on_error: str
    max_attempts_per_task: int
    created_at: str
    updated_at: str
    counts: Mapping[str, int]
    attempts: tuple[StudyAttemptReport, ...]
    total_attempt_count: int
    verified_run_count: int
    event_head_sequence: int
    event_head_sha256: str
    checkpoint_sequence: int
    checkpoint_event_sha256: str
    journal_checkpoint_relation: str
    reconciliation_required: bool
    runnable_task_ids: tuple[str, ...]
    retryable_task_ids: tuple[str, ...]
    runnable_work: bool
    retryable_failed_work: bool
    changed: bool
    issues: tuple[StudyIssue, ...]
    next_actions: tuple[str, ...]

    def as_dict(self) -> dict[str, Any]:
        return {
            "document_type": "vamos.study-report",
            "schema_version": "1.0.0",
            "study_id": self.study_id,
            "plan_id": self.plan_id,
            "state": self.state,
            "policy": {
                "on_error": self.on_error,
                "max_attempts_per_task": self.max_attempts_per_task,
            },
            "timestamps": {"created_at": self.created_at, "updated_at": self.updated_at},
            "counts": dict(self.counts),
            "attempts": [item.as_dict() for item in self.attempts],
            "total_attempt_count": self.total_attempt_count,
            "verified_run_count": self.verified_run_count,
            "journal": {
                "head_sequence": self.event_head_sequence,
                "head_sha256": self.event_head_sha256,
            },
            "checkpoint": {
                "sequence": self.checkpoint_sequence,
                "event_sha256": self.checkpoint_event_sha256,
                "relation": self.journal_checkpoint_relation,
                "reconciliation_required": self.reconciliation_required,
            },
            "runnable_task_ids": list(self.runnable_task_ids),
            "retryable_task_ids": list(self.retryable_task_ids),
            "runnable_work": self.runnable_work,
            "retryable_failed_work": self.retryable_failed_work,
            "changed": self.changed,
            "issues": [item.as_dict() for item in self.issues],
            "next_actions": list(self.next_actions),
        }

StudySummary dataclass

Deterministic, derived, in-memory summary of every planned task.

Source code in src/vamos/experiment/study/report_models.py

@dataclass(frozen=True, slots=True)
class StudySummary:
    """Deterministic, derived, in-memory summary of every planned task."""

    study_id: str
    plan_id: str
    state: str
    generated_at: str
    root_manifest_sha256: str
    event_head_sequence: int
    event_head_sha256: str
    rows: tuple[StudySummaryRow, ...]
    issues: tuple[StudyIssue, ...]

    def as_dict(self) -> dict[str, Any]:
        return {
            "document_type": "vamos.study-summary",
            "schema_version": "1.0.0",
            "study_id": self.study_id,
            "plan_id": self.plan_id,
            "state": self.state,
            "generated_at": self.generated_at,
            "root_manifest_sha256": self.root_manifest_sha256,
            "event_head": {
                "sequence": self.event_head_sequence,
                "sha256": self.event_head_sha256,
            },
            "rows": [item.as_dict() for item in self.rows],
            "issues": [item.as_dict() for item in self.issues],
        }

plan_study(spec, *, output=None)

Resolve and explain spec without creating a study or running a task.

Source code in src/vamos/experiment/study/preflight.py

def plan_study(spec: StudySpec, *, output: str | Path | None = None) -> StudyPlanReport:
    """Resolve and explain ``spec`` without creating a study or running a task."""
    if not isinstance(spec, StudySpec):
        from .errors import InvalidStudySpecError

        raise InvalidStudySpecError(
            operation="plan study",
            reason="INVALID_STUDY_SPEC",
            expected="validated StudySpec",
            actual=type(spec).__name__,
            action="Construct vamos.StudySpec(...) before calling plan_study.",
        )
    plan = resolve_spec(spec)
    destination = inspect_study_output(output)
    summaries = _summarize_plan(plan)
    errors = _output_errors(destination)
    status: PlanStatus = "ready" if not errors else "blocked"
    warnings: tuple[str, ...] = (
        ("Output availability is advisory and is not reserved; another process may occupy it after planning.",)
        if output is not None
        else ()
    )
    if summaries.reconstructable:
        warnings += ("Resolved built-ins are reconstructable; exact replayability is verified only after a run is published.",)
    next_actions = (
        (
            "Call vamos.create_study(spec, output=...) to publish this exact plan."
            if output is None
            else f"Call vamos.create_study(spec, output={os.fspath(output)!r}) to publish this exact plan."
        )
        if status == "ready"
        else "Choose an absent output path, then plan again before calling vamos.create_study(...).",
    )
    return StudyPlanReport(
        plan=plan,
        status=status,
        valid=True,
        total_evaluation_budget=summaries.total_evaluation_budget,
        problem_ids=summaries.problem_ids,
        algorithm_ids=summaries.algorithm_ids,
        operator_ids=summaries.operator_ids,
        backend_ids=summaries.backend_ids,
        seeds=summaries.seeds,
        population_sizes=summaries.population_sizes,
        termination_categories=summaries.termination_categories,
        failure_policy=spec.on_error,
        reconstructable=summaries.reconstructable,
        duplicate_tasks=False,
        output=destination,
        warnings=warnings,
        errors=errors,
        next_actions=next_actions,
    )

create_study(spec, *, output, limits=None)

Resolve and atomically publish a study without executing any task.

Source code in src/vamos/experiment/study/creation.py

def create_study(
    spec: StudySpec,
    *,
    output: str | Path,
    limits: StudyLoadLimits | None = None,
) -> Study:
    """Resolve and atomically publish a study without executing any task."""
    if not isinstance(spec, StudySpec):
        from .errors import InvalidStudySpecError

        raise InvalidStudySpecError(
            operation="create study",
            reason="INVALID_STUDY_SPEC",
            expected="validated StudySpec",
            actual=type(spec).__name__,
            action="Construct vamos.StudySpec(...) before calling create_study.",
        )
    destination = Path(output).absolute()
    try:
        destination.parent.mkdir(parents=True, exist_ok=True)
    except OSError as exc:
        raise StudyInfrastructureError(
            operation="create study",
            reason="ATOMIC_PUBLICATION_FAILED",
            path=destination.parent,
            expected="writable parent directory for sibling staging",
            actual=type(exc).__name__,
            action="Choose a writable output parent; no study directory was published.",
        ) from exc
    _reject_existing(destination)
    plan = resolve_spec(spec)
    return publish_study(spec, plan=plan, destination=destination, limits=limits)

load_study(path, *, limits=None)

Load and fully verify one canonical study without executing code.

Source code in src/vamos/experiment/study/loading.py

def load_study(path: str | Path, *, limits: StudyLoadLimits | None = None) -> Study:
    """Load and fully verify one canonical study without executing code."""
    return _load_study(path, limits=limits, run_verification="all", tolerate_run_errors=False)

Algorithm Configuration

vamos.engine.algorithm.config.nsgaii

NSGA-II configuration.

NSGAIIConfig dataclass

Bases: _SerializableConfig

Source code in src/vamos/engine/algorithm/config/nsgaii.py

@dataclass(frozen=True)
class NSGAIIConfig(_SerializableConfig):
    pop_size: int
    crossover: tuple[str, dict[str, Any]]
    mutation: tuple[str, dict[str, Any]]
    selection: tuple[str, dict[str, Any]]
    offspring_size: int | None = None
    steady_state: bool = False
    replacement_size: int | None = None
    repair: RepairConfigValue = "auto"
    external_archive: ExternalArchiveConfig | None = None
    initializer: dict[str, Any] | None = None
    mutation_prob_factor: float | None = None
    result_mode: ResultMode | None = None
    constraint_mode: ConstraintModeStr = "feasibility"
    track_genealogy: bool = False
    immigration: dict[str, Any] | None = None
    parent_selection_filter: Any | None = None
    live_callback_mode: LiveCallbackMode = "nd_only"
    generation_callback: Any | None = None
    generation_callback_copy: bool = True

    @classmethod
    def default(
        cls,
        pop_size: int = 100,
        n_var: int | None = None,
        encoding: str | None = None,
    ) -> NSGAIIConfig:
        """
        Create a default NSGA-II configuration with sensible defaults.

        Parameters
        ----------
        pop_size
            Population size.
        n_var
            Number of variables used for the default mutation probability.
        encoding
            Problem encoding. If omitted, defaults to ``"real"``.

        Returns
        -------
        NSGAIIConfig
            Frozen configuration ready to use.
        """
        normalized = normalize_encoding(encoding, default="real")
        mut_prob = 1.0 / n_var if n_var else 0.1
        builder = cls.builder().pop_size(pop_size).selection("tournament")

        if normalized == "permutation":
            return builder.crossover("ox").mutation("swap").build()
        if normalized == "binary":
            return builder.crossover("uniform", prob=0.9).mutation("bitflip", prob=mut_prob).build()
        if normalized == "integer":
            return builder.crossover("sbx", prob=0.9, eta=20.0).mutation("pm", prob=mut_prob, eta=20.0).build()
        if normalized == "mixed":
            return builder.crossover("mixed", prob=0.9).mutation("mixed", prob=mut_prob).build()

        return builder.crossover("sbx", prob=1.0, eta=20.0).mutation("pm", prob=mut_prob, eta=20.0).build()

    @classmethod
    def builder(cls) -> _NSGAIIConfigBuilder:
        return _NSGAIIConfigBuilder()

default(pop_size=100, n_var=None, encoding=None) classmethod

Create a default NSGA-II configuration with sensible defaults.

Parameters:

Name	Type	Description	Default
pop_size	int	Population size.	100
n_var	int | None	Number of variables used for the default mutation probability.	None
encoding	str | None	Problem encoding. If omitted, defaults to "real".	None

Returns:

Type	Description
NSGAIIConfig	Frozen configuration ready to use.

Source code in src/vamos/engine/algorithm/config/nsgaii.py

@classmethod
def default(
    cls,
    pop_size: int = 100,
    n_var: int | None = None,
    encoding: str | None = None,
) -> NSGAIIConfig:
    """
    Create a default NSGA-II configuration with sensible defaults.

    Parameters
    ----------
    pop_size
        Population size.
    n_var
        Number of variables used for the default mutation probability.
    encoding
        Problem encoding. If omitted, defaults to ``"real"``.

    Returns
    -------
    NSGAIIConfig
        Frozen configuration ready to use.
    """
    normalized = normalize_encoding(encoding, default="real")
    mut_prob = 1.0 / n_var if n_var else 0.1
    builder = cls.builder().pop_size(pop_size).selection("tournament")

    if normalized == "permutation":
        return builder.crossover("ox").mutation("swap").build()
    if normalized == "binary":
        return builder.crossover("uniform", prob=0.9).mutation("bitflip", prob=mut_prob).build()
    if normalized == "integer":
        return builder.crossover("sbx", prob=0.9, eta=20.0).mutation("pm", prob=mut_prob, eta=20.0).build()
    if normalized == "mixed":
        return builder.crossover("mixed", prob=0.9).mutation("mixed", prob=mut_prob).build()

    return builder.crossover("sbx", prob=1.0, eta=20.0).mutation("pm", prob=mut_prob, eta=20.0).build()

_NSGAIIConfigBuilder

Bases: _ConfigBuilderState, _PopSizeBuilder, _CrossoverBuilder, _MutationBuilder, _SelectionBuilder, _RepairBuilder, _InitializerBuilder, _MutationProbFactorBuilder, _ResultArchiveBuilder, _ConstraintModeBuilder, _TrackGenealogyBuilder

Fluent builder for NSGA-II configs.

Source code in src/vamos/engine/algorithm/config/nsgaii.py

class _NSGAIIConfigBuilder(
    _ConfigBuilderState,
    _PopSizeBuilder,
    _CrossoverBuilder,
    _MutationBuilder,
    _SelectionBuilder,
    _RepairBuilder,
    _InitializerBuilder,
    _MutationProbFactorBuilder,
    _ResultArchiveBuilder,
    _ConstraintModeBuilder,
    _TrackGenealogyBuilder,
):
    """
    Fluent builder for NSGA-II configs.
    """

    def offspring_size(self, value: int) -> _NSGAIIConfigBuilder:
        if value <= 0:
            raise ValueError("offspring size must be positive.")
        self._cfg["offspring_size"] = value
        return self

    def steady_state(self, enabled: bool = True) -> _NSGAIIConfigBuilder:
        """Enable steady-state mode (incremental replacement)."""
        self._cfg["steady_state"] = bool(enabled)
        return self

    def replacement_size(self, value: int) -> _NSGAIIConfigBuilder:
        if value <= 0:
            raise ValueError("replacement size must be positive.")
        self._cfg["replacement_size"] = value
        return self

    def immigration(self, config: dict[str, Any] | None) -> _NSGAIIConfigBuilder:
        if config is None:
            self._cfg["immigration"] = None
        else:
            self._cfg["immigration"] = dict(config)
        return self

    def parent_selection_filter(self, fn: Any | None) -> _NSGAIIConfigBuilder:
        self._cfg["parent_selection_filter"] = fn
        return self

    def live_callback_mode(self, mode: LiveCallbackMode) -> _NSGAIIConfigBuilder:
        self._cfg["live_callback_mode"] = str(mode)
        return self

    def generation_callback(
        self,
        fn: Any | None,
        *,
        copy_arrays: bool = True,
    ) -> _NSGAIIConfigBuilder:
        self._cfg["generation_callback"] = fn
        self._cfg["generation_callback_copy"] = bool(copy_arrays)
        return self

    def build(self) -> NSGAIIConfig:
        _require_fields(
            self._cfg,
            ("crossover", "mutation"),
            "NSGA-II",
        )
        _validate_operators(self._cfg)
        pop_size = int(self._cfg.get("pop_size", 100))
        selection = self._cfg.get("selection", ("tournament", {}))
        return NSGAIIConfig(
            pop_size=pop_size,
            crossover=self._cfg["crossover"],
            mutation=self._cfg["mutation"],
            selection=selection,
            offspring_size=self._cfg.get("offspring_size"),
            steady_state=bool(self._cfg.get("steady_state", False)),
            replacement_size=self._cfg.get("replacement_size"),
            repair=self._cfg.get("repair", "auto"),
            external_archive=self._cfg.get("external_archive"),
            initializer=self._cfg.get("initializer"),
            mutation_prob_factor=self._cfg.get("mutation_prob_factor"),
            result_mode=self._cfg.get("result_mode", "non_dominated"),
            constraint_mode=self._cfg.get("constraint_mode", "feasibility"),
            track_genealogy=bool(self._cfg.get("track_genealogy", False)),
            immigration=self._cfg.get("immigration"),
            parent_selection_filter=self._cfg.get("parent_selection_filter"),
            live_callback_mode=self._cfg.get("live_callback_mode", "nd_only"),
            generation_callback=self._cfg.get("generation_callback"),
            generation_callback_copy=bool(self._cfg.get("generation_callback_copy", True)),
        )

steady_state(enabled=True)

Enable steady-state mode (incremental replacement).

Source code in src/vamos/engine/algorithm/config/nsgaii.py

def steady_state(self, enabled: bool = True) -> _NSGAIIConfigBuilder:
    """Enable steady-state mode (incremental replacement)."""
    self._cfg["steady_state"] = bool(enabled)
    return self

vamos.engine.algorithm.config.moead

MOEA/D configuration.

MOEADConfig dataclass

Bases: _SerializableConfig

Source code in src/vamos/engine/algorithm/config/moead.py

@dataclass(frozen=True)
class MOEADConfig(_SerializableConfig):
    pop_size: int
    batch_size: int
    neighbor_size: int
    delta: float
    replace_limit: int
    crossover: tuple[str, dict[str, Any]]
    mutation: tuple[str, dict[str, Any]]
    aggregation: tuple[str, dict[str, Any]]
    weight_vectors: dict[str, int | str | None] | None
    constraint_mode: ConstraintModeStr = "feasibility"
    repair: RepairConfigValue = "auto"
    initializer: dict[str, Any] | None = None
    mutation_prob_factor: float | None = None
    use_numba_variation: bool | None = None
    track_genealogy: bool = False
    result_mode: ResultMode | None = None
    external_archive: ExternalArchiveConfig | None = None

    @classmethod
    def default(
        cls,
        pop_size: int | None = None,
        n_var: int | None = None,
        n_obj: int = 3,
    ) -> MOEADConfig:
        """Create a default MOEA/D configuration with sensible defaults."""
        divisions = 99 if n_obj == 2 else (12 if n_obj == 3 else 6)
        if pop_size is not None and n_obj == 2:
            divisions = max(1, int(pop_size) - 1)
        if pop_size is None:
            pop_size = divisions + 1 if n_obj == 2 else comb(divisions + n_obj - 1, n_obj - 1)
        mut_prob = 1.0 / n_var if n_var else 0.1
        return (
            cls.builder()
            .pop_size(pop_size)
            .batch_size(1)
            .neighbor_size(20)
            .delta(0.9)
            .replace_limit(2)
            .crossover("de", cr=1.0, f=0.5)
            .mutation("pm", prob=mut_prob, eta=20.0)
            .aggregation("pbi", theta=5.0)
            .weight_vectors(divisions=divisions)
            .build()
        )

    @classmethod
    def builder(cls) -> _MOEADConfigBuilder:
        return _MOEADConfigBuilder()

default(pop_size=None, n_var=None, n_obj=3) classmethod

Create a default MOEA/D configuration with sensible defaults.

Source code in src/vamos/engine/algorithm/config/moead.py

@classmethod
def default(
    cls,
    pop_size: int | None = None,
    n_var: int | None = None,
    n_obj: int = 3,
) -> MOEADConfig:
    """Create a default MOEA/D configuration with sensible defaults."""
    divisions = 99 if n_obj == 2 else (12 if n_obj == 3 else 6)
    if pop_size is not None and n_obj == 2:
        divisions = max(1, int(pop_size) - 1)
    if pop_size is None:
        pop_size = divisions + 1 if n_obj == 2 else comb(divisions + n_obj - 1, n_obj - 1)
    mut_prob = 1.0 / n_var if n_var else 0.1
    return (
        cls.builder()
        .pop_size(pop_size)
        .batch_size(1)
        .neighbor_size(20)
        .delta(0.9)
        .replace_limit(2)
        .crossover("de", cr=1.0, f=0.5)
        .mutation("pm", prob=mut_prob, eta=20.0)
        .aggregation("pbi", theta=5.0)
        .weight_vectors(divisions=divisions)
        .build()
    )

_MOEADConfigBuilder

Bases: _ConfigBuilderState, _PopSizeBuilder, _CrossoverBuilder, _MutationBuilder, _RepairBuilder, _InitializerBuilder, _MutationProbFactorBuilder, _ConstraintModeBuilder, _TrackGenealogyBuilder, _ResultArchiveBuilder

Declarative configuration holder for MOEA/D settings.

Examples: # Fluent builder cfg = MOEADConfig.builder().pop_size(100).neighbor_size(20).build()

# Quick default configuration
cfg = MOEADConfig.default()

Source code in src/vamos/engine/algorithm/config/moead.py

class _MOEADConfigBuilder(
    _ConfigBuilderState,
    _PopSizeBuilder,
    _CrossoverBuilder,
    _MutationBuilder,
    _RepairBuilder,
    _InitializerBuilder,
    _MutationProbFactorBuilder,
    _ConstraintModeBuilder,
    _TrackGenealogyBuilder,
    _ResultArchiveBuilder,
):
    """
    Declarative configuration holder for MOEA/D settings.

    Examples:
        # Fluent builder
        cfg = MOEADConfig.builder().pop_size(100).neighbor_size(20).build()

        # Quick default configuration
        cfg = MOEADConfig.default()
    """

    def batch_size(self, value: int) -> _MOEADConfigBuilder:
        self._cfg["batch_size"] = value
        return self

    def neighbor_size(self, value: int) -> _MOEADConfigBuilder:
        self._cfg["neighbor_size"] = value
        return self

    def delta(self, value: float) -> _MOEADConfigBuilder:
        self._cfg["delta"] = value
        return self

    def replace_limit(self, value: int) -> _MOEADConfigBuilder:
        self._cfg["replace_limit"] = value
        return self

    @overload
    def aggregation(self, method: AggregationName, **kwargs: Any) -> _MOEADConfigBuilder: ...

    @overload
    def aggregation(self, method: str, **kwargs: Any) -> _MOEADConfigBuilder: ...

    def aggregation(self, method: str, **kwargs: Any) -> _MOEADConfigBuilder:
        self._cfg["aggregation"] = (method, kwargs)
        return self

    def weight_vectors(self, *, path: str | None = None, divisions: int | None = None) -> _MOEADConfigBuilder:
        self._cfg["weight_vectors"] = {"path": path, "divisions": divisions}
        return self

    def use_numba_variation(self, enabled: bool = True) -> _MOEADConfigBuilder:
        self._cfg["use_numba_variation"] = bool(enabled)
        return self

    def build(self) -> MOEADConfig:
        _require_fields(
            self._cfg,
            (
                "pop_size",
                "neighbor_size",
                "delta",
                "replace_limit",
                "crossover",
                "mutation",
                "aggregation",
            ),
            "MOEA/D",
        )
        _validate_operators(self._cfg)
        return MOEADConfig(
            pop_size=self._cfg["pop_size"],
            batch_size=int(self._cfg.get("batch_size", 1)),
            neighbor_size=self._cfg["neighbor_size"],
            delta=self._cfg["delta"],
            replace_limit=self._cfg["replace_limit"],
            crossover=self._cfg["crossover"],
            mutation=self._cfg["mutation"],
            aggregation=self._cfg["aggregation"],
            weight_vectors=self._cfg.get("weight_vectors"),
            constraint_mode=self._cfg.get("constraint_mode", "feasibility"),
            repair=self._cfg.get("repair", "auto"),
            initializer=self._cfg.get("initializer"),
            mutation_prob_factor=self._cfg.get("mutation_prob_factor"),
            use_numba_variation=self._cfg.get("use_numba_variation"),
            track_genealogy=bool(self._cfg.get("track_genealogy", False)),
            result_mode=self._cfg.get("result_mode", "non_dominated"),
            external_archive=self._cfg.get("external_archive"),
        )

vamos.engine.algorithm.config.nsgaiii

NSGA-III configuration.

NSGAIIIConfig dataclass

Bases: _SerializableConfig

Source code in src/vamos/engine/algorithm/config/nsgaiii.py

@dataclass(frozen=True)
class NSGAIIIConfig(_SerializableConfig):
    pop_size: int
    crossover: tuple[str, dict[str, Any]]
    mutation: tuple[str, dict[str, Any]]
    selection: tuple[str, dict[str, Any]]
    reference_directions: dict[str, int | str | None]
    enforce_ref_dirs: bool = True
    pop_size_auto: bool = False
    constraint_mode: ConstraintModeStr = "feasibility"
    repair: RepairConfigValue = "auto"
    initializer: dict[str, Any] | None = None
    mutation_prob_factor: float | None = None
    track_genealogy: bool = False
    result_mode: ResultMode | None = None
    external_archive: ExternalArchiveConfig | None = None

    @classmethod
    def default(
        cls,
        pop_size: int | None = None,
        n_var: int | None = None,
        n_obj: int = 3,
    ) -> NSGAIIIConfig:
        """
        Create a default NSGA-III configuration.

        Parameters
        ----------
        pop_size
            Population size. Defaults to the generated reference-direction count.
        n_var
            Number of variables used for the default mutation probability.
        n_obj
            Number of objectives used to choose the default reference directions.
        """
        mut_prob = 1.0 / n_var if n_var else 0.1
        divisions = 12 if n_obj == 3 else 6
        if pop_size is None:
            pop_size = comb(divisions + n_obj - 1, n_obj - 1)
        return (
            cls.builder()
            .pop_size(pop_size)
            .crossover("sbx", prob=1.0, eta=30.0)
            .mutation("pm", prob=mut_prob, eta=20.0)
            .selection("tournament")
            .reference_directions(divisions=divisions)
            .pop_size_auto(True)
            .build()
        )

    @classmethod
    def builder(cls) -> _NSGAIIIConfigBuilder:
        return _NSGAIIIConfigBuilder()

default(pop_size=None, n_var=None, n_obj=3) classmethod

Create a default NSGA-III configuration.

Parameters:

Name	Type	Description	Default
pop_size	int | None	Population size. Defaults to the generated reference-direction count.	None
n_var	int | None	Number of variables used for the default mutation probability.	None
n_obj	int	Number of objectives used to choose the default reference directions.	3

Source code in src/vamos/engine/algorithm/config/nsgaiii.py

@classmethod
def default(
    cls,
    pop_size: int | None = None,
    n_var: int | None = None,
    n_obj: int = 3,
) -> NSGAIIIConfig:
    """
    Create a default NSGA-III configuration.

    Parameters
    ----------
    pop_size
        Population size. Defaults to the generated reference-direction count.
    n_var
        Number of variables used for the default mutation probability.
    n_obj
        Number of objectives used to choose the default reference directions.
    """
    mut_prob = 1.0 / n_var if n_var else 0.1
    divisions = 12 if n_obj == 3 else 6
    if pop_size is None:
        pop_size = comb(divisions + n_obj - 1, n_obj - 1)
    return (
        cls.builder()
        .pop_size(pop_size)
        .crossover("sbx", prob=1.0, eta=30.0)
        .mutation("pm", prob=mut_prob, eta=20.0)
        .selection("tournament")
        .reference_directions(divisions=divisions)
        .pop_size_auto(True)
        .build()
    )

_NSGAIIIConfigBuilder

Bases: _ConfigBuilderState, _PopSizeBuilder, _CrossoverBuilder, _MutationBuilder, _SelectionBuilder, _RepairBuilder, _InitializerBuilder, _MutationProbFactorBuilder, _ResultArchiveBuilder, _ConstraintModeBuilder, _TrackGenealogyBuilder

Declarative configuration holder for NSGA-III settings.

Examples: cfg = NSGAIIIConfig.default(n_obj=3) cfg = NSGAIIIConfig.builder().pop_size(92).crossover("sbx", prob=1.0).build()

Source code in src/vamos/engine/algorithm/config/nsgaiii.py

class _NSGAIIIConfigBuilder(
    _ConfigBuilderState,
    _PopSizeBuilder,
    _CrossoverBuilder,
    _MutationBuilder,
    _SelectionBuilder,
    _RepairBuilder,
    _InitializerBuilder,
    _MutationProbFactorBuilder,
    _ResultArchiveBuilder,
    _ConstraintModeBuilder,
    _TrackGenealogyBuilder,
):
    """
    Declarative configuration holder for NSGA-III settings.

    Examples:
        cfg = NSGAIIIConfig.default(n_obj=3)
        cfg = NSGAIIIConfig.builder().pop_size(92).crossover("sbx", prob=1.0).build()
    """


    def reference_directions(
        self,
        *,
        path: str | None = None,
        divisions: int | None = None,
    ) -> _NSGAIIIConfigBuilder:
        self._cfg["reference_directions"] = {"path": path, "divisions": divisions}
        return self

    def enforce_ref_dirs(self, enabled: bool = True) -> _NSGAIIIConfigBuilder:
        self._cfg["enforce_ref_dirs"] = bool(enabled)
        return self

    def pop_size_auto(self, enabled: bool = True) -> _NSGAIIIConfigBuilder:
        self._cfg["pop_size_auto"] = bool(enabled)
        return self

    def build(self) -> NSGAIIIConfig:
        _require_fields(
            self._cfg,
            ("pop_size", "crossover", "mutation", "selection"),
            "NSGA-III",
        )
        _validate_operators(self._cfg)
        ref_dirs = self._cfg.get("reference_directions", {})
        return NSGAIIIConfig(
            pop_size=self._cfg["pop_size"],
            crossover=self._cfg["crossover"],
            mutation=self._cfg["mutation"],
            selection=self._cfg["selection"],
            reference_directions=ref_dirs,
            enforce_ref_dirs=bool(self._cfg.get("enforce_ref_dirs", True)),
            pop_size_auto=bool(self._cfg.get("pop_size_auto", False)),
            constraint_mode=self._cfg.get("constraint_mode", "feasibility"),
            repair=self._cfg.get("repair", "auto"),
            initializer=self._cfg.get("initializer"),
            mutation_prob_factor=self._cfg.get("mutation_prob_factor"),
            track_genealogy=bool(self._cfg.get("track_genealogy", False)),
            result_mode=self._cfg.get("result_mode", "population"),
            external_archive=self._cfg.get("external_archive"),
        )

vamos.engine.algorithm.config.smsemoa

SMS-EMOA configuration.

SMSEMOAConfig dataclass

Bases: _SerializableConfig

Source code in src/vamos/engine/algorithm/config/smsemoa.py

@dataclass(frozen=True)
class SMSEMOAConfig(_SerializableConfig):
    pop_size: int
    crossover: tuple[str, dict[str, Any]]
    mutation: tuple[str, dict[str, Any]]
    selection: tuple[str, dict[str, Any]]
    reference_point: dict[str, Any]
    eliminate_duplicates: bool = False
    constraint_mode: ConstraintModeStr = "feasibility"
    repair: RepairConfigValue = "auto"
    initializer: dict[str, Any] | None = None
    mutation_prob_factor: float | None = None
    track_genealogy: bool = False
    result_mode: ResultMode | None = None
    external_archive: ExternalArchiveConfig | None = None

    @classmethod
    def default(
        cls,
        pop_size: int = 100,
        n_var: int | None = None,
    ) -> SMSEMOAConfig:
        """Create a default SMS-EMOA configuration."""
        mut_prob = 1.0 / n_var if n_var else 0.1
        return (
            cls.builder()
            .pop_size(pop_size)
            .crossover("sbx", prob=1.0, eta=20.0)
            .mutation("pm", prob=mut_prob, eta=20.0)
            .selection("random")
            .reference_point(adaptive=True)
            .build()
        )

    @classmethod
    def builder(cls) -> _SMSEMOAConfigBuilder:
        return _SMSEMOAConfigBuilder()

default(pop_size=100, n_var=None) classmethod

Create a default SMS-EMOA configuration.

Source code in src/vamos/engine/algorithm/config/smsemoa.py

@classmethod
def default(
    cls,
    pop_size: int = 100,
    n_var: int | None = None,
) -> SMSEMOAConfig:
    """Create a default SMS-EMOA configuration."""
    mut_prob = 1.0 / n_var if n_var else 0.1
    return (
        cls.builder()
        .pop_size(pop_size)
        .crossover("sbx", prob=1.0, eta=20.0)
        .mutation("pm", prob=mut_prob, eta=20.0)
        .selection("random")
        .reference_point(adaptive=True)
        .build()
    )

_SMSEMOAConfigBuilder

Bases: _ConfigBuilderState, _PopSizeBuilder, _CrossoverBuilder, _MutationBuilder, _SelectionBuilder, _RepairBuilder, _InitializerBuilder, _MutationProbFactorBuilder, _ConstraintModeBuilder, _TrackGenealogyBuilder, _ResultArchiveBuilder

Declarative configuration holder for SMS-EMOA settings.

Examples: cfg = SMSEMOAConfig.default() cfg = SMSEMOAConfig.builder().pop_size(100).crossover("sbx", prob=1.0).build()

Source code in src/vamos/engine/algorithm/config/smsemoa.py

class _SMSEMOAConfigBuilder(
    _ConfigBuilderState,
    _PopSizeBuilder,
    _CrossoverBuilder,
    _MutationBuilder,
    _SelectionBuilder,
    _RepairBuilder,
    _InitializerBuilder,
    _MutationProbFactorBuilder,
    _ConstraintModeBuilder,
    _TrackGenealogyBuilder,
    _ResultArchiveBuilder,
):
    """
    Declarative configuration holder for SMS-EMOA settings.

    Examples:
        cfg = SMSEMOAConfig.default()
        cfg = SMSEMOAConfig.builder().pop_size(100).crossover("sbx", prob=1.0).build()
    """

    def eliminate_duplicates(self, enabled: bool = True) -> _SMSEMOAConfigBuilder:
        self._cfg["eliminate_duplicates"] = bool(enabled)
        return self

    def reference_point(
        self,
        *,
        vector: Any = None,
        offset: float = 1.0,
        adaptive: bool = True,
    ) -> _SMSEMOAConfigBuilder:
        self._cfg["reference_point"] = {
            "vector": vector,
            "offset": offset,
            "adaptive": adaptive,
        }
        return self

    def build(self) -> SMSEMOAConfig:
        _require_fields(
            self._cfg,
            ("pop_size", "crossover", "mutation", "selection"),
            "SMS-EMOA",
        )
        _validate_operators(self._cfg)
        reference_point = self._cfg.get("reference_point", {"offset": 1.0, "adaptive": True})
        return SMSEMOAConfig(
            pop_size=self._cfg["pop_size"],
            crossover=self._cfg["crossover"],
            mutation=self._cfg["mutation"],
            selection=self._cfg["selection"],
            reference_point=reference_point,
            eliminate_duplicates=bool(self._cfg.get("eliminate_duplicates", False)),
            constraint_mode=self._cfg.get("constraint_mode", "feasibility"),
            repair=self._cfg.get("repair", "auto"),
            initializer=self._cfg.get("initializer"),
            mutation_prob_factor=self._cfg.get("mutation_prob_factor"),
            track_genealogy=bool(self._cfg.get("track_genealogy", False)),
            result_mode=self._cfg.get("result_mode", "non_dominated"),
            external_archive=self._cfg.get("external_archive"),
        )

vamos.engine.algorithm.config.spea2

SPEA2 configuration.

SPEA2Config dataclass

Bases: _SerializableConfig

Source code in src/vamos/engine/algorithm/config/spea2.py

@dataclass(frozen=True)
class SPEA2Config(_SerializableConfig):
    pop_size: int
    archive_size: int  # Internal archive (part of SPEA2 algorithm)
    crossover: tuple[str, dict[str, Any]]
    mutation: tuple[str, dict[str, Any]]
    selection: tuple[str, dict[str, Any]]
    k_neighbors: int | None = None
    repair: RepairConfigValue = "auto"
    initializer: dict[str, Any] | None = None
    mutation_prob_factor: float | None = None
    constraint_mode: ConstraintModeStr = "feasibility"
    track_genealogy: bool = False
    result_mode: ResultMode | None = None
    external_archive: ExternalArchiveConfig | None = None

    @classmethod
    def default(
        cls,
        pop_size: int = 100,
        n_var: int | None = None,
    ) -> SPEA2Config:
        """Create a default SPEA2 configuration."""
        mut_prob = 1.0 / n_var if n_var else 0.1
        return (
            cls.builder()
            .pop_size(pop_size)
            .archive_size(pop_size)
            .crossover("sbx", prob=1.0, eta=20.0)
            .mutation("pm", prob=mut_prob, eta=20.0)
            .selection("tournament")
            .build()
        )

    @classmethod
    def builder(cls) -> _SPEA2ConfigBuilder:
        return _SPEA2ConfigBuilder()

default(pop_size=100, n_var=None) classmethod

Create a default SPEA2 configuration.

Source code in src/vamos/engine/algorithm/config/spea2.py

@classmethod
def default(
    cls,
    pop_size: int = 100,
    n_var: int | None = None,
) -> SPEA2Config:
    """Create a default SPEA2 configuration."""
    mut_prob = 1.0 / n_var if n_var else 0.1
    return (
        cls.builder()
        .pop_size(pop_size)
        .archive_size(pop_size)
        .crossover("sbx", prob=1.0, eta=20.0)
        .mutation("pm", prob=mut_prob, eta=20.0)
        .selection("tournament")
        .build()
    )

_SPEA2ConfigBuilder

Bases: _ConfigBuilderState, _PopSizeBuilder, _CrossoverBuilder, _MutationBuilder, _SelectionBuilder, _RepairBuilder, _InitializerBuilder, _MutationProbFactorBuilder, _ConstraintModeBuilder, _TrackGenealogyBuilder, _ResultArchiveBuilder

Declarative configuration holder for SPEA2 settings.

Examples: cfg = SPEA2Config.default() cfg = SPEA2Config.builder().pop_size(100).archive_size(100).build()

Source code in src/vamos/engine/algorithm/config/spea2.py

class _SPEA2ConfigBuilder(
    _ConfigBuilderState,
    _PopSizeBuilder,
    _CrossoverBuilder,
    _MutationBuilder,
    _SelectionBuilder,
    _RepairBuilder,
    _InitializerBuilder,
    _MutationProbFactorBuilder,
    _ConstraintModeBuilder,
    _TrackGenealogyBuilder,
    _ResultArchiveBuilder,
):
    """
    Declarative configuration holder for SPEA2 settings.

    Examples:
        cfg = SPEA2Config.default()
        cfg = SPEA2Config.builder().pop_size(100).archive_size(100).build()
    """

    def archive_size(self, value: int) -> _SPEA2ConfigBuilder:
        self._cfg["archive_size"] = value
        return self

    def k_neighbors(self, value: int) -> _SPEA2ConfigBuilder:
        self._cfg["k_neighbors"] = value
        return self
    def external_archive(self, capacity: int | None = None, **kwargs: Any) -> _SPEA2ConfigBuilder:
        """Configure an external archive for result storage.

        Note
        ----
        This is separate from ``archive_size``, which controls the internal SPEA2 archive.

        Parameters
        ----------
        capacity
            Maximum number of solutions. ``None`` means unbounded.
        **kwargs
            Forwarded to :class:`ExternalArchiveConfig`.
        """
        self._cfg["external_archive"] = _build_external_archive_config(capacity, kwargs)
        self._cfg.setdefault("result_mode", "non_dominated")
        return self

    def build(self) -> SPEA2Config:
        _require_fields(
            self._cfg,
            ("pop_size", "archive_size", "crossover", "mutation", "selection"),
            "SPEA2",
        )
        _validate_operators(self._cfg)
        return SPEA2Config(
            pop_size=self._cfg["pop_size"],
            archive_size=self._cfg["archive_size"],
            crossover=self._cfg["crossover"],
            mutation=self._cfg["mutation"],
            selection=self._cfg["selection"],
            k_neighbors=self._cfg.get("k_neighbors"),
            repair=self._cfg.get("repair", "auto"),
            initializer=self._cfg.get("initializer"),
            mutation_prob_factor=self._cfg.get("mutation_prob_factor"),
            constraint_mode=self._cfg.get("constraint_mode", "feasibility"),
            track_genealogy=bool(self._cfg.get("track_genealogy", False)),
            result_mode=self._cfg.get("result_mode", "non_dominated"),
            external_archive=self._cfg.get("external_archive"),
        )

external_archive(capacity=None, **kwargs)

Configure an external archive for result storage.

Note

This is separate from archive_size, which controls the internal SPEA2 archive.

Parameters:

Name	Type	Description	Default
capacity	int | None	Maximum number of solutions. None means unbounded.	None
**kwargs	Any	Forwarded to :class:ExternalArchiveConfig.	{}

Source code in src/vamos/engine/algorithm/config/spea2.py

def external_archive(self, capacity: int | None = None, **kwargs: Any) -> _SPEA2ConfigBuilder:
    """Configure an external archive for result storage.

    Note
    ----
    This is separate from ``archive_size``, which controls the internal SPEA2 archive.

    Parameters
    ----------
    capacity
        Maximum number of solutions. ``None`` means unbounded.
    **kwargs
        Forwarded to :class:`ExternalArchiveConfig`.
    """
    self._cfg["external_archive"] = _build_external_archive_config(capacity, kwargs)
    self._cfg.setdefault("result_mode", "non_dominated")
    return self

vamos.engine.algorithm.config.ibea

IBEA configuration.

IBEAConfig dataclass

Bases: _SerializableConfig

Source code in src/vamos/engine/algorithm/config/ibea.py

@dataclass(frozen=True)
class IBEAConfig(_SerializableConfig):
    pop_size: int
    crossover: tuple[str, dict[str, Any]]
    mutation: tuple[str, dict[str, Any]]
    selection: tuple[str, dict[str, Any]]
    indicator: IndicatorType
    kappa: float
    repair: RepairConfigValue = "auto"
    initializer: dict[str, Any] | None = None
    mutation_prob_factor: float | None = None
    constraint_mode: ConstraintModeStr = "feasibility"
    track_genealogy: bool = False
    result_mode: ResultMode | None = None
    external_archive: ExternalArchiveConfig | None = None

    @classmethod
    def default(
        cls,
        pop_size: int = 100,
        n_var: int | None = None,
    ) -> IBEAConfig:
        """Create a default IBEA configuration."""
        mut_prob = 1.0 / n_var if n_var else 0.1
        return (
            cls.builder()
            .pop_size(pop_size)
            .crossover("sbx", prob=1.0, eta=20.0)
            .mutation("pm", prob=mut_prob, eta=20.0)
            .selection("tournament")
            .indicator("eps")
            .kappa(1.0)
            .build()
        )

    @classmethod
    def builder(cls) -> _IBEAConfigBuilder:
        return _IBEAConfigBuilder()

default(pop_size=100, n_var=None) classmethod

Create a default IBEA configuration.

Source code in src/vamos/engine/algorithm/config/ibea.py

@classmethod
def default(
    cls,
    pop_size: int = 100,
    n_var: int | None = None,
) -> IBEAConfig:
    """Create a default IBEA configuration."""
    mut_prob = 1.0 / n_var if n_var else 0.1
    return (
        cls.builder()
        .pop_size(pop_size)
        .crossover("sbx", prob=1.0, eta=20.0)
        .mutation("pm", prob=mut_prob, eta=20.0)
        .selection("tournament")
        .indicator("eps")
        .kappa(1.0)
        .build()
    )

_IBEAConfigBuilder

Bases: _ConfigBuilderState, _PopSizeBuilder, _CrossoverBuilder, _MutationBuilder, _SelectionBuilder, _RepairBuilder, _InitializerBuilder, _MutationProbFactorBuilder, _ConstraintModeBuilder, _TrackGenealogyBuilder, _ResultArchiveBuilder

Declarative configuration holder for IBEA settings.

Source code in src/vamos/engine/algorithm/config/ibea.py

class _IBEAConfigBuilder(
    _ConfigBuilderState,
    _PopSizeBuilder,
    _CrossoverBuilder,
    _MutationBuilder,
    _SelectionBuilder,
    _RepairBuilder,
    _InitializerBuilder,
    _MutationProbFactorBuilder,
    _ConstraintModeBuilder,
    _TrackGenealogyBuilder,
    _ResultArchiveBuilder,
):
    """Declarative configuration holder for IBEA settings."""

    @overload
    def indicator(self, name: IndicatorType) -> _IBEAConfigBuilder: ...

    @overload
    def indicator(self, name: str) -> _IBEAConfigBuilder: ...

    def indicator(self, name: str) -> _IBEAConfigBuilder:
        self._cfg["indicator"] = name
        return self

    def kappa(self, value: float) -> _IBEAConfigBuilder:
        self._cfg["kappa"] = value
        return self

    def build(self) -> IBEAConfig:
        _require_fields(
            self._cfg,
            ("pop_size", "crossover", "mutation", "selection", "indicator", "kappa"),
            "IBEA",
        )
        _validate_operators(self._cfg)
        return IBEAConfig(
            pop_size=self._cfg["pop_size"],
            crossover=self._cfg["crossover"],
            mutation=self._cfg["mutation"],
            selection=self._cfg["selection"],
            indicator=cast(IndicatorType, str(self._cfg["indicator"])),
            kappa=float(self._cfg["kappa"]),
            repair=self._cfg.get("repair", "auto"),
            initializer=self._cfg.get("initializer"),
            mutation_prob_factor=self._cfg.get("mutation_prob_factor"),
            constraint_mode=self._cfg.get("constraint_mode", "feasibility"),
            track_genealogy=bool(self._cfg.get("track_genealogy", False)),
            result_mode=self._cfg.get("result_mode", "non_dominated"),
            external_archive=self._cfg.get("external_archive"),
        )

vamos.engine.algorithm.config.smpso

SMPSO configuration.

SMPSOConfig dataclass

Bases: _SerializableConfig

Source code in src/vamos/engine/algorithm/config/smpso.py

@dataclass(frozen=True)
class SMPSOConfig(_SerializableConfig):
    pop_size: int
    archive_size: int  # Internal archive (part of SMPSO algorithm)
    mutation: tuple[str, dict[str, Any]]
    inertia: float = 0.1
    c1: float = 1.5
    c2: float = 1.5
    vmax_fraction: float = 0.5
    repair: RepairConfigValue = "auto"
    initializer: dict[str, Any] | None = None
    constraint_mode: ConstraintModeStr = "feasibility"
    track_genealogy: bool = False
    result_mode: ResultMode | None = None
    external_archive: ExternalArchiveConfig | None = None

    @classmethod
    def default(
        cls,
        pop_size: int = 100,
        n_var: int | None = None,
    ) -> SMPSOConfig:
        """Create a default SMPSO configuration."""
        mut_prob = 1.0 / n_var if n_var else 0.1
        return cls.builder().pop_size(pop_size).archive_size(pop_size).mutation("polynomial", prob=mut_prob, eta=20.0).build()

    @classmethod
    def builder(cls) -> _SMPSOConfigBuilder:
        return _SMPSOConfigBuilder()

default(pop_size=100, n_var=None) classmethod

Create a default SMPSO configuration.

Source code in src/vamos/engine/algorithm/config/smpso.py

@classmethod
def default(
    cls,
    pop_size: int = 100,
    n_var: int | None = None,
) -> SMPSOConfig:
    """Create a default SMPSO configuration."""
    mut_prob = 1.0 / n_var if n_var else 0.1
    return cls.builder().pop_size(pop_size).archive_size(pop_size).mutation("polynomial", prob=mut_prob, eta=20.0).build()

_SMPSOConfigBuilder

Bases: _ConfigBuilderState, _PopSizeBuilder, _MutationBuilder, _RepairBuilder, _InitializerBuilder, _ConstraintModeBuilder, _TrackGenealogyBuilder, _ResultArchiveBuilder

Declarative configuration holder for SMPSO settings.

Source code in src/vamos/engine/algorithm/config/smpso.py

class _SMPSOConfigBuilder(
    _ConfigBuilderState,
    _PopSizeBuilder,
    _MutationBuilder,
    _RepairBuilder,
    _InitializerBuilder,
    _ConstraintModeBuilder,
    _TrackGenealogyBuilder,
    _ResultArchiveBuilder,
):
    """Declarative configuration holder for SMPSO settings."""

    def archive_size(self, value: int) -> _SMPSOConfigBuilder:
        self._cfg["archive_size"] = value
        return self

    def inertia(self, value: float) -> _SMPSOConfigBuilder:
        self._cfg["inertia"] = value
        return self

    def c1(self, value: float) -> _SMPSOConfigBuilder:
        self._cfg["c1"] = value
        return self

    def c2(self, value: float) -> _SMPSOConfigBuilder:
        self._cfg["c2"] = value
        return self

    def vmax_fraction(self, value: float) -> _SMPSOConfigBuilder:
        self._cfg["vmax_fraction"] = value
        return self

    def external_archive(self, capacity: int | None = None, **kwargs: Any) -> _SMPSOConfigBuilder:
        """Configure an external archive for result storage.

        Note
        ----
        This is separate from ``archive_size``, which controls the internal SMPSO archive.

        Parameters
        ----------
        capacity
            Maximum number of solutions. ``None`` means unbounded.
        **kwargs
            Forwarded to :class:`ExternalArchiveConfig`.
        """
        self._cfg["external_archive"] = _build_external_archive_config(capacity, kwargs)
        self._cfg.setdefault("result_mode", "non_dominated")
        return self

    def build(self) -> SMPSOConfig:
        _require_fields(
            self._cfg,
            ("pop_size", "archive_size", "mutation"),
            "SMPSO",
        )
        _validate_operators(self._cfg)
        return SMPSOConfig(
            pop_size=self._cfg["pop_size"],
            archive_size=self._cfg["archive_size"],
            mutation=self._cfg["mutation"],
            inertia=float(self._cfg.get("inertia", 0.1)),
            c1=float(self._cfg.get("c1", 1.5)),
            c2=float(self._cfg.get("c2", 1.5)),
            vmax_fraction=float(self._cfg.get("vmax_fraction", 0.5)),
            repair=self._cfg.get("repair", "auto"),
            initializer=self._cfg.get("initializer"),
            constraint_mode=self._cfg.get("constraint_mode", "feasibility"),
            track_genealogy=bool(self._cfg.get("track_genealogy", False)),
            result_mode=self._cfg.get("result_mode", "non_dominated"),
            external_archive=self._cfg.get("external_archive"),
        )

external_archive(capacity=None, **kwargs)

Configure an external archive for result storage.

Note

This is separate from archive_size, which controls the internal SMPSO archive.

Parameters:

Name	Type	Description	Default
capacity	int | None	Maximum number of solutions. None means unbounded.	None
**kwargs	Any	Forwarded to :class:ExternalArchiveConfig.	{}

Source code in src/vamos/engine/algorithm/config/smpso.py

def external_archive(self, capacity: int | None = None, **kwargs: Any) -> _SMPSOConfigBuilder:
    """Configure an external archive for result storage.

    Note
    ----
    This is separate from ``archive_size``, which controls the internal SMPSO archive.

    Parameters
    ----------
    capacity
        Maximum number of solutions. ``None`` means unbounded.
    **kwargs
        Forwarded to :class:`ExternalArchiveConfig`.
    """
    self._cfg["external_archive"] = _build_external_archive_config(capacity, kwargs)
    self._cfg.setdefault("result_mode", "non_dominated")
    return self

vamos.engine.algorithm.config.agemoea

AGE-MOEA configuration.

AGEMOEAConfig dataclass

Bases: _SerializableConfig

Source code in src/vamos/engine/algorithm/config/agemoea.py

@dataclass(frozen=True)
class AGEMOEAConfig(_SerializableConfig):
    pop_size: int
    crossover: tuple[str, dict[str, Any]]
    mutation: tuple[str, dict[str, Any]]
    repair: RepairConfigValue = "auto"
    initializer: dict[str, Any] | None = None
    mutation_prob_factor: float | None = None
    constraint_mode: ConstraintModeStr = "feasibility"
    track_genealogy: bool = False
    result_mode: ResultMode | None = None
    external_archive: ExternalArchiveConfig | None = None

    @classmethod
    def default(
        cls,
        pop_size: int = 100,
        n_var: int | None = None,
    ) -> AGEMOEAConfig:
        """Create a default AGE-MOEA configuration."""
        mut_prob = 1.0 / n_var if n_var else 0.1
        return cls.builder().pop_size(pop_size).crossover("sbx", prob=0.9, eta=15.0).mutation("pm", prob=mut_prob, eta=20.0).build()

    @classmethod
    def builder(cls) -> _AGEMOEAConfigBuilder:
        return _AGEMOEAConfigBuilder()

default(pop_size=100, n_var=None) classmethod

Create a default AGE-MOEA configuration.

Source code in src/vamos/engine/algorithm/config/agemoea.py

@classmethod
def default(
    cls,
    pop_size: int = 100,
    n_var: int | None = None,
) -> AGEMOEAConfig:
    """Create a default AGE-MOEA configuration."""
    mut_prob = 1.0 / n_var if n_var else 0.1
    return cls.builder().pop_size(pop_size).crossover("sbx", prob=0.9, eta=15.0).mutation("pm", prob=mut_prob, eta=20.0).build()

_AGEMOEAConfigBuilder

Bases: _ConfigBuilderState, _PopSizeBuilder, _CrossoverBuilder, _MutationBuilder, _RepairBuilder, _InitializerBuilder, _MutationProbFactorBuilder, _ConstraintModeBuilder, _TrackGenealogyBuilder, _ResultArchiveBuilder

Fluent builder for AGE-MOEA configs.

Source code in src/vamos/engine/algorithm/config/agemoea.py

class _AGEMOEAConfigBuilder(
    _ConfigBuilderState,
    _PopSizeBuilder,
    _CrossoverBuilder,
    _MutationBuilder,
    _RepairBuilder,
    _InitializerBuilder,
    _MutationProbFactorBuilder,
    _ConstraintModeBuilder,
    _TrackGenealogyBuilder,
    _ResultArchiveBuilder,
):
    """
    Fluent builder for AGE-MOEA configs.
    """

    def build(self) -> AGEMOEAConfig:
        _require_fields(
            self._cfg,
            ("pop_size", "crossover", "mutation"),
            "AGE-MOEA",
        )
        _validate_operators(self._cfg)
        return AGEMOEAConfig(
            pop_size=self._cfg["pop_size"],
            crossover=self._cfg["crossover"],
            mutation=self._cfg["mutation"],
            repair=self._cfg.get("repair", "auto"),
            initializer=self._cfg.get("initializer"),
            mutation_prob_factor=self._cfg.get("mutation_prob_factor"),
            constraint_mode=self._cfg.get("constraint_mode", "feasibility"),
            track_genealogy=bool(self._cfg.get("track_genealogy", False)),
            result_mode=self._cfg.get("result_mode", "non_dominated"),
            external_archive=self._cfg.get("external_archive"),
        )

vamos.engine.algorithm.config.rvea

RVEA configuration.

RVEAConfig dataclass

Bases: _SerializableConfig

Source code in src/vamos/engine/algorithm/config/rvea.py

@dataclass(frozen=True)
class RVEAConfig(_SerializableConfig):
    pop_size: int
    n_partitions: int
    alpha: float
    adapt_freq: float | None
    crossover: tuple[str, dict[str, Any]]
    mutation: tuple[str, dict[str, Any]]
    repair: RepairConfigValue = "auto"
    initializer: dict[str, Any] | None = None
    mutation_prob_factor: float | None = None
    constraint_mode: ConstraintModeStr = "feasibility"
    track_genealogy: bool = False
    result_mode: ResultMode | None = None
    external_archive: ExternalArchiveConfig | None = None

    @classmethod
    def default(
        cls,
        pop_size: int | None = None,
        n_var: int | None = None,
        n_obj: int = 3,
    ) -> RVEAConfig:
        """Create a default RVEA configuration."""
        mut_prob = 1.0 / n_var if n_var else 0.1
        n_partitions = 12 if n_obj == 3 else 6
        if pop_size is None:
            pop_size = comb(n_partitions + n_obj - 1, n_obj - 1)
        return (
            cls.builder()
            .pop_size(pop_size)
            .n_partitions(n_partitions)
            .alpha(2.0)
            .adapt_freq(0.1)
            .crossover("sbx", prob=1.0, eta=30.0)
            .mutation("pm", prob=mut_prob, eta=20.0)
            .build()
        )

    @classmethod
    def builder(cls) -> _RVEAConfigBuilder:
        return _RVEAConfigBuilder()

default(pop_size=None, n_var=None, n_obj=3) classmethod

Create a default RVEA configuration.

Source code in src/vamos/engine/algorithm/config/rvea.py

@classmethod
def default(
    cls,
    pop_size: int | None = None,
    n_var: int | None = None,
    n_obj: int = 3,
) -> RVEAConfig:
    """Create a default RVEA configuration."""
    mut_prob = 1.0 / n_var if n_var else 0.1
    n_partitions = 12 if n_obj == 3 else 6
    if pop_size is None:
        pop_size = comb(n_partitions + n_obj - 1, n_obj - 1)
    return (
        cls.builder()
        .pop_size(pop_size)
        .n_partitions(n_partitions)
        .alpha(2.0)
        .adapt_freq(0.1)
        .crossover("sbx", prob=1.0, eta=30.0)
        .mutation("pm", prob=mut_prob, eta=20.0)
        .build()
    )

_RVEAConfigBuilder

Bases: _ConfigBuilderState, _PopSizeBuilder, _CrossoverBuilder, _MutationBuilder, _RepairBuilder, _InitializerBuilder, _MutationProbFactorBuilder, _ConstraintModeBuilder, _TrackGenealogyBuilder, _ResultArchiveBuilder

Declarative configuration holder for RVEA settings.

Examples: cfg = RVEAConfig.default() cfg = RVEAConfig.builder().pop_size(100).n_partitions(12).build()

Source code in src/vamos/engine/algorithm/config/rvea.py

class _RVEAConfigBuilder(
    _ConfigBuilderState,
    _PopSizeBuilder,
    _CrossoverBuilder,
    _MutationBuilder,
    _RepairBuilder,
    _InitializerBuilder,
    _MutationProbFactorBuilder,
    _ConstraintModeBuilder,
    _TrackGenealogyBuilder,
    _ResultArchiveBuilder,
):
    """
    Declarative configuration holder for RVEA settings.

    Examples:
        cfg = RVEAConfig.default()
        cfg = RVEAConfig.builder().pop_size(100).n_partitions(12).build()
    """

    def n_partitions(self, value: int) -> _RVEAConfigBuilder:
        self._cfg["n_partitions"] = value
        return self

    def alpha(self, value: float) -> _RVEAConfigBuilder:
        self._cfg["alpha"] = float(value)
        return self

    def adapt_freq(self, value: float | None) -> _RVEAConfigBuilder:
        self._cfg["adapt_freq"] = None if value is None else float(value)
        return self

    def build(self) -> RVEAConfig:
        _require_fields(
            self._cfg,
            ("pop_size", "n_partitions", "alpha", "crossover", "mutation"),
            "RVEA",
        )
        _validate_operators(self._cfg)
        return RVEAConfig(
            pop_size=self._cfg["pop_size"],
            n_partitions=self._cfg.get("n_partitions", 12),
            alpha=float(self._cfg.get("alpha", 2.0)),
            adapt_freq=self._cfg.get("adapt_freq", 0.1),
            crossover=self._cfg["crossover"],
            mutation=self._cfg["mutation"],
            repair=self._cfg.get("repair", "auto"),
            initializer=self._cfg.get("initializer"),
            mutation_prob_factor=self._cfg.get("mutation_prob_factor"),
            constraint_mode=self._cfg.get("constraint_mode", "feasibility"),
            track_genealogy=bool(self._cfg.get("track_genealogy", False)),
            result_mode=self._cfg.get("result_mode", "non_dominated"),
            external_archive=self._cfg.get("external_archive"),
        )

vamos.engine.algorithm.config.generic

GenericAlgorithmConfig dataclass

Minimal config wrapper for plugin/custom algorithms.

This exists to keep the public-facing APIs strongly typed while still allowing users (or tests) to pass a free-form mapping when integrating a third-party algorithm via the algorithm registry.

Source code in src/vamos/engine/algorithm/config/generic.py

@dataclass(frozen=True, slots=True)
class GenericAlgorithmConfig:
    """
    Minimal config wrapper for plugin/custom algorithms.

    This exists to keep the public-facing APIs strongly typed while still
    allowing users (or tests) to pass a free-form mapping when integrating a
    third-party algorithm via the algorithm registry.
    """

    data: Mapping[str, object]

    def to_dict(self) -> AlgorithmConfigMapping:
        return dict(self.data)

Constraint Handling

vamos.foundation.constraints

ConstraintInfo dataclass

Source code in src/vamos/foundation/constraints/__init__.py

@dataclass
class ConstraintInfo:
    G: np.ndarray
    cv: np.ndarray
    feasible_mask: np.ndarray

FeasibilityFirstStrategy

Bases: ConstraintHandlingStrategy

Source code in src/vamos/foundation/constraints/__init__.py

class FeasibilityFirstStrategy(ConstraintHandlingStrategy):
    def __init__(self, objective_aggregator: str = "sum"):
        self.objective_aggregator = objective_aggregator

    def rank(self, F: np.ndarray, G: np.ndarray | None) -> np.ndarray:
        F = np.asarray(F, dtype=float)
        if G is None:
            return _aggregate_objectives(F, self.objective_aggregator)
        info = compute_constraint_info(G)
        agg = _aggregate_objectives(F, self.objective_aggregator)
        infeasible_penalty = info.cv
        # Feasible get 0 prefix, infeasible 1 + cv to stay worse
        return np.where(info.feasible_mask, agg, agg.max(initial=0.0) + 1.0 + infeasible_penalty)

PenaltyCVStrategy

Bases: ConstraintHandlingStrategy

Source code in src/vamos/foundation/constraints/__init__.py

class PenaltyCVStrategy(ConstraintHandlingStrategy):
    def __init__(self, penalty_lambda: float = 1000.0, objective_aggregator: str = "sum"):
        if penalty_lambda <= 0:
            raise ValueError("penalty_lambda must be positive.")
        self.penalty_lambda = float(penalty_lambda)
        self.objective_aggregator = objective_aggregator

    def rank(self, F: np.ndarray, G: np.ndarray | None) -> np.ndarray:
        F = np.asarray(F, dtype=float)
        agg = _aggregate_objectives(F, self.objective_aggregator)
        if G is None:
            return agg
        info = compute_constraint_info(G)
        return agg + self.penalty_lambda * info.cv

CVAsObjectiveStrategy

Bases: ConstraintHandlingStrategy

Source code in src/vamos/foundation/constraints/__init__.py

class CVAsObjectiveStrategy(ConstraintHandlingStrategy):
    def __init__(self, objective_aggregator: str = "sum", eps: float = 1e-6):
        self.objective_aggregator = objective_aggregator
        self.eps = float(eps)

    def rank(self, F: np.ndarray, G: np.ndarray | None) -> np.ndarray:
        F = np.asarray(F, dtype=float)
        agg = _aggregate_objectives(F, self.objective_aggregator)
        if G is None:
            return agg
        info = compute_constraint_info(G)
        return info.cv + self.eps * agg

EpsilonConstraintStrategy

Bases: ConstraintHandlingStrategy

Source code in src/vamos/foundation/constraints/__init__.py

class EpsilonConstraintStrategy(ConstraintHandlingStrategy):
    def __init__(self, epsilon: float = 0.0, objective_aggregator: str = "sum"):
        self.epsilon = float(epsilon)
        self.objective_aggregator = objective_aggregator

    def rank(self, F: np.ndarray, G: np.ndarray | None) -> np.ndarray:
        F = np.asarray(F, dtype=float)
        if G is None:
            return _aggregate_objectives(F, self.objective_aggregator)
        info = compute_constraint_info(G, eps=self.epsilon)
        agg = _aggregate_objectives(F, self.objective_aggregator)
        infeasible_penalty = info.cv
        return np.where(info.feasible_mask, agg, agg.max(initial=0.0) + 1.0 + infeasible_penalty)

compute_constraint_info(G, eps=0.0)

Compute aggregate constraint violation and feasibility mask.

Parameters:

Name	Type	Description	Default
G	ndarray | None	Constraint values where <= 0 means satisfied.	required
eps	float	Feasibility tolerance. Constraints <= eps are treated as satisfied.	``0.0``

Source code in src/vamos/foundation/constraints/__init__.py

def compute_constraint_info(G: np.ndarray | None, eps: float = 0.0) -> ConstraintInfo:
    """Compute aggregate constraint violation and feasibility mask.

    Parameters
    ----------
    G : np.ndarray | None
        Constraint values where ``<= 0`` means satisfied.
    eps : float, default ``0.0``
        Feasibility tolerance. Constraints ``<= eps`` are treated as satisfied.
    """
    if G is None:
        empty_G = np.zeros((0, 0), dtype=float)
        return ConstraintInfo(
            G=empty_G,
            cv=np.zeros(0, dtype=float),
            feasible_mask=np.ones(0, dtype=bool),
        )
    G = np.asarray(G, dtype=float)
    if G.ndim != 2:
        raise ValueError("G must be a 2D array of shape (n_points, n_constraints).")
    positive = np.maximum(G - eps, 0.0)
    cv = np.asarray(np.sum(positive, axis=1), dtype=float)
    feasible = np.asarray(np.all(G <= eps, axis=1), dtype=bool)
    return ConstraintInfo(G=G, cv=cv, feasible_mask=feasible)

get_constraint_strategy(name, **kwargs)

Source code in src/vamos/foundation/constraints/__init__.py

def get_constraint_strategy(name: str, **kwargs: Any) -> ConstraintHandlingStrategy:
    key = name.lower()
    if key == "feasibility_first":
        return FeasibilityFirstStrategy(**kwargs)
    if key == "penalty_cv":
        return PenaltyCVStrategy(**kwargs)
    if key == "cv_as_objective":
        return CVAsObjectiveStrategy(**kwargs)
    if key == "epsilon":
        return EpsilonConstraintStrategy(**kwargs)
    raise ValueError(f"Unknown constraint strategy '{name}'.")

vamos.foundation.constraints.utils

Utility helpers for constraint handling.

compute_violation(G, *, n=None)

Sum of positive parts per-solution; assumes G shape (N, n_constraints), g<=0 satisfied.

When G is None (unconstrained), n must be provided so the output length is explicit.

Source code in src/vamos/foundation/constraints/utils.py

def compute_violation(G: np.ndarray | None, *, n: int | None = None) -> np.ndarray:
    """Sum of positive parts per-solution; assumes G shape (N, n_constraints), g<=0 satisfied.

    When *G* is ``None`` (unconstrained), ``n`` must be provided so the output
    length is explicit.
    """
    if G is None:
        if n is None:
            raise ValueError("compute_violation() requires 'n' when G is None.")
        return np.zeros(n, dtype=float)
    positive = np.maximum(G, 0.0)
    return np.asarray(np.sum(positive, axis=1), dtype=float)

is_feasible(G, *, n=None, eps=0.0)

Boolean feasibility mask; assumes G shape (N, n_constraints).

When G is None (unconstrained), n must be provided so the output length is explicit.

eps is a feasibility tolerance: constraints with g(x) <= eps are treated as satisfied (default 0.0).

Source code in src/vamos/foundation/constraints/utils.py

def is_feasible(G: np.ndarray | None, *, n: int | None = None, eps: float = 0.0) -> np.ndarray:
    """Boolean feasibility mask; assumes G shape (N, n_constraints).

    When *G* is ``None`` (unconstrained), ``n`` must be provided so the output
    length is explicit.

    *eps* is a feasibility tolerance: constraints with ``g(x) <= eps`` are
    treated as satisfied (default ``0.0``).
    """
    if G is None:
        if n is None:
            raise ValueError("is_feasible() requires 'n' when G is None.")
        return np.ones(n, dtype=bool)
    return np.asarray(np.all(G <= eps, axis=1), dtype=bool)

Encoding

vamos.foundation.encoding

normalize_encoding(value, *, default='real')

Normalize user/problem encoding strings to canonical encoding identifiers.

Canonical encodings are: "real", "binary", "permutation", "integer", "mixed".

Source code in src/vamos/foundation/encoding.py

def normalize_encoding(value: str | None, *, default: Encoding = "real") -> Encoding:
    """
    Normalize user/problem encoding strings to canonical encoding identifiers.

    Canonical encodings are: "real", "binary", "permutation", "integer", "mixed".
    """
    if value is None:
        return default
    key = value.strip().lower()
    if not key:
        return default
    normalized = _ALIASES.get(key)
    if normalized is None:
        all_names = sorted(set(_ALIASES))
        matches = difflib.get_close_matches(key, all_names, n=3, cutoff=0.5)
        if matches:
            raise ValueError(f"Unknown encoding '{value}'. Did you mean: {', '.join(matches)}?")
        expected = ", ".join(all_names)
        raise ValueError(f"Unknown encoding '{value}'. Expected one of: {expected}.")
    return normalized

Problem Registry

vamos.foundation.problem.registry

Problem registry: specs, selection, and factories.

available_problem_names()

Return the sorted names of all registered benchmark problems.

Returns:

Type	Description
tuple[str, ...]	Problem keys that can be passed to optimize() or make_problem_selection().

Source code in src/vamos/foundation/problem/registry/specs.py

def available_problem_names() -> tuple[str, ...]:
    """Return the sorted names of all registered benchmark problems.

    Returns
    -------
    tuple[str, ...]
        Problem keys that can be passed to ``optimize()`` or
        ``make_problem_selection()``.
    """
    return tuple(get_problem_specs().keys())

make_problem_selection(key, *, n_var=None, n_obj=None)

Look up a registered problem by key and resolve its dimensions.

Parameters:

Name	Type	Description	Default
key	str	Registered problem name (e.g. "zdt1", "dtlz2").	required
n_var	int	Override the default number of decision variables.	None
n_obj	int	Override the default number of objectives (only for problems that allow it).	None

Returns:

Type	Description
ProblemSelection	A frozen selection with the resolved spec, n_var, and n_obj ready to instantiate.

Raises:

Type	Description
KeyError	If key does not match any registered problem. The error message includes the full list of valid names and close-match suggestions.

Source code in src/vamos/foundation/problem/registry/selection.py

def make_problem_selection(key: str, *, n_var: int | None = None, n_obj: int | None = None) -> ProblemSelection:
    """Look up a registered problem by *key* and resolve its dimensions.

    Parameters
    ----------
    key : str
        Registered problem name (e.g. ``"zdt1"``, ``"dtlz2"``).
    n_var : int, optional
        Override the default number of decision variables.
    n_obj : int, optional
        Override the default number of objectives (only for problems
        that allow it).

    Returns
    -------
    ProblemSelection
        A frozen selection with the resolved ``spec``, ``n_var``, and
        ``n_obj`` ready to instantiate.

    Raises
    ------
    KeyError
        If *key* does not match any registered problem.  The error
        message includes the full list of valid names and close-match
        suggestions.
    """
    specs = get_problem_specs()
    try:
        spec = specs[key]
    except KeyError as exc:
        available = sorted(specs.keys())
        raise KeyError(_format_unknown_problem(key, available)) from exc

    actual_n_var, actual_n_obj = spec.resolve_dimensions(n_var=n_var, n_obj=n_obj)
    return ProblemSelection(spec=spec, n_var=actual_n_var, n_obj=actual_n_obj)

Tuning (experimental)

Parameters and Tuners for Hyperparameter Optimization.

vamos.engine.tuning.racing.core

RacingTuner

Racing tuner for algorithm configuration.

Source code in src/vamos/engine/tuning/racing/core.py

class RacingTuner:
    """Racing tuner for algorithm configuration."""

    def __init__(
        self,
        task: TuningTask,
        scenario: Scenario,
        seed: int = 0,
        max_initial_configs: int = 20,
        sampler: Sampler | None = None,
        initial_configs: list[dict[str, Any]] | None = None,
    ) -> None:
        self.task = task
        self.scenario = scenario
        self.rng = np.random.default_rng(seed)
        self.max_initial_configs = max_initial_configs
        self._stage_index: int = 0
        self._elite_archive: list[EliteEntry] = []
        self._next_config_id: int = 0  # Re-indexed below
        self._best_score_history: list[float] = []  # For convergence detection

        # Injection of default/user configurations
        self.initial_configs_payload = initial_configs or []

        self.param_space: ParamSpace = task.param_space
        self.instances: Sequence[Instance] = list(task.instances)
        self.seeds: Sequence[int] = list(task.seeds)

        if sampler is None:
            # Prefer model-based sampling by default in large/conditional spaces.
            self.sampler: Sampler = ModelBasedSampler(self.param_space)
        else:
            self.sampler = sampler

        self._schedule: list[tuple[int, int]] = build_schedule(
            self.instances,
            self.seeds,
            start_instances=self.scenario.start_instances,
            instance_order_random=self.scenario.instance_order_random,
            seed_order_random=self.scenario.seed_order_random,
            rng=self.rng,
        )

    def _sample_initial_configs(self) -> list[ConfigState]:
        configs: list[ConfigState] = []
        config_id_counter = 0

        # 1. Add injected configs
        for user_cfg in self.initial_configs_payload:
            state = ConfigState(config_id=config_id_counter, config=user_cfg, alive=True)
            configs.append(state)
            config_id_counter += 1

        # 2. Sample remainder
        needed = max(0, self.max_initial_configs - len(configs))
        for _ in range(needed):
            cfg = self.sampler.sample(self.rng)
            state = ConfigState(config_id=config_id_counter, config=cfg, alive=True)
            configs.append(state)
            config_id_counter += 1

        self._next_config_id = config_id_counter
        return configs

    def _current_budget(self) -> int:
        base_budget = self.task.budget_per_run
        if self.scenario.use_adaptive_budget:
            if self.scenario.initial_budget_per_run is not None:
                base_budget = self.scenario.initial_budget_per_run

            budget = int(round(base_budget * (self.scenario.budget_growth_factor**self._stage_index)))

            if self.scenario.max_budget_per_run is not None:
                budget = min(budget, self.scenario.max_budget_per_run)

            if self.task.budget_per_run is not None:
                budget = min(budget, self.task.budget_per_run)

            if budget <= 0:
                budget = max(1, self.task.budget_per_run or 1)
            return budget

        if base_budget is None or base_budget <= 0:
            raise ValueError("task.budget_per_run must be positive when adaptive budget is disabled")
        return base_budget

    def run(
        self,
        eval_fn: EvalFn,
        verbose: bool | None = None,
    ) -> tuple[dict[str, Any], list[TrialResult]]:
        # Dispatch to multi-fidelity if enabled
        if self.scenario.use_multi_fidelity:
            return _run_multi_fidelity(self, eval_fn, verbose)

        verbose_flag = self.scenario.verbose if verbose is None else verbose

        schedule = list(self._schedule)
        configs: list[ConfigState] = self._sample_initial_configs()
        num_experiments = 0

        for inst_idx, seed_idx in schedule:
            if self.scenario.max_stages is not None and self._stage_index >= self.scenario.max_stages:
                if verbose_flag:
                    _logger().info("[racing] Reached maximum number of stages.")
                break

            if self._num_alive(configs) == 0:
                break

            stage_alive = self._num_alive(configs)
            remaining_budget = self.scenario.max_experiments - num_experiments
            if remaining_budget <= 0:
                if verbose_flag:
                    _logger().info("[racing] Experiment budget exhausted.")
                break
            partial_eval_limit: int | None = None
            if remaining_budget < stage_alive:
                partial_eval_limit = int(remaining_budget)
                if verbose_flag:
                    _logger().info(
                        "[racing] Partial stage due to budget: evaluating %s/%s alive configs.",
                        partial_eval_limit,
                        stage_alive,
                    )

            if verbose_flag:
                _logger().info(
                    "[racing] Stage %s: instance %s, seed idx %s, alive=%s",
                    self._stage_index,
                    inst_idx,
                    seed_idx,
                    stage_alive,
                )

            stage_eval_count = self._run_stage(
                configs,
                inst_idx,
                seed_idx,
                eval_fn,
                max_evals=partial_eval_limit,
            )
            num_experiments += stage_eval_count

            eliminated_any = False
            if partial_eval_limit is None:
                eliminated_any = eliminate_configs(configs, task=self.task, scenario=self.scenario)

            if eliminated_any:
                self._elite_archive = update_elite_archive(
                    configs,
                    task=self.task,
                    scenario=self.scenario,
                    elite_archive=self._elite_archive,
                )
                if self.scenario.use_elitist_restarts:
                    self._next_config_id = refill_population(
                        configs,
                        scenario=self.scenario,
                        param_space=self.param_space,
                        sampler=self.sampler,
                        elite_archive=self._elite_archive,
                        target_population_size=self.scenario.target_population_size or self.max_initial_configs,
                        rng=self.rng,
                        next_config_id=self._next_config_id,
                    )
            if isinstance(self.sampler, ModelBasedSampler):
                survivor_configs = [c.config for c in configs if c.alive]
                self.sampler.update(survivor_configs)

            # Track best score for convergence detection
            current_best = self._get_current_best_score(configs)
            if current_best is not None:
                self._best_score_history.append(current_best)

            reached_budget = num_experiments >= self.scenario.max_experiments
            reached_min_survivors = self._num_alive(configs) <= self.scenario.min_survivors
            reached_convergence = self._check_convergence()

            self._stage_index += 1

            if reached_budget:
                if verbose_flag:
                    _logger().info("[racing] Reached maximum experiment budget.")
                break

            if reached_min_survivors:
                if verbose_flag:
                    _logger().info("[racing] Reached minimum survivors, stopping early.")
                break

            if reached_convergence:
                if verbose_flag:
                    _logger().info("[racing] Converged after %s stages (no improvement).", self._stage_index)
                break

            if partial_eval_limit is not None:
                if verbose_flag:
                    _logger().info("[racing] Budget exhausted after partial stage.")
                break

        best_state, history = self._finalize_results(configs)
        if best_state is None:
            raise RuntimeError("RacingTuner finished without a valid configuration.")

        if verbose_flag and best_state.score is not None:
            _logger().info(
                "[racing] Best score=%.6f after stage %s.",
                best_state.score,
                self._stage_index,
            )

        return best_state.config, history

    def _run_stage(
        self,
        configs: list[ConfigState],
        inst_idx: int,
        seed_idx: int,
        eval_fn: EvalFn,
        max_evals: int | None = None,
    ) -> int:
        instance = self.instances[inst_idx]
        seed = self.seeds[seed_idx]
        budget = self._current_budget()

        # Identify jobs to run
        tasks = []
        indices = []
        for idx, state in enumerate(configs):
            if not state.alive:
                continue
            ctx = EvalContext(instance=instance, seed=seed, budget=budget)
            tasks.append((state.config, ctx))
            indices.append(idx)

        if not tasks:
            return 0

        if max_evals is not None and max_evals < len(tasks):
            take = int(max(max_evals, 0))
            if take <= 0:
                return 0
            chosen = self.rng.choice(len(tasks), size=take, replace=False)
            chosen = np.sort(chosen)
            tasks = [tasks[int(i)] for i in chosen]
            indices = [indices[int(i)] for i in chosen]

        if self.scenario.n_jobs == 1:
            # Sequential execution (avoid overhead)
            for i, (cfg, ctx) in enumerate(tasks):
                try:
                    result = eval_fn(cfg, ctx)
                    score = float(result[0]) if isinstance(result, tuple) else float(result)
                except Exception:
                    _logger().warning(
                        "[racing] eval_fn failed for config_id=%s; assigning failure score.",
                        configs[indices[i]].config_id,
                        exc_info=True,
                    )
                    score = _failure_score(self.task.maximize)
                configs[indices[i]].scores.append(score)
        else:
            # Parallel execution with joblib
            results = Parallel(n_jobs=self.scenario.n_jobs)(
                delayed(_eval_worker)(eval_fn, cfg, ctx, self.task.maximize) for cfg, ctx in tasks
            )

            for i, score in enumerate(results):
                configs[indices[i]].scores.append(score)
        return len(tasks)

    def _num_alive(self, configs: list[ConfigState]) -> int:
        return sum(1 for c in configs if c.alive)

    def _get_current_best_score(self, configs: list[ConfigState]) -> float | None:
        """Get the current best aggregated score among alive configs."""
        best: float | None = None
        for state in configs:
            if not state.alive or not state.scores:
                continue
            agg = float(self.task.aggregator(state.scores))
            if best is None:
                best = agg
            elif self.task.maximize and agg > best:
                best = agg
            elif not self.task.maximize and agg < best:
                best = agg
        return best

    def _check_convergence(self) -> bool:
        """Check if best score has stagnated for convergence_window stages."""
        window = self.scenario.convergence_window
        if window <= 0:
            return False  # Disabled

        history = self._best_score_history
        if len(history) < window:
            return False

        # Get scores over the last 'window' stages
        recent = history[-window:]
        oldest = recent[0]
        newest = recent[-1]
        recent_arr = np.asarray(recent, dtype=float)
        if recent_arr.size == 0 or not np.all(np.isfinite(recent_arr)):
            return False

        # Compute relative improvement
        if self.task.maximize:
            delta = newest - oldest
        else:
            delta = oldest - newest
        if abs(oldest) < 1e-12:
            # Avoid division by zero; when near zero use signed absolute-diff.
            improvement = delta
        else:
            improvement = delta / abs(oldest)

        # Also require low volatility in the recent window to avoid declaring
        # convergence on noisy oscillations around a plateau.
        mean_abs = float(np.mean(np.abs(recent_arr)))
        scale = mean_abs if mean_abs > 1e-12 else 1.0
        volatility = float(np.std(recent_arr) / scale)

        return improvement < self.scenario.convergence_threshold and volatility < self.scenario.convergence_threshold

    def _finalize_results(self, configs: list[ConfigState]) -> tuple[EliteEntry | None, list[TrialResult]]:
        history: list[TrialResult] = []
        best_state: EliteEntry | None = None
        best_score: float | None = None

        for state in configs:
            scores = state.scores
            details = {"num_evals": len(state.scores), "alive": state.alive}
            if self.scenario.use_multi_fidelity:
                final_level = len(self.scenario.fidelity_levels) - 1
                final_scores = state.fidelity_scores.get(final_level) or []
                details["final_fidelity_level"] = int(final_level)
                details["num_final_evals"] = int(len(final_scores))
                if final_scores:
                    scores = final_scores

            if not scores:
                agg_score = float("nan")
            else:
                agg_score = float(self.task.aggregator(scores))

            history.append(
                TrialResult(
                    trial_id=state.config_id,
                    config=state.config,
                    score=agg_score,
                    details=details,
                )
            )

            # Consider all evaluated configs, not only survivors.
            if not scores or np.isnan(agg_score):
                continue

            if best_score is None:
                best_state = EliteEntry(config=state.config, score=agg_score)
                best_score = agg_score
                continue

            if self.task.maximize and agg_score > best_score:
                best_state = EliteEntry(config=state.config, score=agg_score)
                best_score = agg_score
            elif not self.task.maximize and agg_score < best_score:
                best_state = EliteEntry(config=state.config, score=agg_score)
                best_score = agg_score

        return best_state, history

_check_convergence()

Check if best score has stagnated for convergence_window stages.

Source code in src/vamos/engine/tuning/racing/core.py

def _check_convergence(self) -> bool:
    """Check if best score has stagnated for convergence_window stages."""
    window = self.scenario.convergence_window
    if window <= 0:
        return False  # Disabled

    history = self._best_score_history
    if len(history) < window:
        return False

    # Get scores over the last 'window' stages
    recent = history[-window:]
    oldest = recent[0]
    newest = recent[-1]
    recent_arr = np.asarray(recent, dtype=float)
    if recent_arr.size == 0 or not np.all(np.isfinite(recent_arr)):
        return False

    # Compute relative improvement
    if self.task.maximize:
        delta = newest - oldest
    else:
        delta = oldest - newest
    if abs(oldest) < 1e-12:
        # Avoid division by zero; when near zero use signed absolute-diff.
        improvement = delta
    else:
        improvement = delta / abs(oldest)

    # Also require low volatility in the recent window to avoid declaring
    # convergence on noisy oscillations around a plateau.
    mean_abs = float(np.mean(np.abs(recent_arr)))
    scale = mean_abs if mean_abs > 1e-12 else 1.0
    volatility = float(np.std(recent_arr) / scale)

    return improvement < self.scenario.convergence_threshold and volatility < self.scenario.convergence_threshold

_get_current_best_score(configs)

Get the current best aggregated score among alive configs.

Source code in src/vamos/engine/tuning/racing/core.py

def _get_current_best_score(self, configs: list[ConfigState]) -> float | None:
    """Get the current best aggregated score among alive configs."""
    best: float | None = None
    for state in configs:
        if not state.alive or not state.scores:
            continue
        agg = float(self.task.aggregator(state.scores))
        if best is None:
            best = agg
        elif self.task.maximize and agg > best:
            best = agg
        elif not self.task.maximize and agg < best:
            best = agg
    return best

vamos.engine.tuning.racing.param_space

Hyperparameter space definitions for VAMOS tuning.

All parameter types use name as the first argument: - Real(name, low, high, log=False, role="operator_rate") - Int(name, low, high, log=False, role="population") - Categorical(name, choices, role="operator") - Boolean(name, role="adaptive")

All types support: - sample(rng) - draw a random value - to_unit(value) - map to [0, 1] space for optimization - from_unit(value) - map from [0, 1] space back to parameter value

Parameter roles (used by structured tuning workflows): - "structural": algorithm paradigm choices (decomposition, reference points) - "operator": discrete operator family choices (crossover type, mutation type) - "operator_rate": continuous operator parameters (eta, probabilities, sigma) - "population": resource allocation (pop_size, offspring_ratio, archive_size) - "adaptive": meta-parameters (immigration, adaptive operator selection)

Boolean dataclass

Boolean hyperparameter (True/False).

Source code in src/vamos/engine/tuning/racing/param_space.py

@dataclass
class Boolean:
    """Boolean hyperparameter (True/False)."""

    name: str
    role: str = "adaptive"

    def sample(self, rng: np.random.Generator) -> bool:
        return bool(rng.integers(0, 2))

    def to_unit(self, value: bool) -> float:
        return 1.0 if value else 0.0

    def from_unit(self, value: float) -> bool:
        return value >= 0.5

Categorical dataclass

Categorical hyperparameter with discrete choices.

Source code in src/vamos/engine/tuning/racing/param_space.py

@dataclass
class Categorical:
    """Categorical hyperparameter with discrete choices."""

    name: str
    choices: Sequence[Any]
    role: str = "operator"

    def sample(self, rng: np.random.Generator) -> Any:
        return self.choices[int(rng.integers(0, len(self.choices)))]

    def to_unit(self, value: Any) -> float:
        """Map value to [0, 1] space based on choice index."""
        idx = list(self.choices).index(value)
        return 0.0 if len(self.choices) == 1 else idx / float(len(self.choices) - 1)

    def from_unit(self, value: float) -> Any:
        """Map from [0, 1] space to a choice."""
        u = min(max(value, 0.0), 1.0)
        idx = int(round(u * (len(self.choices) - 1)))
        return self.choices[idx]

from_unit(value)

Map from [0, 1] space to a choice.

Source code in src/vamos/engine/tuning/racing/param_space.py

def from_unit(self, value: float) -> Any:
    """Map from [0, 1] space to a choice."""
    u = min(max(value, 0.0), 1.0)
    idx = int(round(u * (len(self.choices) - 1)))
    return self.choices[idx]

to_unit(value)

Map value to [0, 1] space based on choice index.

Source code in src/vamos/engine/tuning/racing/param_space.py

def to_unit(self, value: Any) -> float:
    """Map value to [0, 1] space based on choice index."""
    idx = list(self.choices).index(value)
    return 0.0 if len(self.choices) == 1 else idx / float(len(self.choices) - 1)

Condition dataclass

Simple condition: a parameter is considered active only when expr evaluates to True given the current config.

Expressions are parsed safely (no function calls/attribute access).

Source code in src/vamos/engine/tuning/racing/param_space.py

@dataclass
class Condition:
    """
    Simple condition: a parameter is considered active only when
    `expr` evaluates to True given the current config.

    Expressions are parsed safely (no function calls/attribute access).
    """

    param_name: str
    expr: str  # Python expression using a dict `cfg`

ConditionalBlock dataclass

A block of parameters that are active only when a parent parameter has a specific value.

Source code in src/vamos/engine/tuning/racing/param_space.py

@dataclass
class ConditionalBlock:
    """
    A block of parameters that are active only when a parent parameter
    has a specific value.
    """

    parent_name: str
    parent_value: Any
    params: list[ParamType]  # list of Real, Int, Categorical, etc.

Int dataclass

Integer hyperparameter in [low, high] (inclusive).

Source code in src/vamos/engine/tuning/racing/param_space.py

@dataclass
class Int:
    """Integer hyperparameter in [low, high] (inclusive)."""

    name: str
    low: int
    high: int
    log: bool = False
    role: str = "population"

    def __post_init__(self) -> None:
        if self.high < self.low:
            raise ValueError(f"Int param '{self.name}' has high < low ({self.high} < {self.low}).")
        if self.log and (self.low <= 0 or self.high <= 0):
            raise ValueError(f"Int param '{self.name}' with log=True requires low/high > 0.")

    def sample(self, rng: np.random.Generator) -> int:
        if self.log:
            lo, hi = math.log(self.low), math.log(self.high)
            return int(round(math.exp(rng.uniform(lo, hi))))
        return int(rng.integers(self.low, self.high + 1))

    def to_unit(self, value: int) -> float:
        """Map value to [0, 1] space."""
        v = float(value)
        if self.log:
            lo, hi = math.log(self.low), math.log(self.high)
            return 0.0 if hi == lo else (math.log(v) - lo) / (hi - lo)
        return 0.0 if self.high == self.low else (v - self.low) / (self.high - self.low)

    def from_unit(self, value: float) -> int:
        """Map from [0, 1] space to parameter value."""
        u = min(max(value, 0.0), 1.0)
        if self.log:
            lo, hi = math.log(self.low), math.log(self.high)
            mapped = math.exp(lo + u * (hi - lo))
        else:
            mapped = self.low + u * (self.high - self.low)
        return int(round(mapped))

from_unit(value)

Map from [0, 1] space to parameter value.

Source code in src/vamos/engine/tuning/racing/param_space.py

def from_unit(self, value: float) -> int:
    """Map from [0, 1] space to parameter value."""
    u = min(max(value, 0.0), 1.0)
    if self.log:
        lo, hi = math.log(self.low), math.log(self.high)
        mapped = math.exp(lo + u * (hi - lo))
    else:
        mapped = self.low + u * (self.high - self.low)
    return int(round(mapped))

to_unit(value)

Map value to [0, 1] space.

Source code in src/vamos/engine/tuning/racing/param_space.py

def to_unit(self, value: int) -> float:
    """Map value to [0, 1] space."""
    v = float(value)
    if self.log:
        lo, hi = math.log(self.low), math.log(self.high)
        return 0.0 if hi == lo else (math.log(v) - lo) / (hi - lo)
    return 0.0 if self.high == self.low else (v - self.low) / (self.high - self.low)

ParamSpace dataclass

Defines a hyperparameter space with named parameters.

Example: space = ParamSpace(params={ "lr": Real("lr", 0.001, 0.1, log=True), "epochs": Int("epochs", 10, 100), })

Source code in src/vamos/engine/tuning/racing/param_space.py

@dataclass
class ParamSpace:
    """
    Defines a hyperparameter space with named parameters.

    Example:
        space = ParamSpace(params={
            "lr": Real("lr", 0.001, 0.1, log=True),
            "epochs": Int("epochs", 10, 100),
        })
    """

    params: dict[str, ParamType] = field(default_factory=dict)
    conditions: list[Condition] = field(default_factory=list)

    def sample(self, rng: np.random.Generator | None = None) -> dict[str, Any]:
        """Sample a configuration from the space."""
        rng = np.random.default_rng() if rng is None else rng
        full_cfg = {name: spec.sample(rng) for name, spec in self.params.items()}
        return {name: value for name, value in full_cfg.items() if self.is_active(name, full_cfg)}

    def is_active(self, param_name: str, config: dict[str, Any]) -> bool:
        """Check if param is active given config and conditions."""
        relevant = [c for c in self.conditions if c.param_name == param_name]
        if not relevant:
            return True
        return all(_safe_eval_condition(c.expr, config) for c in relevant)

    def validate(self, config: dict[str, Any]) -> None:
        """Validate that all active params are present and within bounds."""
        for name, spec in self.params.items():
            if not self.is_active(name, config):
                continue
            if name not in config:
                raise ValueError(f"Active parameter '{name}' missing from config")

            value = config[name]
            if isinstance(spec, Real):
                if not (spec.low <= value <= spec.high):
                    raise ValueError(f"Real param '{name}'={value} out of [{spec.low}, {spec.high}]")
            elif isinstance(spec, Int):
                if not (spec.low <= value <= spec.high):
                    raise ValueError(f"Int param '{name}'={value} out of [{spec.low}, {spec.high}]")
            elif isinstance(spec, Categorical):
                if value not in spec.choices:
                    raise ValueError(f"Categorical param '{name}'={value} not in {spec.choices}")

is_active(param_name, config)

Check if param is active given config and conditions.

Source code in src/vamos/engine/tuning/racing/param_space.py

def is_active(self, param_name: str, config: dict[str, Any]) -> bool:
    """Check if param is active given config and conditions."""
    relevant = [c for c in self.conditions if c.param_name == param_name]
    if not relevant:
        return True
    return all(_safe_eval_condition(c.expr, config) for c in relevant)

sample(rng=None)

Sample a configuration from the space.

Source code in src/vamos/engine/tuning/racing/param_space.py

def sample(self, rng: np.random.Generator | None = None) -> dict[str, Any]:
    """Sample a configuration from the space."""
    rng = np.random.default_rng() if rng is None else rng
    full_cfg = {name: spec.sample(rng) for name, spec in self.params.items()}
    return {name: value for name, value in full_cfg.items() if self.is_active(name, full_cfg)}

validate(config)

Validate that all active params are present and within bounds.

Source code in src/vamos/engine/tuning/racing/param_space.py

def validate(self, config: dict[str, Any]) -> None:
    """Validate that all active params are present and within bounds."""
    for name, spec in self.params.items():
        if not self.is_active(name, config):
            continue
        if name not in config:
            raise ValueError(f"Active parameter '{name}' missing from config")

        value = config[name]
        if isinstance(spec, Real):
            if not (spec.low <= value <= spec.high):
                raise ValueError(f"Real param '{name}'={value} out of [{spec.low}, {spec.high}]")
        elif isinstance(spec, Int):
            if not (spec.low <= value <= spec.high):
                raise ValueError(f"Int param '{name}'={value} out of [{spec.low}, {spec.high}]")
        elif isinstance(spec, Categorical):
            if value not in spec.choices:
                raise ValueError(f"Categorical param '{name}'={value} not in {spec.choices}")

Real dataclass

Real-valued hyperparameter in [low, high].

Source code in src/vamos/engine/tuning/racing/param_space.py

@dataclass
class Real:
    """Real-valued hyperparameter in [low, high]."""

    name: str
    low: float
    high: float
    log: bool = False
    role: str = "operator_rate"

    def __post_init__(self) -> None:
        if self.high < self.low:
            raise ValueError(f"Real param '{self.name}' has high < low ({self.high} < {self.low}).")
        if self.log and (self.low <= 0 or self.high <= 0):
            raise ValueError(f"Real param '{self.name}' with log=True requires low/high > 0.")

    def sample(self, rng: np.random.Generator) -> float:
        if self.log:
            lo, hi = math.log(self.low), math.log(self.high)
            return float(math.exp(rng.uniform(lo, hi)))
        return float(rng.uniform(self.low, self.high))

    def to_unit(self, value: float) -> float:
        """Map value to [0, 1] space."""
        v = float(value)
        if self.log:
            lo, hi = math.log(self.low), math.log(self.high)
            return 0.0 if hi == lo else (math.log(v) - lo) / (hi - lo)
        return 0.0 if self.high == self.low else (v - self.low) / (self.high - self.low)

    def from_unit(self, value: float) -> float:
        """Map from [0, 1] space to parameter value."""
        u = min(max(value, 0.0), 1.0)
        if self.log:
            lo, hi = math.log(self.low), math.log(self.high)
            return float(math.exp(lo + u * (hi - lo)))
        return float(self.low + u * (self.high - self.low))

from_unit(value)

Map from [0, 1] space to parameter value.

Source code in src/vamos/engine/tuning/racing/param_space.py

def from_unit(self, value: float) -> float:
    """Map from [0, 1] space to parameter value."""
    u = min(max(value, 0.0), 1.0)
    if self.log:
        lo, hi = math.log(self.low), math.log(self.high)
        return float(math.exp(lo + u * (hi - lo)))
    return float(self.low + u * (self.high - self.low))

to_unit(value)

Map value to [0, 1] space.

Source code in src/vamos/engine/tuning/racing/param_space.py

def to_unit(self, value: float) -> float:
    """Map value to [0, 1] space."""
    v = float(value)
    if self.log:
        lo, hi = math.log(self.low), math.log(self.high)
        return 0.0 if hi == lo else (math.log(v) - lo) / (hi - lo)
    return 0.0 if self.high == self.low else (v - self.low) / (self.high - self.low)

_MissingKeyError

Bases: Exception

Raised when a condition references a cfg key that is absent (inactive parent).

Source code in src/vamos/engine/tuning/racing/param_space.py

class _MissingKeyError(Exception):
    """Raised when a condition references a cfg key that is absent (inactive parent)."""

_safe_eval_condition(expr, cfg)

Evaluate a condition expression against cfg using a restricted AST (no calls/attrs).

If the expression references a cfg key that is missing (because its parent parameter is inactive), the condition evaluates to False.

Source code in src/vamos/engine/tuning/racing/param_space.py

def _safe_eval_condition(expr: str, cfg: dict[str, Any]) -> bool:
    """
    Evaluate a condition expression against cfg using a restricted AST (no calls/attrs).

    If the expression references a cfg key that is missing (because its parent
    parameter is inactive), the condition evaluates to ``False``.
    """
    try:
        tree = ast.parse(expr, mode="eval")
    except SyntaxError as exc:  # pragma: no cover - validated upstream
        raise ValueError(f"Invalid condition syntax: {expr}") from exc

    def _eval(node: ast.AST) -> Any:
        if isinstance(node, ast.Expression):
            return _eval(node.body)
        if isinstance(node, ast.BoolOp):
            values = [_eval(v) for v in node.values]
            if isinstance(node.op, ast.And):
                return all(values)
            if isinstance(node.op, ast.Or):
                return any(values)
            raise ValueError(f"Unsupported boolean operator in: {expr}")
        if isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.Not):
            return not _eval(node.operand)
        if isinstance(node, ast.Compare):
            left = _eval(node.left)
            for op_node, comparator in zip(node.ops, node.comparators):
                rhs = _eval(comparator)
                op_fn = _CMP_OPS.get(type(op_node))
                if op_fn is None:
                    raise ValueError(f"Unsupported comparison in: {expr}")
                if not op_fn(left, rhs):
                    return False
                left = rhs
            return True
        if isinstance(node, ast.Name):
            if node.id != "cfg":
                raise ValueError(f"Only 'cfg' is allowed in conditions (got '{node.id}').")
            return cfg
        if isinstance(node, ast.Constant):
            return node.value
        if isinstance(node, ast.Subscript):
            base = _eval(node.value)
            key = _eval(node.slice)
            try:
                return base[key]
            except KeyError as exc:
                raise _MissingKeyError(key) from exc
            except Exception as exc:
                raise ValueError(f"Failed to access cfg[{key!r}] in condition: {expr}") from exc
        raise ValueError(f"Unsupported expression element in condition: {expr}")

    try:
        result = _eval(tree)
    except _MissingKeyError:
        return False
    if not isinstance(result, bool):
        raise ValueError(f"Condition must evaluate to bool, got {result!r} for: {expr}")
    return result

Diagnostics

vamos.experiment.diagnostics.self_check

Lightweight self-check to verify a VAMOS installation.

Runs tiny NSGA-II jobs on ZDT1 across available backends and reports results. Intended for quick sanity checks; not a benchmark.

run_self_check(verbose=False)

Run a minimal set of smoke checks for each compute backend.

Always exercises NumPy; Numba and MooCore are optional and reported as "skipped" when the dependency is missing.

Parameters:

Name	Type	Description	Default
verbose	bool	If True, log the result of each check at INFO level.	False

Returns:

Type	Description
list[CheckResult]	One entry per backend / encoding variant with status in {"ok", "skipped", "failed"}.

Raises:

Type	Description
RuntimeError	If the mandatory NumPy backend check fails.

Source code in src/vamos/experiment/diagnostics/self_check.py

def run_self_check(verbose: bool = False) -> list[CheckResult]:
    """Run a minimal set of smoke checks for each compute backend.

    Always exercises NumPy; Numba and MooCore are optional and reported
    as ``"skipped"`` when the dependency is missing.

    Parameters
    ----------
    verbose : bool, default False
        If ``True``, log the result of each check at INFO level.

    Returns
    -------
    list[CheckResult]
        One entry per backend / encoding variant with ``status`` in
        ``{"ok", "skipped", "failed"}``.

    Raises
    ------
    RuntimeError
        If the mandatory NumPy backend check fails.
    """
    checks: list[CheckResult] = []
    with tempfile.TemporaryDirectory(prefix="vamos-self-check-") as output_root:
        for engine in ("numpy", "numba", "moocore"):
            result = _run_backend_check(engine, output_root=output_root)
            checks.append(result)
            if verbose:
                status = result.status.upper()
                msg = result.detail or ""
                _logger().info("[self-check] %s: %s %s", result.name, status, msg)

        # Binary and mixed smoke on NumPy only
        for name, label in (("bin_knapsack", "binary"), ("mixed_design", "mixed")):
            selection = make_problem_selection(name)
            cfg = ExperimentConfig(
                population_size=8,
                offspring_population_size=8,
                max_evaluations=20,
                seed=2,
                output_root=output_root,
            )
            try:
                run_single("numpy", "nsgaii", selection, cfg, selection_pressure=2)
                checks.append(CheckResult(name=f"{name}", status="ok"))
            except Exception as exc:  # pragma: no cover - quick smoke only
                checks.append(CheckResult(name=f"{name}", status="failed", detail=str(exc)))

    numpy_ok = next((c for c in checks if c.name == "nsgaii-numpy"), None)
    if numpy_ok is None or numpy_ok.status != "ok":
        detail = numpy_ok.detail if numpy_ok else "NumPy check missing"
        raise RuntimeError(f"VAMOS self-check failed for NumPy backend: {detail}")
    return checks