Privacy-preserving ML будує моделі що навчаються на чутливих даних захищаючи індивідуальну приватність. З регуляціями як GDPR та зростаючою обізнаністю користувачів, приватність не опціональна - це вимога. Виклик полягає в підтримці корисності моделі при наданні значущих гарантій приватності.

Де використовується

Privacy-preserving ML дозволяє чутливі застосування:

Охорона здоров'я: Тренування на даних пацієнтів між лікарнями
Фінанси: Виявлення шахрайства без sharing транзакцій
Мобільні пристрої: Передбачення клавіатури, голосові моделі на пристрої
Реклама: Агрегована аналітика без індивідуального tracking
Уряд: Дані перепису, статистика громадського здоров'я
Дослідження: Sharing результатів без розкриття суб'єктів
IoT: Smart home навчання без хмарного exposure

Огляд технік приватності

Техніка	Що захищає	Компроміс
Диференційна приватність	Статистичні запити	Приватність vs точність
Федеративне навчання	Місце розташування сирих даних	Комунікаційна вартість
Secure MPC	Входи обчислень	Обчислювальний overhead
Гомоморфне шифрування	Дані під час обчислень	Продуктивність
Синтетичні дані	Оригінальні записи	Збереження корисності
K-Anonymity	Прямі ідентифікатори	Стійкість до атак

Що ви робите на практиці

Оцініть вимоги приватності - Регуляції, очікування користувачів
Ідентифікуйте чутливі дані - PII, квазі-ідентифікатори
Оберіть модель приватності - Central vs local, модель загроз
Встановіть бюджет приватності - Epsilon, delta параметри
Імплементуйте механізми - Шум, агрегація, шифрування
Валідуйте корисність - Продуктивність моделі з приватністю
Аудит та тестування - Симуляції атак, membership inference
Документуйте та комунікуйте - Privacy notices, гарантії

Диференційна приватність

import numpy as np
from typing import Callable, Tuple
import torch
import torch.nn as nn

class DifferentialPrivacy:
    """Ядро механізмів диференційної приватності."""

    def __init__(self, epsilon: float, delta: float = 1e-5):
        """
        Ініціалізація DP механізму.

        Args:
            epsilon: Бюджет приватності (нижче = приватніше)
            delta: Ймовірність порушення приватності
        """
        self.epsilon = epsilon
        self.delta = delta

    def laplace_mechanism(self, true_value: float, 
                          sensitivity: float) -> float:
        """Додавання Laplace шуму для epsilon-DP."""
        scale = sensitivity / self.epsilon
        noise = np.random.laplace(0, scale)
        return true_value + noise

    def gaussian_mechanism(self, true_value: float,
                          sensitivity: float) -> float:
        """Додавання Gaussian шуму для (epsilon, delta)-DP."""
        sigma = sensitivity * np.sqrt(2 * np.log(1.25 / self.delta)) / self.epsilon
        noise = np.random.normal(0, sigma)
        return true_value + noise

    def private_mean(self, data: np.ndarray, 
                     lower: float, upper: float) -> float:
        """Обчислення диференційно приватного середнього."""
        n = len(data)
        clipped = np.clip(data, lower, upper)
        true_mean = np.mean(clipped)

        # Чутливість середнього
        sensitivity = (upper - lower) / n

        return self.laplace_mechanism(true_mean, sensitivity)

    def private_count(self, data: np.ndarray, 
                      condition: Callable) -> float:
        """Обчислення диференційно приватного підрахунку."""
        true_count = sum(1 for x in data if condition(x))
        sensitivity = 1  # Додавання/видалення однієї особи змінює count на 1
        return self.laplace_mechanism(true_count, sensitivity)

    def private_histogram(self, data: np.ndarray, 
                          bins: int) -> np.ndarray:
        """Обчислення диференційно приватної гістограми."""
        true_hist, _ = np.histogram(data, bins=bins)

        # Кожен bin має чутливість 1
        noisy_hist = [
            self.laplace_mechanism(count, sensitivity=1)
            for count in true_hist
        ]

        # Post-processing: забезпечення невід'ємності
        return np.maximum(noisy_hist, 0)

class PrivateSGD:
    """Диференційно приватний стохастичний градієнтний спуск."""

    def __init__(self, model: nn.Module, epsilon: float, delta: float,
                 max_grad_norm: float = 1.0, noise_multiplier: float = 1.0):
        self.model = model
        self.epsilon = epsilon
        self.delta = delta
        self.max_grad_norm = max_grad_norm
        self.noise_multiplier = noise_multiplier

    def clip_gradients(self) -> float:
        """Обрізка per-sample градієнтів для обмеження чутливості."""
        total_norm = 0.0

        for param in self.model.parameters():
            if param.grad is not None:
                param_norm = param.grad.data.norm(2)
                total_norm += param_norm.item() ** 2

        total_norm = np.sqrt(total_norm)
        clip_coef = self.max_grad_norm / (total_norm + 1e-6)

        if clip_coef < 1:
            for param in self.model.parameters():
                if param.grad is not None:
                    param.grad.data.mul_(clip_coef)

        return min(total_norm, self.max_grad_norm)

    def add_noise(self, batch_size: int):
        """Додавання каліброваного Gaussian шуму до градієнтів."""
        for param in self.model.parameters():
            if param.grad is not None:
                noise = torch.normal(
                    mean=0,
                    std=self.noise_multiplier * self.max_grad_norm,
                    size=param.grad.shape
                )
                param.grad.data.add_(noise / batch_size)

    def private_step(self, optimizer, batch_size: int):
        """Виконання одного приватного кроку оптимізації."""
        self.clip_gradients()
        self.add_noise(batch_size)
        optimizer.step()

