前两节把「一笔转化算谁的」和「能看见多少」都算清了。本节收口到一个更现实的问题:三个系统报三份数字,你信哪一份。卷三《上线变现与长线运营》
写完了收入模型,但没处理「平台后台和 MMP 对不上」这种运营日常;站内游戏行业栏目
里有复盘与协作,但视角是团队而非数据口径。本节只做一件事:把三个口径摆到一张表上做对账,量化缺口并分类,让「谁在骗你」有据可查。照例声明:这里的 归因 指移动广告归因,与站内讲 AI 特征归因(SHAP)的内容无关。
为什么必然有三个口径
同一批安装,会有三个互相独立的系统各记一笔。它们本该大致相等,实际却常常差出十几个点。原因不是谁坏了,而是每个系统看到的边界不同:
| 口径 | 谁在记 | 记的是什么 | 系统性偏差方向 |
|---|---|---|---|
| MMP | 归因服务商 | 按归因模型认领的安装 | 倾向多认(自然量被误归) |
| 平台后台 | 投放平台 | 平台自认的转化 | 倾向多报(口径更宽) |
| 自有后端 | 你自己 | 真实注册 / 激活 | 倾向少记(埋点漏报) |
对账的价值不在「找出谁对」,而在把偏差的方向和量级固定下来:一旦知道 MMP 系统性比平台高 12%,你在看 MMP 报表时就会自动打个折,而不是每次都被同一个偏差吓一跳。
缺口怎么算、怎么分类
对每个渠道,算三个缺口千分比,再用一个容差阈值把渠道分成「一致」或「某方向不一致」。分类规则写死,避免每次人工判断:
| 条件 | 分类 | 通常意味着 |
|---|---|---|
| |gapMP| ≤ tol 且 |gapBM| ≤ tol | MATCH | 三口径一致,可信 |
| gapMP > tol | MMP_OVER | MMP 多认,疑自然量误归 |
| gapMP < -tol | MMP_UNDER | 平台口径更宽,或 MMP 未覆盖 |
| gapBM < -tol | BACKEND_UNDER | 自有后台低于 MMP,疑埋点漏报 |
容差 tol 本身就是个假设参数——这正是本节要用 B 基准扫它的原因。
S 层仿真:把缺口算出来
下面这段是 S 层仿真脚本 sim/o04_recon.py。它读 config/o04_recon.json,对每个渠道算三个缺口千分比并分类,汇总总量与不一致渠道数,产出 JSON + CSV + SVG。
#!/usr/bin/env python3
# file: sim/o04_recon.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/o04_recon.py --config config/ --out out/
"""三方归因与渠道数据对账:把归因服务商(MMP)、投放平台后台、自有后端三个口径
放到一张表上,算每个渠道的缺口千分比并分类。读 config/o04_recon.json。对每个渠道:
gapMP‰ = (mmpInstalls - platformInstalls) * 1000 // platformInstalls # MMP vs 平台
gapBM‰ = (backendInstalls - mmpInstalls) * 1000 // mmpInstalls # 后端 vs MMP
revGap‰ = (mmpRevenueFen - platformRevenueFen) * 1000 // platformRevenueFen
分类(容差 tol = tolerancePermille):gapMP > tol → MMP_OVER(疑自然量误归);gapMP <
-tol → MMP_UNDER;gapBM < -tol → BACKEND_UNDER(疑埋点漏报);gapBM > tol → BACKEND_OVER;
否则 MATCH。全程整数地板除,无浮点。产出 out/o04_recon.json + .csv + .svg。
三方真实数据需真实业务环境,命令行不可复现(X)。
"""
from __future__ import annotations
import argparse, csv, json, os
import xml.etree.ElementTree as ET
def classify(gap_mp: int, gap_bm: int, tol: int) -> str:
if gap_mp > tol: return "MMP_OVER"
if gap_mp < -tol: return "MMP_UNDER"
if gap_bm < -tol: return "BACKEND_UNDER"
if gap_bm > tol: return "BACKEND_OVER"
return "MATCH"
def reconcile(cfg: dict) -> dict:
tol, rows = cfg["tolerancePermille"], []
for ch in cfg["channels"]:
mmp, plat, back = ch["mmpInstalls"], ch["platformInstalls"], ch["backendInstalls"]
gap_mp = (mmp - plat) * 1000 // plat
gap_bm = (back - mmp) * 1000 // mmp
rev_gap = (ch["mmpRevenueFen"] - ch["platformRevenueFen"]) * 1000 // ch["platformRevenueFen"]
rows.append(dict(ch, gapMpPermille=gap_mp, gapBmPermille=gap_bm, revGapPermille=rev_gap,
status=classify(gap_mp, gap_bm, tol)))
tot_mmp, tot_plat = sum(r["mmpInstalls"] for r in rows), sum(r["platformInstalls"] for r in rows)
tot_back = sum(r["backendInstalls"] for r in rows)
return dict(rows=rows, totMmp=tot_mmp, totPlatform=tot_plat, totBackend=tot_back,
totGapMpPermille=(tot_mmp - tot_plat) * 1000 // tot_plat,
mismatchCount=sum(1 for r in rows if r["status"] != "MATCH"))
def draw_svg(path: str, rows: list, tol: int) -> int:
"""纯 Python SVG:每渠道一根「MMP - 平台」缺口横条,正负分列,容差用虚线。"""
W, PAD, ROW = 760, 330, 52
H = 70 + 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 = "《合成小岛》各渠道 MMP - 平台安装缺口(千分比)"
span, hi = W - PAD - 60, max([abs(r["gapMpPermille"]) for r in rows] + [tol * 2, 1])
zero, half = PAD + span // 2, span // 2
ET.SubElement(root, "line", {"x1": str(zero), "y1": "44", "x2": str(zero), "y2": str(H - 16),
"stroke": "#9ca3af"})
for sign in (-1, 1):
x = zero + sign * half * tol // hi
ET.SubElement(root, "line", {"x1": str(x), "y1": "44", "x2": str(x), "y2": str(H - 16),
"stroke": "#ef4444", "stroke-dasharray": "4 4"})
for i, r in enumerate(rows):
y, g = 60 + i * ROW, r["gapMpPermille"]
ET.SubElement(root, "text", {"x": "16", "y": str(y + 20), "font-size": "12",
"fill": "#374151"}).text = r["name"]
w = half * abs(g) // hi
ET.SubElement(root, "rect", {"x": str(zero if g >= 0 else zero - w), "y": str(y + 6),
"width": str(w), "height": "16", "rx": "3",
"fill": "#2563eb" if r["status"] == "MATCH" else "#ef4444"})
ET.SubElement(root, "text", {"x": str(zero + half + 8), "y": str(y + 19), "font-size": "11",
"fill": "#6b7280"}).text = f'{g}‰ · {r["status"]}'
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/")
args = ap.parse_args()
with open(os.path.join(args.config, "o04_recon.json"), encoding="utf-8") as f:
cfg = json.load(f)
rep = reconcile(cfg)
rows = rep["rows"]
# 跨语言对拍摘要:core/test/o04_recon.test.ts 复算同一串数字,逐位比对。
digest = ",".join(f'{r["id"]}:{r["gapMpPermille"]}:{r["gapBmPermille"]}:{r["status"]}' for r in rows)
os.makedirs(args.out, exist_ok=True)
svg_bytes = draw_svg(os.path.join(args.out, "o04_recon.svg"), rows, cfg["tolerancePermille"])
report = dict(cfg, rows=rows, totMmp=rep["totMmp"], totPlatform=rep["totPlatform"],
totBackend=rep["totBackend"], totGapMpPermille=rep["totGapMpPermille"],
mismatchCount=rep["mismatchCount"], svg=os.path.join(args.out, "o04_recon.svg"),
svgBytes=svg_bytes, digest=digest)
with open(os.path.join(args.out, "o04_recon.json"), "w", encoding="utf-8") as f:
json.dump(report, f, ensure_ascii=False, indent=2)
with open(os.path.join(args.out, "o04_recon.csv"), "w", encoding="utf-8", newline="") as f:
w = csv.writer(f)
cols = ("id", "name", "mmpInstalls", "platformInstalls", "backendInstalls",
"gapMpPermille", "gapBmPermille", "revGapPermille", "status")
w.writerow(cols)
for r in rows: w.writerow([r[k] for k in cols])
print(json.dumps({"totMmp": rep["totMmp"], "totPlatform": rep["totPlatform"],
"totBackend": rep["totBackend"], "totGapMpPermille": rep["totGapMpPermille"],
"mismatchCount": rep["mismatchCount"],
"channels": [{"id": r["id"], "gapMpPermille": r["gapMpPermille"],
"gapBmPermille": r["gapBmPermille"], "revGapPermille": r["revGapPermille"],
"status": r["status"]} for r in rows], "digest": digest},
ensure_ascii=False, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())
三方数字放在 config/o04_recon.json(D 层配置),都是「公开可观察机制」口径的建模假设,不是任何 MMP / 平台 / 后端的实测。
{
"version": 1,
"game": "merge-island",
"observedAt": "2026-10",
"tolerancePermille": 100,
"channels": [
{ "id": "feed", "name": "信息流", "mmpInstalls": 22500, "platformInstalls": 20000, "backendInstalls": 21800, "mmpRevenueFen": 517500, "platformRevenueFen": 440000 },
{ "id": "rewarded", "name": "激励视频", "mmpInstalls": 9800, "platformInstalls": 10200, "backendInstalls": 9550, "mmpRevenueFen": 215600, "platformRevenueFen": 224400 },
{ "id": "kol", "name": "达人合作", "mmpInstalls": 3600, "platformInstalls": 3300, "backendInstalls": 3520, "mmpRevenueFen": 165600, "platformRevenueFen": 151800 },
{ "id": "union", "name": "联盟网盟", "mmpInstalls": 7400, "platformInstalls": 6900, "backendInstalls": 6100, "mmpRevenueFen": 118400, "platformRevenueFen": 110400 }
],
"note": "三方对账建模假设,不是任何 MMP / 平台 / 后端的实测。每渠道给三个安装口径(MMP 回传 / 平台后台 / 自有后端)与两个收入口径(单位分);tolerancePermille 为判定一致的容差。真实三方数据需真实业务环境,命令行不可复现(X)。全整数,无浮点。"
}
在本机跑 /opt/homebrew/bin/python3 sim/o04_recon.py --config config/ --out out/,结果如下(同机相对量级,绝对数值不可移植):
| 渠道 | gapMP‰ | gapBM‰ | revGap‰ | 分类 |
|---|---|---|---|---|
| 信息流 | 125 | -32 | 176 | MMP_OVER |
| 激励视频 | -40 | -26 | -40 | MATCH |
| 达人合作 | 90 | -23 | 90 | MATCH |
| 联盟网盟 | 72 | -176 | 72 | BACKEND_UNDER |
总量上 MMP 报 43300、平台 40400、后端 40970,MMP 比平台高 71‰;四个渠道里 2 个判为不一致。信息流是 MMP_OVER:MMP 比平台多认 12.5%,收入口径更是多报 17.6%——典型的「自然量被算进买量」信号,值得去查这批安装的点击归因是否混入了太多展示窗口内的自然量。联盟网盟是 BACKEND_UNDER:自有后台比 MMP 少 17.6%,MMP 与平台却基本一致,说明问题出在你自己的埋点——很可能是这个渠道的回传链路漏报,而不是渠道虚报。
B 基准:容差扫一遍,看结论稳不稳
上面那张表依赖一个假设参数——容差 100‰。换个容差,结论会不会翻?下面这段是 B 层基准脚本 sim/o04_recon_bench.py,它复用 S 层的 reconcile,把容差取 {50,100,150,200}‰ 各跑一遍,统计不一致渠道数与每个渠道的状态变动。软告警不硬失败。
#!/usr/bin/env python3
# file: sim/o04_recon_bench.py
# tier: B
# verify: /opt/homebrew/bin/python3 sim/o04_recon_bench.py --config config/ --out out/
"""B 基准:对账「不一致」判定对容差阈值的稳健性。判定依赖 tolerancePermille 这个
假设参数,所以要问:「容差从 ±5% 放宽到 ±20%,会多几个渠道被翻案?」做法:对同一份
三方数据,把容差取 {50,100,150,200}‰ 各重跑一次分类,统计不一致渠道数与状态变动。
软告警:相对基准容差(100‰),状态发生变动的渠道 >= 1 即打 WARN。退出码恒为 0——
这是基准,不是门禁。全整数,无浮点。
"""
from __future__ import annotations
import argparse, json, os
from o04_recon import reconcile
TOLERANCES = (50, 100, 150, 200)
BASELINE = 100
SWING_WARN = 1
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--config", default="config/")
ap.add_argument("--out", default="out/")
args = ap.parse_args()
with open(os.path.join(args.config, "o04_recon.json"), encoding="utf-8") as f:
cfg = json.load(f)
cases = []
for tol in TOLERANCES:
rep = reconcile(dict(cfg, tolerancePermille=tol))
cases.append({"tolerancePermille": tol, "mismatchCount": rep["mismatchCount"],
"status": {r["id"]: r["status"] for r in rep["rows"]}})
base = next(c for c in cases if c["tolerancePermille"] == BASELINE)
swing = {ch["id"]: sum(1 for c in cases if c["status"][ch["id"]] != base["status"][ch["id"]])
for ch in cfg["channels"]}
warnings = [f'渠道 {cid} 在容差扫描下状态变动 {n} 次' for cid, n in swing.items() if n >= SWING_WARN]
os.makedirs(args.out, exist_ok=True)
rep = {"version": cfg["version"], "source": "sim/o04_recon_bench.py",
"baselineTolerancePermille": BASELINE, "tolerances": list(TOLERANCES), "swingWarn": SWING_WARN,
"cases": cases, "swingByChannel": swing, "maxSwing": max(swing.values()) if swing else 0,
"warnings": warnings, "verdict": "PASS(soft)" if not warnings else f"WARN(soft) {len(warnings)} 条"}
with open(os.path.join(args.out, "o04_recon_bench.json"), "w", encoding="utf-8") as f:
json.dump(rep, f, ensure_ascii=False, indent=2)
print(json.dumps({"cases": cases, "swingByChannel": swing, "maxSwing": rep["maxSwing"],
"verdict": rep["verdict"], "warnings": warnings}, ensure_ascii=False, indent=2))
return 0 # B 层软告警:恒返回 0
if __name__ == "__main__":
raise SystemExit(main())
在本机跑 /opt/homebrew/bin/python3 sim/o04_recon_bench.py --config config/ --out out/,结果:
| 容差‰ | 不一致渠道数 | 判定 |
|---|---|---|
| 50 | 3 | 更严,达人合作也被判 MMP_OVER |
| 100 | 2 | 基准:信息流 + 联盟网盟 |
| 150 | 1 | 信息流翻案成 MATCH,只剩联盟网盟 |
| 200 | 0 | 全部一致 |
swingByChannel 显示:信息流与联盟网盟各变动 2 次、达人合作 1 次、激励视频 0 次。结论是——信息流的 MMP_OVER 对容差很敏感(容差放宽到 150‰ 就消失了),而联盟网盟的 BACKEND_UNDER 对容差稳健(四个容差下都判不一致)。所以真正该优先处理的是联盟网盟的埋点,信息流那个可以继续观察。这就是 B 基准的作用:把「哪些结论是硬的、哪些是软的」标出来。
P 层内核:跨语言对拍
下面这段是 P 层 TypeScript 内核 core/src/o04/recon.ts,复算同一套缺口与分类。它读同一份 config/o04_recon.json,产出和 Python 逐字符一致的摘要。注意负缺口必须向下取整(-175.67 → -176),用 Math.floor 而非 Math.trunc,否则跨语言会分叉。
// file: core/src/o04/recon.ts
// tier: P
// verify: node --test 'core/test/o04_*.test.ts'
//
// 三方对账内核:与 sim/o04_recon.py 逐位等价的 TypeScript 对偶实现。
// 读同一份 config/o04_recon.json;全整数、地板除、无浮点;只用可擦除语法(禁 enum / namespace)。
export interface ReconConfig {
readonly version: number; readonly game: string; readonly tolerancePermille: number;
readonly channels: readonly {
readonly id: string; readonly name: string;
readonly mmpInstalls: number; readonly platformInstalls: number; readonly backendInstalls: number;
readonly mmpRevenueFen: number; readonly platformRevenueFen: number;
}[];
}
export interface ReconRow {
readonly id: string; readonly gapMpPermille: number; readonly gapBmPermille: number;
readonly revGapPermille: number; readonly status: string;
}
/** 整数地板除(a 可为负,结果向下取整,与 Python // 一致)。 */
export function floorDiv(a: number, b: number): number {
return Math.floor(a / b);
}
export function classify(gapMp: number, gapBm: number, tol: number): string {
if (gapMp > tol) return "MMP_OVER";
if (gapMp < -tol) return "MMP_UNDER";
if (gapBm < -tol) return "BACKEND_UNDER";
if (gapBm > tol) return "BACKEND_OVER";
return "MATCH";
}
export function reconcile(cfg: ReconConfig): ReconRow[] {
const tol = cfg.tolerancePermille;
return cfg.channels.map((ch) => {
const gapMp = floorDiv((ch.mmpInstalls - ch.platformInstalls) * 1000, ch.platformInstalls);
const gapBm = floorDiv((ch.backendInstalls - ch.mmpInstalls) * 1000, ch.mmpInstalls);
const revGap = floorDiv((ch.mmpRevenueFen - ch.platformRevenueFen) * 1000, ch.platformRevenueFen);
return { id: ch.id, gapMpPermille: gapMp, gapBmPermille: gapBm, revGapPermille: revGap,
status: classify(gapMp, gapBm, tol) };
});
}
/** 跨语言对拍摘要:与 sim/o04_recon.py 的 digest 逐字符一致。 */
export function digest(rows: readonly ReconRow[]): string {
return rows.map((r) => `${r.id}:${r.gapMpPermille}:${r.gapBmPermille}:${r.status}`).join(",");
}
配套测试 core/test/o04_recon.test.ts 断言 TS 复算结果逐字符相等。在本机跑 node --test 'core/test/o04_*.test.ts',4 个用例全过,两套实现给出同一个摘要 feed:125:-32:MMP_OVER,rewarded:-40:-26:MATCH,kol:90:-23:MATCH,union:72:-176:BACKEND_UNDER。
X 层:真实三方数据不可复现
上面算的是模型侧:给定三份数字,缺口与分类是多少。而下面这些属于 X 层,需真实业务环境,命令行不可复现:
- 真实的 MMP 回传、平台后台数字与自有后端注册数(含各自的延迟、去重与丢失);
- 真实的收入口径差异(MMP 认的是归因收入,平台报的是自认转化,两者定义不同);
- 真实的容差标准(多少算「一致」是业务约定,不是物理常量)。
对账的正确姿势是把它做成周期动作:每周把三方数字拉齐、跑一遍脚本、记录哪些渠道在漂移。漂移本身不重要,漂移的方向和幅度突然变化才重要——那通常意味着某一边的埋点或口径改了。
自检表
| 检查项 | 合格标准 |
|---|---|
| 三口径齐全 | MMP / 平台 / 后端各一份,不只用一方数据 |
| 缺口有符号 | 算带方向的千分比,不只算绝对值 |
| 分类写死 | 判定规则进脚本,不靠人工拍脑袋 |
| 稳健性 | 容差扫描,标出哪些结论对阈值敏感 |
| 跨语言一致 | Python 与 TS 摘要逐字符相同 |
| 边界清晰 | 真实三方数据标 X,不写成「实测」 |
至此第 4 章的三块拼图——窗口、隐私链路、对账——合起来,就是你判断「买量到底赚不赚钱」的数据地基。地基可信了,下一章我们才敢动真金白银:广告聚合平台怎么接、瀑布流怎么配。
阅读导航:上一节:4.2 ATT、IDFA 与 SKAdNetwork 链路 · 下一节:5.1 聚合架构:SDK、适配器与瀑布流 。
继续阅读
探索更多技术文章
浏览归档,发现更多关于系统设计、工具链和工程实践的内容。