<commit_before>from abc import ABCMeta, abstractmethod
import numpy as np
import scipy.optimize
from scipy.stats import multivariate_normal
import random
from apsis.utilities.logging_utils import get_logger

class AcquisitionFunction(object):
    """
    An acquisition function is used to decide which point to evaluate next.

    For a detailed explanation, see for example "A Tutorial on Bayesian
    Optimization of Expensive Cost Functions, with Application to Active User
    Modeling and Hierarchical Reinforcement Learning", Brochu et.al., 2010

    Internally, each AcquisitionFunction implements a couple of max_searcher and
    multi_searcher functions. These work as follows:
    A max_searcher function takes the gp, experiment and (optionally) a
    good_proposals list. It then uses these to compute a proposal maximizing
    the acquisition function, and returns a tuple of this and its score.
    A multi_searcher function takes the gp, experiment, an (optional)
    good_proposals list and a maximum number of proposals. It then uses these
    to return several proposals in a list.
    Additionally, both max_searcher and multi_searcher functions have to return
    an own good_proposals list as a second return value (or None). These are
    evaluated proposals which are not in the first list, and are used to
    reuse computation.
    """

    _logger = None
    params = None
    LOG_FILE_NAME = "acquisition_functions.log"
    minimizes = True

    default_max_searcher = "random"
    default_multi_searcher = "random_weighted"

    def __init__(self, params=None):
        """
        Initializes the acquisition function.

        Parameters
        ----------
        params : dict or None, optional
            The dictionary of parameters defining the behaviour of the
            acquisition function. Supports at least max_searcher and
            multi_searcher.
        """
        self._logger = get_logger(self, specific_log_name=self.LOG_FILE_NAME)
        if params is None:
            params = {}
        self.params = params

    @abstractmethod
    def evaluate(self, x, gp, experiment):
        """
        Evaluates the gp on the point x.

        Parameters
        ----------
        x : dict
            Dictionary of the parameters on point x
        gp : GPy gp
            The gp on which to evaluate
        experiment : Experiment
            The experiment for further information.

        Returns
        -------
        eval : float
            The value of the acquisition function for the gp on point x.
        """
        pass

    def _compute_minimizing_evaluate(self, x, gp, experiment):
        """
        One problem is that, as a standard, scipy.optimize only searches
        minima. This means we have to convert each acquisition function to
        the minima meaning the best result.
        This the function to do so. Each compute_max can therefore just call
        this function, and know that the returned function has the best value
        as a global minimum.
        Whether to actually return the negative evaluate or not is set by the
        self.minimizes parameter. If true, it will not change the sign.

        Function signature is as evaluate.
        """
        value = self.evaluate(x, gp, experiment)
        if self.minimizes:
            return value
        else:
            return -value

    def compute_proposals(self, gp, experiment, number_proposals=1, return_max=True):
        """
        Computes up to number_proposals proposals.

        If return_max, the first entry in the returned list will be the
        proposal maximizing the acquisition function.

        Parameters
        ----------
        gp : GPy.gp
            The gp.

        experiment : experiment
            The current state of the experiment.

        number_proposals : int, optional
            The maximum number of proposals returned. The acquisition function
             will try its best to return that many proposals, but this cannot
             be guaranteed. By default, returns one proposal.

        return_max : bool, optional
            Whether to try and find the maximum of the proposal. If true (the
            default) will do another evaluation step).

        Returns
        -------
        proposals : list of tuples
            A list of tuples. The first entry of each is a dictionary with the
            parameter name and a 0-1 hypercube entry representing the warped
            parameter value. The second entry is the acquisition function score
            at that point. May return an empty list.
        """
        max_searcher = "none"
        multi_searcher = "none"
        if return_max:
            max_searcher = self.params.get("max_searcher", self.default_max_searcher)
            if number_proposals > 1:
                multi_searcher = self.params.get("multi_searcher", self.default_multi_searcher)
        else:
            multi_searcher = self.params.get("multi_searcher", self.default_multi_searcher)

        proposals = []

        good_results = []

        if max_searcher != "none":
            max_searcher = getattr(self, "max_searcher_" + max_searcher)
            max_prop, good_results_cur = max_searcher(gp, experiment)
            if good_results_cur is not None:
                good_results.extend(good_results_cur)
            proposals.append(max_prop)
        if multi_searcher != "none":
            multi_searcher = getattr(self, "multi_searcher_" + multi_searcher)
            multi_prop, good_results_cur = multi_searcher(gp, experiment,
                                                          good_results=good_results,
                                                          number_proposals=number_proposals-1)
            if good_results_cur is not None:
                good_results.extend(good_results_cur)
            proposals.extend(multi_prop)

        return proposals

    def max_searcher_random(self, gp, experiment, good_results=None):
        """
        Randomly searches the best result.

        Uses optimization_random_steps in self.params, with a default of 1000.

        For signature details see the introduction in the class docs.
        """
        if good_results is None:
            good_results = []
        optimization_random_steps = self.params.get("optimization_random_steps", 1000)

        evaluated_params = []

        best_param_idx = 0
        best_score = float("inf")

        for i in range(optimization_random_steps):
            param_dict_eval = self._gen_random_prop(experiment)
            score = self._compute_minimizing_evaluate(param_dict_eval, gp, experiment)
            if score < best_score:
                best_param_idx = i
                best_score = score<commit_msg>Started logging implementation in acquisition_function.<commit_after>from abc import ABCMeta, abstractmethod
