🙌

UMAP のハイパーパラメータチューニングにおける K-means とシルエットスコアの活用

2024/12/05に公開

■ はじめに

こんにちは。データシステム部 推薦基盤チームのかみけん(上國料)です。よろしくお願いします。

いきなり本題に入りますが、データ分析において、高次元データの可視化やクラスタリングは重要なステップだと思っています。ただ、データ分析をする上で、ラベルがそもそも存在しない(or 大量に存在する)データを扱うことが稀にあります。
今回は、元データにラベルが存在しない、すなわち教師なし学習前提で、次元削減手法である UMAP (Uniform Manifold Approximation and Projection) のハイパーパラメータを最適化する方法を紹介します。どこかで参考になれば幸いです。

■ 手法サマリ

ラベルのないデータに対して、次元削減手法 UMAP のパラメータを最適化します。パラメータ最適化には Optuna を使用し、次元削減後の潜在空間をシルエットスコア で評価・最適化します。

シルエットスコアを最大化することで、UMAPのハイパーパラメータを最適化し、データの内部構造をより良く反映した低次元表現を得ることが可能です。

UMAPやシルエットスコアについて詳しく知りたい方は、以下を参考にしてください。

▼ UMAP とは

UMAP は、高次元データを低次元空間に効率的にマッピングする次元削減手法です。t-SNE と比較されることが多いですが、UMAP はより高速で大規模データにも適応可能です。データの全体的な配置や関係性を保ちながら、局所的な構造も維持することを目指します。

詳しくは論文をご参照ください!:
https://arxiv.org/abs/1802.03426

▼ シルエットスコアとは

シルエットスコアは、各データポイントのクラスタリングの質を評価する指標です。各ポイントが自身のクラスタにどれだけ適切に配置されているか、他のクラスタとどれだけ明確に区別されているかを測ります。スコアは -1 から 1 の範囲で、値が高いほどクラスタリングの質が良いとされます。

詳細については、以下の記事が参考になります:
https://note.com/united_code/n/n8a15ae4063b3

■ 実装 (プロトタイプ)

今回パっとラベルなしの大規模なデータを見つけることができなかったので、sklearn の Digits データセットを使用してラベルを使わずパラメータを最適化していきたいと思います。(本来はラベルがあるので、最後に実際のラベルとクラスタがどのように対応しているかなどもおまけで見ていこうかなと思います。)

クラスタの作成方法は、今回は K-means を使用します(Digits よりも複雑なクラスタになる想定でしたらデータの密度をもとにクラスタリングを実行する DBSCAN (Density-Based Spatial Clustering of Applications with Noise) でもよいかもしれません)。ラベル数を知らない前提なので、K-means のクラスタ数 K に関してもハイパーパラメータチューニングします。

▼ 必要なライブラリのインストール

python 3.11.10 の環境を pyenv で構築し、以下をインストールしました。

%pip install seaborn optuna umap-learn

▼ コード

以下がプロトタイプのコードになります。

import os
import joblib
import numpy as np
import pandas as pd
from sklearn.metrics import silhouette_score
from sklearn.cluster import KMeans
import optuna
import umap
import warnings
import logging
import json
from sklearn.datasets import load_digits
import matplotlib.pyplot as plt
import seaborn as sns

# 警告を無視
warnings.filterwarnings("ignore")

# Optunaのログ設定
optuna.logging.set_verbosity(optuna.logging.INFO)

# 訓練に関連するパラメータの定義
IS_TRAIN = True
N_TRIALS = 50  # ハイパーパラメータ最適化の試行数

# モデル保存ディレクトリ
MODEL_DIR = "./caches/model/"
os.makedirs(MODEL_DIR, exist_ok=True)

# データセットのロードと準備
digits = load_digits()
features = digits['data']
target = digits['target']
feature_names = digits['feature_names'] if 'feature_names' in digits else None

# データフレームに格納
df = pd.DataFrame(features, columns=[f"feature_{i}" for i in range(features.shape[1])])
df['target'] = target  # 実際のラベルを追加(評価用)

# モデルの保存先(Digitsデータセット用)
model_path = os.path.join(MODEL_DIR, "umap_model_digits.pkl")

