Bayesian Optimization

The Bayesian-optimization API ranks finite candidate pools for next-experiment selection. It also supports dataframe-level feasibility constraints for synthesizability, processing windows, safety filters, equipment limits, and diversity-aware experimental batch selection. Candidate trust-region policies keep recommendations near known feasible descriptor space, while duplicate policies remove or annotate materials that were already measured, selected, or queued. Known observation noise can be passed directly or prepared from reported uncertainty columns and replicate measurements before using noisy expected-improvement acquisition functions. Recommendation audit summaries convert ranked BO tables into compact report-ready explanations of acquisition scores, posterior uncertainty, feasibility, trust-region status, duplicate filtering, and batch-selection decisions. For multi-objective campaigns, independent BoTorch GP surrogates can be ranked with expected hypervolume improvement while keeping objective directions in original materials-property units. Greedy sequential batch selection can recompute hypervolume improvement after each selected candidate is treated as a pending experiment. Use the experiment-logging API to record recommendations, selected batches, measured outcomes, and compact campaign summaries across closed-loop iterations. BoTorch is a required dependency, so these helpers work in any matgpr installation. It is imported lazily, only when a BoTorch-backed helper runs.

matgpr.bayesian_optimization

BoTorchSurrogate dataclass

Fitted BoTorch surrogate model and metadata for candidate ranking.

Parameters:

Name	Type	Description	Default
model	Any	Fitted BoTorch model. The first implementation uses botorch.models.SingleTaskGP.	required
train_X	Any	Training features as a torch tensor.	required
train_y	Any	Observed targets in the original target direction.	required
objective_y	Any	Training targets in BoTorch maximization space. For minimization problems this is -train_y.	required
maximize	bool	Whether larger original target values are preferred.	required
best_observed_objective	float	Best observed target in BoTorch maximization space.	required
feature_names	tuple[str, ...] | None	Feature names when X_train was provided as a pandas dataframe.	None
noise_variance	Any | None	Known observation-noise variance passed to BoTorch, if provided.	None

Source code in matgpr/bayesian_optimization.py

@dataclass(frozen=True)
class BoTorchSurrogate:
    """Fitted BoTorch surrogate model and metadata for candidate ranking.

    Parameters
    ----------
    model
        Fitted BoTorch model. The first implementation uses
        `botorch.models.SingleTaskGP`.
    train_X
        Training features as a torch tensor.
    train_y
        Observed targets in the original target direction.
    objective_y
        Training targets in BoTorch maximization space. For minimization
        problems this is `-train_y`.
    maximize
        Whether larger original target values are preferred.
    best_observed_objective
        Best observed target in BoTorch maximization space.
    feature_names
        Feature names when `X_train` was provided as a pandas dataframe.
    noise_variance
        Known observation-noise variance passed to BoTorch, if provided.
    """

    model: Any
    train_X: Any
    train_y: Any
    objective_y: Any
    maximize: bool
    best_observed_objective: float
    feature_names: tuple[str, ...] | None = None
    noise_variance: Any | None = None

BayesianOptimizationResult dataclass

Result from finite-pool Bayesian optimization.

Parameters:

Name	Type	Description	Default
recommendations	DataFrame	Top-ranked candidates to evaluate next.	required
ranked_candidates	DataFrame	Full candidate table sorted by acquisition value.	required
surrogate	BoTorchSurrogate	Fitted BoTorch surrogate and associated metadata.	required
acquisition_function	str	Acquisition function used for ranking.	required
maximize	bool	Whether the original target was maximized.	required
top_k	int	Number of recommendations requested.	required

Source code in matgpr/bayesian_optimization.py

@dataclass(frozen=True)
class BayesianOptimizationResult:
    """Result from finite-pool Bayesian optimization.

    Parameters
    ----------
    recommendations
        Top-ranked candidates to evaluate next.
    ranked_candidates
        Full candidate table sorted by acquisition value.
    surrogate
        Fitted BoTorch surrogate and associated metadata.
    acquisition_function
        Acquisition function used for ranking.
    maximize
        Whether the original target was maximized.
    top_k
        Number of recommendations requested.
    """

    recommendations: pd.DataFrame
    ranked_candidates: pd.DataFrame
    surrogate: BoTorchSurrogate
    acquisition_function: str
    maximize: bool
    top_k: int

MultiObjectiveBoTorchSurrogate dataclass

Fitted independent BoTorch surrogates for multi-objective BO.

Each objective is modeled by an independent SingleTaskGP and combined with BoTorch's ModelListGP. Minimize-type objectives are negated internally so all acquisition functions operate in maximization space.

Parameters:

Name	Type	Description	Default
model	Any	Fitted BoTorch ModelListGP.	required
train_X	Any	Training features as a torch tensor.	required
train_y	Any	Observed targets in original objective directions and units.	required
objective_y	Any	Targets transformed to BoTorch maximization space.	required
objective_names	tuple[str, ...]	Stable objective names used in output column labels.	required
objective_directions	tuple[str, ...]	One direction per objective: "maximize" or "minimize".	required
objective_signs	Any	Numeric signs used to convert original targets to maximization space.	required
reference_point	Any	Hypervolume reference point in BoTorch maximization space.	required
reference_point_original	Any	Same reference point in original objective directions and units.	required
feature_names	tuple[str, ...] | None	Feature names when X_train was provided as a pandas dataframe.	None
noise_variance	Any | None	Known observation-noise variance matrix passed to BoTorch, if provided.	None

Source code in matgpr/bayesian_optimization.py

@dataclass(frozen=True)
class MultiObjectiveBoTorchSurrogate:
    """Fitted independent BoTorch surrogates for multi-objective BO.

    Each objective is modeled by an independent `SingleTaskGP` and combined
    with BoTorch's `ModelListGP`. Minimize-type objectives are negated
    internally so all acquisition functions operate in maximization space.

    Parameters
    ----------
    model
        Fitted BoTorch `ModelListGP`.
    train_X
        Training features as a torch tensor.
    train_y
        Observed targets in original objective directions and units.
    objective_y
        Targets transformed to BoTorch maximization space.
    objective_names
        Stable objective names used in output column labels.
    objective_directions
        One direction per objective: `"maximize"` or `"minimize"`.
    objective_signs
        Numeric signs used to convert original targets to maximization space.
    reference_point
        Hypervolume reference point in BoTorch maximization space.
    reference_point_original
        Same reference point in original objective directions and units.
    feature_names
        Feature names when `X_train` was provided as a pandas dataframe.
    noise_variance
        Known observation-noise variance matrix passed to BoTorch, if provided.
    """

    model: Any
    train_X: Any
    train_y: Any
    objective_y: Any
    objective_names: tuple[str, ...]
    objective_directions: tuple[str, ...]
    objective_signs: Any
    reference_point: Any
    reference_point_original: Any
    feature_names: tuple[str, ...] | None = None
    noise_variance: Any | None = None

MultiObjectiveBayesianOptimizationResult dataclass

Result from finite-pool multi-objective Bayesian optimization.

Parameters:

Name	Type	Description	Default
recommendations	DataFrame	Top-ranked candidates to evaluate next.	required
ranked_candidates	DataFrame	Full candidate table sorted by hypervolume-improvement acquisition.	required
surrogate	MultiObjectiveBoTorchSurrogate	Fitted multi-objective BoTorch surrogate and associated metadata.	required
acquisition_function	str	Multi-objective acquisition function used for ranking.	required
objective_names	tuple[str, ...]	Objective names used for prediction columns.	required
objective_directions	tuple[str, ...]	Optimization direction for each objective.	required
top_k	int	Number of recommendations requested.	required

Source code in matgpr/bayesian_optimization.py

@dataclass(frozen=True)
class MultiObjectiveBayesianOptimizationResult:
    """Result from finite-pool multi-objective Bayesian optimization.

    Parameters
    ----------
    recommendations
        Top-ranked candidates to evaluate next.
    ranked_candidates
        Full candidate table sorted by hypervolume-improvement acquisition.
    surrogate
        Fitted multi-objective BoTorch surrogate and associated metadata.
    acquisition_function
        Multi-objective acquisition function used for ranking.
    objective_names
        Objective names used for prediction columns.
    objective_directions
        Optimization direction for each objective.
    top_k
        Number of recommendations requested.
    """

    recommendations: pd.DataFrame
    ranked_candidates: pd.DataFrame
    surrogate: MultiObjectiveBoTorchSurrogate
    acquisition_function: str
    objective_names: tuple[str, ...]
    objective_directions: tuple[str, ...]
    top_k: int

BORecommendationAudit dataclass

Report-ready audit tables for finite-pool BO recommendations.

Parameters:

Name	Type	Description	Default
overview	DataFrame	One-row summary of recommendation count, candidate filtering, scoring, uncertainty, feasibility, trust-region, duplicate, and batch-selection status.	required
recommendations	DataFrame	Compact per-recommendation table with ranks, score columns, policy columns, and an explanatory note for each recommended candidate.	required
score_summary	DataFrame	Numeric score/prediction summary comparing recommended candidates with the full ranked candidate table.	required
policy_summary	DataFrame	Constraint, trust-region, duplicate, and batch-selection audit counts.	required

Source code in matgpr/bayesian_optimization.py

@dataclass(frozen=True)
class BORecommendationAudit:
    """Report-ready audit tables for finite-pool BO recommendations.

    Parameters
    ----------
    overview
        One-row summary of recommendation count, candidate filtering, scoring,
        uncertainty, feasibility, trust-region, duplicate, and batch-selection
        status.
    recommendations
        Compact per-recommendation table with ranks, score columns, policy
        columns, and an explanatory note for each recommended candidate.
    score_summary
        Numeric score/prediction summary comparing recommended candidates with
        the full ranked candidate table.
    policy_summary
        Constraint, trust-region, duplicate, and batch-selection audit counts.
    """

    overview: pd.DataFrame
    recommendations: pd.DataFrame
    score_summary: pd.DataFrame
    policy_summary: pd.DataFrame

    def overview_frame(self) -> pd.DataFrame:
        """Return a copy of the one-row overview table."""
        return self.overview.copy()

    def recommendation_frame(self) -> pd.DataFrame:
        """Return a copy of the per-recommendation audit table."""
        return self.recommendations.copy()

    def score_summary_frame(self) -> pd.DataFrame:
        """Return a copy of the score and prediction summary table."""
        return self.score_summary.copy()

    def policy_summary_frame(self) -> pd.DataFrame:
        """Return a copy of the policy audit summary table."""
        return self.policy_summary.copy()

