async_optimization_dummies.py

"""Originally by @rhizhiy
https://github.com/bayesian-optimization/BayesianOptimization/issues/347#issuecomment-1273465096
"""
importtime
fromtypingimportCallable

importmatplotlib.pyplotasplt
importnumpyasnp
frombayes_optimportBayesianOptimization
frombayes_optimportacquisition
fromscipy.optimizeimportrosen

ASYNC_METHOD='lie_max'
assertASYNC_METHODin ['lie_max', 'none']
def_closest_distance(point, points):
returnmin(np.linalg.norm(point-p) forpinpointsifpisnotpoint)


defoptimize(
func: Callable[..., float], num_iter: int, bounds: dict[str, tuple[float, float]], num_workers=0
):
init_samples=int(np.sqrt(num_iter))
init_kappa=10
kappa_decay= (0.1/init_kappa) ** (1/num_iter)

acquisition_function=acquisition.UpperConfidenceBound(
kappa=init_kappa,
exploration_decay=kappa_decay,
exploration_decay_delay=0
 )

ifASYNC_METHOD=='lie_max':
acquisition_function=acquisition.ConstantLiar(acquisition_function, 'max')


optimizer=BayesianOptimization(
f=None,
acquisition_function=acquisition_function,
pbounds=bounds,
verbose=0
 )


init_queue= [optimizer.suggest() for_inrange(init_samples)]
result_queue= []
whilelen(optimizer.res) <num_iter:
sample=init_queue.pop(0) ifinit_queueelseoptimizer.suggest()
loss=func(list(sample.values())) *-1
result_queue.append((sample, loss))
iflen(result_queue) >=num_workers:
optimizer.register(*result_queue.pop(0))
returnoptimizer.res


bounds= {"x": [-5, 5], "y": [-5, 5]}

all_times= {}
all_results= {}
workers_each= [1, 2, 4, 8,]# 16]
print(f"Simulating parallel optimization for {workers_each} workers, this can take some time.")
print(f"Async method: {ASYNC_METHOD}.")
fornum_workersinworkers_each:
print(f"\tChecking {num_workers} workers")
results= []
start=time.perf_counter()
results=optimize(rosen, 200, bounds, num_workers)
end=time.perf_counter()
delta=end-start
all_times[num_workers] =delta
samples= [res["params"] forresinresults]
all_results[num_workers] =samples

fig, axs=plt.subplots(2, 2)
ifASYNC_METHOD=='lie_max':
acquisition_function_str="Constant Max Liar (UCB)"
else:
acquisition_function_str="UCB"

fig.suptitle(f"Acquisition function: {acquisition_function_str}")
fig.set_figheight(8)
fig.set_figwidth(8)
axs= [itemforsublistinaxsforiteminsublist]
foridx, (num_workers, samples) inenumerate(all_results.items()):
ifnum_workers>8:
continue
samples= [np.array(list(sample.values())) forsampleinsamples]
axs[idx].scatter(*zip(*samples), s=1)
axs[idx].set_title(f"{num_workers=}")
avg_min_distance=np.mean([_closest_distance(sample, samples) forsampleinsamples])
print(f"{num_workers=}, mean_min_distance={avg_min_distance:.3f}, time={all_times[num_workers]:.3f}")
fig.tight_layout()
plt.savefig(f"corrected_async_{ASYNC_METHOD}.png")