BO Benchmarking

The BO benchmarking API provides lightweight offline simulations for finite candidate pools with known outcomes. Use it to compare acquisition strategies, physics-prior scores, model predictions, or random baselines on historical materials datasets before running an expensive closed-loop campaign.

matgpr.bo_benchmarking

BOBenchmarkStrategy dataclass

Finite-pool strategy definition for offline BO benchmarking.

Parameters:

Name	Type	Description	Default
name	str	Human-readable strategy label.	required
score_column	str | None	Optional dataframe column containing precomputed utility scores.	None
score_function	Callable[[DataFrame, DataFrame, Generator], Any] | None	Optional callable that receives (observed, candidates, rng) and returns one score per remaining candidate. This supports lightweight custom policies without requiring BoTorch inside the benchmark loop.	None
direction	str	Whether larger or smaller strategy scores should be selected first. Use "maximize" for acquisition functions and "minimize" for costs or uncertainty/error scores where lower values are preferred.	'maximize'

Notes

When neither score_column nor score_function is provided, the strategy samples candidates uniformly at random.

Source code in matgpr/bo_benchmarking.py

@dataclass(frozen=True)
class BOBenchmarkStrategy:
    """Finite-pool strategy definition for offline BO benchmarking.

    Parameters
    ----------
    name
        Human-readable strategy label.
    score_column
        Optional dataframe column containing precomputed utility scores.
    score_function
        Optional callable that receives `(observed, candidates, rng)` and
        returns one score per remaining candidate. This supports lightweight
        custom policies without requiring BoTorch inside the benchmark loop.
    direction
        Whether larger or smaller strategy scores should be selected first.
        Use `"maximize"` for acquisition functions and `"minimize"` for costs
        or uncertainty/error scores where lower values are preferred.

    Notes
    -----
    When neither `score_column` nor `score_function` is provided, the strategy
    samples candidates uniformly at random.
    """

    name: str
    score_column: str | None = None
    score_function: Callable[[pd.DataFrame, pd.DataFrame, np.random.Generator], Any] | None = None
    direction: str = "maximize"

    def __post_init__(self) -> None:
        name = str(self.name).strip()
        if not name:
            raise ValueError("BOBenchmarkStrategy.name must be non-empty")
        object.__setattr__(self, "name", name)

        if self.score_column is not None:
            score_column = str(self.score_column).strip()
            if not score_column:
                raise ValueError("score_column must be non-empty when provided")
            object.__setattr__(self, "score_column", score_column)

        if self.score_column is not None and self.score_function is not None:
            raise ValueError("Provide only one of score_column or score_function")

        object.__setattr__(self, "direction", _normalize_direction(self.direction))

BOBenchmarkResult dataclass

Result from one finite-pool BO replay.

Source code in matgpr/bo_benchmarking.py

@dataclass(frozen=True)
class BOBenchmarkResult:
    """Result from one finite-pool BO replay."""

    strategy_name: str
    target_column: str
    maximize: bool
    history: pd.DataFrame
    summary: pd.DataFrame

    def summary_frame(self) -> pd.DataFrame:
        """Return the one-row benchmark summary."""
        return self.summary.copy()

summary_frame()

Return the one-row benchmark summary.

Source code in matgpr/bo_benchmarking.py

def summary_frame(self) -> pd.DataFrame:
    """Return the one-row benchmark summary."""
    return self.summary.copy()

BOBenchmarkComparison dataclass

Result from comparing several finite-pool BO strategies.

Source code in matgpr/bo_benchmarking.py

@dataclass(frozen=True)
class BOBenchmarkComparison:
    """Result from comparing several finite-pool BO strategies."""

    history: pd.DataFrame
    summary: pd.DataFrame
    target_column: str
    maximize: bool

    def summary_by_strategy(self) -> pd.DataFrame:
        """Aggregate benchmark outcomes by strategy."""
        return summarize_bo_benchmark(self.summary)

summary_by_strategy()

Aggregate benchmark outcomes by strategy.

Source code in matgpr/bo_benchmarking.py

def summary_by_strategy(self) -> pd.DataFrame:
    """Aggregate benchmark outcomes by strategy."""
    return summarize_bo_benchmark(self.summary)

simulate_bo_strategy(data, strategy, *, target_column, maximize=True, budget=None, batch_size=1, initial_observed=None, candidate_id_column=None, random_state=None)

Replay one finite-pool strategy against known candidate outcomes.