import numpy as np
import scipy.optimize
from scipy.stats import multivariate_normal
import random
from apsis.utilities.logging_utils import get_logger


class AcquisitionFunction(object):
    """
    An acquisition function is used to decide which point to evaluate next.

    For a detailed explanation, see for example "A Tutorial on Bayesian
    Optimization of Expensive Cost Functions, with Application to Active User
    Modeling and Hierarchical Reinforcement Learning", Brochu et.al., 2010

    Internally, each AcquisitionFunction implements a couple of max_searcher
    and multi_searcher functions. These work as follows:
    A max_searcher function takes the gp, experiment and (optionally) a
    good_proposals list. It then uses these to compute a proposal maximizing
    the acquisition function, and returns a tuple of this and its score.
    A multi_searcher function takes the gp, experiment, an (optional)
    good_proposals list and a maximum number of proposals. It then uses these
    to return several proposals in a list.
    Additionally, both max_searcher and multi_searcher functions have to return
    an own good_proposals list as a second return value (or None). These are
    evaluated proposals which are not in the first list, and are used to
    reuse computation.
    """

    _logger = None
    params = None
    minimizes = True

    default_max_searcher = "random"
    default_multi_searcher = "random_weighted"

    def __init__(self, params=None):
        """
        Initializes the acquisition function.

        Parameters
        ----------
        params : dict or None, optional
            The dictionary of parameters defining the behaviour of the
            acquisition function. Supports at least max_searcher and
            multi_searcher.
        """
        self._logger = get_logger(self)
        self._logger.debug("Initializing acquisition function. params is %s",
                           params)
        if params is None:
            params = {}
        self.params = params

    @abstractmethod
    def evaluate(self, x, gp, experiment):
        """
        Evaluates the gp on the point x.

        Parameters
        ----------
        x : dict
            Dictionary of the parameters on point x
        gp : GPy gp
            The gp on which to evaluate
        experiment : Experiment
            The experiment for further information.

        Returns
        -------
        eval : float
            The value of the acquisition function for the gp on point x.
        """
        pass

    def _compute_minimizing_evaluate(self, x, gp, experiment):
        """
        One problem is that, as a standard, scipy.optimize only searches
        minima. This means we have to convert each acquisition function to
        the minima meaning the best result.
        This the function to do so. Each compute_max can therefore just call
        this function, and know that the returned function has the best value
        as a global minimum.
        Whether to actually return the negative evaluate or not is set by the
        self.minimizes parameter. If true, it will not change the sign.

        Function signature is as evaluate.
        """
        self._logger.debug("Computing minimizing evaluate. x is %s, gp is %s,"
                           "experiment is %s", x, gp, experiment)
        value = self.evaluate(x, gp, experiment)
        if self.minimizes:
            self._logger.debug("Is minimizing, returning %s", value)
            return value
        else:
            self._logger.debug("Is maximizing, returning %s", -value)
            return -value

    def compute_proposals(self, gp, experiment, number_proposals=1,
                          return_max=True):
        """
        Computes up to number_proposals proposals.

        If return_max, the first entry in the returned list will be the
        proposal maximizing the acquisition function.

        Parameters
        ----------
        gp : GPy.gp
            The gp.

        experiment : experiment
            The current state of the experiment.

        number_proposals : int, optional
            The maximum number of proposals returned. The acquisition function
             will try its best to return that many proposals, but this cannot
             be guaranteed. By default, returns one proposal.

        return_max : bool, optional
            Whether to try and find the maximum of the proposal. If true (the
            default) will do another evaluation step).

        Returns
        -------
        proposals : list of tuples
            A list of tuples. The first entry of each is a dictionary with the
            parameter name and a 0-1 hypercube entry representing the warped
            parameter value. The second entry is the acquisition function score
            at that point. May return an empty list.
        """
        self._logger.debug("Computing proposals. gp is %s, experiment is %s, "
                           "number_proposals %s, return_max %s",
                           gp, experiment, number_proposals, return_max)
        max_searcher = "none"
        multi_searcher = "none"
        if return_max:
            max_searcher = self.params.get("max_searcher",
                                           self.default_max_searcher)
            self._logger.debug("Returning maximum. Max_searcher is %s",
                               max_searcher)
            if number_proposals > 1:
                multi_searcher = self.params.get("multi_searcher",
                                                 self.default_multi_searcher)
                self._logger.debug("Also returning multiple results; "
                                   "multi_searcher is %s", multi_searcher)
        else:
            multi_searcher = self.params.get("multi_searcher",
                                             self.default_multi_searcher)
            self._logger.debug("Not returning max. multi_searcher is %s",
                               multi_searcher)

        proposals = []

        good_results = []

        if max_searcher != "none":
            self._logger.debug("Beginning to look for max.")
            max_searcher = getattr(self, "max_searcher_" + max_searcher)
            self._logger.debug("Constructed max_search function. max_searcher "
                               "is %s", max_searcher)
            max_prop, good_results_cur = max_searcher(gp, experiment)
            self._logger.debug("Finished search. max_prop is %s, "
                               "good_results_cur is %s", max_prop,
                               good_results)
            if good_results_cur is not None:
                good_results.extend(good_results_cur)
                self._logger.debug("Extended known good results. Now %s",
                                   good_results)
            proposals.append(max_prop)
            self._logger.debug("Appended %s to max_prop, now %s", max_prop,
                               proposals)
        if multi_searcher != "none":
            self._logger.debug("Starting multi_search.")
            multi_searcher = getattr(self, "multi_searcher_" + multi_searcher)
            self._logger.debug("multi_searcher function generated as %s",
                               multi_searcher)
            multi_prop, good_results_cur = multi_searcher(gp, experiment,
                                          good_results=good_results,
                                          number_proposals=number_proposals-1)
            self._logger.debug("Finished multi search. Multi_prop is %s, "
                               "good_results_cur is %s", multi_prop,
                               good_results_cur)
            if good_results_cur is not None:
                good_results.extend(good_results_cur)
                self._logger.debug("Appended (extend) %s to good_results. Now"
                                   " %s", good_results_cur, good_results)
            proposals.extend(multi_prop)
            self._logger.debug("Appended %s to proposals; now %s",
                               multi_prop, proposals)
        self._logger.debug("Returning proposals %s", proposals)
        return proposals

    def max_searcher_random(self, gp, experiment, good_results=None):
        """
        Randomly searches the best result.

        Uses optimization_random_steps in self.params, with a default of 1000.

        For signature details see the introduction in the class docs.
        """
        if good_results is None:
            good_results = []
        optimization_random_steps = self.params.get("optimization_random_steps", 1000)

        evaluated_params = []

        best_param_idx = 0
        best_score = float("inf")

        for i in range(optimization_random_steps):
            param_dict_eval = self._gen_random_prop(experiment)
            score = self._compute_minimizing_evaluate(param_dict_eval, gp, experiment)
            if score < best_score:
                best_param_idx = i
                best_score = score
{'fraction_non_alphanumeric': 0.060611747044473634, 'fraction_numerical': 0.002251829611559392, 'mean_word_length': 1.9634847080630213, 'pattern_counts': {'":': 4, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 19, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3951718', 'n_tokens_mistral': 4002, 'n_tokens_neox': 3920, 'n_words': 1436}