MLモデルモニタリングとドリフト検知：Evidently AI + MLflow 本番運用ガイド

# Evidently AI インストール（MLflow連携含む）
# pip install evidently mlflow scikit-learn pandas

import pandas as pd
import numpy as np
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split

from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, DataQualityPreset
from evidently.metrics import (
    DatasetDriftMetric,
    DataDriftTable,
    ColumnDriftMetric,
)

# リファレンス / 現在データの準備
data = load_iris(as_frame=True)
df = data.frame
df.columns = ["sepal_length", "sepal_width", "petal_length", "petal_width", "target"]

reference_data = df.sample(frac=0.5, random_state=42)
current_data = df.drop(reference_data.index)

# データドリフトのあるシミュレーションデータ生成
current_drifted = current_data.copy()
current_drifted["sepal_length"] = current_drifted["sepal_length"] + np.random.normal(2.0, 0.5, len(current_drifted))
current_drifted["petal_width"] = current_drifted["petal_width"] * 1.8

# ドリフトレポート生成
drift_report = Report(metrics=[
    DatasetDriftMetric(),
    DataDriftTable(),
])

drift_report.run(
    reference_data=reference_data,
    current_data=current_drifted,
)

# 結果を辞書として抽出（プログラム的活用）
result = drift_report.as_dict()
dataset_drift = result["metrics"][0]["result"]["dataset_drift"]
drift_share = result["metrics"][0]["result"]["share_of_drifted_columns"]

print(f"データセットドリフト検知: {dataset_drift}")
print(f"ドリフトカラム比率: {drift_share:.2%}")

# HTMLレポートとして保存
drift_report.save_html("drift_report.html")

from evidently.test_suite import TestSuite
from evidently.test_preset import DataDriftTestPreset, DataQualityTestPreset
from evidently.tests import (
    TestColumnDrift,
    TestShareOfDriftedColumns,
    TestNumberOfMissingValues,
    TestShareOfOutRangeValues,
    TestMeanInNSigmas,
)

# データドリフト + 品質テストスイートの構成
monitoring_suite = TestSuite(tests=[
    # ドリフトテスト：全カラムの30%以上がドリフトした場合に失敗
    TestShareOfDriftedColumns(lt=0.3),

    # 個別の重要特徴量ドリフト検証
    TestColumnDrift(column_name="sepal_length"),
    TestColumnDrift(column_name="petal_width"),

    # データ品質テスト
    TestNumberOfMissingValues(eq=0),

    # 値範囲検証：sepal_lengthがリファレンスデータ基準±3シグマ以内
    TestMeanInNSigmas(column_name="sepal_length", n=3),
])

monitoring_suite.run(
    reference_data=reference_data,
    current_data=current_drifted,
)

# テスト結果をプログラム的に確認
suite_result = monitoring_suite.as_dict()
all_passed = all(
    test["status"] == "SUCCESS"
    for test in suite_result["tests"]
)

print(f"全テスト通過: {all_passed}")
for test in suite_result["tests"]:
    status_icon = "PASS" if test["status"] == "SUCCESS" else "FAIL"
    print(f"  [{status_icon}] {test['name']}: {test['status']}")

# CI/CDパイプラインでexit codeとして活用
if not all_passed:
    print("ALERT: データドリフトまたは品質異常を検知。再学習パイプラインのトリガーが必要。")
    # sys.exit(1)  # CIでビルド失敗処理

import mlflow
from evidently.report import Report
from evidently.metrics import (
    DatasetDriftMetric,
    DataDriftTable,
    ColumnDriftMetric,
)
import json
from datetime import datetime

# MLflowトラッキングサーバー設定
mlflow.set_tracking_uri("http://mlflow.internal:5000")
mlflow.set_experiment("model-monitoring/fraud-detection-v2")

