차분 프라이버시

# 차분 프라이버시 딥러닝 (DP-SGD with Opacus)
import torch
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset
from opacus import PrivacyEngine
from opacus.validators import ModuleValidator
import numpy as np

# 1. 모델 정의 (DP 호환 가능한 레이어 사용)
class DPCompatibleModel(nn.Module):
    def __init__(self, input_dim, num_classes):
        super().__init__()
        self.layers = nn.Sequential(
            nn.Linear(input_dim, 128),
            nn.ReLU(),
            nn.Linear(128, 64),
            nn.ReLU(),
            nn.Linear(64, num_classes)
        )

    def forward(self, x):
        return self.layers(x)

# 2. 데이터 준비 (예: 민감한 의료 데이터)
np.random.seed(42)
X = torch.randn(1000, 20)
y = torch.randint(0, 2, (1000,))
dataset = TensorDataset(X, y)
dataloader = DataLoader(dataset, batch_size=64, shuffle=True)

# 3. 모델 및 옵티마이저 설정
model = DPCompatibleModel(input_dim=20, num_classes=2)
# DP 호환성 검증 및 수정
model = ModuleValidator.fix(model)
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
criterion = nn.CrossEntropyLoss()

# 4. Privacy Engine 적용
privacy_engine = PrivacyEngine()
model, optimizer, dataloader = privacy_engine.make_private_with_epsilon(
    module=model,
    optimizer=optimizer,
    data_loader=dataloader,
    epochs=10,
    target_epsilon=3.0,      # 프라이버시 예산 (작을수록 강한 보호)
    target_delta=1e-5,       # 허용 실패 확률
    max_grad_norm=1.0,       # 그래디언트 클리핑 임계값
)

print(f"노이즈 승수 (Noise Multiplier): {optimizer.noise_multiplier:.4f}")
print(f"목표 Epsilon: 3.0, Delta: 1e-5")

# 5. DP 학습 실행
def train_with_dp(model, dataloader, optimizer, criterion, epochs=10):
    model.train()
    for epoch in range(epochs):
        total_loss = 0
        for X_batch, y_batch in dataloader:
            optimizer.zero_grad()
            outputs = model(X_batch)
            loss = criterion(outputs, y_batch)
            loss.backward()
            optimizer.step()
            total_loss += loss.item()

        # 현재까지 사용된 프라이버시 예산 확인
        epsilon = privacy_engine.get_epsilon(delta=1e-5)
        print(f"Epoch {epoch+1}: Loss={total_loss:.4f}, Epsilon={epsilon:.2f}")

train_with_dp(model, dataloader, optimizer, criterion, epochs=10)

# 6. 최종 프라이버시 보장 수준
final_epsilon = privacy_engine.get_epsilon(delta=1e-5)
print(f"\n최종 프라이버시 보장: ({final_epsilon:.2f}, 1e-5)-차분 프라이버시")

# 라플라스 메커니즘을 활용한 차분 프라이버시 쿼리
import numpy as np

def laplace_mechanism(true_value, sensitivity, epsilon):
    """
    라플라스 메커니즘으로 차분 프라이버시 적용
    - true_value: 실제 쿼리 결과
    - sensitivity: 쿼리의 민감도 (한 사람이 바뀔 때 최대 변화량)
    - epsilon: 프라이버시 예산
    """
    scale = sensitivity / epsilon
    noise = np.random.laplace(0, scale)
    return true_value + noise

# 예시: 급여 평균 쿼리
salaries = [50000, 55000, 60000, 65000, 70000, 45000, 80000, 75000, 52000, 58000]

# 실제 평균
true_mean = np.mean(salaries)
print(f"실제 평균 급여: ${true_mean:,.0f}")

# 민감도 계산 (한 사람 추가/제거 시 평균 변화 최대치)
# 급여 범위 / n으로 근사
max_salary = 100000
min_salary = 40000
n = len(salaries)
sensitivity = (max_salary - min_salary) / n
print(f"쿼리 민감도: {sensitivity:,.0f}")

# 다양한 epsilon으로 DP 쿼리 수행
print("\n=== Epsilon별 DP 쿼리 결과 ===")
for epsilon in [0.1, 1.0, 5.0, 10.0]:
    dp_results = [laplace_mechanism(true_mean, sensitivity, epsilon) for _ in range(5)]
    print(f"epsilon={epsilon}: ${np.mean(dp_results):,.0f} (std: {np.std(dp_results):,.0f})")

# 프라이버시 vs 유틸리티 트레이드오프
print("\n=> epsilon이 작을수록 프라이버시 보호가 강하지만, 결과의 정확도가 낮아짐")