Nelder-Mead Generator

NelderMeadGenerator

Bases: SequentialGenerator

Nelder-Mead algorithm from SciPy in Xopt's Generator form. Converted to use a state machine to resume in exactly the last state.

Attributes:

Name	Type	Description
name	str	The name of the generator.
initial_point	Optional[Dict[str, float]]	Initial point for the optimization.
initial_simplex	Optional[Dict[str, Union[List[float], ndarray]]]	Initial simplex for the optimization.
adaptive	bool	Change hyperparameters based on dimensionality.
current_state	SimplexState	Current state of the simplex.
future_state	Optional[SimplexState]	Future state of the simplex.
x	Optional[ndarray]	Current x value.
y	Optional[float]	Current y value.

Methods:

Name	Description
add_data	Add new data to the generator.
generate	Generate a specified number of candidate samples.
_call_algorithm	Call the Nelder-Mead algorithm.
simplex	Returns the simplex in the current state.

Source code in xopt/generators/sequential/neldermead.py

class NelderMeadGenerator(SequentialGenerator):
    """
    Nelder-Mead algorithm from SciPy in Xopt's Generator form.
    Converted to use a state machine to resume in exactly the last state.

    Attributes
    ----------
    name : str
        The name of the generator.
    initial_point : Optional[Dict[str, float]]
        Initial point for the optimization.
    initial_simplex : Optional[Dict[str, Union[List[float], np.ndarray]]]
        Initial simplex for the optimization.
    adaptive : bool
        Change hyperparameters based on dimensionality.
    current_state : SimplexState
        Current state of the simplex.
    future_state : Optional[SimplexState]
        Future state of the simplex.
    x : Optional[np.ndarray]
        Current x value.
    y : Optional[float]
        Current y value.

    Methods
    -------
    add_data(self, new_data: pd.DataFrame)
        Add new data to the generator.
    generate(self, n_candidates: int) -> Optional[List[Dict[str, float]]]
        Generate a specified number of candidate samples.
    _call_algorithm(self)
        Call the Nelder-Mead algorithm.
    simplex(self) -> Dict[str, np.ndarray]
        Returns the simplex in the current state.
    """

    name = "neldermead"
    supports_single_objective: bool = True

    initial_point: Optional[Dict[str, float]] = None  # replaces x0 argument
    initial_simplex: Optional[Dict[str, Union[List[float], np.ndarray]]] = (
        None  # This overrides the use of initial_point
    )
    # Same as scipy.optimize._optimize._minimize_neldermead
    adaptive: bool = Field(
        True, description="Change hyperparameters based on dimensionality"
    )
    current_state: SimplexState = SimplexState()
    future_state: Optional[SimplexState] = None

    # Internal data structures
    x: Optional[np.ndarray] = None
    y: Optional[float] = None
    manual_data_cnt: int = Field(
        0, description="How many points are considered manual/not part of simplex run"
    )

    trace: bool = Field(
        False, description="If True, print trace messages during optimization"
    )

    _initial_simplex = None
    _initial_point = None
    _saved_options: Dict = None

    def __init__(self, **kwargs):
        super().__init__(**kwargs)

        self._saved_options = self.model_dump(
            exclude={"current_state", "future_state"}
        ).copy()  # Used to keep track of changed options

        if self.initial_simplex:
            self._initial_simplex = np.array(
                [self.initial_simplex[k] for k in self.vocs.variable_names]
            ).T
        else:
            self._initial_simplex = None

    def _generate(self, first_gen: bool = False) -> Optional[List[Dict[str, float]]]:
        """
        Generate candidate.

        Returns
        -------
        Optional[List[Dict[str, float]]]
            A list of dictionaries containing the generated samples.
        """

        if self.current_state.N is None:
            # fresh start
            pass
        else:
            n_inputs = len(self.data)
            if self.current_state.ngen == n_inputs:
                # We are in a state where result of last point is known
                pass
            else:
                pass

        results = self._call_algorithm()

        x, state_extra = results
        assert len(state_extra) == len(STATE_KEYS)
        stateobj = SimplexState(**{k: v for k, v in zip(STATE_KEYS, state_extra)})
        self.future_state = stateobj

        inputs = dict(zip(self.vocs.variable_names, x))
        if self.vocs.constants is not None:
            inputs.update(self.vocs.constants)

        return [inputs]

    @property
    def x0(self) -> np.ndarray:
        """
        Raw internal initial point for convenience.
        If initial_point is set, it will be used. Otherwise, if data is set, the last point will be used.

        Returns
        -------
        np.ndarray
            The initial point as a NumPy array.
        """
        if self._initial_point is not None:
            return self._initial_point
        elif self.initial_point is not None:
            return np.array([self.initial_point[k] for k in self.vocs.variable_names])
        elif self.data is not None:
            return self._get_initial_point()[0]
        else:
            raise ValueError(
                "No initial point specified in generator or taken from data"
            )

    def _add_data(self, new_data: pd.DataFrame):
        """
        Add new data to the generator.

        Parameters
        ----------
        new_data : pd.DataFrame
            The new data to be added.
        """
        if len(new_data) == 0:
            # empty data, i.e. no steps yet
            assert self.future_state is None
            return

        ndata = len(self.data)
        ngen = self.current_state.ngen

        # Complicated part - need to determine if data corresponds to result of last gen
        if not self.is_active:
            assert self.future_state is None, "Not active, but future state exists?"

            variable_data = get_variable_data(self.vocs, self.data).to_numpy()
            objective_data = get_objective_data(self.vocs, self.data).to_numpy()[:, 0]

            _initial_simplex = variable_data.copy()
            N = self.vocs.n_variables
            if _initial_simplex.shape[0] >= N + 1:
                # TODO: is this really needed?
                # if we have enough, form new simplex and force state to just after it is all probed
                _initial_simplex = _initial_simplex[-(N + 1) :, :]
                objective_data = objective_data[-(N + 1) :]

                fake_initialized_state = _fake_partial_state_gen(
                    _initial_simplex, objective_data
                )
                self.current_state = fake_initialized_state
                self._initial_simplex = _initial_simplex
                self.manual_data_cnt = len(self.data) - (N + 1)
            else:
                # otherwise, just set the last point as the initial point - effectively loses output data
                self.current_state = SimplexState()
                self._initial_simplex = None
                self.manual_data_cnt = len(self.data)

            self._initial_point = _initial_simplex[-1, :]
            self.y = float(objective_data[-1])
        else:
            assert new_data.shape[0] == 1, (
                "Only one point at a time is expected in active mode"
            )
            # new data -> advance state machine 1 step
            n_auto_points = ndata - self.manual_data_cnt
            assert n_auto_points == self.future_state.ngen, (
                f"Internal error {n_auto_points=} {self.future_state=} {ndata=}"
            )
            assert n_auto_points == ngen + 1, f"Internal error {n_auto_points=} {ngen=}"
            self.current_state = self.future_state
            self.future_state = None

            # Can have multiple points if resuming from file, grab last one
            new_data_df = get_objective_data(self.vocs, new_data)
            res = new_data_df.iloc[-1:, :].to_numpy()
            assert np.shape(res) == (1, 1), f"Bad last point [{res}]"

            yt = res[0, 0].item()

            self.y = yt

    def _set_data(self, data: pd.DataFrame):
        # just store data
        self.data = data

        new_data_df = get_objective_data(self.vocs, data)
        res = new_data_df.iloc[-1:, :].to_numpy()
        assert np.shape(res) == (1, 1), f"Bad last point [{res}]"

    def _reset(self):
        self.current_state = SimplexState()
        self.future_state = None
        if self.initial_simplex:
            self._initial_simplex = np.array(
                [self.initial_simplex[k] for k in self.vocs.variable_names]
            ).T
        else:
            self._initial_simplex = None

    def _call_algorithm(self):
        mins, maxs = np.array(self.vocs.bounds).T
        results = _neldermead_generator(
            x0=self.x0,
            state=self.current_state,
            lastval=self.y,
            adaptive=self.adaptive,
            initial_simplex=self._initial_simplex,
            bounds=(mins, maxs),
            trace=self.trace,
        )

        self.y = None
        return results

    @property
    def simplex(self) -> Dict[str, np.ndarray]:
        """
        Returns the simplex in the current state.

        Returns
        -------
        Dict[str, np.ndarray]
            The simplex in the current state.
        """
        sim = self.current_state.sim
        return dict(zip(self.vocs.variable_names, sim.T))

