Candidate Generation

Candidate-generation helpers build and audit finite materials, chemistry, formulation, or processing-condition pools before Bayesian optimization. Use these functions to create realistic candidate tables, summarize descriptor completeness and duplicate keys, check feature/category coverage against measured data, remove rows that have already been measured, and split candidate metadata from numeric model features.

matgpr.candidate_generation

CandidatePoolDiagnostics dataclass

Diagnostic tables for a finite candidate pool.

Attributes:

Name	Type	Description
overview	DataFrame	One-row pool-level summary with candidate counts, missing-feature counts, and duplicate-key counts.
numeric_features	DataFrame	Per-feature numeric descriptor summary.
categorical_features	DataFrame	Per-column categorical metadata summary.
duplicate_keys	DataFrame	Duplicate key groups, empty when no duplicate groups are found or no key columns were supplied.

Source code in matgpr/candidate_generation.py

@dataclass(frozen=True)
class CandidatePoolDiagnostics:
    """Diagnostic tables for a finite candidate pool.

    Attributes
    ----------
    overview
        One-row pool-level summary with candidate counts, missing-feature
        counts, and duplicate-key counts.
    numeric_features
        Per-feature numeric descriptor summary.
    categorical_features
        Per-column categorical metadata summary.
    duplicate_keys
        Duplicate key groups, empty when no duplicate groups are found or no
        key columns were supplied.
    """

    overview: pd.DataFrame
    numeric_features: pd.DataFrame
    categorical_features: pd.DataFrame
    duplicate_keys: pd.DataFrame

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

    def numeric_feature_frame(self) -> pd.DataFrame:
        """Return a copy of the numeric feature summary table."""
        return self.numeric_features.copy()

    def categorical_feature_frame(self) -> pd.DataFrame:
        """Return a copy of the categorical metadata summary table."""
        return self.categorical_features.copy()

    def duplicate_key_frame(self) -> pd.DataFrame:
        """Return a copy of the duplicate-key diagnostics table."""
        return self.duplicate_keys.copy()

overview_frame()

Return a copy of the pool-level overview table.

Source code in matgpr/candidate_generation.py

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

numeric_feature_frame()

Return a copy of the numeric feature summary table.

Source code in matgpr/candidate_generation.py

def numeric_feature_frame(self) -> pd.DataFrame:
    """Return a copy of the numeric feature summary table."""
    return self.numeric_features.copy()

categorical_feature_frame()

Return a copy of the categorical metadata summary table.

Source code in matgpr/candidate_generation.py

def categorical_feature_frame(self) -> pd.DataFrame:
    """Return a copy of the categorical metadata summary table."""
    return self.categorical_features.copy()

duplicate_key_frame()

Return a copy of the duplicate-key diagnostics table.

Source code in matgpr/candidate_generation.py

def duplicate_key_frame(self) -> pd.DataFrame:
    """Return a copy of the duplicate-key diagnostics table."""
    return self.duplicate_keys.copy()

build_cartesian_candidate_grid(parameter_space, *, fixed_values=None, candidate_id_column='candidate_id', candidate_id_prefix='candidate', max_candidates=100000)

Build a finite candidate table from a Cartesian product of choices.

Use this for processing conditions, formulation variables, solvents, categorical choices, or any finite design space that should be ranked by Bayesian optimization.

Parameters:

Name	Type	Description	Default
parameter_space	Mapping[str, Iterable[Any] | Any]	Mapping from output column name to candidate values. Scalars and strings are treated as one-value choices.	required
fixed_values	Mapping[str, Any] | None	Optional columns with constant metadata added to every candidate row.	None
candidate_id_column	str | None	Optional candidate identifier column. Set to None to omit IDs.	'candidate_id'
candidate_id_prefix	str	Prefix used when candidate_id_column is not None.	'candidate'
max_candidates	int | None	Optional safety limit. A ValueError is raised before returning very large grids.	100000

Source code in matgpr/candidate_generation.py

