異常検知パイプラインの構築・運用：ml と組み合わせる

import numpy as np
def psi(expected: np.ndarray, actual: np.ndarray, bins: int = 10) -> float:    """学習データ vs 本番データの分布差を測る。    < 0.1: 安定 / 0.1〜0.25: 軽微な変化 / > 0.25: 大変化"""    # 学習データに基づいて bin を決定    bin_edges = np.percentile(expected, np.linspace(0, 100, bins + 1))    bin_edges = np.unique(bin_edges)  # 重複削除
    expected_pct = np.histogram(expected, bins=bin_edges)[0] / len(expected)    actual_pct = np.histogram(actual, bins=bin_edges)[0] / len(actual)
    # 0 除算防止    expected_pct = np.where(expected_pct == 0, 0.0001, expected_pct)    actual_pct = np.where(actual_pct == 0, 0.0001, actual_pct)
    return np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct))
# 使用例psi_value = psi(train_features, prod_features)print(f"PSI: {psi_value:.3f}")if psi_value > 0.25:    print("⚠️ 大きなドリフト検出 - モデル再学習を検討")

from scipy.stats import wasserstein_distance, ks_2sampimport pandas as pd
def detect_drift(train_df: pd.DataFrame, prod_df: pd.DataFrame, threshold: float = 0.1):    """各特徴量についてドリフトを検出"""    results = []    for col in train_df.select_dtypes(include="number").columns:        # KS 検定        ks_stat, ks_p = ks_2samp(train_df[col].dropna(), prod_df[col].dropna())        # Wasserstein 距離        wd = wasserstein_distance(train_df[col].dropna(), prod_df[col].dropna())        results.append({            "column": col,            "ks_stat": ks_stat, "ks_p": ks_p,            "wasserstein": wd,            "drifted": ks_p < 0.05,        })    return pd.DataFrame(results)
drift_df = detect_drift(train_df, prod_df)print(drift_df[drift_df["drifted"]])

import numpy as np
def champion_vs_challenger(    champion_scores: np.ndarray,    challenger_scores: np.ndarray,    y_true: np.ndarray,):    """両モデルを同じデータで比較し、勝者を決める"""    from sklearn.metrics import average_precision_score    champion_pr = average_precision_score(y_true, champion_scores)    challenger_pr = average_precision_score(y_true, challenger_scores)
    diff = challenger_pr - champion_pr    print(f"Champion  PR-AUC: {champion_pr:.4f}")    print(f"Challenger PR-AUC: {challenger_pr:.4f}")    print(f"差分: {diff:+.4f}")
    # ブートストラップで信頼区間    n_bootstrap = 1000    diffs = []    n = len(y_true)    for _ in range(n_bootstrap):        idx = np.random.choice(n, n, replace=True)        d = (            average_precision_score(y_true[idx], challenger_scores[idx]) -            average_precision_score(y_true[idx], champion_scores[idx])        )        diffs.append(d)    ci_low, ci_high = np.percentile(diffs, [2.5, 97.5])    print(f"95% CI: [{ci_low:+.4f}, {ci_high:+.4f}]")
    if ci_low > 0:        print("→ Challenger 採用（統計的有意）")    else:        print("→ 様子見 or Champion 維持")
# 使用例champion_vs_challenger(champion_scores, challenger_scores, y_true)