Stopping Conditions¶

Stopping conditions for Xopt optimization.

This module contains classes that implement various stopping criteria for optimization processes. Each stopping condition class takes an Xopt dataframe and VOCS object to determine whether optimization should stop.

`CompositeCondition` ¶

Bases: StoppingCondition

Combine multiple stopping conditions with AND/OR logic.

Parameters:

Name	Type	Description	Default
`conditions`	`List[StoppingCondition]`	List of stopping conditions to combine.	required
`logic`	`str`	Logic to combine conditions: "and" or "or" (default: "or").	required

Source code in xopt/stopping_conditions.py

class CompositeCondition(StoppingCondition):
    """
    Combine multiple stopping conditions with AND/OR logic.

    Parameters
    ----------
    conditions : List[StoppingCondition]
        List of stopping conditions to combine.
    logic : str, optional
        Logic to combine conditions: "and" or "or" (default: "or").
    """

    name: Literal["CompositeCondition"] = "CompositeCondition"
    conditions: List["StoppingConditionUnion"] = Field(
        description="List of stopping conditions to combine", min_length=1
    )
    logic: str = Field(
        default="or", description="Logic to combine conditions: 'and' or 'or'"
    )

    @field_serializer("conditions")
    @classmethod
    def serialize_conditions(cls, v):
        serialized_conditions = []
        for condition in v:
            serialized_conditions.append(
                condition.model_dump() | {"name": condition.__class__.__name__}
            )
        return serialized_conditions

    @field_validator("logic")
    @classmethod
    def validate_logic(cls, v):
        if v.lower() not in ["and", "or"]:
            raise ValueError("logic must be 'and' or 'or'")
        return v.lower()

    def should_stop(self, data: pd.DataFrame, vocs: VOCS) -> bool:
        """Combine conditions using specified logic."""
        results = []

        for condition in self.conditions:
            _should_stop = condition.should_stop(data, vocs)
            if _should_stop and self.logic == "or":
                return True
            results.append(_should_stop)

        if self.logic == "and":
            return all(results)
        return False

`should_stop(data, vocs)` ¶

Combine conditions using specified logic.

Source code in xopt/stopping_conditions.py

def should_stop(self, data: pd.DataFrame, vocs: VOCS) -> bool:
    """Combine conditions using specified logic."""
    results = []

    for condition in self.conditions:
        _should_stop = condition.should_stop(data, vocs)
        if _should_stop and self.logic == "or":
            return True
        results.append(_should_stop)

    if self.logic == "and":
        return all(results)
    return False

`yaml(**kwargs)` ¶

serialize first then dump to yaml string

Source code in xopt/pydantic.py

def yaml(self, **kwargs):
    """serialize first then dump to yaml string"""
    output = json.loads(
        self.to_json(
            **kwargs,
        )
    )
    return yaml.dump(output)

`ConvergenceCondition` ¶

Bases: StoppingCondition

Stop when optimization converges (improvement is below threshold).

Parameters:

Name	Type	Description	Default
`objective_name`	`str`	Name of the objective to monitor for convergence.	required
`improvement_threshold`	`float`	Minimum improvement required to continue optimization.	required
`patience`	`int`	Number of evaluations to wait without improvement before stopping.	required
`relative`	`bool`	Whether to use relative improvement (default: False).	required

Source code in xopt/stopping_conditions.py

