필사 모드: AutoML 完全指南：用 AutoGluon、FLAML、Optuna 打造自动化 ML 流水线

from sklearn.model_selection import GridSearchCV
import xgboost as xgb

def grid_search_example(X_train, y_train):
    """网格搜索：完全搜索(只适合较小的参数空间)"""
    param_grid = {
        'max_depth': [3, 5, 7],
        'learning_rate': [0.01, 0.1, 0.3],
        'n_estimators': [100, 300, 500],
        'subsample': [0.7, 0.9],
    }
    # 总组合数：3 * 3 * 3 * 2 = 54 种 x CV 折数

    model = xgb.XGBClassifier(random_state=42, n_jobs=-1)
    grid_search = GridSearchCV(
        model, param_grid,
        cv=5,
        scoring='roc_auc',
        n_jobs=-1,
        verbose=1,
        refit=True  # 用最优参数在全部数据上重新训练
    )
    grid_search.fit(X_train, y_train)

    print(f"最优参数: {grid_search.best_params_}")
    print(f"最优 CV 分数: {grid_search.best_score_:.4f}")

    # 用 DataFrame 分析结果
    results = pd.DataFrame(grid_search.cv_results_)
    results_sorted = results.sort_values('mean_test_score', ascending=False)
    print(results_sorted[['params', 'mean_test_score', 'std_test_score']].head(10))

    return grid_search.best_estimator_

from sklearn.model_selection import RandomizedSearchCV
from scipy.stats import uniform, randint, loguniform

def random_search_example(X_train, y_train, n_iter=100):
    """随机搜索：从连续分布中采样，实现高效搜索"""
    param_distributions = {
        'max_depth': randint(3, 10),
        'learning_rate': loguniform(1e-3, 0.5),   # 对数均匀分布
        'n_estimators': randint(100, 1000),
        'subsample': uniform(0.6, 0.4),           # [0.6, 1.0]
        'colsample_bytree': uniform(0.6, 0.4),
        'reg_alpha': loguniform(1e-4, 10),
        'reg_lambda': loguniform(1e-4, 10),
        'min_child_weight': randint(1, 10),
        'gamma': uniform(0, 0.5),
    }

    model = xgb.XGBClassifier(random_state=42, n_jobs=-1)
    random_search = RandomizedSearchCV(
        model, param_distributions,
        n_iter=n_iter,   # 尝试的组合数
        cv=5,
        scoring='roc_auc',
        n_jobs=-1,
        verbose=1,
        random_state=42,
        refit=True
    )
    random_search.fit(X_train, y_train)

    print(f"最优参数: {random_search.best_params_}")
    print(f"最优 CV 分数: {random_search.best_score_:.4f}")

    return random_search.best_estimator_

pip install optuna optuna-dashboard

import optuna
from optuna.samplers import TPESampler, CmaEsSampler, RandomSampler
from optuna.pruners import MedianPruner, HyperbandPruner
import lightgbm as lgb
from sklearn.model_selection import StratifiedKFold
from sklearn.metrics import roc_auc_score
import numpy as np

# 设置日志级别
optuna.logging.set_verbosity(optuna.logging.WARNING)

def objective_lgbm(trial, X, y):
    """LightGBM 优化用的 Optuna 目标函数"""

    # 定义超参数搜索空间
    params = {
        'objective': 'binary',
        'metric': 'auc',
        'verbosity': -1,
        'boosting_type': trial.suggest_categorical('boosting_type', ['gbdt', 'dart']),
        'num_leaves': trial.suggest_int('num_leaves', 20, 300),
        'max_depth': trial.suggest_int('max_depth', 3, 12),
        'min_child_samples': trial.suggest_int('min_child_samples', 5, 100),
        'learning_rate': trial.suggest_float('learning_rate', 1e-4, 0.3, log=True),
        'n_estimators': trial.suggest_int('n_estimators', 100, 2000),
        'subsample': trial.suggest_float('subsample', 0.5, 1.0),
        'subsample_freq': trial.suggest_int('subsample_freq', 1, 7),
        'colsample_bytree': trial.suggest_float('colsample_bytree', 0.5, 1.0),
        'reg_alpha': trial.suggest_float('reg_alpha', 1e-8, 10.0, log=True),
        'reg_lambda': trial.suggest_float('reg_lambda', 1e-8, 10.0, log=True),
        'min_split_gain': trial.suggest_float('min_split_gain', 0, 1),
        'cat_smooth': trial.suggest_int('cat_smooth', 1, 100),
        'n_jobs': -1,
    }

    # 用 5-Fold CV 评估
    skf = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
    cv_scores = []

    for fold, (train_idx, val_idx) in enumerate(skf.split(X, y)):
        X_train, X_val = X.iloc[train_idx], X.iloc[val_idx]
        y_train, y_val = y.iloc[train_idx], y.iloc[val_idx]

        train_data = lgb.Dataset(X_train, y_train)
        val_data = lgb.Dataset(X_val, y_val, reference=train_data)

        callbacks = [
            lgb.early_stopping(stopping_rounds=50, verbose=False),
            lgb.log_evaluation(-1),
        ]

        model = lgb.train(
            params, train_data,
            num_boost_round=params['n_estimators'],
            valid_sets=[val_data],
            callbacks=callbacks,
        )

        preds = model.predict(X_val)
        fold_score = roc_auc_score(y_val, preds)
        cv_scores.append(fold_score)

        # 剪枝(pruning)
        trial.report(fold_score, fold)
        if trial.should_prune():
            raise optuna.exceptions.TrialPruned()

    return np.mean(cv_scores)