def log_drift_to_mlflow(
    reference_data,
    current_data,
    model_name: str,
    model_version: str,
    batch_id: str,
):
    """ドリフト分析結果をMLflowに記録する関数"""

    # Evidentlyドリフトレポート生成
    drift_report = Report(metrics=[
        DatasetDriftMetric(),
        DataDriftTable(),
    ])
    drift_report.run(
        reference_data=reference_data,
        current_data=current_data,
    )

    result = drift_report.as_dict()
    drift_result = result["metrics"][0]["result"]

    # MLflow Runとして記録
    with mlflow.start_run(run_name=f"drift-check-{batch_id}") as run:
        # 基本ドリフトメトリクス
        mlflow.log_metric("dataset_drift_detected", int(drift_result["dataset_drift"]))
        mlflow.log_metric("drifted_columns_share", drift_result["share_of_drifted_columns"])
        mlflow.log_metric("number_of_drifted_columns", drift_result["number_of_drifted_columns"])
        mlflow.log_metric("total_columns", drift_result["number_of_columns"])

        # 個別カラムドリフトスコアの記録
        column_drift = result["metrics"][1]["result"]["drift_by_columns"]
        for col_name, col_info in column_drift.items():
            safe_col_name = col_name.replace(" ", "_").replace("/", "_")
            mlflow.log_metric(
                f"drift_score_{safe_col_name}",
                col_info.get("drift_score", 0.0),
            )
            mlflow.log_metric(
                f"drift_detected_{safe_col_name}",
                int(col_info.get("column_drift", False)),
            )

        # タグとしてメタデータを記録
        mlflow.set_tags({
            "monitoring.type": "drift_detection",
            "monitoring.model_name": model_name,
            "monitoring.model_version": model_version,
            "monitoring.batch_id": batch_id,
            "monitoring.timestamp": datetime.utcnow().isoformat(),
            "monitoring.reference_size": str(len(reference_data)),
            "monitoring.current_size": str(len(current_data)),
        })

        # HTMLレポートをアーティファクトとして保存
        report_path = f"/tmp/drift_report_{batch_id}.html"
        drift_report.save_html(report_path)
        mlflow.log_artifact(report_path, artifact_path="drift_reports")

        # JSON結果もアーティファクトとして保存
        json_path = f"/tmp/drift_result_{batch_id}.json"
        with open(json_path, "w") as f:
            json.dump(result, f, indent=2, default=str)
        mlflow.log_artifact(json_path, artifact_path="drift_reports")

        print(f"ドリフト結果をMLflowに記録完了。Run ID: {run.info.run_id}")
        return drift_result["dataset_drift"], drift_result["share_of_drifted_columns"]


# 使用例
is_drifted, drift_share = log_drift_to_mlflow(
    reference_data=reference_data,
    current_data=current_drifted,
    model_name="fraud-detector",
    model_version="3",
    batch_id="2026-03-06-batch-001",
)

from mlflow import MlflowClient

client = MlflowClient(tracking_uri="http://mlflow.internal:5000")

MODEL_NAME = "fraud-detector"

def handle_drift_detection(
    is_drifted: bool,
    drift_share: float,
    model_name: str = MODEL_NAME,
    drift_threshold_warn: float = 0.2,
    drift_threshold_critical: float = 0.5,
):
    """ドリフト検知結果に基づくモデルレジストリアクションの実行"""

    # 現在の本番モデルバージョンの確認
    try:
        prod_version = client.get_model_version_by_alias(model_name, "production")
        current_version = prod_version.version
        print(f"現在の本番モデルバージョン: {current_version}")
    except Exception as e:
        print(f"本番モデルエイリアスの取得に失敗: {e}")
        return

    if not is_drifted:
        print("ドリフト未検知。現在のモデルを維持。")
        client.set_model_version_tag(
            model_name, current_version,
            key="last_drift_check",
            value="passed",
        )
        return

    if drift_share >= drift_threshold_critical:
        # 臨界ドリフト：即座にフォールバックモデルに切り替え + 再学習トリガー
        print(f"CRITICAL: ドリフト比率 {drift_share:.1%} - フォールバックモデルへの切り替えと再学習トリガー")
        client.set_model_version_tag(
            model_name, current_version,
            key="drift_status", value="critical",
        )
        # フォールバックモデルがあれば切り替え
        try:
            fallback = client.get_model_version_by_alias(model_name, "fallback")
            client.set_registered_model_alias(model_name, "production", fallback.version)
            print(f"フォールバックモデルバージョン {fallback.version} に切り替え完了")
        except Exception:
            print("WARNING: フォールバックモデルなし。現在のモデルを維持しつつ緊急再学習が必要。")

        # 再学習トリガー（外部システム呼び出し）
        trigger_retraining(model_name, reason="critical_drift")

    elif drift_share >= drift_threshold_warn:
        # 警告レベルドリフト：タグ記録 + 通知
        print(f"WARNING: ドリフト比率 {drift_share:.1%} - モニタリング強化と再学習予約")
        client.set_model_version_tag(
            model_name, current_version,
            key="drift_status", value="warning",
        )
        # スケジュール再学習キューに追加
        schedule_retraining(model_name, priority="normal")


def trigger_retraining(model_name: str, reason: str):
    """緊急再学習トリガー（Airflow DAG、Kubeflow Pipelineなどの呼び出し）"""
    print(f"再学習トリガー: model={model_name}, reason={reason}")
    # requests.post("http://airflow.internal/api/v1/dags/retrain/dagRuns", ...)


def schedule_retraining(model_name: str, priority: str):
    """スケジュール再学習キューへの登録"""
    print(f"再学習スケジュール登録: model={model_name}, priority={priority}")


# 実行
handle_drift_detection(
    is_drifted=True,
    drift_share=0.55,
    model_name=MODEL_NAME,
)