def build_cartesian_candidate_grid(
    parameter_space: Mapping[str, Iterable[Any] | Any],
    *,
    fixed_values: Mapping[str, Any] | None = None,
    candidate_id_column: str | None = "candidate_id",
    candidate_id_prefix: str = "candidate",
    max_candidates: int | None = 100_000,
) -> pd.DataFrame:
    """Build a finite candidate table from a Cartesian product of choices.

    Use this for processing conditions, formulation variables, solvents,
    categorical choices, or any finite design space that should be ranked by
    Bayesian optimization.

    Parameters
    ----------
    parameter_space
        Mapping from output column name to candidate values. Scalars and strings
        are treated as one-value choices.
    fixed_values
        Optional columns with constant metadata added to every candidate row.
    candidate_id_column
        Optional candidate identifier column. Set to `None` to omit IDs.
    candidate_id_prefix
        Prefix used when `candidate_id_column` is not `None`.
    max_candidates
        Optional safety limit. A `ValueError` is raised before returning very
        large grids.
    """
    columns, levels = _validate_parameter_space(parameter_space)
    fixed = _validate_fixed_values(fixed_values)
    conflicts = sorted(set(columns).intersection(fixed))
    if conflicts:
        raise ValueError(f"fixed_values conflict with parameter columns: {conflicts}")
    candidate_id_column = _normalize_candidate_id_column(candidate_id_column)
    _validate_candidate_id_column(candidate_id_column, columns + tuple(fixed))
    _validate_candidate_id_prefix(candidate_id_prefix)
    _validate_max_candidates(max_candidates)

    candidate_count = int(np.prod([len(level) for level in levels], dtype=object))
    if max_candidates is not None and candidate_count > max_candidates:
        raise ValueError(
            "Cartesian candidate grid would contain "
            f"{candidate_count} rows, above max_candidates={max_candidates}"
        )

    rows = [dict(zip(columns, values)) for values in product(*levels)]
    candidates = pd.DataFrame(rows, columns=list(columns))
    for column, value in fixed.items():
        candidates[column] = value

    _insert_candidate_ids(candidates, candidate_id_column, prefix=candidate_id_prefix)
    return candidates.reset_index(drop=True)

build_composition_candidate_grid(elements, *, step=0.25, min_fraction=0.0, max_fraction=1.0, min_components=1, max_components=None, formula_column='formula', fraction_prefix='frac_', candidate_id_column='candidate_id', candidate_id_prefix='composition', max_candidates=100000)

Build inorganic composition candidates on an element-fraction simplex.

The grid uses integer stoichiometric units implied by step. For example, step=0.25 creates compositions with fractions in increments of one quarter and formulas such as "Al3Ni" or "AlNi".

Parameters:

Name	Type	Description	Default
elements	Sequence[str]	Element symbols defining the composition space.	required
step	float	Fractional spacing. The value must divide 1.0, such as 0.5, 0.25, 0.2, or 0.1.	0.25
min_fraction	float	Bounds applied to each nonzero element fraction.	0.0
max_fraction	float	Bounds applied to each nonzero element fraction.	0.0
min_components	int	Bounds on the number of elements present in a candidate.	1
max_components	int	Bounds on the number of elements present in a candidate.	1
formula_column	str	Name of the generated reduced-formula column.	'formula'
fraction_prefix	str	Prefix for element-fraction columns.	'frac_'
candidate_id_column	str | None	Optional candidate identifier column. Set to None to omit IDs.	'candidate_id'
candidate_id_prefix	str	Prefix used when candidate_id_column is not None.	'composition'
max_candidates	int | None	Optional safety limit for generated rows.	100000

Source code in matgpr/candidate_generation.py

