데이터 품질 관리

# 데이터 품질 관리 파이프라인 예제
from great_expectations.core import ExpectationSuite
from great_expectations.dataset import PandasDataset
import pandas as pd
from datetime import datetime
from typing import Dict, List
import json

class AIDataQualityManager:
    """EU AI Act 준수를 위한 데이터 품질 관리 시스템"""

    def __init__(self, quality_threshold: float = 0.8):
        self.quality_threshold = quality_threshold
        self.quality_history: List[Dict] = []

    def create_ai_act_expectations(self) -> ExpectationSuite:
        """EU AI Act 제10조 요구사항 기반 품질 규칙 생성"""
        suite = ExpectationSuite("ai_act_compliance")

        # 완전성 (Completeness) - 결측치 제한
        suite.add_expectation({
            "expectation_type": "expect_column_values_to_not_be_null",
            "kwargs": {"column": "*", "mostly": 0.95}
        })

        # 고유성 (Uniqueness) - 중복 제한
        suite.add_expectation({
            "expectation_type": "expect_compound_columns_to_be_unique",
            "kwargs": {"column_list": ["record_id"]}
        })

        # 유효성 (Validity) - 데이터 타입 및 범위
        suite.add_expectation({
            "expectation_type": "expect_column_values_to_be_between",
            "kwargs": {"column": "age", "min_value": 0, "max_value": 120}
        })

        return suite

    def validate_dataset(self, df: pd.DataFrame,
                          dataset_name: str) -> Dict:
        """데이터셋 품질 검증"""
        ge_df = PandasDataset(df)

        # 품질 차원별 측정
        quality_scores = {
            "completeness": 1 - df.isnull().sum().sum() / df.size,
            "uniqueness": 1 - df.duplicated().sum() / len(df),
            "validity": self._check_validity(df),
            "timeliness": 0.95,  # 실제로는 수집 시점 기반 계산
        }

        # 대표성 검사 (AI 특화)
        if "protected_attribute" in df.columns:
            quality_scores["representativeness"] = self._check_representativeness(df)

        # 종합 점수
        overall = sum(quality_scores.values()) / len(quality_scores)

        result = {
            "dataset_name": dataset_name,
            "timestamp": datetime.now().isoformat(),
            "record_count": len(df),
            "quality_scores": quality_scores,
            "overall_score": round(overall, 3),
            "ai_act_compliant": overall >= self.quality_threshold,
            "threshold": self.quality_threshold
        }

        self.quality_history.append(result)
        return result

    def _check_validity(self, df: pd.DataFrame) -> float:
        """유효성 검사 (커스텀 규칙 기반)"""
        valid_count = 0
        total_checks = 0

        for col in df.select_dtypes(include=['number']).columns:
            # 이상치 검사 (IQR 기반)
            Q1 = df[col].quantile(0.25)
            Q3 = df[col].quantile(0.75)
            IQR = Q3 - Q1
            outliers = ((df[col] < Q1 - 1.5*IQR) | (df[col] > Q3 + 1.5*IQR)).sum()
            valid_count += len(df) - outliers
            total_checks += len(df)

        return valid_count / total_checks if total_checks > 0 else 1.0

    def _check_representativeness(self, df: pd.DataFrame) -> float:
        """대표성 검사 (보호 속성 분포)"""
        # 이상적 분포와 실제 분포 비교
        ideal_dist = {"M": 0.5, "F": 0.5}
        actual_dist = df["protected_attribute"].value_counts(normalize=True)

        divergence = sum(
            abs(actual_dist.get(k, 0) - v)
            for k, v in ideal_dist.items()
        ) / 2

        return 1 - divergence

    def generate_compliance_report(self) -> str:
        """EU AI Act 감사용 컴플라이언스 보고서"""
        report = {
            "report_type": "EU_AI_Act_Article_10_Compliance",
            "generated_at": datetime.now().isoformat(),
            "quality_threshold": self.quality_threshold,
            "validation_history": self.quality_history,
            "overall_compliance_rate": sum(
                1 for h in self.quality_history if h["ai_act_compliant"]
            ) / len(self.quality_history) if self.quality_history else 0
        }
        return json.dumps(report, indent=2, ensure_ascii=False)

    def should_proceed_with_training(self, validation_result: Dict) -> bool:
        """학습 진행 여부 결정 (Circuit Breaker)"""
        if not validation_result["ai_act_compliant"]:
            print(f"[품질 차단] 품질 점수 {validation_result['overall_score']} < "
                  f"임계치 {self.quality_threshold}. 학습이 차단되었습니다.")
            return False
        print(f"[품질 통과] 학습을 진행합니다.")
        return True

# 사용 예시
dqm = AIDataQualityManager(quality_threshold=0.8)

df = pd.DataFrame({
    "record_id": [1, 2, 3, 4, 5],
    "age": [25, 30, 35, 40, 45],
    "income": [50000, None, 55000, 60000, 65000],
    "protected_attribute": ["M", "F", "M", "F", "M"]
})

result = dqm.validate_dataset(df, "training_data_v1")
print(f"품질 점수: {result['quality_scores']}")

if dqm.should_proceed_with_training(result):
    # model.fit(df)  # 학습 진행
    pass