Федеративне навчання

import numpy as np
import torch
import torch.nn as nn
from typing import List, Dict
from copy import deepcopy

class FederatedClient:
    """Локальний клієнт у федеративному навчанні."""

    def __init__(self, client_id: int, local_data: torch.utils.data.Dataset,
                 model: nn.Module, batch_size: int = 32):
        self.client_id = client_id
        self.data_loader = torch.utils.data.DataLoader(
            local_data, batch_size=batch_size, shuffle=True
        )
        self.model = deepcopy(model)

    def local_train(self, global_weights: Dict, 
                    local_epochs: int = 1,
                    learning_rate: float = 0.01) -> Dict:
        """Тренування на локальних даних та повернення оновлень ваг."""
        # Завантаження глобальних ваг
        self.model.load_state_dict(global_weights)

        optimizer = torch.optim.SGD(
            self.model.parameters(), lr=learning_rate
        )
        criterion = nn.CrossEntropyLoss()

        self.model.train()
        for epoch in range(local_epochs):
            for batch_x, batch_y in self.data_loader:
                optimizer.zero_grad()
                outputs = self.model(batch_x)
                loss = criterion(outputs, batch_y)
                loss.backward()
                optimizer.step()

        return self.model.state_dict()

    def get_data_size(self) -> int:
        """Повернення кількості локальних семплів."""
        return len(self.data_loader.dataset)

class FederatedServer:
    """Центральний сервер для federated averaging."""

    def __init__(self, model: nn.Module, clients: List[FederatedClient]):
        self.global_model = model
        self.clients = clients

    def aggregate_weights(self, client_weights: List[Dict],
                         client_sizes: List[int]) -> Dict:
        """Federated averaging ваг клієнтів."""
        total_size = sum(client_sizes)

        # Ініціалізація агрегованих ваг
        aggregated = {}
        for key in client_weights[0].keys():
            aggregated[key] = torch.zeros_like(client_weights[0][key])

        # Зважене середнє
        for weights, size in zip(client_weights, client_sizes):
            weight_factor = size / total_size
            for key in aggregated.keys():
                aggregated[key] += weights[key] * weight_factor

        return aggregated

    def train_round(self, fraction_clients: float = 0.1,
                    local_epochs: int = 1) -> float:
        """Виконання одного раунду федеративного навчання."""
        # Вибір випадкової підмножини клієнтів
        num_selected = max(1, int(len(self.clients) * fraction_clients))
        selected = np.random.choice(
            self.clients, num_selected, replace=False
        )

        # Збір локальних оновлень
        global_weights = self.global_model.state_dict()
        client_weights = []
        client_sizes = []

        for client in selected:
            weights = client.local_train(global_weights, local_epochs)
            client_weights.append(weights)
            client_sizes.append(client.get_data_size())

        # Агрегація
        new_weights = self.aggregate_weights(client_weights, client_sizes)
        self.global_model.load_state_dict(new_weights)

        return self._evaluate()

    def train(self, num_rounds: int, **kwargs) -> List[float]:
        """Повне федеративне тренування."""
        metrics = []
        for round_num in range(num_rounds):
            metric = self.train_round(**kwargs)
            metrics.append(metric)
            print(f"Раунд {round_num + 1}: {metric:.4f}")
        return metrics

