differential-privacy-prod

生产环境中的差分隐私

概述

差分隐私是一种数学框架，用于量化并限制发布数据集统计信息时造成的隐私损失。它提供可证明的保证：计算结果不会显著取决于是否包含任何单个个体的数据。本技能介绍将差分隐私系统部署到生产环境中的实用工程方法。

核心定义

(epsilon, delta)-差分隐私

如果对于所有相邻数据集 D 和 D'（最多相差一条记录）以及所有可能的输出 S，随机机制 M 满足以下条件，则称 M 满足 (epsilon, delta)-差分隐私：

P[M(D) in S] <= e^epsilon * P[M(D') in S] + delta

其中：

• epsilon（隐私损失预算）：量化最大信息泄露量。epsilon 越低，隐私保护越强。
• delta：超出 epsilon 保证范围的额外隐私泄露概率。应小到具有密码学级别的安全性（< 1/n^2，其中 n 是数据集大小）。

敏感度

函数 f 的敏感度用于衡量单个个体对输出的影响程度：

• 全局敏感度：所有相邻数据集 D、D' 的 |f(D) - f(D')| 最大值
• 局部敏感度：对于特定的 D，所有相邻数据集 D' 的 |f(D) - f(D')| 最大值

Epsilon 选择策略

不同使用场景的 Epsilon 指南

使用场景	Epsilon 范围	原因
人口普查／政府统计	0.1 - 1.0	为强制参与提供最大程度的保护
医疗保健分析	0.5 - 2.0	高敏感度，且有监管要求
位置分析	1.0 - 3.0	中等敏感度，关注汇总洞察
产品分析	1.0 - 5.0	敏感度较低，有业务效用需求
A/B 测试	2.0 - 8.0	需要满足统计显著性要求
汇总报告	0.5 - 3.0	面向公众的输出需要更强的保证

影响 Epsilon 选择的因素

1. 数据敏感度：数据越敏感，所需的 epsilon 越低
2. 数据集大小：数据集越大，在效用可接受的情况下可容忍更低的 epsilon
3. 查询频率：查询越多，消耗的预算越多（参见组合）
4. 监管要求：某些法规意味着特定的隐私级别
5. 效用要求：业务需求可能会设定可接受准确度的下限
6. 受众：公开发布相比内部分析需要更强的保证

噪声机制

Laplace 机制

适用于全局敏感度有界的数值查询，可提供纯 epsilon-差分隐私。

import numpy as np

def laplace_mechanism(true_value: float, sensitivity: float, epsilon: float) -> float:
    """
    Apply Laplace noise for epsilon-differential privacy.

    Args:
        true_value: The true query result
        sensitivity: Global sensitivity of the query (L1)
        epsilon: Privacy parameter

    Returns:
        Noisy result satisfying epsilon-differential privacy
    """
    scale = sensitivity / epsilon
    noise = np.random.laplace(loc=0, scale=scale)
    return true_value + noise


def laplace_mechanism_vector(true_values: np.ndarray, sensitivity: float, epsilon: float) -> np.ndarray:
    """Apply Laplace noise to a vector of values."""
    scale = sensitivity / epsilon
    noise = np.random.laplace(loc=0, scale=scale, size=true_values.shape)
    return true_values + noise

Gaussian 机制

适用于数值查询。可提供 (epsilon, delta)-差分隐私，并在高维输出中使用更少的噪声。

import numpy as np
import math

def gaussian_mechanism(true_value: float, sensitivity: float, epsilon: float, delta: float) -> float:
    """
    Apply Gaussian noise for (epsilon, delta)-differential privacy.

    Uses the analytic Gaussian mechanism calibration.

    Args:
        true_value: The true query result
        sensitivity: Global sensitivity of the query (L2)
        epsilon: Privacy parameter
        delta: Failure probability parameter

    Returns:
        Noisy result satisfying (epsilon, delta)-differential privacy
    """
    sigma = sensitivity * math.sqrt(2 * math.log(1.25 / delta)) / epsilon
    noise = np.random.normal(loc=0, scale=sigma)
    return true_value + noise

Exponential 机制

适用于非数值输出（类别选择）；在这种情况下，直接添加噪声没有意义。

import numpy as np

