2.3 止损线与关停规则

放量模型默认了「这个渠道值得继续加」,但现实里很多渠道从第一天就不该继续投。本节把「累计 ROAS 连续低于止损线 N 天就关停」写成一条可跑的整数规则,用 Python 仿真对照优质、平庸、劣质三种流量下规则省下多少亏损,再扫止损线×连续天数的网格并做 LTV 扰动基准,找出那条既不误杀好渠道、又不放跑坏渠道的线。

站内《小游戏开发实战》卷三目录 里讲过 LTV 回本与增长预算分配(那 51 篇的框架),本节不重做回本模型,只做一件更硬的事:给「什么时候关停」定一条可执行的线。上一节的爬坡模型默认了「这个渠道值得继续加」,但现实里很多渠道从第一天起就在亏——运营与团队那批文章 里反复出现的教训是「舍不得砍」,结果把一个本来只亏几万的小口子拖成几十万的大坑。这一节把止损写成规则,让关停不再靠意志力。

止损到底在止什么

流量情形累计 ROAS 走势该不该关规则若失效的后果
优质始终高于回本线不关,继续加误杀好渠道,白丢利润
平庸缓慢下穿回本线中后期关停拖太久,小亏变大亏
劣质开局就低于回本线尽快关停每多投一天都在放大损失

关停规则公式