# Airflow DAG 例：ドリフトチェック + 条件付き再学習
# dag_drift_monitor.py

from airflow import DAG
from airflow.operators.python import PythonOperator, BranchPythonOperator
from airflow.operators.empty import EmptyOperator
from airflow.utils.dates import days_ago
from datetime import timedelta
import pandas as pd

default_args = {
    "owner": "ml-platform",
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
    "execution_timeout": timedelta(minutes=30),
}

dag = DAG(
    dag_id="ml_drift_monitor_fraud_detection",
    default_args=default_args,
    description="日次ドリフトモニタリングと条件付き再学習",
    schedule_interval="0 6 * * *",  # 毎日午前6時
    start_date=days_ago(1),
    catchup=False,
    tags=["ml-monitoring", "drift-detection"],
)


def fetch_data(**context):
    """リファレンスデータと直近24時間のサービングデータをロード"""
    from sqlalchemy import create_engine
    engine = create_engine("postgresql://reader:password@db.internal/features")

    reference = pd.read_sql(
        "SELECT * FROM fraud_features_reference", engine
    )
    current = pd.read_sql(
        """SELECT * FROM fraud_features_serving
           WHERE created_at >= NOW() - INTERVAL '24 hours'""",
        engine,
    )

    # XComでパスを転送（大容量はS3に保存）
    ref_path = "/tmp/reference_data.parquet"
    cur_path = "/tmp/current_data.parquet"
    reference.to_parquet(ref_path)
    current.to_parquet(cur_path)

    context["ti"].xcom_push(key="reference_path", value=ref_path)
    context["ti"].xcom_push(key="current_path", value=cur_path)
    context["ti"].xcom_push(key="current_size", value=len(current))


def run_drift_check(**context):
    """Evidentlyドリフト分析の実行とMLflowへの記録"""
    from evidently.report import Report
    from evidently.metrics import DatasetDriftMetric, DataDriftTable
    import mlflow

    ti = context["ti"]
    ref_path = ti.xcom_pull(key="reference_path")
    cur_path = ti.xcom_pull(key="current_path")

    reference = pd.read_parquet(ref_path)
    current = pd.read_parquet(cur_path)

    # 最小サンプル数の検証
    if len(current) < 100:
        print(f"現在データのサンプル数不足: {len(current)}。ドリフトチェックをスキップ。")
        ti.xcom_push(key="drift_action", value="skip")
        return "skip_retraining"

    report = Report(metrics=[DatasetDriftMetric(), DataDriftTable()])
    report.run(reference_data=reference, current_data=current)
    result = report.as_dict()

    drift_detected = result["metrics"][0]["result"]["dataset_drift"]
    drift_share = result["metrics"][0]["result"]["share_of_drifted_columns"]

    # MLflowに記録
    mlflow.set_tracking_uri("http://mlflow.internal:5000")
    mlflow.set_experiment("monitoring/fraud-detection")
    with mlflow.start_run(run_name=f"drift-{context['ds']}"):
        mlflow.log_metric("drift_detected", int(drift_detected))
        mlflow.log_metric("drift_share", drift_share)

    ti.xcom_push(key="drift_detected", value=drift_detected)
    ti.xcom_push(key="drift_share", value=drift_share)


def decide_action(**context):
    """ドリフトレベルに応じた再学習の判断"""
    ti = context["ti"]
    drift_detected = ti.xcom_pull(key="drift_detected")
    drift_share = ti.xcom_pull(key="drift_share")

    if drift_share is None or drift_share < 0.2:
        return "skip_retraining"
    elif drift_share >= 0.5:
        return "trigger_emergency_retrain"
    else:
        return "trigger_scheduled_retrain"


fetch_task = PythonOperator(
    task_id="fetch_data", python_callable=fetch_data, dag=dag,
)
drift_task = PythonOperator(
    task_id="run_drift_check", python_callable=run_drift_check, dag=dag,
)
branch_task = BranchPythonOperator(
    task_id="decide_action", python_callable=decide_action, dag=dag,
)
skip_task = EmptyOperator(task_id="skip_retraining", dag=dag)
scheduled_retrain = EmptyOperator(task_id="trigger_scheduled_retrain", dag=dag)
emergency_retrain = EmptyOperator(task_id="trigger_emergency_retrain", dag=dag)

fetch_task >> drift_task >> branch_task >> [skip_task, scheduled_retrain, emergency_retrain]

