7장: 프라이버시 보존 합성 데이터

import numpy as np
from dataclasses import dataclass
 
 
@dataclass
class DPConfig:
    epsilon: float = 1.0         # 프라이버시 예산
    delta: float = 1e-5          # 실패 확률
    max_grad_norm: float = 1.0   # 그래디언트 클리핑 노름
    noise_multiplier: float = 1.1 # 노이즈 배수
    num_epochs: int = 50
    batch_size: int = 256
 
 
def add_dp_noise(
    gradients: np.ndarray,
    config: DPConfig,
) -> np.ndarray:
    """그래디언트에 차등 프라이버시 노이즈를 추가합니다."""
    # 1단계: 그래디언트 클리핑
    grad_norm = np.linalg.norm(gradients)
    if grad_norm > config.max_grad_norm:
        gradients = gradients * (config.max_grad_norm / grad_norm)
 
    # 2단계: 가우시안 노이즈 추가
    noise_scale = config.noise_multiplier * config.max_grad_norm
    noise = np.random.normal(0, noise_scale, gradients.shape)
 
    return gradients + noise
 
 
def compute_privacy_budget(
    noise_multiplier: float,
    num_steps: int,
    batch_size: int,
    dataset_size: int,
    delta: float = 1e-5,
) -> float:
    """Renyi DP 기반 프라이버시 예산을 계산합니다."""
    # 실제 구현에서는 opacus 또는 dp-accounting 라이브러리 사용
    sampling_rate = batch_size / dataset_size
    # 근사값 (정확한 계산은 라이브러리 필요)
    epsilon_estimate = (
        noise_multiplier
        * np.sqrt(2 * np.log(1.25 / delta))
        * sampling_rate
        * np.sqrt(num_steps)
    )
    return epsilon_estimate

import re
from dataclasses import dataclass
from enum import Enum
 
 
class PIIType(Enum):
    PHONE = "phone"
    EMAIL = "email"
    SSN = "ssn"             # 주민등록번호
    CREDIT_CARD = "credit_card"
    NAME = "name"
    ADDRESS = "address"
    IP_ADDRESS = "ip_address"
 
 
@dataclass
class PIIEntity:
    text: str
    pii_type: PIIType
    start: int
    end: int
    confidence: float
 
 
# 한국어 PII 패턴
PII_PATTERNS = {
    PIIType.PHONE: [
        r"01[016789]-?\d{3,4}-?\d{4}",
        r"02-?\d{3,4}-?\d{4}",
        r"0\d{1,2}-?\d{3,4}-?\d{4}",
    ],
    PIIType.EMAIL: [
        r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}",
    ],
    PIIType.SSN: [
        r"\d{6}-?[1-4]\d{6}",
    ],
    PIIType.CREDIT_CARD: [
        r"\d{4}-?\d{4}-?\d{4}-?\d{4}",
    ],
    PIIType.IP_ADDRESS: [
        r"\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}",
    ],
}
 
 
def detect_pii_regex(text: str) -> list[PIIEntity]:
    """정규식 기반으로 PII를 탐지합니다."""
    entities = []
 
    for pii_type, patterns in PII_PATTERNS.items():
        for pattern in patterns:
            for match in re.finditer(pattern, text):
                entities.append(PIIEntity(
                    text=match.group(),
                    pii_type=pii_type,
                    start=match.start(),
                    end=match.end(),
                    confidence=0.9,
                ))
 
    return entities
 
 
def mask_pii(text: str, entities: list[PIIEntity]) -> str:
    """탐지된 PII를 마스킹합니다."""
    # 뒤에서부터 치환하여 인덱스 밀림 방지
    sorted_entities = sorted(entities, key=lambda e: e.start, reverse=True)
 
    masked = text
    for entity in sorted_entities:
        mask_token = f"[{entity.pii_type.value.upper()}]"
        masked = masked[:entity.start] + mask_token + masked[entity.end:]
 
    return masked

def detect_pii_with_llm(text: str, model_fn) -> list[PIIEntity]:
    """LLM을 활용하여 PII를 탐지합니다."""
    prompt = f"""다음 텍스트에서 개인 식별 정보(PII)를 모두 찾아주세요.
 
PII 유형: 이름, 전화번호, 이메일, 주민등록번호, 주소,
         신용카드번호, IP 주소, 계좌번호
 
텍스트: {text}
 
JSON 형식으로 출력:
[{{"text": "탐지된 PII", "type": "PII 유형", "start": 시작위치, "end": 끝위치}}]"""
 
    result = model_fn(prompt)
    # 파싱 및 PIIEntity 변환
    return parse_pii_result(result)

