3.3 素材衰减与疲劳替换

素材不是上线就一直好用,跑得越久 CTR 越低。本节把疲劳写成整数幂 ctrAt(d)=baseCtr×decayNum^d//decayDen^d,用阈值反推替换日,再拿 no-replace / fixed-K / threshold 三种策略对照 CPI,并用 B 层扫描验证最优周期是否稳健。

上一节判定完谁赢,素材就进入投放;但同一条素材在账户里待得越久,点击率越低——这是疲劳(fatigue),是默认状态而非意外。素材生命周期「创意侧」怎么迭代,卷三 5.2 素材设计与迭代 与 5.3 素材与玩法一致性 已经讲透,卷三目录 《上线变现与长线运营》 给了完整脉络。本节不重做创意设计,只做衰减的工程化:把疲劳写成一条可算的整数曲线,用它反推替换日,再比较几种替换策略的单位成本。

衰减是默认状态,不是意外

素材不会「用坏了才衰减」,而是从第一天起就在衰减:对新素材敏感的那部分人群被反复触达后消耗殆尽,剩下的点击来自转化意愿更低的长尾,可观察的规律是前几天下滑最快、之后趋缓。运营上要回答的只有一个问题——什么时候把它换掉。换早了浪费制作与测试成本,换晚了持续买贵量。

衰减曲线:一条整数幂

把「第 d 日的 CTR」写成一条整数幂,衰减率用整数比表示(config 里禁止浮点):

ctrAt(d)    = baseCtr × decayNum^d // decayDen^d                        # 千分比
replaceDay  = 首个满足 ctrAt(d) < baseCtr × thresholdPermille // 1000 的 d

decayNum/decayDen = 970/1000 表示每天保留 97% 的 CTR。注意 970^30 ≈ 10^45,远超 JS 安全整数 2^53——所以 Python 侧用原生大整数、TS 侧必须用 BigInt,否则跨语言一定对不上。

S 层仿真:三种替换策略的 CPI 对照

S 层仿真脚本 sim/o03_fatigue.py 在 30 天窗口里比较三种策略:no-replace(一条跑满)、fixed-K(每 K 天强制换新)、threshold(CTR 跌破基线的 70% 才换)。成本 = 曝光成本(各策略相同)+ 素材条数 × 单条制作成本,最终看 CPI(每次安装成本)。

#!/usr/bin/env python3
# file: sim/o03_fatigue.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/o03_fatigue.py --config config/ --out out/
"""素材衰减与疲劳替换仿真:衰减曲线 → 替换时机 → 各策略 CPI 对照。

    ctrAt(d)   = baseCtr * decayNum**d // decayDen**d     # 素材服役第 d 日的 CTR(千分比)
    replaceDay = 首个满足 ctrAt(d) < baseCtr*thresholdPermille//1000 的 d
    第 d 日安装 = (imp*ctrAt(age)//1000) * ipm // 1000      # age = 该素材服役天数
    策略:no-replace 跑满 / fixed-K 每 K 天换新 / threshold 跌到阈值即换;成本 = 曝光成本 +
    素材条数*creativeCostFen;CPI = 总成本 // 总安装,最低者最优。

幂运算用 Python 大整数(decayNum**30 ≈ 10^45,远超 2^53,TS 侧必须 BigInt)。
真实衰减曲线与真实替换收益需真实业务环境,命令行不可复现(X)。产出 out/o03_fatigue.json + .csv + .svg。
"""

from __future__ import annotations

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


def ctr_at(cfg: dict, d: int) -> int:
    return cfg["baseCtrPermille"] * cfg["decayNum"] ** d // cfg["decayDen"] ** d


def replacement_day(cfg: dict) -> int:
    target = cfg["baseCtrPermille"] * cfg["thresholdPermille"] // 1000
    for d in range(0, cfg["horizonDays"] + 1):
        if ctr_at(cfg, d) < target:
            return d
    return cfg["horizonDays"]


