import logging
import os
from typing import Union, Tuple, Dict
import numpy as np
import pandas as pd
import pyswarms.backend as P
from pyswarms.backend.topology import Star
# remove logging to filesystem of pyswarms
logger = logging.getLogger()
for handler in logger.handlers[:]:
if handler.name == "file_default":
logger.removeHandler(handler)
import logging.config
logging.config.dictConfig({
'version': 1,
'disable_existing_loggers': True,
'loggers': {'file_default': {'level': 'INFO'}}
})
try:
if os.path.exists("report.log"):
os.remove("report.log")
except:
pass
from sklearn.model_selection._search import BaseSearchCV
import mocharts as mc
from .utils import METRIC_DICT
from .utils import visualize_parallel_plot
from ...utils.constants import (REGRESSION,
CLASSIFICATION,
REGRESSION_METRIC_DEFAULT,
CLASSIFICATION_METRIC_DEFAULT)
from ...utils.results import ValidationResult
from ...auth import auth
class PSOSearchCV(BaseSearchCV):
def __init__(
self,
estimator,
param_bounds,
*,
n_iter=10,
n_particles=10,
param_types=None,
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_bounds = param_bounds
self.param_types = param_types
self.n_iter = n_iter
self.n_particles = n_particles
self.random_state = random_state
self._init_pso()
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 _init_pso(self):
min_ = []
max_ = []
dimensions = 0
index_list = []
self.param_types = dict() if self.param_types is None else self.param_types
for key, param in self.param_bounds.items():
if key in self.param_types and self.param_types[key] == "categorical":
min_.append(-0.5)
max_.append(len(param) - 0.5)
else:
min_.append(param[0])
max_.append(param[1])
if isinstance(param[0], np.ndarray):
size = param[0].size
else:
size = 1
index_list.append(np.arange(dimensions, dimensions + size))
dimensions += size
self.dimensions = dimensions
self.index_list = index_list
self.bounds = (np.hstack(min_), np.hstack(max_))
self.options = {'c1': 0.3, 'c2': 0.5, 'w': 0.9}
def _get_params(self, position):
params = []
for param in position:
param_dict = {}
param = np.clip(param, self.bounds[0], self.bounds[1])
for i, (key, item) in enumerate(self.param_bounds.items()):
idx = self.index_list[i]
if key in self.param_types:
if key in self.param_types and self.param_types[key] == "int":
param_dict[key] = np.array(param[idx]).round().astype(int)
elif key in self.param_types and self.param_types[key] == "categorical":
param_dict[key] = np.array(item)[np.array(param[idx]).round().astype(int)]
else:
param_dict[key] = param[idx]
# convert to scaler if there is only one element
if param_dict[key].size == 1:
param_dict[key] = param_dict[key].item()
params.append(param_dict)
return params
def _run_search(self, evaluate_candidates):
"""Search n_iter candidates from param_bounds"""
np.random.seed(self.random_state)
objective = "mean_test_%s" % list(self.scoring.keys())[0]
my_topology = Star() # The Topology Class
my_swarm = P.create_swarm(n_particles=self.n_particles,
dimensions=self.dimensions,
options=self.options,
bounds=self.bounds)
for i in range(self.n_iter):
# Part 1: Update personal best
if self.verbose:
print("PSO iteration %d." % i)
res = evaluate_candidates(self._get_params(my_swarm.position))
my_swarm.current_cost = - res[objective][-my_swarm.n_particles:]
res = evaluate_candidates(self._get_params(my_swarm.pbest_pos))
my_swarm.pbest_cost = - res[objective][-my_swarm.n_particles:]
# Update and store
my_swarm.pbest_pos, my_swarm.pbest_cost = P.compute_pbest(my_swarm)
# Part 2: Update global best
# Note that gbest computation is dependent on your topology
if np.min(my_swarm.pbest_cost) < my_swarm.best_cost:
my_swarm.best_pos, my_swarm.best_cost = my_topology.compute_gbest(my_swarm)
# Part 3: Update position and velocity matrices
# Note that position and velocity updates are dependent on your topology
my_swarm.velocity = my_topology.compute_velocity(my_swarm)
my_swarm.position = my_topology.compute_position(my_swarm)
[docs]
class ModelTunePSO:
"""
A class for performing hyperparameter tuning using the PSO algorithm.
"""
def __init__(self, dataset, model):
self.dataset = dataset
self.model = model
self.key = "model_tune_pso"
[docs]
def run(self,
param_bounds: Dict,
param_types: Dict = None,
dataset: str = "train",
n_iter: int = 10,
n_particles: int = 10,
metric: Union[str, Tuple] = None,
n_jobs: int = None,
cv=None,
error_score=np.nan,
random_state: int = 0):
"""Particle Swarm Optimization (PSO) for model tuning.
This method performs hyperparameter optimization using PSO 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_bounds : dict
Dictionary with parameters names (str) as keys and lists of parameter min and max values (for numerical parameters) or list of parameter categories (for categorical parameters). For example,
.. code-block:: python
{'max_depth': [1, 5],
'n_estimators': [100, 200],
'eta': [0.01, 1.0],
'tree_method': ["exact", "approx", "hist"],
'centroids': [np.repeat(ds.train_x.min(0).reshape(1, -1), repeats=n_clusters, axis=0).ravel(),
np.repeat(ds.train_x.max(0).reshape(1, -1), repeats=n_clusters, axis=0).ravel()]
}
In this setting, parameters can also be an 1D array.
param_types : dict, default=None
Dictionary with parameters names (str) as keys and types of parameter. Available types include
"float", "int", and "categorical". For example,
.. code-block:: python
{'max_depth': "int",
'n_estimators': "int",
'tree_method': "categorical"}
Note that this is optional, and it may not contain all the keys shown in param_bounds; if a parameter's type is not specified, then it is treated as float.
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.
n_particles : int, default=10
Number of particles in each iteration of PSO.
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_pso"
- 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*/*pso*.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 = PSOSearchCV(estimator=self.model,
param_bounds=param_bounds,
param_types=param_types,
n_iter=n_iter,
n_particles=n_particles,
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_
scalar_cols = ["_".join(col.split("_")[1:]) for col, item in search.cv_results_.items()
if col in ["param_" + key for key in list(param_bounds.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 (PSO Search)")
options.set_tooltip(precision=4)
options_all[(param, metric_name)] = options.render()
categorical_params_dict = {key: item for key, item in param_bounds.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