Monitoring callbacks

Early stopping callbacks

"""Monitor and influence the optimization procedure via callbacks.Callbacks are callables which are invoked after each iteration of the optimizerand are passed the results "so far". Callbacks can monitor progress, or stopthe optimization early by returning `True`.Monitoring callbacks--------------------* VerboseCallback* TimerCallbackEarly stopping callbacks------------------------* DeltaXStopper"""fromcollectionsimportCallablefromtimeimporttimeimportnumpyasnpfromskopt.utilsimportdumpdefcheck_callback(callback):""" Check if callback is a callable or a list of callables. """ifcallbackisnotNone:ifisinstance(callback,Callable):return[callback]elif(isinstance(callback,list)andall([isinstance(c,Callable)forcincallback])):returncallbackelse:raiseValueError("callback should be either a callable or ""a list of callables.")else:return[]classVerboseCallback(object):""" Callback to control the verbosity. Parameters ---------- * `n_init` [int, optional]: Number of points provided by the user which are yet to be evaluated. This is equal to `len(x0)` when `y0` is None * `n_random` [int, optional]: Number of points randomly chosen. * `n_total` [int]: Total number of func calls. Attributes ---------- * `iter_no`: [int]: Number of iterations of the optimization routine. """def__init__(self,n_total,n_init=0,n_random=0):self.n_init=n_initself.n_random=n_randomself.n_total=n_totalself.iter_no=1self._start_time=time()self._print_info(start=True)def_print_info(self,start=True):iter_no=self.iter_noifstart:status="started"eval_status="Evaluating function"search_status="Searching for the next optimal point."else:status="ended"eval_status="Evaluation done"search_status="Search finished for the next optimal point."ifiter_no<=self.n_init:print("Iteration No: %d%s. %s at provided point."%(iter_no,status,eval_status))elifself.n_init<iter_no<=(self.n_random+self.n_init):print("Iteration No: %d%s. %s at random point."%(iter_no,status,eval_status))else:print("Iteration No: %d%s. %s"%(iter_no,status,search_status))def__call__(self,res):""" Parameters ---------- * `res` [`OptimizeResult`, scipy object]: The optimization as a OptimizeResult object. """time_taken=time()-self._start_timeself._print_info(start=False)curr_y=res.func_vals[-1]curr_min=res.funprint("Time taken: %0.4f"%time_taken)print("Function value obtained: %0.4f"%curr_y)print("Current minimum: %0.4f"%curr_min)self.iter_no+=1ifself.iter_no<=self.n_total:self._print_info(start=True)self._start_time=time()classTimerCallback(object):""" Log the elapsed time between each iteration of the minimization loop. The time for each iteration is stored in the `iter_time` attribute which you can inspect after the minimization has completed. Attributes ---------- * `iter_time`: [list, shape=(n_iter,)]: `iter_time[i-1]` gives the time taken to complete iteration `i` """def__init__(self):self._time=time()self.iter_time=[]def__call__(self,res):""" Parameters ---------- * `res` [`OptimizeResult`, scipy object]: The optimization as a OptimizeResult object. """elapsed_time=time()-self._timeself.iter_time.append(elapsed_time)self._time=time()classEarlyStopper(object):"""Decide to continue or not given the results so far. The optimization procedure will be stopped if the callback returns True. """def__call__(self,result):""" Parameters ---------- * `result` [`OptimizeResult`, scipy object]: The optimization as a OptimizeResult object. """returnself._criterion(result)def_criterion(self,result):"""Compute the decision to stop or not. Classes inheriting from `EarlyStop` should use this method to implement their decision logic. Parameters ---------- * `result` [`OptimizeResult`, scipy object]: The optimization as a OptimizeResult object. Returns ------- * `decision`: Return True/False if the criterion can make a decision or `None` if there is not enough data yet to make a decision. """raiseNotImplementedError("The _criterion method should be implemented"" by subclasses of EarlyStopper.")classDeltaXStopper(EarlyStopper):"""Stop the optimization when |x1 - x2| < `delta` If the last two positions at which the objective has been evaluated are less than `delta` apart stop the optimization procedure. """def__init__(self,delta):super(EarlyStopper,self).__init__()self.delta=deltadef_criterion(self,result):iflen(result.x_iters)>=2:returnresult.space.distance(result.x_iters[-2],result.x_iters[-1])<self.deltaelse:returnNoneclassDeltaYStopper(EarlyStopper):"""Stop the optimization if the `n_best` minima are within `delta` Stop the optimizer if the absolute difference between the `n_best` objective values is less than `delta`. """def__init__(self,delta,n_best=5):super(EarlyStopper,self).__init__()self.delta=deltaself.n_best=n_bestdef_criterion(self,result):iflen(result.func_vals)>=self.n_best:func_vals=np.sort(result.func_vals)worst=func_vals[self.n_best-1]best=func_vals[0]# worst is always larger, so no need for abs()returnworst-best<self.deltaelse:returnNoneclassDeadlineStopper(EarlyStopper):""" Stop the optimization before running out of a fixed budget of time. Attributes ---------- * `iter_time`: [list, shape=(n_iter,)]: `iter_time[i-1]` gives the time taken to complete iteration `i` Parameters ---------- * `total_time`: fixed budget of time (seconds) that the optimization must finish within. """def__init__(self,total_time):super(DeadlineStopper,self).__init__()self._time=time()self.iter_time=[]self.total_time=total_timedef_criterion(self,result):elapsed_time=time()-self._timeself.iter_time.append(elapsed_time)self._time=time()ifresult.x_iters:time_remaining=self.total_time-np.sum(self.iter_time)returntime_remaining<=np.max(self.iter_time)else:returnNoneclassCheckpointSaver(object):""" Save current state after each iteration with `skopt.dump`. Example usage: import skopt checkpoint_callback = skopt.callbacks.CheckpointSaver("./result.pkl") skopt.gp_minimize(obj_fun, dims, callback=[checkpoint_callback]) Parameters ---------- * `checkpoint_path`: location where checkpoint will be saved to; * `dump_options`: options to pass on to `skopt.dump`, like `compress=9` """def__init__(self,checkpoint_path,**dump_options):self.checkpoint_path=checkpoint_pathself.dump_options=dump_optionsdef__call__(self,res):""" Parameters ---------- * `res` [`OptimizeResult`, scipy object]: The optimization as a OptimizeResult object. """dump(res,self.checkpoint_path,**self.dump_options)

