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분 이상 실행되지 않았습니다. 모니터링 파이프라인 점검이 필요합니다.'