Experiment Logging

Closed-loop experiment logging helpers record Bayesian-optimization recommendations, selected experiments, measured outcomes, and compact campaign summaries. The functions are dataframe-based so they can be used with BoTorch recommendations, manually selected batches, or external laboratory results. Restart helpers infer the next BO ask iteration, recover pending selections, separate completed observations, and filter finite candidate pools before a new session starts.

matgpr.experiment_logging

BOCampaignState dataclass

Restart state for a finite-pool closed-loop BO campaign.

Parameters:

Name	Type	Description	Default
campaign_id	str	Stable campaign identifier.	required
key_columns	tuple[str, ...]	Candidate key columns used to match recommendations, selections, observations, and candidate-pool rows.	required
current_iteration	int | None	Largest iteration number currently present in the campaign log, or None when the campaign has not logged any rows.	required
last_recommendation_iteration	int | None	Largest recommendation iteration currently present in the campaign log, or None when no BO ask has been logged yet.	required
next_iteration	int	Recommended iteration number for the next BO ask. This is inferred from recommendation rows, so observation rows can be logged as the iteration that will be used for the next model update.	required
log	DataFrame	Campaign-specific closed-loop log rows.	required
recommendations	DataFrame	Logged recommendation rows.	required
selections	DataFrame	Logged selection rows.	required
observations	DataFrame	Logged observation rows.	required
pending_experiments	DataFrame	Selected candidates that do not yet have matching observation rows.	required
completed_experiments	DataFrame	Observation rows, de-duplicated by key with the latest row kept.	required
unavailable_candidates	DataFrame	Candidate-pool rows or log rows that should not be recommended again. This includes completed observations and pending selections.	required
available_candidates	DataFrame	Candidate-pool rows that remain available after removing unavailable keys. Empty when candidate_pool was not provided.	required
candidate_pool_size	int | None	Number of rows in the original candidate pool, or None when no pool was provided.	None

Source code in matgpr/experiment_logging.py

@dataclass(frozen=True)
class BOCampaignState:
    """Restart state for a finite-pool closed-loop BO campaign.

    Parameters
    ----------
    campaign_id
        Stable campaign identifier.
    key_columns
        Candidate key columns used to match recommendations, selections,
        observations, and candidate-pool rows.
    current_iteration
        Largest iteration number currently present in the campaign log, or
        `None` when the campaign has not logged any rows.
    last_recommendation_iteration
        Largest recommendation iteration currently present in the campaign log,
        or `None` when no BO ask has been logged yet.
    next_iteration
        Recommended iteration number for the next BO ask. This is inferred
        from recommendation rows, so observation rows can be logged as the
        iteration that will be used for the next model update.
    log
        Campaign-specific closed-loop log rows.
    recommendations
        Logged recommendation rows.
    selections
        Logged selection rows.
    observations
        Logged observation rows.
    pending_experiments
        Selected candidates that do not yet have matching observation rows.
    completed_experiments
        Observation rows, de-duplicated by key with the latest row kept.
    unavailable_candidates
        Candidate-pool rows or log rows that should not be recommended again.
        This includes completed observations and pending selections.
    available_candidates
        Candidate-pool rows that remain available after removing unavailable
        keys. Empty when `candidate_pool` was not provided.
    candidate_pool_size
        Number of rows in the original candidate pool, or `None` when no pool
        was provided.
    """

    campaign_id: str
    key_columns: tuple[str, ...]
    current_iteration: int | None
    last_recommendation_iteration: int | None
    next_iteration: int
    log: pd.DataFrame
    recommendations: pd.DataFrame
    selections: pd.DataFrame
    observations: pd.DataFrame
    pending_experiments: pd.DataFrame
    completed_experiments: pd.DataFrame
    unavailable_candidates: pd.DataFrame
    available_candidates: pd.DataFrame
    candidate_pool_size: int | None = None

    def duplicate_policy(
        self,
        *,
        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,
    ) -> Any:
        """Build a duplicate policy for the next BO recommendation step."""
        from .bayesian_optimization import CandidateDuplicatePolicy

        resolved_keys = (
            self.key_columns if key_columns is None else _validate_key_columns(key_columns)
        )
        return CandidateDuplicatePolicy(
            existing_candidates=self.unavailable_candidates,
            key_columns=resolved_keys,
            feature_columns=feature_columns,
            feature_tolerance=feature_tolerance,
            metric=metric,
            feature_scales=feature_scales,
        )