def exponential_mechanism(
    candidates: list,
    utility_scores: np.ndarray,
    sensitivity: float,
    epsilon: float
) -> object:
    """
    Select an output using the exponential mechanism.

    Args:
        candidates: List of possible outputs
        utility_scores: Utility score for each candidate
        sensitivity: Global sensitivity of the utility function
        epsilon: Privacy parameter

    Returns:
        Selected candidate satisfying epsilon-differential privacy
    """
    # Calculate selection probabilities
    probabilities = np.exp(epsilon * utility_scores / (2 * sensitivity))
    probabilities = probabilities / probabilities.sum()

    # Sample according to probabilities
    index = np.random.choice(len(candidates), p=probabilities)
    return candidates[index]

随机响应（本地差分隐私）

适用于以本地差分隐私方式收集个体数据点。

import random
import math

def randomized_response(true_bit: bool, epsilon: float) -> bool:
    """
    Apply randomized response for local differential privacy.

    Args:
        true_bit: The individual's true binary response
        epsilon: Privacy parameter

    Returns:
        Randomized response satisfying epsilon-local-DP
    """
    p = math.exp(epsilon) / (math.exp(epsilon) + 1)

    if random.random() < p:
        return true_bit  # Report truthfully
    else:
        return not true_bit  # Flip the answer


def estimate_from_randomized_responses(
    responses: list,
    epsilon: float
) -> float:
    """
    Estimate true proportion from randomized responses.

    Args:
        responses: List of randomized boolean responses
        epsilon: The epsilon used during collection

    Returns:
        Estimated true proportion
    """
    n = len(responses)
    observed_proportion = sum(responses) / n
    p = math.exp(epsilon) / (math.exp(epsilon) + 1)

    # Correct for randomization bias
    estimated_proportion = (observed_proportion - (1 - p)) / (2 * p - 1)
    return max(0.0, min(1.0, estimated_proportion))

隐私预算跟踪

预算管理器实现

import threading
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
import math


@dataclass
class BudgetAllocation:
    query_id: str
    epsilon_spent: float
    delta_spent: float
    timestamp: datetime
    query_description: str
    analyst_id: str


class PrivacyBudgetManager:
    """
    Track and enforce differential privacy budget across queries.

    Supports both basic and advanced composition theorems.
    """

    def __init__(
        self,
        total_epsilon: float,
        total_delta: float,
        composition_method: str = "advanced"
    ):
        self.total_epsilon = total_epsilon
        self.total_delta = total_delta
        self.composition_method = composition_method
        self.allocations: list[BudgetAllocation] = []
        self._lock = threading.Lock()

    def remaining_budget(self) -> tuple[float, float]:
        """Calculate remaining (epsilon, delta) budget."""
        if self.composition_method == "basic":
            return self._basic_composition_remaining()
        elif self.composition_method == "advanced":
            return self._advanced_composition_remaining()
        elif self.composition_method == "rdp":
            return self._rdp_composition_remaining()
        else:
            raise ValueError(f"Unknown composition method: {self.composition_method}")

    def _basic_composition_remaining(self) -> tuple[float, float]:
        """Basic sequential composition: epsilons and deltas sum."""
        spent_epsilon = sum(a.epsilon_spent for a in self.allocations)
        spent_delta = sum(a.delta_spent for a in self.allocations)
        return (self.total_epsilon - spent_epsilon, self.total_delta - spent_delta)

    def _advanced_composition_remaining(self) -> tuple[float, float]:
        """
        Advanced composition theorem:
        k queries each with epsilon_i satisfy
        (sqrt(2k * ln(1/delta')) * max(epsilon_i) + k * epsilon_i * (e^epsilon_i - 1),
         k * delta_i + delta')-DP
        """
        k = len(self.allocations)
        if k == 0:
            return (self.total_epsilon, self.total_delta)

        epsilons = [a.epsilon_spent for a in self.allocations]
        deltas = [a.delta_spent for a in self.allocations]

        max_eps = max(epsilons)
        sum_delta = sum(deltas)

        # Reserve delta' for composition overhead
        delta_prime = (self.total_delta - sum_delta) / 2
        if delta_prime <= 0:
            return (0.0, 0.0)

        composed_epsilon = (
            math.sqrt(2 * k * math.log(1 / delta_prime)) * max_eps
            + k * max_eps * (math.exp(max_eps) - 1)
        )
        composed_delta = sum_delta + delta_prime

        return (
            max(0.0, self.total_epsilon - composed_epsilon),
            max(0.0, self.total_delta - composed_delta)
        )

    def _rdp_composition_remaining(self) -> tuple[float, float]:
        """Renyi DP composition (simplified)."""
        # RDP provides tighter bounds through Renyi divergence
        spent_epsilon = sum(a.epsilon_spent for a in self.allocations)
        spent_delta = sum(a.delta_spent for a in self.allocations)
        return (self.total_epsilon - spent_epsilon, self.total_delta - spent_delta)

    def request_budget(
        self,
        query_id: str,
        epsilon_requested: float,
        delta_requested: float,
        query_description: str,
        analyst_id: str
    ) -> bool:
        """
        Request budget allocation for a query.

        Returns True if budget is available and allocated, False otherwise.
        """
        with self._lock:
            remaining_eps, remaining_delta = self.remaining_budget()

            if epsilon_requested > remaining_eps or delta_requested > remaining_delta:
                return False

            allocation = BudgetAllocation(
                query_id=query_id,
                epsilon_spent=epsilon_requested,
                delta_spent=delta_requested,
                timestamp=datetime.utcnow(),
                query_description=query_description,
                analyst_id=analyst_id
            )
            self.allocations.append(allocation)
            return True

    def get_usage_report(self) -> dict:
        """Generate budget usage report."""
        remaining_eps, remaining_delta = self.remaining_budget()
        return {
            "total_epsilon": self.total_epsilon,
            "total_delta": self.total_delta,
            "remaining_epsilon": remaining_eps,
            "remaining_delta": remaining_delta,
            "utilization_pct": (1 - remaining_eps / self.total_epsilon) * 100,
            "num_queries": len(self.allocations),
            "composition_method": self.composition_method,
            "allocations": [
                {
                    "query_id": a.query_id,
                    "epsilon": a.epsilon_spent,
                    "delta": a.delta_spent,
                    "timestamp": a.timestamp.isoformat(),
                    "analyst": a.analyst_id,
                }
                for a in self.allocations
            ],
        }

