Optunaインテグレーションによる材料配合最適化

import pandas as pd
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
from sklearn.model_selection import train_test_split
from sklearn import linear_model
from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score
import wandb
from joblib import dump

wb_project_name = "concrete-strength-optimization-demo"

run = wandb.init(project=wb_project_name)

df_train = pd.read_excel("dataset/Concrete_Data.xls")

X = df_train.iloc[:, :-1]
y = df_train.iloc[:, -1]

X_train, X_test, y_train, y_test = train_test_split(X, y, random_state=0)

# 学習データをW&Bのテーブルとして登録します
run.log({"training_data": wandb.Table(dataframe=pd.concat([X_train, y_train], axis=1).sort_index())})

# 学習実行
model = Pipeline([('scaler', StandardScaler()), ('ridge', linear_model.Ridge(alpha=0.0))]).fit(X_train, y_train)

# モデル出力
model_file_path = 'model.joblib'
dump(pipe, model_file_path)

# モデルをArtifactに保存
model_artifact = wandb.Artifact(
            "ridge", type="model",
            description="Simple Ridge with StandardScaler",
            metadata=dict(wandb.config))
model_artifact.add_file(model_file_path)
run.log_artifact(model_artifact)

# 予測実行・精度評価
y_test_pred = pipe.predict(X_test)
run.log({
    "RMSE": mean_squared_error(y_test, y_test_pred, squared=False),
    "MAE": mean_absolute_error(y_test, y_test_pred),
    "R2": r2_score(y_test, y_test_pred),
})

# 検定データをW&Bのテーブルで保存
df_valid = pd.concat([X_test, y_test, pd.Series(y_test_pred, name="pred", index=X_test.index)], axis=1).sort_index()
run.log({"valid_data": wandb.Table(dataframe=df_valid)})

from joblib import load
from pathlib import Path
import pandas as pd
import plotly.express as px
import wandb
import optuna
from optuna.integration.wandb import WeightsAndBiasesCallback

wb_project_name = "concrete-strength-optimization-demo"

def load_model():
    with wandb.init(project=wb_project_name) as run:
        artifact = run.use_artifact("ridge:latest", type="model")
        artifact_dir = artifact.download()
        return load(Path(artifact_dir) / "model.joblib")

model = load_model()

# 学習データを読み込む
df_train = pd.read_excel("dataset/Concrete_Data.xls")

# 学習データの各変数の最大値・最小値の辞書
limits = {col: [df_train.min()[col], df_train.max()[col]] for col in df_train.columns}

# 目的変数
target = 'Concrete compressive strength(MPa, megapascals) '

# 目的変数を除く特徴量のリスト
features = list(df_train.drop(columns=[target]).columns)

# 固定パラメータの辞書（今回は'Age (day)'を60で固定）
fixed_features = {'Age (day)': 60}

# 探索対象パラメータ名のリスト
control_features = [col for col in features if col not in fixed_features]

def suggest(trial, col):
    if col in control_features:
        return trial.suggest_float(col, limits[col][0], limits[col][1])
    return fixed_features[col]

def optimize():
    wandb_kwargs = {"project": wb_project_name}

    wandbc = WeightsAndBiasesCallback(wandb_kwargs=wandb_kwargs, as_multirun=True)
    
    @wandbc.track_in_wandb()
    def objective(trial):
        df_suggest = pd.DataFrame({col: [suggest(trial, col)] for col in features})
        return - model.predict(df_suggest)
    
    study = optuna.create_study()
    study.optimize(objective, n_trials=100, callbacks=[wandbc])

optimize()