if IS_TRAIN:
    # Optuna の目的関数
    def objective(trial):
        """ 目的関数: シルエットスコアを最大化 """
        # UMAPのハイパーパラメータのサンプリング
        n_neighbors = trial.suggest_int("n_neighbors", 5, 50)
        min_dist = trial.suggest_uniform("min_dist", 0.0, 0.99)
        metric = trial.suggest_categorical("metric", ["euclidean", "cosine"])
        
        # KMeansのハイパーパラメータのサンプリング
        n_clusters = trial.suggest_int("n_clusters", 5, 20)
        init = trial.suggest_categorical("init", ["k-means++", "random"])
        n_init = trial.suggest_int("n_init", 10, 50)
        
        # UMAPによる次元削減
        reducer = umap.UMAP(n_neighbors=n_neighbors,
                            min_dist=min_dist,
                            metric=metric,
                            random_state=42)
        reduced_features = reducer.fit_transform(df.drop(columns=['target']))
        
        # KMeansによるクラスタリング
        kmeans = KMeans(n_clusters=n_clusters, init=init, n_init=n_init, random_state=42)
        labels = kmeans.fit_predict(reduced_features)
        
        # クラスタ数を計算(ユニークなラベル数)
        unique_labels = set(labels)
        n_clusters_found = len(unique_labels)
        
        if n_clusters_found <= 1:
            return -1  # 不適切なクラスタリング
        
        # シルエットスコアを計算
        try:
            score = silhouette_score(reduced_features, labels)
        except:
            return -1  # シルエットスコアが計算できない場合
        
        return score

    # Optuna のコールバック関数で進捗を表示
    def optuna_callback(study, trial):
        logging.info(f"Trial {trial.number}: Score={trial.value}, Params={trial.params}")

    # Optuna で探索を実行
    study = optuna.create_study(direction="maximize")
    study.optimize(objective, n_trials=N_TRIALS, callbacks=[optuna_callback])

    # 最適なパラメータ
    best_params = study.best_params
    best_score = study.best_value
    print("最適なパラメータ:", best_params)
    print("最適なシルエットスコア:", best_score)

    # 最適なパラメータで次元削減モデルを作成
    reducer = umap.UMAP(n_neighbors=best_params["n_neighbors"],
                        min_dist=best_params["min_dist"],
                        metric=best_params["metric"],
                        random_state=42)
    reduced = reducer.fit_transform(df.drop(columns=['target']))
    df["umap_x"], df["umap_y"] = zip(*reduced)

    # 最適なパラメータで KMeans を実行
    kmeans = KMeans(n_clusters=best_params["n_clusters"],
                    init=best_params["init"],
                    n_init=best_params["n_init"],
                    random_state=42)
    labels = kmeans.fit_predict(reduced)
    df["cluster"] = labels
    print("クラスタリング結果の概要:")
    print(df["cluster"].value_counts())

    # モデルを保存
    joblib.dump(reducer, model_path)
    print(f"UMAPモデルを保存しました: {model_path}")
    joblib.dump(kmeans, os.path.join(MODEL_DIR, "kmeans_model_digits.pkl"))
    print(f"KMeansモデルを保存しました: {os.path.join(MODEL_DIR, 'kmeans_model_digits.pkl')}")

    # ベストモデルの詳細を保存(オプション)
    best_model_details = {
        "umap_params": {
            "n_neighbors": best_params["n_neighbors"],
            "min_dist": best_params["min_dist"],
            "metric": best_params["metric"]
        },
        "kmeans_params": {
            "n_clusters": best_params["n_clusters"],
            "init": best_params["init"],
            "n_init": best_params["n_init"]
        },
        "score": best_score,
        "cluster_counts": df["cluster"].value_counts().to_dict()
    }
    details_path = os.path.join(MODEL_DIR, "best_umap_kmeans_model_details_digits.json")
    with open(details_path, "w") as f:
        json.dump(best_model_details, f, indent=4, ensure_ascii=False)
    print(f"ベストモデルの詳細を保存しました: {details_path}")

else:
    # 訓練をスキップして既存モデルをロード
    if os.path.exists(model_path) and os.path.exists(os.path.join(MODEL_DIR, "kmeans_model_digits.pkl")):
        reducer = joblib.load(model_path)
        kmeans = joblib.load(os.path.join(MODEL_DIR, "kmeans_model_digits.pkl"))
        print(f"UMAPモデルとKMeansモデルをロードしました: {model_path}, {os.path.join(MODEL_DIR, 'kmeans_model_digits.pkl')}")
        reduced = reducer.transform(df.drop(columns=['target']))
        df["umap_x"], df["umap_y"] = zip(*reduced)
        
        labels = kmeans.predict(reduced)
        df["cluster"] = labels
        print("クラスタリング結果の概要:")
        print(df["cluster"].value_counts())
    else:
        print(f"必要なモデルが見つかりません。まずモデルを訓練してください。")

# クラスタリング結果と元のラベルの可視化
def visualize_clusters(df, title_suffix=''):
    plt.figure(figsize=(14, 6))

    # クラスタリング結果を色分け
    plt.subplot(1, 2, 1)
    sns.scatterplot(x='umap_x', y='umap_y', hue='cluster', data=df, palette='tab10', legend='full', alpha=0.7)
    plt.title(f'UMAP + KMeans Clustering Results {title_suffix}')
    plt.xlabel('UMAP Dimension 1')
    plt.ylabel('UMAP Dimension 2')
    plt.legend(title='Cluster', bbox_to_anchor=(1.05, 1), loc='upper left')

    # 元のラベルを色分け
    plt.subplot(1, 2, 2)
    sns.scatterplot(x='umap_x', y='umap_y', hue='target', data=df, palette='tab10', legend='full', alpha=0.7)
    plt.title(f'Original Labels {title_suffix}')
    plt.xlabel('UMAP Dimension 1')
    plt.ylabel('UMAP Dimension 2')
    plt.legend(title='Label', bbox_to_anchor=(1.05, 1), loc='upper left')

    plt.tight_layout()
    plt.show()

