10.1 讲了未授权侧走聚合回传，10.2 讲了没有 IDFA 时怎么反推渠道贡献。两节都默认「聚合数据是干净的」（合规底线见站内数据安全与隐私合规 ）。本节把这个假设拆开：聚合上报本身是被噪声污染过的——平台为了保护个体，会在计数上加噪声、抑制小样本、并按隐私预算限制你能问多少次。工程上必须把这套机制算清楚，否则你会把噪声当成趋势。
聚合上报的三个机制
| 机制 | 作用 | 工程含义 |
|---|---|---|
| 加噪 | 给每个计数加随机噪声 | 小桶的相对误差极大，不能直接看单点 |
| 最小计数抑制 | 真实计数低于阈值不发 | 小地区 / 小众行为会整片缺失，不是 0 |
| 聚合预算 | 累计查询消耗超过总额即拒绝 | 查询要提前规划，不能想到就问 |
三者叠加的结果是：聚合数据能用来看大盘，不能用来追单点。下面用一个整数、可播种的差分隐私机制把这三件事都跑出来。
D 层：机制与预算清单
D 层配置 config/g10_dp_agg.json 把噪声强度、预算总额、最小计数阈值与查询计划写成整数表：
{
"version": 1,
"game": "merge-island",
"observedAt": "2026-10",
"note": "隐私沙盒聚合上报的建模假设（观察时点 2026-10），不是任一平台聚合服务实测。mechanism.pPermille=整数离散拉普拉斯机制的伯努利成功阈值（越小噪声越大）；budget.totalEpsilonMilli=聚合预算总额；queries[].epsilonMilli=每个查询消耗的预算；minCount=最小计数阈值（低于即抑制）。真实平台的噪声机制与预算口径需真实业务环境，命令行不可复现（X）。全整数、可播种（seed + LCG），无浮点。",
"seed": 20261013,
"mechanism": { "pPermille": 400 },
"budget": { "totalEpsilonMilli": 4000 },
"minCount": 5,
"buckets": [
{ "id": "us", "trueCount": 4200 },
{ "id": "de", "trueCount": 900 },
{ "id": "jp", "trueCount": 1500 },
{ "id": "kr", "trueCount": 700 },
{ "id": "br", "trueCount": 300 },
{ "id": "mx", "trueCount": 6 },
{ "id": "nz", "trueCount": 3 }
],
"queries": [
{ "id": "q_region_installs", "epsilonMilli": 800, "buckets": ["us", "de", "jp", "kr", "br", "mx", "nz"] },
{ "id": "q_region_payers", "epsilonMilli": 800, "buckets": ["us", "de", "jp", "kr", "br"] },
{ "id": "q_region_retained", "epsilonMilli": 800, "buckets": ["us", "de", "jp", "kr"] },
{ "id": "q_region_ads", "epsilonMilli": 800, "buckets": ["us", "de", "jp"] },
{ "id": "q_region_iap", "epsilonMilli": 800, "buckets": ["us", "de"] },
{ "id": "q_region_sessions", "epsilonMilli": 800, "buckets": ["us"] },
{ "id": "q_region_extra", "epsilonMilli": 800, "buckets": ["us", "de", "jp"] }
]
}
下面这段是 D 层脚本 sim/g10_dp_agg.py。噪声用整数离散拉普拉斯（双几何）：先抽符号，再数「成功前的失败次数」得到幅度——全程整数、可播种、无浮点，因此跨语言可逐位对拍：
#!/usr/bin/env python3
# file: sim/g10_dp_agg.py
# tier: D
# verify: /opt/homebrew/bin/python3 sim/g10_dp_agg.py --config config/ --out out/
"""隐私沙盒聚合上报：整数差分隐私噪声 + 聚合预算记账（D 层：机制与预算的可跑清单）。
读 config/g10_dp_agg.json，模拟一次「按地区聚合上报」的发布流程：
1) 噪声机制：整数离散拉普拉斯（双几何）。先按 LCG 抽符号 s=±1，再数「成功前失败
次数」m（每次 LCG 抽 state%1000 < pPermille 即成功），噪声 = s*m。pPermille 越小
噪声越大、隐私越强。全程整数、可播种、可复现。
2) 最小计数阈值：真实计数 < minCount 的桶直接抑制（不发），防重识别。
3) 聚合预算：每个查询消耗 epsilonMilli，累计超过 totalEpsilonMilli 的查询被拒。
产出 out/g10_dp_agg.json + .csv。全整数、无浮点。
口径：机制参数与预算口径是「公开可观察机制」的建模假设（观察时点 2026-10），不是任一
平台聚合服务实测；真实噪声机制与预算需真实业务环境，命令行不可复现（X）。
"""
from __future__ import annotations
import argparse
import csv
import json
import os
MASK = 0x7FFFFFFF
MAX_MAG = 64
def lcg(s: int) -> int:
return (s * 1103515245 + 12345) & MASK
def sample_noise(state: int, p_permille: int) -> tuple:
"""整数离散拉普拉斯：符号 + 几何幅度。返回 (噪声, 新状态)。"""
state = lcg(state)
sign = 1 if (state & 1) == 0 else -1
mag = 0
while mag < MAX_MAG:
state = lcg(state)
if (state % 1000) < p_permille:
break
mag += 1
return sign * mag, state
def publish(cfg: dict) -> dict:
counts = {b["id"]: b["trueCount"] for b in cfg["buckets"]}
total, p, min_count = cfg["budget"]["totalEpsilonMilli"], cfg["mechanism"]["pPermille"], cfg["minCount"]
state, spent, draws = cfg["seed"], 0, 0
rel = den = sup = 0
queries = []
for q in cfg["queries"]:
if spent + q["epsilonMilli"] > total:
den += 1
queries.append({"id": q["id"], "status": "denied", "epsilonMilli": q["epsilonMilli"],
"released": {}})
continue
spent += q["epsilonMilli"]
rel += 1
released = {}
for bid in q["buckets"]:
if counts[bid] < min_count:
released[bid] = "sup"
sup += 1
continue
noise, state = sample_noise(state, p)
draws += 1
released[bid] = max(0, counts[bid] + noise)
queries.append({"id": q["id"], "status": "ok", "epsilonMilli": q["epsilonMilli"],
"released": released})
return {"spent": spent, "total": total, "released": rel, "denied": den, "suppressed": sup,
"draws": draws, "queries": queries}
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--config", default="config/")
ap.add_argument("--out", default="out/")
a = ap.parse_args()
with open(os.path.join(a.config, "g10_dp_agg.json"), encoding="utf-8") as f:
cfg = json.load(f)
rep = publish(cfg)
body = ";".join(f'{q["id"]}:{q["status"]}:' + ",".join(f"{k}={v}" for k, v in q["released"].items())
for q in rep["queries"])
digest = (f'budget={rep["total"]},spent={rep["spent"]},queries={len(cfg["queries"])},'
f'released={rep["released"]},denied={rep["denied"]},suppressed={rep["suppressed"]},'
f'draws={rep["draws"]};' + body)
os.makedirs(a.out, exist_ok=True)
counts = {b["id"]: b["trueCount"] for b in cfg["buckets"]}
with open(os.path.join(a.out, "g10_dp_agg.csv"), "w", encoding="utf-8", newline="") as f:
w = csv.writer(f)
w.writerow(["query", "status", "bucket", "trueCount", "released"])
for q in rep["queries"]:
if q["status"] == "denied":
w.writerow([q["id"], "denied", "", "", ""])
else:
for bid, val in q["released"].items():
w.writerow([q["id"], "ok", bid, counts[bid], val])
out = {"version": cfg["version"], "game": cfg["game"], "observedAt": cfg["observedAt"],
"mechanism": cfg["mechanism"], "budget": cfg["budget"], "minCount": cfg["minCount"],
"spent": rep["spent"], "released": rep["released"], "denied": rep["denied"],
"suppressed": rep["suppressed"], "draws": rep["draws"], "queries": rep["queries"],
"digest": digest, "csv": os.path.join(a.out, "g10_dp_agg.csv")}
with open(os.path.join(a.out, "g10_dp_agg.json"), "w", encoding="utf-8") as f:
json.dump(out, f, ensure_ascii=False, indent=2)
print(json.dumps({"digest": digest, "spent": rep["spent"], "released": rep["released"],
"denied": rep["denied"], "suppressed": rep["suppressed"], "draws": rep["draws"],
"queries": rep["queries"]}, ensure_ascii=False, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())
在本机跑 /opt/homebrew/bin/python3 sim/g10_dp_agg.py --config config/ --out out/，七个查询里 5 个发布、2 个被预算拒绝，另有 1 个桶被抑制：
| 查询 | 状态 | 发布内容 |
|---|---|---|
| q_region_installs | ok | us=4201 de=900 jp=1500 kr=699 br=307 mx=6 nz=sup |
| q_region_payers | ok | us=4199 de=902 jp=1500 kr=701 br=300 |
| q_region_retained | ok | us=4197 de=901 jp=1495 kr=702 |
| q_region_ads | ok | us=4205 de=900 jp=1500 |
| q_region_iap | ok | us=4200 de=899 |
| q_region_sessions | denied | 预算已耗尽 |
| q_region_extra | denied | 预算已耗尽 |
三件事都体现在这张表里。加噪：us 的真实计数 4200，前五个查询分别发布 4201 / 4199 / 4197 / 4205 / 4200——同一个真实值被问五次，出来五个不同的数。抑制：nz 真实计数 3，低于阈值 5，直接发 sup 而不是 0 或 3。预算：前 5 个查询各消耗 800‰，累计正好 4000‰，第 6、7 个被拒。注意 mx 真实计数恰好等于阈值 5，被放行——边界是「小于」而不是「小于等于」。
P 层：跨语言对拍，证明噪声可复现
噪声是随机的，但必须是可复现的随机——否则报表今天一个数、明天一个数，无法对账。下面这段是 P 层 TypeScript 内核 core/src/g10/dp_agg.ts，用 BigInt LCG 复刻同一套噪声，产出与 Python 逐字符一致的摘要。用 BigInt 而非 number 是必需的：LCG 的中间乘积会超过 2^53，number 会丢精度、两侧对不上：
// file: core/src/g10/dp_agg.ts
// tier: P
// verify: node --test 'core/test/g10_*.test.ts'
//
// 隐私沙盒聚合上报内核：与 sim/g10_dp_agg.py 逐位等价的 TypeScript 对偶实现。
// 读同一份 config/g10_dp_agg.json；整数离散拉普拉斯噪声用 BigInt LCG（可播种、可复现）；
// 全整数、无浮点、无 Math.random；只用可擦除语法（禁 enum / namespace）。
export interface Bucket { readonly id: string; readonly trueCount: number }
export interface Query { readonly id: string; readonly epsilonMilli: number; readonly buckets: readonly string[] }
export interface DpConfig {
readonly version: number; readonly game: string; readonly seed: number;
readonly mechanism: { readonly pPermille: number };
readonly budget: { readonly totalEpsilonMilli: number };
readonly minCount: number; readonly buckets: readonly Bucket[]; readonly queries: readonly Query[];
}
export interface QueryResult {
readonly id: string; readonly status: string; readonly epsilonMilli: number;
readonly released: Record<string, number | string>;
}
export interface DpResult {
readonly spent: number; readonly total: number; readonly released: number; readonly denied: number;
readonly suppressed: number; readonly draws: number; readonly queries: QueryResult[];
}
const MASK = 0x7fffffffn;
const MAX_MAG = 64;
/** 整数线性同余，BigInt 版：与 Python 大整数逐位等价。 */
export function lcg(state: bigint): bigint {
return (state * 1103515245n + 12345n) & MASK;
}
/** 整数离散拉普拉斯：符号 + 几何幅度。返回 [噪声, 新状态]。 */
export function sampleNoise(state: bigint, pPermille: number): [number, bigint] {
state = lcg(state);
const sign = (state & 1n) === 0n ? 1 : -1;
let mag = 0;
while (mag < MAX_MAG) {
state = lcg(state);
if (Number(state % 1000n) < pPermille) break;
mag += 1;
}
return [sign * mag, state];
}
export function publish(cfg: DpConfig): DpResult {
const counts = new Map(cfg.buckets.map((b) => [b.id, b.trueCount]));
const total = cfg.budget.totalEpsilonMilli, p = cfg.mechanism.pPermille, minCount = cfg.minCount;
let state = BigInt(cfg.seed), spent = 0, draws = 0, released = 0, denied = 0, suppressed = 0;
const queries: QueryResult[] = [];
for (const q of cfg.queries) {
if (spent + q.epsilonMilli > total) {
denied += 1;
queries.push({ id: q.id, status: "denied", epsilonMilli: q.epsilonMilli, released: {} });
continue;
}
spent += q.epsilonMilli;
released += 1;
const rel: Record<string, number | string> = {};
for (const bid of q.buckets) {
const trueCount = counts.get(bid) ?? 0;
if (trueCount < minCount) { rel[bid] = "sup"; suppressed += 1; continue; }
const [noise, next] = sampleNoise(state, p);
state = next; draws += 1;
rel[bid] = Math.max(0, trueCount + noise);
}
queries.push({ id: q.id, status: "ok", epsilonMilli: q.epsilonMilli, released: rel });
}
return { spent, total, released, denied, suppressed, draws, queries };
}
/** 跨语言对拍摘要：与 sim/g10_dp_agg.py 的 digest 逐字符一致。 */
export function digest(cfg: DpConfig): string {
const r = publish(cfg);
const head = `budget=${r.total},spent=${r.spent},queries=${cfg.queries.length},`
+ `released=${r.released},denied=${r.denied},suppressed=${r.suppressed},draws=${r.draws};`;
return head + r.queries.map((q) =>
`${q.id}:${q.status}:` + Object.entries(q.released).map(([k, v]) => `${k}=${v}`).join(","),
).join(";");
}
配套测试 core/test/g10_dp_agg.test.ts 把 Python 落盘的摘要写成金标准，并验「同一种子两次采样完全一致」「最小计数抑制」「预算拒绝」「发布值非负」。在本机跑 node --test 'core/test/g10_*.test.ts'，本节 5 个用例全过：
✔ 跨语言一致：TS 聚合摘要 == Python 金标准
✔ 噪声可播种可复现：同一种子两次采样完全一致
✔ 最小计数抑制：真实计数低于阈值不发
✔ 聚合预算：累计超过总额的查询被拒
✔ 发布值非负：加噪后不会出现负数计数
ℹ tests 5 / pass 5 / fail 0（本节 5 个；该 glob 连同 10.1 的 6 个共 11 个）
两套实现给出同一个摘要（截取首段）：budget=4000,spent=4000,queries=7,released=5,denied=2,suppressed=1,draws=20;q_region_installs:ok:us=4201,de=900,jp=1500,kr=699,br=307,mx=6,nz=sup;…。
X 层：真实机制不可复现
必须划清边界。上面算的是机制侧：给定参数与查询计划，噪声、抑制与预算的行为是什么。而下面这些属于 X 层，需真实业务环境：
- 真实的噪声机制与参数（平台可能用不同的分布、不同的敏感度标定）；
- 真实的聚合预算口径（总额、单查询消耗、是否按用户分摊，均由平台定义）；
- 真实的抑制阈值与上报延迟（会随版本与地区调整）。
工程上正确的做法是把机制层与业务层分开：机制层回答「这个桶会不会被抑制、这次查询会不会被拒」，业务层用来在拿到数据后判断「这个差异是趋势还是噪声」。把噪声当趋势是这类数据最常见的误用——同一个 4200 被问出 4197 到 4205 的抖动，你若拿两个查询的差值当「增长」，就会得出完全错误的结论。
自检表
| 检查项 | 合格标准 |
|---|---|
| 噪声可复现 | 种子固定、同机重跑逐位一致，不用 Math.random |
| 小样本识别 | 抑制值单独标记，不与 0 混为一谈 |
| 预算记账 | 累计消耗与拒绝都在结果里显式可见 |
| 跨语言一致 | Python 与 TS 摘要逐字符相同 |
| 边界清晰 | 真实机制与预算口径标 X，不写成「实测」 |
把机制算清楚之后，还有一个更硬的问题：钱怎么从平台真正落到你账上——结算方式、提现门槛与到账周期。下一章处理这件事。
阅读导航：上一节：10.2 无 IDFA 时代的归因替代 · 下一节：11.1 平台结算方式与提现门槛 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。