duplicate_policy(*, key_columns=None, feature_columns=None, feature_tolerance=None, metric='euclidean', feature_scales=None)

Build a duplicate policy for the next BO recommendation step.

Source code in matgpr/experiment_logging.py

def duplicate_policy(
    self,
    *,
    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,
) -> Any:
    """Build a duplicate policy for the next BO recommendation step."""
    from .bayesian_optimization import CandidateDuplicatePolicy

    resolved_keys = (
        self.key_columns if key_columns is None else _validate_key_columns(key_columns)
    )
    return CandidateDuplicatePolicy(
        existing_candidates=self.unavailable_candidates,
        key_columns=resolved_keys,
        feature_columns=feature_columns,
        feature_tolerance=feature_tolerance,
        metric=metric,
        feature_scales=feature_scales,
    )

append_closed_loop_records(records, *, path, campaign_id, iteration, record_type, metadata=None, timestamp=None)

Append annotated rows to a closed-loop experiment CSV log.

Parameters:

Name	Type	Description	Default
records	DataFrame	Dataframe of recommendations, selected experiments, or measured observations to record.	required
path	str | Path	CSV path for the campaign log. Existing logs are read and rewritten so new record types can add new columns while preserving previous rows.	required
campaign_id	str	Stable label for the optimization campaign.	required
iteration	int	Non-negative closed-loop iteration number.	required
record_type	str	Short label such as "recommendation", "selection", or "observation".	required
metadata	Mapping[str, Any] | None	Optional scalar metadata written to matgpr_metadata_* columns.	None
timestamp	datetime | str | None	Optional timestamp. Datetime values are converted to UTC. If omitted, the current UTC time is used.	None

Returns:

Type	Description
DataFrame	The annotated rows that were appended to the log.

Source code in matgpr/experiment_logging.py

def append_closed_loop_records(
    records: pd.DataFrame,
    *,
    path: str | Path,
    campaign_id: str,
    iteration: int,
    record_type: str,
    metadata: Mapping[str, Any] | None = None,
    timestamp: datetime | str | None = None,
) -> pd.DataFrame:
    """Append annotated rows to a closed-loop experiment CSV log.

    Parameters
    ----------
    records
        Dataframe of recommendations, selected experiments, or measured
        observations to record.
    path
        CSV path for the campaign log. Existing logs are read and rewritten so
        new record types can add new columns while preserving previous rows.
    campaign_id
        Stable label for the optimization campaign.
    iteration
        Non-negative closed-loop iteration number.
    record_type
        Short label such as `"recommendation"`, `"selection"`, or
        `"observation"`.
    metadata
        Optional scalar metadata written to `matgpr_metadata_*` columns.
    timestamp
        Optional timestamp. Datetime values are converted to UTC. If omitted,
        the current UTC time is used.

    Returns
    -------
    pandas.DataFrame
        The annotated rows that were appended to the log.
    """
    raw_records = _validate_records(records)
    campaign_id = _validate_campaign_id(campaign_id)
    iteration = _validate_iteration(iteration)
    record_type = _validate_record_type(record_type)
    timestamp_text = _format_timestamp(timestamp)
    metadata_columns = _metadata_to_columns(metadata)

    collisions = [
        column
        for column in (*_RESERVED_LOG_COLUMNS, *metadata_columns)
        if column in raw_records.columns
    ]
    if collisions:
        raise ValueError(f"records contain columns reserved for closed-loop logging: {collisions}")

    prefix = pd.DataFrame(
        {
            "matgpr_campaign_id": campaign_id,
            "matgpr_iteration": iteration,
            "matgpr_record_type": record_type,
            "matgpr_timestamp_utc": timestamp_text,
            **metadata_columns,
        },
        index=raw_records.index,
    )
    annotated = pd.concat(
        [prefix.reset_index(drop=True), raw_records.reset_index(drop=True)],
        axis=1,
    )

    log_path = Path(path)
    log_path.parent.mkdir(parents=True, exist_ok=True)
    if log_path.exists() and log_path.stat().st_size > 0:
        existing = pd.read_csv(log_path)
        combined = pd.concat([existing, annotated], ignore_index=True, sort=False)
    else:
        combined = annotated.reset_index(drop=True)
    combined.to_csv(log_path, index=False)

    return annotated.reset_index(drop=True)