def simulate(cfg: dict, mode: str, cycle: int) -> dict:
    imp, ipm, days = cfg["impressionsPerDay"], cfg["ipmPermille"], cfg["horizonDays"]
    total = 0
    for d in range(days):
        age = d if mode == "no-replace" else d % cycle
        total += (imp * ctr_at(cfg, age) // 1000) * ipm // 1000
    creatives = 1 if mode == "no-replace" else (days + cycle - 1) // cycle
    cost = imp * days * cfg["cpmFen"] // 1000 + creatives * cfg["creativeCostFen"]
    name = "no-replace" if mode == "no-replace" else f"{mode}-{cycle}"
    return {"strategy": name, "creatives": creatives, "totalInstalls": total,
            "impressionsCostFen": imp * days * cfg["cpmFen"] // 1000, "totalCostFen": cost,
            "cpiFen": cost // total if total else 10 ** 9}


def run(cfg: dict) -> dict:
    rday = replacement_day(cfg)
    strategies = [simulate(cfg, "no-replace", 0)]
    for k in cfg["replacementCycles"]:
        strategies.append(simulate(cfg, "fixed", k))
    strategies.append(simulate(cfg, "threshold", rday))
    best = min(strategies, key=lambda s: (s["cpiFen"], s["strategy"]))
    curve = [{"day": d, "ctrPermille": ctr_at(cfg, d)} for d in range(cfg["horizonDays"])]
    digest = (f'replaceDay:{rday}|' + ",".join(f'{s["strategy"]}:{s["cpiFen"]}' for s in strategies)
              + f'|best:{best["strategy"]}:{best["cpiFen"]}')
    return {"replaceDay": rday, "strategies": strategies, "best": best["strategy"],
            "bestCpiFen": best["cpiFen"], "curve": curve, "digest": digest}


def draw_svg(path: str, rep: dict) -> int:
    """纯 Python SVG:CTR 衰减折线 + 阈值横线 + 替换日竖线。"""
    curve, W, H, L, R, T, B = rep["curve"], 760, 320, 56, 24, 44, 46
    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 = "素材 CTR 日衰减曲线(千分比)与替换阈值"
    hi = max(c["ctrPermille"] for c in curve) or 1
    spanx, spany, last = W - L - R, H - T - B, len(curve) - 1
    thresh = hi * 700 // 1000
    yt = T + spany - spany * thresh // hi
    ET.SubElement(root, "line", {"x1": str(L), "y1": str(yt), "x2": str(W - R), "y2": str(yt),
                                 "stroke": "#ef4444", "stroke-dasharray": "4 4"})
    ET.SubElement(root, "text", {"x": str(W - R - 130), "y": str(yt - 4), "font-size": "11",
                                 "fill": "#ef4444"}).text = f"阈值 70% ≈ {thresh}‰"
    pts = [f'{L + spanx * c["day"] // last},{T + spany - spany * c["ctrPermille"] // hi}' for c in curve]
    ET.SubElement(root, "polyline", {"points": " ".join(pts), "fill": "none",
                                     "stroke": "#2563eb", "stroke-width": "2"})
    xr = L + spanx * rep["replaceDay"] // last
    ET.SubElement(root, "line", {"x1": str(xr), "y1": str(T), "x2": str(xr), "y2": str(T + spany),
                                 "stroke": "#16a34a", "stroke-dasharray": "3 3"})
    ET.SubElement(root, "text", {"x": str(xr + 4), "y": str(T + 14), "font-size": "11",
                                 "fill": "#16a34a"}).text = f"替换 D{rep['replaceDay']}"
    ET.SubElement(root, "line", {"x1": str(L), "y1": str(T + spany), "x2": str(W - R),
                                 "y2": str(T + spany), "stroke": "#9ca3af"})
    ET.SubElement(root, "text", {"x": str(L), "y": str(H - 16), "font-size": "11",
                                 "fill": "#6b7280"}).text = "D0 → D29;最优策略 " + rep["best"]
    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, "o03_fatigue.json"), encoding="utf-8") as f:
        cfg = json.load(f)
    rep = run(cfg)
    os.makedirs(args.out, exist_ok=True)
    svg_bytes = draw_svg(os.path.join(args.out, "o03_fatigue.svg"), rep)
    report = {"version": cfg["version"], "game": cfg["game"], "observedAt": cfg["observedAt"],
              "baseCtrPermille": cfg["baseCtrPermille"], "decayNum": cfg["decayNum"], "decayDen": cfg["decayDen"],
              "horizonDays": cfg["horizonDays"], **rep, "svg": os.path.join(args.out, "o03_fatigue.svg"),
              "svgBytes": svg_bytes}
    with open(os.path.join(args.out, "o03_fatigue.json"), "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False, indent=2)
    with open(os.path.join(args.out, "o03_fatigue.csv"), "w", encoding="utf-8", newline="") as f:
        w = csv.writer(f)
        w.writerow(["strategy", "creatives", "totalInstalls", "impressionsCostFen", "totalCostFen", "cpiFen"])
        for s in rep["strategies"]:
            w.writerow([s["strategy"], s["creatives"], s["totalInstalls"],
                        s["impressionsCostFen"], s["totalCostFen"], s["cpiFen"]])
    print(json.dumps({"replaceDay": rep["replaceDay"], "strategies": rep["strategies"],
                      "best": rep["best"], "bestCpiFen": rep["bestCpiFen"],
                      "curveHead": rep["curve"][:6], "digest": rep["digest"]},
                     ensure_ascii=False, indent=2))
    return 0


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

策略参数与阈值放在 config/o03_fatigue.json:

{
  "version": 1,
  "game": "merge-island",
  "observedAt": "2026-10",
  "horizonDays": 30,
  "baseCtrPermille": 32,
  "decayNum": 970,
  "decayDen": 1000,
  "impressionsPerDay": 200000,
  "cpmFen": 1600,
  "ipmPermille": 140,
  "thresholdPermille": 700,
  "creativeCostFen": 400000,
  "replacementCycles": [5, 7, 10, 14, 30],
  "sweep": [
    { "decayNum": 985, "decayDen": 1000 },
    { "decayNum": 970, "decayDen": 1000 },
    { "decayNum": 955, "decayDen": 1000 }
  ],
  "note": "素材衰减与替换参数,口径是「公开可观察机制」的建模假设,不是任何渠道后台实测。decayNum/decayDen = 每日 CTR 保留率(整数比,970/1000 即日衰减 3%);ctrAt(d) = baseCtr * decayNum^d // decayDen^d,幂运算必须用大整数(BigInt)。thresholdPermille = 替换阈值:当 CTR 跌到新鲜的 70% 以下即换新。creativeCostFen = 每条新素材的制作+测试成本。真实疲劳曲线与真实替换效果需真实业务环境,命令行不可复现(X)。全整数,无浮点。"
}

在本机跑 /opt/homebrew/bin/python3 sim/o03_fatigue.py --config config/ --out out/,各策略结果(同机相对量级,绝对数值不可移植):

策略素材条数总安装总成本(分)CPI(分)
no-replace11750010000000571
fixed-562520012000000476
fixed-752450011600000473
fixed-1032326810800000464
fixed-1432220410800000486
threshold-1332268010800000476

(fixed-30 与 no-replace 同为 571,从略。)替换日由阈值反推得到 D13。注意最优既不是「换得越勤越好」,也不是「阈值触发」:fixed-10 的 CPI 464 最低,比 threshold-13 的 476 还低,比不换的 571 低得多——换太勤(fixed-5/7)制作成本吃掉了收益,换太懒则一直在买高价量。

B 层基准:换一个衰减率,结论还成立吗

一个「最优周期」如果只在某一条衰减率下成立,它就是拟合噪声。B 层基准 sim/o03_fatigue_bench.py 对衰减率做敏感度扫描(decayNum 985 / 970 / 955),用完全相同的整数模型复算,检查最优策略是否翻盘:

#!/usr/bin/env python3
# file: sim/o03_fatigue_bench.py
# tier: B
# verify: /opt/homebrew/bin/python3 sim/o03_fatigue_bench.py --config config/ --out out/
"""B 基准:对素材衰减率做敏感度扫描,检查「最优替换周期」结论是否稳健。

对 config/o03_fatigue.json 的 sweep 逐档覆盖 (decayNum, decayDen),用与 sim/o03_fatigue.py
完全相同的整数模型复算各策略 CPI,输出每档的 replaceDay、最优策略与 CPI 极差,并检查最优
策略是否在扫描区间内保持不变(软告警,不硬失败)。

B 层对抗「编数字」:若换一个衰减率结论就翻盘,说明这个「最优周期」是拟合噪声。
数字全部来自本机实跑;真实衰减率需真实业务环境,命令行不可复现(X)。产出 out/o03_fatigue_bench.json。
"""

from __future__ import annotations

import argparse
import json
import os


def ctr_at(cfg: dict, d: int) -> int:
    return cfg["baseCtrPermille"] * cfg["decayNum"] ** d // cfg["decayDen"] ** d


def replacement_day(cfg: dict) -> int:
    target = cfg["baseCtrPermille"] * cfg["thresholdPermille"] // 1000
    for d in range(0, cfg["horizonDays"] + 1):
        if ctr_at(cfg, d) < target:
            return d
    return cfg["horizonDays"]


def simulate(cfg: dict, mode: str, cycle: int) -> dict:
    imp, ipm, days = cfg["impressionsPerDay"], cfg["ipmPermille"], cfg["horizonDays"]
    total = sum((imp * ctr_at(cfg, d if mode == "no-replace" else d % cycle) // 1000) * ipm // 1000
                for d in range(days))
    creatives = 1 if mode == "no-replace" else (days + cycle - 1) // cycle
    cost = imp * days * cfg["cpmFen"] // 1000 + creatives * cfg["creativeCostFen"]
    name = "no-replace" if mode == "no-replace" else f"{mode}-{cycle}"
    return {"strategy": name, "creatives": creatives, "totalInstalls": total,
            "totalCostFen": cost, "cpiFen": cost // total if total else 10 ** 9}


def run_one(cfg: dict) -> dict:
    rday = replacement_day(cfg)
    strategies = [simulate(cfg, "no-replace", 0)]
    for k in cfg["replacementCycles"]:
        strategies.append(simulate(cfg, "fixed", k))
    strategies.append(simulate(cfg, "threshold", rday))
    best = min(strategies, key=lambda s: (s["cpiFen"], s["strategy"]))
    cpis = [s["cpiFen"] for s in strategies]
    return {"decayNum": cfg["decayNum"], "decayDen": cfg["decayDen"], "replaceDay": rday,
            "best": best["strategy"], "bestCpiFen": best["cpiFen"], "cpiSpread": max(cpis) - min(cpis),
            "strategies": {s["strategy"]: s["cpiFen"] for s in strategies}}


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, "o03_fatigue.json"), encoding="utf-8") as f:
        base = json.load(f)
    rows = [run_one(dict(base, decayNum=s["decayNum"], decayDen=s["decayDen"]))
            for s in base["sweep"]]
    bests = {r["best"] for r in rows}
    stable = len(bests) == 1
    warnings = []
    if not stable:
        warnings.append(f"最优策略在扫描区间内不唯一:{sorted(bests)}")
    if max(r["cpiSpread"] for r in rows) < 40:
        warnings.append("各策略 CPI 极差过小,结论对成本结构不敏感")
    report = {"version": base["version"], "game": base["game"], "observedAt": base["observedAt"],
              "sweep": rows, "stableBest": stable, "bestStrategies": sorted(bests),
              "warnings": warnings}
    os.makedirs(args.out, exist_ok=True)
    with open(os.path.join(args.out, "o03_fatigue_bench.json"), "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False, indent=2)
    print(json.dumps({"sweep": [{"decayNum": r["decayNum"], "replaceDay": r["replaceDay"], "best": r["best"],
                                 "bestCpiFen": r["bestCpiFen"], "cpiSpread": r["cpiSpread"]} for r in rows],
                      "stableBest": stable, "bestStrategies": sorted(bests), "warnings": warnings},
                     ensure_ascii=False, indent=2))
    return 0


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

扫描结果:

衰减档替换日最优策略最优 CPI策略极差
985D25fixed-1043533
970D13fixed-10464107
955D9fixed-10496190

三档衰减率下最优策略都是 fixed-10,stableBest = true、无告警。衰减越快,替换日越早(D25 → D13 → D9)、CPI 越高、策略间极差越大——这正是预期:衰减越猛,「换不换」这件事越值钱。结论在扫描区间内稳健,不是拟合噪声。

X 层:真实衰减与真实收益不可复现

必须把边界划清。上面算的是替换决策侧:给定一条衰减曲线,哪种替换周期最省。下面这些属于 X 层,需真实业务环境,命令行不可复现:真实的衰减曲线(受素材、人群、竞争、季节共同影响,且不同素材衰减速度不同);真实的制作成本与再投放后的「新素材红利」幅度;真实收益里素材对留存与付费的长期影响。工程上的正确用法是把可复现的决策层当骨架,投放后用真实数据反标定 baseCtr 与 decayNum,再回代这套模型。

自检表

检查项合格标准
衰减可算曲线写成整数幂,config 里无浮点
成本算全曝光成本 + 素材制作成本都进 CPI
结论稳健换一档衰减率最优策略不翻盘(B 层)
边界清晰真实衰减与收益标 X,不写成「实测」

素材衰减是这一章最容易被忽略的「隐形税」:它不写在任何账单上,却每天都在抬高你的实际 CPI。把衰减写成曲线、把替换写成策略、把结论放进扫描里验证,是把这笔税从「凭感觉」变成「可预算」的唯一办法。

阅读导航:上一节:3.2 素材测试矩阵与优胜劣汰 · 下一节:4.1 归因窗口、点击归因与展示归因 。

继续阅读

探索更多技术文章

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

全部文章 返回首页

「minigame」更多文章

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