知道很多，却不泄露任何人

math.log1p(...)

import math
import random
import secrets


def laplace_noise(scale, rng=None):
    """
    从Laplace(0, scale)采样。

    默认使用操作系统支持的安全随机源。

    重要：
    这只是教学采样实现，不代表经过
    完整浮点安全性审计的生产级DP机制。
    """

    if scale <= 0 or not math.isfinite(scale):
        raise ValueError("scale必须是正有限数")

    if rng is None:
        rng = secrets.SystemRandom()

    # 理论逆CDF要求采样点位于开区间(0,1)
    while True:
        r = rng.random()

        if 0 < r < 1:
            break

    u = r - 0.5

    sign = 1.0 if u >= 0 else -1.0

    return (
        -scale
        * sign
        * math.log1p(-2.0 * abs(u))
    )

def private_count(records, epsilon, rng=None):
    """
    对二元属性记录进行带噪声计数。

    假设：
        一个人至多贡献一条记录。
        相邻数据库差一个人的加入或删除。
        每个值只能为0或1。

    敏感度：
        Δf = 1
    """

    if epsilon <= 0:
        raise ValueError("epsilon必须大于0")

    if any(x not in (0, 1) for x in records):
        raise ValueError("记录必须为0或1")

    true_count = sum(records)

    scale = 1.0 / epsilon

    return true_count + laplace_noise(
        scale, rng
    )


synthetic_records = (
    [1]*63
    + [0]*37
)

noisy_answer = private_count(
    synthetic_records,
    epsilon=0.5,
)

print("发布结果:", noisy_answer)

def private_mean_fixed_n(
    values,
    epsilon,
    public_n,
    lower=0.0,
    upper=100.0,
    rng=None,
):
    """
    公开固定人数条件下的私有平均值。

    邻接关系：
        固定大小n的数据集中替换一条记录。

    假设：
        n事先公开，不由未经保护的私有输出泄露。
        数值先按公开范围进行裁剪。
    """

    if epsilon <= 0:
        raise ValueError("epsilon必须大于0")

    if public_n <= 0:
        raise ValueError("public_n必须大于0")

    if len(values) != public_n:
        raise ValueError("教学数据长度与公开n不一致")

    if upper <= lower:
        raise ValueError("上下界错误")

    bounded = [
        min(upper, max(lower, x))
        for x in values
    ]

    true_mean = sum(bounded) / public_n

    sensitivity = (
        upper - lower
    ) / public_n

    scale = sensitivity / epsilon

    return true_mean + laplace_noise(
        scale, rng
    )


ages = [20, 35, 40, 50, 55]

answer = private_mean_fixed_n(
    ages,
    epsilon=1.0,
    public_n=5,
)

print("带噪声平均年龄:", answer)

def private_histogram(
    labels,
    categories,
    epsilon,
    rng=None,
):
    """
    一人只属于一个类别。

    增删一人邻接关系下：
        整个向量的L1敏感度为1。

    独立地给每个类别计数加入
        Laplace(0, 1/epsilon)

    整个向量按一次epsilon预算发布。
    """

    if epsilon <= 0:
        raise ValueError("epsilon必须大于0")

    if len(set(categories)) != len(categories):
        raise ValueError("类别不能重复")

    counts = {
        category: 0
        for category in categories
    }

    for label in labels:
        if label not in counts:
            raise ValueError("存在未知类别")

        counts[label] += 1

    if rng is None:
        rng = secrets.SystemRandom()

    scale = 1.0 / epsilon

    return {
        category: (
            counts[category]
            + laplace_noise(scale, rng)
        )
        for category in categories
    }


interests = (
    ["编程"]*40
    + ["音乐"]*25
    + ["绘画"]*20
    + ["运动"]*15
)

hist = private_histogram(
    interests,
    ["编程", "音乐", "绘画", "运动"],
    epsilon=1.0,
)

for name, value in hist.items():
    print(name, round(value, 2))

class PrivacyBudget:
    """
    仅演示纯DP的基本顺序组合。

    不包含完整的多用户身份管理、
    贡献界定、持久化账本和并发控制。
    """

    def __init__(self, epsilon_limit):
        if epsilon_limit <= 0:
            raise ValueError("预算必须大于0")

        self.limit = float(epsilon_limit)
        self.spent = 0.0

    @property
    def remaining(self):
        return max(0.0, self.limit - self.spent)

    def release_count(self, records, epsilon):
        if epsilon <= 0:
            raise ValueError("epsilon必须大于0")

        if self.spent + epsilon > self.limit + 1e-12:
            raise RuntimeError("隐私预算不足")

        result = private_count(
            records,
            epsilon=epsilon,
        )

        self.spent += epsilon

        return result


ledger = PrivacyBudget(
    epsilon_limit=1.5
)

for i in range(3):

    result = ledger.release_count(
        synthetic_records,
        epsilon=0.5,
    )

    print(
        f"第{i+1}次发布:",
        round(result, 3),
        "剩余预算:",
        round(ledger.remaining, 3),
    )

try:
    ledger.release_count(
        synthetic_records,
        epsilon=0.5,
    )
except RuntimeError as error:
    print("第四次查询被拒绝:", error)

def laplace_pdf(y, center, scale):
    return (
        math.exp(-abs(y-center)/scale)
        / (2*scale)
    )


epsilon = 0.5
scale = 1 / epsilon

bound = math.exp(epsilon)

for i in range(201):

    y = 45.0 + i*0.2

    p63 = laplace_pdf(
        y, 63.0, scale
    )

    p64 = laplace_pdf(
        y, 64.0, scale
    )

    ratio = p63 / p64

    assert ratio <= bound + 1e-12
    assert ratio >= 1/bound - 1e-12

print("数值密度比检查通过")
print("理论上界:", bound)

数值密度比检查通过
理论上界: 1.6487212707...

count + laplace_noise(...)