overview_frame()

Return a copy of the one-row overview table.

Source code in matgpr/bayesian_optimization.py

def overview_frame(self) -> pd.DataFrame:
    """Return a copy of the one-row overview table."""
    return self.overview.copy()

recommendation_frame()

Return a copy of the per-recommendation audit table.

Source code in matgpr/bayesian_optimization.py

def recommendation_frame(self) -> pd.DataFrame:
    """Return a copy of the per-recommendation audit table."""
    return self.recommendations.copy()

score_summary_frame()

Return a copy of the score and prediction summary table.

Source code in matgpr/bayesian_optimization.py

def score_summary_frame(self) -> pd.DataFrame:
    """Return a copy of the score and prediction summary table."""
    return self.score_summary.copy()

policy_summary_frame()

Return a copy of the policy audit summary table.

Source code in matgpr/bayesian_optimization.py

def policy_summary_frame(self) -> pd.DataFrame:
    """Return a copy of the policy audit summary table."""
    return self.policy_summary.copy()

CandidateTrustRegion dataclass

Descriptor-space trust region for finite-pool BO candidates.

Parameters:

Name	Type	Description	Default
centers	Any	One or more center points in the same feature space as the candidate features. This is commonly a dataframe or array of measured features, the current best material, or a manually chosen local search center.	required
radius	float	Maximum allowed distance from the nearest center.	required
metric	str	Distance metric: "euclidean", "manhattan", or "chebyshev".	'euclidean'
feature_columns	tuple[str, ...] | None	Optional feature columns to use when centers or candidate features are dataframes.	None
feature_scales	Any | None	Optional feature scales used before distance calculation. Use None for raw distances, "std" for standard-deviation scaling over centers and candidates, or one positive scale per feature.	None

Source code in matgpr/bayesian_optimization.py

@dataclass(frozen=True)
class CandidateTrustRegion:
    """Descriptor-space trust region for finite-pool BO candidates.

    Parameters
    ----------
    centers
        One or more center points in the same feature space as the candidate
        features. This is commonly a dataframe or array of measured features,
        the current best material, or a manually chosen local search center.
    radius
        Maximum allowed distance from the nearest center.
    metric
        Distance metric: `"euclidean"`, `"manhattan"`, or `"chebyshev"`.
    feature_columns
        Optional feature columns to use when `centers` or candidate features
        are dataframes.
    feature_scales
        Optional feature scales used before distance calculation. Use `None`
        for raw distances, `"std"` for standard-deviation scaling over centers
        and candidates, or one positive scale per feature.
    """

    centers: Any
    radius: float
    metric: str = "euclidean"
    feature_columns: tuple[str, ...] | None = None
    feature_scales: Any | None = None

    def __post_init__(self) -> None:
        radius = float(self.radius)
        if not np.isfinite(radius) or radius < 0.0:
            raise ValueError("CandidateTrustRegion.radius must be finite and non-negative")
        object.__setattr__(self, "radius", radius)
        object.__setattr__(self, "metric", _normalize_distance_metric(self.metric))
        if self.feature_columns is not None:
            columns = tuple(str(column).strip() for column in self.feature_columns)
            if not columns or any(not column for column in columns):
                raise ValueError("feature_columns must contain non-empty column names")
            if len(set(columns)) != len(columns):
                raise ValueError("feature_columns must be unique")
            object.__setattr__(self, "feature_columns", columns)

CandidateDuplicatePolicy dataclass

Duplicate-avoidance policy for finite-pool BO candidates.

Parameters:

Name	Type	Description	Default
existing_candidates	Any	Existing measured, already-selected, or otherwise unavailable candidates. Use metadata rows for key matching, feature rows for distance matching, or both.	required
key_columns	tuple[str, ...] | None	Optional metadata columns used for exact duplicate matching.	None
feature_columns	tuple[str, ...] | None	Optional numeric feature columns used when dataframe features are compared by distance.	None
feature_tolerance	float | None	Optional maximum descriptor-space distance for feature duplicates.	None
metric	str	Distance metric: "euclidean", "manhattan", or "chebyshev".	'euclidean'
feature_scales	Any | None	Optional feature scales used before distance calculation. Use None for raw distances, "std" for standard-deviation scaling over candidates and existing rows, or one positive scale per feature.	None

Source code in matgpr/bayesian_optimization.py

@dataclass(frozen=True)
class CandidateDuplicatePolicy:
    """Duplicate-avoidance policy for finite-pool BO candidates.

    Parameters
    ----------
    existing_candidates
        Existing measured, already-selected, or otherwise unavailable
        candidates. Use metadata rows for key matching, feature rows for
        distance matching, or both.
    key_columns
        Optional metadata columns used for exact duplicate matching.
    feature_columns
        Optional numeric feature columns used when dataframe features are
        compared by distance.
    feature_tolerance
        Optional maximum descriptor-space distance for feature duplicates.
    metric
        Distance metric: `"euclidean"`, `"manhattan"`, or `"chebyshev"`.
    feature_scales
        Optional feature scales used before distance calculation. Use `None`
        for raw distances, `"std"` for standard-deviation scaling over
        candidates and existing rows, or one positive scale per feature.
    """

    existing_candidates: Any
    key_columns: tuple[str, ...] | None = None
    feature_columns: tuple[str, ...] | None = None
    feature_tolerance: float | None = None
    metric: str = "euclidean"
    feature_scales: Any | None = None

    def __post_init__(self) -> None:
        if self.key_columns is None and self.feature_tolerance is None:
            raise ValueError("CandidateDuplicatePolicy requires key_columns or feature_tolerance")
        if self.key_columns is not None:
            columns = tuple(str(column).strip() for column in self.key_columns)
            if not columns or any(not column for column in columns):
                raise ValueError("key_columns must contain non-empty column names")
            if len(set(columns)) != len(columns):
                raise ValueError("key_columns must be unique")
            object.__setattr__(self, "key_columns", columns)
        if self.feature_columns is not None:
            columns = tuple(str(column).strip() for column in self.feature_columns)
            if not columns or any(not column for column in columns):
                raise ValueError("feature_columns must contain non-empty column names")
            if len(set(columns)) != len(columns):
                raise ValueError("feature_columns must be unique")
            object.__setattr__(self, "feature_columns", columns)
        if self.feature_tolerance is not None:
            tolerance = float(self.feature_tolerance)
            if not np.isfinite(tolerance) or tolerance < 0.0:
                raise ValueError(
                    "CandidateDuplicatePolicy.feature_tolerance must be finite and non-negative"
                )
            object.__setattr__(self, "feature_tolerance", tolerance)
        object.__setattr__(self, "metric", _normalize_distance_metric(self.metric))

CandidateConstraint dataclass

Finite-pool feasibility constraint for candidate materials.

Parameters:

Name	Type	Description	Default
name	str	Short label used in violation reports.	required
column	str	Candidate dataframe column to evaluate. The column can come from candidate_data or from X_candidates when it is a dataframe and no separate candidate_data is supplied.	required
lower_bound	float | None	Optional numeric lower bound.	None
upper_bound	float | None	Optional numeric upper bound.	None
allowed_values	tuple[Any, ...] | None	Optional categorical set of allowed values.	None
inclusive	bool	If True, numeric bounds are inclusive. If False, they are strict.	True
allow_missing	bool	If True, missing values satisfy the constraint. The default treats missing values as infeasible.	False

Source code in matgpr/bayesian_optimization.py

@dataclass(frozen=True)
class CandidateConstraint:
    """Finite-pool feasibility constraint for candidate materials.

    Parameters
    ----------
    name
        Short label used in violation reports.
    column
        Candidate dataframe column to evaluate. The column can come from
        `candidate_data` or from `X_candidates` when it is a dataframe and no
        separate `candidate_data` is supplied.
    lower_bound
        Optional numeric lower bound.
    upper_bound
        Optional numeric upper bound.
    allowed_values
        Optional categorical set of allowed values.
    inclusive
        If `True`, numeric bounds are inclusive. If `False`, they are strict.
    allow_missing
        If `True`, missing values satisfy the constraint. The default treats
        missing values as infeasible.
    """

    name: str
    column: str
    lower_bound: float | None = None
    upper_bound: float | None = None
    allowed_values: tuple[Any, ...] | None = None
    inclusive: bool = True
    allow_missing: bool = False

    def __post_init__(self) -> None:
        name = str(self.name).strip()
        column = str(self.column).strip()
        if not name:
            raise ValueError("CandidateConstraint.name must be non-empty")
        if not column:
            raise ValueError("CandidateConstraint.column must be non-empty")
        object.__setattr__(self, "name", name)
        object.__setattr__(self, "column", column)
        if self.lower_bound is None and self.upper_bound is None and self.allowed_values is None:
            raise ValueError(
                "CandidateConstraint requires at least one of lower_bound, "
                "upper_bound, or allowed_values"
            )
        if self.allowed_values is not None and len(self.allowed_values) == 0:
            raise ValueError("allowed_values must not be empty when provided")
        if (
            self.lower_bound is not None
            and self.upper_bound is not None
            and self.lower_bound > self.upper_bound
        ):
            raise ValueError("lower_bound must be <= upper_bound")

    def evaluate(self, candidates: pd.DataFrame) -> pd.Series:
        """Return a boolean feasibility mask for this constraint."""
        if self.column not in candidates.columns:
            raise ValueError(f"Constraint column {self.column!r} is missing")

        values = candidates[self.column]
        feasible = pd.Series(True, index=candidates.index, dtype=bool)

        if self.lower_bound is not None or self.upper_bound is not None:
            numeric = pd.to_numeric(values, errors="coerce")
            if self.lower_bound is not None:
                if self.inclusive:
                    feasible &= numeric >= self.lower_bound
                else:
                    feasible &= numeric > self.lower_bound
            if self.upper_bound is not None:
                if self.inclusive:
                    feasible &= numeric <= self.upper_bound
                else:
                    feasible &= numeric < self.upper_bound

        if self.allowed_values is not None:
            feasible &= values.isin(self.allowed_values)

        if self.allow_missing:
            feasible |= values.isna()
        else:
            feasible &= ~values.isna()

        return feasible.fillna(False).astype(bool)

evaluate(candidates)

Return a boolean feasibility mask for this constraint.

Source code in matgpr/bayesian_optimization.py

