sklearn-advanced

scikit-learn - 高度なアーキテクチャ

単純なスクリプトから先に進むには、Pipeline API を使いこなす必要があります。これにより、前処理とモデリングの一連の処理全体を単一のオブジェクトとして扱え、学習ロジックと本番環境での推論ロジックを確実に一致させることができます。

使用する場面

• 異種データを扱う複雑な特徴量エンジニアリングのフローを構築する場合。
• 再利用可能なカスタム前処理ステップ（ドメイン固有のクリーニングなど）を作成する場合。
• データリークを起こさずに、厳密なハイパーパラメーター調整を実施する場合。
• 標準的な Random Forest を超えるアンサンブル手法を実装する場合。
• モデルの判断を監視・解釈する場合（部分依存、Permutation Importance）。
• 高性能な本番環境向けにモデルをエクスポートする場合。

参考ドキュメント

• Pipeline ガイド: https://scikit-learn.org/stable/modules/compose.html
• カスタム推定器: https://scikit-learn.org/stable/developers/develop.html
• モデル評価: https://scikit-learn.org/stable/modules/model_evaluation.html
• 検索パターン: sklearn.base.BaseEstimator、sklearn.compose.make_column_selector、sklearn.model_selection.GridSearchCV

基本原則

すべてをオブジェクトとして扱う

ワークフローの各ステップはすべて、推定器にする必要があります。学習とテストの間に手作業で pandas の操作を行っている場合は、データリークのリスクがあります。

Pipeline の契約

Pipeline を使うと、.fit() は学習データに対してのみ呼び出され、.transform() は学習データとテストデータの両方に一貫して適用されます。

異種データの処理

ColumnTransformer を使うと、数値データ、カテゴリーデータ、テキストデータにそれぞれ異なる処理を並行して適用し、結果を自動的に結合できます。

クイックリファレンス

標準的なインポート

import numpy as np
import pandas as pd
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.pipeline import Pipeline, FeatureUnion
from sklearn.compose import ColumnTransformer, make_column_selector
from sklearn.preprocessing import FunctionTransformer, StandardScaler, OneHotEncoder
from sklearn.model_selection import cross_validate, StratifiedKFold

基本パターン - 実用的な Pipeline

# 1. Define Preprocessor
preprocessor = ColumnTransformer(
    transformers=[
        ('num', StandardScaler(), make_column_selector(dtype_include=np.number)),
        ('cat', OneHotEncoder(handle_unknown='ignore'), make_column_selector(dtype_include=object))
    ])

# 2. Create the Full Pipeline
clf = Pipeline(steps=[
    ('preprocessor', preprocessor),
    ('classifier', RandomForestClassifier())
])

# 3. Fit and Tune (The entire pipeline is tuned together)
# Use 'classifier__' prefix to access parameters inside the pipeline
param_grid = {'classifier__n_estimators': [100, 200]}
grid = GridSearchCV(clf, param_grid, cv=5).fit(X_train, y_train)

重要なルール

✅ 推奨事項

• BaseEstimator と TransformerMixin を継承する - これにより、.fit_transform() と .get_params() を追加の実装なしで利用できます。
• check_is_fitted を使用する - カスタム変換器では、.transform() を実行する前にモデルが学習済みかどうかを必ず確認します。
• handle_unknown='ignore' を設定する - OneHotEncoder でこれを設定すると、本番環境で新しいカテゴリーが現れた際のクラッシュを防げます。
• TransformedTargetRegressor を使用する - 目的変数（Y）を対数変換する必要がある場合は、予測値を自動で逆変換するために使用します。
• cross_val_score より cross_validate を優先する - 複数の評価指標を指定でき、過学習を検出するための学習スコアも返されます。
• n_jobs=-1 を設定する - GridSearch と交差検証中の CPU 使用率を最大化します。

❌ 禁止事項

• テストデータに fit_transform を使用しない - これは、過度に楽観的な結果を招く最大の原因です。
• 不要な fit を実装しない - log-transform のような状態を持たない変換には、FunctionTransformer を使用します。
• 列名をハードコードしない - make_column_selector を使用すると、新しい列にも対応できる堅牢な Pipeline を構築できます。
• Pipeline のインデックスを無視しない - Pipeline が失敗した場合は、pipe.named_steps['step_name'] を使用して内部状態を調べます。

カスタム推定器の開発

カスタム特徴量セレクターの作成

from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.utils.validation import check_is_fitted

class VarianceSelector(BaseEstimator, TransformerMixin):
    def __init__(self, threshold=0.01):
        self.threshold = threshold

    def fit(self, X, y=None):
        X = pd.DataFrame(X)
        self.variances_ = X.var()
        self.columns_to_keep_ = self.variances_[self.variances_ > self.threshold].index
        self.n_features_in_ = X.shape[1]
        return self

    def transform(self, X):
        check_is_fitted(self)
        X = pd.DataFrame(X)
        return X[self.columns_to_keep_]