组合定理

基本组合

如果 M1 满足 (e1, d1)-DP，且 M2 满足 (e2, d2)-DP，那么同时发布 M1(D) 和 M2(D) 满足 (e1 + e2, d1 + d2)-DP。

高级组合

对于 k 个各自满足 (epsilon, delta)-DP 的机制，组合后的机制满足 (epsilon', k*delta + delta')-DP，其中：

epsilon' = sqrt(2k * ln(1/delta')) * epsilon + k * epsilon * (e^epsilon - 1)

Renyi 差分隐私（RDP）组合

RDP 通过跟踪隐私损失的 Renyi 散度（阶数为 alpha）来提供更紧的组合界限。最后将其转换为 (epsilon, delta)-DP：

For alpha > 1: epsilon(delta) = RDP_alpha - ln(delta) / (alpha - 1)

Moments Accountant

由 TensorFlow Privacy 和 Opacus 使用。它会跟踪隐私损失变量的矩母函数的对数，并为迭代机制（例如 DP-SGD）提供已知最紧的界限。

生产部署模式

模式 1：集中式 DP 分析管道

Raw Data Store --> Sensitivity Calibration --> DP Mechanism --> Result Cache
                         |                         |
                         v                         v
                  Budget Manager <------------- Audit Log

模式 2：本地 DP 数据收集

Client Device --> Local Randomizer --> Aggregation Server --> Estimator
                  (epsilon-LDP)              |                    |
                                             v                    v
                                       Budget Tracker      Utility Monitor

模式 3：差分隐私模型训练（DP-SGD）

Training Data --> Mini-batch Sampling --> Per-example Gradient
                  (Poisson sampling)            |
                                                v
                                        Gradient Clipping (norm bound C)
                                                |
                                                v
                                        Gaussian Noise Addition (sigma * C)
                                                |
                                                v
                                        Model Update --> Privacy Accountant

效用度量

指标	描述	公式
平均绝对误差	与真实值之间的平均绝对差值	MAE = (1/n) * sum(
相对误差	误差占真实值的比例	RE =
覆盖率	置信区间内真实值所占的比例	Count(true in CI) / n
效用比率	噪声信号噪声比与真实信号噪声比之比	SNR_noisy / SNR_true

参考资料

• Dwork, C. 和 Roth, A.《差分隐私的算法基础》。理论计算机科学前沿与趋势，9(3-4):211-407，2014。
• Mironov, I.《Rényi 差分隐私》。IEEE CSF，2017。
• Abadi, M. 等。《深度学习与差分隐私》。ACM CCS，2016。
• Google 差分隐私库：github.com/google/differential-privacy
• OpenDP 项目：opendp.org
• Apple 差分隐私技术概览（2017 年）
• 美国人口普查局披露规避系统（2020 年人口普查）