def evaluate(self, candidates: pd.DataFrame) -> pd.Series:
    """Return a boolean feasibility mask for this constraint."""
    if self.column not in candidates.columns:
        raise ValueError(f"Constraint column {self.column!r} is missing")

    values = candidates[self.column]
    feasible = pd.Series(True, index=candidates.index, dtype=bool)

    if self.lower_bound is not None or self.upper_bound is not None:
        numeric = pd.to_numeric(values, errors="coerce")
        if self.lower_bound is not None:
            if self.inclusive:
                feasible &= numeric >= self.lower_bound
            else:
                feasible &= numeric > self.lower_bound
        if self.upper_bound is not None:
            if self.inclusive:
                feasible &= numeric <= self.upper_bound
            else:
                feasible &= numeric < self.upper_bound

    if self.allowed_values is not None:
        feasible &= values.isin(self.allowed_values)

    if self.allow_missing:
        feasible |= values.isna()
    else:
        feasible &= ~values.isna()

    return feasible.fillna(False).astype(bool)

summarize_bo_recommendation_audit(recommendations, *, ranked_candidates=None, candidate_count=None, identifier_columns=None)

Summarize why finite-pool BO candidates were recommended.

Use this after suggest_next_experiments, suggest_multi_objective_next_experiments, or a custom ranked candidate table. The audit summarizes acquisition scores, posterior uncertainty, feasibility constraints, trust-region status, duplicate filtering, and batch-selection annotations without requiring another BoTorch call.

Source code in matgpr/bayesian_optimization.py

def summarize_bo_recommendation_audit(
    recommendations: BayesianOptimizationResult
    | MultiObjectiveBayesianOptimizationResult
    | pd.DataFrame,
    *,
    ranked_candidates: pd.DataFrame | None = None,
    candidate_count: int | None = None,
    identifier_columns: Sequence[str] | None = None,
) -> BORecommendationAudit:
    """Summarize why finite-pool BO candidates were recommended.

    Use this after `suggest_next_experiments`,
    `suggest_multi_objective_next_experiments`, or a custom ranked candidate
    table. The audit summarizes acquisition scores, posterior uncertainty,
    feasibility constraints, trust-region status, duplicate filtering, and
    batch-selection annotations without requiring another BoTorch call.
    """
    recommendation_frame, ranked_frame, metadata = _coerce_bo_audit_inputs(
        recommendations,
        ranked_candidates=ranked_candidates,
    )
    candidate_count = _resolve_audit_candidate_count(candidate_count, ranked_frame)
    identifier_columns = _resolve_audit_identifier_columns(
        recommendation_frame,
        identifier_columns,
    )
    score_columns = _bo_audit_score_columns(recommendation_frame, ranked_frame)
    std_columns = [column for column in score_columns if column.startswith("matgpr_predicted_std")]

    recommendation_audit = _bo_recommendation_audit_frame(
        recommendation_frame,
        ranked_frame=ranked_frame,
        identifier_columns=identifier_columns,
        score_columns=score_columns,
        std_columns=std_columns,
    )
    score_summary = _bo_score_summary(
        recommendation_frame,
        ranked_frame,
        score_columns=score_columns,
    )
    policy_summary = _bo_policy_summary(
        recommendation_frame,
        ranked_frame=ranked_frame,
        candidate_count=candidate_count,
    )
    overview = _bo_audit_overview(
        recommendation_frame,
        ranked_frame=ranked_frame,
        candidate_count=candidate_count,
        metadata=metadata,
        score_summary=score_summary,
        policy_summary=policy_summary,
        std_columns=std_columns,
    )
    return BORecommendationAudit(
        overview=overview,
        recommendations=recommendation_audit,
        score_summary=score_summary,
        policy_summary=policy_summary,
    )

apply_candidate_constraints(candidates, constraints, *, require_all=True, feasible_column='matgpr_feasible', violations_column='matgpr_constraint_violations')

Annotate a candidate table with finite-pool feasibility constraints.

Parameters:

Name	Type	Description	Default
candidates	DataFrame	Candidate table to annotate.	required
constraints	CandidateConstraint | list[CandidateConstraint] | tuple[CandidateConstraint, ...]	One or more CandidateConstraint objects.	required
require_all	bool	If True, candidates must satisfy every constraint. If False, satisfying at least one constraint is enough.	True
feasible_column	str	Name of the boolean feasibility column to add.	'matgpr_feasible'
violations_column	str	Name of the semicolon-separated violation-label column to add.	'matgpr_constraint_violations'

Source code in matgpr/bayesian_optimization.py

def apply_candidate_constraints(
    candidates: pd.DataFrame,
    constraints: CandidateConstraint | list[CandidateConstraint] | tuple[CandidateConstraint, ...],
    *,
    require_all: bool = True,
    feasible_column: str = "matgpr_feasible",
    violations_column: str = "matgpr_constraint_violations",
) -> pd.DataFrame:
    """Annotate a candidate table with finite-pool feasibility constraints.

    Parameters
    ----------
    candidates
        Candidate table to annotate.
    constraints
        One or more `CandidateConstraint` objects.
    require_all
        If `True`, candidates must satisfy every constraint. If `False`,
        satisfying at least one constraint is enough.
    feasible_column
        Name of the boolean feasibility column to add.
    violations_column
        Name of the semicolon-separated violation-label column to add.
    """
    if not isinstance(candidates, pd.DataFrame):
        raise TypeError("candidates must be a pandas DataFrame")
    constraint_list = _as_constraint_tuple(constraints)
    if not constraint_list:
        raise ValueError("constraints must contain at least one CandidateConstraint")

    evaluated = [(constraint, constraint.evaluate(candidates)) for constraint in constraint_list]
    mask_frame = pd.DataFrame(
        {constraint.name: mask.to_numpy(dtype=bool) for constraint, mask in evaluated},
        index=candidates.index,
    )
    if require_all:
        feasible = mask_frame.all(axis=1)
    else:
        feasible = mask_frame.any(axis=1)

    result = candidates.copy()
    result[feasible_column] = feasible.to_numpy(dtype=bool)
    result[violations_column] = [
        "; ".join(mask_frame.columns[~row.to_numpy(dtype=bool)]) for _, row in mask_frame.iterrows()
    ]
    return result

apply_candidate_trust_region(candidates, trust_region, *, candidate_features=None, policy='filter', in_region_column='matgpr_in_trust_region', distance_column='matgpr_trust_region_distance')

Annotate or filter candidates by a descriptor-space trust region.

Use this helper to keep a finite BO campaign near known feasible chemistry, processing conditions, or local optima. Distances are measured from each candidate to the nearest trust-region center.

Source code in matgpr/bayesian_optimization.py

def apply_candidate_trust_region(
    candidates: pd.DataFrame,
    trust_region: CandidateTrustRegion,
    *,
    candidate_features: pd.DataFrame | np.ndarray | None = None,
    policy: str = "filter",
    in_region_column: str = "matgpr_in_trust_region",
    distance_column: str = "matgpr_trust_region_distance",
) -> pd.DataFrame:
    """Annotate or filter candidates by a descriptor-space trust region.

    Use this helper to keep a finite BO campaign near known feasible chemistry,
    processing conditions, or local optima. Distances are measured from each
    candidate to the nearest trust-region center.
    """
    if not isinstance(candidates, pd.DataFrame):
        raise TypeError("candidates must be a pandas DataFrame")
    if not isinstance(trust_region, CandidateTrustRegion):
        raise TypeError("trust_region must be a CandidateTrustRegion")
    policy = _validate_candidate_policy(policy)
    in_region_column = _validate_output_name(in_region_column, name="in_region_column")
    distance_column = _validate_output_name(distance_column, name="distance_column")
    if in_region_column == distance_column:
        raise ValueError("in_region_column and distance_column must be distinct")

    candidate_data = candidates if candidate_features is None else candidate_features
    distances, in_region = _trust_region_annotations(
        candidate_data,
        trust_region,
        expected_rows=candidates.shape[0],
    )
    result = candidates.copy()
    result[distance_column] = distances
    result[in_region_column] = in_region
    if policy == "filter":
        result = result.loc[in_region].copy()
    return result.reset_index(drop=True)

apply_candidate_duplicate_policy(candidates, duplicate_policy, *, candidate_features=None, policy='filter', duplicate_column='matgpr_is_duplicate', distance_column='matgpr_duplicate_distance', reason_column='matgpr_duplicate_reason')

Annotate or filter duplicate candidates in a finite BO pool.

Duplicates can be detected by exact metadata keys, by descriptor-space tolerance, or by both. This helps closed-loop BO avoid recommending materials that were already measured, selected, or queued.

Source code in matgpr/bayesian_optimization.py

def apply_candidate_duplicate_policy(
    candidates: pd.DataFrame,
    duplicate_policy: CandidateDuplicatePolicy,
    *,
    candidate_features: pd.DataFrame | np.ndarray | None = None,
    policy: str = "filter",
    duplicate_column: str = "matgpr_is_duplicate",
    distance_column: str = "matgpr_duplicate_distance",
    reason_column: str = "matgpr_duplicate_reason",
) -> pd.DataFrame:
    """Annotate or filter duplicate candidates in a finite BO pool.

    Duplicates can be detected by exact metadata keys, by descriptor-space
    tolerance, or by both. This helps closed-loop BO avoid recommending
    materials that were already measured, selected, or queued.
    """
    if not isinstance(candidates, pd.DataFrame):
        raise TypeError("candidates must be a pandas DataFrame")
    if not isinstance(duplicate_policy, CandidateDuplicatePolicy):
        raise TypeError("duplicate_policy must be a CandidateDuplicatePolicy")
    policy = _validate_candidate_policy(policy)
    duplicate_column = _validate_output_name(duplicate_column, name="duplicate_column")
    distance_column = _validate_output_name(distance_column, name="distance_column")
    reason_column = _validate_output_name(reason_column, name="reason_column")
    _validate_distinct_names(duplicate_column, distance_column, reason_column)

    candidate_data = candidates if candidate_features is None else candidate_features
    is_duplicate, distances, reasons = _duplicate_policy_annotations(
        candidates,
        candidate_data,
        duplicate_policy,
        expected_rows=candidates.shape[0],
    )
    result = candidates.copy()
    result[duplicate_column] = is_duplicate
    result[reason_column] = reasons
    if distances is not None:
        result[distance_column] = distances
    if policy == "filter":
        result = result.loc[~is_duplicate].copy()
    return result.reset_index(drop=True)