def build_composition_candidate_grid(
    elements: Sequence[str],
    *,
    step: float = 0.25,
    min_fraction: float = 0.0,
    max_fraction: float = 1.0,
    min_components: int = 1,
    max_components: int | None = None,
    formula_column: str = "formula",
    fraction_prefix: str = "frac_",
    candidate_id_column: str | None = "candidate_id",
    candidate_id_prefix: str = "composition",
    max_candidates: int | None = 100_000,
) -> pd.DataFrame:
    """Build inorganic composition candidates on an element-fraction simplex.

    The grid uses integer stoichiometric units implied by `step`. For example,
    `step=0.25` creates compositions with fractions in increments of one
    quarter and formulas such as `"Al3Ni"` or `"AlNi"`.

    Parameters
    ----------
    elements
        Element symbols defining the composition space.
    step
        Fractional spacing. The value must divide 1.0, such as `0.5`, `0.25`,
        `0.2`, or `0.1`.
    min_fraction, max_fraction
        Bounds applied to each nonzero element fraction.
    min_components, max_components
        Bounds on the number of elements present in a candidate.
    formula_column
        Name of the generated reduced-formula column.
    fraction_prefix
        Prefix for element-fraction columns.
    candidate_id_column
        Optional candidate identifier column. Set to `None` to omit IDs.
    candidate_id_prefix
        Prefix used when `candidate_id_column` is not `None`.
    max_candidates
        Optional safety limit for generated rows.
    """
    symbols = _validate_element_symbols(elements)
    total_units = _fraction_step_units(step)
    _validate_fraction_bounds(min_fraction, max_fraction)
    min_components, max_components = _validate_component_bounds(
        min_components,
        max_components,
        n_elements=len(symbols),
    )
    if not str(formula_column).strip():
        raise ValueError("formula_column must be non-empty")
    if not str(fraction_prefix).strip():
        raise ValueError("fraction_prefix must be non-empty")
    formula_column = str(formula_column).strip()
    fraction_prefix = str(fraction_prefix).strip()

    fraction_columns = tuple(f"{fraction_prefix}{symbol}" for symbol in symbols)
    reserved_columns = (formula_column, "n_components") + fraction_columns
    candidate_id_column = _normalize_candidate_id_column(candidate_id_column)
    _validate_candidate_id_column(candidate_id_column, reserved_columns)
    _validate_candidate_id_prefix(candidate_id_prefix)
    _validate_max_candidates(max_candidates)

    rows: list[dict[str, Any]] = []
    for counts in _integer_simplex_points(len(symbols), total_units):
        fractions = np.asarray(counts, dtype=float) / float(total_units)
        n_components = int(np.count_nonzero(counts))
        if n_components < min_components or n_components > max_components:
            continue
        nonzero = fractions[fractions > 0.0]
        if np.any(nonzero < min_fraction) or np.any(nonzero > max_fraction):
            continue

        row: dict[str, Any] = {
            formula_column: _reduced_formula(symbols, counts),
            "n_components": n_components,
        }
        row.update({column: fraction for column, fraction in zip(fraction_columns, fractions)})
        rows.append(row)
        if max_candidates is not None and len(rows) > max_candidates:
            raise ValueError(f"Composition candidate grid exceeded max_candidates={max_candidates}")

    candidates = pd.DataFrame(rows, columns=[formula_column, "n_components", *fraction_columns])
    _insert_candidate_ids(candidates, candidate_id_column, prefix=candidate_id_prefix)
    return candidates.reset_index(drop=True)

exclude_existing_candidates(candidates, observed, key_columns, *, keep_indicator=False, indicator_column='matgpr_is_observed')

Remove or annotate candidates that already exist in observed data.

Parameters:

Name	Type	Description	Default
candidates	DataFrame	Candidate table to filter.	required
observed	DataFrame	Existing measured, simulated, or already-selected rows.	required
key_columns	Sequence[str]	Columns used to identify duplicates, such as ("formula",) or ("formula", "temperature_c").	required
keep_indicator	bool	If True, keep all rows and add indicator_column. If False, return only candidates not present in observed.	False
indicator_column	str	Name of the boolean indicator column used when keep_indicator=True.	'matgpr_is_observed'

Source code in matgpr/candidate_generation.py

def exclude_existing_candidates(
    candidates: pd.DataFrame,
    observed: pd.DataFrame,
    key_columns: Sequence[str],
    *,
    keep_indicator: bool = False,
    indicator_column: str = "matgpr_is_observed",
) -> pd.DataFrame:
    """Remove or annotate candidates that already exist in observed data.

    Parameters
    ----------
    candidates
        Candidate table to filter.
    observed
        Existing measured, simulated, or already-selected rows.
    key_columns
        Columns used to identify duplicates, such as `("formula",)` or
        `("formula", "temperature_c")`.
    keep_indicator
        If `True`, keep all rows and add `indicator_column`. If `False`, return
        only candidates not present in `observed`.
    indicator_column
        Name of the boolean indicator column used when `keep_indicator=True`.
    """
    if not isinstance(candidates, pd.DataFrame):
        raise TypeError("candidates must be a pandas DataFrame")
    if not isinstance(observed, pd.DataFrame):
        raise TypeError("observed must be a pandas DataFrame")
    keys = _validate_columns(key_columns, candidates, name="key_columns")
    missing_observed = [column for column in keys if column not in observed.columns]
    if missing_observed:
        raise ValueError(f"observed is missing key columns: {missing_observed}")
    if not str(indicator_column).strip():
        raise ValueError("indicator_column must be non-empty")

    observed_keys = set(_row_keys(observed, keys))
    is_observed = pd.Series(
        [key in observed_keys for key in _row_keys(candidates, keys)],
        index=candidates.index,
        dtype=bool,
    )

    result = candidates.copy()
    if keep_indicator:
        result[indicator_column] = is_observed.to_numpy(dtype=bool)
        return result.reset_index(drop=True)
    return result.loc[~is_observed].reset_index(drop=True)

