7.1 新手漏斗与 D1/D7/D30 留存

把「留存」拆成两件可算的事:首日新手漏斗(启动→加载→引导→首关→合成→离线收益→首日结算)与跨天留存曲线(D1/D7/D30 锚点整数插值)。给出可跑的 Python 漏斗仿真、同口径的 TypeScript 留存内核与 4 项单元测试,并用三档改造对照回答「改哪一步最划算」。

第 6 章讲完了「人怎么进来」——自然增长、平台流量、预算分配。但一款小游戏真正决定生死的,不是某一天来了多少人,而是来的人第二天还回不回来。本章(第 7 章)只讲留存与漏斗这条线:新手漏斗、流失预警与召回、留存与付费的联合分析。增长裂变是第 6 章,A/B 实验是第 8 章,本节不越界。

先说分工。本站有一篇独立开发者的手游广告失败复盘 ,那是从失败视角讲「留存没做好就先买量」的教训——它告诉你坑在哪;本节把同样的教训变成可跑的漏斗与留存曲线,把「留存差」量化到具体哪一步、哪一天。本站还有一篇小游戏开发综述 ,那是单篇综述口径(引擎选型 / 性能 / 变现各扫一遍);本节是成体系、可运行的专题里的一节,只做一件事:把留存算成数字。

留存不是一个数,是两条曲线

「留存率」被当成一个数来谈时,几乎一定谈错。它其实是两条不同的曲线叠在一起。第一条是首日漏斗:同一批新用户在第一天之内,从「启动游戏」一路走到「完成首日结算」,每一步都掉人,衡量的是新手体验的摩擦力。第二条是跨天留存曲线:首日结算完的人里,第二天回来多少、第七天回来多少、第三十天还回来多少,衡量的是核心循环的黏性。

两条曲线要分开算,因为改善手段完全不同:漏斗靠改引导、改节奏、改首关;留存靠改核心循环、加目标、加社交。把两者混成一个「留存率」,就永远不知道下一步该动哪里。本节先把两条曲线都做成可跑的模型。

D 层:把漏斗与留存锚点落成配置

下面这份是 D 层(静态层)产物,全整数、没有任何浮点。cohortSize 是进入漏斗的初始人数,每个步骤带一个 passPermille(通过千分比);retentionPermille 是 D1/D7/D30 三个留存锚点;scenarios 对某些步骤覆盖 passPermille,用来做改造对照。参数均为示例假设值,真实平台漏斗与留存需真实业务环境,命令行不可复现(X)。

{
  "version": 1,
  "game": "merge-island",
  "seed": 20261012,
  "observedAt": "2026-10",
  "note": "新手漏斗与留存锚点:cohortSize 为进入漏斗的初始人数,每步乘 passPermille//1000(整数地板除)得到该步人数;D1/D7/D30 为留存锚点(千分比),锚点之间整数线性插值得到日留存曲线。passPermille 与留存锚点均为示例假设参数;真实平台漏斗与留存需真实业务环境,命令行不可复现(X)。全整数,禁浮点。",
  "cohortSize": 100000,
  "days": 30,
  "steps": [
    { "id": "launch", "name": "启动游戏", "passPermille": 1000 },
    { "id": "loaded", "name": "资源加载完成", "passPermille": 960 },
    { "id": "tutorial-start", "name": "进入新手引导", "passPermille": 980 },
    { "id": "tutorial-done", "name": "完成新手引导", "passPermille": 720 },
    { "id": "first-level", "name": "通过首关", "passPermille": 850 },
    { "id": "first-merge", "name": "首次合成", "passPermille": 930 },
    { "id": "first-offline", "name": "领取首次离线收益", "passPermille": 760 },
    { "id": "d0-complete", "name": "完成首日结算", "passPermille": 880 }
  ],
  "retentionPermille": { "d1": 420, "d7": 190, "d30": 90 },
  "scenarios": [
    { "id": "baseline", "name": "基线", "stepOverrides": {} },
    { "id": "short-tutorial", "name": "引导缩短", "stepOverrides": { "tutorial-done": 860 } },
    { "id": "auto-merge", "name": "首关自动合成引导", "stepOverrides": { "first-level": 920 } },
    { "id": "offline-onboarding", "name": "离线收益前置教学", "stepOverrides": { "first-offline": 900 } }
  ]
}

八个步骤覆盖了《合成小岛》首日的完整路径:启动 → 加载 → 进引导 → 完成引导 → 过首关 → 首次合成 → 领首次离线收益 → 首日结算。留存锚点 420 / 190 / 90(‰)表示首日结算完的人里,D1 回来 42%、D7 回来 19%、D30 回来 9%。