select_diverse_batch(candidates, *, top_k, score_column='matgpr_acquisition', feature_columns=None, diversity_weight=0.25, min_distance=None, higher_is_better=True, standardize_features=True, return_all=False)

Select a diversity-aware batch from a ranked candidate table.

This helper is useful when several high-acquisition candidates are nearly identical in descriptor space, but the next experimental batch should cover a broader region of the candidate pool.

Parameters:

Name	Type	Description	Default
candidates	DataFrame	Candidate table containing an acquisition or utility score.	required
top_k	int	Maximum number of candidates to select.	required
score_column	str	Column used as the primary utility score.	'matgpr_acquisition'
feature_columns	list[str] | tuple[str, ...] | None	Numeric columns used to compute diversity distances. If omitted, numeric candidate columns other than reserved matgpr_* output columns and score_column are used.	None
diversity_weight	float	Weight applied to the normalized distance from already selected candidates. Set to 0.0 for ordinary score-only top-k selection.	0.25
min_distance	float | None	Optional minimum distance from the already selected batch. If this cannot be satisfied, fewer than top_k candidates are returned.	None
higher_is_better	bool	If True, larger scores are preferred. If False, lower scores are preferred.	True
standardize_features	bool	If True, standardize diversity features before computing distances.	True
return_all	bool	If True, return the full candidate table with batch annotations. Otherwise return only selected candidates in batch order.	False

Source code in matgpr/bayesian_optimization.py

def select_diverse_batch(
    candidates: pd.DataFrame,
    *,
    top_k: int,
    score_column: str = "matgpr_acquisition",
    feature_columns: list[str] | tuple[str, ...] | None = None,
    diversity_weight: float = 0.25,
    min_distance: float | None = None,
    higher_is_better: bool = True,
    standardize_features: bool = True,
    return_all: bool = False,
) -> pd.DataFrame:
    """Select a diversity-aware batch from a ranked candidate table.

    This helper is useful when several high-acquisition candidates are nearly
    identical in descriptor space, but the next experimental batch should cover
    a broader region of the candidate pool.

    Parameters
    ----------
    candidates
        Candidate table containing an acquisition or utility score.
    top_k
        Maximum number of candidates to select.
    score_column
        Column used as the primary utility score.
    feature_columns
        Numeric columns used to compute diversity distances. If omitted,
        numeric candidate columns other than reserved `matgpr_*` output columns
        and `score_column` are used.
    diversity_weight
        Weight applied to the normalized distance from already selected
        candidates. Set to `0.0` for ordinary score-only top-k selection.
    min_distance
        Optional minimum distance from the already selected batch. If this
        cannot be satisfied, fewer than `top_k` candidates are returned.
    higher_is_better
        If `True`, larger scores are preferred. If `False`, lower scores are
        preferred.
    standardize_features
        If `True`, standardize diversity features before computing distances.
    return_all
        If `True`, return the full candidate table with batch annotations.
        Otherwise return only selected candidates in batch order.
    """
    if not isinstance(candidates, pd.DataFrame):
        raise TypeError("candidates must be a pandas DataFrame")
    if top_k < 1:
        raise ValueError("top_k must be at least 1")
    if diversity_weight < 0.0:
        raise ValueError("diversity_weight must be non-negative")
    if min_distance is not None and min_distance < 0.0:
        raise ValueError("min_distance must be non-negative when provided")
    if score_column not in candidates.columns:
        raise ValueError(f"score_column {score_column!r} is missing")

    result = candidates.copy()
    result["matgpr_batch_selected"] = False
    result["matgpr_batch_order"] = pd.Series(pd.NA, index=result.index, dtype="Int64")
    result["matgpr_batch_score"] = np.nan
    result["matgpr_diversity_distance"] = np.nan

    if result.empty:
        return result.reset_index(drop=True)

    score_values = _as_finite_numeric_vector(result[score_column], name=score_column)
    effective_scores = score_values if higher_is_better else -score_values
    normalized_scores = _normalize_vector(effective_scores)

    if diversity_weight == 0.0 and min_distance is None:
        selected_positions = _top_score_positions(effective_scores, top_k)
        selection_scores = normalized_scores[selected_positions]
        diversity_distances = np.full(len(selected_positions), np.nan)
    else:
        selected_positions, selection_scores, diversity_distances = _greedy_diverse_positions(
            result,
            top_k=top_k,
            score_column=score_column,
            feature_columns=feature_columns,
            normalized_scores=normalized_scores,
            diversity_weight=diversity_weight,
            min_distance=min_distance,
            standardize_features=standardize_features,
        )

    for order, (position, selection_score, distance) in enumerate(
        zip(selected_positions, selection_scores, diversity_distances),
        start=1,
    ):
        label = result.index[position]
        result.loc[label, "matgpr_batch_selected"] = True
        result.loc[label, "matgpr_batch_order"] = order
        result.loc[label, "matgpr_batch_score"] = selection_score
        result.loc[label, "matgpr_diversity_distance"] = distance

    if return_all:
        return result.sort_values(
            ["matgpr_batch_selected", "matgpr_batch_order"],
            ascending=[False, True],
            na_position="last",
        ).reset_index(drop=True)

    return (
        result[result["matgpr_batch_selected"]]
        .sort_values("matgpr_batch_order")
        .reset_index(drop=True)
    )

observation_noise_variance(data, *, variance_column=None, std_column=None, sem_column=None, replicate_group_column=None, target_column=None, min_variance=1e-12, ddof=1)

Build known observation-noise variances for BoTorch GPR.

This helper converts common experimental uncertainty formats into the per-training-row variance vector expected by fit_botorch_surrogate and suggest_next_experiments.

Provide exactly one source of known noise:

• variance_column: target variance in squared target units.
• std_column: target standard deviation in target units.
• sem_column: standard error of the mean in target units.
• replicate_group_column with target_column: estimate replicate-group target variance from repeated measurements.

Replicate groups with too few rows to estimate variance receive the pooled replicate variance when available, otherwise min_variance.

Source code in matgpr/bayesian_optimization.py

def observation_noise_variance(
    data: pd.DataFrame,
    *,
    variance_column: str | None = None,
    std_column: str | None = None,
    sem_column: str | None = None,
    replicate_group_column: str | None = None,
    target_column: str | None = None,
    min_variance: float = 1e-12,
    ddof: int = 1,
) -> pd.Series:
    """Build known observation-noise variances for BoTorch GPR.

    This helper converts common experimental uncertainty formats into the
    per-training-row variance vector expected by `fit_botorch_surrogate` and
    `suggest_next_experiments`.

    Provide exactly one source of known noise:

    - `variance_column`: target variance in squared target units.
    - `std_column`: target standard deviation in target units.
    - `sem_column`: standard error of the mean in target units.
    - `replicate_group_column` with `target_column`: estimate replicate-group
      target variance from repeated measurements.

    Replicate groups with too few rows to estimate variance receive the pooled
    replicate variance when available, otherwise `min_variance`.
    """
    if not isinstance(data, pd.DataFrame):
        raise TypeError("data must be a pandas DataFrame")
    if not np.isfinite(min_variance) or min_variance < 0.0:
        raise ValueError("min_variance must be a finite non-negative value")
    if ddof < 0:
        raise ValueError("ddof must be non-negative")

    replicate_requested = replicate_group_column is not None or target_column is not None
    source_count = sum(
        value is not None
        for value in (
            variance_column,
            std_column,
            sem_column,
        )
    ) + int(replicate_requested)
    if source_count != 1:
        raise ValueError(
            "Provide exactly one observation-noise source: variance_column, "
            "std_column, sem_column, or replicate_group_column with target_column"
        )
    if replicate_requested and (replicate_group_column is None or target_column is None):
        raise ValueError("replicate_group_column and target_column must be provided together")

    if variance_column is not None:
        variance = _nonnegative_numeric_column(
            data,
            variance_column,
            label="variance_column",
        )
    elif std_column is not None:
        std = _nonnegative_numeric_column(data, std_column, label="std_column")
        variance = std**2
    elif sem_column is not None:
        sem = _nonnegative_numeric_column(data, sem_column, label="sem_column")
        variance = sem**2
    else:
        variance = _replicate_group_variance(
            data,
            group_column=str(replicate_group_column),
            target_column=str(target_column),
            min_variance=min_variance,
            ddof=ddof,
        )

    variance = variance.astype(float).clip(lower=min_variance)
    variance.name = "matgpr_noise_variance"
    return variance

fit_botorch_surrogate(X_train, y_train, *, maximize=True, noise_variance=None, normalize_features=True, standardize_target=True, fit_model=True, device=None)

Fit a BoTorch SingleTaskGP surrogate for Bayesian optimization.

BoTorch is a required dependency, so this helper is available in any matgpr installation. BoTorch itself is imported lazily, so it is only loaded when a Bayesian-optimization helper is called.

Parameters:

Name	Type	Description	Default
X_train	DataFrame | ndarray	Numeric training features with shape (n_samples, n_features).	required
y_train	Series | ndarray	Numeric target values with shape (n_samples,).	required
maximize	bool	If True, larger target values are preferred. If False, the target is negated internally so BoTorch can solve the equivalent maximization problem.	True
noise_variance	float | ndarray | Series | None	Optional known observation-noise variance. Provide either a scalar or one variance per training sample. These are variances, not standard deviations.	None
normalize_features	bool	If True, use BoTorch's Normalize input transform.	True
standardize_target	bool	If True, use BoTorch's Standardize outcome transform.	True
fit_model	bool	If True, optimize GP hyperparameters with fit_gpytorch_mll. Disable only for tests or advanced workflows that fit the model later.	True
device	str | None	Optional torch device string such as "cpu" or "cuda".	None

Source code in matgpr/bayesian_optimization.py

