Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions dpsynth/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@
from dpsynth.domain import NumericalAttribute
from dpsynth.domain import OpenSetCategoricalAttribute
from dpsynth.domain import Schema
from dpsynth.reporting import AuditReport
from dpsynth.reporting import PrivacyReport
from dpsynth.serialize import from_yaml
from dpsynth.serialize import to_yaml
Expand All @@ -58,6 +59,7 @@
MultiTableMechanism = relational.MultiTableMechanism

__all__ = [
'AuditReport',
'CategoricalAttribute',
'ForeignKeyRelation',
'MultiDataGenerationResult',
Expand Down
79 changes: 78 additions & 1 deletion dpsynth/reporting.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,11 +19,15 @@
from collections.abc import Callable, Sequence
import dataclasses
import functools
from typing import TypeAlias
from typing import TYPE_CHECKING, TypeAlias

import dp_accounting
import numpy as np

# Guarded by TYPE_CHECKING so jax_privacy remains an optional dependency.
if TYPE_CHECKING:
import jax_privacy # pylint: disable=g-import-not-at-top

Pair: TypeAlias = tuple[float, float]

DEFAULT_TARGET_DELTAS = (1e-10, 1e-9, 1e-8, 1e-7, 1e-6, 1e-5, 1e-4, 1e-3)
Expand Down Expand Up @@ -142,6 +146,79 @@ def from_dp_event(
)


@dataclasses.dataclass(frozen=True)
class AuditReport:
"""Empirical privacy audit summary computed from a CanaryScoreAuditor.

This class complements ``PrivacyReport``, providing an empirical rather than
theoretical measure of privacy loss.
"""

epsilon_deltas: tuple[Pair, ...]
auroc: float
max_accuracy: float
trade_off_curve: tuple[Pair, ...] | None = None
significance: float = 0.05
notes: str | None = None

@classmethod
def from_auditor(
cls,
auditor: jax_privacy.auditing.CanaryScoreAuditor,
*,
significance: float = 0.05,
target_deltas: float | Sequence[float] = DEFAULT_TARGET_DELTAS,
target_false_positive_rates: Sequence[float] | None = DEFAULT_TARGET_FPRS,
notes: str | None = None,
) -> 'AuditReport':
"""Computes an AuditReport from a CanaryScoreAuditor.

Evaluates one-run empirical epsilon lower bounds across multiple delta
values (taking the maximum of the Steinke et al. and f-DP bounds), along
with attack AUROC, maximum accuracy, and the empirical ROC curve (TPR vs.
FPR).

Args:
auditor: The CanaryScoreAuditor to summarize.
significance: Allowed failure probability (1 - confidence) for epsilon.
target_deltas: Evaluation delta(s) for empirical (epsilon, delta)-DP.
target_false_positive_rates: Optional FPRs for the (TPR, FPR) curve.
notes: Optional string capturing custom notes about the audit.

Returns:
An AuditReport containing empirical privacy audit metrics.
"""
if isinstance(target_deltas, (int, float)):
target_deltas = [target_deltas]

assert all(0 < d <= 1 for d in target_deltas), 'target_deltas not in (0, 1]'
assert 0 < significance < 1, 'significance not in (0, 1)'

epsilons = []
for d in target_deltas:
eps_steinke = auditor.epsilon_one_run(significance, d)
eps_fdp = auditor.epsilon_one_run_fdp(significance, d)
epsilons.append(max(eps_steinke, eps_fdp))

auroc = float(auditor.attack_auroc())
max_accuracy = float(auditor.max_accuracy())

trade_off_curve = None
if target_false_positive_rates is not None:
fprs = target_false_positive_rates
tprs = np.asarray(auditor.tpr_at_given_fpr(fprs)).tolist()
trade_off_curve = tuple(zip(tprs, fprs))

return cls(
epsilon_deltas=tuple(zip(epsilons, target_deltas)),
auroc=auroc,
max_accuracy=max_accuracy,
trade_off_curve=trade_off_curve,
significance=significance,
notes=notes,
)


def _get_epsilons(
event: dp_accounting.DpEvent,
make_accountant: Callable[[], dp_accounting.PrivacyAccountant],
Expand Down
50 changes: 50 additions & 0 deletions tests/reporting_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,8 @@
import dpsynth
from dpsynth import reporting
from dpsynth import serialize
import jax_privacy
import numpy as np


class ReportingTest(parameterized.TestCase):
Expand Down Expand Up @@ -150,6 +152,54 @@ def test_serialize_yaml_roundtrip(self):

def test_public_reexport(self):
self.assertIs(dpsynth.PrivacyReport, reporting.PrivacyReport)
self.assertIs(dpsynth.AuditReport, reporting.AuditReport)

def test_from_auditor_defaults_and_yaml_roundtrip(self):
in_scores = np.arange(100, dtype=float) + 20.0
out_scores = np.arange(100, dtype=float)
auditor = jax_privacy.auditing.CanaryScoreAuditor(in_scores, out_scores)
report = reporting.AuditReport.from_auditor(
auditor,
significance=0.05,
notes='Synthetic canary audit.',
)
self.assertEqual(report.significance, 0.05)
self.assertEqual(report.notes, 'Synthetic canary audit.')
self.assertLen(report.epsilon_deltas, len(reporting.DEFAULT_TARGET_DELTAS))
for eps, delta in report.epsilon_deltas:
expected_eps = max(
auditor.epsilon_one_run(0.05, delta),
auditor.epsilon_one_run_fdp(0.05, delta),
)
self.assertAlmostEqual(eps, expected_eps)
self.assertAlmostEqual(report.auroc, float(auditor.attack_auroc()))
self.assertAlmostEqual(report.max_accuracy, float(auditor.max_accuracy()))
self.assertIsNotNone(report.trade_off_curve)
assert report.trade_off_curve is not None
self.assertLen(report.trade_off_curve, len(reporting.DEFAULT_TARGET_FPRS))
for tpr, fpr in report.trade_off_curve:
self.assertAlmostEqual(tpr, float(auditor.tpr_at_given_fpr(fpr)))

yaml_str = serialize.to_yaml(report)
loaded = serialize.from_yaml(yaml_str)
self.assertEqual(loaded, report)

def test_from_auditor_custom_args(self):
auditor = jax_privacy.auditing.CanaryScoreAuditor(
np.arange(50, dtype=float) + 10.0,
np.arange(50, dtype=float),
)
report = reporting.AuditReport.from_auditor(
auditor,
significance=0.1,
target_deltas=1e-5,
target_false_positive_rates=None,
)
self.assertEqual(report.significance, 0.1)
self.assertLen(report.epsilon_deltas, 1)
self.assertEqual(report.epsilon_deltas[0][1], 1e-5)
self.assertIsNone(report.trade_off_curve)
self.assertIsNone(report.notes)


def load_tests(loader, tests, ignore):
Expand Down
Loading