from typing import Tuple, Dict, Union
import mocharts as mc
import numpy as np
import pandas as pd
from sklearn.model_selection import RandomizedSearchCV
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
[docs]
class ModelTuneRandomSearch:
"""
A class for performing hyperparameter tuning using random search.
"""
def __init__(self, dataset, model):
auth.run()
self.dataset = dataset
self.model = model
self.key = "model_tune_random_search"
[docs]
def run(self,
param_distributions: Dict,
dataset: str = "train",
n_iter: int = 10,
metric: Union[str, Tuple] = None,
n_jobs: int = None,
cv=None,
error_score=np.nan,
random_state: int = 0):
"""Random Search for model tuning.
This method performs hyperparameter optimization using random search on the specified model and dataset. It evaluates the model's performance based on the provided metrics and returns the results in a structured format.
Parameters
----------
param_distributions : dict
Dictionary with parameters names (str) as keys and distributions or lists of parameters to try. Distributions must provide a rvs method for sampling (such as those from scipy.stats.distributions). If a list is given, it is sampled uniformly. If a list of dicts is given, first a dict is sampled uniformly, and then a parameter is sampled using that dict as above.
dataset : {"main", "train", "test"}, default="train"
The data set for model fitting.
n_iter : int, default=10
Number of parameter settings that are sampled. 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
Number of jobs to run in parallel. None means 1 unless in a joblib.parallel_backend context. -1 means using all processors. See Glossary for more details.
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 random seed for reproducibility.
Returns
-------
ValidationResult
A container object with the following components:
- key: "model_tune_random_search"
- 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*/*random*.py
"""
inputs = locals()
inputs.pop('self')
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)
search = RandomizedSearchCV(estimator=self.model,
n_iter=n_iter,
param_distributions=param_distributions,
scoring=scoring,
cv=cv,
refit=False,
n_jobs=n_jobs,
error_score=error_score,
random_state=random_state)
search.fit(X=X, y=y.ravel(), sample_weight=sample_weight)
res_value = search.cv_results_
cols = ["param_" + col for col in param_distributions]
res_table = pd.DataFrame(search.cv_results_)[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.columns = ([col for col in param_distributions] +
["%s" % metric_name for metric_name in metric] +
["%s_rank" % metric_name for metric_name in metric] +
["Time"])
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 param_distributions:
group = res_table.groupby(param)
for metric_name in metric:
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 (Random 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,
inputs=inputs,
value=res_value,
table=res_table,
options=options_all)
return result