def fit_botorch_surrogate(
    X_train: pd.DataFrame | np.ndarray,
    y_train: pd.Series | np.ndarray,
    *,
    maximize: bool = True,
    noise_variance: float | np.ndarray | pd.Series | None = None,
    normalize_features: bool = True,
    standardize_target: bool = True,
    fit_model: bool = True,
    device: str | None = None,
) -> BoTorchSurrogate:
    """Fit a BoTorch `SingleTaskGP` surrogate for Bayesian optimization.

    BoTorch is a required dependency, so this helper is available in any
    `matgpr` installation. BoTorch itself is imported lazily, so it is only
    loaded when a Bayesian-optimization helper is called.

    Parameters
    ----------
    X_train
        Numeric training features with shape `(n_samples, n_features)`.
    y_train
        Numeric target values with shape `(n_samples,)`.
    maximize
        If `True`, larger target values are preferred. If `False`, the target
        is negated internally so BoTorch can solve the equivalent maximization
        problem.
    noise_variance
        Optional known observation-noise variance. Provide either a scalar or
        one variance per training sample. These are variances, not standard
        deviations.
    normalize_features
        If `True`, use BoTorch's `Normalize` input transform.
    standardize_target
        If `True`, use BoTorch's `Standardize` outcome transform.
    fit_model
        If `True`, optimize GP hyperparameters with `fit_gpytorch_mll`.
        Disable only for tests or advanced workflows that fit the model later.
    device
        Optional torch device string such as `"cpu"` or `"cuda"`.
    """
    import torch
    from botorch.fit import fit_gpytorch_mll
    from botorch.models import SingleTaskGP
    from botorch.models.transforms.input import Normalize
    from botorch.models.transforms.outcome import Standardize
    from gpytorch.mlls import ExactMarginalLogLikelihood

    train_array, feature_names, _ = _as_numeric_matrix(X_train, name="X_train")
    target_array = _as_target_vector(y_train, expected_length=train_array.shape[0], name="y_train")
    objective_array = target_array if maximize else -target_array

    torch_device = torch.device(device) if device is not None else None
    train_X = _to_tensor(torch, train_array, device=torch_device)
    train_Y = _to_tensor(torch, objective_array.reshape(-1, 1), device=torch_device)
    train_y_original = _to_tensor(torch, target_array.reshape(-1, 1), device=torch_device)

    model_kwargs: dict[str, Any] = {}
    noise_tensor = None
    if noise_variance is not None:
        noise_array = _as_noise_variance(
            noise_variance,
            expected_length=train_array.shape[0],
        )
        noise_tensor = _to_tensor(
            torch,
            noise_array.reshape(-1, 1),
            device=torch_device,
        )
        model_kwargs["train_Yvar"] = noise_tensor
    if normalize_features:
        model_kwargs["input_transform"] = Normalize(d=train_X.shape[-1])
    if standardize_target:
        model_kwargs["outcome_transform"] = Standardize(m=1)

    model = SingleTaskGP(train_X=train_X, train_Y=train_Y, **model_kwargs)
    if torch_device is not None:
        model = model.to(torch_device)

    if fit_model:
        mll = ExactMarginalLogLikelihood(model.likelihood, model)
        fit_gpytorch_mll(mll)

    return BoTorchSurrogate(
        model=model,
        train_X=train_X,
        train_y=train_y_original,
        objective_y=train_Y,
        maximize=maximize,
        best_observed_objective=float(train_Y.max().detach().cpu().item()),
        feature_names=feature_names,
        noise_variance=noise_tensor,
    )

fit_multi_objective_botorch_surrogate(X_train, y_train, *, objective_directions='maximize', objective_names=None, reference_point=None, noise_variance=None, normalize_features=True, standardize_targets=True, fit_model=True, device=None)

Fit independent BoTorch GP surrogates for multi-objective BO.

Parameters:

Name	Type	Description	Default
X_train	DataFrame | ndarray	Numeric training features with shape (n_samples, n_features).	required
y_train	DataFrame | ndarray	Numeric target matrix with shape (n_samples, n_objectives). At least two objectives are required.	required
objective_directions	str | Sequence[str]	Either one direction applied to every objective or one direction per objective. Use "maximize" for objectives where larger values are better and "minimize" for objectives where smaller values are better.	'maximize'
objective_names	Sequence[str] | None	Optional objective names used in output columns. If omitted, dataframe column names are used when available.	None
reference_point	Sequence[float] | ndarray | Series | None	Optional hypervolume reference point in original objective units and directions. If omitted, a conservative point worse than the observed training targets is estimated.	None
noise_variance	float | ndarray | DataFrame | None	Optional known observation-noise variance. Provide a scalar, a vector with one variance per training row, or a matrix with one column per objective.	None
normalize_features	bool	If True, use BoTorch's Normalize input transform for each GP.	True
standardize_targets	bool	If True, use BoTorch's Standardize outcome transform for each GP.	True
fit_model	bool	If True, optimize GP hyperparameters with fit_gpytorch_mll.	True
device	str | None	Optional torch device string such as "cpu" or "cuda".	None

Source code in matgpr/bayesian_optimization.py

def fit_multi_objective_botorch_surrogate(
    X_train: pd.DataFrame | np.ndarray,
    y_train: pd.DataFrame | np.ndarray,
    *,
    objective_directions: str | Sequence[str] = "maximize",
    objective_names: Sequence[str] | None = None,
    reference_point: Sequence[float] | np.ndarray | pd.Series | None = None,
    noise_variance: float | np.ndarray | pd.DataFrame | None = None,
    normalize_features: bool = True,
    standardize_targets: bool = True,
    fit_model: bool = True,
    device: str | None = None,
) -> MultiObjectiveBoTorchSurrogate:
    """Fit independent BoTorch GP surrogates for multi-objective BO.

    Parameters
    ----------
    X_train
        Numeric training features with shape `(n_samples, n_features)`.
    y_train
        Numeric target matrix with shape `(n_samples, n_objectives)`. At least
        two objectives are required.
    objective_directions
        Either one direction applied to every objective or one direction per
        objective. Use `"maximize"` for objectives where larger values are
        better and `"minimize"` for objectives where smaller values are better.
    objective_names
        Optional objective names used in output columns. If omitted, dataframe
        column names are used when available.
    reference_point
        Optional hypervolume reference point in original objective units and
        directions. If omitted, a conservative point worse than the observed
        training targets is estimated.
    noise_variance
        Optional known observation-noise variance. Provide a scalar, a vector
        with one variance per training row, or a matrix with one column per
        objective.
    normalize_features
        If `True`, use BoTorch's `Normalize` input transform for each GP.
    standardize_targets
        If `True`, use BoTorch's `Standardize` outcome transform for each GP.
    fit_model
        If `True`, optimize GP hyperparameters with `fit_gpytorch_mll`.
    device
        Optional torch device string such as `"cpu"` or `"cuda"`.
    """
    import torch
    from botorch.fit import fit_gpytorch_mll
    from botorch.models import SingleTaskGP
    from botorch.models.model_list_gp_regression import ModelListGP
    from botorch.models.transforms.input import Normalize
    from botorch.models.transforms.outcome import Standardize
    from gpytorch.mlls import SumMarginalLogLikelihood

    train_array, feature_names, _ = _as_numeric_matrix(X_train, name="X_train")
    target_array, inferred_names = _as_target_matrix(
        y_train,
        expected_length=train_array.shape[0],
        name="y_train",
    )
    objective_names = _resolve_objective_names(
        objective_names,
        inferred_names=inferred_names,
        n_objectives=target_array.shape[1],
    )
    directions, signs = _resolve_objective_directions(
        objective_directions,
        n_objectives=target_array.shape[1],
    )
    objective_array = target_array * signs.reshape(1, -1)
    ref_array, ref_original = _resolve_reference_point(
        reference_point,
        objective_array=objective_array,
        signs=signs,
    )

    torch_device = torch.device(device) if device is not None else None
    train_X = _to_tensor(torch, train_array, device=torch_device)
    train_y_original = _to_tensor(torch, target_array, device=torch_device)
    train_Y = _to_tensor(torch, objective_array, device=torch_device)
    reference_tensor = _to_tensor(torch, ref_array, device=torch_device)
    reference_original_tensor = _to_tensor(torch, ref_original, device=torch_device)
    signs_tensor = _to_tensor(torch, signs, device=torch_device)

    noise_tensor = None
    noise_array = None
    if noise_variance is not None:
        noise_array = _as_noise_variance_matrix(
            noise_variance,
            expected_length=train_array.shape[0],
            n_objectives=target_array.shape[1],
        )
        noise_tensor = _to_tensor(torch, noise_array, device=torch_device)

    models = []
    for objective_index in range(target_array.shape[1]):
        model_kwargs: dict[str, Any] = {}
        if noise_array is not None:
            model_kwargs["train_Yvar"] = _to_tensor(
                torch,
                noise_array[:, [objective_index]],
                device=torch_device,
            )
        if normalize_features:
            model_kwargs["input_transform"] = Normalize(d=train_X.shape[-1])
        if standardize_targets:
            model_kwargs["outcome_transform"] = Standardize(m=1)
        models.append(
            SingleTaskGP(
                train_X=train_X,
                train_Y=train_Y[:, [objective_index]],
                **model_kwargs,
            )
        )

    model = ModelListGP(*models)
    if torch_device is not None:
        model = model.to(torch_device)

    if fit_model:
        mll = SumMarginalLogLikelihood(model.likelihood, model)
        fit_gpytorch_mll(mll)

    return MultiObjectiveBoTorchSurrogate(
        model=model,
        train_X=train_X,
        train_y=train_y_original,
        objective_y=train_Y,
        objective_names=objective_names,
        objective_directions=directions,
        objective_signs=signs_tensor,
        reference_point=reference_tensor,
        reference_point_original=reference_original_tensor,
        feature_names=feature_names,
        noise_variance=noise_tensor,
    )

rank_discrete_candidates(surrogate, X_candidates, *, acquisition_function='log_expected_improvement', top_k=None, beta=0.2, candidate_data=None, constraints=None, constraint_policy='filter', trust_region=None, trust_region_policy='filter', duplicate_policy=None, duplicate_policy_action='filter')

Rank a finite pool of candidate materials using a BoTorch acquisition.

This is the first Bayesian-optimization workflow in matgpr: users provide already-featurized candidate materials, and the function ranks them for the next experimental or simulation batch.

Parameters:

