Privacy-preserving ML builds models that learn from sensitive data while protecting individual privacy. With regulations like GDPR and growing user awareness, privacy isn't optional - it's a requirement. The challenge is maintaining model utility while providing meaningful privacy guarantees.

Where It Is Used

Privacy-preserving ML enables sensitive applications:

Healthcare: Training on patient data across hospitals
Finance: Fraud detection without sharing transactions
Mobile devices: Keyboard prediction, voice models on-device
Advertising: Aggregate analytics without individual tracking
Government: Census data, public health statistics
Research: Sharing findings without exposing subjects
IoT: Smart home learning without cloud exposure

Privacy Techniques Overview

Technique	What It Protects	Trade-off
Differential Privacy	Statistical queries	Privacy vs accuracy
Federated Learning	Raw data location	Communication cost
Secure MPC	Computation inputs	Computational overhead
Homomorphic Encryption	Data during computation	Performance
Synthetic Data	Original records	Utility preservation
K-Anonymity	Direct identifiers	Attack resistance

What You Do in Practice

Assess privacy requirements - Regulations, user expectations
Identify sensitive data - PII, quasi-identifiers
Choose privacy model - Central vs local, threat model
Set privacy budget - Epsilon, delta parameters
Implement mechanisms - Noise, aggregation, encryption
Validate utility - Model performance with privacy
Audit and test - Attack simulations, membership inference
Document and communicate - Privacy notices, guarantees

Differential Privacy

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

class DifferentialPrivacy:
    """Core differential privacy mechanisms."""

    def __init__(self, epsilon: float, delta: float = 1e-5):
        """
        Initialize DP mechanism.

        Args:
            epsilon: Privacy budget (lower = more private)
            delta: Probability of privacy breach
        """
        self.epsilon = epsilon
        self.delta = delta

    def laplace_mechanism(self, true_value: float, 
                          sensitivity: float) -> float:
        """Add Laplace noise for 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:
        """Add Gaussian noise for (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:
        """Compute differentially private mean."""
        n = len(data)
        clipped = np.clip(data, lower, upper)
        true_mean = np.mean(clipped)

        # Sensitivity of mean
        sensitivity = (upper - lower) / n

        return self.laplace_mechanism(true_mean, sensitivity)

    def private_count(self, data: np.ndarray, 
                      condition: Callable) -> float:
        """Compute differentially private count."""
        true_count = sum(1 for x in data if condition(x))
        sensitivity = 1  # Adding/removing one person changes count by 1
        return self.laplace_mechanism(true_count, sensitivity)

    def private_histogram(self, data: np.ndarray, 
                          bins: int) -> np.ndarray:
        """Compute differentially private histogram."""
        true_hist, _ = np.histogram(data, bins=bins)

        # Each bin has sensitivity 1
        noisy_hist = [
            self.laplace_mechanism(count, sensitivity=1)
            for count in true_hist
        ]

        # Post-processing: ensure non-negative
        return np.maximum(noisy_hist, 0)

class PrivateSGD:
    """Differentially private stochastic gradient descent."""

    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:
        """Clip per-sample gradients to bound sensitivity."""
        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):
        """Add calibrated Gaussian noise to gradients."""
        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):
        """Perform one private optimization step."""
        self.clip_gradients()
        self.add_noise(batch_size)
        optimizer.step()

    def compute_epsilon(self, num_steps: int, batch_size: int, 
                        dataset_size: int) -> float:
        """Compute actual epsilon using RDP accountant."""
        sampling_probability = batch_size / dataset_size

        # Simplified epsilon calculation
        # In practice, use Opacus RDP accountant
        orders = [1 + x / 10.0 for x in range(1, 100)]

        rdp = []
        for alpha in orders:
            rdp_step = alpha * sampling_probability ** 2 / (2 * self.noise_multiplier ** 2)
            rdp.append(num_steps * rdp_step)

        eps = min(
            rdp_val - np.log(self.delta) / (alpha - 1)
            for rdp_val, alpha in zip(rdp, orders)
        )

        return eps

Federated Learning

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

