Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix add typehint without generic #507

Merged
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
31 commits
Select commit Hold shift + click to select a range
caac1bd
fix: acquisition.py
phi-friday Jul 25, 2024
c764e9e
fix: constraint.py
phi-friday Jul 25, 2024
064d060
fix: target_space.py
phi-friday Jul 25, 2024
e390684
fix: domain_reduction.py
phi-friday Jul 25, 2024
5fa63bb
fix: logger.py
phi-friday Jul 25, 2024
ee29f0e
fix: observer.py
phi-friday Jul 25, 2024
24471d5
fix: bayesian_optimization.py
phi-friday Jul 25, 2024
1938b61
fix: util.py
phi-friday Jul 25, 2024
c47f4fb
fix
phi-friday Jul 25, 2024
7c7527b
fix: dict -> Mapping
phi-friday Jul 25, 2024
4668e8e
fix: allow Sequence
phi-friday Jul 25, 2024
f459560
fix: diallow null
phi-friday Jul 25, 2024
d35f4e2
fix: allow scipy constraints
phi-friday Jul 25, 2024
2640c6e
fix: revert ArrayLike
phi-friday Jul 25, 2024
ba14e10
fix
phi-friday Jul 25, 2024
2bea86c
fix: codecov
phi-friday Sep 8, 2024
84e2831
fix: deque, suggest
phi-friday Sep 8, 2024
3ce5298
fix: allow null target_func
phi-friday Sep 8, 2024
59605f5
fix: nullable ConstraintModel.fun
phi-friday Sep 8, 2024
c2e7b8f
fix: NonlinearConstraint
phi-friday Sep 8, 2024
d129a00
fix: docs
phi-friday Sep 8, 2024
8b0698c
fix: deps. errors
phi-friday Sep 8, 2024
c9ca71e
chore: revert numpy version
phi-friday Sep 8, 2024
5f7f098
tests: codecov
phi-friday Sep 8, 2024
1d96462
chore: rm myst-parse
phi-friday Sep 12, 2024
c6847c1
fix: review(only single point)
phi-friday Sep 12, 2024
4b8b74f
fix: review(params type)
phi-friday Sep 12, 2024
7b62feb
docs: add ext options comment
phi-friday Sep 12, 2024
de081cd
fix: rm unnecessary overload
phi-friday Sep 12, 2024
da6ba37
fix: rm whitespace
phi-friday Sep 12, 2024
1924116
fix: rm autodoc docstring hook
phi-friday Sep 12, 2024
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
136 changes: 91 additions & 45 deletions bayes_opt/acquisition.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,8 +23,7 @@
import abc
import warnings
from copy import deepcopy
from numbers import Number
from typing import TYPE_CHECKING, Callable
from typing import TYPE_CHECKING, Any, Literal, NoReturn

import numpy as np
from numpy.random import RandomState
Expand All @@ -41,8 +40,15 @@
from bayes_opt.target_space import TargetSpace

if TYPE_CHECKING:
from collections.abc import Callable

from numpy.typing import NDArray
from scipy.optimize import OptimizeResult

from bayes_opt.constraint import ConstraintModel

Float = np.floating[Any]


class AcquisitionFunction(abc.ABC):
"""Base class for acquisition functions.
Expand All @@ -53,7 +59,7 @@ class AcquisitionFunction(abc.ABC):
Set the random state for reproducibility.
"""

def __init__(self, random_state=None):
def __init__(self, random_state: int | RandomState | None = None) -> None:
if random_state is not None:
if isinstance(random_state, RandomState):
self.random_state = random_state
Expand All @@ -64,7 +70,7 @@ def __init__(self, random_state=None):
self.i = 0

@abc.abstractmethod
def base_acq(self, *args, **kwargs):
def base_acq(self, *args: Any, **kwargs: Any) -> NDArray[Float]:
"""Provide access to the base acquisition function."""