cpiFen(d)  = baseCpiFen * (1000 + driftPermillePerDay * d) // 1000   # 成本随天数膨胀(素材疲劳)
roas(d)    = (ltvD30DeciFen // 10) * 1000 // cpiFen(d)                # 当日 ROAS 千分比
cumRoas(d) = cumRevenue(d) * 1000 // cumSpend(d)                      # 累计 ROAS 千分比
关停        = cumRoas(d) < stopFloorPermille 连续 consecutiveDays 天

stopFloorPermille 是止损线(1000 = 回本线),consecutiveDays 是耐心天数。连续计数而不是单日判定,是为了滤掉单日噪音——这也是它比「看一天关一天」更稳的原因。

S 层仿真:规则省下多少亏损

下面这段是 S 层仿真脚本 sim/o02_stoploss.py。它读 config/o02_stoploss.json 的三个流量场景,逐日推进成本漂移与累计 ROAS,对照「不设规则(跑满 21 天)」与「设规则(触发即关停)」的净盈亏,再扫一遍止损线 × 连续天数网格,产出 JSON + CSV + SVG。

#!/usr/bin/env python3
# file: sim/o02_stoploss.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/o02_stoploss.py --config config/ --out out/
"""止损线与关停规则仿真:累计 ROAS 连续低于止损线 N 天即关停,量化「少亏多少」。

    cpiFen(d)  = baseCpiFen * (1000 + driftPermillePerDay * d) // 1000   # 成本随天数膨胀(素材疲劳)
    cpiFen(d)  = cpiFen(d) * (1000 + jitter(d)) // 1000                   # jitter ∈ [-noise, +noise]
    roas(d)    = (ltvD30DeciFen // 10) * 1000 // cpiFen(d)                # 当日 ROAS 千分比
    cumRoas(d) = cumRevenue(d) * 1000 // cumSpend(d)
    关停        = cumRoas(d) < stopFloorPermille 连续 consecutiveDays 天

对照「不设规则(跑满)」与「设规则(触发即关停)」的净盈亏,再扫 stopFloorPermille ×
consecutiveDays 网格看规则松紧的取舍。全程整数、地板除,无浮点。产出 out/o02_stoploss.json
+ .csv + .svg。真实平台关停与重跑行为需真实业务环境,命令行不可复现(X)。
"""

from __future__ import annotations

import argparse
import csv
import json
import os
import xml.etree.ElementTree as ET


def lcg_next(s: int, mod: int, mul: int) -> int:
    return (s * mul) % mod


def run_scenario(cfg: dict, sc: dict, floor: int, k: int) -> dict:
    """按给定止损线 floor 与连续天数 k 跑一个场景,返回关停日与净盈亏。"""
    mod, mul, noise = cfg["lcgModulus"], cfg["lcgMultiplier"], cfg["noisePermille"]
    per_install = cfg["ltvD30DeciFen"] // 10
    s = sc["seed"]
    cum_spend = cum_rev = 0
    below = 0
    stop_day = -1
    series = []
    for d in range(1, cfg["days"] + 1):
        cpi = sc["baseCpiFen"] * (1000 + sc["driftPermillePerDay"] * d) // 1000
        s = lcg_next(s, mod, mul)
        jitter = s % (2 * noise + 1) - noise
        cpi = cpi * (1000 + jitter) // 1000
        roas = per_install * 1000 // cpi
        cum_spend += cfg["dailyBudgetFen"]
        cum_rev += cfg["dailyBudgetFen"] * roas // 1000
        cum_roas = cum_rev * 1000 // cum_spend
        below = below + 1 if cum_roas < floor else 0
        series.append(dict(day=d, cpiFen=cpi, roasPermille=roas, cumRoasPermille=cum_roas))
        if below >= k:
            stop_day = d
            break
    return dict(scenario=sc["id"], name=sc["name"], floorPermille=floor, consecutiveDays=k,
                stopDay=stop_day, spentFen=cum_spend, revenueFen=cum_rev,
                netFen=cum_rev - cum_spend, series=series)


def sweep(cfg: dict) -> list[dict]:
    grid = []
    for floor in cfg["thresholdGrid"]:
        for k in cfg["consecutiveGrid"]:
            runs = [run_scenario(cfg, sc, floor, k) for sc in cfg["scenarios"]]
            grid.append(dict(floorPermille=floor, consecutiveDays=k,
                             totalNetFen=sum(r["netFen"] for r in runs),
                             stopDays={r["scenario"]: r["stopDay"] for r in runs}))
    return grid


def draw_svg(path: str, rows: list[dict]) -> int:
    """纯 Python SVG:每个场景两根横条——不设规则 vs 设规则(默认止损线)的净盈亏。"""
    W, H, PAD, ROW = 760, 70 + 60 * len(rows), 190, 60
    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": "28", "font-size": "15", "fill": "#111827"}
                  ).text = "《合成小岛》不设规则 vs 设止损规则的净盈亏(分,越大越好)"
    span = W - PAD - 24
    lo = min([min(r["netNoRuleFen"], r["netRuleFen"]) for r in rows])
    hi = max([max(r["netNoRuleFen"], r["netRuleFen"]) for r in rows] + [1])
    zero = PAD + span * (0 - lo) // (hi - lo)
    ET.SubElement(root, "line", {"x1": str(zero), "y1": "44", "x2": str(zero), "y2": str(H - 16),
                                 "stroke": "#9ca3af"})
    for i, r in enumerate(rows):
        y = 60 + i * ROW
        ET.SubElement(root, "text", {"x": "16", "y": str(y + 22), "font-size": "12",
                                     "fill": "#374151"}).text = r["name"]
        for j, (key, color, label) in enumerate([("netNoRuleFen", "#fca5a5", "不设规则"),
                                                 ("netRuleFen", "#34d399", "设规则")]):
            val = r[key]
            x0 = PAD + span * (min(0, val) - lo) // (hi - lo)
            w = span * abs(val) // (hi - lo)
            ET.SubElement(root, "rect", {"x": str(x0), "y": str(y + 6 + j * 22), "width": str(w),
                                         "height": "18", "rx": "3", "fill": color})
            ET.SubElement(root, "text", {"x": str(PAD + span + 4), "y": str(y + 20 + j * 22),
                                         "font-size": "10", "fill": "#6b7280"}).text = (
                f'{label} {val}')
    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 main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--config", default="config/")
    ap.add_argument("--out", default="out/")
    args = ap.parse_args()
    with open(os.path.join(args.config, "o02_stoploss.json"), encoding="utf-8") as f:
        cfg = json.load(f)

    floor, k = cfg["stopFloorPermille"], cfg["consecutiveDays"]
    rows = []
    for sc in cfg["scenarios"]:
        no_rule = run_scenario(cfg, sc, 0, cfg["days"] + 1)  # floor=0 永不触发 → 跑满
        rule = run_scenario(cfg, sc, floor, k)
        rows.append(dict(id=sc["id"], name=sc["name"], netNoRuleFen=no_rule["netFen"],
                         netRuleFen=rule["netFen"], stopDay=rule["stopDay"],
                         savedFen=rule["netFen"] - no_rule["netFen"],
                         spentRuleFen=rule["spentFen"], spentNoRuleFen=no_rule["spentFen"]))
    grid = sweep(cfg)
    best = max(grid, key=lambda g: g["totalNetFen"])
    digest = ",".join(f'{r["id"]}:{r["netNoRuleFen"]}:{r["netRuleFen"]}:{r["stopDay"]}'
                      f':{r["savedFen"]}' for r in rows)

    os.makedirs(args.out, exist_ok=True)
    svg_bytes = draw_svg(os.path.join(args.out, "o02_stoploss.svg"), rows)
    report = {"version": cfg["version"], "game": cfg["game"], "observedAt": cfg["observedAt"],
              "stopFloorPermille": floor, "consecutiveDays": k, "days": cfg["days"],
              "scenarios": rows, "grid": grid, "bestGrid": best, "digest": digest,
              "svg": os.path.join(args.out, "o02_stoploss.svg"), "svgBytes": svg_bytes}
    with open(os.path.join(args.out, "o02_stoploss.json"), "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False, indent=2)
    with open(os.path.join(args.out, "o02_stoploss.csv"), "w", encoding="utf-8", newline="") as f:
        w = csv.writer(f)
        cols = ("id", "name", "netNoRuleFen", "netRuleFen", "stopDay", "savedFen",
                "spentRuleFen", "spentNoRuleFen")
        w.writerow(cols)
        for r in rows:
            w.writerow([r[c] for c in cols])

    print(json.dumps({"scenarios": [{"id": r["id"], "netNoRuleFen": r["netNoRuleFen"],
                                     "netRuleFen": r["netRuleFen"], "stopDay": r["stopDay"],
                                     "savedFen": r["savedFen"]} for r in rows],
                      "digest": digest, "bestGrid": best}, ensure_ascii=False, indent=2))
    return 0


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

参数放在 config/o02_stoploss.json(默认止损线 1000、连续 3 天;网格是 B 层要扫的止损线 × 天数)。同样,这些是「公开可观察机制」口径的建模假设(观察时点 2026-10):

{
  "version": 1,
  "game": "merge-island",
  "observedAt": "2026-10",
  "dailyBudgetFen": 100000,
  "days": 21,
  "ltvD30DeciFen": 3800,
  "stopFloorPermille": 1000,
  "consecutiveDays": 3,
  "noisePermille": 40,
  "lcgModulus": 2147483647,
  "lcgMultiplier": 48271,
  "thresholdGrid": [800, 900, 1000, 1100, 1200],
  "consecutiveGrid": [1, 2, 3, 5],
  "note": "止损线与关停规则参数,口径是「公开可观察机制」的建模假设。dailyBudgetFen=每日固定投放;ltvD30DeciFen=单安装 D30 收入(0.1 分);stopFloorPermille=累计 ROAS 止损线(千分比);consecutiveDays=连续低于止损线多少天才关停;baseCpiFen=起始单安装成本;driftPermillePerDay=单安装成本日漂移(正=成本膨胀/素材疲劳);noisePermille=日噪音;seed 驱动自实现整数 LCG。thresholdGrid/consecutiveGrid 是 B 层稳健性扫描网格。真实平台关停与重跑行为需真实业务环境,命令行不可复现(X)。全整数,无浮点。",
  "scenarios": [
    { "id": "good", "name": "优质流量", "baseCpiFen": 300, "driftPermillePerDay": 5, "seed": 6001 },
    { "id": "flat", "name": "平庸流量", "baseCpiFen": 340, "driftPermillePerDay": 15, "seed": 6002 },
    { "id": "bad", "name": "劣质流量", "baseCpiFen": 380, "driftPermillePerDay": 30, "seed": 6003 }
  ]
}

在本机跑 /opt/homebrew/bin/python3 sim/o02_stoploss.py --config config/ --out out/,默认规则(1000 / 连续 3 天)的效果(同机相对量级,绝对数值不可移植):

流量场景不设规则净盈亏(分)设规则净盈亏(分)关停日规则省下(分)
优质流量438400438400不触发0
平庸流量-61200-35600D1925600
劣质流量-496100-14600D3481500

规则的表现完全符合预期:优质流量一次都没触发(累计 ROAS 从不下穿 1000),白丢 0 分;劣质流量在 D3 就被砍掉,把一次要亏 49.6 万分的投放压到只亏 1.46 万分,单场景省下 481500 分。平庸流量 D19 才关停,省得少(25600 分)——因为它的累计 ROAS 是慢慢下穿的,规则已经尽力。

再扫一遍网格,看止损线该设在哪(表里是三个场景的合计净盈亏,单位分):

止损线‰ \ 连续天数135
80027300-39800-118900
900285200237000185500
1000426800388200339900
1100465700469300449200
1200446100453500446600

最优是 1100 / 连续 3 天(合计 469300 分),比默认的 1000 / 3 天(388200 分)多 81100 分。两个反直觉点值得记:止损线设太松(800)几乎等于不设——劣质流量拖到 D19 才关,合计只剩 27300 分甚至转负;设太紧(1200)反而比 1100 差,因为平庸流量前几天还在赚钱,第 3 天就砍掉会少赚。止损线不是越紧越好,是「刚好在坏渠道的赚钱期结束后落下」最好。

B 层基准:止损线对收入假设稳不稳

上面选出的 1100 依赖 ltvD30DeciFen 这个收入假设,而它最容易被真实数据推翻。下面这段是 B 层基准脚本 sim/o02_stoploss_bench.py——它复用 S 层的 sweep(),只把 LTV 上下扰动 ±20%,看最优止损线是否漂移。B 层软告警不硬失败。

#!/usr/bin/env python3
# file: sim/o02_stoploss_bench.py
# tier: B
# verify: /opt/homebrew/bin/python3 sim/o02_stoploss_bench.py --config config/ --out out/
"""止损规则稳健性基准(B):把 LTV 假设上下扰动 ±20%,看「最优止损线」是否漂移。

复用 sim/o02_stoploss.py 的 sweep(),只换 ltvD30DeciFen。软告警不硬失败:最优止损线在
三个 LTV 档位下若不一致,说明该止损线对收入假设敏感,规则应留缓冲。产出 out/o02_stoploss_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("o02_stoploss", 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/")
    args = ap.parse_args()
    with open(os.path.join(args.config, "o02_stoploss.json"), encoding="utf-8") as f:
        base = json.load(f)
    sim = load_sim(os.path.join(os.path.dirname(os.path.abspath(__file__)), "o02_stoploss.py"))

    rows = []
    for sc in (800, 1000, 1200):  # LTV 千分比:-20% / 基准 / +20%
        cfg = json.loads(json.dumps(base))
        cfg["ltvD30DeciFen"] = base["ltvD30DeciFen"] * sc // 1000
        best = max(sim.sweep(cfg), key=lambda g: g["totalNetFen"])
        rows.append(dict(ltvScalePermille=sc, bestFloor=best["floorPermille"],
                         bestK=best["consecutiveDays"], totalNetFen=best["totalNetFen"]))

    floors = sorted({r["bestFloor"] for r in rows})
    stable = len(floors) == 1
    report = dict(version=base["version"], game=base["game"], ltvBase=base["ltvD30DeciFen"],
                  defaultFloor=base["stopFloorPermille"], rows=rows, distinctBestFloors=floors,
                  stable=stable, note="软告警:最优止损线跨 LTV 档位一致才算稳健。")
    os.makedirs(args.out, exist_ok=True)
    with open(os.path.join(args.out, "o02_stoploss_bench.json"), "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False, indent=2)
    if not stable:
        print(f"WARN 最优止损线跨 LTV 档位漂移:{floors}(软告警,不硬失败)")
    print(json.dumps({"rows": rows, "distinctBestFloors": floors, "stable": stable},
                     ensure_ascii=False, indent=2))
    return 0


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

跑 /opt/homebrew/bin/python3 sim/o02_stoploss_bench.py --config config/ --out out/,三个 LTV 档位(-20% / 基准 / +20%)下最优止损线都落在 1100,连续天数分别取 1 / 3 / 1,合计净盈亏 -33200 / 469300 / 1350900 分。

最优止损线在三个 LTV 档位下都稳定在 1100(stable: true),说明这条线不依赖收入假设的精确值——这是它可用的前提。注意连续天数会从 3 漂到 1:当收入整体下移 20%,合计净盈亏转负(-33200 分),此时「早关」比「多等两天」更划算。所以稳健的纪律是止损线钉在 1100,连续天数留 1~3 的弹性。

X 层:真实关停行为不可复现

上面算的是模型侧。下面这些属于 X 层——需真实业务环境,命令行不可复现:

  • 平台真实的「关停」语义(暂停计划、删除计划、还是仅降预算,对后续重跑的影响不同);
  • 真实成本漂移的形状(driftPermillePerDay 的线性假设是简化,真实素材疲劳可能前慢后快)。

正确的用法是把规则当纪律:模型定出「累计 ROAS 连续低于 1100‰ 三天就关」这条线,投放后用真实数据回看——如果线上关掉的渠道后来自己回本了,说明线太紧;如果没关的渠道越亏越多,说明太松。规则是可调的,情绪不是。

自检表

检查项合格标准
判据稳定用累计 ROAS 而非单日,滤掉噪音
有耐心参数连续天数可调,不靠单日触发
不误杀优质流量全程不触发
能止损劣质流量在开局几天内关停
线可标定止损线对收入假设稳健(B 层验证)
边界清晰真实关停语义标 X,不写成「实测」

到这里,出价方式(2.1)、放量节奏(2.2)、止损规则(2.3)三件事连成了一条完整的投放决策链:选对出价、按安全步长爬、越线就关。但三者都建立在一个前提上——你的素材能撑住 CTR 与留存。下一章处理素材的工业化产出与优胜劣汰。

阅读导航:上一节:2.2 放量节奏与预算爬坡 · 下一节:3.1 素材生产线与批量产出 。

继续阅读

探索更多技术文章

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

全部文章 返回首页

「minigame」更多文章

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