# 可視化の実行
visualize_clusters(df, title_suffix='on Digits Dataset')

# クラスタリング結果と元のラベルの一致度を定量的に評価
from sklearn.metrics import adjusted_rand_score, normalized_mutual_info_score

adjusted_rand = adjusted_rand_score(df["target"], df["cluster"])
nmi = normalized_mutual_info_score(df["target"], df["cluster"])

print(f"Adjusted Rand Index (ARI): {adjusted_rand:.4f}")
print(f"Normalized Mutual Information (NMI): {nmi:.4f}")

ちなみに、optuna でチューニングしているパラメータは以下です(結局チューニングすれば良いかなと思っているので、あんまり深堀りしていません🙏):

UMAPのハイパーパラメータ
パラメータ名 説明 可能な値 影響
n_neighbors 各データポイントの近傍をどれだけ考慮するかを決定します。 整数: 5 〜 50 小さい値(e.g. 5)は局所的な構造に焦点を当て、大きい値(e.g. 50)は全体的な構造を捉えます。
min_dist 低次元空間でのデータポイント間の最小距離を制御します。 浮動小数点数: 0.0 〜 0.99 小さい値(e.g. 0.0)はクラスタ間の境界を明確にし、クラスタ内部を凝集させます。大きい値(e.g. 0.99)はデータポイントをより広く分散させ、クラスタ間の境界を曖昧にします。
metric データポイント間の距離を計算する際に使用する距離指標を選択します。 カテゴリカル: "euclidean", "cosine" 距離指標の選択により、次元削減後のデータ表現が大きく変わります。"euclidean" は直線距離、"cosine" はベクトルの方向性に基づく距離を使用します。
K-meansクラスタリングのハイパーパラメータ
パラメータ名 説明 可能な値 影響
n_clusters データを分割するクラスタの数を指定します。 整数: 2 〜 20 クラスタ数が少ないとデータの大まかな構造を捉え、多いと詳細な構造を捉えます。ただし、多すぎるとノイズがクラスタに含まれる可能性があります。
init 初期クラスタ中心の選択方法を指定します。 カテゴリカル: "k-means++", "random" 初期化方法により、クラスタリングの収束速度や結果の安定性が変わります。"k-means++" は初期中心をいい感じに選択し、収束を早める傾向があります。"random" はランダムに選択します。
n_init 初期化の試行回数を指定します。 整数: 10 〜 50 多くの初期化を試行することで、より良いクラスタリング結果が得られる可能性が高まりますが、計算コストも増加します。

▼ 結果

50 回施行させて、最適なパラメータとして選ばれたのは以下のパラメータでした。

最適なパラメータ: {'n_neighbors': 11, 'min_dist': 0.008871245237427692, 'metric': 'cosine', 'n_clusters': 10, 'init': 'random', 'n_init': 41}

シルエットスコアとクラスタリング結果の概要は以下になります。

最適なシルエットスコア: 0.8316881060600281
クラスタリング結果の概要:
cluster
2    352
1    194
8    183
7    181
4    181
3    178
5    178
9    177
6    146
0     27
Name: count, dtype: int64

クラスタ数をハイパーパラメータとして調整した上で、 10 が最適と選ばれたのは、UMAP がデータの固有の構造をしっかりと抑えている証拠とも言えそうです。また、シルエットスコアが 0.8317 と高いため、各クラスタが適切に分離されつつ内部の密度も高いという、理想的なクラスタリングができていそうです。

以下が、UMAP で Digits の特徴量を次元削減した結果を可視化したものになります。
左側のグラフが、K-means でクラスタリングした結果で色付けしたもので、右側のグラフが、元のラベル(Digits データセットの真のクラスラベル)で色付けしたものになります。

綺麗にクラスタに分かれていることがわかりますが、1 と 7 の構造を誤判別するケースが見られていそうです。(若干似ているし仕方ないか…)

ちなみに、Adjusted Rand Index (ARI) や Normalized Mutual Information (NMI) の指標も計算しました。まずまずの結果が得られました。

Adjusted Rand Index (ARI): 0.8410
Normalized Mutual Information (NMI): 0.9187

■ まとめ

今回、ラベルのないデータに対して UMAP のハイパーパラメータをシルエットスコアを用いて最適化する手法を紹介しました。Digits データがクラスタリングが簡単なのもありますが、真のラベルと K-means のクラスタ数を入れていなくても、綺麗にクラスタリングすることができました。
より大規模なデータセットを適用した場合に、どのような結果になるのか楽しみです。

株式会社ZOZO

Discussion