def run_optuna_study(X, y, n_trials=100, n_jobs=1):
    """执行 Optuna Study"""

    # 选择 Sampler
    sampler = TPESampler(
        n_startup_trials=20,       # 初始随机搜索次数
        n_ei_candidates=24,        # EI 候选数
        multivariate=True,         # 多变量 TPE
        seed=42
    )

    # 设置 Pruner
    pruner = MedianPruner(
        n_startup_trials=5,        # 开始剪枝前的最少 Trial 数
        n_warmup_steps=10,         # 剪枝前的预热步数
        interval_steps=1
    )

    study = optuna.create_study(
        direction='maximize',      # 最大化 AUC
        sampler=sampler,
        pruner=pruner,
        study_name='lgbm_optimization',
        # storage='sqlite:///optuna.db',   # 持久化保存结果
        # load_if_exists=True,             # 恢复已有 Study
    )

    # 执行优化
    study.optimize(
        lambda trial: objective_lgbm(trial, X, y),
        n_trials=n_trials,
        n_jobs=n_jobs,             # 并行执行(注意：需要 DB storage)
        show_progress_bar=True,
        callbacks=[
            lambda study, trial: print(
                f"Trial {trial.number}: {trial.value:.4f} "
                f"(Best: {study.best_value:.4f})"
            ) if trial.value is not None else None
        ],
    )

    print(f"\n最优参数:")
    for key, value in study.best_params.items():
        print(f"  {key}: {value}")
    print(f"最优 AUC: {study.best_value:.4f}")
    print(f"已完成的 Trial: {len(study.trials)}")
    print(f"被剪枝的 Trial: {len([t for t in study.trials if t.state == optuna.trial.TrialState.PRUNED])}")

    return study

# Optuna 可视化
def visualize_optuna_study(study):
    """可视化 Optuna 优化结果"""
    import optuna.visualization as vis

    # 优化历史
    fig1 = vis.plot_optimization_history(study)
    fig1.show()

    # 参数重要性
    fig2 = vis.plot_param_importances(study)
    fig2.show()

    # 参数关系
    fig3 = vis.plot_parallel_coordinate(study)
    fig3.show()

    # 参数切片
    fig4 = vis.plot_slice(study)
    fig4.show()

    # 等高线图
    fig5 = vis.plot_contour(study, params=['learning_rate', 'num_leaves'])
    fig5.show()

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset
import optuna

def create_model(trial, input_dim):
    """在 Optuna Trial 中动态生成神经网络架构"""
    n_layers = trial.suggest_int('n_layers', 1, 4)
    dropout = trial.suggest_float('dropout', 0.1, 0.5)
    activation_name = trial.suggest_categorical('activation', ['relu', 'tanh', 'elu'])

    activation_map = {
        'relu': nn.ReLU(),
        'tanh': nn.Tanh(),
        'elu': nn.ELU()
    }

    layers = []
    in_features = input_dim

    for i in range(n_layers):
        out_features = trial.suggest_int(f'n_units_l{i}', 32, 512)
        layers.extend([
            nn.Linear(in_features, out_features),
            nn.BatchNorm1d(out_features),
            activation_map[activation_name],
            nn.Dropout(dropout),
        ])
        in_features = out_features

    layers.append(nn.Linear(in_features, 1))
    layers.append(nn.Sigmoid())

    return nn.Sequential(*layers)