S 层:把两条曲线算出来

下面这段是 S 层仿真脚本,读 config/l07_funnel.json:先把漏斗逐级乘出每步人数,再用 D1/D7/D30 三个锚点做整数线性插值得到日留存曲线,最后对四档场景各跑一遍漏斗。产出 JSON + CSV + SVG。

#!/usr/bin/env python3
# file: sim/l07_funnel.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/l07_funnel.py --config config/ --out out/
"""新手漏斗与 D1/D7/D30 留存:读 config/l07_funnel.json。

漏斗:从 cohortSize 起逐级乘 passPermille // 1000(整数地板除)得到每步人数;
留存:D1/D7/D30 三个锚点做整数线性插值,得到第 1..days 日的日留存千分比;
D1 / D7 / D30 人数 = 漏斗末步人数 × 对应锚点 // 1000。
scenarios 覆盖若干步的 passPermille,用来回答「改哪一步最划算」。

产出 JSON + CSV + SVG。全整数地板除,无浮点。
真实平台漏斗与留存锚点需真实业务环境,命令行不可复现(X)。
"""
from __future__ import annotations
import argparse
import csv
import json
import os
import xml.etree.ElementTree as ET


def load(d: str) -> dict:
    with open(os.path.join(d, "l07_funnel.json"), encoding="utf-8") as f:
        return json.load(f)


def step_counts(steps: list, cohort: int, overrides: dict | None = None) -> list[int]:
    """逐级漏斗人数;第 0 步恒为 cohort(passPermille 应为 1000)。"""
    ov, counts, cur = overrides or {}, [], cohort
    for i, s in enumerate(steps):
        if i > 0:
            cur = cur * ov.get(s["id"], s["passPermille"]) // 1000
        counts.append(cur)
    return counts


def daily_retention(a: dict, d: int) -> int:
    """第 d 日留存千分比(d<=0 安装日恒 1000);锚点之间整数线性插值,D30 之后持平。"""
    if d <= 0:
        return 1000
    if d <= 7:
        return a["d1"] + (a["d7"] - a["d1"]) * (d - 1) // 6
    if d <= 30:
        return a["d7"] + (a["d30"] - a["d7"]) * (d - 7) // 23
    return a["d30"]


def funnel_rows(steps: list, counts: list[int], cohort: int) -> list[dict]:
    rows = []
    for i, s in enumerate(steps):
        prev = counts[i - 1] if i > 0 else cohort
        rows.append({"id": s["id"], "name": s["name"], "passPermille": s["passPermille"],
                     "count": counts[i], "relPermille": counts[i] * 1000 // cohort,
                     "dropFromPrev": prev - counts[i]})
    return rows