simplex property

Returns the simplex in the current state.

Returns:

Type	Description
Dict[str, ndarray]	The simplex in the current state.

x0 property

Raw internal initial point for convenience. If initial_point is set, it will be used. Otherwise, if data is set, the last point will be used.

Returns:

Type	Description
ndarray	The initial point as a NumPy array.

add_data(new_data)

Add new data to the generator.

Parameters:

Name	Type	Description	Default
new_data	DataFrame	The new data to add.	required

Raises:

Type	Description
ValueError	If the generator is active but no candidate was generated, or if the new data does not contain the last candidate.

Source code in xopt/generators/sequential/sequential_generator.py

def add_data(self, new_data: pd.DataFrame):
    """
    Add new data to the generator.

    Parameters
    ----------
    new_data : pd.DataFrame
        The new data to add.

    Raises
    ------
    ValueError
        If the generator is active but no candidate was generated, or if the new data does not contain the last candidate.
    """
    # if the generator is active then the new data must contain the last candidate
    if self.is_active:
        if self._last_candidate is None:
            raise SeqGeneratorError(
                "Generator is active, but no candidate was generated. Cannot add data."
            )
        if len(new_data) > 1:
            raise SeqGeneratorError(
                "Cannot add more than one data point when generator is active."
            )
        else:
            # check if the last candidate is in the new data
            self.validate_point(new_data.iloc[0].to_dict())

    # do not call super, this will likely need to be customized for some generators
    if self.data is not None:
        self.data = pd.concat([self.data, new_data], axis=0, ignore_index=True)
    else:
        self.data = new_data

    # update internal state of the generator
    self._add_data(new_data)