class SecureAggregation:
    """Secure aggregation для федеративного навчання."""

    def __init__(self, num_clients: int, threshold: int = None):
        self.num_clients = num_clients
        self.threshold = threshold or (num_clients // 2 + 1)

    def generate_masks(self) -> List[np.ndarray]:
        """Генерація pairwise masking seeds."""
        # На практиці використовуйте Diffie-Hellman key exchange
        masks = []
        for i in range(self.num_clients):
            client_mask = np.zeros(self.num_clients)
            for j in range(self.num_clients):
                if i < j:
                    seed = np.random.randint(0, 2**32)
                    client_mask[j] = seed
                elif i > j:
                    client_mask[j] = -masks[j][i]  # Canceling mask
            masks.append(client_mask)
        return masks

    def aggregate_secure(self, masked_updates: List[np.ndarray]) -> np.ndarray:
        """Сума masked updates (masks відміняються)."""
        return sum(masked_updates) / len(masked_updates)

Компроміси приватність-корисність

class PrivacyUtilityAnalyzer:
    """Аналіз компромісів приватність-корисність."""

    def epsilon_sweep(self, train_func: Callable, 
                      epsilons: List[float],
                      num_trials: int = 5) -> Dict:
        """Оцінка продуктивності моделі на різних рівнях приватності."""
        results = {'epsilon': [], 'accuracy': [], 'std': []}

        for eps in epsilons:
            trial_accuracies = []

            for _ in range(num_trials):
                accuracy = train_func(epsilon=eps)
                trial_accuracies.append(accuracy)

            results['epsilon'].append(eps)
            results['accuracy'].append(np.mean(trial_accuracies))
            results['std'].append(np.std(trial_accuracies))

        return results

    def _interpret_epsilon(self, epsilon: float) -> str:
        """Людино-читабельна інтерпретація epsilon."""
        if epsilon < 0.1:
            return "Дуже сильна приватність - мінімальний витік інформації"
        elif epsilon < 1.0:
            return "Сильна приватність - добрий баланс для чутливих даних"
        elif epsilon < 5.0:
            return "Помірна приватність - прийнятно для багатьох застосувань"
        else:
            return "Слабка приватність - розгляньте нижчий epsilon"

Membership Inference Attack

class MembershipInferenceAttack:
    """Тестування приватності моделі через membership inference."""

    def __init__(self, target_model: nn.Module):
        self.target_model = target_model

    def train_attack_model(self, member_data, non_member_data):
        """Тренування attack model для розрізнення members."""
        member_features = self._extract_features(member_data, label=1)
        non_member_features = self._extract_features(non_member_data, label=0)

        X = np.vstack([member_features, non_member_features])
        y = np.concatenate([
            np.ones(len(member_features)),
            np.zeros(len(non_member_features))
        ])

        from sklearn.linear_model import LogisticRegression
        self.attack_model = LogisticRegression()
        self.attack_model.fit(X, y)

    def _extract_features(self, data, label):
        """Екстракція фіч для attack model."""
        features = []

        self.target_model.eval()
        with torch.no_grad():
            for x, y in data:
                output = self.target_model(x.unsqueeze(0))
                probs = torch.softmax(output, dim=1).numpy()[0]

                feature = [
                    probs[y],  # Confidence на true label
                    -np.sum(probs * np.log(probs + 1e-10)),  # Entropy
                    np.max(probs),  # Max confidence
                ]
                features.append(feature)

        return np.array(features)

    def attack_accuracy(self, test_members, test_non_members) -> float:
        """Оцінка успішності атаки."""
        member_features = self._extract_features(test_members, label=1)
        non_member_features = self._extract_features(test_non_members, label=0)

        X = np.vstack([member_features, non_member_features])
        y_true = np.concatenate([
            np.ones(len(member_features)),
            np.zeros(len(non_member_features))
        ])

        y_pred = self.attack_model.predict(X)
        return np.mean(y_pred == y_true)

Типові помилки та чек-лист

[ ] Вичерпання бюджету приватності: Epsilon занадто високий після багатьох запитів
[ ] Колапс корисності: Занадто багато шуму для корисної моделі
[ ] Вибух градієнтів: Обрізка занадто агресивна
[ ] Гетерогенність клієнтів: Non-IID дані у федеративному навчанні
[ ] Комунікаційне вузьке місце: Повільні федеративні раунди
[ ] Side channels: Витоки timing або іншої інформації
[ ] Атаки реконструкції: Меморизація моделі

Чек-лист для продакшену

Перед деплоєм: - [ ] Визначення вимог приватності та моделі загроз - [ ] Обчислення та документація бюджету приватності - [ ] Тестування з membership inference атаками - [ ] Валідація корисності з приватним тренуванням - [ ] Legal/compliance review

Після деплою: - [ ] Відстеження кумулятивного бюджету приватності - [ ] Моніторинг нових технік атак - [ ] Регулярні аудити приватності - [ ] Оновлення механізмів за потреби - [ ] Контролі приватності користувача та прозорість

Інструменти та стек

Категорія	Інструменти
Диференційна приватність	Opacus (PyTorch), TensorFlow Privacy
Федеративне навчання	Flower, PySyft, TensorFlow Federated
Secure Computation	CrypTen, PySEAL
Аудит приватності	ML Privacy Meter, ART
Compliance	OneTrust, BigID

Приклад міні-кейсу

Сценарій: Федеративна модель здоров'я між лікарнями.

Підхід: - Федеративне навчання (дані залишаються локально) - Диференційна приватність для захисту градієнтів - Secure aggregation для оновлень моделі - Epsilon = 2.0 за раунд тренування

Результати: - Точність моделі: 89% (vs 92% централізована) - Гарантія приватності: (2.0, 10^-5)-DP за раунд - Нуль sharing даних між лікарнями - Досягнуто regulatory compliance

Пов'язані сторінки

Індекс: ML Field Guide
ML Fairness
Стиснення моделей
Інференс на пристрої

Додаткові матеріали

Спробуйте наживо

Дослідіть пов'язану модель на практиці: відкрийте Privacy-Preserving AI: Federated Learning Simulator і змінюйте параметри під час роботи. Усе лишається у вашому браузері.

▶ Відкрити симуляцію Privacy-Preserving AI: Federated Learning Simulator

Машинне навчання зі збереженням приватності