from typing import Union, Tuple, Dict
import mocharts as mc
import numpy as np
import pandas as pd
from sklearn.model_selection import GridSearchCV
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 ModelTuneGridSearch:
"""
A class for performing hyperparameter tuning using grid search.
"""
def __init__(self, dataset, model):
auth.run()
self.dataset = dataset
self.model = model
self.key = "model_tune_grid_search"
[docs]
def run(self,
param_grid: Dict,
dataset: str = "train",
metric: Union[str, Tuple] = None,
n_jobs: int = None,
cv=None,
error_score=np.nan):
"""Executes a grid search for model tuning.
This method performs hyperparameter optimization using grid 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_grid : dict
A dictionary where the keys are parameter names (str) and the values are lists of settings to try. Alternatively, it can be a list of such dictionaries to explore multiple grids.
dataset : {"main", "train", "test"}, default="train"
Specifies the dataset to be used for model fitting, with options for main, training, or testing datasets.
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
Defines the cross-validation strategy. It can be an integer specifying the number of folds, a CV splitter, or an iterable yielding (train, test) splits as arrays of indices.
n_jobs : int, default=None
The number of jobs to run in parallel. If None, it defaults to 1 unless in a joblib.parallel_backend context. -1 indicates 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.
Returns
-------
ValidationResult
A container object with the following components:
- key: "model_tune_grid_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*/*grid*.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 = GridSearchCV(estimator=self.model,
param_grid=param_grid,
scoring=scoring,
cv=cv,
refit=False,
n_jobs=n_jobs)
search.fit(X=X, y=y.ravel(), sample_weight=sample_weight)
res_value = search.cv_results_
cols = ["param_" + col for col in param_grid]
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_grid] +
["%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_grid:
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 (Grid Search)")
options.set_tooltip(precision=4)
options_all[(param, metric_name)] = options.render()
categorical_params_dict = {key: item for key, item in param_grid.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