# prometheus_drift_exporter.py
from prometheus_client import start_http_server, Gauge, Counter, Histogram
from evidently.report import Report
from evidently.metrics import DatasetDriftMetric, DataDriftTable
import pandas as pd
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Prometheusメトリクス定義
DRIFT_DETECTED = Gauge(
    "ml_model_drift_detected",
    "データセットドリフト検知ステータス (0/1)",
    ["model_name", "model_version"],
)
DRIFT_SHARE = Gauge(
    "ml_model_drift_column_share",
    "ドリフト検知カラムの比率",
    ["model_name", "model_version"],
)
COLUMN_DRIFT_SCORE = Gauge(
    "ml_model_column_drift_score",
    "個別カラムドリフトスコア",
    ["model_name", "model_version", "column_name"],
)
DRIFT_CHECK_TOTAL = Counter(
    "ml_model_drift_checks_total",
    "ドリフトチェック実行回数",
    ["model_name"],
)
DRIFT_CHECK_DURATION = Histogram(
    "ml_model_drift_check_duration_seconds",
    "ドリフトチェック所要時間",
    ["model_name"],
    buckets=[0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0],
)

MODEL_NAME = "fraud-detector"
MODEL_VERSION = "3"


def run_periodic_drift_check(
    reference_path: str,
    current_query_fn,
    interval_seconds: int = 300,
):
    """定期ドリフトチェックとPrometheusメトリクス更新"""
    reference = pd.read_parquet(reference_path)

    while True:
        try:
            start_time = time.time()

            # 最新データのロード
            current = current_query_fn()
            if current is None or len(current) < 50:
                logger.warning(f"現在データ不足: {len(current) if current is not None else 0}件")
                time.sleep(interval_seconds)
                continue

            # 特徴量カラムのみフィルタリング（ターゲット、メタデータカラムを除外）
            feature_cols = [c for c in reference.columns if c not in ["target", "id", "timestamp"]]
            ref_features = reference[feature_cols]
            cur_features = current[feature_cols]

            # ドリフト分析
            report = Report(metrics=[DatasetDriftMetric(), DataDriftTable()])
            report.run(reference_data=ref_features, current_data=cur_features)
            result = report.as_dict()

            drift_result = result["metrics"][0]["result"]
            column_results = result["metrics"][1]["result"]["drift_by_columns"]

            # Prometheusメトリクス更新
            DRIFT_DETECTED.labels(MODEL_NAME, MODEL_VERSION).set(
                int(drift_result["dataset_drift"])
            )
            DRIFT_SHARE.labels(MODEL_NAME, MODEL_VERSION).set(
                drift_result["share_of_drifted_columns"]
            )

            for col_name, col_info in column_results.items():
                COLUMN_DRIFT_SCORE.labels(MODEL_NAME, MODEL_VERSION, col_name).set(
                    col_info.get("drift_score", 0.0)
                )

            DRIFT_CHECK_TOTAL.labels(MODEL_NAME).inc()

            duration = time.time() - start_time
            DRIFT_CHECK_DURATION.labels(MODEL_NAME).observe(duration)

            logger.info(
                f"ドリフトチェック完了: drift={drift_result['dataset_drift']}, "
                f"share={drift_result['share_of_drifted_columns']:.2%}, "
                f"duration={duration:.1f}s"
            )

        except Exception as e:
            logger.error(f"ドリフトチェック失敗: {e}", exc_info=True)

        time.sleep(interval_seconds)


if __name__ == "__main__":
    # PrometheusメトリクスHTTPサーバー起動（ポート8000）
    start_http_server(8000)
    logger.info("Prometheusメトリクスエクスポーター起動（ポート8000）")

    # 定期ドリフトチェック開始（5分間隔）
    run_periodic_drift_check(
        reference_path="/data/reference/fraud_features_v3.parquet",
        current_query_fn=lambda: pd.read_parquet("/data/serving/latest_batch.parquet"),
        interval_seconds=300,
    )

# prometheus-alerts.yaml
groups:
  - name: ml_model_drift_alerts
    rules:
      - alert: MLModelDriftDetected
        expr: ml_model_drift_detected == 1
        for: 5m
        labels:
          severity: warning
          team: ml-platform
        annotations:
          summary: 'MLモデルデータドリフト検知'
          description: 'モデル {{ $labels.model_name }} v{{ $labels.model_version }} でデータドリフトが検知されました。ドリフトカラム比率: {{ $value }}'

      - alert: MLModelCriticalDrift
        expr: ml_model_drift_column_share > 0.5
        for: 0m
        labels:
          severity: critical
          team: ml-platform
        annotations:
          summary: 'MLモデル臨界ドリフト - 即時対応が必要'
          description: 'モデル {{ $labels.model_name }} のドリフトカラム比率が {{ $value | humanizePercentage }} です。即時再学習またはフォールバック切替が必要です。'

      - alert: MLDriftCheckStalled
        expr: rate(ml_model_drift_checks_total[1h]) == 0
        for: 30m
        labels:
          severity: warning
          team: ml-platform
        annotations:
          summary: 'MLドリフトチェック停止'
          description: 'モデル {{ $labels.model_name }} のドリフトチェックが30分以上実行されていません。モニタリングパイプラインの点検が必要です。'