Coverage for moptipy / algorithms / so / vector / surrogate / rbf_interpolation.py: 96%
98 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-11-24 08:49 +0000
1"""
2A meta-algorithm for model-assisted optimization using SciPy's Interpolation.
4This algorithm wraps another numerical optimization algorithm `A` and proceeds
5in two stages.
6First, it samples and evaluates a set of initial points during the warmup
7phase. These points are directly sampled by `A` on the original process, the
8meta-algorithm just collects them.
10Then, in the second stage, for each iteration, a model is constructed from
11all previously sampled and evaluated points. The model is used for
12interpolating the actual objective function.
13In each step, the inner algorithm `A` is applied to this model. It strictly
14works on the model and does not invoke the original objective. Instead, we
15maintain the best point that `A` has sampled on the model based on the modeled
16objective function. This best point is then evaluated on the actual objective
17function. Together with its actual objective value, it is added to the set of
18evaluated points. In the next step, a new model will be constructed based on
19all the points we have now. This model is then the basis for the next
20"simulated" run of `A`, and so on.
22Thus, in the second stage, each execution of `A` on the model problem yields
23one new point that is actually evaluated. The new point is used to create a
24better model, and so on. If the models reflect the actual objective function
25well, this may allow us to achieve better overall solution qualities or to
26reduce the number of actual objective function evaluations to reach certain
27goals.
29However, this only works if a) we will not do too many actual objective
30function evaluations (FEs) overall, as the memory requirement grows
31quadratically with the number of FEs and b) if the dimensionality of the
32problem is not too high, as the number of points needed to create a reasonably
33accurate model rises with the dimensions of the search space.
34"""
36from math import inf
37from typing import Callable, Final
39import numpy as np
40from pycommons.types import check_int_range, type_error
41from scipy.interpolate import RBFInterpolator # type: ignore
42from scipy.special import comb # type: ignore
44from moptipy.algorithms.so.vector.surrogate._processes import (
45 _SurrogateApply,
46 _SurrogateWarmup,
47)
48from moptipy.api.algorithm import Algorithm, check_algorithm
49from moptipy.api.process import Process
50from moptipy.spaces.vectorspace import VectorSpace
51from moptipy.utils.logger import KeyValueLogSection
52from moptipy.utils.nputils import DEFAULT_FLOAT
53from moptipy.utils.strings import num_to_str_for_name, sanitize_names
55#: the permitted RBF kernels
56_RBF_KERNELS: Final[dict[str, str]] = {
57 "linear": "l",
58 "thin_plate_spline": "tps",
59 "cubic": "c",
60 "quintic": "q",
61}
64class RBFInterpolation(Algorithm):
65 """
66 A meta algorithm using an RBF-interpolation based surrogate model.
68 This algorithm uses :class:`scipy.interpolate.RBFInterpolator` as
69 interpolator surrogate model.
70 """
72 def __init__(self,
73 space: VectorSpace,
74 inner: Algorithm,
75 fes_for_warmup: int,
76 fes_per_interpolation: int,
77 kernel: str = "thin_plate_spline",
78 degree: int = 2,
79 name="RBF") -> None:
80 """
81 Create an interpolation-based surrogate algorithm.
83 :param name: the base name of this algorithm
84 :param inner: the algorithm to be applied in the inner optimization
85 loop
86 :param fes_for_warmup: the number of objective function evaluations to
87 be spent on the initial warmup period
88 :param fes_per_interpolation: the number of FEs to be performed
89 for each interpolation run
90 """
91 super().__init__()
93 if not isinstance(space, VectorSpace):
94 raise type_error(space, "space", VectorSpace)
95 if not isinstance(kernel, str):
96 raise type_error(kernel, "kernel", str)
97 if kernel not in _RBF_KERNELS:
98 raise ValueError(
99 f"kernel={kernel!r} not permitted, must be one "
100 f"of {_RBF_KERNELS.keys()}.")
102 degree = check_int_range(degree, "degree", -1, 20)
103 dimensions = check_int_range(
104 space.dimension, "space.dimensions", 1, 10_000)
105 fes_for_warmup = check_int_range(
106 fes_for_warmup, "fes_for_warmup", 1, 1_000_000_000)
107 if degree >= 0:
108 min_points: Final[int] \
109 = int(comb(degree + dimensions, dimensions, exact=True))
110 fes_for_warmup = max(fes_for_warmup, min_points)
112 #: the vector space
113 self._space = space
114 #: the inner algorithm
115 self.__inner = check_algorithm(inner)
116 #: the warmup FEs
117 self.__warmup_fes = fes_for_warmup
118 #: the FEs per interpolation run
119 self.__interpolation_fes = check_int_range(
120 fes_per_interpolation, "fes_per_interpolation", 1,
121 1_000_000_000_000)
122 #: the name of this surrogate assisted
123 self.__name = sanitize_names((
124 f"{name}{_RBF_KERNELS[kernel]}{num_to_str_for_name(degree)}",
125 str(fes_for_warmup), str(fes_per_interpolation), str(inner)))
126 #: the kernel name
127 self.__kernel: Final[str] = kernel
128 #: the degree
129 self.__degree: Final[int] = degree
131 def __str__(self) -> str:
132 """
133 Get the name of this surrogate-assisted algorithm.
135 :returns: the name of this surrogate assisted algorithm
136 """
137 return self.__name
139 def log_parameters_to(self, logger: KeyValueLogSection) -> None:
140 """
141 Log all parameters of this surrogate-assisted algorithm.
143 :param logger: the logger for the parameters
144 """
145 super().log_parameters_to(logger)
146 logger.key_value("warmupFEs", self.__warmup_fes)
147 logger.key_value("interpolationFEs", self.__interpolation_fes)
148 logger.key_value("kernel", self.__kernel)
149 logger.key_value("degree", self.__degree)
150 with logger.scope("inner") as inner:
151 self.__inner.log_parameters_to(inner)
152 with logger.scope("space") as space:
153 self._space.log_parameters_to(space)
155 def solve(self, process: Process) -> None:
156 """
157 Apply the surrogate-assisted optimization method to the given process.
159 :param process: the process to solve
160 """
161 # fast calls
162 should_terminate: Final[Callable[[], bool]] = process.should_terminate
163 inner: Final[Callable[[Process], None]] = self.__inner.solve
164 evaluate: Final[Callable[[np.ndarray], int | float]] = \
165 process.evaluate
166 init: Final[Callable[[], None]] = self.__inner.initialize
167 uniform: Final[Callable[[
168 np.ndarray, np.ndarray, int], np.ndarray]] = \
169 process.get_random().uniform
170 empty: Final[Callable[[int, np.dtype], np.ndarray]] \
171 = np.empty
173 # constants
174 lb: Final[np.ndarray] = self._space.lower_bound
175 ub: Final[np.ndarray] = self._space.upper_bound
176 dim: Final[int] = self._space.dimension
177 dtype: Final[np.dtype] = self._space.dtype
178 run_fes: Final[int] = self.__interpolation_fes
179 kernel: Final[str] = self.__kernel
180 degree: Final[int] = self.__degree
182# the containers for the points that we have sampled
183 x: Final[list[np.ndarray]] = [] # the points that were sampled so far
184 z: Final[list[int | float]] = [] # their objective values
186# Perform the initial warm-up process. Here, the inner algorithm will directly
187# be applied to the original problem. All the points that it samples are
188# collected and will later be used to construct the model.
189 with _SurrogateWarmup(process, self.__warmup_fes,
190 x.append, z.append) as p2:
191 p2._fes_left = p2.max_fes # type: ignore # store the budget
192 p2._terminated = False # type: ignore # not terminated yet
193 init() # initialize the inner algorithm
194 inner(p2) # apply the inner algorithm to the real model
195 del p2
197# Now, we have collected self.__warmup_fes points from the search space.
199 if should_terminate():
200 return
202# We can now perform the optimization on the model. The model is constructed
203# based on all points in the search space that were sampled and evaluated with
204# the actual objective function. In each iteration, we apply the inner
205# algorithm to the model from scratch. After it has terminated, then take the
206# best point it found (based on the modeled objective function) and evaluate
207# it with the actual objective function. This point and its objective value
208# are then added to the internal list and used, together with all previous
209# points, to construct the model for the next iteration.
210 model: Final[_SurrogateApply] = _SurrogateApply(process, run_fes)
212 while True:
213 while True:
214 # We always begin by building the surrogate model anew.
215 # However, this may sometimes fail. Maybe a parameter matrix
216 # becomes singular or whatever.
217 try:
218 f: Callable[[np.ndarray], np.ndarray] = \
219 RBFInterpolator(np.array(x, dtype=DEFAULT_FLOAT),
220 np.array(z, dtype=DEFAULT_FLOAT),
221 kernel=kernel,
222 degree=degree)
223 break # success: quit innermost loop
224 except: # noqa # pylint: disable=[W0702]
225 # If we get here, the model construction has failed.
226 # This means that the points that we have collected are
227 # somehow insufficient.
228 # We try to fix this by sampling one additional point
229 # uniformly at random and evaluate it.
230 # If this does not exhaust the FEs that we have, we can
231 # then try again.
232 tmp = uniform(lb, ub, dim)
233 x.append(tmp) # add random point to list of points
234 z.append(process.evaluate(tmp)) # and its objective value
235 if process.should_terminate(): # did we exhaust budget?
236 return # yes ... so we return
238 model._fes_left = run_fes # assign the budget for the run
239 model._terminated = False # the run has not terminated
240 model._evaluate = f # forward evaluation to the model
241 model._best_f = inf # no best-so-far solution exists yet
242 init() # initialize the inner algorithm
243 inner(model) # apply the inner algorithm to the model
244 tmp = empty(dim, dtype) # allocate holder for result
245 model.get_copy_of_best_x(tmp) # get best solution
246 z2 = evaluate(tmp) # evaluate it on the actual problem
247 if should_terminate(): # should we quit?
248 return # yes, so we return
249 x.append(tmp) # add the best solution to the list of points
250 z.append(z2) # and also add the objective value
251 del f # dispose old model