Name	Type	Description	Default
surrogate	BoTorchSurrogate	Fitted surrogate returned by fit_botorch_surrogate.	required
X_candidates	DataFrame | ndarray	Numeric candidate features with shape (n_candidates, n_features).	required
acquisition_function	str	Acquisition function name. Supported values are "log_expected_improvement", "log_noisy_expected_improvement", "expected_improvement", "noisy_expected_improvement", "upper_confidence_bound", and "probability_of_improvement".	'log_expected_improvement'
top_k	int | None	Optional number of rows to return after ranking. If omitted, all candidates are returned.	None
beta	float	Exploration weight for upper confidence bound.	0.2
candidate_data	DataFrame | None	Optional metadata dataframe to carry through to the ranked output, such as formulas, SMILES strings, synthesis conditions, or sample IDs.	None
constraints	CandidateConstraint | list[CandidateConstraint] | tuple[CandidateConstraint, ...] | None	Optional finite-pool feasibility constraints applied after candidate metadata and acquisition scores are assembled.	None
constraint_policy	str	"filter" returns only feasible candidates. "annotate" keeps all candidates and adds feasibility columns.	'filter'
trust_region	CandidateTrustRegion | None	Optional descriptor-space trust region for local candidate filtering.	None
trust_region_policy	str	"filter" returns only candidates inside the trust region. "annotate" keeps all candidates and adds trust-region columns.	'filter'
duplicate_policy	CandidateDuplicatePolicy | None	Optional duplicate-avoidance policy for already measured, selected, or queued candidates.	None
duplicate_policy_action	str	"filter" removes duplicate candidates. "annotate" keeps all candidates and adds duplicate-audit columns.	'filter'

Source code in matgpr/bayesian_optimization.py

def rank_discrete_candidates(
    surrogate: BoTorchSurrogate,
    X_candidates: pd.DataFrame | np.ndarray,
    *,
    acquisition_function: str = "log_expected_improvement",
    top_k: int | None = None,
    beta: float = 0.2,
    candidate_data: pd.DataFrame | None = None,
    constraints: CandidateConstraint
    | list[CandidateConstraint]
    | tuple[CandidateConstraint, ...]
    | None = None,
    constraint_policy: str = "filter",
    trust_region: CandidateTrustRegion | None = None,
    trust_region_policy: str = "filter",
    duplicate_policy: CandidateDuplicatePolicy | None = None,
    duplicate_policy_action: str = "filter",
) -> pd.DataFrame:
    """Rank a finite pool of candidate materials using a BoTorch acquisition.

    This is the first Bayesian-optimization workflow in `matgpr`: users provide
    already-featurized candidate materials, and the function ranks them for the
    next experimental or simulation batch.

    Parameters
    ----------
    surrogate
        Fitted surrogate returned by `fit_botorch_surrogate`.
    X_candidates
        Numeric candidate features with shape `(n_candidates, n_features)`.
    acquisition_function
        Acquisition function name. Supported values are
        `"log_expected_improvement"`, `"log_noisy_expected_improvement"`,
        `"expected_improvement"`, `"noisy_expected_improvement"`,
        `"upper_confidence_bound"`, and `"probability_of_improvement"`.
    top_k
        Optional number of rows to return after ranking. If omitted, all
        candidates are returned.
    beta
        Exploration weight for upper confidence bound.
    candidate_data
        Optional metadata dataframe to carry through to the ranked output, such
        as formulas, SMILES strings, synthesis conditions, or sample IDs.
    constraints
        Optional finite-pool feasibility constraints applied after candidate
        metadata and acquisition scores are assembled.
    constraint_policy
        `"filter"` returns only feasible candidates. `"annotate"` keeps all
        candidates and adds feasibility columns.
    trust_region
        Optional descriptor-space trust region for local candidate filtering.
    trust_region_policy
        `"filter"` returns only candidates inside the trust region.
        `"annotate"` keeps all candidates and adds trust-region columns.
    duplicate_policy
        Optional duplicate-avoidance policy for already measured, selected, or
        queued candidates.
    duplicate_policy_action
        `"filter"` removes duplicate candidates. `"annotate"` keeps all
        candidates and adds duplicate-audit columns.
    """
    if top_k is not None and top_k < 1:
        raise ValueError("top_k must be at least 1 when provided")
    constraint_policy = _validate_constraint_policy(constraint_policy)

    import torch

    candidate_array, _, candidate_index = _as_numeric_matrix(
        X_candidates,
        name="X_candidates",
    )
    if candidate_array.shape[1] != surrogate.train_X.shape[-1]:
        raise ValueError(
            "X_candidates must have the same number of features as X_train "
            f"({candidate_array.shape[1]} != {surrogate.train_X.shape[-1]})"
        )
    ranked = _candidate_frame(
        X_candidates=X_candidates,
        candidate_data=candidate_data,
        candidate_index=candidate_index,
    )

    candidate_X = _to_tensor(
        torch,
        candidate_array,
        device=surrogate.train_X.device,
    )
    acquisition = _build_acquisition_function(
        acquisition_function,
        surrogate=surrogate,
        beta=beta,
    )

    surrogate.model.eval()
    with torch.no_grad():
        scores = acquisition(candidate_X.unsqueeze(-2)).detach().cpu().numpy().reshape(-1)
        posterior = surrogate.model.posterior(candidate_X)
        transformed_mean = posterior.mean.detach().cpu().numpy().reshape(-1)
        std = posterior.variance.clamp_min(0.0).sqrt().detach().cpu().numpy().reshape(-1)

    predicted_mean = transformed_mean if surrogate.maximize else -transformed_mean
    ranked["matgpr_predicted_mean"] = predicted_mean
    ranked["matgpr_predicted_std"] = std
    ranked["matgpr_acquisition"] = scores
    if constraints is not None:
        ranked = apply_candidate_constraints(ranked, constraints)
        if constraint_policy == "filter":
            keep_mask = ranked["matgpr_feasible"].to_numpy(dtype=bool)
            ranked = ranked.loc[keep_mask].copy()
            candidate_array = candidate_array[keep_mask]
    ranked = _apply_candidate_domain_policies(
        ranked,
        candidate_array,
        trust_region=trust_region,
        trust_region_policy=trust_region_policy,
        duplicate_policy=duplicate_policy,
        duplicate_policy_action=duplicate_policy_action,
    )
    ranked = ranked.sort_values("matgpr_acquisition", ascending=False).reset_index(drop=True)
    ranked.insert(0, "matgpr_rank", np.arange(1, ranked.shape[0] + 1))

    if top_k is not None:
        return ranked.head(top_k).reset_index(drop=True)
    return ranked

rank_multi_objective_discrete_candidates(surrogate, X_candidates, *, acquisition_function='q_log_expected_hypervolume_improvement', top_k=None, candidate_data=None, constraints=None, constraint_policy='filter', trust_region=None, trust_region_policy='filter', duplicate_policy=None, duplicate_policy_action='filter', mc_samples=128, sampler_seed=42)

Rank a finite pool with multi-objective hypervolume improvement.

The returned table is sorted by acquisition value and includes posterior mean and standard-deviation columns for every objective in original units. Acquisition functions operate in BoTorch maximization space, so minimize-type objectives are handled by the fitted surrogate.

Source code in matgpr/bayesian_optimization.py

def rank_multi_objective_discrete_candidates(
    surrogate: MultiObjectiveBoTorchSurrogate,
    X_candidates: pd.DataFrame | np.ndarray,
    *,
    acquisition_function: str = "q_log_expected_hypervolume_improvement",
    top_k: int | None = None,
    candidate_data: pd.DataFrame | None = None,
    constraints: CandidateConstraint
    | list[CandidateConstraint]
    | tuple[CandidateConstraint, ...]
    | None = None,
    constraint_policy: str = "filter",
    trust_region: CandidateTrustRegion | None = None,
    trust_region_policy: str = "filter",
    duplicate_policy: CandidateDuplicatePolicy | None = None,
    duplicate_policy_action: str = "filter",
    mc_samples: int = 128,
    sampler_seed: int | None = 42,
) -> pd.DataFrame:
    """Rank a finite pool with multi-objective hypervolume improvement.

    The returned table is sorted by acquisition value and includes posterior
    mean and standard-deviation columns for every objective in original units.
    Acquisition functions operate in BoTorch maximization space, so
    minimize-type objectives are handled by the fitted surrogate.
    """
    if top_k is not None and top_k < 1:
        raise ValueError("top_k must be at least 1 when provided")
    constraint_policy = _validate_constraint_policy(constraint_policy)
    mc_samples = _validate_mc_samples(mc_samples)

    import torch

    candidate_array, _, candidate_index = _as_numeric_matrix(
        X_candidates,
        name="X_candidates",
    )
    if candidate_array.shape[1] != surrogate.train_X.shape[-1]:
        raise ValueError(
            "X_candidates must have the same number of features as X_train "
            f"({candidate_array.shape[1]} != {surrogate.train_X.shape[-1]})"
        )
    ranked = _candidate_frame(
        X_candidates=X_candidates,
        candidate_data=candidate_data,
        candidate_index=candidate_index,
    )

    candidate_X = _to_tensor(
        torch,
        candidate_array,
        device=surrogate.train_X.device,
    )
    acquisition = _build_multi_objective_acquisition_function(
        acquisition_function,
        surrogate=surrogate,
        mc_samples=mc_samples,
        sampler_seed=sampler_seed,
    )

    surrogate.model.eval()
    with torch.no_grad():
        scores = acquisition(candidate_X.unsqueeze(-2)).detach().cpu().numpy().reshape(-1)
        posterior = surrogate.model.posterior(candidate_X)
        objective_mean = posterior.mean.detach().cpu()
        std = posterior.variance.clamp_min(0.0).sqrt().detach().cpu().numpy()

    signs = surrogate.objective_signs.detach().cpu()
    predicted_mean = (objective_mean * signs).numpy()
    objective_columns = []
    for objective_index, objective_name in enumerate(surrogate.objective_names):
        slug = _slugify_objective_name(objective_name)
        mean_column = f"matgpr_predicted_mean_{slug}"
        std_column = f"matgpr_predicted_std_{slug}"
        ranked[mean_column] = predicted_mean[:, objective_index]
        ranked[std_column] = std[:, objective_index]
        objective_columns.append(mean_column)

    ranked["matgpr_acquisition"] = scores
    ranked["matgpr_predicted_pareto_front"] = pareto_front_mask(
        ranked,
        [
            ObjectiveSpec(
                name=surrogate.objective_names[index],
                column=column,
                goal=surrogate.objective_directions[index],
            )
            for index, column in enumerate(objective_columns)
        ],
    ).to_numpy(dtype=bool)

    if constraints is not None:
        ranked = apply_candidate_constraints(ranked, constraints)
        if constraint_policy == "filter":
            keep_mask = ranked["matgpr_feasible"].to_numpy(dtype=bool)
            ranked = ranked.loc[keep_mask].copy()
            candidate_array = candidate_array[keep_mask]
    ranked = _apply_candidate_domain_policies(
        ranked,
        candidate_array,
        trust_region=trust_region,
        trust_region_policy=trust_region_policy,
        duplicate_policy=duplicate_policy,
        duplicate_policy_action=duplicate_policy_action,
    )
    ranked = ranked.sort_values("matgpr_acquisition", ascending=False).reset_index(drop=True)
    ranked.insert(0, "matgpr_rank", np.arange(1, ranked.shape[0] + 1))

    if top_k is not None:
        return ranked.head(top_k).reset_index(drop=True)
    return ranked

