데이터 드리프트

# 데이터 드리프트 감지 - Evidently AI 활용
import pandas as pd
import numpy as np
from evidently import ColumnMapping
from evidently.report import Report
from evidently.metrics import DataDriftTable, DatasetDriftMetric
from scipy import stats

# 1. 통계적 드리프트 감지 (기본 구현)
def detect_drift_ks_test(reference_data: np.ndarray,
                         current_data: np.ndarray,
                         threshold: float = 0.05) -> dict:
    """
    Kolmogorov-Smirnov 테스트로 드리프트 감지
    귀무가설: 두 분포가 동일
    """
    statistic, p_value = stats.ks_2samp(reference_data, current_data)

    return {
        "statistic": statistic,
        "p_value": p_value,
        "drift_detected": p_value < threshold,
        "severity": "high" if statistic > 0.2 else "medium" if statistic > 0.1 else "low"
    }

# 2. PSI (Population Stability Index) 계산
def calculate_psi(reference: np.ndarray, current: np.ndarray,
                  buckets: int = 10) -> float:
    """
    PSI 계산 - 금융권에서 많이 사용
    PSI < 0.1: 변화 없음
    0.1 <= PSI < 0.2: 약간의 변화
    PSI >= 0.2: 유의미한 변화 (재훈련 필요)
    """
    # 버킷 경계 설정 (reference 기준)
    breakpoints = np.percentile(reference, np.linspace(0, 100, buckets + 1))
    breakpoints[0], breakpoints[-1] = -np.inf, np.inf

    # 각 버킷의 비율 계산
    ref_counts = np.histogram(reference, breakpoints)[0] / len(reference)
    cur_counts = np.histogram(current, breakpoints)[0] / len(current)

    # 0 방지
    ref_counts = np.clip(ref_counts, 0.0001, None)
    cur_counts = np.clip(cur_counts, 0.0001, None)

    psi = np.sum((cur_counts - ref_counts) * np.log(cur_counts / ref_counts))
    return psi

# 3. Evidently AI 활용 (실무 권장)
def create_drift_report(reference_df: pd.DataFrame,
                        current_df: pd.DataFrame,
                        target_column: str = None):
    """Evidently AI로 종합 드리프트 리포트 생성"""
    column_mapping = ColumnMapping(target=target_column)

    report = Report(metrics=[
        DatasetDriftMetric(),  # 전체 데이터셋 드리프트
        DataDriftTable()       # 컬럼별 드리프트
    ])

    report.run(reference_data=reference_df,
               current_data=current_df,
               column_mapping=column_mapping)

    return report

# 사용 예시
np.random.seed(42)
# 훈련 시점 데이터 (정규분포)
reference = np.random.normal(50, 10, 10000)
# 운영 시점 데이터 (분포 이동 발생!)
current = np.random.normal(55, 12, 10000)

# 드리프트 감지
ks_result = detect_drift_ks_test(reference, current)
psi_value = calculate_psi(reference, current)

print(f"KS Test - 드리프트 감지: {ks_result['drift_detected']}")
print(f"PSI 값: {psi_value:.4f} ({'재훈련 필요!' if psi_value >= 0.2 else '정상'})")