-
Notifications
You must be signed in to change notification settings - Fork 441
/
base_skopt_service.py
98 lines (83 loc) · 3.88 KB
/
base_skopt_service.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
import json
import numpy as np
import skopt
import logging
from pkg.suggestion.v1alpha3.internal.search_space import *
from pkg.suggestion.v1alpha3.internal.trial import *
logger = logging.getLogger("BaseSkoptService")
class BaseSkoptService(object):
"""
Refer to https://github.com/scikit-optimize/scikit-optimize .
"""
def __init__(self, algorithm_name="skopt-bayesian-optimization",
base_estimator="GP",
n_initial_points=10,
acq_func="gp_hedge",
acq_optimizer="auto",
random_state=None):
self.base_estimator = base_estimator
self.n_initial_points = n_initial_points
self.acq_func = acq_func
self.acq_optimizer = acq_optimizer
self.random_state = random_state
self.algorithm_name = algorithm_name
def getSuggestions(self, search_space, trials, request_number):
"""
Get the new suggested trials with skopt algorithm.
"""
skopt_search_space = []
for param in search_space.params:
if param.type == INTEGER:
skopt_search_space.append(skopt.space.Integer(
int(param.min), int(param.max), name=param.name))
elif param.type == DOUBLE:
skopt_search_space.append(skopt.space.Real(
float(param.min), float(param.max), "log-uniform", name=param.name))
elif param.type == CATEGORICAL or param.type == DISCRETE:
skopt_search_space.append(
skopt.space.Categorical(param.list, name=param.name))
if self.algorithm_name != "bayesianoptimization":
raise Exception(
'"Failed to create the algortihm: {}'.format(self.algorithm_name))
skopt_optimizer = skopt.Optimizer(skopt_search_space,
base_estimator=self.base_estimator,
n_initial_points=self.n_initial_points,
acq_func=self.acq_func,
acq_optimizer=self.acq_optimizer,
random_state=self.random_state)
for trial in trials:
skopt_suggested = []
for param in search_space.params:
parameter_value = None
for assignment in trial.assignments:
if assignment.name == param.name:
parameter_value = assignment.value
break
if param.type == INTEGER:
skopt_suggested.append(int(parameter_value))
elif param.type == DOUBLE:
skopt_suggested.append(float(parameter_value))
else:
skopt_suggested.append(parameter_value)
loss_for_skopt = float(trial.target_metric.value)
if search_space.goal == MAX_GOAL:
loss_for_skopt = -1 * loss_for_skopt
skopt_optimizer.tell(skopt_suggested, loss_for_skopt)
return_trial_list = []
for i in range(request_number):
skopt_suggested = skopt_optimizer.ask()
return_trial_list.append(
BaseSkoptService.convert(search_space, skopt_suggested))
return return_trial_list
@staticmethod
def convert(search_space, skopt_suggested):
assignments = []
for i in range(len(search_space.params)):
param = search_space.params[i]
if param.type == INTEGER:
assignments.append(Assignment(param.name, skopt_suggested[i]))
elif param.type == DOUBLE:
assignments.append(Assignment(param.name, skopt_suggested[i]))
elif param.type == CATEGORICAL or param.type == DISCRETE:
assignments.append(Assignment(param.name, skopt_suggested[i]))
return assignments