def objective_pytorch(trial, X_train_t, y_train_t, X_val_t, y_val_t, input_dim):
    """PyTorch 神经网络的 Optuna 目标函数"""
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

    model = create_model(trial, input_dim).to(device)

    # 优化参数
    lr = trial.suggest_float('lr', 1e-5, 1e-1, log=True)
    optimizer_name = trial.suggest_categorical('optimizer', ['Adam', 'RMSprop', 'SGD'])
    batch_size = trial.suggest_categorical('batch_size', [32, 64, 128, 256])
    weight_decay = trial.suggest_float('weight_decay', 1e-8, 1e-2, log=True)

    if optimizer_name == 'Adam':
        optimizer = optim.Adam(model.parameters(), lr=lr, weight_decay=weight_decay)
    elif optimizer_name == 'RMSprop':
        optimizer = optim.RMSprop(model.parameters(), lr=lr, weight_decay=weight_decay)
    else:
        optimizer = optim.SGD(model.parameters(), lr=lr, weight_decay=weight_decay,
                              momentum=0.9)

    scheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=50)
    criterion = nn.BCELoss()

    train_dataset = TensorDataset(
        X_train_t.to(device), y_train_t.to(device).float().unsqueeze(1)
    )
    train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)

    best_val_loss = float('inf')
    patience = 10
    patience_counter = 0

    for epoch in range(100):
        model.train()
        for X_batch, y_batch in train_loader:
            optimizer.zero_grad()
            output = model(X_batch)
            loss = criterion(output, y_batch)
            loss.backward()
            optimizer.step()
        scheduler.step()

        # 验证
        model.eval()
        with torch.no_grad():
            val_output = model(X_val_t.to(device))
            val_loss = criterion(
                val_output, y_val_t.to(device).float().unsqueeze(1)
            ).item()

        # 提前剪枝报告
        trial.report(val_loss, epoch)
        if trial.should_prune():
            raise optuna.exceptions.TrialPruned()

        # Early stopping
        if val_loss < best_val_loss:
            best_val_loss = val_loss
            patience_counter = 0
        else:
            patience_counter += 1
            if patience_counter >= patience:
                break

    return best_val_loss

def run_pytorch_optuna(X_train, y_train, X_val, y_val, n_trials=50):
    """PyTorch + Optuna 集成运行"""
    X_train_t = torch.FloatTensor(X_train.values if hasattr(X_train, 'values') else X_train)
    y_train_t = torch.FloatTensor(y_train.values if hasattr(y_train, 'values') else y_train)
    X_val_t = torch.FloatTensor(X_val.values if hasattr(X_val, 'values') else X_val)
    y_val_t = torch.FloatTensor(y_val.values if hasattr(y_val, 'values') else y_val)

    input_dim = X_train_t.shape[1]

    study = optuna.create_study(
        direction='minimize',
        pruner=optuna.pruners.HyperbandPruner(
            min_resource=5, max_resource=100, reduction_factor=3
        ),
        sampler=TPESampler(seed=42)
    )

    study.optimize(
        lambda trial: objective_pytorch(
            trial, X_train_t, y_train_t, X_val_t, y_val_t, input_dim
        ),
        n_trials=n_trials,
        show_progress_bar=True,
    )

    print(f"最优验证损失: {study.best_value:.4f}")
    print(f"最优参数: {study.best_params}")

    return study

# 在连续空间中更高效的 CMA-ES
study_cmaes = optuna.create_study(
    direction='maximize',
    sampler=CmaEsSampler(
        n_startup_trials=10,    # 初始随机搜索次数
        restart_strategy='ipop', # 重启策略
        seed=42
    )
)

pip install ray[tune] ray[air]

import ray
from ray import tune
from ray.tune import CLIReporter
from ray.tune.schedulers import ASHAScheduler, HyperBandScheduler
from ray.tune.search.optuna import OptunaSearch
from ray.tune.search.bayesopt import BayesOptSearch
import torch
import torch.nn as nn

ray.init(ignore_reinit_error=True)

