上一节把留存做成了两条曲线，但它们都是事后统计——告诉你昨天掉了多少人，却不能在玩家真的要掉之前把他捞回来。本节讲事前那一步：流失预警（谁快要走了）与召回（把他叫回来值不值）。本站那篇独立开发者的手游广告失败复盘 讲的正是「留存没做好就先买量」的教训；本节把它反过来——先识别出要流失的人，再用可算的 ROI 决定要不要花钱召回。
预警的价值在于「早」
等一个玩家彻底不来了，再想召回就晚了：他已经卸载、已经忘了这款游戏、已经装上了别的。预警的价值就在于早——在他还装着、还偶尔打开的时候，就把他标成「高风险」，用一条推送或一份礼包去挽留。所以预警模型要回答的不是「他会不会走」，而是「在还来得及的时候，我能不能认出他」。
认人要靠行为特征：最近多久没登录、会话时长是不是在缩短、卡在同一关多久了、广告观看是不是在减少、社交互动是不是冷了。这些信号单独看都不强，但加权合起来就有判别力。下面先用一份配置把这些特征和阈值固定下来。
D 层：把特征、分档与召回渠道落成配置
下面这份是 D 层（静态层）产物，全整数、没有任何浮点。features 是五个预警特征及其权重；riskBands 是按分数切的三档；recallChannels 是三条召回渠道的成本与回流率；reactivatedValueDeciFen 是一个被召回用户的期望价值。全部为示例假设参数，真实特征分布与召回效果需真实业务环境，命令行不可复现（X）。
{
"version": 1,
"game": "merge-island",
"seed": 20261012,
"observedAt": "2026-10",
"note": "流失预警与召回：用可播种 LCG 生成一批带真实标签的合成用户，按加权特征打分、分风险档，算混淆矩阵（精确率/召回率/提升度），再对高风险人群跑召回渠道的成本-回收账。权重、阈值、触发概率、召回成本与回流率均为示例假设参数；真实特征分布与召回效果需真实业务环境，命令行不可复现（X）。全整数，禁浮点；随机数用配置 seed + 自实现 LCG。",
"cohortSize": 20000,
"churnBasePermille": 350,
"triggerIfChurnPermille": 700,
"triggerIfStayPermille": 250,
"features": [
{ "id": "idleDays", "name": "最近登录间隔偏长", "weight": 30 },
{ "id": "sessionDrop", "name": "会话时长下降", "weight": 25 },
{ "id": "levelStall", "name": "关卡停滞", "weight": 20 },
{ "id": "adDrop", "name": "广告观看下降", "weight": 15 },
{ "id": "socialDrop", "name": "社交互动下降", "weight": 10 }
],
"riskBands": [
{ "id": "high", "name": "高风险", "minScore": 60 },
{ "id": "mid", "name": "中风险", "minScore": 35 },
{ "id": "low", "name": "低风险", "minScore": 0 }
],
"recallChannels": [
{ "id": "push", "name": "站内推送", "costCent": 1, "reachPermille": 900, "reactivatePermille": 60 },
{ "id": "sms", "name": "短信召回", "costCent": 6, "reachPermille": 700, "reactivatePermille": 110 },
{ "id": "gift", "name": "回流礼包", "costCent": 25, "reachPermille": 950, "reactivatePermille": 210 }
],
"reactivatedValueDeciFen": 3200
}
注意配置里那两个触发概率：triggerIfChurnPermille 700 与 triggerIfStayPermille 250。它们的含义是——真正会流失的人，某个特征被触发的概率是 70%；不会流失的人，同一特征误触发的概率是 25%。预警的难度全在这两个数的差距里：差距越大，模型越好分；差距越小，误报越多。
S 层：合成用户、打分、分档、算账
下面这段是 S 层仿真脚本，读 config/l07_churn.json：用配置 seed 跑一个 32 位 LCG 合成 2 万个带真实标签的用户，按权重打分分档，以「高风险」为预测正类算混淆矩阵与精确率 / 召回率 / 提升度，再对高风险人群跑三条召回渠道的成本-回收账。产出 JSON + CSV + SVG。
#!/usr/bin/env python3
# file: sim/l07_churn.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/l07_churn.py --config config/ --out out/
"""流失预警与召回：读 config/l07_churn.json。
用配置 seed + 自实现 32 位 LCG 生成带真实标签的合成用户；每个特征在「已流失 / 未流失」
两群人身上有不同触发概率（triggerIfChurn / triggerIfStay）。风险分 = 触发特征权重之和，
按 minScore 分档；以「高风险」为预测正类算混淆矩阵、精确率、召回率与提升度；再对高风险
人群算召回渠道账：reached = 高风险人数 × reachPermille // 1000，reactivated = reached ×
reactivatePermille // 1000，回收 = reactivated × reactivatedValueDeciFen，成本 = reached ×
costCent × 10（0.1 分），ROI = 回收 // 成本。产出 JSON + CSV + SVG。
全整数地板除，无浮点，无 Math.random。真实特征分布与召回效果需真实业务环境，命令行不可复现（X）。
"""
from __future__ import annotations
import argparse
import csv
import json
import os
import xml.etree.ElementTree as ET
MASK32 = 0xFFFFFFFF
def load(d: str) -> dict:
with open(os.path.join(d, "l07_churn.json"), encoding="utf-8") as f:
return json.load(f)
def lcg(state: int) -> int:
"""32 位线性同余发生器：state = (state*1664525 + 1013904223) mod 2^32。"""
return (state * 1664525 + 1013904223) & MASK32
def make_cohort(cfg: dict) -> list[dict]:
"""先抽真实标签，再按标签决定每个特征是否触发。"""
state, users = cfg["seed"], []
for _ in range(cfg["cohortSize"]):
state = lcg(state)
churned = (state % 1000) < cfg["churnBasePermille"]
trig = []
for _f in cfg["features"]:
state = lcg(state)
p = cfg["triggerIfChurnPermille"] if churned else cfg["triggerIfStayPermille"]
trig.append((state % 1000) < p)
users.append({"churned": churned, "trig": trig})
return users
def score_of(cfg: dict, trig: list) -> int:
return sum(f["weight"] for f, t in zip(cfg["features"], trig) if t)
def band_of(cfg: dict, score: int) -> str:
for b in cfg["riskBands"]:
if score >= b["minScore"]:
return b["id"]
return cfg["riskBands"][-1]["id"]
def draw_svg(path: str, bands: list[dict]) -> int:
"""纯 Python SVG：每个风险档一根横条（人数），标注档内真实流失率。"""
W, PAD, ROW = 720, 176, 52
H = 74 + ROW * len(bands)
r = ET.Element("svg", {"xmlns": "http://www.w3.org/2000/svg", "width": str(W),
"height": str(H), "viewBox": f"0 0 {W} {H}"})
ET.SubElement(r, "rect", {"width": str(W), "height": str(H), "fill": "#ffffff"})
ET.SubElement(r, "text", {"x": "16", "y": "26", "font-size": "15", "fill": "#111827"}).text = \
"风险分档：档内真实流失率随分数升高（示例参数）"
span, hi = W - PAD - 210, max((b["count"] for b in bands), default=1) or 1
colors = {"high": "#dc2626", "mid": "#f59e0b", "low": "#10b981"}
for i, b in enumerate(bands):
y, w = 50 + i * ROW, span * b["count"] // hi
ET.SubElement(r, "text", {"x": "16", "y": str(y + 22), "font-size": "12",
"fill": "#374151"}).text = b["name"]
ET.SubElement(r, "rect", {"x": str(PAD), "y": str(y + 6), "width": str(w), "height": "22",
"rx": "4", "fill": colors.get(b["id"], "#6b7280")})
ET.SubElement(r, "text", {"x": str(PAD + w + 8), "y": str(y + 22), "font-size": "12",
"fill": "#6b7280"}).text = f'{b["count"]} 人 · 流失 {b["churnRatePermille"] / 10:.1f}%'
os.makedirs(os.path.dirname(path), exist_ok=True)
ET.ElementTree(r).write(path, encoding="utf-8", xml_declaration=True)
return os.path.getsize(path)
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--config", default="config/")
ap.add_argument("--out", default="out/")
a = ap.parse_args()
cfg = load(a.config)
users = make_cohort(cfg)
bands = {b["id"]: {"id": b["id"], "name": b["name"], "minScore": b["minScore"],
"count": 0, "churned": 0} for b in cfg["riskBands"]}
tp = fp = fn = tn = 0
for u in users:
sc = score_of(cfg, u["trig"])
b = bands[band_of(cfg, sc)]
b["count"] += 1
b["churned"] += 1 if u["churned"] else 0
pred = sc >= cfg["riskBands"][0]["minScore"] # 高风险为预测正类
if pred and u["churned"]:
tp += 1
elif pred:
fp += 1
elif u["churned"]:
fn += 1
else:
tn += 1
band_rows = []
for b in cfg["riskBands"]:
x = bands[b["id"]]
x["churnRatePermille"] = x["churned"] * 1000 // x["count"] if x["count"] else 0
band_rows.append(x)
base_rate = (tp + fn) * 1000 // len(users)
precision = tp * 1000 // (tp + fp) if (tp + fp) else 0
recall = tp * 1000 // (tp + fn) if (tp + fn) else 0
high_count = tp + fp
channels = []
for ch in cfg["recallChannels"]:
reached = high_count * ch["reachPermille"] // 1000
reactivated = reached * ch["reactivatePermille"] // 1000
value, cost = reactivated * cfg["reactivatedValueDeciFen"], reached * ch["costCent"] * 10
channels.append({"id": ch["id"], "name": ch["name"], "costCent": ch["costCent"],
"reached": reached, "reactivated": reactivated, "valueDeciFen": value,
"costDeciFen": cost, "netDeciFen": value - cost,
"roiPermille": value * 1000 // cost if cost else 0})
best = max(channels, key=lambda c: c["netDeciFen"])
os.makedirs(a.out, exist_ok=True)
svg = draw_svg(os.path.join(a.out, "l07_churn.svg"), band_rows)
with open(os.path.join(a.out, "l07_churn.csv"), "w", encoding="utf-8", newline="") as f:
w = csv.writer(f)
w.writerow(["riskBand", "name", "minScore", "count", "churned", "churnRatePermille"])
for b in band_rows:
w.writerow([b["id"], b["name"], b["minScore"], b["count"], b["churned"], b["churnRatePermille"]])
w.writerow([])
w.writerow(["channel", "name", "costCent", "reached", "reactivated", "valueDeciFen",
"costDeciFen", "netDeciFen", "roiPermille"])
for c in channels:
w.writerow([c["id"], c["name"], c["costCent"], c["reached"], c["reactivated"],
c["valueDeciFen"], c["costDeciFen"], c["netDeciFen"], c["roiPermille"]])
digest = f"tp:{tp}:fp:{fp}:fn:{fn}:tn:{tn}|prec:{precision}:rec:{recall}|best:{best['id']}"
rep = {"version": cfg["version"], "game": cfg["game"], "observedAt": cfg["observedAt"],
"cohortSize": cfg["cohortSize"], "churnBasePermille": cfg["churnBasePermille"],
"confusion": {"tp": tp, "fp": fp, "fn": fn, "tn": tn}, "baseRatePermille": base_rate,
"precisionPermille": precision, "recallPermille": recall,
"liftPermille": precision * 1000 // base_rate if base_rate else 0,
"highRiskCount": high_count, "bands": band_rows, "channels": channels,
"bestChannel": best["id"], "digest": digest, "csv": os.path.join(a.out, "l07_churn.csv"),
"svg": os.path.join(a.out, "l07_churn.svg"), "svgBytes": svg}
with open(os.path.join(a.out, "l07_churn.json"), "w", encoding="utf-8") as f:
json.dump(rep, f, ensure_ascii=False, indent=2)
print(json.dumps({"confusion": rep["confusion"], "baseRatePermille": base_rate,
"precisionPermille": precision, "recallPermille": recall,
"liftPermille": rep["liftPermille"], "highRiskCount": high_count,
"bandChurnRate": {b["id"]: b["churnRatePermille"] for b in band_rows},
"bestChannel": best["id"], "bestNetDeciFen": best["netDeciFen"],
"digest": digest}, ensure_ascii=False))
return 0
if __name__ == "__main__":
raise SystemExit(main())
在本机跑它，输出如下（同机相对量级，绝对数值不可移植）：
{
"confusion": { "tp": 5153, "fp": 929, "fn": 1984, "tn": 11934 },
"baseRatePermille": 356,
"precisionPermille": 847,
"recallPermille": 722,
"liftPermille": 2379,
"highRiskCount": 6082,
"bandChurnRate": { "high": 847, "mid": 337, "low": 42 },
"bestChannel": "gift",
"bestNetDeciFen": 2437350
}
读混淆矩阵：预警模型到底准不准
2 万人里真实流失 7137 人（基率 35.6%），模型把 6082 人标成高风险。混淆矩阵是 TP 5153 / FP 929 / FN 1984 / TN 11934，于是：
| 指标 | 值 | 含义 |
|---|---|---|
| 精确率 | 84.7% | 标成高风险的人里，84.7% 真的流失了 |
| 召回率 | 72.2% | 真正流失的人里，72.2% 被标成了高风险 |
| 提升度 | 2.38× | 相对 35.6% 的基率，命中率抬高了 2.38 倍 |
精确率 84.7% 意味着误报率只有 15.3%——发出去 100 条召回推送，只有约 15 条发给了本来不会走的人，浪费可控。召回率 72.2% 意味着近三成的真实流失者漏网，他们没被标进高风险，也就不会收到召回。这是预警模型永恒的权衡：想多抓人就得降低阈值（召回率升、精确率降），想少误报就得抬高阈值（反之）。分档表把这条权衡摊开：
| 风险档 | 分数区间 | 人数 | 档内真实流失率 |
|---|---|---|---|
| 高风险 | ≥ 60 | 6082 | 84.7% |
| 中风险 | 35–59 | 4721 | 33.7% |
| 低风险 | 0–34 | 9197 | 4.2% |
三档的流失率 84.7% / 33.7% / 4.2% 呈单调下降，这正是模型有判别力的证据：分数越高，真的越会走。低风险档 9197 人里只有 4.2% 会流失，对他们发召回几乎全是浪费——所以召回资源应该只投给高风险档。
P 层：把打分与分档锁进内核
预警要能上线，就不能只是一段离线 Python。本节对应的 P 层产物是 core/src/l07/churn.ts：把 LCG、加权打分、分档、混淆矩阵与召回账都封成可复用函数，与 Python 脚本同一份 config、同一串 LCG，因此逐位一致——LCG 用 Math.imul 实现 32 位取模，第一次取值 1891695331 与 Python 的 & 0xFFFFFFFF 完全相同。全整数、无 Math.random()，只用可擦除语法。
配套测试 core/test/l07_churn.test.ts 直接读同一份 config，把混淆矩阵（5153 / 929 / 1984 / 11934）、高风险人数（6082）与三条召回账锁进断言，并验证「同输入两次结果逐位相同」。node --test 'core/test/l07_*.test.ts' 本机 4 项全过：
✔ LCG 是 32 位取模、确定可复现
✔ 风险分 = 触发特征权重之和；分档按 minScore 降序
✔ 与 Python 仿真同口径：混淆矩阵与召回账逐位一致
✔ 同输入两次结果逐位相同（无隐藏随机）
ℹ tests 4
ℹ pass 4
ℹ fail 0
逐位一致不是巧合，而是「同一份 config + 同一串 LCG」的必然结果：LCG 的乘法会超过 2^53，TS 侧必须用 Math.imul 做 32 位取模，Python 侧用 & 0xFFFFFFFF，两者在数学上是同一个运算。
召回账：ROI 最高 ≠ 净收益最高
把三条召回渠道的账算完（高风险 6082 人为基数）：
| 渠道 | 触达 | 召回 | 回收（元） | 成本（元） | 净收益（元） | ROI |
|---|---|---|---|---|---|---|
| 站内推送 | 5473 | 328 | 1049.6 | 54.7 | 994.9 | 19.2× |
| 短信召回 | 4257 | 468 | 1497.6 | 255.4 | 1242.2 | 5.86× |
| 回流礼包 | 5777 | 1213 | 3881.6 | 1444.3 | 2437.4 | 2.69× |
（回收 / 成本按 reactivatedValueDeciFen 3200 即 3.2 元 / 召回用户折算。）
这张表里藏着一个反直觉的结论：站内推送的 ROI 最高（19.2×），但净收益最低（994.9 元）；回流礼包 ROI 最低（2.69×），净收益却最高（2437.4 元）。原因是 ROI 是个比率——推送几乎不花钱（1 分 / 人），分母小，比率自然大；礼包单价高（25 分 / 人），但回流率高（21% vs 6%），乘出来的绝对量最大。所以「该选哪条渠道」取决于目标：预算紧、只想证明召回有效，选推送；要的是绝对回收额、预算不设限，选礼包。把这两个数放在一张表里，正是为了不让「ROI 最高」这一个数替你做决定。
小结与下一步
这一节把「流失」从事后统计变成了事前预警：合成用户 → 加权打分 → 风险分档 → 混淆矩阵（精确率 84.7% / 召回率 72.2% / 提升 2.38 倍）→ 召回账。核心结论有两个：预警的难度在「会走的人触发特征」与「不会走的人误触发」这两个概率的差距里；选召回渠道要同时看 ROI 和净收益，二者常常指向不同的渠道。
但这一节只算了「召回多少人」，没算「召回回来的人值多少钱」。下一节把留存与付费放在一起看：不同留存档 × 不同付费档的用户，ARPU 差多少；抬高留存和抬高付费率，哪个更能撬动收入。
阅读导航：上一节：新手漏斗与 D1/D7/D30 留存 · 下一节：留存与付费的联合分析 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。