12.1 的漏斗是事后的——人已经走了，我们回头数缺口。这一节做事前的：在玩家真正流失之前，用他过去 7 天的行为算出「未来 7 天流失」的概率，把概率高的人挑出来，触发一次召回。模型不用任何机器学习库，手写一个纯 Python 逻辑回归（全批量梯度下降），这样每一行都可控、可解释、可复算；同时跑一个规则基线（「停滞天数 ≥ 8 就告警」）做同口径对照，看模型到底值不值这些复杂度。
先定义「流失」
「流失」必须先定义清楚，否则模型学的是你的定义、而不是玩家的行为。本节的定义是未来 7 天内不再登录（churnDefinitionDays: 7）。这个定义有两个后果要提前认下：①它会把「偶尔隔三天上线一次的老玩家」也算成流失，带来标签噪声；②定义窗口越长，正样本越稀疏、预测越难。真实项目里通常同时看 7 天与 30 天两套标签，本节只用 7 天。
特征与合成样本
五个特征都是运营侧能直接取到的：activeDays7（近 7 天活跃天数）、stallDays（当前停滞天数）、paidCents（累计付费，分）、inAlliance（是否在联盟）、levelGain7（近 7 天等级增益）。配置 config/n12_churn.json（D 层）给每个特征标了区间、分布与「真实权重」——权重是生成样本用的真值，不是模型要学的答案；labelBiasBp 与 labelNoiseBp 控制基础流失率与标签噪声。
样本是合成的：先按真值权重算一个潜在分数 z，再叠加均匀噪声，z + noise > 0 记为流失。为什么不用伯努利采样？ 因为伯努利会把信号抹掉——同样一份特征，抽签得到的标签 AUC 只有 0.6 出头，噪声模型才能把「特征确实有区分度」这件事保留下来。合成数据的价值不在数值本身，而在它有一个已知的真值，可以用来验证实现对不对。
手写逻辑回归
下面这段是 S 层脚本，包含四件事：合成样本生成、全批量梯度下降训练、测试集指标计算（精确率 / 召回率 / F1）、以及 AUC（用秩和 Mann-Whitney 口径，不调库）。训练在归一化特征上做，加了 L2 正则。
# file: sim/n12_churn.py
# tier: S
# verify: python3 sim/n12_churn.py --config config/ --out out/
"""流失预警：合成带标签样本 + 手写逻辑回归（全批量梯度下降，纯 Python，不引入 sklearn）。
读 config/n12_churn.json：特征定义与「真实」生成权重用于合成样本；手写逻辑回归在归一化
特征上训练，输出测试集精确率 / 召回率 / AUC，以及阈值扫描与召回活动 ROI。
产出 out/n12_churn.json + out/n12_churn.csv + out/n12_churn.svg。
"""
from __future__ import annotations
import argparse
import csv
import json
import math
import os
from svg_line import render
MASK = 0xFFFFFFFF
def imul(a: int, b: int) -> int:
return ((a & MASK) * (b & MASK)) & MASK
def mulberry32(seed: int):
a = seed & MASK
def nxt() -> float:
nonlocal a
a = (a + 0x6D2B79F5) & MASK
t = imul(a ^ (a >> 15), a | 1)
t = (t ^ ((t + imul(t ^ (t >> 7), t | 61)) & MASK)) & MASK
return ((t ^ (t >> 14)) & MASK) / 4294967296.0
return nxt
def sigmoid(z: float) -> float:
return 0.0 if z < -60 else 1.0 if z > 60 else 1.0 / (1.0 + math.exp(-z))
def norm(f: dict, v: int) -> float:
return (v - f["min"]) / (f["max"] - f["min"])
def gen_dataset(cfg: dict, rng) -> list[dict]:
feats, rows = cfg["features"], []
for _ in range(cfg["players"]):
raw = {}
for f in feats:
if f["kind"] == "binary":
v = 1 if int(rng() * 100) < f["posPct"] else 0
elif f["kind"] == "paid":
v = 0 if int(rng() * 100) < f["zeroPct"] else int(rng() * (f["max"] - f["min"] + 1)) + f["min"]
else:
v = int(rng() * (f["max"] - f["min"] + 1)) + f["min"]
raw[f["id"]] = v
z = cfg["labelBiasBp"] / 10000.0 + sum(f["trueWeightBp"] / 10000.0 * norm(f, raw[f["id"]]) for f in feats)
y = 1 if z + (rng() * 2.0 - 1.0) * cfg["labelNoiseBp"] / 10000.0 > 0.0 else 0
rows.append({"x": [norm(f, raw[f["id"]]) for f in feats], "y": y})
return rows
def train(rows: list[dict], epochs: int, lr: float, l2: float) -> tuple[list[float], float]:
d, w, b, n = len(rows[0]["x"]), [0.0] * len(rows[0]["x"]), 0.0, len(rows)
for _ in range(epochs):
gw, gb = [0.0] * d, 0.0
for r in rows:
e = sigmoid(sum(w[j] * r["x"][j] for j in range(d)) + b) - r["y"]
for j in range(d):
gw[j] += e * r["x"][j]
gb += e
for j in range(d):
w[j] -= lr * (gw[j] / n + l2 * w[j])
b -= lr * gb / n
return w, b
def predict(w: list[float], b: float, r: dict) -> float:
return sigmoid(sum(w[j] * r["x"][j] for j in range(len(w))) + b)
def pr_at(scores: list[float], labels: list[int], threshold: float) -> dict:
tp = sum(1 for s, y in zip(scores, labels) if s >= threshold and y == 1)
fp = sum(1 for s, y in zip(scores, labels) if s >= threshold and y == 0)
fn = sum(1 for s, y in zip(scores, labels) if s < threshold and y == 1)
prec = tp * 1000 // (tp + fp) if tp + fp else 0
rec = tp * 1000 // (tp + fn) if tp + fn else 0
f1 = 2 * prec * rec // (prec + rec) if prec + rec else 0
return {"thresholdPct": int(threshold * 100), "alerts": tp + fp, "tp": tp, "fp": fp, "fn": fn,
"precisionPermille": prec, "recallPermille": rec, "f1Permille": f1}
def auc_permille(scores: list[float], labels: list[int]) -> int:
pairs = sorted(zip(scores, labels))
pos, neg = sum(labels), len(labels) - sum(labels)
if pos == 0 or neg == 0:
return 500
rank_sum, i = 0, 0
while i < len(pairs):
j = i
while j < len(pairs) and pairs[j][0] == pairs[i][0]:
j += 1
ar = (i + j + 1) / 2.0
rank_sum += sum(ar for k in range(i, j) if pairs[k][1] == 1)
i = j
return int(round((rank_sum - pos * (pos + 1) / 2.0) / (pos * neg) * 1000))
def main() -> int:
here = os.path.dirname(os.path.abspath(__file__))
ap = argparse.ArgumentParser()
ap.add_argument("--config", default=os.path.join(here, "..", "config"))
ap.add_argument("--out", default=os.path.join(here, "..", "out"))
args = ap.parse_args()
os.makedirs(args.out, exist_ok=True)
cfg = json.load(open(os.path.join(args.config, "n12_churn.json"), encoding="utf-8"))
rows = gen_dataset(cfg, mulberry32(cfg["prngSeed"]))
split, tr = cfg["players"] * cfg["trainPct"] // 100, cfg["training"]
w, b = train(rows[:split], tr["epochs"], tr["learningRateMilli"] / 1000.0, tr["l2Milli"] / 1000.0)
test = rows[split:]
scores, labels = [predict(w, b, r) for r in test], [r["y"] for r in test]
pos = sum(labels)
model = pr_at(scores, labels, tr["thresholdPct"] / 100.0)
model["aucPermille"] = auc_permille(scores, labels)
ridx = [f["id"] for f in cfg["features"]].index("stallDays")
rf, rule = cfg["features"][ridx], cfg["ruleBaseline"]
hits = [1 if r["x"][ridx] * (rf["max"] - rf["min"]) + rf["min"] >= rule["stallDaysGte"] else 0 for r in test]
rt = pr_at([float(h) for h in hits], labels, 0.5)
k = rt["alerts"]
top = set(sorted(range(len(scores)), key=lambda i: -scores[i])[:k])
tp = sum(1 for i in top if labels[i] == 1)
matched = {"alerts": k, "tp": tp, "precisionPermille": tp * 1000 // k if k else 0,
"recallPermille": tp * 1000 // pos if pos else 0}
sweep = [pr_at(scores, labels, t / 100.0) for t in range(10, 91, 10)]
render([s["recallPermille"] for s in sweep], os.path.join(args.out, "n12_churn.svg"),
"recall permille vs alert threshold pct (10..90)")
camp, alerts = cfg["recallCampaign"], model["alerts"]
saved = alerts * model["precisionPermille"] // 1000 * camp["successPct"] // 100
report = {"version": 1, "players": cfg["players"], "trainN": split, "testN": len(test),
"baseChurnPermille": pos * 1000 // len(labels),
"model": {"weights": {f["id"]: round(w[j], 4) for j, f in enumerate(cfg["features"])},
"bias": round(b, 4), **model},
"ruleBaseline": {**rule, **rt}, "matchedVolume": matched, "sweep": sweep,
"recallCampaign": {"alerts": alerts, "savedPlayers": saved, "costCents": alerts * camp["costCentsPerAlert"],
"recoveredValueCents": saved * camp["avgRecoveredValueCents"],
"roiPermille": saved * camp["avgRecoveredValueCents"] * 1000 // (alerts * camp["costCentsPerAlert"]) if alerts else 0}}
json.dump(report, open(os.path.join(args.out, "n12_churn.json"), "w", encoding="utf-8"), ensure_ascii=False, indent=2)
with open(os.path.join(args.out, "n12_churn.csv"), "w", encoding="utf-8", newline="") as fh:
wr = csv.writer(fh)
wr.writerow(["thresholdPct", "alerts", "tp", "fp", "fn", "precisionPermille", "recallPermille", "f1Permille"])
for s in sweep:
wr.writerow([s["thresholdPct"], s["alerts"], s["tp"], s["fp"], s["fn"], s["precisionPermille"], s["recallPermille"], s["f1Permille"]])
print(json.dumps({"baseChurnPermille": report["baseChurnPermille"], "model": model, "ruleBaseline": rt,
"matchedVolume": matched, "recallCampaign": report["recallCampaign"],
"weights": report["model"]["weights"], "bias": report["model"]["bias"]}, ensure_ascii=False))
return 0
if __name__ == "__main__":
raise SystemExit(main())
注意 sigmoid 的裁剪（±60）——纯 Python 的 math.exp 在 z 很大时会溢出，裁剪之后既保数值稳定又不改变预测。
实跑结果
python3 sim/n12_churn.py --config config/ --out out/ 的 stdout：
{"baseChurnPermille": 445, "model": {"thresholdPct": 50, "alerts": 1534, "tp": 1188, "fp": 346, "fn": 415, "precisionPermille": 774, "recallPermille": 741, "f1Permille": 757, "aucPermille": 881}, "ruleBaseline": {"thresholdPct": 50, "alerts": 977, "tp": 659, "fp": 318, "fn": 944, "precisionPermille": 674, "recallPermille": 411, "f1Permille": 510}, "matchedVolume": {"alerts": 977, "tp": 860, "precisionPermille": 880, "recallPermille": 536}, "recallCampaign": {"alerts": 1534, "savedPlayers": 178, "costCents": 30680, "recoveredValueCents": 267000, "roiPermille": 8702}, "weights": {"activeDays7": -2.2114, "stallDays": 1.8, "paidCents": -0.8674, "inAlliance": -0.9885, "levelGain7": -0.6345}, "bias": 0.8518}
测试集 3600 人（训练集 8400 人，70/30 划分），基础流失率 44.5%。阈值 0.5 下：
| 指标 | 模型 | 规则基线 |
|---|---|---|
| 告警数 | 1534 | 977 |
| 命中（TP） | 1188 | 659 |
| 误报（FP） | 346 | 318 |
| 漏报（FN） | 415 | 944 |
| 精确率 | 77.4% | 67.4% |
| 召回率 | 74.1% | 41.1% |
| F1 | 75.7% | 51.0% |
模型 AUC 88.1%，说明特征里确实有信号。同阈值下，模型把召回率从规则的 41.1% 抬到 74.1%，精确率也从 67.4% 抬到 77.4%——两头都赢，不是「牺牲精确换召回」。学到的权重（归一化特征上）：stallDays +1.80、activeDays7 −2.21、inAlliance −0.99、paidCents −0.87、levelGain7 −0.63，偏置 +0.85。符号全部符合直觉，这就是可解释性的好处：你能拿着这张权重表去问策划「为什么停滞天数的权重这么高」。
阈值扫描：告警量才是旋钮
阈值越高，告警越少、精确率越高、召回越低。out/n12_churn.csv 扫了 10% 到 90%：
| 阈值 | 告警数 | 精确率 | 召回率 | F1 |
|---|---|---|---|---|
| 10% | 3467 | 46.2% | 100.0% | 63.2% |
| 30% | 2517 | 61.1% | 95.9% | 74.6% |
| 50% | 1534 | 77.4% | 74.1% | 75.7% |
| 70% | 619 | 94.9% | 36.6% | 52.8% |
| 90% | 27 | 100.0% | 1.6% | 3.1% |
这张表才是预警系统真正的「旋钮」：你不是在选一个模型，而是在选一个告警量。运营侧能承受多少条告警、客服有多少人力，决定了阈值该定在哪。
B 基准：模型 vs 规则的公平对照
上面的对照有一处不公平：两者告警量不同（1534 vs 977），告警量越多、召回自然越高。所以 B 层基准补了第二种口径——取模型分数最高的 977 人（与规则同量），再比召回：
# file: sim/n12_churn_bench.py
# tier: B
# verify: python3 sim/n12_churn_bench.py --config config/ --out out/
"""流失预警基准（软告警，退出码恒 0）：手写逻辑回归 vs 规则基线的同口径对照。
① 同阈值（0.5）：模型 vs 「停滞天数规则」的精确率 / 召回率 / F1；
② 同告警量：取模型分数最高的 K 名（K = 规则告警数），比较召回率提升。
产出 out/n12_churn_bench.json。模型 AUC 低于阈值只软告警，不硬失败。
"""
from __future__ import annotations
import argparse
import json
import os
import sys
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
from n12_churn import auc_permille, gen_dataset, mulberry32, pr_at, predict, train # noqa: E402
AUC_WARN_PERMILLE = 700
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--config", default=os.path.join(HERE, "..", "config"))
ap.add_argument("--out", default=os.path.join(HERE, "..", "out"))
args = ap.parse_args()
os.makedirs(args.out, exist_ok=True)
cfg = json.load(open(os.path.join(args.config, "n12_churn.json"), encoding="utf-8"))
rows = gen_dataset(cfg, mulberry32(cfg["prngSeed"]))
split, tr = cfg["players"] * cfg["trainPct"] // 100, cfg["training"]
w, b = train(rows[:split], tr["epochs"], tr["learningRateMilli"] / 1000.0, tr["l2Milli"] / 1000.0)
test = rows[split:]
labels, scores = [r["y"] for r in test], [predict(w, b, r) for r in test]
model = pr_at(scores, labels, tr["thresholdPct"] / 100.0)
rule = cfg["ruleBaseline"]
ridx = [f["id"] for f in cfg["features"]].index("stallDays")
rf = cfg["features"][ridx]
hits = [1 if r["x"][ridx] * (rf["max"] - rf["min"]) + rf["min"] >= rule["stallDaysGte"] else 0 for r in test]
base = pr_at([float(h) for h in hits], labels, 0.5)
k, pos = base["alerts"], sum(labels)
top = set(sorted(range(len(scores)), key=lambda i: -scores[i])[:k])
tp = sum(1 for i in top if labels[i] == 1)
matched = {"alerts": k, "tp": tp, "precisionPermille": tp * 1000 // k if k else 0,
"recallPermille": tp * 1000 // pos if pos else 0}
auc = auc_permille(scores, labels)
rep = {"version": 1, "thresholdPct": tr["thresholdPct"],
"model": {x: model[x] for x in ("precisionPermille", "recallPermille", "f1Permille", "alerts")},
"rule": {x: base[x] for x in ("precisionPermille", "recallPermille", "f1Permille", "alerts")},
"matchedVolume": matched, "matchedRecallGainPermille": matched["recallPermille"] - base["recallPermille"],
"aucPermille": auc, "aucWarnThresholdPermille": AUC_WARN_PERMILLE, "warn": auc < AUC_WARN_PERMILLE}
json.dump(rep, open(os.path.join(args.out, "n12_churn_bench.json"), "w", encoding="utf-8"), ensure_ascii=False, indent=2)
print(f"{'WARN' if rep['warn'] else 'OK'} AUC={auc}permille model(p={model['precisionPermille']},"
f"r={model['recallPermille']}) rule(p={base['precisionPermille']},r={base['recallPermille']}) "
f"matchedRecallGain={rep['matchedRecallGainPermille']}permille")
return 0
if __name__ == "__main__":
raise SystemExit(main())
python3 sim/n12_churn_bench.py --config config/ --out out/ 输出：
OK AUC=881permille model(p=774,r=741) rule(p=674,r=411) matchedRecallGain=125permille
同告警量下，模型召回 53.6%、规则 41.1%，净增益 12.5 个百分点——即使在最保守的同量口径下模型依然赢。这就是 B 层基准的意义：它给「模型比规则好」这个结论加了一个可复现的下界，而不是靠一个数字自我感觉良好。脚本设了软告警线（AUC < 70% 打 WARN），但退出码恒为 0——基准是用来观察的，不是用来卡构建的。
把告警换算成钱
最后一步是运营闭环：告警 → 召回 → 回本。配置给了三个参数：每条告警成本 20 分、召回成功率 15%、成功挽回的玩家价值 1500 分。在阈值 0.5（1534 条告警）下，命中 1188 人 × 15% ≈ 178 人真正被挽回，成本 30680 分、回收价值 267000 分，两者比值约 8.7。
但这里有个陷阱：比值对「召回成功率」极其敏感，而 15% 是拍的。真实项目里成功率要靠第 13 章的 A/B 实验测出来。在测出来之前，这个 8.7 只是一个假设，不是一个结论。
小结
流失预警的完整链路是：定义流失 → 取特征 → 训练模型 → 选阈值 → 触发召回 → 算 ROI。本节用纯 Python 手写逻辑回归跑出 AUC 88.1%，同阈值下精确率 / 召回率 77.4% / 74.1%（规则基线 67.4% / 41.1%），同告警量下召回净增益 12.5 个百分点。所有数字来自合成样本，绝对数值不可移植；能搬走的是流程，以及那句最重要的提醒——预警系统真正的旋钮是告警量，不是模型。
阅读导航：上一节：12.1 新手漏斗与 D1/D7/D30 · 下一节：12.3 留存与付费的联合分析 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。