This tutorial shows how to implement a Bayesian Hyperparameter Optimization workflow with Hyperopt The Tree-structured Parzen Estimator algorithm (TPE). Hyperopt’s ability to handle hierarchical parameter graphs is demonstrated by constructing a conditional space which dynamically switches among different models families. Cross-validation is used to build an objective function that can be evaluated realistically. In addition, we implement early stopping on the basis of stagnating improvement in loss and thoroughly inspect the Trials objects to analyze optimization paths. We will not only understand Hyperopt’s internal tracking, evaluation, and refinement of the search process by the time we finish this tutorial but we will also have found the optimal model configuration. The framework is scalable, reproducible and can be extended for deeper learning or in distributed settings.
Install -U Hyperopt scikit learn pandas Matplotlib
import time
The Importance of Math
Numpy can be imported as np
import pandas as pd
Matplotlib.pyplot can be imported as a plt
Import load_breast_cancer from sklearn.datasets
from sklearn.model_selection import StratifiedKFold, cross_val_score
Import Pipeline from sklearn.pipeline
from sklearn.preprocessing import StandardScaler
Import LogisticRegression from sklearn.linear_model
SVC import sklearn.svm
From hyperopt, import tpe and hp. Trials. STATUS_OK.
Importation of hyperopt.pyll.base
Hyperopt.early_stop Import no_progress_loss
X, y = load_breast_cancer(return_X_y=True)
cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)Installation of dependencies is required to import the libraries needed for modeling and visualization. The Breast Cancer dataset is loaded and stratified cross validation prepared to achieve a balanced evaluation. It is the basis of our Bayesian structured optimization.
If you want to know more about this, click here."model_family", [
{
"model": "logreg",
"scaler": True,
"C": hp.loguniform("lr_C", np.log(1e-4), np.log(1e2)),
"penalty": hp.choice("lr_penalty", ["l2"]),
"solver": hp.choice("lr_solver", ["lbfgs", "liblinear"]),
"max_iter": scope.int(hp.quniform("lr_max_iter", 200, 2000, 50)),
"class_weight": hp.choice("lr_class_weight", [None, "balanced"]),
},
{
"model": "svm",
"scaler": True,
"kernel": hp.choice("svm_kernel", ["rbf", "poly"]),
"C": hp.loguniform("svm_C", np.log(1e-4), np.log(1e2)),
"gamma": hp.loguniform("svm_gamma", np.log(1e-6), np.log(1e0)),
"degree": scope.int(hp.quniform("svm_degree", 2, 5, 1)),
"class_weight": hp.choice("svm_class_weight", [None, "balanced"]),
}
])
This allows Hyperopt to choose between SVM and Logistic Regression. The search is tree-structured, with each branch having its own subspace of parameters. In order to avoid floating-point configuration errors, we have also used scope.int correctly for integer parameter conversion.
def build_pipeline(params: dict) -> Pipeline:
Steps = []
If params.get()"scaler", True):
steps.append(("scaler"StandardScaler()))
If params["model"] == "logreg":
Clf = LogisticRegression
C=float(params["C"]),
penalty=params["penalty"],
solver=params["solver"],
max_iter=int(params["max_iter"]),
class_weight=params["class_weight"],
n_jobs=None,
)
Element params["model"] == "svm":
kernel = params["kernel"]
Clf = (SVC)
kernel=kernel,
C=float(params["C"]),
gamma=float(params["gamma"]),
degree=int(params["degree"]) if kernel == "poly" Then 3,
class_weight=params["class_weight"],
probability=True,
)
else:
raise ValueError(f"Unknown model type: {params['model']}")
steps.append(("clf", clf))
return Pipeline(steps)
def objective(params: dict):
t0 = time.time()
try:
pipe = build_pipeline(params)
scores = cross_val_score(
pipe,
X, y,
cv=cv,
scoring="roc_auc",
n_jobs=-1,
error_score="raise",
)
mean_auc = float(np.mean(scores))
std_auc = float(np.std(scores))
loss = 1.0 - mean_auc
Elapsed = time.time.floatReturn() - t0)
return {
"loss": loss,
"status": STATUS_OK,
"attachments": {
"mean_auc": mean_auc,
"std_auc": std_auc,
"elapsed_sec": elapsed,
},
}
Exception to the rule:
Elapsed = time.time.floatReturn() - t0)
return {
"loss": 1.0,
"status": STATUS_FAIL,
"attachments": {
"error": repr(e),
"elapsed_sec": elapsed,
},
}Implement the constructor of the pipeline and the objective. We evaluate models using cross-validated ROC-AUC and convert the optimization problem into a minimization task by defining loss as 1 – mean_auc. Additionally, we attach structured metadata for each trial to enable richer post-optimization analyses.
trials = Trials()
rstate = np.random.default_rng(123)
max_evals = 80
The best is fmin (
fn=objective,
space=space,
algo=tpe.suggest,
max_evals=max_evals,
trials=trials,
rstate=rstate,
early_stop_fn=no_progress_loss(20),
)
print("nRaw `best` (note: includes choice indices):")
print(best)We can run TPE using fmin by specifying the number of maximum evaluations, and stopping early conditions. To ensure reproducibility we seed randomness and use a Trials object to track all evaluations. This snippet performs the entire Bayesian Search process.
def decode_best(space, best):
from hyperopt.pyll.stochastic import sample
fake = {}
def _fill(node):
Return node
cfg = sample(space, rng=np.random.default_rng(0))
Return Nonedecoded =
best_trial = trials.best_trial
best_params = best_trial["result"].get("attachments", {}).copy()
best_used_params = best_trial["misc"]["vals"].copy()
best_used_params = {k: (v[0] if isinstance(v, list) and len(v) else v) for k, v in best_used_params.items()}
MODEL_FAMILY = ["logreg", "svm"]
LR_PENALTY = ["l2"]
LR_SOLVER = ["lbfgs", "liblinear"]
LR_CLASS_WEIGHT = [None, "balanced"]
SVM_KERNEL = ["rbf", "poly"]
SVM_CLASS_WEIGHT = [None, "balanced"]
mf = int(best_used_params.get("model_family", 0))
decoded = {"model": MODEL_FAMILY[mf]}
if decoded["model"] == "logreg":
decoded.update({
"C": float(best_used_params["lr_C"]),
"penalty": LR_PENALTY[int(best_used_params["lr_penalty"])],
"solver": LR_SOLVER[int(best_used_params["lr_solver"])],
"max_iter": int(best_used_params["lr_max_iter"]),
"class_weight": LR_CLASS_WEIGHT[int(best_used_params["lr_class_weight"])],
"scaler": True,
})
else:
decoded.update({
"kernel": SVM_KERNEL[int(best_used_params["svm_kernel"])],
"C": float(best_used_params["svm_C"]),
"gamma": float(best_used_params["svm_gamma"]),
"degree": int(best_used_params["svm_degree"]),
"class_weight": SVM_CLASS_WEIGHT[int(best_used_params["svm_class_weight"])],
"scaler": True,
})
print("nDecoded best configuration:")
print(decoded)
print("nBest trial metrics:")
print(best_params)Decoding Hyperopt’s internal choices indices to human-readable configurations. In order to manually assign parameter names, we map hp.choice index values. It produces an easily interpreted best configuration.
The rows are equal to []
Trials for the t
Res = T.get("result", {})
att = res.get("attachments", {}) if isinstance(res, dict) else {}
Status = res.get"status"Then None if (res,dict) is not true.
loss = res.get("loss"Then None if (res,dict) is not true.
vals = t.get("misc", {}).get("vals", {})
vals = {k: (v[0] vals.items: if len() and isinstance(v list), otherwise None) for K, V.()}
rows.append({
"tid": t.get("tid"),
"status": status,
"loss": loss,
"mean_auc": att.get("mean_auc"),
"std_auc": att.get("std_auc"),
"elapsed_sec": att.get("elapsed_sec"),
**{f"p_{k}": v for k, v in vals.items()},
})
df = pd.DataFrame(rows).sort_values("tid").reset_index(drop=True)
print("nTop 10 trials by best loss:")
print(df[df["status"] == STATUS_OK].sort_values("loss").head(10)[
["tid", "loss", "mean_auc", "std_auc", "elapsed_sec", "p_model_family"]
])
OK = df[df["status"] == STATUS_OK].copy()
Ok["best_so_far"] = Ok["loss"].cummin()
plt.figure()
plt.plot(ok["tid"]Okay["loss"], marker="o", linestyle="none")
plt.xlabel("trial id")
plt.ylabel("loss = 1 - mean_auc")
plt.title("Trial losses")
plt.show()
plt.figure()
plt.plot(ok["tid"]Okay["best_so_far"])
plt.xlabel("trial id")
plt.ylabel("best-so-far loss")
plt.title("Best-so-far trajectory")
plt.show()
final_pipe = build_pipeline(decoded)
final_pipe.fit(X, y)
print("nFinal model fitted on full dataset.")
print(final_pipe)
print("nNOTE: SparkTrials is primarily useful on Spark/Databricks environments.")
print("Hyperopt SparkTrials docs exist, but Colab is typically not the right place for it.")The Trials object is transformed into a DataFrame that can be used for analysis. Visualizing loss progressions and the best performance so far helps us understand convergence behaviour. The final step is to train the model with the complete dataset. This confirms the final pipeline.
As a conclusion, using Hyperopt’s algorithm TPE we created a Bayesian Hyperparameter Optimization System that was fully structured. In this paper, we showed how to build conditional search space, use robust objective functions, perform early stopping and analyse trial metadata. Instead of treating hyperparameter optimization as a “black box”, we reveal and inspect each component in the optimization pipeline. Our framework is now scalable and extendable, allowing it to be easily adapted for gradient boosting, deep learning, reinforcement agents or Spark distributed environments. We achieved efficient model optimization for both production and research environments by combining intelligent sampling with structured search spaces.
Check out the Full Codes with Notebook here. Also, feel free to follow us on Twitter Join our Facebook group! 130k+ ML SubReddit Subscribe Now our Newsletter. Wait! Are you using Telegram? now you can join us on telegram as well.
Want to promote your GitHub repo, Hugging Face page, Product release or Webinar?? Connect with us
Post A Coding Implementation to Build a Conditional Bayesian Hyperparameter Optimization Pipeline with Hyperopt, TPE, and Early Stopping The first time that appeared on MarkTechPost.