Functions

defcheck_callback(callback):""" Check if callback is a callable or a list of callables. """ifcallbackisnotNone:ifisinstance(callback,Callable):return[callback]elif(isinstance(callback,list)andall([isinstance(c,Callable)forcincallback])):returncallbackelse:raiseValueError("callback should be either a callable or ""a list of callables.")else:return[]

Parameters

classDeadlineStopper(EarlyStopper):""" Stop the optimization before running out of a fixed budget of time. Attributes ---------- * `iter_time`: [list, shape=(n_iter,)]: `iter_time[i-1]` gives the time taken to complete iteration `i` Parameters ---------- * `total_time`: fixed budget of time (seconds) that the optimization must finish within. """def__init__(self,total_time):super(DeadlineStopper,self).__init__()self._time=time()self.iter_time=[]self.total_time=total_timedef_criterion(self,result):elapsed_time=time()-self._timeself.iter_time.append(elapsed_time)self._time=time()ifresult.x_iters:time_remaining=self.total_time-np.sum(self.iter_time)returntime_remaining<=np.max(self.iter_time)else:returnNone

classDeltaXStopper(EarlyStopper):"""Stop the optimization when |x1 - x2| < `delta` If the last two positions at which the objective has been evaluated are less than `delta` apart stop the optimization procedure. """def__init__(self,delta):super(EarlyStopper,self).__init__()self.delta=deltadef_criterion(self,result):iflen(result.x_iters)>=2:returnresult.space.distance(result.x_iters[-2],result.x_iters[-1])<self.deltaelse:returnNone

classDeltaYStopper(EarlyStopper):"""Stop the optimization if the `n_best` minima are within `delta` Stop the optimizer if the absolute difference between the `n_best` objective values is less than `delta`. """def__init__(self,delta,n_best=5):super(EarlyStopper,self).__init__()self.delta=deltaself.n_best=n_bestdef_criterion(self,result):iflen(result.func_vals)>=self.n_best:func_vals=np.sort(result.func_vals)worst=func_vals[self.n_best-1]best=func_vals[0]# worst is always larger, so no need for abs()returnworst-best<self.deltaelse:returnNone

Instance variables

classEarlyStopper(object):"""Decide to continue or not given the results so far. The optimization procedure will be stopped if the callback returns True. """def__call__(self,result):""" Parameters ---------- * `result` [`OptimizeResult`, scipy object]: The optimization as a OptimizeResult object. """returnself._criterion(result)def_criterion(self,result):"""Compute the decision to stop or not. Classes inheriting from `EarlyStop` should use this method to implement their decision logic. Parameters ---------- * `result` [`OptimizeResult`, scipy object]: The optimization as a OptimizeResult object. Returns ------- * `decision`: Return True/False if the criterion can make a decision or `None` if there is not enough data yet to make a decision. """raiseNotImplementedError("The _criterion method should be implemented"" by subclasses of EarlyStopper.")

Attributes

classTimerCallback(object):""" Log the elapsed time between each iteration of the minimization loop. The time for each iteration is stored in the `iter_time` attribute which you can inspect after the minimization has completed. Attributes ---------- * `iter_time`: [list, shape=(n_iter,)]: `iter_time[i-1]` gives the time taken to complete iteration `i` """def__init__(self):self._time=time()self.iter_time=[]def__call__(self,res):""" Parameters ---------- * `res` [`OptimizeResult`, scipy object]: The optimization as a OptimizeResult object. """elapsed_time=time()-self._timeself.iter_time.append(elapsed_time)self._time=time()