10.2 无 IDFA 时代的归因替代

IDFA 没了,用户级归因断了,但渠道贡献仍然可估。本节给出两条不依赖设备标识的路:媒体组合模型(MMM)用精确有理数解正规方程反推各渠道边际贡献,地区增量实验(DiD)用双重差分量出广告真正多带来的安装。S 层仿真落 JSON + CSV + SVG,B 层再把噪声与观测周数扫成 20 格矩阵,告诉你 MMM 在什么条件下才还能用。

上一节 把链路接对了:授权走 IDFA、未授权走聚合回传。但聚合回传给的是粗粒度、有延迟、只有部分字段的数据——它不会告诉你「这个安装是哪条广告带来的」。本节回答的就是这个问题:没有 IDFA,还能靠什么把渠道贡献估出来——渠道怎么选、漏斗怎么搭,站内增长渠道与漏斗拆解 已经讲过,本节只补「估」这一步。与站内卷四 4.2 的分工是:卷四讲归因窗口与三方对账(数据已经有了怎么对齐),本节讲数据拿不到时怎么反推。

归因断了,不等于不能估

用户级归因(把某个安装精确归到某次点击)依赖设备标识,标识没了,这条路就断了。但「估渠道贡献」还有两条不依赖设备级标识的路:一是看总量之间的关系,二是主动做实验。两条路的共同点是都只需要聚合数据,因此与隐私新规不冲突。

第二条路还顺带解决了归因的根本缺陷——归因不等于增量。一个用户可能本来就会装你的游戏,归因却把这份功劳记在某次广告上。只有把一部分地区故意不给广告当对照,才能量出「广告真正多带来了多少安装」。

两条可算的路:MMM 与地区增量实验

方法需要的数据回答的问题主要风险
媒体组合模型 MMM周度花费 × 周度安装总量各渠道的边际贡献是多少渠道花费高度相关时反推不稳
地区增量实验 DiD实验 / 对照地区的安装序列广告真正多带来多少增量地区间不可比、外部冲击

MMM 用最小二乘把「安装 = 基线 + Σ 渠道花费 × 系数」里的系数解出来;DiD 用「实验组变化 − 对照组变化」消掉共同趋势。下面用一份可机读配置把两条路都跑一遍。

S 层:建模配置

S 层配置 config/g10_alt.json 把种子、周数、花费区间、渠道真值与地区分组全部写成整数,保证同机可复现:

{
  "version": 1,
  "game": "merge-island",
  "observedAt": "2026-10",
  "note": "无 IDFA 时代的归因替代建模假设(观察时点 2026-10),不是任一渠道后台实测。seed + 整数 LCG 生成确定性的周度花费与地区安装流;channels[].trueEffectPermille=每 1 单位花费带来的安装数(千分比);noisePermille=周安装噪声幅度;geo 段做双重差分(DiD)增量实验。真实渠道贡献与真实增量需真实业务环境,命令行不可复现(X)。全整数生成,回归用精确有理数(Fraction),无浮点。",
  "seed": 20261013,
  "weeks": 12,
  "baseInstallsPerWeek": 900,
  "noisePermille": 150,
  "spendMin": 200,
  "spendMax": 1400,
  "channels": [
    { "id": "search", "label": "搜索", "trueEffectPermille": 900 },
    { "id": "social", "label": "社交", "trueEffectPermille": 620 },
    { "id": "video", "label": "视频", "trueEffectPermille": 780 },
    { "id": "display", "label": "展示", "trueEffectPermille": 300 }
  ],
  "geo": {
    "preWeeks": 4,
    "postWeeks": 4,
    "liftPermille": 200,
    "geos": [
      { "id": "g1", "group": "treatment", "baseline": 520 },
      { "id": "g2", "group": "treatment", "baseline": 410 },
      { "id": "g3", "group": "treatment", "baseline": 640 },
      { "id": "g4", "group": "control", "baseline": 500 },
      { "id": "g5", "group": "control", "baseline": 460 },
      { "id": "g6", "group": "control", "baseline": 580 }
    ]
  }
}

S 层:MMM 与 DiD 仿真

下面这段是 S 层仿真脚本 sim/g10_alt_attribution.py。它用 seed + 整数 LCG 生成确定性的花费与噪声,MMM 的正规方程用精确有理数 Fraction 解——这样结果与 BLAS / numpy 版本无关,逐位一致;DiD 全程整数运算。产出 JSON + CSV + SVG:

#!/usr/bin/env python3
# file: sim/g10_alt_attribution.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/g10_alt_attribution.py --config config/ --out out/
"""无 IDFA 时代的归因替代仿真(S 层):媒体组合模型(MMM)+ 地区增量实验(DiD)。

IDFA 消失后拿不到用户级触点,只剩两条可算的路:
  1) MMM:只看「周度花费 × 周度安装」两条总量序列,最小二乘反推每个渠道的边际贡献。
     用精确有理数(Fraction)解正规方程,结果与 BLAS / numpy 版本无关,逐位一致。
         installs[w] = base + Σ_c (spend[w][c] * effect_c // 1000) + noise[w]
         recovered_c = round(coef_c * 1000)   (千分比,每 1 单位花费的安装数)
  2) Geo 增量实验(DiD):地区分实验 / 对照两组,用「实验组变化 − 对照组变化」估增量安装。
         est_incremental = (Δ_treat * n_ctrl − Δ_ctrl * n_treat) // n_ctrl
确定性:seed + 整数 LCG 生成花费与噪声;回归用 Fraction,无浮点。产出
out/g10_alt_attribution.json + .csv + .svg。
口径:渠道真实贡献与真实增量是「公开可观察机制」的建模假设(观察时点 2026-10),不是任一
渠道后台实测;真实增量需真实业务环境,命令行不可复现(X)。
"""
from __future__ import annotations

import argparse
import csv
import json
import os
import xml.etree.ElementTree as ET
from fractions import Fraction

MASK = 0x7FFFFFFF


def lcg(s):
    return (s * 1103515245 + 12345) & MASK


def solve(a, b):
    """高斯消元解线性方程组,全程 Fraction,精确且与 BLAS 无关。"""
    n = len(a)
    m = [[Fraction(a[i][j]) for j in range(n)] + [Fraction(b[i])] for i in range(n)]
    for col in range(n):
        piv = next(r for r in range(col, n) if m[r][col] != 0)
        m[col], m[piv] = m[piv], m[col]
        m[col] = [v / m[col][col] for v in m[col]]
        for r in range(n):
            if r != col and m[r][col] != 0:
                f = m[r][col]
                m[r] = [m[r][k] - f * m[col][k] for k in range(n + 1)]
    return [m[i][n] for i in range(n)]


