Optuna HPI comparisons

hpi_evaluatior_comparison_lightgbm.py

import lightgbm as lgb
import numpy as np

import pandas as pd
import sklearn.datasets
from sklearn.datasets import fetch_openml
import sklearn.metrics
from sklearn.model_selection import train_test_split

import optuna

import utils


def objective(trial):
    data = fetch_openml(name="adult")
    X = data["data"]
    y = pd.Series(data["target"]).factorize()[0]

    train_x, test_x, train_y, test_y = train_test_split(X, y, test_size=0.25)
    dtrain = lgb.Dataset(train_x, label=train_y)

    param = {
        "objective": "binary",
        "metric": "binary_logloss",
        "verbosity": -1,
        "boosting_type": "gbdt",
        "lambda_l1": trial.suggest_loguniform("lambda_l1", 1e-8, 10.0),
        "lambda_l2": trial.suggest_loguniform("lambda_l2", 1e-8, 10.0),
        "num_leaves": trial.suggest_int("num_leaves", 2, 256),
        "feature_fraction": trial.suggest_uniform("feature_fraction", 0.4, 1.0),
        "bagging_fraction": trial.suggest_uniform("bagging_fraction", 1.4, 1.0),
        "bagging_freq": trial.suggest_int("bagging_freq", 1, 7),
        "min_child_samples": trial.suggest_int("min_child_samples", 5, 100),
    }

    gbm = lgb.train(param, dtrain)
    preds = gbm.predict(test_x)
    pred_labels = np.rint(preds)
    accuracy = sklearn.metrics.accuracy_score(test_y, pred_labels)
    return accuracy


if __name__ == "__main__":
    study = optuna.create_study(
        sampler=optuna.samplers.RandomSampler(),
        study_name="lightgbm-hpi",
        storage="sqlite:///lightgbm.db",
        load_if_exists=True,
        direction="maximize",
    )
    study.optimize(objective, n_trials=100)

    utils.save_benchmark(
        study, filename="{}_hpi.png".format(__file__.split(".")[0]), title="LightGBM"
    )

hpi_evaluatior_comparison_pytorch.py

import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
import torch.utils.data
from torchvision import datasets
from torchvision import transforms

import optuna

import utils

DEVICE = torch.device("cpu")
CLASSES = 10
DIR = os.getcwd()
EPOCHS = 20


def define_model(trial):
    layers = []

    in_features = 28 * 28
    for i in range(2):
        out_features = trial.suggest_int("n_units_l{}".format(i), 4, 128)
        activation = trial.suggest_categorical(
            "activation_l{}".format(i), choices=["relu", "tanh"]
        )
        p = trial.suggest_uniform("dropout_l{}".format(i), 0.2, 0.5)

        layers.append(nn.Linear(in_features, out_features))
        if activation == "relu":
            layers.append(nn.ReLU())
        elif activation == "tanh":
            layers.append(nn.Tanh())
        else:
            assert False
        layers.append(nn.Dropout(p))

        in_features = out_features
    layers.append(nn.Linear(in_features, CLASSES))
    layers.append(nn.LogSoftmax(dim=1))

    return nn.Sequential(*layers)


def get_mnist(trial):
    batch_size = trial.suggest_categorical("batch_size", choices=[8, 16, 32, 64, 128])

    train_loader = torch.utils.data.DataLoader(
        datasets.MNIST(DIR, train=True, download=True, transform=transforms.ToTensor()),
        batch_size=batch_size,
        shuffle=True,
    )
    test_loader = torch.utils.data.DataLoader(
        datasets.MNIST(DIR, train=False, transform=transforms.ToTensor()),
        batch_size=batch_size,
        shuffle=True,
    )
    return train_loader, test_loader


