卷三「买量渠道与投放机制」 与「素材设计与创意迭代」 做的是渠道机制与素材设计框架(国内口径):前者算单渠道从曝光到回本整条 ROI 漏斗,后者讲素材怎么设计、怎么迭代。本节不重做那条漏斗、也不重做素材设计方法论,而是把视角抬到组合:五个「地区 × 渠道」单元同时投,钱该怎么分、哪个单元该关、什么时候关。站内那篇独立开发者买量失败复盘 讲的是「钱花出去了、留存没起来」的事后教训;本节给你一台能算止损省了多少钱的机器。
止损不是「关掉」,是「按规则关掉」
「投不动就停」听起来简单,做起来难在两件事:停早了会把学习期的正常波动当成失败(新计划前几天成本天然虚高,停早了永远起不了量);停晚了会让一个持续失血的单元把好单元赚的钱吃光。所以止损必须是一条可复现的规则,而不是运营拍脑袋。本节用的规则是三段式:学习期不判(前 warmupDays 天不看)、单日设下限(日 ROAS 低于 minRoasPermille 记一次)、连续触发才关(连续 consecutiveDays 天低于下限,从次日停)。
一个单元一天怎么算
每个单元一天用整数运算:ramp(d) = d < warmup ? rampPenalty × (warmup − d) ÷ warmup : 0 是学习期折损(新计划第一天 ROAS 最低,随 warmup 线性恢复到 baseRoasPermille);obs(d) = max(0, baseRoasPermille − ramp(d) + noise(d)),其中 noise 是可播种噪声——用配置里的 seed 驱动 PCG32,所以同一份配置两次运行的日序列逐位相同;当日回收 rev(d) = dailySpendFen × obs(d) ÷ 1000。关键点:「止损」与「不止损」必须跑在同一条日序列上,否则比的不是策略,是两次不同的随机数。
S 层:日粒度 ROI 与止损仿真
下面这段是 S 层仿真脚本 sim/g13_roi_stop_loss.py。它读 config/g13_roi_stop_loss.json,为五个单元各生成 30 天日 ROAS,在同一条序列上跑两种策略,产出 JSON + CSV + SVG:
#!/usr/bin/env python3
# file: sim/g13_roi_stop_loss.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/g13_roi_stop_loss.py --config config/ --out out/
"""海外投放 ROI 与止损仿真(S):多单元组合的日粒度收益与关停规则。
每个「地区 × 渠道」单元生成 horizonDays 天日 ROAS(学习期折损 + 可播种 PCG32 噪声),在同一条序列上跑
「止损」与「不止损」两策略(学习期后连续 consecutiveDays 天 obs < minRoasPermille 则次日停)。全整数、
地板除、无浮点、无 Math.random,产出 json + csv + svg;真实 ROAS 与真实关停决策需真实业务环境(X)。
"""
from __future__ import annotations
import argparse
import csv
import json
import os
import xml.etree.ElementTree as ET
MULT = 6364136223846793005
MASK64 = (1 << 64) - 1
MASK32 = (1 << 32) - 1
class Pcg32:
"""与 core/ 侧逐位等价的可播种 PCG32(确定性、跨语言一致)。"""
def __init__(self, seed: int, seq: int = 54) -> None:
self.state = 0
self.inc = ((seq << 1) | 1) & MASK64
self._u32()
self.state = (self.state + seed) & MASK64
self._u32()
def _u32(self) -> int:
old = self.state
self.state = (old * MULT + self.inc) & MASK64
x = (((old >> 18) ^ old) >> 27) & MASK32
rot = old >> 59
return ((x >> rot) | (x << ((32 - rot) & 31))) & MASK32
def span(self, lo: int, hi: int) -> int:
return lo + self._u32() % (hi - lo + 1)
def series_of(c: dict, horizon: int, warmup: int, rng: Pcg32) -> list[int]:
"""一个单元的 horizon 天日 ROAS 序列(千分比);学习期折损随 warmup 线性衰减到 0。"""
out = []
for d in range(horizon):
ramp = c["rampPenaltyPermille"] * (warmup - d) // warmup if d < warmup else 0
vol = c["volatilityPermille"]
out.append(max(0, c["baseRoasPermille"] - ramp + rng.span(-vol, vol)))
return out
def simulate(c: dict, series: list[int], sl: dict) -> dict:
"""在同一条日 ROAS 序列上跑「止损」与「不止损」两种策略。"""
spend, warmup = c["dailySpendFen"], sl["warmupDays"]
floor, need = sl["minRoasPermille"], sl["consecutiveDays"]
rev_all, spent_all = sum(spend * r // 1000 for r in series), spend * len(series)
stopped, spent, rev = -1, 0, 0
for d, r in enumerate(series):
spent += spend
rev += spend * r // 1000
if d >= warmup - 1 and d + 1 >= need and all(x < floor for x in series[d + 1 - need:d + 1]):
stopped = d + 1
break
return {"stoppedDay": stopped, "spentWithFen": spent, "revenueWithFen": rev,
"profitWithFen": rev - spent, "spentWithoutFen": spent_all, "revenueWithoutFen": rev_all,
"profitWithoutFen": rev_all - spent_all,
"deltaProfitFen": (rev - spent) - (rev_all - spent_all)}
def draw_svg(path: str, rows: list[dict]) -> int:
"""纯 Python SVG:每单元两根横条——不止损利润(灰)与止损后利润(蓝)。"""
W, ROW, LEFT, SPAN = 840, 56, 150, 560
H = 76 + ROW * len(rows)
root = ET.Element("svg", {"xmlns": "http://www.w3.org/2000/svg", "width": str(W),
"height": str(H), "viewBox": f"0 0 {W} {H}"})
ET.SubElement(root, "rect", {"width": str(W), "height": str(H), "fill": "#ffffff"})
ET.SubElement(root, "text", {"x": "16", "y": "26", "font-size": "15", "fill": "#111827"}).text = (
"《合成小岛》海外投放:止损(蓝)与不止损(灰)的 30 天利润(分)")
hi = max([abs(r["profitWithoutFen"]) for r in rows] + [1])
bars = [("profitWithoutFen", "#cbd5e1", "不止损"), ("profitWithFen", "#2563eb", "止损")]
for i, r in enumerate(rows):
y = 52 + i * ROW
ET.SubElement(root, "text", {"x": "16", "y": str(y + 22), "font-size": "12", "fill": "#374151"}).text = r["id"]
for j, (key, color, tag) in enumerate(bars):
w = SPAN * abs(r[key]) // hi
ET.SubElement(root, "rect", {"x": str(LEFT), "y": str(y + 6 + j * 20), "width": str(w),
"height": "16", "rx": "3", "fill": color})
tail = f' · 停于 D{r["stoppedDay"]}' if (j == 1 and r["stoppedDay"] != -1) else ""
ET.SubElement(root, "text", {"x": str(LEFT + w + 6), "y": str(y + 18 + j * 20), "font-size": "11",
"fill": "#6b7280"}).text = f"{tag} {r[key]}{tail}"
os.makedirs(os.path.dirname(path), exist_ok=True)
ET.ElementTree(root).write(path, encoding="utf-8", xml_declaration=True)
return len(ET.tostring(root))
def run(cfg: dict) -> list[dict]:
"""按配置顺序消费同一个 PCG32 流,生成全部单元结果(供 S 与 B 复用)。"""
rng = Pcg32(cfg["seed"])
sl = cfg["stopLoss"]
return [dict(c, **simulate(c, series_of(c, cfg["horizonDays"], sl["warmupDays"], rng), sl))
for c in cfg["campaigns"]]
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--config", default="config/")
ap.add_argument("--out", default="out/")
a = ap.parse_args()
with open(os.path.join(a.config, "g13_roi_stop_loss.json"), encoding="utf-8") as f:
cfg = json.load(f)
rows = run(cfg)
stopped = [r["id"] for r in rows if r["stoppedDay"] != -1]
tot_with = sum(r["profitWithFen"] for r in rows)
tot_without = sum(r["profitWithoutFen"] for r in rows)
parts = ";".join(f'{r["id"]}:{r["stoppedDay"]}:{r["profitWithFen"]}:{r["profitWithoutFen"]}' for r in rows)
digest = (f'campaigns={len(rows)},stopped={len(stopped)},profitWithFen={tot_with},'
f'profitWithoutFen={tot_without},deltaFen={tot_with - tot_without};{parts}')
os.makedirs(a.out, exist_ok=True)
svg_bytes = draw_svg(os.path.join(a.out, "g13_roi_stop_loss.svg"), rows)
cols = ("id", "region", "channel", "dailySpendFen", "baseRoasPermille", "stoppedDay",
"spentWithFen", "revenueWithFen", "profitWithFen", "spentWithoutFen",
"revenueWithoutFen", "profitWithoutFen", "deltaProfitFen")
with open(os.path.join(a.out, "g13_roi_stop_loss.csv"), "w", encoding="utf-8", newline="") as f:
w = csv.writer(f)
w.writerow(cols)
for r in rows:
w.writerow([r[k] for k in cols])
rep = {"version": cfg["version"], "game": cfg["game"], "observedAt": cfg["observedAt"],
"horizonDays": cfg["horizonDays"], "stopLoss": cfg["stopLoss"], "seed": cfg["seed"],
"stopped": stopped, "totalProfitWithFen": tot_with, "totalProfitWithoutFen": tot_without,
"campaigns": rows, "digest": digest,
"svg": os.path.join(a.out, "g13_roi_stop_loss.svg"), "svgBytes": svg_bytes}
with open(os.path.join(a.out, "g13_roi_stop_loss.json"), "w", encoding="utf-8") as f:
json.dump(rep, f, ensure_ascii=False, indent=2)
print(json.dumps({"digest": digest, "stopped": stopped,
"campaigns": [{k: r[k] for k in ("id", "baseRoasPermille", "stoppedDay",
"profitWithFen", "profitWithoutFen")}
for r in rows]}, ensure_ascii=False, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())
配置放在 config/g13_roi_stop_loss.json。这些数字是「公开可观察机制」口径的建模假设,不是任何投放后台实测:
{
"version": 1,
"game": "merge-island",
"observedAt": "2026-10",
"seed": 20261013,
"horizonDays": 30,
"stopLoss": { "warmupDays": 7, "minRoasPermille": 700, "consecutiveDays": 2 },
"note": "海外投放 ROI 与止损配置。口径是「公开可观察机制」的建模假设(观察时点 2026-10),不是任一投放后台实测。campaigns[] 是一个「地区 × 渠道」投放单元:dailySpendFen=日预算(分);baseRoasPermille=长期日 ROAS 假设(千分比,1000=当天回本);rampPenaltyPermille=学习期首日的 ROAS 折损,随 warmupDays 线性衰减到 0;volatilityPermille=日 ROAS 波动幅度(可播种噪声的上下界)。stopLoss:warmupDays=学习期天数(期内不触发止损);minRoasPermille=单日 ROAS 下限;consecutiveDays=连续低于下限多少天触发关停。真实 ROAS 与真实关停决策需真实业务环境,命令行不可复现(X)。全整数,无浮点。",
"campaigns": [
{ "id": "us_meta", "region": "US", "channel": "meta_feed", "dailySpendFen": 100000, "baseRoasPermille": 1150, "rampPenaltyPermille": 400, "volatilityPermille": 180 },
{ "id": "jp_apple", "region": "JP", "channel": "apple_search", "dailySpendFen": 60000, "baseRoasPermille": 1300, "rampPenaltyPermille": 250, "volatilityPermille": 150 },
{ "id": "de_tiktok", "region": "DE", "channel": "tiktok_feed", "dailySpendFen": 80000, "baseRoasPermille": 800, "rampPenaltyPermille": 300, "volatilityPermille": 260 },
{ "id": "br_adnet", "region": "BR", "channel": "adnet_cpi", "dailySpendFen": 40000, "baseRoasPermille": 520, "rampPenaltyPermille": 300, "volatilityPermille": 260 },
{ "id": "kr_kol", "region": "KR", "channel": "kol_cpc", "dailySpendFen": 50000, "baseRoasPermille": 980, "rampPenaltyPermille": 200, "volatilityPermille": 100 }
]
}
在本机跑 /opt/homebrew/bin/python3 sim/g13_roi_stop_loss.py --config config/ --out out/,五个单元的结果(同机相对量级,绝对数值不可移植):
| 单元 | 地区·渠道 | 日预算(分) | 基准 ROAS‰ | 关停日 | 止损利润(分) | 不止损利润(分) |
|---|---|---|---|---|---|---|
| us_meta | US·Meta | 100000 | 1150 | 未关 | 277900 | 277900 |
| jp_apple | JP·Apple | 60000 | 1300 | 未关 | 477120 | 477120 |
| de_tiktok | DE·TikTok | 80000 | 800 | D8 | -217680 | -660880 |
| br_adnet | BR·网盟 | 40000 | 520 | D7 | -178040 | -609360 |
| kr_kol | KR·达人 | 50000 | 980 | 未关 | -86250 | -86250 |
摘要行:
digest = campaigns=5,stopped=2,profitWithFen=273050,profitWithoutFen=-601470,deltaFen=874520;us_meta:-1:277900:277900;jp_apple:-1:477120:477120;de_tiktok:8:-217680:-660880;br_adnet:7:-178040:-609360;kr_kol:-1:-86250:-86250
这张表有两个关键读法。第一,止损把组合从亏 601470 分拉到赚 273050 分(差额 874520 分)——省下来的钱全部来自 de_tiktok 与 br_adnet 两个被关停的单元,它们一个基准 ROAS 只有 800‰、一个只有 520‰,都低于回本线 1000‰。第二,kr_kol 没被关停,却一直在小幅失血(-86250 分):它的基准 ROAS 是 980‰,只比回本线低 20‰,日 ROAS 从没跌破下限 700‰,所以规则抓不到它——这是**「慢性失血」型单元**,规则越松(下限越低)越抓不到。
B 层基准:扰动阈值,看「该关谁」稳不稳
上面那张表依赖一组阈值假设(minRoasPermille=700、warmupDays=7、consecutiveDays=2)。假设是拍的,所以必须问:换个阈值,「该关停谁」会不会变? 下面这段是 B 层基准脚本 sim/g13_roi_stop_loss_bench.py,它复用 S 层的 run(),对四类假设各做多档扰动,重算关停集合:
#!/usr/bin/env python3
# file: sim/g13_roi_stop_loss_bench.py
# tier: B
# verify: /opt/homebrew/bin/python3 sim/g13_roi_stop_loss_bench.py --config config/ --out out/
"""海外投放止损规则的稳健性基准(B):扰动阈值与波动,看「关停集合」是否翻转。
复用 sim/g13_roi_stop_loss.py 的 run(),对四类假设各做多档扰动:minRoasPermille(600/700/800)、
warmupDays(5/7/9)、consecutiveDays(1/2/3)、volatilityPermille(各单元 ±50%)。每档重算一遍,
收集「被关停的单元集合」与组合总利润。软告警不硬失败:若关停集合随扰动漂移,说明「该关停谁」
的结论对阈值假设敏感,应以多档位交叉验证、并对临界单元留人工复核。产出
out/g13_roi_stop_loss_bench.json。
"""
from __future__ import annotations
import argparse
import importlib.util
import json
import os
def load_sim(path: str):
spec = importlib.util.spec_from_file_location("g13_roi_stop_loss", path)
mod = importlib.util.module_from_spec(spec)
assert spec.loader is not None
spec.loader.exec_module(mod)
return mod
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--config", default="config/")
ap.add_argument("--out", default="out/")
a = ap.parse_args()
with open(os.path.join(a.config, "g13_roi_stop_loss.json"), encoding="utf-8") as f:
base = json.load(f)
sim = load_sim(os.path.join(os.path.dirname(os.path.abspath(__file__)), "g13_roi_stop_loss.py"))
base_rows = sim.run(base)
base_stopped = sorted(r["id"] for r in base_rows if r["stoppedDay"] != -1)
def probe(label: str, mutate) -> dict:
cfg = json.loads(json.dumps(base))
mutate(cfg)
rows = sim.run(cfg)
return {"perturb": label, "stopped": sorted(r["id"] for r in rows if r["stoppedDay"] != -1),
"totalProfitWithFen": sum(r["profitWithFen"] for r in rows)}
rows = []
for v in (600, 700, 800):
rows.append(probe(f"minRoas={v}", lambda c, v=v: c["stopLoss"].update(minRoasPermille=v)))
for v in (5, 7, 9):
rows.append(probe(f"warmup={v}", lambda c, v=v: c["stopLoss"].update(warmupDays=v)))
for v in (1, 2, 3):
rows.append(probe(f"consecutive={v}", lambda c, v=v: c["stopLoss"].update(consecutiveDays=v)))
for v in (500, 1500):
rows.append(probe(f"volatility×{v // 1000}", lambda c, v=v: [
camp.update(volatilityPermille=camp["volatilityPermille"] * v // 1000) for camp in c["campaigns"]]))
sets = sorted({tuple(r["stopped"]) for r in rows})
stable = len(sets) == 1
report = {"version": base["version"], "game": base["game"], "baseStopped": base_stopped,
"baseTotalProfitWithFen": sum(r["profitWithFen"] for r in base_rows),
"rows": rows, "distinctStopped": [list(s) for s in sets], "stable": stable,
"note": "软告警:关停集合跨全部阈值 / 波动扰动均不变,止损结论才稳健。"}
os.makedirs(a.out, exist_ok=True)
with open(os.path.join(a.out, "g13_roi_stop_loss_bench.json"), "w", encoding="utf-8") as f:
json.dump(report, f, ensure_ascii=False, indent=2)
if not stable:
print(f"WARN 关停集合随扰动漂移:{len(sets)} 种不同结果(软告警,不硬失败)")
print(json.dumps({"baseStopped": base_stopped, "stable": stable,
"distinctStopped": [list(s) for s in sets],
"changedRows": [{"perturb": r["perturb"], "stopped": r["stopped"]}
for r in rows if sorted(r["stopped"]) != base_stopped]},
ensure_ascii=False, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())
在本机跑 /opt/homebrew/bin/python3 sim/g13_roi_stop_loss_bench.py --config config/ --out out/,结果是软告警(B 层不硬失败):
WARN 关停集合随扰动漂移:2 种不同结果(软告警,不硬失败)
{
"baseStopped": ["br_adnet", "de_tiktok"],
"stable": false,
"distinctStopped": [["br_adnet"], ["br_adnet", "de_tiktok"]],
"changedRows": [{"perturb": "minRoas=600", "stopped": ["br_adnet"]}]
}
两种关停集合分别是基准的 {br_adnet, de_tiktok} 与只关 {br_adnet}。翻转只发生在一处:把单日下限从 700 降到 600,de_tiktok 就逃脱了——它的基准 ROAS 是 800‰,只比下限高 100‰,下限一松它就够不着规则了。所以正确的读法不是「基准关了 de_tiktok 就照关」,而是:br_adnet 是稳的(任何扰动都关),de_tiktok 卡在下限线上、关不关取决于你把线画在哪儿——这种临界单元应该按实时 ROAS 再拍一次,而不是套死一个阈值。B 层的意义正是把这个摇摆显式暴露。
X 层:真实 ROAS 与真实关停决策不可复现
上面算的是模型侧:给定基准 ROAS 与噪声,止损省下多少钱。而下面这些属于 X 层,需真实业务环境,命令行不可复现:
- 真实的日 ROAS(受竞价、人群、素材衰减共同影响,平台侧不透明);
- 真实的关停决策(关停会重置计划的学习期,实际执行还要考虑账户结构、返点、渠道关系);
- 真实的结算口径(本节把「ROAS」当作已对齐口径的回收比,真实回收还受归因窗口与结算周期影响,见「无 IDFA 时代的归因替代」 )。
工程上正确的做法:把可复现的规则层与不可复现的投放层分开——规则层固定「学习期 + 单日下限 + 连续触发」这套止损算法,投放层在拿到真实日 ROAS 后反标定(把实测基准与波动填回 config,重跑立刻看到新的关停集合与盈亏)。
自检表
| 检查项 | 合格标准 |
|---|---|
| 规则可复现 | 止损是「学习期 + 下限 + 连续触发」的显式规则,非拍脑袋 |
| 同序列对照 | 止损与不止损跑在同一条日序列上,可比 |
| 学习期保护 | 学习期内不触发止损,避免误杀正常波动 |
| 慢性失血可见 | 指出「没跌破下限但持续小亏」的单元,规则抓不到 |
| 稳健性显式 | 用 B 层扰动阈值,暴露临界单元的关停摇摆 |
| 边界清晰 | 真实 ROAS 与关停决策标 X,不写成「实测」 |
止损最贵的一课是:它省的不是「一笔钱」,是「一个不肯认错的决策」。到这里,第 13 章的海外买量与素材就闭环了:先把渠道口径拉平(13.1),再把素材合规判定清楚(13.2),最后用可跑的模型决定投多少、什么时候停(13.3)。下一章把视角从「投放」转到「市场」——海外各平台的榜单结构,以及竞品在本地化上的机会。
阅读导航:上一节:13.2 素材本地化与合规 · 下一节:14.1 各平台榜单结构 。
继续阅读
探索更多技术文章
浏览归档,发现更多关于系统设计、工具链和工程实践的内容。