def make_spend(cfg):
    """确定性周度花费矩阵与安装序列。"""
    s, weeks, nc = cfg["seed"], cfg["weeks"], len(cfg["channels"])
    lo, hi = cfg["spendMin"], cfg["spendMax"]
    spend = []
    for _w in range(weeks):
        row = []
        for _c in range(nc):
            s = lcg(s)
            row.append(lo + s % (hi - lo + 1))
        spend.append(row)
    base, amp = cfg["baseInstallsPerWeek"], cfg["baseInstallsPerWeek"] * cfg["noisePermille"] // 1000
    eff = [c["trueEffectPermille"] for c in cfg["channels"]]
    installs = []
    for w in range(weeks):
        v = base + sum(spend[w][c] * eff[c] // 1000 for c in range(nc))
        s = lcg(s)
        installs.append(v + (s % (2 * amp + 1)) - amp)
    return spend, installs


def mmm_fit(cfg, spend, installs):
    """正规方程最小二乘(精确有理数),返回各渠道边际贡献(千分比)。"""
    weeks, nc = cfg["weeks"], len(cfg["channels"])
    p = nc + 1
    x = [[1] + [spend[w][c] for c in range(nc)] for w in range(weeks)]
    xtx = [[sum(x[w][i] * x[w][j] for w in range(weeks)) for j in range(p)] for i in range(p)]
    xty = [sum(x[w][i] * installs[w] for w in range(weeks)) for i in range(p)]
    return [round(coef * 1000) for coef in solve(xtx, xty)[1:]]


def geo_lift(cfg):
    """地区双重差分:整数生成 + 整数 DiD 估计。"""
    g = cfg["geo"]
    pre, post, lift = g["preWeeks"], g["postWeeks"], g["liftPermille"]
    treat = [x for x in g["geos"] if x["group"] == "treatment"]
    ctrl = [x for x in g["geos"] if x["group"] == "control"]
    s = lcg(cfg["seed"] ^ 0x5A5A5A5A)
    pre_t = post_t = pre_c = post_c = base_t = 0
    rows = []
    for x in g["geos"]:
        base, amp = x["baseline"], x["baseline"] * cfg["noisePermille"] // 1000
        tot_pre = tot_post = 0
        for _ in range(pre):
            s = lcg(s)
            tot_pre += base + (s % (2 * amp + 1)) - amp
        for _ in range(post):
            s = lcg(s)
            v = base + (s % (2 * amp + 1)) - amp
            tot_post += v * (1000 + lift) // 1000 if x["group"] == "treatment" else v
        rows.append({"id": x["id"], "group": x["group"], "preTotal": tot_pre, "postTotal": tot_post})
        if x["group"] == "treatment":
            pre_t += tot_pre; post_t += tot_post; base_t += base * post
        else:
            pre_c += tot_pre; post_c += tot_post
    d_treat, d_ctrl = post_t - pre_t, post_c - pre_c
    est = (d_treat * len(ctrl) - d_ctrl * len(treat)) // len(ctrl)
    true = post_t - base_t
    return {"estimatedIncremental": est, "trueIncremental": true,
            "errorPermille": (est - true) * 1000 // true if true else 0,
            "deltaTreat": d_treat, "deltaCtrl": d_ctrl, "geos": rows}


def draw_svg(path, rows):
    """纯 Python SVG:每渠道两根横条——真实边际贡献 vs MMM 反推。"""
    W, PAD, ROW, span = 760, 170, 60, 530
    H = 92 + ROW * len(rows)
    root = ET.Element("svg", {"xmlns": "http://www.w3.org/2000/svg", "width": str(W),
                              "height": str(H), "viewBox": f"0 0 {W} {H}"})
    ET.SubElement(root, "rect", {"width": str(W), "height": str(H), "fill": "#ffffff"})
    ET.SubElement(root, "text", {"x": "16", "y": "28", "font-size": "15", "fill": "#111827"}
                  ).text = "《合成小岛》MMM 反推渠道边际贡献(千分比)vs 真值"
    hi = max(max(r["trueEffectPermille"], r["recoveredPermille"]) for r in rows) or 1
    for i, r in enumerate(rows):
        y = 54 + i * ROW
        ET.SubElement(root, "text", {"x": "16", "y": str(y + 22), "font-size": "12",
                                     "fill": "#374151"}).text = r["label"]
        for j, key in enumerate(("trueEffectPermille", "recoveredPermille")):
            w = span * r[key] // hi
            ET.SubElement(root, "rect", {"x": str(PAD), "y": str(y + 6 + j * 20), "width": str(w),
                                         "height": "16", "rx": "3",
                                         "fill": ("#93c5fd", "#2563eb")[j]})
            ET.SubElement(root, "text", {"x": str(PAD + w + 6), "y": str(y + 18 + j * 20),
                                         "font-size": "11", "fill": "#6b7280"}).text = f'{r[key]}‰'
    os.makedirs(os.path.dirname(path), exist_ok=True)
    ET.ElementTree(root).write(path, encoding="utf-8", xml_declaration=True)
    return len(ET.tostring(root))


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_alt.json"), encoding="utf-8") as f:
        cfg = json.load(f)
    spend, installs = make_spend(cfg)
    recovered = mmm_fit(cfg, spend, installs)
    rows = [{"id": c["id"], "label": c["label"], "trueEffectPermille": c["trueEffectPermille"],
             "recoveredPermille": rec, "absErrorPermille": abs(rec - c["trueEffectPermille"]),
             "relErrorPermille": abs(rec - c["trueEffectPermille"]) * 1000 // c["trueEffectPermille"]}
            for c, rec in zip(cfg["channels"], recovered)]
    mape = sum(r["relErrorPermille"] for r in rows) // len(rows)
    geo = geo_lift(cfg)
    digest = ("mmm=" + ",".join(f'{r["id"]}:{r["trueEffectPermille"]}:{r["recoveredPermille"]}'
                                for r in rows)
              + f';mape={mape};geo={geo["estimatedIncremental"]}:{geo["trueIncremental"]}'
              + f':{geo["errorPermille"]}')
    os.makedirs(a.out, exist_ok=True)
    svg = draw_svg(os.path.join(a.out, "g10_alt_attribution.svg"), rows)
    cols = ("id", "label", "trueEffectPermille", "recoveredPermille", "absErrorPermille", "relErrorPermille")
    with open(os.path.join(a.out, "g10_alt_attribution.csv"), "w", encoding="utf-8", newline="") as f:
        w = csv.writer(f)
        w.writerow(cols)
        for r in rows:
            w.writerow([r[c] for c in cols])
    rep = {"version": cfg["version"], "game": cfg["game"], "observedAt": cfg["observedAt"],
           "weeks": cfg["weeks"], "channels": rows, "mmmMapePermille": mape, "geo": geo,
           "digest": digest, "svg": os.path.join(a.out, "g10_alt_attribution.svg"), "svgBytes": svg}
    with open(os.path.join(a.out, "g10_alt_attribution.json"), "w", encoding="utf-8") as f:
        json.dump(rep, f, ensure_ascii=False, indent=2)
    print(json.dumps({"digest": digest, "mmmMapePermille": mape, "channels": rows,
                      "geo": {"estimatedIncremental": geo["estimatedIncremental"],
                              "trueIncremental": geo["trueIncremental"],
                              "errorPermille": geo["errorPermille"]}},
                     ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

在本机跑 /opt/homebrew/bin/python3 sim/g10_alt_attribution.py --config config/ --out out/,四个渠道的边际贡献反推如下(单位 ‰,即每 1 单位花费带来的安装数):

渠道真值MMM 反推绝对误差相对误差
搜索 search9009343437‰
社交 social62053783133‰
视频 video78077467‰
展示 display300410110366‰

渠道平均相对误差 mape=135‰(约 13.5%)。花费少的渠道误差最大——展示渠道花费区间窄、信号弱,反推最不稳,这正是 MMM 的已知弱点。地区 DiD 给出 estimated=969 对 true=1261,低估约 23%:噪声把实验组的增量稀释了。这些数字都来自本机实跑,不是估计。

B 层:稳健性基准(软告警)

上面那组数字只在一个噪声档、一个观测周数上成立。要判断结论稳不稳,得把这两个旋钮一起扫。下面这段是 B 层基准脚本 sim/g10_alt_bench.py 的核心网格函数(完整脚本在工作区,--config config/ --out out/ 可复现):

# snippet: sim/g10_alt_bench.py(核心网格;NOISE_GRID × WEEK_GRID 共 20 格)
NOISE_GRID = [0, 75, 150, 300, 600]
WEEK_GRID = [8, 12, 16, 24]
WARN_MAPE = 200

def cell(cfg, noise, weeks):
    c = copy.deepcopy(cfg)
    c["noisePermille"], c["weeks"] = noise, weeks
    spend, installs = make_spend(c)
    rec = mmm_fit(c, spend, installs)
    errs = [abs(r - ch["trueEffectPermille"]) * 1000 // ch["trueEffectPermille"]
            for r, ch in zip(rec, c["channels"])]
    geo = geo_lift(c)
    return {"noisePermille": noise, "weeks": weeks, "mmmMapePermille": sum(errs) // len(errs),
            "geoErrorPermille": geo["errorPermille"]}

跑出来 20 格的 MMM 平均相对误差(‰)矩阵如下:

噪声 \ 周数8121624
0‰(无噪)0000
75‰134833547
150‰(基准)2501358039
300‰85323695127
600‰1549593942239

读法很直接:噪声越低、观测越长,反推越准;基准档(150‰ / 12 周)是 135‰,低于软告警阈值 200‰,所以 warns=0。但 600‰ / 8 周那格高达 1549‰——误差比真值还大,说明噪声档太差时 MMM 会彻底失效。这张矩阵比一句「MMM 很稳」有用得多:它告诉你在什么条件下 MMM 还能用。

X 层:真实渠道贡献不可复现

必须划清边界。上面算的是模型侧:给定真值与噪声,两种估计方法能恢复到什么程度。而下面这些属于 X 层,需真实业务环境:

  • 真实渠道的边际贡献(随素材、受众、竞争与季节变化,且各渠道会互相蚕食);
  • 真实的花费与安装序列(受预算节奏、平台竞价、自然量波动影响);
  • 地区增量实验的真实增量(要求实验地区之间基线可比,现实中很难满足)。

工程上正确的做法是先做小规模地区实验标定 MMM:用 DiD 量出的真实增量去校准 MMM 系数,再用校准后的 MMM 指导日常预算分配。单靠 MMM 做预算决策,在渠道花费高度相关时会给出看似精确、实则不稳的结论。

自检表

检查项合格标准
归因 ≠ 增量明确区分「归到某渠道」与「真正多带来」,用实验验证
无设备级标识只依赖聚合序列,不依赖 IDFA / OAID
回归可复现用精确有理数或固定精度,结果与 BLAS 版本无关
参数敏感性至少给出噪声 × 观测长度的对照矩阵,不报单点结论
边界清晰真实渠道贡献标 X,不写成「实测」

下一节换到上游:即便你什么都不做,平台侧的聚合上报本身也在消耗用户的隐私预算——怎么把噪声与预算算清楚,是下一节的事。

阅读导航:上一节:10.1 ATT、IDFA 与 SKAdNetwork · 下一节:10.3 隐私沙盒与聚合上报 。

继续阅读

探索更多技术文章

浏览归档,发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「minigame」更多文章

  1. 附录 F:全书公式、参数与数据集索引
  2. 9.3 同意管理与数据出境
  3. 9.2 COPPA 与儿童向游戏的限制