colosseum.hyperopt.selection

  1import csv
  2import os
  3from glob import glob
  4from typing import Dict, Type, Iterable, Callable, Collection
  5
  6import numpy as np
  7
  8from colosseum import config
  9from colosseum.agent.agents.base import BaseAgent
 10from colosseum.benchmark.utils import retrieve_agent_configs
 11from colosseum.utils.miscellanea import ensure_folder
 12
 13
 14def retrieve_best_agent_config_from_hp_folder(
 15    agent_classes: Iterable[Type["BaseAgent"]] = None,
 16    indicator="normalized_cumulative_regret",
 17    reduce_seeds: Callable[[Collection], float] = np.mean,
 18    folder: str = None,
 19) -> Dict[Type["BaseAgent"], str]:
 20    """
 21    retrieve the best agents configurations from a folder with the results of a hyperparameter optimization procedure.
 22    Note that. by default, the indicator is minimized. If you want to maximise the indicator you can pass a
 23    `reduce_seeds` function that inverts the sign of the indicators, e.g. `lambda x : -np.mean(x)`.
 24
 25    Parameters
 26    ----------
 27    agent_classes : Iterable[Type["BaseAgent"]]
 28        The agent classes for which the function retrieves the best config. By default, the agent classes are retrieved
 29         from the hyper_opt folder.
 30    indicator : str
 31        The code name of the performance indicator that will be used in the choice of the best parameters. Check
 32        `MDPLoop.get_indicators()` to get a list of the available indicators. By default, the indicator is the
 33        'normalized_cumulative_regret'.
 34    reduce_seeds : Callable[[Collection], float]
 35        The function that reduces the values of different seeds. By default, the average is used.
 36    folder : str
 37        The folder where the parameters optimization results are stored. By default, the one configured in the
 38        package is used.
 39
 40    Returns
 41    -------
 42    Dict[Type["BaseAgent"], str]
 43        A dictionary that associates to each agent class its best configuration.
 44    """
 45
 46    if folder is None:
 47        folder = config.get_hyperopt_folder()
 48    else:
 49        folder = ensure_folder(folder)
 50
 51    latest_hyprms_folder = folder + "latest_hyprms" + os.sep
 52
 53    # Retrive the agent classes from the folder if no agent classes is given
 54    if agent_classes is None:
 55        agent_classes = []
 56        if os.path.isdir(folder + "hyperopt_continuous"):
 57            agent_classes += list(
 58                retrieve_agent_configs(folder + "hyperopt_continuous").keys()
 59            )
 60        if os.path.isdir(folder + "hyperopt_episodic"):
 61            agent_classes += list(
 62                retrieve_agent_configs(folder + "hyperopt_episodic").keys()
 63            )
 64        assert len(agent_classes) > 0, f"No agent classes found in the {folder}"
 65
 66    agent_config = dict()
 67    for agent_class in agent_classes:
 68        current_hp_folder = (
 69            folder
 70            + "hyperopt_"
 71            + ("episodic" if agent_class.is_episodic() else "continuous")
 72            + os.sep
 73        )
 74
 75        if os.path.isfile(latest_hyprms_folder + agent_class.__name__ + ".gin"):
 76            with open(latest_hyprms_folder + agent_class.__name__ + ".gin", "r") as f:
 77                agent_config[agent_class] = f.read()
 78        elif os.path.isdir(current_hp_folder):
 79            agent_config[agent_class] = get_best_agent_gin_config(
 80                agent_class, current_hp_folder, indicator, reduce_seeds
 81            )
 82        else:
 83            raise FileNotFoundError(
 84                f"The hyperoptimization folder for {agent_class.__name__} could not be found "
 85                f"neither in the cache ({latest_hyprms_folder}) nor in the full logs ({current_hp_folder})"
 86            )
 87    return agent_config
 88
 89
 90def get_best_agent_gin_config(
 91    agent_class: Type["BaseAgent"],
 92    hp_folder: str,
 93    indicator="normalized_cumulative_regret",
 94    reduce_seeds: Callable[[Collection], float] = np.mean,
 95) -> str:
 96    """
 97    retrieve the best agents configurations from the folder with the results of a hyperparameter optimization procedure
 98    given an indicator. Note that. by default, the indicator is minimized. If you want to maximise the indicator you
 99    can pass a `reduce_seeds` function that inverts the sign of the indicators, e.g. `lambda x : -np.mean(x)`.
100
101    Parameters
102    ----------
103    agent_class : Type["BaseAgent"]
104        The agent class for which the function retrieves the best config.
105    hp_folder : str
106        The folder where the results of the parameters optimization procedure are located.
107    indicator : str
108        The code name of the performance indicator that will be used in the choice of the best parameters. Check
109        `MDPLoop.get_indicators()` to get a list of the available indicators. By default, the indicator is the
110        'normalized_cumulative_regret'.
111    reduce_seeds : Callable[[Collection], float]
112        The function that reduces the values of different seeds. By default, the average is used.
113    Returns
114    -------
115    str
116        The gin config of the best parameters.
117    """
118    agents_configs = retrieve_agent_configs(hp_folder, False)[agent_class]
119
120    prms_scores = dict()
121    for prm in agents_configs:
122        agent_prm_logs = glob(
123            hp_folder
124            + f"logs/*{prm}{config.EXPERIMENT_SEPARATOR_PRMS}{agent_class.__name__}/*.csv",
125            recursive=True,
126        )
127
128        scores = []
129        for l_f in agent_prm_logs:
130            with open(l_f) as f:
131                reader = csv.DictReader(f)
132                for p in reader:
133                    pass
134                scores.append(float(p[indicator]))
135        score = reduce_seeds(scores)
136        prms_scores[prm] = score
137    best_prms = min(prms_scores, key=lambda k: prms_scores[k])
138
139    return agents_configs[best_prms]