class ConvergenceCondition(StoppingCondition):
    """
    Stop when optimization converges (improvement is below threshold).

    Parameters
    ----------
    objective_name : str
        Name of the objective to monitor for convergence.
    improvement_threshold : float
        Minimum improvement required to continue optimization.
    patience : int
        Number of evaluations to wait without improvement before stopping.
    relative : bool, optional
        Whether to use relative improvement (default: False).
    """

    name: Literal["ConvergenceCondition"] = "ConvergenceCondition"
    objective_name: str = Field(description="Name of the objective to monitor")
    improvement_threshold: PositiveFloat = Field(
        description="Minimum improvement required to continue optimization"
    )
    patience: PositiveInt = Field(
        description="Number of evaluations to wait without improvement"
    )
    relative: bool = Field(
        default=False, description="Whether to use relative improvement"
    )

    def should_stop(self, data: pd.DataFrame, vocs: VOCS) -> bool:
        """Stop if no improvement for specified patience."""
        if data.empty or self.objective_name not in vocs.objectives:
            return False

        if self.objective_name not in data.columns:
            return False

        if len(data) < self.patience + 1:
            return False

        objective_type = vocs.objectives[self.objective_name]
        objective_values = data[self.objective_name].dropna()

        # Check improvement over the last 'patience' evaluations
        recent_values = objective_values.iloc[-(self.patience + 1) :]

        if isinstance(objective_type, MinimizeObjective):
            best_recent = recent_values.min()
            baseline = recent_values.iloc[0]
            improvement = baseline - best_recent
        else:  # MAXIMIZE
            best_recent = recent_values.max()
            baseline = recent_values.iloc[0]
            improvement = best_recent - baseline

        if self.relative and abs(baseline) > 1e-12:
            improvement = improvement / abs(baseline)

        return improvement < self.improvement_threshold

`should_stop(data, vocs)` ¶

Stop if no improvement for specified patience.

Source code in xopt/stopping_conditions.py

def should_stop(self, data: pd.DataFrame, vocs: VOCS) -> bool:
    """Stop if no improvement for specified patience."""
    if data.empty or self.objective_name not in vocs.objectives:
        return False

    if self.objective_name not in data.columns:
        return False

    if len(data) < self.patience + 1:
        return False

    objective_type = vocs.objectives[self.objective_name]
    objective_values = data[self.objective_name].dropna()

    # Check improvement over the last 'patience' evaluations
    recent_values = objective_values.iloc[-(self.patience + 1) :]

    if isinstance(objective_type, MinimizeObjective):
        best_recent = recent_values.min()
        baseline = recent_values.iloc[0]
        improvement = baseline - best_recent
    else:  # MAXIMIZE
        best_recent = recent_values.max()
        baseline = recent_values.iloc[0]
        improvement = best_recent - baseline

    if self.relative and abs(baseline) > 1e-12:
        improvement = improvement / abs(baseline)

    return improvement < self.improvement_threshold

`yaml(**kwargs)` ¶

serialize first then dump to yaml string

Source code in xopt/pydantic.py

def yaml(self, **kwargs):
    """serialize first then dump to yaml string"""
    output = json.loads(
        self.to_json(
            **kwargs,
        )
    )
    return yaml.dump(output)

`FeasibilityCondition` ¶

Bases: StoppingCondition

Stop when a feasible solution is found.

Parameters:

Name	Type	Description	Default
`require_all_constraints`	`bool`	Whether all constraints must be satisfied (default: True).	required

Source code in xopt/stopping_conditions.py

class FeasibilityCondition(StoppingCondition):
    """
    Stop when a feasible solution is found.

    Parameters
    ----------
    require_all_constraints : bool, optional
        Whether all constraints must be satisfied (default: True).
    """

    name: Literal["FeasibilityCondition"] = "FeasibilityCondition"
    require_all_constraints: bool = Field(
        default=True, description="Whether all constraints must be satisfied"
    )

    def should_stop(self, data: pd.DataFrame, vocs: VOCS) -> bool:
        """Stop when a feasible solution is found."""
        if data.empty or not vocs.constraints:
            return False

        # Use VOCS to determine feasibility
        feasibility_data = get_feasibility_data(vocs, data)

        if self.require_all_constraints:
            # Stop if any point is fully feasible
            return feasibility_data["feasible"].any()
        else:
            # Stop if any individual constraint is satisfied
            constraint_columns = [
                col
                for col in feasibility_data.columns
                if col.startswith("feasible_") and col != "feasible"
            ]
            if constraint_columns:
                return feasibility_data[constraint_columns].any().any()

