데이터 품질

# 데이터 품질 측정 및 검증 예제
import pandas as pd
import numpy as np
from dataclasses import dataclass
from typing import Dict, List, Optional

@dataclass
class DataQualityReport:
    """데이터 품질 측정 결과"""
    dataset_name: str
    record_count: int

    # 6대 품질 차원 점수 (0.0 ~ 1.0)
    accuracy: float
    completeness: float
    consistency: float
    timeliness: float
    validity: float
    uniqueness: float

    # AI 특화 품질 차원
    representativeness: float
    bias_score: Dict[str, float]

    # 종합 점수
    @property
    def overall_score(self) -> float:
        weights = {
            "accuracy": 0.2,
            "completeness": 0.15,
            "consistency": 0.15,
            "timeliness": 0.1,
            "validity": 0.15,
            "uniqueness": 0.1,
            "representativeness": 0.15
        }
        return sum([
            self.accuracy * weights["accuracy"],
            self.completeness * weights["completeness"],
            self.consistency * weights["consistency"],
            self.timeliness * weights["timeliness"],
            self.validity * weights["validity"],
            self.uniqueness * weights["uniqueness"],
            self.representativeness * weights["representativeness"]
        ])

    def is_ai_act_compliant(self, threshold: float = 0.8) -> bool:
        """EU AI Act 제10조 요구 수준 충족 여부"""
        return self.overall_score >= threshold

class DataQualityChecker:
    """데이터 품질 측정기"""

    def __init__(self, df: pd.DataFrame, dataset_name: str):
        self.df = df
        self.dataset_name = dataset_name

    def check_completeness(self) -> float:
        """완전성: 결측치 비율 기반"""
        total_cells = self.df.size
        missing_cells = self.df.isnull().sum().sum()
        return 1 - (missing_cells / total_cells)

    def check_uniqueness(self, key_columns: List[str] = None) -> float:
        """고유성: 중복 레코드 비율"""
        if key_columns:
            duplicates = self.df.duplicated(subset=key_columns).sum()
        else:
            duplicates = self.df.duplicated().sum()
        return 1 - (duplicates / len(self.df))

    def check_validity(self, rules: Dict[str, callable]) -> float:
        """유효성: 비즈니스 규칙 준수율"""
        valid_counts = []
        for column, rule in rules.items():
            if column in self.df.columns:
                valid = self.df[column].apply(rule).sum()
                valid_counts.append(valid / len(self.df))
        return np.mean(valid_counts) if valid_counts else 1.0

    def check_representativeness(self, protected_column: str,
                                  expected_dist: Dict[str, float]) -> float:
        """대표성: 실제 분포와 기대 분포의 일치도"""
        if protected_column not in self.df.columns:
            return 1.0

        actual_dist = self.df[protected_column].value_counts(normalize=True)
        chi_squared = 0
        for category, expected_ratio in expected_dist.items():
            actual_ratio = actual_dist.get(category, 0)
            chi_squared += (actual_ratio - expected_ratio) ** 2 / expected_ratio

        # chi-squared 값을 0-1 점수로 변환 (낮을수록 좋음)
        return max(0, 1 - chi_squared / len(expected_dist))

    def check_bias(self, label_column: str,
                   protected_columns: List[str]) -> Dict[str, float]:
        """편향 측정: 보호 속성별 레이블 불균형"""
        bias_scores = {}
        for col in protected_columns:
            if col in self.df.columns and label_column in self.df.columns:
                # 각 그룹별 긍정 레이블 비율
                group_rates = self.df.groupby(col)[label_column].mean()
                # 그룹 간 최대 차이
                bias_scores[col] = group_rates.max() - group_rates.min()
        return bias_scores

    def generate_report(self,
                        validity_rules: Dict[str, callable] = None,
                        protected_column: str = None,
                        expected_dist: Dict[str, float] = None,
                        label_column: str = None,
                        protected_columns: List[str] = None) -> DataQualityReport:
        """종합 품질 리포트 생성"""
        return DataQualityReport(
            dataset_name=self.dataset_name,
            record_count=len(self.df),
            accuracy=0.95,  # 실제로는 외부 검증 필요
            completeness=self.check_completeness(),
            consistency=0.98,  # 실제로는 교차 검증 필요
            timeliness=0.90,  # 실제로는 수집 시점 기반 계산
            validity=self.check_validity(validity_rules or {}),
            uniqueness=self.check_uniqueness(),
            representativeness=self.check_representativeness(
                protected_column or "", expected_dist or {}
            ),
            bias_score=self.check_bias(
                label_column or "", protected_columns or []
            )
        )

# 사용 예시
df = pd.DataFrame({
    "age": [25, 30, None, 45, 35],
    "gender": ["M", "F", "M", "F", "M"],
    "income": [50000, 60000, 55000, 70000, 45000],
    "approved": [1, 1, 0, 1, 0]
})

checker = DataQualityChecker(df, "대출심사데이터셋_v1")
report = checker.generate_report(
    validity_rules={"age": lambda x: 18 <= x <= 100 if pd.notna(x) else False},
    protected_column="gender",
    expected_dist={"M": 0.5, "F": 0.5},
    label_column="approved",
    protected_columns=["gender"]
)

print(f"종합 품질 점수: {report.overall_score:.2f}")
print(f"AI Act 준수 여부: {report.is_ai_act_compliant()}")
print(f"성별 편향 점수: {report.bias_score.get('gender', 0):.3f}")