generate(n_candidates=1)

Generate a new candidate point.

Parameters:

Name	Type	Description	Default
n_candidates	int	Number of candidates to generate, by default 1.	1

Returns:

Type	Description
dict	A dictionary representing the candidate point.

Raises:

Type	Description
ValueError	If more than one candidate is requested.

Source code in xopt/generators/sequential/sequential_generator.py

def generate(self, n_candidates: int = 1) -> dict:
    """
    Generate a new candidate point.

    Parameters
    ----------
    n_candidates : int, optional
        Number of candidates to generate, by default 1.

    Returns
    -------
    dict
        A dictionary representing the candidate point.

    Raises
    ------
    ValueError
        If more than one candidate is requested.
    """
    # we cannot generate more than one candidate at a time
    if n_candidates > 1:
        raise SeqGeneratorError(
            "Sequential generators can only generate one candidate at a time."
        )

    # if the generator is not active, we need to start it
    if not self.is_active:
        candidate = self._generate(True)
        self.is_active = True
    else:
        candidate = self._generate()

    # need to store the candidate to validate adding data to the generator
    self._last_candidate = candidate

    return candidate

model_dump(*args, **kwargs)

overwrite model dump to remove faux class attrs

Source code in xopt/generator.py

def model_dump(self, *args: Any, **kwargs: Any) -> dict[str, Any]:
    """overwrite model dump to remove faux class attrs"""

    res = super().model_dump(*args, **kwargs)

    res.pop("supports_batch_generation", None)
    res.pop("supports_multi_objective", None)

    return res

reset()

Reset the generator.

Source code in xopt/generators/sequential/sequential_generator.py

def reset(self):
    """
    Reset the generator.
    """
    self.is_active = False
    self._last_candidate = None
    self._reset()

set_data(data)

Set the full dataset for the generator. Typically only used when loading from a save file. This skips active generator lockout.

Parameters:

Name	Type	Description	Default
data	DataFrame	The data to set.	required

Source code in xopt/generators/sequential/sequential_generator.py

def set_data(self, data: pd.DataFrame):
    """
    Set the full dataset for the generator. Typically only used when loading from a save file. This skips active
    generator lockout.

    Parameters
    ----------
    data : pd.DataFrame
        The data to set.
    """
    # TODO: make a flag for generator that support multiple data sets
    if self._data_set:
        raise SeqGeneratorError(
            "Data has already been initialized for this generator."
        )
    self._set_data(data)
    self._data_set = True

validate_point(point)

determine if an input point was generated by the generator

Source code in xopt/generators/sequential/sequential_generator.py

def validate_point(self, point: Dict[str, float]):
    """determine if an input point was generated by the generator"""
    last_candidate = np.array(
        [self._last_candidate[0][ele] for ele in self.vocs.variable_names]
    )
    point_variables = np.array(
        [point[ele] for ele in self.vocs.variable_names]
    ).flatten()
    if not np.allclose(last_candidate, point_variables, atol=0.0, rtol=1e-6):
        raise SeqGeneratorError(
            "Cannot add data that was not generated by the generator when generator is active. "
            "Call reset() to reset the generator first in order to add data via other methods."
        )

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)

SimplexState

Bases: XoptBaseModel

Container model for all simplex parameters

Source code in xopt/generators/sequential/neldermead.py

class SimplexState(XoptBaseModel):
    """Container model for all simplex parameters"""

    astg: int = -1
    N: Optional[int] = None
    kend: int = 0
    jend: int = 0
    ind: Optional[np.ndarray] = None
    sim: Optional[np.ndarray] = None
    fsim: Optional[np.ndarray] = None
    fxr: Optional[float] = None
    x: Optional[np.ndarray] = None
    xr: Optional[np.ndarray] = None
    xe: Optional[np.ndarray] = None
    xc: Optional[np.ndarray] = None
    xcc: Optional[np.ndarray] = None
    xbar: Optional[np.ndarray] = None
    doshrink: int = 0
    ngen: int = 0
    model_config = ConfigDict(arbitrary_types_allowed=True)

    @field_validator(
        "ind", "fsim", "sim", "x", "xr", "xe", "xc", "xcc", "xbar", mode="before"
    )
    def to_numpy(cls, v):
        return np.array(v, dtype=np.float64)

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)