in bayesmark/experiment.py [0:0]
def run_study(optimizer, test_problem, n_calls, n_suggestions, n_obj=1, callback=None):
"""Run a study for a single optimizer on a single test problem.
This function can be used for benchmarking on general stateless objectives (not just `sklearn`).
Parameters
----------
optimizer : :class:`.abstract_optimizer.AbstractOptimizer`
Instance of one of the wrapper optimizers.
test_problem : :class:`.sklearn_funcs.TestFunction`
Instance of test function to attempt to minimize.
n_calls : int
How many iterations of minimization to run.
n_suggestions : int
How many parallel evaluation we run each iteration. Must be ``>= 1``.
n_obj : int
Number of different objectives measured, only objective 0 is seen by optimizer. Must be ``>= 1``.
callback : callable
Optional callback taking the current best function evaluation, and the number of iterations finished. Takes
array of shape `(n_obj,)`.
Returns
-------
function_evals : :class:`numpy:numpy.ndarray` of shape (n_calls, n_suggestions, n_obj)
Value of objective for each evaluation.
timing_evals : (:class:`numpy:numpy.ndarray`, :class:`numpy:numpy.ndarray`, :class:`numpy:numpy.ndarray`)
Tuple of 3 timing results: ``(suggest_time, eval_time, observe_time)`` with shapes ``(n_calls,)``,
``(n_calls, n_suggestions)``, and ``(n_calls,)``. These are the time to make each suggestion, the time for each
evaluation of the objective function, and the time to make an observe call.
suggest_log : list(list(dict(str, object)))
Log of the suggestions corresponding to the `function_evals`.
"""
assert n_suggestions >= 1, "batch size must be at least 1"
assert n_obj >= 1, "Must be at least one objective"
space_for_validate = JointSpace(test_problem.get_api_config())
if callback is not None:
# First do initial log at inf score, in case we don't even get to first eval before crash/job timeout
callback(np.full((n_obj,), np.inf, dtype=float), 0)
suggest_time = np.zeros(n_calls)
observe_time = np.zeros(n_calls)
eval_time = np.zeros((n_calls, n_suggestions))
function_evals = np.zeros((n_calls, n_suggestions, n_obj))
suggest_log = [None] * n_calls
for ii in range(n_calls):
tt = time()
try:
next_points = optimizer.suggest(n_suggestions)
except Exception as e:
logger.warning("Failure in optimizer suggest. Falling back to random search.")
logger.exception(e, exc_info=True)
print(json.dumps({"optimizer_suggest_exception": {ITER: ii}}))
api_config = test_problem.get_api_config()
next_points = rs.suggest_dict([], [], api_config, n_suggestions=n_suggestions)
suggest_time[ii] = time() - tt
logger.info("suggestion time taken %f iter %d next_points %s" % (suggest_time[ii], ii, str(next_points)))
assert len(next_points) == n_suggestions, "invalid number of suggestions provided by the optimizer"
# We could put this inside the TestProblem class, but ok here for now.
try:
space_for_validate.validate(next_points) # Fails if suggestions outside allowed range
except Exception:
raise ValueError("Optimizer suggestion is out of range.")
for jj, next_point in enumerate(next_points):
tt = time()
try:
f_current_eval = test_problem.evaluate(next_point)
except Exception as e:
logger.warning("Failure in function eval. Setting to inf.")
logger.exception(e, exc_info=True)
f_current_eval = np.full((n_obj,), np.inf, dtype=float)
eval_time[ii, jj] = time() - tt
assert np.shape(f_current_eval) == (n_obj,)
suggest_log[ii] = next_points
function_evals[ii, jj, :] = f_current_eval
logger.info(
"function_evaluation time %f value %f suggestion %s"
% (eval_time[ii, jj], f_current_eval[0], str(next_point))
)
# Note: this could be inf in the event of a crash in f evaluation, the optimizer must be able to handle that.
# Only objective 0 is seen by optimizer.
eval_list = function_evals[ii, :, 0].tolist()
if callback is not None:
idx_ii, idx_jj = argmin_2d(function_evals[: ii + 1, :, 0])
callback(function_evals[idx_ii, idx_jj, :], ii + 1)
tt = time()
try:
optimizer.observe(next_points, eval_list)
except Exception as e:
logger.warning("Failure in optimizer observe. Ignoring these observations.")
logger.exception(e, exc_info=True)
print(json.dumps({"optimizer_observe_exception": {ITER: ii}}))
observe_time[ii] = time() - tt
logger.info(
"observation time %f, current best %f at iter %d"
% (observe_time[ii], np.min(function_evals[: ii + 1, :, 0]), ii)
)
return function_evals, (suggest_time, eval_time, observe_time), suggest_log