8.3 离线收益与回归设计

把「不在线的时间」也算进成长:在线速率、离线比例、逐段衰减与时长封顶的整数递推模型,逐分钟算出累计收益并产出 JSON、CSV 与纯 Python SVG 对比曲线,再用上限敏感性基准回答「离线补偿到底该给多长」,与 8.1 成长线、8.2 续玩机制接成闭环。

8.3 收尾「成长、进度与放置收益」:玩家离线一段时间再回来,凭什么还愿意继续?通用的放置经济框架——离线时长、产能、衰减、封顶——在《游戏经济与数值平衡》 一文里已经讲透;本节不重复它,而是给出一份可运行、可调参、可做敏感性分析的离线收益模型,把它与 8.1 的成长线、8.2 的「再来一局」接成一条闭环。

离线收益:把「不在线」也算进成长

离线收益的本质是「时间换资源」的补偿:在线每分钟产 55,离线只按 60% 起手(33/分钟),之后每满 60 分钟衰减 10%,最低不低于 5,总时长封顶 480 分钟。衰减与封顶缺一不可——离线收益是留存补偿,不是挂机刷分;没有衰减,玩家会退化成「只上线收菜」,把在线行为挤没了。

D 层:离线收益配置。 一份纯整数配置,定义在线速率、离线比例、满速率时长、衰减步长与速率下限、总时长上限。

{
  "version": 1,
  "game": "merge-island",
  "note": "离线收益模型配置。在线速率 onlineRatePerMinute;离线前 fullRateMinutes 分钟按 offlineRateNum/offlineRateDen 的比例累积,之后每 decayStepMinutes 分钟把速率乘 decayNum/decayDen(不低于 rateFloor),总时长封顶 capMinutes。全整数递推,禁浮点。",
  "onlineRatePerMinute": 55,
  "offlineRateNum": 60,
  "offlineRateDen": 100,
  "fullRateMinutes": 60,
  "decayStepMinutes": 60,
  "decayNum": 90,
  "decayDen": 100,
  "rateFloor": 5,
  "capMinutes": 480,
  "benchPerturbPct": 25
}