`should_stop(data, vocs)` ¶

Stop when a feasible solution is found.

Source code in xopt/stopping_conditions.py

def should_stop(self, data: pd.DataFrame, vocs: VOCS) -> bool:
    """Stop when a feasible solution is found."""
    if data.empty or not vocs.constraints:
        return False

    # Use VOCS to determine feasibility
    feasibility_data = get_feasibility_data(vocs, data)

    if self.require_all_constraints:
        # Stop if any point is fully feasible
        return feasibility_data["feasible"].any()
    else:
        # Stop if any individual constraint is satisfied
        constraint_columns = [
            col
            for col in feasibility_data.columns
            if col.startswith("feasible_") and col != "feasible"
        ]
        if constraint_columns:
            return feasibility_data[constraint_columns].any().any()

`yaml(**kwargs)` ¶

serialize first then dump to yaml string

Source code in xopt/pydantic.py

def yaml(self, **kwargs):
    """serialize first then dump to yaml string"""
    output = json.loads(
        self.to_json(
            **kwargs,
        )
    )
    return yaml.dump(output)

`MaxEvaluationsCondition` ¶

Bases: StoppingCondition

Stop after a maximum number of evaluations. Evaluations can be counted in different ways based on parameters. If count_valid_only is True, only evaluations without errors are counted. If use_dataframe_index is True, the dataframe index is used to count evaluations instead of the number of rows.

Parameters:

Name	Type	Description	Default
`max_evaluations`	`int`	Maximum number of function evaluations before stopping.	required
`count_valid_only`	`bool`	Whether to count only valid evaluations (default: False).	required
`use_dataframe_index`	`bool`	Whether to use the dataframe index to count evaluations (default: False).	required

Source code in xopt/stopping_conditions.py

class MaxEvaluationsCondition(StoppingCondition):
    """
    Stop after a maximum number of evaluations. Evaluations can be counted
    in different ways based on parameters. If count_valid_only is True, only
    evaluations without errors are counted. If use_dataframe_index is True,
    the dataframe index is used to count evaluations instead of the number of rows.

    Parameters
    ----------
    max_evaluations : int
        Maximum number of function evaluations before stopping.
    count_valid_only : bool, optional
        Whether to count only valid evaluations (default: False).
    use_dataframe_index : bool, optional
        Whether to use the dataframe index to count evaluations (default: False).
    """

    name: Literal["MaxEvaluationsCondition"] = "MaxEvaluationsCondition"
    max_evaluations: PositiveInt = Field(
        description="Maximum number of function evaluations"
    )
    count_valid_only: bool = Field(
        default=False,
        description="Whether to count only valid evaluations that do not raise errors",
    )
    use_dataframe_index: bool = Field(
        default=False,
        description="Whether to use the dataframe index to count evaluations",
    )

    def should_stop(self, data: pd.DataFrame, vocs: VOCS) -> bool:
        """Stop if we've reached the maximum number of evaluations."""
        if data.empty:
            return False

        if self.use_dataframe_index:
            # assume index starts at 0
            return max(data.index) + 1 >= self.max_evaluations

        if self.count_valid_only:
            valid_data = data[data["error"].isna()] if "error" in data.columns else data
            return len(valid_data) >= self.max_evaluations

        return len(data) >= self.max_evaluations

`should_stop(data, vocs)` ¶

Stop if we've reached the maximum number of evaluations.

Source code in xopt/stopping_conditions.py

def should_stop(self, data: pd.DataFrame, vocs: VOCS) -> bool:
    """Stop if we've reached the maximum number of evaluations."""
    if data.empty:
        return False

    if self.use_dataframe_index:
        # assume index starts at 0
        return max(data.index) + 1 >= self.max_evaluations

    if self.count_valid_only:
        valid_data = data[data["error"].isna()] if "error" in data.columns else data
        return len(valid_data) >= self.max_evaluations

    return len(data) >= self.max_evaluations