split_candidate_features(candidates, feature_columns, *, metadata_columns=None)

Split a candidate table into numeric features and candidate metadata.

The returned (X_candidates, candidate_data) tuple can be passed directly to suggest_next_experiments.

Source code in matgpr/candidate_generation.py

def split_candidate_features(
    candidates: pd.DataFrame,
    feature_columns: Sequence[str],
    *,
    metadata_columns: Sequence[str] | None = None,
) -> tuple[pd.DataFrame, pd.DataFrame]:
    """Split a candidate table into numeric features and candidate metadata.

    The returned `(X_candidates, candidate_data)` tuple can be passed directly
    to `suggest_next_experiments`.
    """
    if not isinstance(candidates, pd.DataFrame):
        raise TypeError("candidates must be a pandas DataFrame")
    features = _validate_columns(feature_columns, candidates, name="feature_columns")
    non_numeric = [
        column for column in features if not pd.api.types.is_numeric_dtype(candidates[column])
    ]
    if non_numeric:
        raise ValueError(f"feature_columns must be numeric: {non_numeric}")

    if metadata_columns is None:
        metadata = tuple(column for column in candidates.columns if column not in features)
    else:
        metadata = _validate_columns(metadata_columns, candidates, name="metadata_columns")

    X_candidates = candidates.loc[:, list(features)].copy()
    candidate_data = candidates.loc[:, list(metadata)].copy()
    return X_candidates.reset_index(drop=True), candidate_data.reset_index(drop=True)

summarize_candidate_pool(candidates, *, feature_columns=None, categorical_columns=None, key_columns=None)

Summarize a finite candidate pool before Bayesian optimization.

This helper is intended for search-space audits: descriptor completeness, numeric feature ranges, categorical metadata diversity, and duplicate candidate keys. It does not rank candidates and does not require BoTorch.

Source code in matgpr/candidate_generation.py

def summarize_candidate_pool(
    candidates: pd.DataFrame,
    *,
    feature_columns: Sequence[str] | None = None,
    categorical_columns: Sequence[str] | None = None,
    key_columns: Sequence[str] | None = None,
) -> CandidatePoolDiagnostics:
    """Summarize a finite candidate pool before Bayesian optimization.

    This helper is intended for search-space audits: descriptor completeness,
    numeric feature ranges, categorical metadata diversity, and duplicate
    candidate keys. It does not rank candidates and does not require BoTorch.
    """
    frame = _validate_candidate_frame(candidates, name="candidates")
    features = _resolve_numeric_feature_columns(frame, feature_columns)
    categorical = _resolve_categorical_columns(
        frame,
        categorical_columns,
        exclude_columns=features,
    )
    duplicates = (
        summarize_candidate_duplicates(frame, key_columns)
        if key_columns is not None
        else _empty_duplicate_summary()
    )
    numeric_summary = _summarize_numeric_candidate_features(frame, features)
    categorical_summary = _summarize_categorical_candidate_columns(frame, categorical)
    overview = _summarize_candidate_pool_overview(
        frame,
        feature_columns=features,
        categorical_columns=categorical,
        duplicate_keys=duplicates,
    )
    return CandidatePoolDiagnostics(
        overview=overview,
        numeric_features=numeric_summary,
        categorical_features=categorical_summary,
        duplicate_keys=duplicates,
    )

summarize_candidate_feature_coverage(candidates, reference_data, feature_columns, *, tolerance=0.0)

Compare numeric candidate-feature ranges with reference/training data.

The returned table reports whether the finite candidate pool covers the reference range for each feature, and how much of the candidate pool lies outside the reference feature range. Use this before BO to identify extrapolative candidate pools or features with poor search-space coverage.

Source code in matgpr/candidate_generation.py

