站内已有一篇H5 游戏产业全景 ，那篇是产业综述口径，讲的是这条赛道整体怎么长、钱在哪儿流；本节给的是可跑的预算分配模型：同样一笔固定日预算，怎么在买量 / 平台 / 裂变 / 内容四个渠道之间分配，总新增最大。区别在写法：那篇描述「行业怎么分配资源」，本节把「一笔钱该怎么分」变成一段能跑出数字的贪心水填。
增长预算分配不是「平均分」
上一节算清了各渠道的效率，但没回答一个更实际的问题：手里只有一笔固定的日预算，该怎么分？ 最直觉的做法是平均分给每个渠道，但这几乎一定不是最优——因为每个渠道的边际收益是递减的。
边际收益递减的意思是：往一个渠道投第一笔钱，回报最高；投第二笔，回报低一点；投得越多，下一笔的回报越低。原因很实在：买量的第一批用户最容易转化，越往后越贵；裂变的第一批玩家好友最多，越往后邀请越难。既然每个渠道的边际回报都在下降，那么最优分配一定是「每一步都把钱投给当前边际回报最高的那个渠道」——这就是贪心水填（water-filling）。
D 层：把渠道效率落成配置
下面这份是 D 层（静态层）产物，全整数。每个渠道有三个参数：baseNewPerUnit（第一份预算的边际新增）、decayNum/decayDen（每多投一份，边际新增乘这个比例），另有日预算 dailyBudgetCent 与最小单位 unitCent。
{
"version": 1,
"game": "merge-island",
"seed": 20261012,
"note": "增长预算分配：把日预算按固定最小单位（unitCent）拆成若干份，逐份投给「当前边际新增最高」的渠道。每渠道边际新增按几何衰减（baseNewPerUnit × decayNum^(m-1) / decayDen^(m-1)）递减，体现边际收益递减。全整数，禁浮点。baseNewPerUnit / decay 是假设参数，真实渠道效率需真实业务环境，命令行不可复现（X）。",
"dailyBudgetCent": 60000,
"unitCent": 5000,
"channels": [
{ "id": "ua", "name": "买量投放", "baseNewPerUnit": 220, "decayNum": 4, "decayDen": 5 },
{ "id": "platform", "name": "平台流量位", "baseNewPerUnit": 300, "decayNum": 3, "decayDen": 5 },
{ "id": "viral", "name": "裂变激励", "baseNewPerUnit": 180, "decayNum": 7, "decayDen": 10 },
{ "id": "content", "name": "内容与私域", "baseNewPerUnit": 120, "decayNum": 9, "decayDen": 10 }
]
}
S 层：贪心水填
下面这段是 S 层分配脚本，读 config/l06_budget.json，逐份投给边际新增最高的渠道，再与「平均分」对照，产出 JSON + CSV + SVG。
#!/usr/bin/env python3
# file: sim/l06_budget_alloc.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/l06_budget_alloc.py --config config/ --out out/
"""增长预算分配：读 config/l06_budget.json。
把日预算按最小单位（unitCent）拆成若干份，逐份投给「当前边际新增最高」的渠道，
即贪心水填（water-filling）。第 m 份的边际新增 = baseNewPerUnit × decayNum^(m-1) // decayDen^(m-1)，
这就是「边际收益递减」：同一渠道投得越多，下一份的边际新增越低。
贪心结果与「平均分给每个渠道」对照，证明同样预算下水填拿到的总新增更高。
产出 JSON + CSV + SVG。真实渠道效率（baseNewPerUnit / decay）需真实业务环境，命令行不可复现（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, "l06_budget.json"), encoding="utf-8") as f:
return json.load(f)
def marginal_new(base: int, dn: int, dd: int, m: int) -> int:
"""第 m 份预算的边际新增（整数，几何衰减）。m 从 1 开始。"""
return base * (dn ** (m - 1)) // (dd ** (m - 1))
def channel_total(ch: dict, units: int) -> int:
return sum(marginal_new(ch["baseNewPerUnit"], ch["decayNum"], ch["decayDen"], m)
for m in range(1, units + 1))
def allocate(channels: list, budget_units: int) -> dict:
"""贪心水填：每一步把一份预算给边际新增最高的渠道（并列时取列表靠前者）。"""
used = {c["id"]: 0 for c in channels}
trace = []
for step in range(budget_units):
best, best_val = None, -1
for c in channels:
val = marginal_new(c["baseNewPerUnit"], c["decayNum"], c["decayDen"], used[c["id"]] + 1)
if val > best_val:
best_val, best = val, c
used[best["id"]] += 1
trace.append({"step": step + 1, "channel": best["id"], "marginalNew": best_val})
per = [{"id": c["id"], "name": c["name"], "units": used[c["id"]],
"newUsers": channel_total(c, used[c["id"]])} for c in channels]
return {"allocation": used, "perChannel": per, "trace": trace,
"totalNew": sum(p["newUsers"] for p in per),
"lastMarginal": trace[-1]["marginalNew"] if trace else 0}
def even_split(channels: list, budget_units: int) -> dict:
"""对照：把预算尽量平均分给每个渠道。"""
base, rem = divmod(budget_units, len(channels))
per = [{"id": c["id"], "name": c["name"], "units": base + (1 if i < rem else 0)}
for i, c in enumerate(channels)]
for p in per:
ch = next(c for c in channels if c["id"] == p["id"])
p["newUsers"] = channel_total(ch, p["units"])
return {"allocation": {p["id"]: p["units"] for p in per}, "perChannel": per,
"totalNew": sum(p["newUsers"] for p in per)}
def draw_svg(path: str, rows: list[dict], budget_units: int) -> int:
W, H, PAD, ROW = 720, 70 + 46 * len(rows), 200, 46
peak = max((r["units"] for r in rows), default=1) or 1
svg = ET.Element("svg", {"xmlns": "http://www.w3.org/2000/svg", "width": str(W),
"height": str(H), "viewBox": f"0 0 {W} {H}"})
ET.SubElement(svg, "rect", {"width": str(W), "height": str(H), "fill": "#ffffff"})
ET.SubElement(svg, "text", {"x": "16", "y": "28", "font-size": "15", "fill": "#111827"}).text = \
f"贪心水填的预算分配（共 {budget_units} 份）"
for i, r in enumerate(rows):
y, w = 54 + i * ROW, (W - PAD - 40) * r["units"] // peak
ET.SubElement(svg, "text", {"x": "16", "y": str(y + 20), "font-size": "12",
"fill": "#374151"}).text = r["name"]
ET.SubElement(svg, "rect", {"x": str(PAD), "y": str(y + 6), "width": str(w), "height": "22",
"rx": "4", "fill": "#2563eb"})
ET.SubElement(svg, "text", {"x": str(PAD + w + 8), "y": str(y + 22), "font-size": "12",
"fill": "#6b7280"}).text = f'{r["units"]} 份 · 新增 {r["newUsers"]}'
os.makedirs(os.path.dirname(path), exist_ok=True)
ET.ElementTree(svg).write(path, encoding="utf-8", xml_declaration=True)
return len(ET.tostring(svg))
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)
units = cfg["dailyBudgetCent"] // cfg["unitCent"]
channels = cfg["channels"]
greedy = allocate(channels, units)
even = even_split(channels, units)
budget_cent = units * cfg["unitCent"]
os.makedirs(a.out, exist_ok=True)
svg = draw_svg(os.path.join(a.out, "l06_budget_alloc.svg"), greedy["perChannel"], units)
even_by_id = {p["id"]: p for p in even["perChannel"]}
with open(os.path.join(a.out, "l06_budget_alloc.csv"), "w", encoding="utf-8", newline="") as f:
w = csv.writer(f)
w.writerow(["channelId", "name", "units", "newUsers", "evenUnits", "evenNewUsers"])
for p in greedy["perChannel"]:
e = even_by_id[p["id"]]
w.writerow([p["id"], p["name"], p["units"], p["newUsers"], e["units"], e["newUsers"]])
rep = {"version": cfg["version"], "game": cfg["game"], "seed": cfg["seed"],
"dailyBudgetCent": cfg["dailyBudgetCent"], "unitCent": cfg["unitCent"], "budgetUnits": units,
"greedy": {"allocation": greedy["allocation"], "perChannel": greedy["perChannel"],
"totalNew": greedy["totalNew"], "lastMarginal": greedy["lastMarginal"],
"blendedCacCent": budget_cent // greedy["totalNew"] if greedy["totalNew"] else 0},
"even": {"allocation": even["allocation"], "totalNew": even["totalNew"]},
"greedyGainOverEven": greedy["totalNew"] - even["totalNew"],
"csv": os.path.join(a.out, "l06_budget_alloc.csv"),
"svg": os.path.join(a.out, "l06_budget_alloc.svg"), "svgBytes": svg}
with open(os.path.join(a.out, "l06_budget_alloc.json"), "w", encoding="utf-8") as f:
json.dump(rep, f, ensure_ascii=False, indent=2)
print(json.dumps({"budgetUnits": units, "allocation": greedy["allocation"],
"totalNew": greedy["totalNew"], "evenTotalNew": even["totalNew"],
"greedyGain": rep["greedyGainOverEven"], "blendedCacCent": rep["greedy"]["blendedCacCent"],
"lastMarginal": greedy["lastMarginal"]}, ensure_ascii=False))
return 0
if __name__ == "__main__":
raise SystemExit(main())
脚本的核心是 allocate：每一步扫一遍所有渠道，算出「如果再给这个渠道一份，能带来多少新增」，然后把这一份给最高的那个。因为有并列，规则里写死「并列时取列表靠前者」，保证结果确定。在本机跑它（假设参数），输出如下（同机相对量级，不可移植）：
{"budgetUnits": 12, "allocation": {"ua": 4, "platform": 3, "viral": 2, "content": 3}, "totalNew": 1867, "evenTotalNew": 1843, "greedyGain": 24, "blendedCacCent": 32, "lastMarginal": 97}
12 份预算的分配是 ua=4 / platform=3 / viral=2 / content=3，总新增 1867。对照的「平均分」是每渠道 3 份、总新增 1843——水填比平均分多拿 24 个新增，混合 CAC 32 分。注意 platform 只拿了 3 份：它第一份边际新增最高（300），但衰减最快（每份 ×0.6），到第 4 份就掉到 64，还不如 ua 的第 4 份（112）。「第一份最强」不等于「该投最多」——这正是边际递减的陷阱。
B 层：分配结果对假设敏感吗
但这份分配依赖一堆假设参数——baseNewPerUnit 是估的、衰减是猜的。所以必须问一句：如果某个渠道的真实效率比假设高 / 低 20%，分配会翻盘吗？下面这段是 B 层基准脚本，逐个渠道扰动 ±20%、重跑水填、比较各渠道拿到的份数。它是基准，无论多少告警退出码恒为 0。
#!/usr/bin/env python3
# file: sim/l06_budget_bench.py
# tier: B
# verify: /opt/homebrew/bin/python3 sim/l06_budget_bench.py --config config/ --out out/
"""B 基准：增长预算分配对渠道效率假设的稳健性。
分配结果依赖 baseNewPerUnit / decay 这些**假设参数**，所以要问：
「如果某渠道的真实效率比假设高 / 低 20%，分配会翻盘吗？」
做法：逐个渠道把 baseNewPerUnit 上下各扰动 20%，重跑贪心水填，比较各渠道拿到的份数。
软告警：任一渠道在 ±20% 扰动下份数变动 >= 2 份即打 WARN（说明份额对假设敏感）。
退出码恒为 0——这是基准，不是门禁。全整数，无浮点。
"""
from __future__ import annotations
import argparse
import json
import os
from l06_budget_alloc import allocate, load
SWING_WARN_UNITS = 2
PERTURB_PCT = 20
def perturbed(channels: list, target_id: str, pct: int) -> list:
"""把指定渠道的 baseNewPerUnit 乘 (100+pct)/100（整数），其余不变。"""
out = []
for c in channels:
if c["id"] == target_id:
c = dict(c, baseNewPerUnit=c["baseNewPerUnit"] * (100 + pct) // 100)
out.append(c)
return out
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)
units = cfg["dailyBudgetCent"] // cfg["unitCent"]
channels = cfg["channels"]
base_alloc = allocate(channels, units)["allocation"]
cases = []
for c in channels:
for pct in (-PERTURB_PCT, PERTURB_PCT):
alloc = allocate(perturbed(channels, c["id"], pct), units)["allocation"]
deltas = {k: alloc[k] - base_alloc[k] for k in base_alloc}
cases.append({"channel": c["id"], "pct": pct, "allocation": alloc, "deltas": deltas,
"maxAbsDelta": max(abs(v) for v in deltas.values())})
swing = {c["id"]: max(cs["maxAbsDelta"] for cs in cases if cs["channel"] == c["id"]) for c in channels}
warnings = [f"渠道 {cid} 在 ±{PERTURB_PCT}% 扰动下份数变动 {v} >= {SWING_WARN_UNITS}"
for cid, v in swing.items() if v >= SWING_WARN_UNITS]
os.makedirs(a.out, exist_ok=True)
rep = {"version": cfg["version"], "source": "sim/l06_budget_bench.py", "budgetUnits": units,
"perturbPct": PERTURB_PCT, "swingWarnUnits": SWING_WARN_UNITS, "baseAllocation": base_alloc,
"cases": cases, "swingByChannel": swing, "maxSwing": max(swing.values()) if swing else 0,
"warnings": warnings,
"verdict": "PASS(soft)" if not warnings else f"WARN(soft) {len(warnings)} 条"}
with open(os.path.join(a.out, "l06_budget_bench.json"), "w", encoding="utf-8") as f:
json.dump(rep, f, ensure_ascii=False, indent=2)
print(json.dumps({"baseAllocation": base_alloc, "swingByChannel": swing, "maxSwing": rep["maxSwing"],
"verdict": rep["verdict"], "warnings": warnings}, ensure_ascii=False))
return 0 # B 层软告警：恒返回 0
if __name__ == "__main__":
raise SystemExit(main())
在本机跑它，输出如下：
{"baseAllocation": {"ua": 4, "platform": 3, "viral": 2, "content": 3}, "swingByChannel": {"ua": 1, "platform": 1, "viral": 1, "content": 2}, "maxSwing": 2, "verdict": "WARN(soft) 1 条", "warnings": ["渠道 content 在 ±20% 扰动下份数变动 2 >= 2"]}
结果是一句软告警：content 渠道在 ±20% 扰动下份数变动 2 份，触及告警线；其余三个渠道都只变动 1 份。这说明分配整体是稳健的——前三个渠道的份额几乎不受假设影响，只有垫底的 content 会在 2~3 份之间摇摆。这符合直觉：垫底渠道的边际新增本来就和相邻渠道咬得很紧，参数一动就换位。运营含义很直接：预算分配的大头可以信，尾巴不必纠结。
小结与下一步
这一节把增长预算分配做成了一个可跑的水填模型：边际收益递减 → 每份投给最高边际 → 贪心拿到的总新增高于平均分；再用 B 基准验证分配对假设的敏感性，确认大头稳健、尾巴可动。
到这里，第 6 章把「增长」这条线走完了：自然增长与裂变（6.1）、平台流量（6.2）、预算分配（6.3）。但增长只解决「人怎么进来」；人进来之后能不能留下，是另一条完全不同的曲线。下一章讲留存与漏斗：从 D1 到 D30 的留存结构、流失预警与召回，以及留存与付费的联合建模。
阅读导航：上一节：平台推荐与流量获取 · 下一节：D1 / D7 / D30 留存漏斗 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。