def train_with_tune(config, data=None):
    """Ray Tune 训练函数"""
    X_train, y_train, X_val, y_val = data

    # 创建模型
    model = nn.Sequential(
        nn.Linear(X_train.shape[1], config['hidden_size']),
        nn.ReLU(),
        nn.Dropout(config['dropout']),
        nn.Linear(config['hidden_size'], config['hidden_size'] // 2),
        nn.ReLU(),
        nn.Linear(config['hidden_size'] // 2, 1),
        nn.Sigmoid()
    )

    optimizer = torch.optim.Adam(
        model.parameters(),
        lr=config['lr'],
        weight_decay=config['weight_decay']
    )
    criterion = nn.BCELoss()

    X_train_t = torch.FloatTensor(X_train)
    y_train_t = torch.FloatTensor(y_train).unsqueeze(1)
    X_val_t = torch.FloatTensor(X_val)
    y_val_t = torch.FloatTensor(y_val).unsqueeze(1)

    for epoch in range(config['max_epochs']):
        model.train()
        optimizer.zero_grad()
        output = model(X_train_t)
        loss = criterion(output, y_train_t)
        loss.backward()
        optimizer.step()

        if epoch % 5 == 0:
            model.eval()
            with torch.no_grad():
                val_output = model(X_val_t)
                val_loss = criterion(val_output, y_val_t).item()

            # 向 Ray Tune 报告中间结果
            tune.report(
                val_loss=val_loss,
                training_iteration=epoch,
            )

def run_ray_tune(X_train, y_train, X_val, y_val, num_samples=50):
    """执行 Ray Tune 分布式 HPO"""

    config = {
        'hidden_size': tune.choice([64, 128, 256, 512]),
        'dropout': tune.uniform(0.1, 0.5),
        'lr': tune.loguniform(1e-5, 1e-1),
        'weight_decay': tune.loguniform(1e-8, 1e-3),
        'max_epochs': tune.choice([50, 100, 200]),
    }

    # ASHA(Asynchronous Successive Halving Algorithm)调度器
    scheduler = ASHAScheduler(
        metric='val_loss',
        mode='min',
        max_t=200,              # 最大 epoch 数
        grace_period=10,        # 最少运行 epoch 数
        reduction_factor=3,     # 淘汰比例
    )

    reporter = CLIReporter(
        metric_columns=['val_loss', 'training_iteration'],
        max_progress_rows=10
    )

    # 集成 OptunaSearch
    search_alg = OptunaSearch(metric='val_loss', mode='min')

    result = tune.run(
        tune.with_parameters(
            train_with_tune,
            data=(X_train, y_train, X_val, y_val)
        ),
        config=config,
        num_samples=num_samples,
        scheduler=scheduler,
        search_alg=search_alg,
        progress_reporter=reporter,
        verbose=1,
        resources_per_trial={'cpu': 2, 'gpu': 0},
    )

    best_trial = result.get_best_trial('val_loss', 'min', 'last')
    print(f"最优验证损失: {best_trial.last_result['val_loss']:.4f}")
    print(f"最优参数: {best_trial.config}")

    return result

# Population Based Training(PBT)
from ray.tune.schedulers import PopulationBasedTraining

def run_pbt(X_train, y_train, X_val, y_val):
    """PBT：在训练过程中动态改变超参数"""

    pbt_scheduler = PopulationBasedTraining(
        time_attr='training_iteration',
        metric='val_loss',
        mode='min',
        perturbation_interval=20,    # 每 N 步变异一次
        hyperparam_mutations={
            'lr': tune.loguniform(1e-5, 1e-1),
            'dropout': tune.uniform(0.1, 0.5),
        },
        quantile_fraction=0.25,      # 用前 25% 的参数替换后 25%
    )

    result = tune.run(
        tune.with_parameters(
            train_with_tune,
            data=(X_train, y_train, X_val, y_val)
        ),
        config={
            'hidden_size': 256,
            'dropout': tune.uniform(0.1, 0.5),
            'lr': tune.loguniform(1e-4, 1e-1),
            'weight_decay': 1e-5,
            'max_epochs': 200,
        },
        num_samples=8,
        scheduler=pbt_scheduler,
        verbose=1,
    )

    return result

pip install autogluon

from autogluon.tabular import TabularPredictor
import pandas as pd

def autogluon_tabular_example(train_df, test_df, target_col, eval_metric='roc_auc'):
    """AutoGluon 表格数据训练"""

    # 基本用法(3 行就能训练！)
    predictor = TabularPredictor(
        label=target_col,
        eval_metric=eval_metric,
        path='autogluon_models/',     # 模型保存路径
        problem_type='binary',        # 'binary'、'multiclass'、'regression'、'softclass'
    )

    predictor.fit(
        train_data=train_df,
        time_limit=3600,              # 最大训练时间(秒)
        presets='best_quality',       # 质量预设
        # 'best_quality'：最高性能(慢)
        # 'good_quality'：性能/速度平衡
        # 'medium_quality'：快速训练
        # 'optimize_for_deployment'：针对快速预测优化
        excluded_model_types=['KNN'], # 要排除的模型类型
        verbosity=2,
    )

    # 输出排行榜
    leaderboard = predictor.leaderboard(test_df, silent=True)
    print(leaderboard[['model', 'score_test', 'score_val', 'pred_time_test']].head(10))

    # 预测
    predictions = predictor.predict(test_df)
    pred_proba = predictor.predict_proba(test_df)

    # 特征重要性
    feature_importance = predictor.feature_importance(test_df)
    print(feature_importance.head(20))

    return predictor, predictions, pred_proba

# 高级设置：自定义超参数
def autogluon_advanced(train_df, test_df, target_col):
    """AutoGluon 高级设置"""
    import lightgbm as lgb
    import xgboost as xgb

    hyperparameters = {
        'GBM': [
            {'num_boost_round': 300, 'ag_args': {'name_suffix': 'fast'}},
            {'num_boost_round': 1000, 'learning_rate': 0.03,
             'ag_args': {'name_suffix': 'slow', 'priority': 0}},
        ],
        'XGB': [
            {'n_estimators': 300, 'max_depth': 6},
        ],
        'CAT': [
            {'iterations': 500, 'depth': 6},
        ],
        'NN_TORCH': [
            {'num_epochs': 50, 'learning_rate': 1e-3,
             'dropout_prob': 0.1},
        ],
        'RF': [
            {'n_estimators': 300},
        ],
    }

    predictor = TabularPredictor(
        label=target_col,
        eval_metric='roc_auc',
        path='autogluon_advanced/',
    )

    predictor.fit(
        train_data=train_df,
        hyperparameters=hyperparameters,
        time_limit=7200,
        num_stack_levels=1,       # 堆叠层数
        num_bag_folds=5,          # bagging 折数(0 表示不使用 bagging)
        num_bag_sets=1,           # bagging 集数
        verbosity=3,
    )

    return predictor

from autogluon.multimodal import MultiModalPredictor

def autogluon_image_classification(train_df, test_df, label_col, image_col):
    """AutoGluon 图像分类"""
    predictor = MultiModalPredictor(label=label_col)

    predictor.fit(
        train_data=train_df,
        time_limit=3600,
        hyperparameters={
            'model.timm_image.checkpoint_name': 'efficientnet_b4',
            'optimization.learning_rate': 1e-4,
            'optimization.max_epochs': 20,
        }
    )

    predictions = predictor.predict(test_df)
    return predictor, predictions

# 文本 + 表格多模态
def autogluon_multimodal(train_df, test_df, target_col):
    """AutoGluon 多模态训练(文本 + 数值型特征)"""
    predictor = MultiModalPredictor(
        label=target_col,
        problem_type='binary',
    )

    predictor.fit(
        train_data=train_df,
        time_limit=3600,
        hyperparameters={
            'model.hf_text.checkpoint_name': 'bert-base-uncased',
        }
    )

    return predictor

pip install flaml

from flaml import AutoML
import pandas as pd
import numpy as np

def flaml_basic_example(X_train, y_train, X_test, task='classification'):
    """FLAML 基本用法"""
    automl = AutoML()

    automl_settings = {
        'time_budget': 300,           # 最大训练时间(秒)
        'metric': 'roc_auc',          # 优化指标
        'task': task,                 # 'classification'、'regression'、'ranking'
        'estimator_list': [           # 要尝试的模型列表
            'lgbm', 'xgboost', 'catboost',
            'rf', 'extra_tree', 'lrl1', 'lrl2',
            'kneighbor', 'prophet', 'arima'
        ],
        'log_file_name': 'flaml_log.log',
        'seed': 42,
        'n_jobs': -1,
        'verbose': 1,

        # 成本效率高的搜索设置
        'retrain_full': True,         # 用全部数据重新训练最终模型
        'max_iter': 100,              # 最大迭代次数
        'ensemble': True,             # 是否使用集成
        'eval_method': 'cv',          # 'cv' 或 'holdout'
        'n_splits': 5,                # CV 折数
    }

    automl.fit(X_train, y_train, **automl_settings)

    print(f"最优模型: {automl.best_estimator}")
    print(f"最优损失: {automl.best_loss:.4f}")
    print(f"最优配置: {automl.best_config}")
    print(f"总训练时间: {automl.time_to_find_best_model:.1f} 秒")

    predictions = automl.predict(X_test)
    pred_proba = automl.predict_proba(X_test)

    return automl, predictions, pred_proba

# FLAML + scikit-learn 流水线
def flaml_sklearn_pipeline(X_train, y_train, X_test):
    """将 FLAML 集成进 scikit-learn 流水线"""
    from sklearn.pipeline import Pipeline
    from sklearn.preprocessing import StandardScaler

    automl = AutoML()

    pipeline = Pipeline([
        ('scaler', StandardScaler()),
        ('automl', automl),
    ])

    pipeline.fit(
        X_train, y_train,
        automl__time_budget=120,
        automl__metric='roc_auc',
        automl__task='classification',
    )

    return pipeline

# FLAML 自定义目标函数
def flaml_custom_objective(X_train, y_train):
    """FLAML 自定义评估指标"""
    from flaml.automl.data import get_output_from_log

    def custom_metric(
        X_val, y_val, estimator, labels, X_train, y_train,
        weight_val=None, weight_train=None, *args
    ):
        """优化 F-beta 分数的自定义指标"""
        from sklearn.metrics import fbeta_score
        y_pred = estimator.predict(X_val)
        score = fbeta_score(y_val, y_pred, beta=2, average='weighted')
        return -score, {'f2_score': score}  # (损失, 指标字典)

    automl = AutoML()
    automl.fit(
        X_train, y_train,
        metric=custom_metric,
        task='classification',
        time_budget=120,
    )

    return automl

pip install h2o

import h2o
from h2o.automl import H2OAutoML
import pandas as pd

def h2o_automl_example(train_df, test_df, target_col, max_models=20):
    """使用 H2O AutoML"""

    # 启动 H2O 集群
    h2o.init(
        nthreads=-1,              # 使用所有 CPU
        max_mem_size='8G',        # 最大内存
        port=54321,
    )

    # 转换为 H2O Frame
    train_h2o = h2o.H2OFrame(train_df)
    test_h2o = h2o.H2OFrame(test_df)

    # 将目标列转换为类别型(分类任务)
    train_h2o[target_col] = train_h2o[target_col].asfactor()

    feature_cols = [col for col in train_df.columns if col != target_col]

    # 执行 AutoML
    aml = H2OAutoML(
        max_models=max_models,     # 尝试的最大模型数
        max_runtime_secs=3600,     # 最大运行时间
        seed=42,
        sort_metric='AUC',         # 排序基准指标
        balance_classes=False,     # 类别不平衡处理

        # 包含/排除的算法
        include_algos=[
            'GBM', 'GLM', 'DRF', 'DeepLearning',
            'StackedEnsemble', 'XGBoost'
        ],
        # exclude_algos=['DeepLearning'],

        # 堆叠设置
        keep_cross_validation_predictions=True,
        keep_cross_validation_models=True,
        nfolds=5,

        verbosity='info',
    )

    aml.fit(
        x=feature_cols,
        y=target_col,
        training_frame=train_h2o,
        validation_frame=None,     # None 表示使用 CV
        leaderboard_frame=test_h2o,
    )

    # 输出排行榜
    lb = aml.leaderboard
    print("H2O AutoML 排行榜:")
    print(lb.head(20))

    # 最优模型
    best_model = aml.leader
    print(f"\n最优模型: {best_model.model_id}")

    # 预测
    predictions = best_model.predict(test_h2o)
    pred_df = predictions.as_data_frame()

    # 模型解释
    explainability = aml.explain(test_h2o)

    # SHAP 值(支持的模型)
    if hasattr(best_model, 'shap_values'):
        shap_vals = best_model.shap_values(test_h2o)

    # 保存模型
    model_path = h2o.save_model(model=best_model, path='h2o_models/', force=True)
    print(f"模型已保存: {model_path}")

    return aml, best_model, pred_df

# 关闭 H2O
def cleanup_h2o():
    h2o.cluster().shutdown()

import torch
import torch.nn as nn
import torch.nn.functional as F

class MixedOperation(nn.Module):
    """DARTS 的混合运算(加权和)"""
    def __init__(self, operations):
        super().__init__()
        self.ops = nn.ModuleList(operations)
        # 架构参数(alpha)
        self.alphas = nn.Parameter(torch.randn(len(operations)))

    def forward(self, x):
        weights = F.softmax(self.alphas, dim=0)
        return sum(w * op(x) for w, op in zip(weights, self.ops))

class DARTSCell(nn.Module):
    """DARTS 单元结构"""
    def __init__(self, in_channels, out_channels):
        super().__init__()

        # 定义可用的运算
        operations = [
            nn.Conv2d(in_channels, out_channels, 3, padding=1),
            nn.Conv2d(in_channels, out_channels, 5, padding=2),
            nn.MaxPool2d(3, stride=1, padding=1),
            nn.AvgPool2d(3, stride=1, padding=1),
            nn.Identity() if in_channels == out_channels else
                nn.Conv2d(in_channels, out_channels, 1),
        ]

        self.mixed_op = MixedOperation(operations)
        self.bn = nn.BatchNorm2d(out_channels)

    def forward(self, x):
        return F.relu(self.bn(self.mixed_op(x)))

class SimpleDARTS(nn.Module):
    """简单的 DARTS 架构"""
    def __init__(self, num_classes=10, num_cells=6):
        super().__init__()
        self.stem = nn.Conv2d(3, 64, 3, padding=1)

        self.cells = nn.ModuleList([
            DARTSCell(64, 64) for _ in range(num_cells)
        ])

        self.classifier = nn.Linear(64, num_classes)

    def forward(self, x):
        x = self.stem(x)
        for cell in self.cells:
            x = cell(x)
        x = x.mean([2, 3])  # 全局平均池化
        return self.classifier(x)

    def arch_parameters(self):
        """只返回架构参数"""
        return [p for n, p in self.named_parameters() if 'alphas' in n]

    def model_parameters(self):
        """只返回权重参数"""
        return [p for n, p in self.named_parameters() if 'alphas' not in n]

def train_darts(model, train_loader, val_loader, epochs=50):
    """DARTS 双层优化训练"""
    # 权重优化器
    w_optimizer = torch.optim.SGD(
        model.model_parameters(), lr=0.025, momentum=0.9, weight_decay=3e-4
    )
    # 架构参数优化器
    a_optimizer = torch.optim.Adam(
        model.arch_parameters(), lr=3e-4, betas=(0.5, 0.999), weight_decay=1e-3
    )

    w_scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(
        w_optimizer, T_max=epochs
    )

    for epoch in range(epochs):
        model.train()
        train_iter = iter(train_loader)
        val_iter = iter(val_loader)

        for step in range(min(len(train_loader), len(val_loader))):
            # 1. 更新架构参数(用验证数据)
            try:
                X_val, y_val = next(val_iter)
            except StopIteration:
                val_iter = iter(val_loader)
                X_val, y_val = next(val_iter)

            a_optimizer.zero_grad()
            val_logits = model(X_val)
            val_loss = F.cross_entropy(val_logits, y_val)
            val_loss.backward()
            a_optimizer.step()

            # 2. 更新权重参数(用训练数据)
            X_train, y_train = next(train_iter)

            w_optimizer.zero_grad()
            train_logits = model(X_train)
            train_loss = F.cross_entropy(train_logits, y_train)
            train_loss.backward()
            nn.utils.clip_grad_norm_(model.model_parameters(), 5.0)
            w_optimizer.step()

        w_scheduler.step()

        if epoch % 10 == 0:
            print(f"Epoch {epoch}: Train Loss = {train_loss.item():.4f}")

    # 提取最优架构
    for i, cell in enumerate(model.cells):
        weights = F.softmax(cell.mixed_op.alphas, dim=0).detach()
        best_op = weights.argmax().item()
        print(f"Cell {i}: 最优运算索引 = {best_op}, 权重 = {weights}")

    return model

class SuperNetwork(nn.Module):
    """One-Shot NAS：从单一超网络中采样子网络"""

    def __init__(self, num_classes=10, max_channels=256):
        super().__init__()
        self.max_channels = max_channels

        # 可能的通道数选项
        self.channel_options = [64, 128, 256]

        # 用全部参数定义各层
        self.conv1 = nn.Conv2d(3, max_channels, 3, padding=1)
        self.conv2 = nn.Conv2d(max_channels, max_channels, 3, padding=1)
        self.conv3 = nn.Conv2d(max_channels, max_channels, 3, padding=1)
        self.bn1 = nn.BatchNorm2d(max_channels)
        self.bn2 = nn.BatchNorm2d(max_channels)
        self.bn3 = nn.BatchNorm2d(max_channels)
        self.classifier = nn.Linear(max_channels, num_classes)

    def forward(self, x, arch_config=None):
        """arch_config: 各层的通道数"""
        if arch_config is None:
            # 随机采样子网络
            arch_config = {
                'conv1_out': torch.randint(0, len(self.channel_options), (1,)).item(),
                'conv2_out': torch.randint(0, len(self.channel_options), (1,)).item(),
            }

        # 用切片后的通道做前向传播
        c1 = self.channel_options[arch_config['conv1_out']]
        c2 = self.channel_options[arch_config['conv2_out']]

        x = F.relu(self.bn1(self.conv1(x)[:, :c1]))
        x = F.relu(self.bn2(self.conv2(
            F.pad(x, (0, 0, 0, 0, 0, self.max_channels - c1))
        )[:, :c2]))
        x = F.relu(self.bn3(self.conv3(
            F.pad(x, (0, 0, 0, 0, 0, self.max_channels - c2))
        )))
        x = x.mean([2, 3])
        return self.classifier(x)

pip install auto-sklearn

import autosklearn.classification
import autosklearn.regression
from autosklearn.metrics import roc_auc, mean_squared_error

def auto_sklearn_example(X_train, y_train, X_test, task='classification'):
    """Auto-sklearn：基于 scikit-learn 的 AutoML"""

    if task == 'classification':
        automl = autosklearn.classification.AutoSklearnClassifier(
            time_left_for_this_task=3600,    # 整体时间限制(秒)
            per_run_time_limit=360,          # 单个模型时间限制
            n_jobs=-1,
            memory_limit=8192,               # 内存限制(MB)
            ensemble_size=50,                # 集成大小
            ensemble_nbest=50,               # 用于集成的最优模型数
            max_models_on_disc=50,

            # 包含/排除的算法
            include={
                'classifier': [
                    'random_forest', 'gradient_boosting',
                    'extra_trees', 'liblinear_svc'
                ]
            },
            # exclude={'classifier': ['k_nearest_neighbors']},

            # 指标
            metric=roc_auc,
            resampling_strategy='cv',
            resampling_strategy_arguments={'folds': 5},

            seed=42,
        )
    else:
        automl = autosklearn.regression.AutoSklearnRegressor(
            time_left_for_this_task=3600,
            per_run_time_limit=360,
            n_jobs=-1,
            metric=mean_squared_error,
            seed=42,
        )

    automl.fit(X_train, y_train)

    # 输出统计信息
    print(automl.sprint_statistics())
    print(automl.leaderboard())

    predictions = automl.predict(X_test)
    if task == 'classification':
        pred_proba = automl.predict_proba(X_test)

    return automl, predictions

# 基于 LLM 的超参数优化(概念代码)
from openai import OpenAI

def llm_hyperparameter_suggestion(dataset_description, model_type, previous_results=None):
    """利用 LLM 进行超参数建议"""
    client = OpenAI()

    prompt = f"""
    数据集特征：
    {dataset_description}

    模型类型：{model_type}

    之前的尝试结果：
    {previous_results if previous_results else '无'}

    请基于以上信息，以 JSON 格式为 {model_type} 提出最优超参数。
    """

    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system",
             "content": "你是一名机器学习专家。请帮助进行超参数优化。"},
            {"role": "user", "content": prompt}
        ],
        response_format={"type": "json_object"}
    )

    return response.choices[0].message.content

