装配清单让方案「接得住」，但接住之后还有第二个问题：整条链真的跑得通吗？一份方案如果漏斗算不出末段人数、ARPDAU 凑不出回本日、月度预测一到幂运算就爆掉精度，那它只是一张好看的表格。本节做验收——把三卷的模型串成一条链，逐段跑出数字、逐段过门禁，用可复现的方式回答「这份方案能不能上线」。先说分工：本站有一篇小程序小游戏开发综述 ，那是单篇概览，把引擎、性能与变现并列罗列；本套三卷给的是可运行、可复现的链路——同一条链在 Python 与 TypeScript 里逐位对齐，不是两段互不相干的代码。
16.2 全链路仿真验收
16.2.1 验收对象是一条链，不是十五份报告
三卷各自都有验收：卷一验设计案（见交付物验收与下一卷衔接
），卷二验 CI 流水线（见持续集成与发布流水线
），卷三验漏斗与收入模型。但装配后的方案是一条链：漏斗末段人数喂给留存锚点，留存曲线喂给 LTV，LTV 与 ARPDAU 一起决定回本日，回本日再喂给月度预测——链上任何一段算错，后面全塌。所以验收的粒度不是「十五份报告各自及格」，而是**「一条链逐段过门」**：本节把这条链写成 config/l16_fullchain.json 里的五段与六道门——G1 漏斗末段通过率、G2 D1 留存、G3 D30 留存、G4 ARPDAU、G5 回本日、G6 LTV30 对 CPA 的覆盖（软门禁）。硬门禁全过，整条链才判可上线。
16.2.2 全链路仿真（S 层）
下面这段是 S 层全链路仿真脚本。它读 l16_fullchain、l07_funnel、l03_revenue、l03_ltv、l03_sensitivity、economy 六份配置，依次跑漏斗、留存、ARPDAU、LTV 与回本、月度预测，把链上每个中间量喂给六道门禁，产出 JSON、CSV 与一张漏斗图。全整数、地板除、幂用大整数。
#!/usr/bin/env python3
# file: sim/l16_fullchain.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/l16_fullchain.py --config config/ --out out/
"""全链路仿真验收（S 层）：把三卷各自的模型串成一条链，逐段跑出数字再用门禁判定。
① 漏斗 l07_funnel ② 留存锚点 ③ ARPDAU l03_revenue ④ LTV/回本 l03_ltv ⑤ 月度预测 l03_sensitivity + economy。
链上每个中间量喂给 config/l16_fullchain.json 的 gates；任一硬门禁不过即整条链判 FAIL。
产出 JSON + CSV + SVG。全整数、地板除、幂用大整数。真实参数需真实业务环境（X）。
"""
from __future__ import annotations
import argparse
import csv
import json
import os
import xml.etree.ElementTree as ET
def load(cfg_dir: str, name: str) -> dict:
with open(os.path.join(cfg_dir, f"{name}.json"), encoding="utf-8") as f:
return json.load(f)
def funnel(cfg: dict) -> tuple[int, list[dict]]:
users, steps = cfg["cohortSize"], []
for s in cfg["steps"]:
users = users * s["passPermille"] // 1000
steps.append({"id": s["id"], "name": s["name"], "users": users})
return users, steps
def ltv_deci_fen(arpdau: int, curve: list[int], days: int) -> int:
return sum(arpdau * curve[d] // 1000 for d in range(min(days, len(curve))))
def payback_day(arpdau: int, curve: list[int], cpa_fen: int) -> int:
target, cum = cpa_fen * 10, 0
for d in range(len(curve)):
cum += arpdau * curve[d] // 1000
if cum >= target:
return d + 1
return -1
def svg(path: str, steps: list[dict], cohort: int, gates: list[dict]) -> int:
"""纯 Python SVG：漏斗逐段人数横条 + 门禁判定文字。"""
W, PAD, ROW = 720, 150, 30
H = 90 + ROW * len(steps) + 24 * len(gates)
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 = "《合成小岛》全链路验收：新手漏斗逐段人数（示例参数）"
bmax, hi = W - PAD - 150, max((s["users"] for s in steps), default=1) or 1
for i, s in enumerate(steps):
y, w = 90 + i * ROW, bmax * s["users"] // hi
ET.SubElement(r, "text", {"x": "16", "y": str(y + 18), "font-size": "11", "fill": "#374151"}).text = s["name"]
ET.SubElement(r, "rect", {"x": str(PAD), "y": str(y + 4), "width": str(w), "height": "18", "rx": "4", "fill": "#2563eb"})
ET.SubElement(r, "text", {"x": str(PAD + w + 8), "y": str(y + 18), "font-size": "11", "fill": "#6b7280"}).text = f'{s["users"]} 人 · {s["users"] * 1000 // cohort}‰'
y0 = 90 + ROW * len(steps) + 4
for k, g in enumerate(gates):
ET.SubElement(r, "text", {"x": "16", "y": str(y0 + k * 24), "font-size": "12", "fill": "#16a34a" if g["pass"] else "#dc2626"}).text = f'[{"PASS" if g["pass"] else "FAIL"}] {g["id"]} {g["name"]}：{g["metric"]}={g["value"]} {g["op"]} {g["threshold"]}'
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()
fc, fn = load(a.config, "l16_fullchain"), load(a.config, "l07_funnel")
rev, ltv = load(a.config, "l03_revenue"), load(a.config, "l03_ltv")
sens, econ = load(a.config, "l03_sensitivity"), load(a.config, "economy")
end_users, steps = funnel(fn)
cohort = fn["cohortSize"]
iaa = rev["adRequestsPerDau"] * rev["fillPct"] * rev["ecpmFen"] // 10000
iap = rev["arppuFen"] * rev["payRatePermille"] // 100
arpdau = iaa + iap
daily = rev["dau"] * arpdau // 10
ltv30 = ltv_deci_fen(ltv["arpdauDeciFen"], ltv["retentionPermille"], ltv["horizonDays"])
payback = payback_day(ltv["arpdauDeciFen"], ltv["retentionPermille"], ltv["cpaFen"])
cover = ltv30 * 1000 // ltv["cpaFen"] if ltv["cpaFen"] else 0
m6 = sens["forecastMonths"] - 1
monthly = daily * 30 * (sens["growthNum"] ** m6) // (sens["growthDen"] ** m6)
e = econ["economy"]
max_out = e["baseOutputPerMinute"] * (e["levelBonusNum"] ** (e["maxLevel"] - 1)) // (e["levelBonusDen"] ** (e["maxLevel"] - 1))
metrics = {"funnelEndPermille": end_users * 1000 // cohort, "d1Permille": fn["retentionPermille"]["d1"],
"d30Permille": fn["retentionPermille"]["d30"], "arpdauDeciFen": arpdau,
"paybackDay": payback, "ltvCoverCpaPermille": cover}
gates = [{**g, "value": metrics[g["metric"]],
"pass": (metrics[g["metric"]] >= g["threshold"]) if g["op"] == ">=" else (metrics[g["metric"]] <= g["threshold"])}
for g in fc["gates"]]
ok = all(g["pass"] for g in gates if g["hard"])
os.makedirs(a.out, exist_ok=True)
sb = svg(os.path.join(a.out, "l16_fullchain.svg"), steps, cohort, gates)
summary = {"version": fc["version"], "source": "sim/l16_fullchain.py", "game": fc["game"], "observedAt": fc["observedAt"],
"chain": fc["chain"], "cohort": cohort, "funnelSteps": steps, "endUsers": end_users,
"funnelEndPermille": metrics["funnelEndPermille"], "retention": fn["retentionPermille"],
"iaaDeciFen": iaa, "iapDeciFen": iap, "arpdauDeciFen": arpdau, "dailyFen": daily,
"ltv30DeciFen": ltv30, "paybackDay": payback, "ltvCoverCpaPermille": cover,
"monthlyFen": monthly, "maxLevelOutput": max_out, "gates": gates, "ok": ok, "svgBytes": sb}
with open(os.path.join(a.out, "l16_fullchain.json"), "w", encoding="utf-8") as f:
json.dump(summary, f, ensure_ascii=False, indent=2)
with open(os.path.join(a.out, "l16_fullchain.csv"), "w", encoding="utf-8", newline="") as f:
w = csv.writer(f)
w.writerow(["gate", "name", "metric", "value", "op", "threshold", "hard", "pass"])
for g in gates:
w.writerow([g["id"], g["name"], g["metric"], g["value"], g["op"], g["threshold"], g["hard"], g["pass"]])
print(json.dumps({"endUsers": end_users, "funnelEndPermille": metrics["funnelEndPermille"],
"d1Permille": metrics["d1Permille"], "d30Permille": metrics["d30Permille"],
"arpdauDeciFen": arpdau, "dailyFen": daily, "ltv30DeciFen": ltv30, "paybackDay": payback,
"ltvCoverCpaPermille": cover, "monthlyFen": monthly, "maxLevelOutput": max_out,
"gates": {g["id"]: g["pass"] for g in gates}, "ok": ok, "svgBytes": sb}, ensure_ascii=False, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())
本机实跑输出（同机相对量级，绝对数值不可移植）：
{
"endUsers": 35810,
"funnelEndPermille": 358,
"d1Permille": 420,
"d30Permille": 90,
"arpdauDeciFen": 655,
"dailyFen": 1310000,
"ltv30DeciFen": 2300,
"paybackDay": 8,
"ltvCoverCpaPermille": 15333,
"monthlyFen": 45559471,
"maxLevelOutput": 170,
"gates": { "G1": true, "G2": true, "G3": true, "G4": true, "G5": true, "G6": true },
"ok": true,
"svgBytes": 2759
}
六道门全 PASS，"ok": true。注意链上两个跨语言风险点：ARPDAU 是 655（十进分），回本日 paybackDay 落在第 8 天——这两个数后面要在 TypeScript 里逐位重算，一位都不能差。
16.2.3 跨语言验收内核（P 层）
下面这段是 P 层跨语言验收内核。它把同一条链在 TypeScript 里重写一遍，只依赖 ../l03/revenue.ts 里已有的收入函数，用 BigInt 做幂运算、用整数地板除，保证与 Python 侧逐位一致。配套的 core/test/l16_acceptance.test.ts 有六条断言，把 35810 / 358 / 655 / 8 / 15333 / 45559471 全部钉死。
// file: core/src/l16/acceptance.ts
// tier: P
// verify: node --test 'core/test/l16_*.test.ts'
//
// 全链路验收内核（本卷 16.2 的 P 层）。把三卷的模型串成一条链——新手漏斗 → 留存锚点 →
// ARPDAU → LTV 与回本 → 月度预测——并逐段用门禁判定。与 sim/l16_fullchain.py 消费同一份
// config/*.json，逐位一致。只用可擦除语法（禁 enum / namespace / 参数属性 / 装饰器）；
// 配置里禁浮点；幂运算用 BigInt；除法一律整数地板除。
import { arpdauDeciFen, ltvDeciFen, paybackDay, type RevenueInputs } from "../l03/revenue.ts";
export type GateOp = ">=" | "<=";
export interface GateCfg { readonly id: string; readonly name: string; readonly metric: string; readonly op: GateOp; readonly threshold: number; readonly hard: boolean; }
export interface FullChainCfg { readonly chain: readonly string[]; readonly gates: readonly GateCfg[]; }
export interface GateResult { readonly id: string; readonly metric: string; readonly value: number; readonly op: GateOp; readonly threshold: number; readonly hard: boolean; readonly pass: boolean; }
export interface ChainMetrics { readonly funnelEndPermille: number; readonly d1Permille: number; readonly d30Permille: number; readonly arpdauDeciFen: number; readonly paybackDay: number; readonly ltvCoverCpaPermille: number; }
export interface ChainInputs {
readonly cohort: number;
readonly funnelPassPermille: readonly number[];
readonly retentionAnchors: { readonly d1: number; readonly d7: number; readonly d30: number };
readonly revenue: RevenueInputs;
readonly ltvArpdauDeciFen: number;
readonly ltvCurve: readonly number[];
readonly cpaFen: number;
readonly horizonDays: number;
}
/** ① 漏斗：每步人数 = 上一步 × passPermille // 1000。 */
export function funnelEnd(cohort: number, passPermille: readonly number[]): number {
let users = cohort;
for (const p of passPermille) users = Math.floor((users * p) / 1000);
return users;
}
/** ① 漏斗末段通过率（千分比）。 */
export function funnelEndPermille(cohort: number, passPermille: readonly number[]): number {
return Math.floor((funnelEnd(cohort, passPermille) * 1000) / cohort);
}
/** ⑤ 满级每分钟产出：base × (num/den)^(maxLevel-1)，幂用 BigInt 地板除。 */
export function maxLevelOutput(base: number, num: number, den: number, maxLevel: number): number {
const n = BigInt(num) ** BigInt(maxLevel - 1);
const d = BigInt(den) ** BigInt(maxLevel - 1);
return Number((BigInt(base) * n) / d);
}
/** ④ LTV30 对 CPA 的覆盖（千分比）：LTV30 × 1000 // CPA。 */
export function ltvCoverCpaPermille(ltv30: number, cpaFen: number): number {
return cpaFen === 0 ? 0 : Math.floor((ltv30 * 1000) / cpaFen);
}
/** 单条门禁判定：op 为 >= 或 <=。 */
export function evalGate(g: GateCfg, value: number): boolean {
return g.op === ">=" ? value >= g.threshold : value <= g.threshold;
}
/** 跑完整条链，返回链上所有中间量。 */
export function runFullChain(i: ChainInputs): ChainMetrics {
const ltv30 = ltvDeciFen(i.ltvArpdauDeciFen, [...i.ltvCurve], i.horizonDays);
return { funnelEndPermille: funnelEndPermille(i.cohort, i.funnelPassPermille),
d1Permille: i.retentionAnchors.d1, d30Permille: i.retentionAnchors.d30,
arpdauDeciFen: arpdauDeciFen(i.revenue),
paybackDay: paybackDay(i.ltvArpdauDeciFen, [...i.ltvCurve], i.cpaFen),
ltvCoverCpaPermille: ltvCoverCpaPermille(ltv30, i.cpaFen) };
}
/** 逐条判定门禁；整条链是否可上线 = 所有硬门禁通过。 */
export function evaluateChain(cfg: FullChainCfg, m: ChainMetrics): { results: GateResult[]; ok: boolean } {
const table: Record<string, number> = { ...m };
const results: GateResult[] = cfg.gates.map((g) => {
const value = table[g.metric];
return { id: g.id, metric: g.metric, value, op: g.op, threshold: g.threshold, hard: g.hard, pass: evalGate(g, value) };
});
return { results, ok: results.every((r) => r.pass || !r.hard) };
}
node --test 'core/test/l16_*.test.ts' 六条断言全绿：末段 35810、通过率 358‰、ARPDAU 655、回本第 8 天、覆盖 15333‰、第 6 个月 45559471 分，六道门全 PASS；另有一条反例断言——把 G4 阈值抬到 700，整条链立刻判 FAIL。Python 与 TypeScript 逐位一致，是这条链「可复现」的硬证据。
16.2.4 稳健性基准（B 层）
下面这段是 B 层稳健性基准。它把五个收入参数（广告请求、填充率、eCPM、付费率、ARPPU）同时乘 pct//100，用同一个 ARPDAU 重算六道门，看哪道硬门禁在 ±10% 扰动下先翻红。它是基准不是门禁：退出码恒为 0，只打软告警。
#!/usr/bin/env python3
# file: sim/l16_fullchain_bench.py
# tier: B
# verify: /opt/homebrew/bin/python3 sim/l16_fullchain_bench.py --config config/ --out out/
"""B 基准：全链路验收对收入参数的稳健性（软告警不硬失败）。
把五个收入参数同时乘 pct//100（带上限截断），用同一个 ARPDAU 重算门禁；任一**硬门禁**
在 ±10%（pct 90 或 110）扰动下翻红即打 WARN。退出码恒为 0——这是基准，不是门禁。
全整数、无浮点。参数均为示例假设值，真实业务波动需真实环境，命令行不可复现（X）。
"""
from __future__ import annotations
import argparse
import json
import os
from l16_fullchain import load, ltv_deci_fen, payback_day, funnel
PCTS, WARN, CAPS = [80, 90, 110, 120], {90, 110}, {"fillPct": 100, "payRatePermille": 1000}
KEYS = ("adRequestsPerDau", "fillPct", "ecpmFen", "payRatePermille", "arppuFen")
def arpdau(rev: dict) -> int:
return rev["adRequestsPerDau"] * rev["fillPct"] * rev["ecpmFen"] // 10000 + rev["arppuFen"] * rev["payRatePermille"] // 100
def gates_state(fc: dict, fn: dict, ltv: dict, a: int) -> dict[str, bool]:
end_users, _ = funnel(fn)
curve = ltv["retentionPermille"]
m = {"funnelEndPermille": end_users * 1000 // fn["cohortSize"], "d1Permille": fn["retentionPermille"]["d1"],
"d30Permille": fn["retentionPermille"]["d30"], "arpdauDeciFen": a,
"paybackDay": payback_day(a, curve, ltv["cpaFen"]),
"ltvCoverCpaPermille": ltv_deci_fen(a, curve, ltv["horizonDays"]) * 1000 // ltv["cpaFen"]}
return {g["id"]: (m[g["metric"]] >= g["threshold"]) if g["op"] == ">=" else (m[g["metric"]] <= g["threshold"]) for g in fc["gates"]}
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--config", default="config/")
ap.add_argument("--out", default="out/")
a = ap.parse_args()
fc, fn, rev, ltv = (load(a.config, "l16_fullchain"), load(a.config, "l07_funnel"),
load(a.config, "l03_revenue"), load(a.config, "l03_ltv"))
base = gates_state(fc, fn, ltv, arpdau(rev))
cases = []
for pct in PCTS:
pr = {k: (min(rev[k] * pct // 100, CAPS[k]) if k in CAPS else rev[k] * pct // 100) for k in KEYS}
st = gates_state(fc, fn, ltv, arpdau(pr))
cases.append({"pct": pct, "arpdauDeciFen": arpdau(pr), "flips": [g for g in st if st[g] != base[g]]})
hard = {g["id"] for g in fc["gates"] if g["hard"]}
warn = [f'±{abs(c["pct"] - 100)}% 扰动下硬门禁翻红：{[g for g in c["flips"] if g in hard]}'
for c in cases if c["pct"] in WARN and any(g in hard for g in c["flips"])]
os.makedirs(a.out, exist_ok=True)
rep = {"source": "sim/l16_fullchain_bench.py", "game": fc["game"], "perturbPcts": PCTS,
"baseArpdauDeciFen": arpdau(rev), "cases": cases, "maxFlipCount": max(len(c["flips"]) for c in cases),
"warnings": warn, "verdict": "PASS(soft)" if not warn else f"WARN(soft) {len(warn)} 条"}
with open(os.path.join(a.out, "l16_fullchain_bench.json"), "w", encoding="utf-8") as f:
json.dump(rep, f, ensure_ascii=False, indent=2)
print(json.dumps({"baseArpdauDeciFen": rep["baseArpdauDeciFen"],
"cases": [{"pct": c["pct"], "arpdauDeciFen": c["arpdauDeciFen"], "flips": c["flips"]} for c in cases],
"maxFlipCount": rep["maxFlipCount"], "warnings": warn, "verdict": rep["verdict"]},
ensure_ascii=False, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())
真跑结果（同机相对量级，绝对数值不可移植）：基准 ARPDAU 655（G4 阈值 600），乘 80% 掉到 360、乘 90% 掉到 493，两次都触发 G4 翻红；乘 110% 涨到 812、乘 120% 涨到 1004，门禁全稳。verdict 打出 WARN(soft) 1 条，退出码仍是 0。这条告警正是基准的价值：G4 是整条链最脆的一环——收入端只要缩水 10%，ARPDAU 就跌破硬门禁。方案交付时要把这句写进风险页，而不是假装链很结实。
16.2.5 小结
- 验收的对象是一条链：漏斗 → 留存 → ARPDAU → LTV 回本 → 月度预测，链上每一段都是下一段的输入。S 层脚本真跑出
35810 / 358 / 655 / 8 / 15333，六道门全 PASS；P 层内核在 TypeScript 里逐位重算同一组数字并钉进测试。 - B 层基准诚实暴露
G4是最脆的硬门禁：收入缩水 10% 即翻红，退出码仍为 0——基准只报警、不拦截。
阅读导航：上一节：16.1 把三卷装配成完整方案 · 下一节：16.3 原型回放与方案交付 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。