This is useful for historical or synthetic benchmarks where every candidate has a known target value. The simulator selects candidates from the finite pool, reveals their known outcomes, and records the best value found as a function of experimental budget.

Source code in matgpr/bo_benchmarking.py

def simulate_bo_strategy(
    data: pd.DataFrame,
    strategy: BOBenchmarkStrategy,
    *,
    target_column: str,
    maximize: bool = True,
    budget: int | None = None,
    batch_size: int = 1,
    initial_observed: Sequence[Any] | None = None,
    candidate_id_column: str | None = None,
    random_state: int | None = None,
) -> BOBenchmarkResult:
    """Replay one finite-pool strategy against known candidate outcomes.

    This is useful for historical or synthetic benchmarks where every
    candidate has a known target value. The simulator selects candidates from
    the finite pool, reveals their known outcomes, and records the best value
    found as a function of experimental budget.
    """
    frame = _validate_benchmark_data(
        data,
        target_column=target_column,
        candidate_id_column=candidate_id_column,
    )
    if not isinstance(strategy, BOBenchmarkStrategy):
        raise TypeError("strategy must be a BOBenchmarkStrategy")
    batch_size = validate_positive_int(batch_size, name="batch_size")

    candidate_ids = _candidate_ids(frame, candidate_id_column)
    target_values = _numeric_column(frame, target_column)
    initial_observed_values = None if initial_observed is None else list(initial_observed)
    observed_mask = _initial_observed_mask(
        frame,
        candidate_ids,
        initial_observed=initial_observed_values,
        candidate_id_column=candidate_id_column,
    )
    initial_mask = observed_mask.copy()
    remaining_count = int((~observed_mask).sum())
    budget = remaining_count if budget is None else validate_positive_int(budget, name="budget")
    budget = min(budget, remaining_count)

    optimum_position = _best_position(target_values, maximize=maximize)
    optimum_value = float(target_values[optimum_position])
    optimum_candidate_id = candidate_ids[optimum_position]

    rng = np.random.default_rng(random_state)
    history_rows: list[dict[str, Any]] = []
    selected_count = 0
    iteration = 0

    while selected_count < budget and np.any(~observed_mask):
        iteration += 1
        batch_limit = min(batch_size, budget - selected_count, int((~observed_mask).sum()))
        selected_positions, selected_scores = _select_strategy_batch(
            frame,
            strategy,
            observed_mask=observed_mask,
            batch_size=batch_limit,
            rng=rng,
        )
        best_before = _best_observed_value(target_values, observed_mask, maximize=maximize)

        for batch_order, (position, score) in enumerate(
            zip(selected_positions, selected_scores),
            start=1,
        ):
            selected_count += 1
            observed_mask[position] = True
            best_so_far = _best_observed_value(target_values, observed_mask, maximize=maximize)
            regret = _simple_regret(
                best_so_far,
                optimum_value=optimum_value,
                maximize=maximize,
            )
            history_rows.append(
                {
                    "matgpr_strategy": strategy.name,
                    "matgpr_iteration": iteration,
                    "matgpr_batch_order": batch_order,
                    "matgpr_evaluation": selected_count,
                    "matgpr_candidate_index": frame.index[position],
                    "matgpr_candidate_id": candidate_ids[position],
                    "matgpr_strategy_score": float(score),
                    "matgpr_target_value": float(target_values[position]),
                    "matgpr_best_before_batch": best_before,
                    "matgpr_best_so_far": best_so_far,
                    "matgpr_simple_regret": regret,
                    "matgpr_is_optimum": bool(position == optimum_position),
                }
            )

    history = pd.DataFrame(history_rows, columns=list(_HISTORY_COLUMNS))
    final_best = _best_observed_value(target_values, observed_mask, maximize=maximize)
    final_regret = _simple_regret(
        final_best,
        optimum_value=optimum_value,
        maximize=maximize,
    )
    evaluations_to_optimum = (
        0 if initial_mask[optimum_position] else _evaluations_to_optimum(history)
    )
    hit_optimum = not pd.isna(evaluations_to_optimum)

    summary = pd.DataFrame(
        [
            {
                "matgpr_strategy": strategy.name,
                "matgpr_target_column": target_column,
                "matgpr_goal": "maximize" if maximize else "minimize",
                "matgpr_n_candidates": frame.shape[0],
                "matgpr_n_initial": int(_initial_count(initial_observed_values)),
                "matgpr_budget": budget,
                "matgpr_batch_size": batch_size,
                "matgpr_n_selected": selected_count,
                "matgpr_n_iterations": iteration,
                "matgpr_optimum_value": optimum_value,
                "matgpr_optimum_candidate_id": optimum_candidate_id,
                "matgpr_final_best": final_best,
                "matgpr_final_regret": final_regret,
                "matgpr_hit_optimum": bool(hit_optimum),
                "matgpr_evaluations_to_optimum": evaluations_to_optimum,
            }
        ]
    )
    return BOBenchmarkResult(
        strategy_name=strategy.name,
        target_column=target_column,
        maximize=maximize,
        history=history,
        summary=summary,
    )