import numpy as np
from sklearn.metrics import roc_auc_score
from sklearn.neighbors import NearestNeighbors
 
 
def membership_inference_test(
    synthetic_data: np.ndarray,
    training_data: np.ndarray,
    holdout_data: np.ndarray,
    k: int = 5,
) -> dict:
    """멤버십 추론 공격 시뮬레이션으로 프라이버시를 평가합니다."""
    nn_model = NearestNeighbors(n_neighbors=k, metric="euclidean")
    nn_model.fit(synthetic_data)
 
    # 학습 데이터(멤버)의 최근접 거리
    member_distances, _ = nn_model.kneighbors(training_data)
    member_avg_dist = member_distances.mean(axis=1)
 
    # 홀드아웃 데이터(비멤버)의 최근접 거리
    nonmember_distances, _ = nn_model.kneighbors(holdout_data)
    nonmember_avg_dist = nonmember_distances.mean(axis=1)
 
    # 라벨: 멤버=1, 비멤버=0
    labels = np.concatenate([
        np.ones(len(member_avg_dist)),
        np.zeros(len(nonmember_avg_dist)),
    ])
 
    # 거리가 짧을수록 멤버일 가능성이 높으므로 부호 반전
    scores = np.concatenate([
        -member_avg_dist,
        -nonmember_avg_dist,
    ])
 
    # AUC 계산: 0.5에 가까울수록 안전
    auc = roc_auc_score(labels, scores)
 
    return {
        "mia_auc": auc,
        "privacy_risk": "high" if auc > 0.6 else "low",
        "member_mean_distance": member_avg_dist.mean(),
        "nonmember_mean_distance": nonmember_avg_dist.mean(),
    }

import numpy as np
from sklearn.neighbors import NearestNeighbors
 
 
def similarity_filter(
    synthetic: np.ndarray,
    real: np.ndarray,
    min_distance: float = 0.1,
) -> tuple[np.ndarray, np.ndarray]:
    """원본과 너무 유사한 합성 데이터를 필터링합니다."""
    nn = NearestNeighbors(n_neighbors=1, metric="euclidean")
    nn.fit(real)
 
    distances, indices = nn.kneighbors(synthetic)
    distances = distances.flatten()
 
    # 최소 거리 이상인 데이터만 유지
    safe_mask = distances >= min_distance
    safe_synthetic = synthetic[safe_mask]
    filtered_synthetic = synthetic[~safe_mask]
 
    print(
        f"필터링 결과: {safe_mask.sum()}/{len(synthetic)} 유지 "
        f"({(~safe_mask).sum()}건 제거)"
    )
 
    return safe_synthetic, filtered_synthetic

from dataclasses import dataclass
 
 
@dataclass
class PrivacyUtilityResult:
    epsilon: float
    tstr_score: float
    mia_auc: float
    pii_leak_rate: float
 
 
def find_optimal_privacy_level(
    real_data,
    epsilon_candidates: list[float] = [0.1, 0.5, 1.0, 2.0, 5.0, 10.0],
    min_tstr_threshold: float = 0.90,
    max_mia_auc_threshold: float = 0.55,
) -> PrivacyUtilityResult | None:
    """최적의 프라이버시 수준을 탐색합니다."""
    results: list[PrivacyUtilityResult] = []
 
    for epsilon in epsilon_candidates:
        # 1. DP 적용 합성 데이터 생성
        synthetic = generate_dp_synthetic(real_data, epsilon)
 
        # 2. 유용성 평가
        tstr = evaluate_tstr(synthetic, real_data)
 
        # 3. 프라이버시 평가
        mia = membership_inference_test(synthetic, real_data)
        pii_rate = detect_pii_leak_rate(synthetic, real_data)
 
        results.append(PrivacyUtilityResult(
            epsilon=epsilon,
            tstr_score=tstr,
            mia_auc=mia["mia_auc"],
            pii_leak_rate=pii_rate,
        ))
 
    # 조건을 만족하는 결과 중 가장 강한 프라이버시 선택
    valid = [
        r for r in results
        if r.tstr_score >= min_tstr_threshold
        and r.mia_auc <= max_mia_auc_threshold
    ]
 
    if valid:
        return min(valid, key=lambda r: r.epsilon)
    return None