select_sequential_multi_objective_batch(surrogate, X_candidates, *, top_k, acquisition_function='q_log_expected_hypervolume_improvement', candidate_data=None, constraints=None, constraint_policy='filter', trust_region=None, trust_region_policy='filter', duplicate_policy=None, duplicate_policy_action='filter', mc_samples=128, sampler_seed=42, return_all=False)

Select a greedy multi-objective batch from a finite candidate pool.

Candidates are selected one at a time. After each selection, the selected candidates are treated as pending points and the hypervolume-improvement acquisition is recomputed for the remaining candidates. This is useful when recommending a small experimental batch, because the second and later choices account for what was already selected rather than simply taking the top individual acquisition scores.

Parameters:

Name	Type	Description	Default
surrogate	MultiObjectiveBoTorchSurrogate	Fitted multi-objective surrogate returned by fit_multi_objective_botorch_surrogate.	required
X_candidates	DataFrame | ndarray	Numeric candidate features with shape (n_candidates, n_features).	required
top_k	int	Maximum number of candidates to select.	required
acquisition_function	str	Multi-objective acquisition function name. Supported values match rank_multi_objective_discrete_candidates.	'q_log_expected_hypervolume_improvement'
candidate_data	DataFrame | None	Optional metadata dataframe to carry through to the output.	None
constraints	CandidateConstraint | list[CandidateConstraint] | tuple[CandidateConstraint, ...] | None	Optional finite-pool feasibility constraints.	None
constraint_policy	str	"filter" selects only feasible candidates. "annotate" keeps all candidates and adds feasibility columns.	'filter'
mc_samples	int	Number of Monte Carlo samples for hypervolume acquisition estimates.	128
sampler_seed	int | None	Optional seed for reproducible Sobol sampling.	42
return_all	bool	If True, return the full annotated candidate table. Otherwise return only selected candidates in sequential batch order.	False

Source code in matgpr/bayesian_optimization.py

def select_sequential_multi_objective_batch(
    surrogate: MultiObjectiveBoTorchSurrogate,
    X_candidates: pd.DataFrame | np.ndarray,
    *,
    top_k: int,
    acquisition_function: str = "q_log_expected_hypervolume_improvement",
    candidate_data: pd.DataFrame | None = None,
    constraints: CandidateConstraint
    | list[CandidateConstraint]
    | tuple[CandidateConstraint, ...]
    | None = None,
    constraint_policy: str = "filter",
    trust_region: CandidateTrustRegion | None = None,
    trust_region_policy: str = "filter",
    duplicate_policy: CandidateDuplicatePolicy | None = None,
    duplicate_policy_action: str = "filter",
    mc_samples: int = 128,
    sampler_seed: int | None = 42,
    return_all: bool = False,
) -> pd.DataFrame:
    """Select a greedy multi-objective batch from a finite candidate pool.

    Candidates are selected one at a time. After each selection, the selected
    candidates are treated as pending points and the hypervolume-improvement
    acquisition is recomputed for the remaining candidates. This is useful
    when recommending a small experimental batch, because the second and later
    choices account for what was already selected rather than simply taking the
    top individual acquisition scores.

    Parameters
    ----------
    surrogate
        Fitted multi-objective surrogate returned by
        `fit_multi_objective_botorch_surrogate`.
    X_candidates
        Numeric candidate features with shape `(n_candidates, n_features)`.
    top_k
        Maximum number of candidates to select.
    acquisition_function
        Multi-objective acquisition function name. Supported values match
        `rank_multi_objective_discrete_candidates`.
    candidate_data
        Optional metadata dataframe to carry through to the output.
    constraints
        Optional finite-pool feasibility constraints.
    constraint_policy
        `"filter"` selects only feasible candidates. `"annotate"` keeps all
        candidates and adds feasibility columns.
    mc_samples
        Number of Monte Carlo samples for hypervolume acquisition estimates.
    sampler_seed
        Optional seed for reproducible Sobol sampling.
    return_all
        If `True`, return the full annotated candidate table. Otherwise return
        only selected candidates in sequential batch order.
    """
    if top_k < 1:
        raise ValueError("top_k must be at least 1")
    constraint_policy = _validate_constraint_policy(constraint_policy)
    mc_samples = _validate_mc_samples(mc_samples)

    import torch

    candidate_array, _, candidate_index = _as_numeric_matrix(
        X_candidates,
        name="X_candidates",
    )
    if candidate_array.shape[1] != surrogate.train_X.shape[-1]:
        raise ValueError(
            "X_candidates must have the same number of features as X_train "
            f"({candidate_array.shape[1]} != {surrogate.train_X.shape[-1]})"
        )

    result = _candidate_frame(
        X_candidates=X_candidates,
        candidate_data=candidate_data,
        candidate_index=candidate_index,
    )
    candidate_X = _to_tensor(
        torch,
        candidate_array,
        device=surrogate.train_X.device,
    )
    single_scores, predicted_mean, predicted_std = _multi_objective_scores_and_predictions(
        surrogate,
        candidate_X,
        acquisition_function=acquisition_function,
        mc_samples=mc_samples,
        sampler_seed=sampler_seed,
        x_pending=None,
    )
    objective_columns = _add_multi_objective_prediction_columns(
        result,
        surrogate=surrogate,
        predicted_mean=predicted_mean,
        predicted_std=predicted_std,
    )
    result["matgpr_acquisition"] = single_scores
    result["matgpr_predicted_pareto_front"] = pareto_front_mask(
        result,
        [
            ObjectiveSpec(
                name=surrogate.objective_names[index],
                column=column,
                goal=surrogate.objective_directions[index],
            )
            for index, column in enumerate(objective_columns)
        ],
    ).to_numpy(dtype=bool)

    if constraints is not None:
        result = apply_candidate_constraints(result, constraints)
        if constraint_policy == "filter":
            keep_mask = result["matgpr_feasible"].to_numpy(dtype=bool)
            result = result.loc[keep_mask].copy()
            candidate_array = candidate_array[keep_mask]
            candidate_X = candidate_X[torch.as_tensor(keep_mask, device=candidate_X.device)]

    result, keep_mask = _apply_candidate_domain_policies_with_mask(
        result,
        candidate_array,
        trust_region=trust_region,
        trust_region_policy=trust_region_policy,
        duplicate_policy=duplicate_policy,
        duplicate_policy_action=duplicate_policy_action,
    )
    candidate_X = candidate_X[torch.as_tensor(keep_mask, device=candidate_X.device)]
    result = result.reset_index(drop=True)
    result.insert(0, "matgpr_rank", _rank_descending(result["matgpr_acquisition"]))
    result["matgpr_batch_selected"] = False
    result["matgpr_batch_order"] = pd.Series(pd.NA, index=result.index, dtype="Int64")
    result["matgpr_batch_score"] = np.nan

    if result.empty:
        return result.reset_index(drop=True)

    selected_positions, selected_scores = _greedy_multi_objective_batch_positions(
        surrogate,
        candidate_X,
        top_k=min(top_k, result.shape[0]),
        acquisition_function=acquisition_function,
        mc_samples=mc_samples,
        sampler_seed=sampler_seed,
    )
    for order, (position, score) in enumerate(
        zip(selected_positions, selected_scores),
        start=1,
    ):
        result.loc[position, "matgpr_batch_selected"] = True
        result.loc[position, "matgpr_batch_order"] = order
        result.loc[position, "matgpr_batch_score"] = score

    if return_all:
        return result.sort_values(
            ["matgpr_batch_selected", "matgpr_batch_order", "matgpr_rank"],
            ascending=[False, True, True],
            na_position="last",
        ).reset_index(drop=True)

    return (
        result[result["matgpr_batch_selected"]]
        .sort_values("matgpr_batch_order")
        .reset_index(drop=True)
    )

suggest_next_experiments(X_train, y_train, X_candidates, *, candidate_data=None, top_k=5, acquisition_function='log_expected_improvement', maximize=True, noise_variance=None, normalize_features=True, standardize_target=True, beta=0.2, fit_model=True, device=None, constraints=None, constraint_policy='filter', trust_region=None, trust_region_policy='filter', duplicate_policy=None, duplicate_policy_action='filter', batch_selection='top', batch_feature_columns=None, diversity_weight=0.25, min_batch_distance=None)

Fit a surrogate and recommend the next materials to evaluate.

Use this high-level function when the candidate space is a finite list of materials, compositions, molecules, polymers, or processing conditions that have already been featurized.

Parameters:

Name	Type	Description	Default
X_train	DataFrame | ndarray	Numeric features for materials with measured target values.	required
y_train	Series | ndarray	Measured target values.	required
X_candidates	DataFrame | ndarray	Numeric features for unmeasured candidate materials.	required
candidate_data	DataFrame | None	Optional metadata for the candidates. This is returned with the recommendation table.	None
top_k	int	Number of candidates to recommend.	5
acquisition_function	str	Acquisition function used for ranking. See rank_discrete_candidates.	'log_expected_improvement'
maximize	bool	If True, recommend candidates expected to maximize the target. If False, recommend candidates expected to minimize the target.	True
noise_variance	float | ndarray | Series | None	Optional known target-noise variance for each training row or a scalar variance shared by all rows.	None
normalize_features	bool	If True, use BoTorch's feature normalization transform.	True
standardize_target	bool	If True, use BoTorch's target standardization transform.	True
beta	float	Exploration weight for upper confidence bound.	0.2
fit_model	bool	If True, fit GP hyperparameters before ranking.	True
device	str | None	Optional torch device string.	None
constraints	CandidateConstraint | list[CandidateConstraint] | tuple[CandidateConstraint, ...] | None	Optional finite-pool feasibility constraints.	None
constraint_policy	str	"filter" recommends only feasible candidates. "annotate" ranks all candidates and marks feasibility.	'filter'
trust_region	CandidateTrustRegion | None	Optional descriptor-space trust region.	None
trust_region_policy	str	"filter" recommends only candidates inside the trust region. "annotate" keeps all candidates and marks trust-region membership.	'filter'
duplicate_policy	CandidateDuplicatePolicy | None	Optional duplicate-avoidance policy.	None
duplicate_policy_action	str	"filter" removes duplicate candidates. "annotate" keeps all candidates and marks duplicates.	'filter'
batch_selection	str	"top" returns the highest-acquisition candidates. "diverse" uses greedy diversity-aware batch selection.	'top'
batch_feature_columns	list[str] | tuple[str, ...] | None	Numeric columns in the ranked candidate table used for diversity-aware batch selection.	None
diversity_weight	float	Diversity weight used when batch_selection="diverse".	0.25
min_batch_distance	float | None	Optional minimum descriptor-space distance between selected batch members when batch_selection="diverse".	None

