Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_get_next_by_random_search_sorted(self,
patch_sample,
patch_ei,
patch_impute):
values = (10, 1, 9, 2, 8, 3, 7, 4, 6, 5)
patch_sample.return_value = [ConfigurationMock(i) for i in values]
patch_ei.return_value = np.array([[_] for _ in values], dtype=float)
patch_impute.side_effect = lambda l: values
cs = ConfigurationSpace()
ei = EI(None)
rs = RandomSearch(ei, cs)
rval = rs._maximize(
runhistory=None, stats=None, num_points=10, _sorted=True
)
self.assertEqual(len(rval), 10)
for i in range(10):
self.assertIsInstance(rval[i][1], ConfigurationMock)
self.assertEqual(rval[i][1].value, 10 - i)
self.assertEqual(rval[i][0], 10 - i)
self.assertEqual(rval[i][1].origin, 'Random Search (sorted)')
# Check that config.get_array works as desired and imputation is used
# in between, we therefore have to retrieve the value from the mock!
np.testing.assert_allclose([v.value for v in patch_ei.call_args[0][0]],
np.array(values, dtype=float))
def test_comp_builder(self):
conf = {"model": "RF", "acq_func": "EI", "y_transform": "y"}
model, acqf, rh2epm = self.acs._component_builder(conf)
self.assertIsInstance(acqf, EI)
self.assertIsInstance(model, RandomForestWithInstances)
self.assertIsInstance(rh2epm, AbstractRunHistory2EPM)
conf = {"model": "GP", "acq_func": "EI", "y_transform": "y"}
model, acqf, rh2epm = self.acs._component_builder(conf)
self.assertIsInstance(acqf, EI)
self.assertIsInstance(model, GaussianProcessMCMC)
self.assertIsInstance(rh2epm, AbstractRunHistory2EPM)
@unittest.mock.patch.object(EI, '__call__')
@unittest.mock.patch.object(ConfigurationSpace, 'sample_configuration')
def test_get_next_by_random_search_sorted(self,
patch_sample,
patch_ei,
patch_impute):
values = (10, 1, 9, 2, 8, 3, 7, 4, 6, 5)
patch_sample.return_value = [ConfigurationMock(i) for i in values]
patch_ei.return_value = np.array([[_] for _ in values], dtype=float)
patch_impute.side_effect = lambda l: values
cs = ConfigurationSpace()
ei = EI(None)
rs = RandomSearch(ei, cs)
rval = rs._maximize(
runhistory=None, stats=None, num_points=10, _sorted=True
)
self.assertEqual(len(rval), 10)
def test_init_only_scenario_quality(self):
epils = EPILS(self.scenario).solver
self.assertIsInstance(epils.model, RandomForestWithInstances)
self.assertIsInstance(epils.rh2EPM, RunHistory2EPM4Cost)
self.assertIsInstance(epils.acquisition_func, EI)
self.logger.warning("Predicted std is 0.0 for at least one sample.")
s_copy = np.copy(s)
s[s_copy == 0.0] = 1.0
f = calculate_f()
f[s_copy == 0.0] = 0.0
else:
f = calculate_f()
if (f < 0).any():
raise ValueError(
"Expected Improvement is smaller than 0 for at least one "
"sample.")
return f
class EIPS(EI):
def __init__(self,
model: AbstractEPM,
par: float=0.0):
r"""Computes for a given x the expected improvement as
acquisition value.
:math:`EI(X) := \frac{\mathbb{E}\left[ \max\{0, f(\mathbf{X^+}) - f_{t+1}(\mathbf{X}) - \xi\right] \} ]} {np.log(r(x))}`,
with :math:`f(X^+)` as the incumbent and :math:`r(x)` as runtime.
Parameters
----------
model : AbstractEPM
A model that implements at least
- predict_marginalized_over_instances(X) returning a tuples of
predicted cost and running time
par : float, default=0.0
Controls the balance between exploration and exploitation of the
cs.random = np.random.RandomState(seed=seed)
print("Run SMAC on %s with seed %d" %(benchmark.get_meta_information()["name"], seed))
print(smac_config)
# Build Configuration Space which defines all parameters and their ranges
# Scenario object
scenario = Scenario({"run_obj": "quality", # we optimize quality (alternatively runtime)
"runcount-limit": num_function_evals, # maximum function evaluations
"cs": cs, # configuration space
"deterministic": "true",
"output_dir": None
})
# string translations
acq = {"LCB": LCB, "EI": EI, "PI": PI}
r2e = {"InvScaled(y)": RunHistory2EPM4InvScaledCost,
"LogScaled(y)": RunHistory2EPM4LogScaledCost,
"y": RunHistory2EPM4Cost}
init_design = {"LHD": LHDesign,
"Sobol": SobolDesign,
"Default": DefaultConfiguration}
# Optimize, using a SMAC-object
print("Optimizing! Depending on your machine, this might take a few minutes.")
smac = SMAC(scenario=scenario,
rng=np.random.RandomState(seed),
tae_runner=wrapper,
initial_design=init_design[smac_config['init_design']],
initial_design_kwargs={'n_configs_x_params':smac_config['n_configs_x_params'],
'max_config_fracs':1.0},
random_configuration_chooser_kwargs={'prob': smac_config['rand_prob']},
def __init__(self,
model: AbstractEPM,
par: float=0.0):
"""Constructor
Parameters
----------
model : AbstractEPM
A model that implements at least
- predict_marginalized_over_instances(X)
par : float, default=0.0
Controls the balance between exploration and exploitation of the
acquisition function.
"""
super(EI, self).__init__(model)
self.long_name = 'Expected Improvement'
self.par = par
self.eta = None
cs.add_hyperparameters([x0, x1])
# Scenario object
scenario = Scenario({"run_obj": "quality", # we optimize quality (alternatively runtime)
"runcount-limit": 10, # max. number of function evaluations; for this example set to a low number
"cs": cs, # configuration space
"deterministic": "true"
})
# Example call of the function
# It returns: Status, Cost, Runtime, Additional Infos
def_value = rosenbrock_2d(cs.get_default_configuration())
print("Default Value: %.2f" % def_value)
# Optimize, using a SMAC-object
for acquisition_func in (LCB, EI, PI):
print("Optimizing with %s! Depending on your machine, this might take a few minutes." % acquisition_func)
smac = SMAC4HPO(scenario=scenario, rng=np.random.RandomState(42),
tae_runner=rosenbrock_2d,
initial_design=LHDesign,
initial_design_kwargs={'n_configs_x_params': 4,
'max_config_fracs': 1.0},
runhistory2epm=RunHistory2EPM4InvScaledCost,
acquisition_function_optimizer_kwargs={'max_steps': 100},
acquisition_function=acquisition_func,
acquisition_function_kwargs={'par': 0.01}
)
smac.optimize()
model = GaussianProcessMCMC(
types=types,
bounds=bounds,
kernel=kernel,
prior=prior,
n_hypers=n_hypers,
chain_length=200,
burnin_steps=100,
normalize_input=True,
normalize_output=True,
seed=self.rng.randint(low=0, high=10000),
)
if conf["acq_func"] == "EI":
acq = EI(model=model,
par=conf.get("par_ei", 0))
elif conf["acq_func"] == "LCB":
acq = LCB(model=model,
par=conf.get("par_lcb", 0))
elif conf["acq_func"] == "PI":
acq = PI(model=model,
par=conf.get("par_pi", 0))
elif conf["acq_func"] == "LogEI":
# par value should be in log-space
acq = LogEI(model=model,
par=conf.get("par_logei", 0))
return acq, model
types=types,
bounds=bounds,
kernel=kernel,
prior=prior,
n_hypers=n_hypers,
chain_length=200,
burnin_steps=100,
normalize_input=True,
normalize_output=True,
rng=self.rng,
)
else:
raise ValueError(conf['model'])
if conf["acq_func"] == "EI":
acq = EI(model=model, par=conf.get("par_ei", 0))
elif conf["acq_func"] == "LCB":
acq = LCB(model=model, par=conf.get("par_lcb", 0.05))
elif conf["acq_func"] == "PI":
acq = PI(model=model, par=conf.get("par_pi", 0))
elif conf["acq_func"] == "LogEI":
# par value should be in log-space
acq = LogEI(model=model, par=conf.get("par_logei", 0))
else:
raise ValueError(conf['acq_func'])
num_params = len(self.scenario.cs.get_hyperparameters())
success_states = [StatusType.SUCCESS, StatusType.CRASHED]
#TODO: only designed for black box problems without instances
if conf["y_transform"] == "y":
rh2epm = RunHistory2EPM4Cost(scenario=self.scenario,
num_params=num_params,