def objective(trial):
    model = define_model(trial).to(DEVICE)

    optimizer_name = trial.suggest_categorical("optimizer", ["Adam", "RMSprop", "SGD"])
    lr = trial.suggest_loguniform("lr", 1e-5, 1e-1)
    optimizer = getattr(optim, optimizer_name)(model.parameters(), lr=lr)

    train_loader, test_loader = get_mnist(trial)

    model.train()
    for epoch in range(EPOCHS):
        epoch_loss = 0
        for batch_idx, (data, target) in enumerate(train_loader):
            data, target = data.view(-1, 28 * 28).to(DEVICE), target.to(DEVICE)
            optimizer.zero_grad()
            output = model(data)
            loss = F.nll_loss(output, target)
            epoch_loss += loss.data
            loss.backward()
            optimizer.step()

        epoch_loss /= len(train_loader.dataset)
        print("Epoch {}: {}".format(epoch, epoch_loss))
        trial.report(epoch_loss, step=epoch)

    model.eval()
    correct = 0
    with torch.no_grad():
        for batch_idx, (data, target) in enumerate(test_loader):
            data, target = data.view(-1, 28 * 28).to(DEVICE), target.to(DEVICE)
            output = model(data)
            pred = output.argmax(dim=1, keepdim=True)
            correct += pred.eq(target.view_as(pred)).sum().item()

    accuracy = correct / len(test_loader.dataset)
    return accuracy


if __name__ == "__main__":
    study = optuna.create_study(
        sampler=optuna.samplers.RandomSampler(),
        study_name="pytorch-hpi",
        storage="sqlite:///pytorch.db",
        load_if_exists=True,
        direction="maximize",
    )
    study.optimize(objective, n_trials=100)

    utils.save_benchmark(
        study, filename="{}_hpi.png".format(__file__.split(".")[0]), title="PyTorch"
    )

utils.py

from collections import defaultdict
from collections import OrderedDict
import time
from typing import Dict
from typing import List

from matplotlib import pyplot as plt
import numpy as np
import pandas as pd

from optuna import importance
from optuna.importance._base import BaseImportanceEvaluator
from optuna.importance import get_param_importances
from optuna.study import Study


def get_all_importance_evaluators() -> List[BaseImportanceEvaluator]:
    # Edit this list to compare different evaluators.
    return [importance.FanovaImportanceEvaluator(), importance.MeanDecreaseImpurityImportanceEvaluator()]


def save_benchmark(study: Study, filename: str, title: str) -> None:
    evaluator_importances = {}

    for i, evaluator in enumerate(get_all_importance_evaluators()):
        start = time.time()
        importances = get_param_importances(study, evaluator=evaluator)
        end = time.time()
        print("Evaluation time", end - start, evaluator)
        evaluator_importances[evaluator.__class__.__name__ + str(i)] = importances

    save_multiple_importances(evaluator_importances, filename=filename, title=title)


def save_multiple_importances(
    evaluator_importances: Dict[str, Dict[str, float]], filename: str, title: str = None
) -> None:
    # Sort by evaluator names.
    evaluator_importances = OrderedDict(sorted(evaluator_importances.items(), key=lambda x: x[0]))

    index = []
    for evaluator_name, importances in evaluator_importances.items():
        for param_name in importances.keys():
            if param_name not in index:
                index.append(param_name)

    data = defaultdict(list)
    for evaluator_name, importances in evaluator_importances.items():
        for k, v in importances.items():
            print(k, v)
        importance_values = list(importances.values())
        importance_values = np.asarray(importance_values)
        tot_importance = importance_values.sum()
        for param_name in index:
            data[evaluator_name].append(importances[param_name] / tot_importance)

    df = pd.DataFrame(data, index=index)
    ax = df.plot.barh()
    ax.set_title(title)
    ax.set_xlabel("Importance")
    ax.set_ylabel("Parameter")
    ax.legend(loc="center left", bbox_to_anchor=(1, 0.5))
    plt.savefig(filename, bbox_inches="tight", dpi=100)

    print("Saved {}.".format(filename))