Source code in matgpr/bayesian_optimization.py

def suggest_next_experiments(
    X_train: pd.DataFrame | np.ndarray,
    y_train: pd.Series | np.ndarray,
    X_candidates: pd.DataFrame | np.ndarray,
    *,
    candidate_data: pd.DataFrame | None = None,
    top_k: int = 5,
    acquisition_function: str = "log_expected_improvement",
    maximize: bool = True,
    noise_variance: float | np.ndarray | pd.Series | None = None,
    normalize_features: bool = True,
    standardize_target: bool = True,
    beta: float = 0.2,
    fit_model: bool = True,
    device: str | None = None,
    constraints: CandidateConstraint
    | list[CandidateConstraint]
    | tuple[CandidateConstraint, ...]
    | None = None,
    constraint_policy: str = "filter",
    trust_region: CandidateTrustRegion | None = None,
    trust_region_policy: str = "filter",
    duplicate_policy: CandidateDuplicatePolicy | None = None,
    duplicate_policy_action: str = "filter",
    batch_selection: str = "top",
    batch_feature_columns: list[str] | tuple[str, ...] | None = None,
    diversity_weight: float = 0.25,
    min_batch_distance: float | None = None,
) -> BayesianOptimizationResult:
    """Fit a surrogate and recommend the next materials to evaluate.

    Use this high-level function when the candidate space is a finite list of
    materials, compositions, molecules, polymers, or processing conditions that
    have already been featurized.

    Parameters
    ----------
    X_train
        Numeric features for materials with measured target values.
    y_train
        Measured target values.
    X_candidates
        Numeric features for unmeasured candidate materials.
    candidate_data
        Optional metadata for the candidates. This is returned with the
        recommendation table.
    top_k
        Number of candidates to recommend.
    acquisition_function
        Acquisition function used for ranking. See
        `rank_discrete_candidates`.
    maximize
        If `True`, recommend candidates expected to maximize the target. If
        `False`, recommend candidates expected to minimize the target.
    noise_variance
        Optional known target-noise variance for each training row or a scalar
        variance shared by all rows.
    normalize_features
        If `True`, use BoTorch's feature normalization transform.
    standardize_target
        If `True`, use BoTorch's target standardization transform.
    beta
        Exploration weight for upper confidence bound.
    fit_model
        If `True`, fit GP hyperparameters before ranking.
    device
        Optional torch device string.
    constraints
        Optional finite-pool feasibility constraints.
    constraint_policy
        `"filter"` recommends only feasible candidates. `"annotate"` ranks all
        candidates and marks feasibility.
    trust_region
        Optional descriptor-space trust region.
    trust_region_policy
        `"filter"` recommends only candidates inside the trust region.
        `"annotate"` keeps all candidates and marks trust-region membership.
    duplicate_policy
        Optional duplicate-avoidance policy.
    duplicate_policy_action
        `"filter"` removes duplicate candidates. `"annotate"` keeps all
        candidates and marks duplicates.
    batch_selection
        `"top"` returns the highest-acquisition candidates. `"diverse"` uses
        greedy diversity-aware batch selection.
    batch_feature_columns
        Numeric columns in the ranked candidate table used for diversity-aware
        batch selection.
    diversity_weight
        Diversity weight used when `batch_selection="diverse"`.
    min_batch_distance
        Optional minimum descriptor-space distance between selected batch
        members when `batch_selection="diverse"`.
    """
    if top_k < 1:
        raise ValueError("top_k must be at least 1")
    constraint_policy = _validate_constraint_policy(constraint_policy)
    batch_selection = _validate_batch_selection(batch_selection)

    surrogate = fit_botorch_surrogate(
        X_train,
        y_train,
        maximize=maximize,
        noise_variance=noise_variance,
        normalize_features=normalize_features,
        standardize_target=standardize_target,
        fit_model=fit_model,
        device=device,
    )
    ranked_candidates = rank_discrete_candidates(
        surrogate,
        X_candidates,
        acquisition_function=acquisition_function,
        top_k=None,
        beta=beta,
        candidate_data=candidate_data,
        constraints=constraints,
        constraint_policy=constraint_policy,
        trust_region=trust_region,
        trust_region_policy=trust_region_policy,
        duplicate_policy=duplicate_policy,
        duplicate_policy_action=duplicate_policy_action,
    )
    if batch_selection == "diverse":
        recommendations = select_diverse_batch(
            ranked_candidates,
            top_k=top_k,
            feature_columns=batch_feature_columns,
            diversity_weight=diversity_weight,
            min_distance=min_batch_distance,
            return_all=False,
        )
    else:
        recommendations = ranked_candidates.head(top_k).reset_index(drop=True)

    return BayesianOptimizationResult(
        recommendations=recommendations,
        ranked_candidates=ranked_candidates,
        surrogate=surrogate,
        acquisition_function=acquisition_function,
        maximize=maximize,
        top_k=top_k,
    )

suggest_multi_objective_next_experiments(X_train, y_train, X_candidates, *, objective_directions='maximize', objective_names=None, reference_point=None, candidate_data=None, top_k=5, acquisition_function='q_log_expected_hypervolume_improvement', noise_variance=None, normalize_features=True, standardize_targets=True, fit_model=True, device=None, constraints=None, constraint_policy='filter', trust_region=None, trust_region_policy='filter', duplicate_policy=None, duplicate_policy_action='filter', batch_selection='top', batch_feature_columns=None, diversity_weight=0.25, min_batch_distance=None, mc_samples=128, sampler_seed=42)

Fit multi-objective surrogates and recommend finite-pool candidates.

Use this when a next experiment must trade off two or more measured objectives, such as maximizing conductivity while minimizing degradation rate, toxicity, synthesis cost, or processing burden.

Source code in matgpr/bayesian_optimization.py

def suggest_multi_objective_next_experiments(
    X_train: pd.DataFrame | np.ndarray,
    y_train: pd.DataFrame | np.ndarray,
    X_candidates: pd.DataFrame | np.ndarray,
    *,
    objective_directions: str | Sequence[str] = "maximize",
    objective_names: Sequence[str] | None = None,
    reference_point: Sequence[float] | np.ndarray | pd.Series | None = None,
    candidate_data: pd.DataFrame | None = None,
    top_k: int = 5,
    acquisition_function: str = "q_log_expected_hypervolume_improvement",
    noise_variance: float | np.ndarray | pd.DataFrame | None = None,
    normalize_features: bool = True,
    standardize_targets: bool = True,
    fit_model: bool = True,
    device: str | None = None,
    constraints: CandidateConstraint
    | list[CandidateConstraint]
    | tuple[CandidateConstraint, ...]
    | None = None,
    constraint_policy: str = "filter",
    trust_region: CandidateTrustRegion | None = None,
    trust_region_policy: str = "filter",
    duplicate_policy: CandidateDuplicatePolicy | None = None,
    duplicate_policy_action: str = "filter",
    batch_selection: str = "top",
    batch_feature_columns: list[str] | tuple[str, ...] | None = None,
    diversity_weight: float = 0.25,
    min_batch_distance: float | None = None,
    mc_samples: int = 128,
    sampler_seed: int | None = 42,
) -> MultiObjectiveBayesianOptimizationResult:
    """Fit multi-objective surrogates and recommend finite-pool candidates.

    Use this when a next experiment must trade off two or more measured
    objectives, such as maximizing conductivity while minimizing degradation
    rate, toxicity, synthesis cost, or processing burden.
    """
    if top_k < 1:
        raise ValueError("top_k must be at least 1")
    constraint_policy = _validate_constraint_policy(constraint_policy)
    batch_selection = _validate_batch_selection(batch_selection, allow_sequential=True)

    surrogate = fit_multi_objective_botorch_surrogate(
        X_train,
        y_train,
        objective_directions=objective_directions,
        objective_names=objective_names,
        reference_point=reference_point,
        noise_variance=noise_variance,
        normalize_features=normalize_features,
        standardize_targets=standardize_targets,
        fit_model=fit_model,
        device=device,
    )
    ranked_candidates = rank_multi_objective_discrete_candidates(
        surrogate,
        X_candidates,
        acquisition_function=acquisition_function,
        top_k=None,
        candidate_data=candidate_data,
        constraints=constraints,
        constraint_policy=constraint_policy,
        trust_region=trust_region,
        trust_region_policy=trust_region_policy,
        duplicate_policy=duplicate_policy,
        duplicate_policy_action=duplicate_policy_action,
        mc_samples=mc_samples,
        sampler_seed=sampler_seed,
    )
    if batch_selection == "sequential":
        recommendations = select_sequential_multi_objective_batch(
            surrogate,
            X_candidates,
            top_k=top_k,
            acquisition_function=acquisition_function,
            candidate_data=candidate_data,
            constraints=constraints,
            constraint_policy=constraint_policy,
            trust_region=trust_region,
            trust_region_policy=trust_region_policy,
            duplicate_policy=duplicate_policy,
            duplicate_policy_action=duplicate_policy_action,
            mc_samples=mc_samples,
            sampler_seed=sampler_seed,
            return_all=False,
        )
    elif batch_selection == "diverse":
        recommendations = select_diverse_batch(
            ranked_candidates,
            top_k=top_k,
            feature_columns=batch_feature_columns,
            diversity_weight=diversity_weight,
            min_distance=min_batch_distance,
            return_all=False,
        )
    else:
        recommendations = ranked_candidates.head(top_k).reset_index(drop=True)

    return MultiObjectiveBayesianOptimizationResult(
        recommendations=recommendations,
        ranked_candidates=ranked_candidates,
        surrogate=surrogate,
        acquisition_function=acquisition_function,
        objective_names=surrogate.objective_names,
        objective_directions=surrogate.objective_directions,
        top_k=top_k,
    )