log_bo_recommendations(recommendations, *, path, campaign_id, iteration, model_name=None, acquisition_function=None, metadata=None, timestamp=None)

Log Bayesian-optimization recommendations for a campaign iteration.

Source code in matgpr/experiment_logging.py

def log_bo_recommendations(
    recommendations: pd.DataFrame,
    *,
    path: str | Path,
    campaign_id: str,
    iteration: int,
    model_name: str | None = None,
    acquisition_function: str | None = None,
    metadata: Mapping[str, Any] | None = None,
    timestamp: datetime | str | None = None,
) -> pd.DataFrame:
    """Log Bayesian-optimization recommendations for a campaign iteration."""
    merged_metadata = _merge_metadata(
        metadata,
        {
            "model_name": model_name,
            "acquisition_function": acquisition_function,
        },
    )
    return append_closed_loop_records(
        recommendations,
        path=path,
        campaign_id=campaign_id,
        iteration=iteration,
        record_type="recommendation",
        metadata=merged_metadata,
        timestamp=timestamp,
    )

log_selected_experiments(selected_experiments, *, path, campaign_id, iteration, selection_policy=None, metadata=None, timestamp=None)

Log the recommendations actually selected for experimental execution.

Source code in matgpr/experiment_logging.py

def log_selected_experiments(
    selected_experiments: pd.DataFrame,
    *,
    path: str | Path,
    campaign_id: str,
    iteration: int,
    selection_policy: str | None = None,
    metadata: Mapping[str, Any] | None = None,
    timestamp: datetime | str | None = None,
) -> pd.DataFrame:
    """Log the recommendations actually selected for experimental execution."""
    merged_metadata = _merge_metadata(
        metadata,
        {"selection_policy": selection_policy},
    )
    return append_closed_loop_records(
        selected_experiments,
        path=path,
        campaign_id=campaign_id,
        iteration=iteration,
        record_type="selection",
        metadata=merged_metadata,
        timestamp=timestamp,
    )

log_observations(observations, *, path, campaign_id, iteration, target_column=None, metadata=None, timestamp=None)

Log measured outcomes returned from selected experiments.

Source code in matgpr/experiment_logging.py

def log_observations(
    observations: pd.DataFrame,
    *,
    path: str | Path,
    campaign_id: str,
    iteration: int,
    target_column: str | None = None,
    metadata: Mapping[str, Any] | None = None,
    timestamp: datetime | str | None = None,
) -> pd.DataFrame:
    """Log measured outcomes returned from selected experiments."""
    merged_metadata = _merge_metadata(
        metadata,
        {"target_column": target_column},
    )
    return append_closed_loop_records(
        observations,
        path=path,
        campaign_id=campaign_id,
        iteration=iteration,
        record_type="observation",
        metadata=merged_metadata,
        timestamp=timestamp,
    )

load_closed_loop_log(path)

Load a closed-loop experiment CSV log.

Source code in matgpr/experiment_logging.py

def load_closed_loop_log(path: str | Path) -> pd.DataFrame:
    """Load a closed-loop experiment CSV log."""
    return pd.read_csv(Path(path))