def summarize_candidate_feature_coverage(
    candidates: pd.DataFrame,
    reference_data: pd.DataFrame,
    feature_columns: Sequence[str],
    *,
    tolerance: float = 0.0,
) -> pd.DataFrame:
    """Compare numeric candidate-feature ranges with reference/training data.

    The returned table reports whether the finite candidate pool covers the
    reference range for each feature, and how much of the candidate pool lies
    outside the reference feature range. Use this before BO to identify
    extrapolative candidate pools or features with poor search-space coverage.
    """
    candidate_frame = _validate_candidate_frame(candidates, name="candidates")
    reference_frame = _validate_candidate_frame(reference_data, name="reference_data")
    features = _validate_columns(feature_columns, candidate_frame, name="feature_columns")
    missing_reference = [column for column in features if column not in reference_frame.columns]
    if missing_reference:
        raise ValueError(f"reference_data is missing feature columns: {missing_reference}")
    _validate_nonnegative_finite(tolerance, name="tolerance")

    rows: list[dict[str, Any]] = []
    for feature in features:
        candidate_values, candidate_invalid = _finite_numeric_values(
            candidate_frame[feature],
            label=f"candidates[{feature!r}]",
        )
        reference_values, reference_invalid = _finite_numeric_values(
            reference_frame[feature],
            label=f"reference_data[{feature!r}]",
        )
        rows.append(
            _feature_coverage_row(
                feature,
                candidate_values=candidate_values,
                reference_values=reference_values,
                candidate_invalid_count=candidate_invalid,
                reference_invalid_count=reference_invalid,
                tolerance=float(tolerance),
            )
        )
    return pd.DataFrame(rows)

summarize_candidate_category_coverage(candidates, reference_data, categorical_columns, *, max_levels=20)

Compare categorical candidate levels with reference/training data.

This is useful for finite BO pools that include solvents, processing labels, synthesis routes, supplier choices, or discrete formulation classes. The summary highlights candidate-only levels and reference levels that are absent from the candidate pool.

Source code in matgpr/candidate_generation.py

def summarize_candidate_category_coverage(
    candidates: pd.DataFrame,
    reference_data: pd.DataFrame,
    categorical_columns: Sequence[str],
    *,
    max_levels: int = 20,
) -> pd.DataFrame:
    """Compare categorical candidate levels with reference/training data.

    This is useful for finite BO pools that include solvents, processing
    labels, synthesis routes, supplier choices, or discrete formulation
    classes. The summary highlights candidate-only levels and reference levels
    that are absent from the candidate pool.
    """
    candidate_frame = _validate_candidate_frame(candidates, name="candidates")
    reference_frame = _validate_candidate_frame(reference_data, name="reference_data")
    columns = _validate_columns(categorical_columns, candidate_frame, name="categorical_columns")
    missing_reference = [column for column in columns if column not in reference_frame.columns]
    if missing_reference:
        raise ValueError(f"reference_data is missing categorical columns: {missing_reference}")
    max_levels = validate_positive_int(max_levels, name="max_levels")

    rows = []
    for column in columns:
        candidate_levels = _nonmissing_level_set(candidate_frame[column])
        reference_levels = _nonmissing_level_set(reference_frame[column])
        shared = candidate_levels.intersection(reference_levels)
        candidate_only = candidate_levels.difference(reference_levels)
        reference_only = reference_levels.difference(candidate_levels)
        reference_count = len(reference_levels)
        rows.append(
            {
                "column": column,
                "candidate_unique_count": len(candidate_levels),
                "reference_unique_count": reference_count,
                "shared_unique_count": len(shared),
                "reference_levels_covered_fraction": (
                    np.nan if reference_count == 0 else len(shared) / reference_count
                ),
                "candidate_new_level_count": len(candidate_only),
                "reference_missing_level_count": len(reference_only),
                "candidate_new_levels": _format_levels(candidate_only, max_levels=max_levels),
                "reference_missing_levels": _format_levels(reference_only, max_levels=max_levels),
            }
        )
    return pd.DataFrame(rows)

summarize_candidate_duplicates(candidates, key_columns)

Return duplicate candidate-key groups sorted by duplicate count.

Source code in matgpr/candidate_generation.py

def summarize_candidate_duplicates(
    candidates: pd.DataFrame,
    key_columns: Sequence[str],
) -> pd.DataFrame:
    """Return duplicate candidate-key groups sorted by duplicate count."""
    frame = _validate_candidate_frame(candidates, name="candidates")
    keys = _validate_columns(key_columns, frame, name="key_columns")
    if frame.empty:
        return _empty_duplicate_summary(keys)

    duplicates = (
        frame.groupby(list(keys), dropna=False, sort=False)
        .size()
        .reset_index(name="matgpr_duplicate_count")
    )
    duplicates = duplicates.loc[duplicates["matgpr_duplicate_count"] > 1].copy()
    if duplicates.empty:
        return _empty_duplicate_summary(keys)

    duplicates["matgpr_duplicate_fraction"] = duplicates["matgpr_duplicate_count"] / frame.shape[0]
    return duplicates.sort_values(
        "matgpr_duplicate_count",
        ascending=False,
    ).reset_index(drop=True)