compare_bo_strategies(data, strategies, *, target_column, maximize=True, budget=None, batch_size=1, initial_observed=None, candidate_id_column=None, n_repeats=1, random_state=None)

Compare several finite-pool BO strategies over repeated replays.

Source code in matgpr/bo_benchmarking.py

def compare_bo_strategies(
    data: pd.DataFrame,
    strategies: BOBenchmarkStrategy | Sequence[BOBenchmarkStrategy],
    *,
    target_column: str,
    maximize: bool = True,
    budget: int | None = None,
    batch_size: int = 1,
    initial_observed: Sequence[Any] | None = None,
    candidate_id_column: str | None = None,
    n_repeats: int = 1,
    random_state: int | None = None,
) -> BOBenchmarkComparison:
    """Compare several finite-pool BO strategies over repeated replays."""
    strategy_tuple = _as_strategy_tuple(strategies)
    n_repeats = validate_positive_int(n_repeats, name="n_repeats")
    root_rng = np.random.default_rng(random_state)
    history_tables = []
    summary_tables = []

    for repeat in range(n_repeats):
        for strategy in strategy_tuple:
            seed = int(root_rng.integers(0, np.iinfo(np.uint32).max))
            result = simulate_bo_strategy(
                data,
                strategy,
                target_column=target_column,
                maximize=maximize,
                budget=budget,
                batch_size=batch_size,
                initial_observed=initial_observed,
                candidate_id_column=candidate_id_column,
                random_state=seed,
            )
            history = result.history.copy()
            summary = result.summary.copy()
            history.insert(1, "matgpr_repeat", repeat)
            summary.insert(1, "matgpr_repeat", repeat)
            history_tables.append(history)
            summary_tables.append(summary)

    history = pd.concat(history_tables, ignore_index=True, sort=False)
    summary = pd.concat(summary_tables, ignore_index=True, sort=False)
    return BOBenchmarkComparison(
        history=history,
        summary=summary,
        target_column=target_column,
        maximize=maximize,
    )

summarize_bo_benchmark(summary)

Aggregate repeated BO benchmark summaries by strategy.

Source code in matgpr/bo_benchmarking.py

def summarize_bo_benchmark(summary: pd.DataFrame) -> pd.DataFrame:
    """Aggregate repeated BO benchmark summaries by strategy."""
    if not isinstance(summary, pd.DataFrame):
        raise TypeError("summary must be a pandas DataFrame")
    required = {
        "matgpr_strategy",
        "matgpr_final_best",
        "matgpr_final_regret",
        "matgpr_hit_optimum",
        "matgpr_evaluations_to_optimum",
    }
    missing = sorted(required.difference(summary.columns))
    if missing:
        raise ValueError(f"summary is missing benchmark columns: {missing}")

    working = summary.copy()
    working["matgpr_hit_optimum"] = working["matgpr_hit_optimum"].astype(bool)
    working["matgpr_evaluations_to_optimum"] = pd.to_numeric(
        working["matgpr_evaluations_to_optimum"],
        errors="coerce",
    )
    grouped = working.groupby("matgpr_strategy", dropna=False)
    return (
        grouped.agg(
            matgpr_n_runs=("matgpr_strategy", "size"),
            matgpr_final_best_mean=("matgpr_final_best", "mean"),
            matgpr_final_best_std=("matgpr_final_best", "std"),
            matgpr_final_regret_mean=("matgpr_final_regret", "mean"),
            matgpr_final_regret_std=("matgpr_final_regret", "std"),
            matgpr_hit_optimum_rate=("matgpr_hit_optimum", "mean"),
            matgpr_evaluations_to_optimum_mean=(
                "matgpr_evaluations_to_optimum",
                "mean",
            ),
        )
        .reset_index()
        .sort_values(
            ["matgpr_final_regret_mean", "matgpr_hit_optimum_rate"],
            ascending=[True, False],
        )
        .reset_index(drop=True)
    )