class FederatedClient:
    """Local client in federated learning."""

    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:
        """Train on local data and return weight updates."""
        # Load global weights
        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 number of local samples."""
        return len(self.data_loader.dataset)

class FederatedServer:
    """Central server for 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 of client weights."""
        total_size = sum(client_sizes)

        # Initialize aggregated weights
        aggregated = {}
        for key in client_weights[0].keys():
            aggregated[key] = torch.zeros_like(client_weights[0][key])

        # Weighted average
        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:
        """Execute one round of federated learning."""
        # Select random subset of clients
        num_selected = max(1, int(len(self.clients) * fraction_clients))
        selected = np.random.choice(
            self.clients, num_selected, replace=False
        )

        # Collect local updates
        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())

        # Aggregate
        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]:
        """Full federated training."""
        metrics = []
        for round_num in range(num_rounds):
            metric = self.train_round(**kwargs)
            metrics.append(metric)
            print(f"Round {round_num + 1}: {metric:.4f}")
        return metrics

    def _evaluate(self) -> float:
        """Evaluate global model (placeholder)."""
        return 0.0  # Implement with test data

class SecureAggregation:
    """Secure aggregation for federated learning."""

    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]:
        """Generate pairwise masking seeds."""
        # In practice, use 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 mask_update(self, update: np.ndarray, mask_seed: int) -> np.ndarray:
        """Mask client update with PRG."""
        np.random.seed(mask_seed)
        mask = np.random.randn(*update.shape)
        return update + mask

    def aggregate_secure(self, masked_updates: List[np.ndarray]) -> np.ndarray:
        """Sum masked updates (masks cancel out)."""
        # Masks are designed to sum to zero across clients
        return sum(masked_updates) / len(masked_updates)

Privacy-Utility Trade-offs

class PrivacyUtilityAnalyzer:
    """Analyze privacy-utility trade-offs."""

    def epsilon_sweep(self, train_func: Callable, 
                      epsilons: List[float],
                      num_trials: int = 5) -> Dict:
        """Evaluate model performance across privacy levels."""
        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 compute_privacy_loss(self, epsilon: float, 
                            delta: float = 1e-5) -> Dict:
        """Interpret privacy guarantee."""
        # Privacy amplification by subsampling
        # and composition theorems

        # Intuition: e^epsilon bounds ratio of probabilities
        ratio_bound = np.exp(epsilon)

        return {
            'epsilon': epsilon,
            'delta': delta,
            'probability_ratio': ratio_bound,
            'interpretation': self._interpret_epsilon(epsilon)
        }

    def _interpret_epsilon(self, epsilon: float) -> str:
        """Human-readable epsilon interpretation."""
        if epsilon < 0.1:
            return "Very strong privacy - minimal information leakage"
        elif epsilon < 1.0:
            return "Strong privacy - good balance for sensitive data"
        elif epsilon < 5.0:
            return "Moderate privacy - acceptable for many applications"
        else:
            return "Weak privacy - consider lower epsilon"

Membership Inference Attack

class MembershipInferenceAttack:
    """Test model privacy via membership inference."""

    def __init__(self, target_model: nn.Module, 
                 shadow_model: nn.Module = None):
        self.target_model = target_model
        self.shadow_model = shadow_model or deepcopy(target_model)

    def train_attack_model(self, member_data, non_member_data):
        """Train attack model to distinguish members."""
        # Get model outputs on training vs non-training data
        member_features = self._extract_features(member_data, label=1)
        non_member_features = self._extract_features(non_member_data, label=0)

        # Train binary classifier
        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):
        """Extract features for 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]

                # Features: prediction confidence, entropy, etc.
                feature = [
                    probs[y],  # Confidence on 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:
        """Evaluate attack success rate."""
        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)

Failure Modes and Debugging Checklist

[ ] Privacy budget exhaustion: Epsilon too high after many queries
[ ] Utility collapse: Too much noise for useful model
[ ] Gradient explosion: Clipping too aggressive
[ ] Client heterogeneity: Non-IID data in federated learning
[ ] Communication bottleneck: Slow federated rounds
[ ] Side channels: Timing or other information leaks
[ ] Reconstruction attacks: Model memorization

Production Checklist

Before deployment: - [ ] Define privacy requirements and threat model - [ ] Compute and document privacy budget - [ ] Test with membership inference attacks - [ ] Validate utility with private training - [ ] Legal/compliance review

After deployment: - [ ] Track cumulative privacy budget - [ ] Monitor for new attack techniques - [ ] Regular privacy audits - [ ] Update mechanisms as needed - [ ] User privacy controls and transparency

Tools and Stack

Category	Tools
Differential Privacy	Opacus (PyTorch), TensorFlow Privacy
Federated Learning	Flower, PySyft, TensorFlow Federated
Secure Computation	CrypTen, PySEAL
Privacy Auditing	ML Privacy Meter, ART
Compliance	OneTrust, BigID

Example Mini-Case

Scenario: Federated health model across hospitals.

Approach: - Federated learning (data stays local) - Differential privacy for gradient protection - Secure aggregation for model updates - Epsilon = 2.0 per training round

Results: - Model accuracy: 89% (vs 92% centralized) - Privacy guarantee: (2.0, 10^-5)-DP per round - Zero data sharing between hospitals - Regulatory compliance achieved

Index: ML Field Guide
ML Fairness
Model Compression
On-Device Inference

Try it live

Everything above runs in your browser — open Privacy-Preserving AI: Federated Learning Simulator and change the parameters while it is running. Nothing is installed, nothing is uploaded, the whole model lives in one tab.

▶ Open Privacy-Preserving AI: Federated Learning Simulator simulation

Privacy-Preserving Machine Learning