infer_next_bo_iteration(log_data, *, campaign_id=None, recommendation_record_type='recommendation')

Infer the next BO ask iteration from a closed-loop campaign log.

The next iteration is based on logged recommendation rows. If a campaign has no recommendation rows yet, the next ask iteration is 0.

Source code in matgpr/experiment_logging.py

def infer_next_bo_iteration(
    log_data: pd.DataFrame | str | Path,
    *,
    campaign_id: str | None = None,
    recommendation_record_type: str = "recommendation",
) -> int:
    """Infer the next BO ask iteration from a closed-loop campaign log.

    The next iteration is based on logged recommendation rows. If a campaign
    has no recommendation rows yet, the next ask iteration is `0`.
    """
    log = _coerce_campaign_log_data(log_data)
    if log.empty:
        return 0
    _validate_log_schema(log)

    if campaign_id is not None:
        campaign_id = _validate_campaign_id(campaign_id)
        log = log.loc[log["matgpr_campaign_id"] == campaign_id].copy()
    if log.empty:
        return 0

    recommendation_record_type = _validate_record_type(recommendation_record_type)
    recommendations = log.loc[log["matgpr_record_type"] == recommendation_record_type].copy()
    if recommendations.empty:
        return 0
    iterations = _numeric_iteration_series(recommendations, label="recommendation log")
    return int(iterations.max()) + 1

resume_bo_campaign(log_data, *, campaign_id, key_columns=('candidate_id',), candidate_pool=None, recommendation_record_type='recommendation', selection_record_type='selection', observation_record_type='observation')

Build restart state for an ask-tell Bayesian-optimization campaign.

Use this helper at the beginning of a new session to recover the next iteration number, pending selections, completed observations, unavailable candidate keys, and the remaining candidate pool.

Source code in matgpr/experiment_logging.py

def resume_bo_campaign(
    log_data: pd.DataFrame | str | Path,
    *,
    campaign_id: str,
    key_columns: tuple[str, ...] = ("candidate_id",),
    candidate_pool: pd.DataFrame | None = None,
    recommendation_record_type: str = "recommendation",
    selection_record_type: str = "selection",
    observation_record_type: str = "observation",
) -> BOCampaignState:
    """Build restart state for an ask-tell Bayesian-optimization campaign.

    Use this helper at the beginning of a new session to recover the next
    iteration number, pending selections, completed observations, unavailable
    candidate keys, and the remaining candidate pool.
    """
    campaign_id = _validate_campaign_id(campaign_id)
    key_columns = _validate_key_columns(key_columns)
    recommendation_record_type = _validate_record_type(recommendation_record_type)
    selection_record_type = _validate_record_type(selection_record_type)
    observation_record_type = _validate_record_type(observation_record_type)

    log = _coerce_campaign_log_data(log_data)
    if not log.empty:
        _validate_log_schema(log)
        log = log.loc[log["matgpr_campaign_id"] == campaign_id].reset_index(drop=True)
    else:
        log = _empty_closed_loop_log()

    candidate_pool_data = _validate_candidate_pool(candidate_pool, key_columns)

    recommendations = _records_by_type(log, recommendation_record_type)
    selections = _records_by_type(log, selection_record_type)
    observations = _records_by_type(log, observation_record_type)
    completed = _latest_records_by_key(observations, key_columns, label="observation records")
    pending = _pending_selection_records(selections, completed, key_columns)

    unavailable_records = pd.concat(
        [completed, pending],
        ignore_index=True,
        sort=False,
    )
    if unavailable_records.empty:
        unavailable_records = pd.DataFrame(columns=list(key_columns))
    unavailable_keys = _row_key_set(unavailable_records, key_columns)

    if candidate_pool_data is None:
        available_candidates = pd.DataFrame()
        unavailable_candidates = unavailable_records.reset_index(drop=True)
        candidate_pool_size = None
    else:
        candidate_keys = _row_keys(candidate_pool_data, key_columns, label="candidate_pool")
        available_mask = [key not in unavailable_keys for key in candidate_keys]
        unavailable_mask = [not keep for keep in available_mask]
        available_candidates = candidate_pool_data.loc[available_mask].reset_index(drop=True)
        unavailable_candidates = candidate_pool_data.loc[unavailable_mask].reset_index(drop=True)
        candidate_pool_size = int(candidate_pool_data.shape[0])

    current_iteration = _latest_iteration(log)
    last_recommendation_iteration = _latest_iteration(recommendations)
    next_iteration = infer_next_bo_iteration(
        log,
        campaign_id=campaign_id,
        recommendation_record_type=recommendation_record_type,
    )

    return BOCampaignState(
        campaign_id=campaign_id,
        key_columns=key_columns,
        current_iteration=current_iteration,
        last_recommendation_iteration=last_recommendation_iteration,
        next_iteration=next_iteration,
        log=log.reset_index(drop=True),
        recommendations=recommendations.reset_index(drop=True),
        selections=selections.reset_index(drop=True),
        observations=observations.reset_index(drop=True),
        pending_experiments=pending.reset_index(drop=True),
        completed_experiments=completed.reset_index(drop=True),
        unavailable_candidates=unavailable_candidates,
        available_candidates=available_candidates,
        candidate_pool_size=candidate_pool_size,
    )

