differential-privacy-prod

本番環境における差分プライバシー

概要

差分プライバシーは、データセットに関する統計情報を公開する際に生じるプライバシー損失を定量化し、上限を定める数学的なフレームワークです。ある個人のデータが含まれるかどうかによって、計算結果が大きく変わらないことを、証明可能な形で保証します。このスキルでは、本番環境へのデプロイに向けた差分プライバシーシステムの実践的なエンジニアリングを扱います。

基本的な定義

(epsilon, delta)-差分プライバシー

ランダム化メカニズム M は、隣接するすべてのデータセット D および D'（最大1レコードのみ異なる）と、考えられるすべての出力 S に対して、次の条件を満たす場合、(epsilon, delta)-差分プライバシーを満たします。

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

ここで：

• epsilon（プライバシー損失予算）：情報漏えいの最大量を定量化します。epsilon が低いほど、プライバシーは強化されます。
• delta：epsilon の保証を超える、さらなるプライバシー侵害が起こる確率です。暗号学的に十分小さい値（< 1/n^2、n はデータセットのサイズ）にする必要があります。

感度

関数 f の感度は、1人の個人が出力にどれほど影響を与えうるかを測定します。

• 大域感度：隣接するすべての 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. 対象者：一般公開には、社内分析より強い保証が必要です

ノイズメカニズム

ラプラスメカニズム

有界な大域感度を持つ数値クエリに使用します。純粋な 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

ガウスメカニズム

数値クエリに使用します。高次元の出力では、より少ないノイズで (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

指数メカニズム

ノイズを直接加えることに意味がない、数値以外の出力（カテゴリの選択）に使用します。

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]

ランダム化応答（ローカル DP）

ローカル差分プライバシーを用いて個人のデータポイントを収集する場合に使用します。

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 を満たします。

高度な合成

それぞれが (epsilon, delta)-DP を満たす k 個のメカニズムを合成した場合、合成メカニズムは (epsilon', k*delta + delta')-DP を満たします。ここで：

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

レニー差分プライバシー（RDP）の合成

RDP は、次数 alpha のレニー発散を用いてプライバシー損失を追跡することで、より厳密な合成境界を提供します。最後に (epsilon, delta)-DP に変換します。

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

モーメントアカウンタ

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 モデルトレーニング（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. and Roth, A.『差分プライバシーのアルゴリズム基盤』。Foundations and Trends in Theoretical Computer Science, 9(3-4):211-407, 2014.
• Mironov, I.「Rényi 差分プライバシー」。IEEE CSF, 2017.
• Abadi, M. et al.「差分プライバシーを用いたディープラーニング」。ACM CCS, 2016.
• Google 差分プライバシーライブラリ：github.com/google/differential-privacy
• OpenDP プロジェクト：opendp.org
• Apple 差分プライバシー技術概要（2017）
• 米国国勢調査局の開示回避システム（2020年国勢調査）