# 利用 Few-shot learning 的 AutoML
def llm_feature_engineering(df_description, target_description):
    """LLM 提出的特征工程建议"""
    client = OpenAI()

    prompt = f"""
    数据框列：
    {df_description}

    预测目标：
    {target_description}

    请提出可能有用的衍生特征及对应的 Python 代码。
    """

    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system",
             "content": "作为机器学习特征工程专家，请提出有用的衍生特征。"},
            {"role": "user", "content": prompt}
        ]
    )

    return response.choices[0].message.content

# AutoML 智能体(实验性)
class AutoMLAgent:
    """基于 LLM 的 AutoML 智能体(概念实现)"""

    def __init__(self, llm_client, X_train, y_train, X_val, y_val, max_iterations=10):
        self.client = llm_client
        self.X_train = X_train
        self.y_train = y_train
        self.X_val = X_val
        self.y_val = y_val
        self.max_iterations = max_iterations
        self.history = []
        self.best_score = 0
        self.best_params = None

    def get_next_config(self):
        """向 LLM 请求下一个要尝试的配置"""
        history_str = "\n".join([
            f"Iteration {i+1}: params={h['params']}, score={h['score']:.4f}"
            for i, h in enumerate(self.history[-5:])  # 最近 5 个
        ])

        prompt = f"""
        目前为止尝试过的 LightGBM 参数与结果：
        {history_str if history_str else '无(首次尝试)'}

        请以 JSON 格式提出下一个要尝试的参数组合。
        可行范围：num_leaves(10-300)、learning_rate(0.001-0.3)、
        n_estimators(100-2000)、subsample(0.5-1.0)、colsample_bytree(0.5-1.0)
        """

        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "你是一名 HPO 专家。"},
                {"role": "user", "content": prompt}
            ],
            response_format={"type": "json_object"}
        )

        import json
        return json.loads(response.choices[0].message.content)

    def evaluate(self, params):
        """评估参数"""
        import lightgbm as lgb
        from sklearn.metrics import roc_auc_score

        model = lgb.LGBMClassifier(**params, random_state=42, verbose=-1)
        model.fit(self.X_train, self.y_train)
        preds = model.predict_proba(self.X_val)[:, 1]
        return roc_auc_score(self.y_val, preds)

    def run(self):
        """运行智能体"""
        for i in range(self.max_iterations):
            config = self.get_next_config()
            score = self.evaluate(config)
            self.history.append({'params': config, 'score': score})

            if score > self.best_score:
                self.best_score = score
                self.best_params = config
                print(f"Iteration {i+1}: 出现新的最优分数 {score:.4f}")

        print(f"\n最优分数: {self.best_score:.4f}")
        print(f"最优参数: {self.best_params}")
        return self.best_params