def draw_svg(path: str, rows: list[dict], end_count: int, cohort: int) -> int:
    """纯 Python SVG:每步一根横条,长度正比于该步人数。"""
    W, PAD, ROW = 760, 150, 40
    H = 84 + ROW * len(rows)
    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 = \
        f"《合成小岛》新手漏斗(初始 {cohort} → 末步 {end_count})"
    span = W - PAD - 190
    for i, s in enumerate(rows):
        y, w = 54 + i * ROW, span * s["count"] // cohort
        ET.SubElement(r, "text", {"x": "16", "y": str(y + 20), "font-size": "12",
                                  "fill": "#374151"}).text = s["name"]
        ET.SubElement(r, "rect", {"x": str(PAD), "y": str(y + 6), "width": str(w), "height": "20",
                                  "rx": "4", "fill": "#2563eb"})
        ET.SubElement(r, "text", {"x": str(PAD + w + 8), "y": str(y + 21), "font-size": "12",
                                  "fill": "#6b7280"}).text = f'{s["count"]} · {s["relPermille"] / 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)
    cohort, anchors = cfg["cohortSize"], cfg["retentionPermille"]
    scenarios = []
    for sc in cfg["scenarios"]:
        end = step_counts(cfg["steps"], cohort, sc["stepOverrides"])[-1]
        scenarios.append({"id": sc["id"], "name": sc["name"], "stepOverrides": sc["stepOverrides"],
                          "endCount": end, "endRelPermille": end * 1000 // cohort,
                          "d1Count": end * anchors["d1"] // 1000,
                          "d7Count": end * anchors["d7"] // 1000,
                          "d30Count": end * anchors["d30"] // 1000})
    base = scenarios[0]
    base_rows = funnel_rows(cfg["steps"], step_counts(cfg["steps"], cohort), cohort)
    curve = [{"day": d, "retentionPermille": daily_retention(anchors, d)}
             for d in range(1, cfg["days"] + 1)]
    worst = max(base_rows, key=lambda s: s["dropFromPrev"])
    os.makedirs(a.out, exist_ok=True)
    svg = draw_svg(os.path.join(a.out, "l07_funnel.svg"), base_rows, base["endCount"], cohort)
    with open(os.path.join(a.out, "l07_funnel.csv"), "w", encoding="utf-8", newline="") as f:
        w = csv.writer(f)
        w.writerow(["id", "name", "passPermille", "count", "relPermille", "dropFromPrev"])
        for s in base_rows:
            w.writerow([s["id"], s["name"], s["passPermille"], s["count"], s["relPermille"], s["dropFromPrev"]])
    digest = ",".join(f'{s["id"]}:{s["count"]}' for s in base_rows) + \
        f'|d1:{base["d1Count"]}:d7:{base["d7Count"]}:d30:{base["d30Count"]}'
    rep = {"version": cfg["version"], "game": cfg["game"], "observedAt": cfg["observedAt"],
           "cohortSize": cohort, "days": cfg["days"], "steps": base_rows,
           "retentionPermille": anchors, "retentionCurve": curve,
           "worstDropStep": {"id": worst["id"], "name": worst["name"], "dropFromPrev": worst["dropFromPrev"]},
           "scenarios": scenarios, "digest": digest, "csv": os.path.join(a.out, "l07_funnel.csv"),
           "svg": os.path.join(a.out, "l07_funnel.svg"), "svgBytes": svg}
    with open(os.path.join(a.out, "l07_funnel.json"), "w", encoding="utf-8") as f:
        json.dump(rep, f, ensure_ascii=False, indent=2)
    print(json.dumps({"cohortSize": cohort, "endCount": base["endCount"],
                      "endRelPermille": base["endRelPermille"], "d1": base["d1Count"],
                      "d7": base["d7Count"], "d30": base["d30Count"],
                      "worstDrop": [worst["id"], worst["dropFromPrev"]],
                      "scenarioEnd": {s["id"]: s["endCount"] for s in scenarios},
                      "digest": digest}, ensure_ascii=False))
    return 0


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

脚本的核心只有两处:step_counts 逐级乘 passPermille(整数地板除),daily_retention 在锚点之间做整数线性插值。在本机跑它,输出如下(同机相对量级,绝对数值不可移植):

{
  "cohortSize": 100000,
  "endCount": 35810,
  "endRelPermille": 358,
  "d1": 15040,
  "d7": 6803,
  "d30": 3222,
  "worstDrop": ["tutorial-done", 26343],
  "scenarioEnd": { "baseline": 35810, "short-tutorial": 42774, "auto-merge": 38759, "offline-onboarding": 42407 }
}

10 万启动用户走到「完成首日结算」只剩 35810,首日漏斗整体转化 35.8%。三个跨天留存人数是 15040 / 6803 / 3222,对应 D1 42%、D7 19%、D30 9% 的锚点。注意留存人数是以漏斗末步为分母的——首日没走完的人不会进入留存统计,这也是为什么漏斗和留存必须一起看:漏斗漏掉的 64%,在留存曲线上根本不会出现。

读表:最大的流失口在哪

脚本会标出流失最大的一步。本例是 tutorial-done(完成新手引导),一步流失 26343 人——占全部启动用户的四分之一强。这不是偶然:新手引导是所有小游戏最大的漏斗口,因为它同时叠加了「加载等待」「陌生操作」「看不懂的目标」三重摩擦。

改造覆盖步骤末步人数相对基线
基线—35810—
引导缩短(720→860‰)tutorial-done42774+6964
离线收益前置教学(760→900‰)first-offline42407+6597
首关自动合成引导(850→920‰)first-level38759+2949

这张表回答了一个运营上非常实际的问题:同样的改造力气,投在哪一步回报最大。 引导缩短把末步人数抬到 42774,比首关优化(38759)多出 4000 多人。原因不在单步通过率本身,而在位置:越靠前的步骤,后面所有步骤都以它为基数往下乘,所以靠前步骤的通过率提升会被整条链放大。这就是「漏斗前段优先」的量化依据。

P 层:把口径锁进内核

仿真跑通了,但它只是 Python 里的一段验证。要让游戏真正用上这套口径——比如在运营后台实时算「今天这批人的 D1 预测」——还需要一份能在客户端 / 服务端跑的 TypeScript 内核。下面这段是 P 层产物 core/src/l07/funnel.ts,与 Python 脚本同一份 config、同一套整数地板除,因此逐位一致。全整数、无 Math.random()。

// file: core/src/l07/funnel.ts
// tier: P
// verify: node --test 'core/test/l07_*.test.ts'
// 《合成小岛》新手漏斗与留存曲线内核:逐级漏斗 + 锚点插值留存。
// 与 sim/l07_funnel.py 同一份 config、同一套整数地板除,逐位一致。
// 全整数、无 Math.random()、只用可擦除语法(禁 enum / namespace / 参数属性 / 装饰器)。

export interface FunnelStep { readonly id: string; readonly name: string; readonly passPermille: number; }
export interface RetentionAnchors { readonly d1: number; readonly d7: number; readonly d30: number; }
export interface FunnelConfig {
  readonly cohortSize: number; readonly days: number;
  readonly steps: readonly FunnelStep[]; readonly retentionPermille: RetentionAnchors;
}
export interface FunnelStepRow {
  readonly id: string; readonly name: string; readonly count: number;
  readonly relPermille: number; readonly dropFromPrev: number;
}

const fdiv = (a: number, b: number): number => Math.floor(a / b);

/** 逐级漏斗人数:第 0 步恒为 cohort,其后每步乘 passPermille // 1000。 */
export function stepCounts(steps: readonly FunnelStep[], cohort: number,
  overrides: Record<string, number> = {}): number[] {
  const counts: number[] = [];
  let cur = cohort;
  for (let i = 0; i < steps.length; i += 1) {
    if (i > 0) {
      const p = overrides[steps[i].id] ?? steps[i].passPermille;
      cur = fdiv(cur * p, 1000);
    }
    counts.push(cur);
  }
  return counts;
}

/** 第 d 日留存千分比(d<=0 恒 1000);D1/D7/D30 锚点之间整数线性插值,D30 之后持平。 */
export function dailyRetention(a: RetentionAnchors, d: number): number {
  if (d <= 0) return 1000;
  if (d <= 7) return a.d1 + fdiv((a.d7 - a.d1) * (d - 1), 6);
  if (d <= 30) return a.d7 + fdiv((a.d30 - a.d7) * (d - 7), 23);
  return a.d30;
}

/** 留存曲线:第 1..days 日的日留存千分比。 */
export function retentionCurve(a: RetentionAnchors, days: number): number[] {
  const out: number[] = [];
  for (let d = 1; d <= days; d += 1) out.push(dailyRetention(a, d));
  return out;
}

/** 漏斗行:人数、相对首步转化(‰)、本步流失。 */
export function funnelRows(steps: readonly FunnelStep[], counts: readonly number[],
  cohort: number): FunnelStepRow[] {
  return steps.map((s, i) => {
    const prev = i > 0 ? counts[i - 1] : cohort;
    return { id: s.id, name: s.name, count: counts[i],
      relPermille: fdiv(counts[i] * 1000, cohort), dropFromPrev: prev - counts[i] };
  });
}

配套的 P 层测试 core/test/l07_funnel.test.ts 把四条性质锁进断言——逐级漏斗整数地板除、锚点命中且 D30 之后持平、留存曲线单调不增、以及与 Python 逐位一致(末步 35810,D1/D7/D30 = 15040 / 6803 / 3222)。node --test 'core/test/l07_*.test.ts' 本机 4 项全过:

✔ 逐级漏斗:每步整数地板除,首步恒为 cohort
✔ 留存锚点插值:D1/D7/D30 命中锚点,D30 之后持平
✔ 留存曲线单调不增(锚点递减 + 线性插值)
✔ 与 Python 仿真同口径 + 无隐藏随机
ℹ tests 4
ℹ pass 4
ℹ fail 0

逐位一致不是巧合,而是「同一份 config + 同一套整数地板除」的必然结果——包括负数插值那一行,Python 的 // 与 TS 的 Math.floor 在负方向上的取整行为也一致。

小结与下一步

这一节把「留存」从一个笼统的百分比,拆成了两条可算的曲线:首日漏斗(同一批人第一天的逐级流失)与跨天留存(首日结算完的人第 2/7/30 天回来多少)。Python 仿真、TS 内核、4 项测试三处同口径,并证明「漏斗前段优先」——靠前步骤的通过率提升会被整条链放大。

但漏斗与留存都是事后统计:它们告诉你昨天掉的人多,却不能在玩家真的要掉之前把他捞回来。下一节讲流失预警与召回:用一批可播种的合成用户训练一个加权风险分,算清精确率 / 召回率,再对高风险人群算召回渠道的成本-回收账。

阅读导航:上一节:目录 · 下一节:流失预警与召回 。

继续阅读

探索更多技术文章

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

全部文章 返回首页

「minigame」更多文章

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