`yaml(**kwargs)` ¶

serialize first then dump to yaml string

Source code in xopt/pydantic.py

def yaml(self, **kwargs):
    """serialize first then dump to yaml string"""
    output = json.loads(
        self.to_json(
            **kwargs,
        )
    )
    return yaml.dump(output)

`StagnationCondition` ¶

Bases: StoppingCondition

Stop when the best objective value hasn't improved for a number of evaluations.

Parameters:

Name	Type	Description	Default
`objective_name`	`str`	Name of the objective to monitor.	required
`patience`	`int`	Number of evaluations without improvement before stopping.	required
`tolerance`	`float`	Minimum improvement considered significant (default: 1e-8).	required

Source code in xopt/stopping_conditions.py

class StagnationCondition(StoppingCondition):
    """
    Stop when the best objective value hasn't improved for a number of evaluations.

    Parameters
    ----------
    objective_name : str
        Name of the objective to monitor.
    patience : int
        Number of evaluations without improvement before stopping.
    tolerance : float, optional
        Minimum improvement considered significant (default: 1e-8).
    """

    name: Literal["StagnationCondition"] = "StagnationCondition"
    objective_name: str = Field(description="Name of the objective to monitor")
    patience: PositiveInt = Field(
        description="Number of evaluations without improvement before stopping"
    )
    tolerance: PositiveFloat = Field(
        default=1e-8, description="Minimum improvement considered significant"
    )

    def should_stop(self, data: pd.DataFrame, vocs: VOCS) -> bool:
        """Stop if no improvement in best value for specified patience."""
        if data.empty or self.objective_name not in vocs.objectives:
            return False

        if self.objective_name not in data.columns:
            return False

        if len(data) < self.patience + 1:
            return False

        objective_type = vocs.objectives[self.objective_name]
        objective_values = data[self.objective_name].dropna()

        # Track the best value seen so far
        if isinstance(objective_type, MinimizeObjective):
            cumulative_best = objective_values.cummin()
        else:  # MAXIMIZE
            cumulative_best = objective_values.cummax()

        recent_best = cumulative_best.iloc[-1]
        past_best = cumulative_best.iloc[-(self.patience + 1)]

        if isinstance(objective_type, MinimizeObjective):
            improvement = past_best - recent_best
        else:  # MAXIMIZE
            improvement = recent_best - past_best

        return improvement < self.tolerance

`should_stop(data, vocs)` ¶

Stop if no improvement in best value for specified patience.

Source code in xopt/stopping_conditions.py

def should_stop(self, data: pd.DataFrame, vocs: VOCS) -> bool:
    """Stop if no improvement in best value for specified patience."""
    if data.empty or self.objective_name not in vocs.objectives:
        return False

    if self.objective_name not in data.columns:
        return False

    if len(data) < self.patience + 1:
        return False

    objective_type = vocs.objectives[self.objective_name]
    objective_values = data[self.objective_name].dropna()

    # Track the best value seen so far
    if isinstance(objective_type, MinimizeObjective):
        cumulative_best = objective_values.cummin()
    else:  # MAXIMIZE
        cumulative_best = objective_values.cummax()

    recent_best = cumulative_best.iloc[-1]
    past_best = cumulative_best.iloc[-(self.patience + 1)]

    if isinstance(objective_type, MinimizeObjective):
        improvement = past_best - recent_best
    else:  # MAXIMIZE
        improvement = recent_best - past_best

    return improvement < self.tolerance

`yaml(**kwargs)` ¶

serialize first then dump to yaml string

Source code in xopt/pydantic.py

def yaml(self, **kwargs):
    """serialize first then dump to yaml string"""
    output = json.loads(
        self.to_json(
            **kwargs,
        )
    )
    return yaml.dump(output)