def _fit_gp(self, gp: GaussianProcessRegressor, target_space: TargetSpace) -> None:
Expand All @@ -80,10 +86,10 @@ def suggest(
self,
gp: GaussianProcessRegressor,
target_space: TargetSpace,
n_random=10_000,
n_l_bfgs_b=10,
n_random: int = 10_000,
n_l_bfgs_b: int = 10,
fit_gp: bool = True,
):
) -> NDArray[Float]:
"""Suggest a promising point to probe next.

Parameters
Expand Down Expand Up @@ -123,7 +129,9 @@ def suggest(
acq = self._get_acq(gp=gp, constraint=target_space.constraint)
return self._acq_min(acq, target_space.bounds, n_random=n_random, n_l_bfgs_b=n_l_bfgs_b)

def _get_acq(self, gp: GaussianProcessRegressor, constraint: ConstraintModel | None = None) -> Callable:
def _get_acq(
self, gp: GaussianProcessRegressor, constraint: ConstraintModel | None = None
) -> Callable[[NDArray[Float]], NDArray[Float]]:
"""Prepare the acquisition function for minimization.

Transforms a base_acq Callable, which takes `mean` and `std` as
Expand All @@ -148,25 +156,36 @@ def _get_acq(self, gp: GaussianProcessRegressor, constraint: ConstraintModel | N
dim = gp.X_train_.shape[1]
if constraint is not None:

def acq(x):
def acq(x: NDArray[Float]) -> NDArray[Float]:
x = x.reshape(-1, dim)
with warnings.catch_warnings():
warnings.simplefilter("ignore")
mean: NDArray[Float]
std: NDArray[Float]
p_constraints: NDArray[Float]
mean, std = gp.predict(x, return_std=True)
p_constraints = constraint.predict(x)
return -1 * self.base_acq(mean, std) * p_constraints
else:

def acq(x):
def acq(x: NDArray[Float]) -> NDArray[Float]:
x = x.reshape(-1, dim)
with warnings.catch_warnings():
warnings.simplefilter("ignore")
mean: NDArray[Float]
std: NDArray[Float]
mean, std = gp.predict(x, return_std=True)
return -1 * self.base_acq(mean, std)

return acq

def _acq_min(self, acq: Callable, bounds: np.ndarray, n_random=10_000, n_l_bfgs_b=10) -> np.ndarray:
def _acq_min(
self,
acq: Callable[[NDArray[Float]], NDArray[Float]],
bounds: NDArray[Float],
n_random: int = 10_000,
n_l_bfgs_b: int = 10,
) -> NDArray[Float]:
"""Find the maximum of the acquisition function.

Uses a combination of random sampling (cheap) and the 'L-BFGS-B'
Expand Down Expand Up @@ -200,13 +219,14 @@ def _acq_min(self, acq: Callable, bounds: np.ndarray, n_random=10_000, n_l_bfgs_
raise ValueError(error_msg)
x_min_r, min_acq_r = self._random_sample_minimize(acq, bounds, n_random=n_random)
x_min_l, min_acq_l = self._l_bfgs_b_minimize(acq, bounds, n_x_seeds=n_l_bfgs_b)
# Either n_random or n_l_bfgs_b is not 0 => at least one of x_min_r and x_min_l is not None
if min_acq_r < min_acq_l:
return x_min_r
return x_min_l

def _random_sample_minimize(
self, acq: Callable, bounds: np.ndarray, n_random: int
) -> tuple[np.ndarray, float]:
self, acq: Callable[[NDArray[Float]], NDArray[Float]], bounds: NDArray[Float], n_random: int
) -> tuple[NDArray[Float] | None, float]:
"""Random search to find the minimum of `acq` function.

Parameters
Expand Down Expand Up @@ -239,8 +259,8 @@ def _random_sample_minimize(
return x_min, min_acq

def _l_bfgs_b_minimize(
self, acq: Callable, bounds: np.ndarray, n_x_seeds: int = 10
) -> tuple[np.ndarray, float]:
self, acq: Callable[[NDArray[Float]], NDArray[Float]], bounds: NDArray[Float], n_x_seeds: int = 10
) -> tuple[NDArray[Float] | None, float]:
"""Random search to find the minimum of `acq` function.

Parameters
Expand Down Expand Up @@ -268,10 +288,12 @@ def _l_bfgs_b_minimize(
return None, np.inf
x_seeds = self.random_state.uniform(bounds[:, 0], bounds[:, 1], size=(n_x_seeds, bounds.shape[0]))

min_acq = None
min_acq: float | None = None
x_try: NDArray[Float]
x_min: NDArray[Float]
for x_try in x_seeds:
# Find the minimum of minus the acquisition function
res = minimize(acq, x_try, bounds=bounds, method="L-BFGS-B")
res: OptimizeResult = minimize(acq, x_try, bounds=bounds, method="L-BFGS-B")

# See if success
if not res.success:
Expand Down Expand Up @@ -317,7 +339,11 @@ class UpperConfidenceBound(AcquisitionFunction):
"""

def __init__(
self, kappa=2.576, exploration_decay=None, exploration_decay_delay=None, random_state=None
self,
kappa: float = 2.576,
exploration_decay: float | None = None,
exploration_decay_delay: int | None = None,
random_state: int | RandomState | None = None,
) -> None:
if kappa < 0:
error_msg = "kappa must be greater than or equal to 0."
Expand All @@ -328,7 +354,7 @@ def __init__(
self.exploration_decay = exploration_decay
self.exploration_decay_delay = exploration_decay_delay

def base_acq(self, mean, std):
def base_acq(self, mean: NDArray[Float], std: NDArray[Float]) -> NDArray[Float]:
"""Calculate the upper confidence bound.

Parameters
Expand All @@ -350,10 +376,10 @@ def suggest(
self,
gp: GaussianProcessRegressor,
target_space: TargetSpace,
n_random=10_000,
n_l_bfgs_b=10,
n_random: int = 10_000,
n_l_bfgs_b: int = 10,
fit_gp: bool = True,
) -> np.ndarray:
) -> NDArray[Float]:
"""Suggest a promising point to probe next.

Parameters
Expand Down Expand Up @@ -432,14 +458,20 @@ class ProbabilityOfImprovement(AcquisitionFunction):
Set the random state for reproducibility.
"""

def __init__(self, xi, exploration_decay=None, exploration_decay_delay=None, random_state=None) -> None:
def __init__(
self,
xi: float,
exploration_decay: float | None = None,
exploration_decay_delay: int | None = None,
random_state: int | RandomState | None = None,
) -> None:
super().__init__(random_state=random_state)
self.xi = xi
self.exploration_decay = exploration_decay
self.exploration_decay_delay = exploration_decay_delay
self.y_max = None

def base_acq(self, mean, std):
def base_acq(self, mean: NDArray[Float], std: NDArray[Float]) -> NDArray[Float]:
"""Calculate the probability of improvement.

Parameters
Expand Down Expand Up @@ -473,10 +505,10 @@ def suggest(
self,
gp: GaussianProcessRegressor,
target_space: TargetSpace,
n_random=10_000,
n_l_bfgs_b=10,
n_random: int = 10_000,
n_l_bfgs_b: int = 10,
fit_gp: bool = True,
) -> np.ndarray:
) -> NDArray[Float]:
"""Suggest a promising point to probe next.

Parameters
Expand Down Expand Up @@ -565,14 +597,20 @@ class ExpectedImprovement(AcquisitionFunction):
Set the random state for reproducibility.
"""

def __init__(self, xi, exploration_decay=None, exploration_decay_delay=None, random_state=None) -> None:
def __init__(
self,
xi: float,
exploration_decay: float | None = None,
exploration_decay_delay: int | None = None,
random_state: int | RandomState | None = None,
) -> None:
super().__init__(random_state=random_state)
self.xi = xi
self.exploration_decay = exploration_decay
self.exploration_decay_delay = exploration_decay_delay
self.y_max = None

def base_acq(self, mean, std):
def base_acq(self, mean: NDArray[Float], std: NDArray[Float]) -> NDArray[Float]:
"""Calculate the expected improvement.

Parameters
Expand Down Expand Up @@ -607,10 +645,10 @@ def suggest(
self,
gp: GaussianProcessRegressor,
target_space: TargetSpace,
n_random=10_000,
n_l_bfgs_b=10,
n_random: int = 10_000,
n_l_bfgs_b: int = 10,
fit_gp: bool = True,
) -> np.ndarray:
) -> NDArray[Float]:
"""Suggest a promising point to probe next.

Parameters
Expand Down Expand Up @@ -701,19 +739,24 @@ class ConstantLiar(AcquisitionFunction):
"""

def __init__(
self, base_acquisition: AcquisitionFunction, strategy="max", random_state=None, atol=1e-5, rtol=1e-8
self,
base_acquisition: AcquisitionFunction,
strategy: Literal["min", "mean", "max"] | float = "max",
random_state: int | RandomState | None = None,
atol: float = 1e-5,
rtol: float = 1e-8,
) -> None:
super().__init__(random_state)
self.base_acquisition = base_acquisition
self.dummies = []
if not isinstance(strategy, Number) and strategy not in ["min", "mean", "max"]:
if not isinstance(strategy, float) and strategy not in ["min", "mean", "max"]:
error_msg = f"Received invalid argument {strategy} for strategy."
raise ValueError(error_msg)
self.strategy = strategy
self.strategy: Literal["min", "mean", "max"] | float = strategy
self.atol = atol
self.rtol = rtol

def base_acq(self, *args, **kwargs):
def base_acq(self, *args: Any, **kwargs: Any) -> NDArray[Float]:
"""Calculate the acquisition function.

Calls the base acquisition function's `base_acq` method.
Expand Down Expand Up @@ -774,10 +817,10 @@ def suggest(
self,
gp: GaussianProcessRegressor,
target_space: TargetSpace,
n_random=10_000,
n_l_bfgs_b=10,
n_random: int = 10_000,
n_l_bfgs_b: int = 10,
fit_gp: bool = True,
) -> np.ndarray:
) -> NDArray[Float]:
"""Suggest a promising point to probe next.

Parameters
Expand Down Expand Up @@ -824,8 +867,9 @@ def suggest(
# Create a copy of the target space
dummy_target_space = self._copy_target_space(target_space)

dummy_target: float
# Choose the dummy target value
if isinstance(self.strategy, Number):
if isinstance(self.strategy, float):
dummy_target = self.strategy
elif self.strategy == "min":
dummy_target = target_space.target.min()
Expand Down Expand Up @@ -875,14 +919,16 @@ class GPHedge(AcquisitionFunction):
Set the random state for reproducibility.
"""

def __init__(self, base_acquisitions: list[AcquisitionFunction], random_state=None) -> None:
def __init__(
self, base_acquisitions: list[AcquisitionFunction], random_state: int | RandomState | None = None
) -> None:
super().__init__(random_state)
self.base_acquisitions = base_acquisitions
self.n_acq = len(self.base_acquisitions)
self.gains = np.zeros(self.n_acq)
self.previous_candidates = None

def base_acq(self, *args, **kwargs):
def base_acq(self, *args: Any, **kwargs: Any) -> NoReturn:
"""Raise an error, since the base acquisition function is ambiguous."""
msg = (
"GPHedge base acquisition function is ambiguous."
Expand All @@ -909,10 +955,10 @@ def suggest(
self,
gp: GaussianProcessRegressor,
target_space: TargetSpace,
n_random=10_000,
n_l_bfgs_b=10,
n_random: int = 10_000,
n_l_bfgs_b: int = 10,
fit_gp: bool = True,
) -> np.ndarray:
) -> NDArray[Float]:
"""Suggest a promising point to probe next.

Parameters
Expand Down
Loading
Loading