summarize_closed_loop_log(log_data, *, campaign_id=None, target_column=None)

Summarize closed-loop records by campaign, iteration, and record type.

The summary always reports record counts. When target_column is provided, the summary also reports target count, mean, minimum, and maximum for each group. This is useful for quickly auditing whether a BO campaign has moved from recommendation to selection to measured outcome.

Source code in matgpr/experiment_logging.py

def summarize_closed_loop_log(
    log_data: pd.DataFrame | str | Path,
    *,
    campaign_id: str | None = None,
    target_column: str | None = None,
) -> pd.DataFrame:
    """Summarize closed-loop records by campaign, iteration, and record type.

    The summary always reports record counts. When `target_column` is provided,
    the summary also reports target count, mean, minimum, and maximum for each
    group. This is useful for quickly auditing whether a BO campaign has
    moved from recommendation to selection to measured outcome.
    """
    log = _coerce_log_data(log_data)
    _validate_log_schema(log)

    if campaign_id is not None:
        campaign_id = _validate_campaign_id(campaign_id)
        log = log.loc[log["matgpr_campaign_id"] == campaign_id].copy()

    summary_columns = [*_SUMMARY_GROUP_COLUMNS, "matgpr_record_count"]
    if log.empty:
        if target_column is not None:
            summary_columns.extend(
                [
                    "matgpr_target_count",
                    "matgpr_target_mean",
                    "matgpr_target_min",
                    "matgpr_target_max",
                ]
            )
        return pd.DataFrame(columns=summary_columns)

    summary = (
        log.groupby(list(_SUMMARY_GROUP_COLUMNS), dropna=False)
        .size()
        .reset_index(name="matgpr_record_count")
    )

    if target_column is not None:
        target_column = str(target_column).strip()
        if not target_column:
            raise ValueError("target_column must be non-empty when provided")
        if target_column not in log.columns:
            raise ValueError(f"target_column {target_column!r} is missing from the log")

        target_log = log.assign(_matgpr_target=pd.to_numeric(log[target_column], errors="coerce"))
        target_summary = (
            target_log.groupby(list(_SUMMARY_GROUP_COLUMNS), dropna=False)["_matgpr_target"]
            .agg(
                matgpr_target_count=lambda values: int(values.notna().sum()),
                matgpr_target_mean="mean",
                matgpr_target_min="min",
                matgpr_target_max="max",
            )
            .reset_index()
        )
        summary = summary.merge(
            target_summary,
            on=list(_SUMMARY_GROUP_COLUMNS),
            how="left",
        )

    return summary.sort_values(list(_SUMMARY_GROUP_COLUMNS)).reset_index(drop=True)