前十一节讲的是数值、概率与商业化，从这一节起把镜头对准人。留存曲线的形状与口径在第 11 章已经交代过，本章不重复，而是讲分群诊断与预警：不是画一条留存曲线，而是回答「哪一类新玩家在漏斗的哪一步掉了、D1/D7/D30 在哪个分群上塌了」。站内既有的 长线运营留存设计 是一份留存设计清单——把哪些机制、哪些节点摆上去能把人留住；本节给的是可跑的分群诊断工具——漏斗各步的人数、分群队列的 D1/D7/D30，全部由脚本算出来。清单管「该做什么」，脚本管「现在掉在哪」。
先把本章三节的分工说清楚：12.1（本节）算漏斗与队列留存，做描述性诊断；12.2 用特征预测谁会流失并触发召回；12.3 把留存与付费放在一起做联合分析。LTV 建模在第 11 章，实验设计在第 13 章，都不在这里重复。
新手漏斗：把「首日」拆成几步
新手漏斗记录新玩家在首日经过的每一个关键步骤，以及每一步留下多少人。它和留存曲线的区别是：留存曲线是时间轴（第几天还回来），漏斗是流程轴（走到第几步就停了）。两者要一起看——漏斗告诉你人在哪一步走丢，留存告诉你走丢之后还会不会回来。
本套把《王朝纪元》的首日拆成六步：安装 → 首次启动 → 完成新手引导 → 建成首个建筑 → 完成首次出征 → 完成首日会话。每一步用一个整数留存率描述「上一步的人有多少走完这一步」，累积起来就是安装到首日会话的转化率。
配置放在 config/n12_retention.json（D 层），它同时定义漏斗各步与三类分群的逐日生存参数。漏斗各步的留存率依次是 93 / 78 / 95 / 90 / 84；retention.segments 里是三个相互重叠的分群：all 是全体新玩家（D1 存活 42%、日存活 88%），guided 是完成了引导的人（55% / 91%），alliance 是首日就加入联盟的人（71% / 93%）。它们不是互斥划分，而是三个「视角」——同一个玩家可能既在 guided 也在 alliance。这样做是为了对比「带不同行为标签的人群，留存差多少」。
S 层仿真：漏斗人数与分群 D1/D7/D30
下面这段是 S 层仿真脚本，读上面这份配置，一次算出两件事：漏斗各步的人数（确定性整数运算），以及每个分群的 D1/D7/D30 队列留存（用自实现 mulberry32 确定性 PRNG 逐日抽签）。
# file: sim/n12_funnel.py
# tier: S
# verify: python3 sim/n12_funnel.py --config config/ --out out/
"""新手漏斗与 D1/D7/D30 队列留存仿真。
读 config/n12_retention.json：漏斗各步整数留存率（确定性），以及分群逐日生存模型
（自实现 mulberry32 确定性 PRNG，与 proto/src/n12_retention.ts 同种子逐位一致）。
产出 out/n12_funnel.json + out/n12_funnel.csv + out/n12_funnel.svg。
"""
from __future__ import annotations
import argparse
import csv
import json
import os
from svg_line import render
MASK = 0xFFFFFFFF
def imul(a: int, b: int) -> int:
"""Math.imul 的 Python 等价：低 32 位乘积。"""
return ((a & MASK) * (b & MASK)) & MASK
def mulberry32(seed: int):
"""与 proto 侧同算法同种子，随机序列逐位一致。"""
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 build_funnel(steps: list[dict]) -> list[dict]:
"""漏斗各步：第一步给绝对人数，其后每步 = 上一步 × keepPct // 100。"""
rows = []
for i, s in enumerate(steps):
if i == 0:
enter, keep = s["enter"], 100
else:
enter, keep = rows[-1]["enter"] * s["keepPct"] // 100, s["keepPct"]
rows.append({"id": s["id"], "name": s["name"], "enter": enter, "keepPct": keep})
install = rows[0]["enter"]
for r in rows:
r["pctOfInstallPermille"] = r["enter"] * 1000 // install
return rows
def survive_pct(seg: dict, day: int) -> int:
if day == 1:
return seg["d1SurvivePct"]
return min(seg["capPct"], seg["dailySurviveBasePct"] + (day - 2) * seg["decayStepPct"])
def simulate_segment(seg: dict, cohort: int, days: int, seed: int) -> list[int]:
"""逐日生存仿真：每名玩家 day1..days 逐日抽一次随机，抽中流失即退出（不再抽）。"""
rng = mulberry32(seed)
active = [0] * (days + 1)
for _ in range(cohort):
for day in range(1, days + 1):
if int(rng() * 100) >= survive_pct(seg, day):
break
active[day] += 1
return active
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)
with open(os.path.join(args.config, "n12_retention.json"), encoding="utf-8") as fh:
cfg = json.load(fh)
cohort, days, seed0 = cfg["cohortSize"], cfg["horizonDays"], cfg["prngSeed"]
funnel = build_funnel(cfg["funnel"])
segments = []
for idx, seg in enumerate(cfg["retention"]["segments"]):
active = simulate_segment(seg, cohort, days, seed0 + idx)
d = {f"d{day}": active[day] * 1000 // cohort for day in (1, 7, 30)}
segments.append({
"id": seg["id"], "name": seg["name"],
"d1Permille": d["d1"], "d7Permille": d["d7"], "d30Permille": d["d30"],
"dailyActive": active[1:],
})
rnd = mulberry32(seed0)
report = {
"version": 1, "cohortSize": cohort, "horizonDays": days, "seed": seed0,
"funnel": funnel,
"funnelOverallPermille": funnel[-1]["enter"] * 1000 // funnel[0]["enter"],
"segments": segments,
"firstRolls": [round(rnd(), 12) for _ in range(5)],
}
with open(os.path.join(args.out, "n12_funnel.json"), "w", encoding="utf-8") as fh:
json.dump(report, fh, ensure_ascii=False, indent=2)
with open(os.path.join(args.out, "n12_funnel.csv"), "w", encoding="utf-8", newline="") as fh:
w = csv.writer(fh)
w.writerow(["step", "name", "enter", "keepPct", "pctOfInstallPermille"])
for r in funnel:
w.writerow([r["id"], r["name"], r["enter"], r["keepPct"], r["pctOfInstallPermille"]])
render([v * 1000 // cohort for v in segments[0]["dailyActive"]],
os.path.join(args.out, "n12_funnel.svg"), "retention curve: daily active permille (all new players)")
print(json.dumps({
"funnelOverallPermille": report["funnelOverallPermille"],
"funnelDrops": [r["keepPct"] for r in funnel[1:]],
"segments": [{k: s[k] for k in ("id", "d1Permille", "d7Permille", "d30Permille")} for s in segments],
"firstRolls": report["firstRolls"],
}, ensure_ascii=False))
return 0
if __name__ == "__main__":
raise SystemExit(main())
脚本的核心是 simulate_segment：每名玩家从 day1 起逐日抽一次随机数，抽中流失就退出，不再抽后面的天。survive_pct 给出某一天的存活率——day1 用 d1SurvivePct，之后用 min(capPct, dailySurviveBasePct + (day-2)*decayStepPct)，也就是存活越久、日存活率越高（幸存者偏差），曲线自然「先陡后缓」。同一份随机种子下，Python 与 TypeScript 的抽签序列逐位一致。
实跑结果
本机运行 python3 sim/n12_funnel.py --config config/ --out out/（Python 3，零依赖），stdout 一行 JSON：
{"funnelOverallPermille": 520, "funnelDrops": [93, 78, 95, 90, 84], "segments": [{"id": "all", "d1Permille": 417, "d7Permille": 228, "d30Permille": 70}, {"id": "guided", "d1Permille": 548, "d7Permille": 363, "d30Permille": 140}, {"id": "alliance", "d1Permille": 713, "d7Permille": 534, "d30Permille": 266}], "firstRolls": [0.341123865684, 0.017803263851, 0.364578908775, 0.008977568476, 0.681191388052]}
漏斗各步人数（来自 out/n12_funnel.csv）：
| 步骤 | 人数 | 相对上一步 | 占安装 |
|---|---|---|---|
| 安装 | 100000 | — | 100% |
| 首次启动 | 93000 | 93% | 93% |
| 完成新手引导 | 72540 | 78% | 72.5% |
| 建成首个建筑 | 68913 | 95% | 68.9% |
| 完成首次出征 | 62021 | 90% | 62.0% |
| 完成首日会话 | 52097 | 84% | 52.0% |
安装到首日会话的整体转化是 52.0%。缺口不在第一步：首次启动 93%、建成首建筑 95%、首战 90% 都还算稳，最大的单步流失在「完成新手引导」——只有 78% 的人走完，一步掉掉 20460 人（占安装量的 20.5%）。这就是漏斗要抓的东西：不是笼统地说「留存差」，而是定位到「引导这一关把人卡住了」。
分群队列留存（千分比，来自 out/n12_funnel.json）：
| 分群 | D1 | D7 | D30 |
|---|---|---|---|
| 全体新玩家 | 417 | 228 | 70 |
| 完成引导 | 548 | 363 | 140 |
| 首日入盟 | 713 | 534 | 266 |
三行对比把分群诊断的价值讲清楚了：全体的 D1/D7/D30 是 41.7% / 22.8% / 7.0%；完成引导的人是 54.8% / 36.3% / 14.0%；首日入盟的人是 71.3% / 53.4% / 26.6%。入盟党的 D7 留存（53.4%）是全体（22.8%）的两倍多，引导党也明显高于全体。
分群差异 ≠ 因果
这里要小心一个陷阱：分群之间的留存差异，不等于「做了这个动作就能留住人」。首日入盟的人留存高，可能有两种解释：①入盟本身带来社交绑定，让人更粘；②本来就更投入的玩家才会首日入盟（选择效应）。本节的仿真只能算出「差异有多大」，算不出「哪一半是因果」——要分因果，得靠第 13 章的 A/B 实验，把入盟入口随机对一半人开放。这个陷阱在 12.3 会再用一次，那里有可量化的做法。
但即便是描述性诊断，分群也有用：它告诉你「哪一群人值得优先服务」。引导党与入盟党都远高于全体，说明「走完引导」和「建立社交」是两个高价值的留存节点——这正好和站内那份留存设计清单的结论对得上。
P 层：TypeScript 复算与逐位一致
下面这段是 P 层原型，用 TypeScript 复刻同一份漏斗与生存模型。它和 Python 侧读同一份 config、用同一个 mulberry32 与同一个种子：
// file: proto/src/n12_retention.ts
// tier: P
// verify: node --test proto/test/n12.test.ts
// 与 sim/n12_funnel.py 同构：读同一份 config/n12_retention.json，用同算法同种子 mulberry32
// 复算新手漏斗与分群 D1/D7/D30 逐日生存曲线，与 Python 侧逐位一致。
import { loadConfig } from "./core.ts";
export interface FunnelStep { id: string; name: string; enter?: number; keepPct?: number }
export interface Segment {
id: string; name: string;
d1SurvivePct: number; dailySurviveBasePct: number; decayStepPct: number; capPct: number;
}
export interface RetentionConfig {
prngSeed: number; cohortSize: number; horizonDays: number;
funnel: FunnelStep[];
retention: { segments: Segment[] };
}
export function loadRetention(): RetentionConfig {
return loadConfig<RetentionConfig>("n12_retention");
}
// 与 sim/n12_funnel.py 同算法同种子，随机序列逐位一致；不吃运行时随机数实现。
export function mulberry32(seed: number): () => number {
let a = seed >>> 0;
return () => {
a = (a + 0x6D2B79F5) >>> 0;
let t = Math.imul(a ^ (a >>> 15), a | 1);
t = (t ^ ((t + Math.imul(t ^ (t >>> 7), t | 61)) >>> 0)) >>> 0;
return ((t ^ (t >>> 14)) >>> 0) / 4294967296;
};
}
export interface FunnelRow { id: string; name: string; enter: number; keepPct: number; pctOfInstallPermille: number }
export function buildFunnel(steps: FunnelStep[]): FunnelRow[] {
const rows: FunnelRow[] = [];
steps.forEach((s, i) => {
const enter = i === 0 ? s.enter! : Math.floor((rows[i - 1].enter * s.keepPct!) / 100);
rows.push({ id: s.id, name: s.name, enter, keepPct: i === 0 ? 100 : s.keepPct!, pctOfInstallPermille: 0 });
});
const install = rows[0].enter;
for (const r of rows) r.pctOfInstallPermille = Math.floor((r.enter * 1000) / install);
return rows;
}
function survivePct(seg: Segment, day: number): number {
if (day === 1) return seg.d1SurvivePct;
return Math.min(seg.capPct, seg.dailySurviveBasePct + (day - 2) * seg.decayStepPct);
}
export function simulateSegment(seg: Segment, cohort: number, days: number, seed: number): number[] {
const rng = mulberry32(seed);
const active = new Array<number>(days + 1).fill(0);
for (let i = 0; i < cohort; i++) {
for (let day = 1; day <= days; day++) {
if (Math.floor(rng() * 100) >= survivePct(seg, day)) break;
active[day]++;
}
}
return active;
}
export function retentionReport() {
const cfg = loadRetention();
const cohort = cfg.cohortSize, days = cfg.horizonDays, seed0 = cfg.prngSeed;
const funnel = buildFunnel(cfg.funnel);
const segments = cfg.retention.segments.map((seg, idx) => {
const active = simulateSegment(seg, cohort, days, seed0 + idx);
return {
id: seg.id, name: seg.name,
d1Permille: Math.floor((active[1] * 1000) / cohort),
d7Permille: Math.floor((active[7] * 1000) / cohort),
d30Permille: Math.floor((active[30] * 1000) / cohort),
dailyActive: active.slice(1),
};
});
const rnd = mulberry32(seed0);
const firstRolls = Array.from({ length: 5 }, () => Math.round(rnd() * 1e12) / 1e12);
return {
cohortSize: cohort, horizonDays: days, seed: seed0, funnel,
funnelOverallPermille: Math.floor((funnel[funnel.length - 1].enter * 1000) / funnel[0].enter),
segments, firstRolls,
};
}
proto/test/n12.test.ts 断言了漏斗各步人数、PRNG 前 5 个随机数与三类分群的 D1/D7/D30 与 Python 侧逐位一致。本机 node --test proto/test/n12.test.ts 输出：
✔ 12.1 漏斗各步人数与占安装比与 Python 一致
✔ 12.1 PRNG 前 5 个随机数与 Python 一致
✔ 12.1 分群 D1/D7/D30 与 Python 一致
✔ 12.1 漏斗由配置确定：改 keepPct 会改结果
ℹ tests 4
ℹ pass 4
ℹ fail 0
这就是「一份配置、两处消费」要保证的性质——如果 Python 和 TypeScript 对同一份配置算出不同的漏斗，那这份配置就不是唯一真源。
三个口径陷阱
最后交代三个容易踩的口径，脚本覆盖不到，但决定你读到的数对不对：
- 自然日 vs 24 小时。D1 是「次日自然日回访」还是「首次登录后 24 小时内回访」，两条曲线形状完全不同。跨版本比较时口径必须固定，否则会以为改对了，其实是换了尺子。
- 队列留存 vs 截面留存。D7 有两种算法：同一天进入的新玩家队列（cohort）在第 7 天还剩多少（队列留存），和「今天所有活跃玩家中登录满 7 天的占比」（截面留存）。前者才反映真实流失，后者会被新玩家涌入稀释。本节算的是队列留存。
- 分群重叠。本节三个分群是重叠视角，不能相加。要算「入盟对留存的净贡献」，必须做 12.3 那样的同口径对照。
这三条属于 X 层——需真实业务环境，命令行不可复现：脚本能算出给定口径下的数，但「你线上埋的口径对不对」只能靠真实埋点数据核对。
小结
新手漏斗 = 把首日拆成可数的几步，D1/D7/D30 = 分群队列的留存快照。本节的 S 层脚本用 config/n12_retention.json 跑出漏斗整体转化 52.0%、全体 D1/D7/D30 = 41.7%/22.8%/7.0%，并给出引导党与入盟党的对照；P 层用 TypeScript 复算，断言逐位一致。数字是同机相对量级，绝对数值不可移植，但「先分群、再看漏斗、最后追缺口」这套流程可以搬到你自己的项目。
一句话记住本节：漏斗定位缺口，分群定位人群，口径决定你信不信这两个数。下一节把「事后诊断」换成「事前预警」——用特征预测谁会在未来 7 天流失。
阅读导航：上一节：目录 · 下一节：12.2 流失预警与召回 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。