高度な前処理

Target Encoding（カーディナリティの高いカテゴリーの処理）

from sklearn.preprocessing import TargetEncoder

# Efficiently encodes categories like 'City' or 'ZipCode' 
# based on the average target value, with internal cross-validation
encoder = TargetEncoder(smooth="auto")
X_encoded = encoder.fit_transform(X_cat, y)

Stacking と Voting アンサンブル

from sklearn.ensemble import StackingClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.svm import SVC

estimators = [
    ('rf', RandomForestClassifier()),
    ('svc', Pipeline([('scaler', StandardScaler()), ('svr', SVC())]))
]

# Use a meta-learner (LogisticRegression) to combine base model predictions
stack_clf = StackingClassifier(
    estimators=estimators, final_estimator=LogisticRegression()
)

モデル評価と診断

厳密な交差検証

from sklearn.model_selection import cross_validate

scoring = ['accuracy', 'precision_macro', 'recall_macro', 'f1_macro']
results = cross_validate(clf, X, y, cv=5, scoring=scoring, return_train_score=True)

print(f"Test F1: {results['test_f1_macro'].mean():.4f}")
print(f"Train F1: {results['train_f1_macro'].mean():.4f}") # Check for gap (overfitting)

キャリブレーション曲線（確率の妥当性を確認）

from sklearn.calibration import CalibrationDisplay

# A well-calibrated model's predicted probability matches the actual frequency
CalibrationDisplay.from_estimator(clf, X_test, y_test, n_bins=10)

本番環境と永続化

大規模モデルに Joblib を使用する

import joblib

# Save model
joblib.dump(clf, 'final_model.joblib', compress=3)

# Load model
loaded_model = joblib.load('final_model.joblib')

ONNX へのエクスポート（高速推論）

# requires skl2onnx
from skl2onnx import convert_sklearn
from skl2onnx.common.data_types import FloatTensorType

initial_type = [('float_input', FloatTensorType([None, X.shape[1]]))]
onx = convert_sklearn(clf, initial_types=initial_type)
with open("model.onnx", "wb") as f:
    f.write(onx.SerializeToString())

実践的なワークフロー

1. 欠損データと外れ値の自動処理

from sklearn.impute import KNNImputer
from sklearn.ensemble import IsolationForest

def build_robust_pipe():
    return Pipeline([
        ('imputer', KNNImputer(n_neighbors=5)),
        # FunctionTransformer for outlier removal is tricky because 
        # it changes row count. IsolationForest is better used for filtering.
        ('scaler', StandardScaler()),
        ('model', GradientBoostingClassifier())
    ])

2. 時系列分割（未来のデータリークを防止）

from sklearn.model_selection import TimeSeriesSplit

tscv = TimeSeriesSplit(n_splits=5)
# Use this cv object in GridSearchCV
grid = GridSearchCV(model, params, cv=tscv)

3. Feature Union（特徴量の並列抽出）

from sklearn.pipeline import FeatureUnion
from sklearn.decomposition import PCA
from sklearn.feature_selection import SelectKBest

# Extract PCA features AND SelectKBest features in parallel
combined_features = FeatureUnion([
    ("pca", PCA(n_components=2)),
    ("univ_select", SelectKBest(k=5))
])

pipe = Pipeline([
    ("features", combined_features),
    ("clf", RandomForestClassifier())
])

パフォーマンスの最適化

Pipeline の結果をキャッシュする

前処理（KNNImputer など）が遅く、GridSearch を実行している場合は、memory を使用して変換器の出力をキャッシュします。

from tempfile import mkdtemp
from shutil import rmtree

cachedir = mkdtemp()
pipe = Pipeline(steps=[...], memory=cachedir)
# Clean up after
# rmtree(cachedir)

よくある落とし穴と解決策

「X に LabelEncoder」を使用する際のエラー

LabelEncoder はラベル（y）専用です。特徴量（X）には、必ず OrdinalEncoder または OneHotEncoder を使用してください。

本番環境での列の不一致

Pipeline には学習時の列順が保存されます。本番環境で列順が異なる DataFrame を渡すと、失敗したり、誤った結果が返されたりする場合があります。

# ✅ Solution: Ensure your pipeline is the first point of entry 
# for raw data, or use a custom transformer that sorts columns.

ハイパーパラメーター調整中のデータリーク

Pipeline 内での標準的な交差検証は安全です。ただし、Pipeline の前に SelectKBest などで特徴量選択を行うと、データセット全体の情報がモデルに漏れてしまいます。

# ✅ Solution: Always include feature selection AS A STEP in the Pipeline.

scikit-learn の高度な活用には規律が必要です。すべてのデータ変換を Pipeline／Transformer アーキテクチャに組み込むことで、高精度なだけでなく、堅牢で保守しやすく、実環境へのデプロイに適したモデルを作成できます。