HyperImpute

The Imputers.get() method is the primary way to instantiate an imputer. It supports different optimization modes:

AutoML Mode (Hyperband)

Use optimizer="hyperband" to enable automated model selection. You can constrain the search space using regression_seed and classifier_seed.

Simple Mode

Use optimizer="simple" to bypass the hyperband optimization process, which is useful for testing specific model seeds without the overhead of hyperparameter optimization.

# AutoML with specific model seeds
model = imputers.get(
    "hyperimpute",
    optimizer="hyperband",
    regression_seed=["linear_regression"],
    classifier_seed=["logistic_regression"],
    select_model_by_column=False,
    select_model_by_iteration=False,
    select_lazy=True,
)

# Simple mode with specific model seeds
model = imputers.get(
    "hyperimpute",
    optimizer="simple",
    regression_seed=["random_forest_regressor"],
    classifier_seed=["random_forest"],
    select_model_by_column=False,
    select_model_by_iteration=False,
    select_lazy=True,
)

To simulate the effect of multiple imputations, you can iterate through different random seeds. For each seed, you can generate a new dataset split, simulate missingness (amputation), and apply an imputer plugin.

Key steps in the workflow:

1. Amputation: Use simulate_nan to create missing values based on a mechanism (e.g., 'MCAR') and a missingness probability p_miss.
2. Imputation: Retrieve a plugin using Imputers().get(plugin_name, random_state=seed) and call .fit_transform() on the incomplete data.
3. Evaluation: Compare the imputed data against the original ground truth using metrics like AUROC.

from hyperimpute.plugins.imputers import Imputers

# Configuration
plugin = "ice"
ampute_mechanism = "MCAR"
p_miss = 0.5

for seed in range(5):
    # 1. Setup data and simulate missingness
    # (Assuming dataset() and ampute() functions are defined as per tutorial)
    X_train, X_test, y_train, y_test = dataset(random_state=seed)
    x, x_miss, mask = ampute(X_train, ampute_mechanism, p_miss)

    # 2. Get the imputer plugin and impute
    model = Imputers().get(plugin, random_state=seed)
    X_train_imp = model.fit_transform(x_miss.copy()).astype(float)

    # 3. Evaluate (e.g., using an XGBoost classifier)
    # ... evaluation logic ...

To understand how HyperImpute scales, you can evaluate its performance on subsamples of increasing sizes. The following pattern demonstrates how to iterate through different sample sizes, run the evaluation, and cache the results in a JSON file to avoid redundant computations.

import json

results_by_df_size = {}
for subsample in [1000, 4000, 7000, 10000, 13000, 17000, 20000]:
    try:
        with open("general_results/error_by_df_size.json") as f:
            results_by_df_size = json.load(f)
    except BaseException:
        pass

    if str(subsample) in results_by_df_size:
        print("subsample already cached", subsample)
        continue

    print("subsample eval", subsample)
    X_local = X_raw.sample(subsample)
    y_local = y[X_local.index]

    results_by_df_size[subsample] = evaluate_dataset_repeated(
        f"subsample_{subsample}", X_local, y_local
    )

    with open("general_results/error_by_df_size.json", "w") as f:
        json.dump(results_by_df_size, f)

To evaluate how the number of features affects imputation performance, you can subsample a dataset and then progressively include more columns (features).

import json

results_by_feat_count = {}
for feat_count in [2, 4, 6, 8, 10, 12, 14]:
    try:
        with open("general_results/error_by_df_feat_count.json") as f:
            results_by_feat_count = json.load(f)
    except BaseException:
        pass

    if str(feat_count) in results_by_feat_count:
        print("feat count cached", feat_count)
        continue

    print("eval feat count", feat_count)
    X_local = X_raw.sample(2000)
    y_local = y[X_local.index]
    X_local = X_local.iloc[:, :feat_count]

    results_by_feat_count[feat_count] = evaluate_dataset_repeated(
        f"feature_subsample_{feat_count}", X_local, y_local
    )

    with open("general_results/error_by_df_feat_count.json", "w") as f:
        json.dump(results_by_feat_count, f)

You can use hyperimpute in conjunction with optuna to find the best imputer and hyperparameters for a given dataset. The workflow involves:

1. Identify Plugins: Use Imputers().list() to see available methods.
2. Define Objective: Create an objective function that uses plugin.sample_hyperparameters(trial) to suggest parameters.
3. Evaluate: Use imputer.fit_transform() and a metric like RMSE to score the results.
4. Optimize: Use create_study from hyperimpute.utils.optimizer to manage the Optuna study and handle early stopping.

from hyperimpute.plugins.imputers import Imputers
from hyperimpute.utils.optimizer import create_study
import optuna

imputers = Imputers()

def objective(trial):
    # Get the plugin class
    plugin_t = imputers.get_type('some_imputer')
    # Sample hyperparameters
    args = plugin_t.sample_hyperparameters(trial)
    # Instantiate and run
    imputer = plugin_t(**args)
    imputed = imputer.fit_transform(X_miss)
    # Return metric (e.g., RMSE)
    return score

study = create_study(study_name='imputer_opt', direction='minimize')
study.optimize(objective, n_trials=50)