from typing import Union, Tuple, Dict
import mocharts as mc
import numpy as np
import pandas as pd
from scipy.stats import rv_continuous, rv_discrete
from scipy.stats._distn_infrastructure import rv_continuous_frozen, rv_discrete_frozen
from sklearn.model_selection._search import BaseSearchCV
from .utils import METRIC_DICT
from .utils import visualize_parallel_plot
from ...auth import auth
from ...utils.constants import (REGRESSION,
CLASSIFICATION,
REGRESSION_METRIC_DEFAULT,
CLASSIFICATION_METRIC_DEFAULT)
from ...utils.results import ValidationResult
class OptunaSearchCV(BaseSearchCV):
def __init__(self,
estimator,
param_distributions,
sampler,
*,
n_iter=10,
scoring=None,
n_jobs=None,
refit=True,
cv=None,
verbose=0,
pre_dispatch="2*n_jobs",
random_state=None,
error_score=np.nan,
return_train_score=False):
auth.run()
self.param_distributions = param_distributions
self.sampler = sampler
self.n_iter = n_iter
self.study = None
self.best_params_ = None
self.best_score_ = None
self.random_state = random_state
super().__init__(estimator=estimator,
scoring=scoring,
n_jobs=n_jobs,
refit=refit,
cv=cv,
verbose=verbose,
pre_dispatch=pre_dispatch,
error_score=error_score,
return_train_score=return_train_score)
def _suggest_param(self, trial, param_name, param_distribution):
EPS = 0.00001
if isinstance(param_distribution, list):
return trial.suggest_categorical(param_name, param_distribution)
elif isinstance(param_distribution, rv_continuous) or isinstance(param_distribution, rv_continuous_frozen):
return trial.suggest_float(param_name, param_distribution.ppf(EPS), param_distribution.ppf(1 - EPS))
elif isinstance(param_distribution, rv_discrete) or isinstance(param_distribution, rv_discrete_frozen):
return trial.suggest_int(param_name, param_distribution.ppf(EPS), param_distribution.ppf(1 - EPS))
else:
raise ValueError(f"Unsupported parameter type for {param_name}: {type(param_distribution)}")
def _run_search(self, evaluate_candidates):
try:
import optuna
except:
print('This API requires optuna. Please run "pip install optuna" in terminal.')
return
np.random.seed(self.random_state)
objective = "mean_test_%s" % list(self.scoring.keys())[0]
def objective_func(trial):
params = {k: self._suggest_param(trial, k, v) for k, v in self.param_distributions.items()}
return evaluate_candidates([params])[objective]
self.study = optuna.create_study(direction='maximize', sampler=self.sampler)
self.study.optimize(objective_func, n_trials=self.n_iter)
self.best_params_ = self.study.best_params
self.best_score_ = self.study.best_value
[docs]
class ModelTuneOptuna:
"""
A class for performing hyperparameter tuning using the optuna Python package.
"""
def __init__(self, dataset, model):
self.dataset = dataset
self.model = model
self.key = "model_tune_optuna"
[docs]
def run(self,
param_distributions: Dict,
dataset: str = "train",
n_iter: int = 10,
sampler: str = "tpe",
sampler_args: dict = None,
metric: Union[str, Tuple] = None,
n_jobs: int = None,
cv=None,
error_score=np.nan,
random_state: int = 0):
"""Runs model tuning using Optuna for hyperparameter optimization.
This method performs hyperparameter optimization for the specified model using the Optuna library.
It allows users to define parameter distributions, choose a sampling strategy, and specify evaluation metrics.
The results of the optimization process are returned in a structured format, including the best parameters and their corresponding scores.
Parameters
----------
param_distributions : dict
A dictionary where keys are parameter names (str) and values are distributions or lists of parameters to try. The distributions must provide a method for sampling (e.g., from `scipy.stats`), and if a list is provided, it will be sampled uniformly.
dataset : {"main", "train", "test"}, default="train"
Specifies which dataset to use for model fitting. Options include "main" for the entire dataset, "train" for the training subset, and "test" for the testing subset.
n_iter : int, default=10
The number of iterations for the optimization process, controlling the trade-off between runtime and the quality of the solution.
sampler : {"grid", "random", "tpe", "gp", "cma-es", "qmc"}, default="tpe"
The sampling strategy used in optuna.
- "grid" : Grid Search implemented in GridSampler
- "random" : Random Search implemented in RandomSampler
- "tpe" : Tree-structured Parzen Estimator algorithm implemented in TPESampler
- "gp" : Gaussian process-based algorithm implemented in GPSampler
- "cma-es": CMA-ES based algorithm implemented in CmaEsSampler
- "qmc" : A Quasi Monte Carlo sampling algorithm implemented in QMCSampler
sampler_args : dict, default=None
The arguments passed to the sampler.
dataset : {"main", "train", "test"}, default="train"
The data set for model fitting.
n_iter : int, default=10
Number of iterations of PSO. n_iter trades off runtime vs quality of the solution.
metric : str or tuple, default=None
The performance metric(s). If None,
- For classification: "ACC", "AUC", "F1", "LogLoss", "Precision", "Recall", and "Brier"
- For regression: "MSE", "MAE", and "R2"
Note that only the first one is used as the optimization objective.
cv : int, cross-validation generator or an iterable, default=None
Determines the cross-validation splitting strategy.
Possible inputs for cv are:
- None, to use the default 5-fold cross validation,
- integer, to specify the number of folds in a `(Stratified)KFold`,
- CV splitter,
- An iterable yielding (train, test) splits as arrays of indices.
n_jobs : int, default=None
The number of jobs to run in parallel. None means 1 unless in a joblib.parallel_backend context; -1 means using all processors.
error_score : 'raise' or numeric, default=np.nan
Value to assign to the score if an error occurs in estimator fitting.
If set to 'raise', the error is raised. If a numeric value is given,
FitFailedWarning is raised. This parameter does not affect the refit
step, which will always raise the error.
random_state : int, default=0
The seed used for random number generation to ensure reproducibility of results.
Returns
-------
ValidationResult
A container object with the following components:
- key: "model_tune_optuna"
- data: Name of the dataset used
- model: Name of the model used
- inputs: Input parameters
- value: Dictionary containing the optimization history
- table: Tabular format of the optimization history
- options: Dictionary of visualizations configuration. Run `results.plot()` to show all plots; Run `results.plot(name=xxx)` to display one preferred plot; and the following names are available:
- "parallel": Parallel plot of the hyperparameter settings and final performance.
- "(<parameter>, <metric>)": Bar plot showing the performance metric against parameter values.
Examples
--------
.. minigallery:: ../galleries/*/*hpo*/*optuna*.py
"""
inputs = locals()
inputs.pop('self')
try:
import optuna
except:
print('This API requires optuna. Please run "pip install optuna" in terminal.')
return
if metric is None:
if self.dataset.task_type == REGRESSION:
metric = [REGRESSION_METRIC_DEFAULT]
elif self.dataset.task_type == CLASSIFICATION:
metric = [CLASSIFICATION_METRIC_DEFAULT]
elif isinstance(metric, str):
metric = [metric]
scoring = {metric_name: METRIC_DICT[metric_name] for metric_name in metric}
X, y, sample_weight = self.dataset.get_X_y_data(dataset=dataset)
if sampler_args is None:
sampler_args = {}
if sampler == "grid":
sampler_obj = optuna.samplers.GridSampler(**sampler_args)
elif sampler == "random":
sampler_obj = optuna.samplers.RandomSampler(**sampler_args)
elif sampler == "tpe":
sampler_obj = optuna.samplers.TPESampler(**sampler_args)
elif sampler == "cma-es":
sampler_obj = optuna.samplers.CmaEsSampler(**sampler_args)
elif sampler == "gp":
sampler_obj = optuna.samplers.GPSampler(**sampler_args)
elif sampler == "qmc":
sampler_obj = optuna.samplers.QMCSampler(**sampler_args)
search = OptunaSearchCV(estimator=self.model,
param_distributions=param_distributions,
n_iter=n_iter,
sampler=sampler_obj,
scoring=scoring,
cv=cv,
refit=False,
n_jobs=n_jobs,
random_state=random_state)
search.fit(X=X, y=y.ravel(), sample_weight=sample_weight)
res_value = search.cv_results_
scalar_cols = ["_".join(col.split("_")[1:]) for col, item in search.cv_results_.items()
if col in ["param_" + key for key in list(param_distributions.keys())] and item.ndim == 1]
all_cols = (["param_" + col for col in scalar_cols] +
["mean_test_%s" % metric_name for metric_name in metric] +
["rank_test_%s" % metric_name for metric_name in metric] +
["mean_fit_time"])
res_table = pd.DataFrame({col: search.cv_results_[col] for col in all_cols})
res_table = res_table.rename(columns={"param_%s" % param: "%s" % param for param in scalar_cols})
res_table = res_table.rename(
columns={"mean_test_%s" % metric_name: "%s" % metric_name for metric_name in metric})
res_table = res_table.rename(
columns={"rank_test_%s" % metric_name: "%s_rank" % metric_name for metric_name in metric})
for metric_name in metric:
res_table[metric_name] = res_table[metric_name] * scoring[metric_name]._sign
res_table = res_table.sort_values("%s_rank" % metric[0])
options_all = {}
for param in scalar_cols:
group = res_table.groupby(param)
for metric_name in metric:
if isinstance(res_table[param].values[0], (list, tuple, np.ndarray)):
continue
options = mc.boxplot(x=res_table[param].values,
y=res_table[metric_name].values,
orient="vertical")
options.set_yaxis(axis_name=metric_name)
options.set_xaxis(axis_name=param)
options.set_title("HPO (Optuna Search)")
options.set_tooltip(precision=4)
options_all[(param, metric_name)] = options.render()
categorical_params_dict = {key: item for key, item in param_distributions.items()
if isinstance(item, (list, tuple)) and all(
not isinstance(v, (int, float)) for v in item)}
options_all["parallel"] = visualize_parallel_plot(search_result=res_value,
categorical_params_dict=categorical_params_dict,
metric_list=list(metric),
index=True)
result = ValidationResult(key=self.key,
data=self.dataset.name,
model=self.model.name,
value=res_value,
table=res_table,
options=options_all,
inputs=inputs)
return result