S 层:离线收益仿真。 下面这段 Python 逐分钟做整数递推(// 地板除,无浮点),产出 out/ch08_offline.json + out/ch08_offline.csv + out/ch08_offline.svg;SVG 用纯 xml.etree 手绘,对比「离线」与「一直在线」两条曲线。

#!/usr/bin/env python3
# file: sim/ch08_offline.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/ch08_offline.py --config config/ --out out/
"""离线收益模型:离线时长 → 累计收益。带衰减与上限的整数递推。

  base      = onlineRate * offlineRateNum // offlineRateDen      离线起手的每分钟收益
  第 m 分钟:total += rate
  若 m >= fullRateMinutes 且 (m - fullRateMinutes) % decayStepMinutes == 0:
      rate = max(rateFloor, rate * decayNum // decayDen)          速率逐段衰减
  minutes 上限 capMinutes。

产出 out/ch08_offline.json + out/ch08_offline.csv + out/ch08_offline.svg。
对照:同一时长若玩家一直在线,收益 = onlineRate * minutes(线性)。
"""

from __future__ import annotations

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


def load(cfg_dir):
    with open(os.path.join(cfg_dir, "ch08_offline.json"), encoding="utf-8") as f:
        return json.load(f)


def offline_base(cfg):
    return cfg["onlineRatePerMinute"] * cfg["offlineRateNum"] // cfg["offlineRateDen"]


def offline_earnings(minutes, cfg):
    """离线 minutes 分钟的累计收益(整数递推,逐分钟累加)。"""
    rate = offline_base(cfg)
    total = 0
    for m in range(1, minutes + 1):
        total += rate
        if m >= cfg["fullRateMinutes"] and (m - cfg["fullRateMinutes"]) % cfg["decayStepMinutes"] == 0:
            rate = max(cfg["rateFloor"], rate * cfg["decayNum"] // cfg["decayDen"])
    return total


def rate_at_minute(m, cfg):
    """第 m 分钟这一分钟实际使用的离线速率(m<=fullRateMinutes 为起手速率)。"""
    rate = offline_base(cfg)
    if m > cfg["fullRateMinutes"]:
        steps = (m - 1 - cfg["fullRateMinutes"]) // cfg["decayStepMinutes"] + 1
        for _ in range(steps):
            rate = max(cfg["rateFloor"], rate * cfg["decayNum"] // cfg["decayDen"])
    return rate


def segments(cfg):
    """按衰减段切分:每段的起止分钟、该段速率、该段小计。"""
    rate = offline_base(cfg)
    segs, start = [], 1
    subtotal = 0
    for m in range(1, cfg["capMinutes"] + 1):
        subtotal += rate
        if m >= cfg["fullRateMinutes"] and (m - cfg["fullRateMinutes"]) % cfg["decayStepMinutes"] == 0:
            segs.append({"fromMin": start, "toMin": m, "rate": rate, "subtotal": subtotal})
            start = m + 1
            subtotal = 0
            rate = max(cfg["rateFloor"], rate * cfg["decayNum"] // cfg["decayDen"])
    if start <= cfg["capMinutes"]:
        segs.append({"fromMin": start, "toMin": cfg["capMinutes"], "rate": rate, "subtotal": subtotal})
    return segs


def draw_svg(path, cfg):
    W, H, PAD = 760, 420, 60
    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": "26", "font-size": "15", "fill": "#111827"}).text = (
        "离线收益(蓝)与一直在线(灰虚线):离线时长 → 累计收益"
    )
    x0, x1, y0, y1 = PAD, W - 24, H - PAD, 54
    cap = cfg["capMinutes"]
    online_max = cfg["onlineRatePerMinute"] * cap or 1
    sx = lambda m: x0 + (x1 - x0) * m / cap
    sy = lambda v: y0 - (y0 - y1) * v / online_max
    for i in range(5):
        y = y1 + (y0 - y1) * i / 4
        ET.SubElement(svg, "line", {"x1": str(x0), "y1": f"{y:.1f}", "x2": str(x1), "y2": f"{y:.1f}",
                                    "stroke": "#e5e7eb"})
        ET.SubElement(svg, "text", {"x": str(x0 - 8), "y": f"{y + 4:.1f}", "font-size": "11",
                                    "text-anchor": "end", "fill": "#6b7280"}).text = str(
            int(online_max * (1 - i / 4)))
    ET.SubElement(svg, "line", {"x1": str(x0), "y1": str(y0), "x2": str(x1), "y2": str(y0), "stroke": "#9ca3af"})
    ET.SubElement(svg, "line", {"x1": str(x0), "y1": str(y1), "x2": str(x0), "y2": str(y0), "stroke": "#9ca3af"})
    # 在线(线性)
    ET.SubElement(svg, "line", {"x1": f"{sx(0):.1f}", "y1": f"{sy(0):.1f}",
                                "x2": f"{sx(cap):.1f}", "y2": f"{sy(online_max):.1f}",
                                "stroke": "#9ca3af", "stroke-width": "2", "stroke-dasharray": "6 4"})
    # 离线(分段折线,每段内线性)
    pts = [f"{sx(0):.2f},{sy(0):.2f}"]
    for s in segments(cfg):
        pts.append(f"{sx(s['toMin']):.2f},{sy(offline_earnings(s['toMin'], cfg)):.2f}")
    ET.SubElement(svg, "polyline", {"points": " ".join(pts), "fill": "none", "stroke": "#2563eb",
                                    "stroke-width": "2"})
    for s in segments(cfg):
        v = offline_earnings(s["toMin"], cfg)
        ET.SubElement(svg, "circle", {"cx": f"{sx(s['toMin']):.1f}", "cy": f"{sy(v):.1f}", "r": "3",
                                      "fill": "#2563eb"})
    # x 轴刻度(每 60 分钟)
    for m in range(0, cap + 1, 60):
        ET.SubElement(svg, "text", {"x": f"{sx(m):.1f}", "y": str(y0 + 16), "font-size": "10",
                                    "text-anchor": "middle", "fill": "#6b7280"}).text = str(m)
    ET.SubElement(svg, "text", {"x": str(x1), "y": str(y0 + 30), "font-size": "10",
                                "text-anchor": "end", "fill": "#6b7280"}).text = "离线分钟"
    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/")
    args = ap.parse_args()

    cfg = load(args.config)
    cap = cfg["capMinutes"]
    segs = segments(cfg)
    offline_at_cap = offline_earnings(cap, cfg)
    online_at_cap = cfg["onlineRatePerMinute"] * cap

    os.makedirs(args.out, exist_ok=True)
    with open(os.path.join(args.out, "ch08_offline.csv"), "w", encoding="utf-8", newline="") as f:
        w = csv.writer(f)
        w.writerow(["minutes", "offlineEarnings", "onlineEquivalent", "offlineRateNow"])
        for m in range(0, cap + 1, 15):
            w.writerow([m, offline_earnings(m, cfg), cfg["onlineRatePerMinute"] * m, rate_at_minute(m, cfg)])

    svg_bytes = draw_svg(os.path.join(args.out, "ch08_offline.svg"), cfg)
    report = {
        "version": cfg["version"],
        "config": os.path.join(args.config, "ch08_offline.json"),
        "onlineRatePerMinute": cfg["onlineRatePerMinute"],
        "offlineBaseRate": offline_base(cfg),
        "capMinutes": cap,
        "earningsAtCap": offline_at_cap,
        "onlineEquivalentAtCap": online_at_cap,
        "offlineVsOnlinePermille": (offline_at_cap * 1000) // online_at_cap,
        "effectiveOfflineRateAtCap": offline_at_cap // cap,
        "segments": segs,
        "svg": os.path.join(args.out, "ch08_offline.svg"),
        "svgBytes": svg_bytes,
    }
    with open(os.path.join(args.out, "ch08_offline.json"), "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False, indent=2)
    print(json.dumps({k: report[k] for k in (
        "onlineRatePerMinute", "offlineBaseRate", "capMinutes", "earningsAtCap",
        "onlineEquivalentAtCap", "offlineVsOnlinePermille", "effectiveOfflineRateAtCap")}, ensure_ascii=False))
    return 0


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

跑出来(同机相对量级,绝对数值不可移植):在线 55/分钟,离线起手 33/分钟,480 分钟封顶累计 10740;同样 480 分钟一直在线是 26400,离线只拿到 406‰(约 40.6%),折算到每分钟的有效离线速率是 22。分段看更清楚:

段分钟区间该段速率该段小计
11–60331980
261–120291740
3121–180261560
4181–240231380
5241–300201200
6301–360181080
7361–42016960
8421–48014840

八段小计相加正好是 10740。这里最值得盯的不是「最后只剩 14/分钟」,而是第 1 段就贡献了 1980(18.4%)——离线收益的重心天然压在回归的头一小时,这正是「离线补偿」该有的形状:奖励「回来」,而不是奖励「离线」。

B 层:上限敏感性。 把 capMinutes 上下扰动 ±25%(360 / 600),看满时长收益与有效速率各漂多少。B 层只做软告警,恒返回 0。

#!/usr/bin/env python3
# file: sim/ch08_offline_bench.py
# tier: B
# verify: /opt/homebrew/bin/python3 sim/ch08_offline_bench.py --config config/ --out out/
"""B 基准:离线时长上限(capMinutes)敏感性。

把 capMinutes 上下扰动 ±benchPerturbPct%,看两个结论漂多少:
  1) 离线满时长能拿到的累计收益;
  2) 折算到每分钟的有效离线速率。
软告警:漂移超过阈值只打 WARN,恒返回 0(B 层不硬失败)。
产出 out/ch08_offline_bench.json
"""

from __future__ import annotations

import argparse
import json
import os

from ch08_offline import load, offline_base, offline_earnings, segments

WARN_EARNINGS_DRIFT_PCT = 20  # 满时长收益漂移超过 20% 就打软告警


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--config", default="config/")
    ap.add_argument("--out", default="out/")
    args = ap.parse_args()

    cfg = load(args.config)
    pct = cfg["benchPerturbPct"]
    base_cap = cfg["capMinutes"]
    base_earn = offline_earnings(base_cap, cfg)
    base_eff = base_earn // base_cap

    rows = []
    for sign, tag in ((+1, "up"), (-1, "down")):
        cap = (base_cap * (100 + sign * pct)) // 100
        c2 = dict(cfg)
        c2["capMinutes"] = cap
        earn = offline_earnings(cap, c2)
        eff = earn // cap
        rows.append(
            {
                "dir": tag,
                "capMinutes": cap,
                "earnings": earn,
                "effectiveRate": eff,
                "earningsDriftPct": (earn - base_earn) * 100 // base_earn,
                "effectiveRateDriftPct": (eff - base_eff) * 100 // base_eff,
                "segmentCount": len(segments(c2)),
            }
        )

    earn_drift_max = max(abs(r["earningsDriftPct"]) for r in rows)
    eff_drift_max = max(abs(r["effectiveRateDriftPct"]) for r in rows)
    within = earn_drift_max <= WARN_EARNINGS_DRIFT_PCT

    os.makedirs(args.out, exist_ok=True)
    report = {
        "version": cfg["version"],
        "onlineRatePerMinute": cfg["onlineRatePerMinute"],
        "offlineBaseRate": offline_base(cfg),
        "baseCapMinutes": base_cap,
        "baseEarnings": base_earn,
        "baseEffectiveRate": base_eff,
        "perturbPct": pct,
        "cases": rows,
        "maxEarningsDriftPct": earn_drift_max,
        "maxEffectiveRateDriftPct": eff_drift_max,
        "warnThresholdPct": WARN_EARNINGS_DRIFT_PCT,
        "withinTolerance": within,
    }
    with open(os.path.join(args.out, "ch08_offline_bench.json"), "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False, indent=2)

    print(json.dumps({"baseEarnings": base_earn, "baseEffectiveRate": base_eff,
                      "maxEarningsDriftPct": earn_drift_max,
                      "maxEffectiveRateDriftPct": eff_drift_max,
                      "withinTolerance": within}, ensure_ascii=False))
    if not within:
        print(f"WARN 上限扰动 {pct}% 后满时长收益漂移 {earn_drift_max}% > {WARN_EARNINGS_DRIFT_PCT}%")
    return 0  # B 层软告警:恒返回 0


if __name__ == "__main__":
    raise SystemExit(main())
{"baseEarnings": 10740, "baseEffectiveRate": 22, "maxEarningsDriftPct": 17, "maxEffectiveRateDriftPct": 10, "withinTolerance": true}

上限 ±25% 时满时长收益漂 17%、有效速率只漂 10%,都在 20% 阈值内。收益漂得比上限多、有效速率漂得比上限少,正是衰减曲线在起作用:上限拉长时多出来的那段时间仍在拿递减后的收益(拉高总额),但摊到更长的分钟数上(压低均值)。这条「总额对上限更敏感、均值对上限更钝」的结论,是判断「离线补偿该给多长」的直接依据:把上限从 8 小时拉到 10 小时,玩家感知到的是总额上涨,而单分钟效率其实被稀释了——所以上限该按「一次自然离线的最长间隔」定,而不是按「能多发多少」定。至此,8.1 的成长线、8.2 的「再来一局」与本节离线收益,共同构成小游戏「成长—进度—放置」三件套的可运行闭环。

阅读导航:上一节:进度感与「再来一局」 · 下一节:小游戏特有的社交传播面 。

继续阅读

探索更多技术文章

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

全部文章 返回首页

「minigame」更多文章

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