15.3 联盟健康度指标

联盟从健康到瓦解不是一夜之间。这一节用「健康 vs 衰退」两个剧本逐日对照,量出谁先亮红灯:贡献率第 27 天、参与率第 29 天就先越线,而流失率要拖到第 52 天才超标——等流失率起来人早就心散了,预警真正该盯的是前两个领先指标,不是流失率。

联盟是 SLG 的留存基本盘。玩家为联盟打 GVG、为联盟捐资源、为联盟留在一个区服——联盟一散,人就走。但「联盟要散了」这件事从来不是突然发生的,它是一个逐日下滑的过程,只不过大多数运营等到「人跑光了」才反应过来。

这一节要回答一个具体问题:在联盟瓦解的过程中,哪些指标先亮红灯,哪些指标最后才亮? 答案决定了预警该盯哪个数。

五个指标

把联盟状态拆成五个可算的数。前四个是「当下状态」,最后一个是「分布是否畸形」:

指标阈值含义方向
活跃率 activeRatePct≥ 60%当天在线成员占比低于阈值告警
GVG 参与率 gvgParticipatePct≥ 50%当天参与 GVG / 贡献的成员占比低于阈值告警
贡献产出 contributionPct≥ 85%相对基线产出的百分比(7 天平滑)低于阈值告警
流失率 churnPct≤ 10%连续 5 天不在线者的累计占比高于阈值告警
基尼系数 giniMaxPct≤ 70%贡献分布的集中度高于阈值告警

前四个是时间序列,看的是「今天比昨天差多少」;基尼是截面,看的是「是不是少数核心在扛全盟」。两者结合,才叫健康度——一个产出很高但基尼 90% 的联盟,比一个产出中等但基尼 40% 的联盟更脆,因为它全靠几个核心,核心一走就崩。

S 层:健康 vs 衰退的逐日对照

要找出「谁先亮红灯」,最干净的办法是造一个对照实验:同一批成员、同一个 PRNG 流,跑两个剧本。健康剧本里,在线与参与概率全程恒定;衰退剧本里,从第 20 天起,参与概率按每天 2%、在线概率按每天 1% 下滑——参与先瓦解,在线后瓦解,模拟「先心散、后散伙」。连续 5 天不在线者记流失。20 次试验配对平均以压住单日噪声。

# file: sim/n15_health.py
# tier: S
# verify: python3 sim/n15_health.py --config config/ --out out/
"""联盟健康度指标(15.3):健康 vs 衰退两剧本逐日观测,找领先指标。

逐日判定成员「在线」(present)与「参与」(engaged:打 GVG / 贡献);
健康剧本两者概率恒定,衰退剧本自 declineStartDay 起参与概率按 engageDeclineRatePct、
在线概率按 presenceDeclineRatePct 下滑(参与先于在线瓦解,符合「先心散后散伙」);
连续 missChurnDays 天不在线者记流失。20 次试验配对平均以压噪声。
产出 out/n15_health.json + .csv + .svg。
"""
import argparse
import csv
import json
import os
import sys
import xml.etree.ElementTree as ET

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from svg_line import polyline_points  # noqa: E402

SVG_NS = "http://www.w3.org/2000/svg"


def mulberry32(seed):
    state = seed & 0xFFFFFFFF

    def nxt():
        nonlocal state
        state = (state + 0x6D2B79F5) & 0xFFFFFFFF
        t = ((state ^ (state >> 15)) * (state | 1)) & 0xFFFFFFFF
        t = (t ^ (t + (((t ^ (t >> 7)) * (t | 61)) & 0xFFFFFFFF))) & 0xFFFFFFFF
        return ((t ^ (t >> 14)) & 0xFFFFFFFF) / 4294967296.0

    return nxt


def probs(day, scenario, h):
    p0 = h["activeProbPct"]
    if scenario == "health" or day < h["declineStartDay"]:
        return p0, p0
    k = day - h["declineStartDay"] + 1
    return (max(h["presentFloorPct"], p0 - k * h["presenceDeclineRatePct"]),
            max(h["engageFloorPct"], p0 - k * h["engageDeclineRatePct"]))


def simulate(cfg, rng, scenario):
    h = cfg["health"]
    days, n, base = h["seasonDays"], h["members"], h["baseContribution"]
    mult = [60 + int(rng() * 100) for _ in range(n)]  # 固定成员产出系数 60..159
    gone = [False] * n
    miss = [0] * n
    rows = []
    for day in range(1, days + 1):
        present_p, engage_p = probs(day, scenario, h)
        present = engaged = contrib = 0
        for i in range(n):
            if gone[i]:
                continue
            if int(rng() * 100) < present_p:
                miss[i] = 0
                present += 1
                if int(rng() * 100) < engage_p:
                    engaged += 1
                    contrib += base * mult[i] // 100
            else:
                miss[i] += 1
                if miss[i] >= h["missChurnDays"]:
                    gone[i] = True
        rows.append({"day": day, "activeRatePct": round(100.0 * present / n, 1),
                     "gvgParticipatePct": round(100.0 * engaged / n, 1),
                     "contribution": contrib, "churnPct": round(100.0 * sum(gone) / n, 1)})
    return rows


def first_breach(rows, key, threshold, mode, start_day):
    for r in rows:
        if r["day"] >= start_day and ((mode == "below" and r[key] < threshold) or (mode == "above" and r[key] > threshold)):
            return r["day"]
    return None


def render_multi(path, series, title):
    allv = [v for _, ys, _ in series for v in ys]
    lo, hi = min(allv), max(allv)
    svg = ET.Element("svg", {"xmlns": SVG_NS, "viewBox": "0 0 640 360", "width": "640", "height": "360"})
    ET.SubElement(svg, "rect", {"width": "640", "height": "360", "fill": "#ffffff"})
    for _, ys, color in series:
        ET.SubElement(svg, "polyline", {"points": polyline_points(ys, lo, hi),
                                        "fill": "none", "stroke": color, "stroke-width": "2"})
    ET.SubElement(svg, "text", {"x": "56", "y": "16", "font-size": "13", "fill": "#111"}).text = title
    os.makedirs(os.path.dirname(path), exist_ok=True)
    ET.ElementTree(svg).write(path, encoding="utf-8", xml_declaration=True)
    return path


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--config", default="config/")
    ap.add_argument("--out", default="out/")
    ap.add_argument("--seed", type=int, default=None)
    args = ap.parse_args()
    cfg = json.load(open(os.path.join(args.config, "n15_alliance.json"), encoding="utf-8"))
    h = cfg["health"]
    th = h["thresholds"]
    seed = args.seed if args.seed is not None else cfg["seed"]
    trials = h["trials"]
    keys = ("activeRatePct", "gvgParticipatePct", "contribution", "churnPct")

    def run_avg(scenario):
        acc = [dict(dict.fromkeys(keys, 0.0), day=d + 1) for d in range(h["seasonDays"])]
        for t in range(trials):
            rows = simulate(cfg, mulberry32(seed + 104729 * t + 11), scenario)
            for i, r in enumerate(rows):
                for k in keys:
                    acc[i][k] += r[k]
        for r in acc:
            for k in keys:
                r[k] /= trials
            r.update(activeRatePct=round(r["activeRatePct"], 1), gvgParticipatePct=round(r["gvgParticipatePct"], 1),
                     churnPct=round(r["churnPct"], 1), contribution=round(r["contribution"]))
        return acc

    health = run_avg("health")
    decline = run_avg("decline")
    contrib_base = sum(r["contribution"] for r in health if r["day"] < h["declineStartDay"]) / (h["declineStartDay"] - 1)
    win = h["contributionSmoothDays"]
    for rows in (health, decline):
        for i, r in enumerate(rows):
            r["contributionPct"] = round(100.0 * r["contribution"] / contrib_base, 1)
            seg = [rows[j]["contributionPct"] for j in range(max(0, i - win + 1), i + 1)]
            r["contributionPctSmooth"] = round(sum(seg) / len(seg), 1)

    ds = h["declineStartDay"]
    breaches = {
        "gvgParticipatePct": first_breach(decline, "gvgParticipatePct", th["gvgParticipatePct"], "below", ds),
        "contributionPctSmooth": first_breach(decline, "contributionPctSmooth", th["contributionPct"], "below", ds),
        "activeRatePct": first_breach(decline, "activeRatePct", th["activeRatePct"], "below", ds),
        "churnPct": first_breach(decline, "churnPct", th["churnMaxPct"], "above", ds),
    }
    os.makedirs(args.out, exist_ok=True)
    with open(os.path.join(args.out, "n15_health.json"), "w", encoding="utf-8") as f:
        json.dump({"version": 1, "seed": seed, "trials": trials, "thresholds": th,
                   "contributionBase": round(contrib_base, 1), "breachDay": breaches,
                   "healthDay60": health[-1], "declineDay60": decline[-1]}, f, ensure_ascii=False, indent=2)
    with open(os.path.join(args.out, "n15_health.csv"), "w", encoding="utf-8", newline="") as f:
        w = csv.writer(f)
        w.writerow(["scenario", "day", "activeRatePct", "gvgParticipatePct", "contribution",
                    "contributionPct", "contributionPctSmooth", "churnPct"])
        for name, rows in (("health", health), ("decline", decline)):
            for r in rows:
                w.writerow([name, r["day"], r["activeRatePct"], r["gvgParticipatePct"], r["contribution"],
                            r["contributionPct"], r["contributionPctSmooth"], r["churnPct"]])
    render_multi(os.path.join(args.out, "n15_health.svg"),
                 [("健康·参与率", [r["gvgParticipatePct"] for r in health], "#1a7f37"),
                  ("衰退·参与率", [r["gvgParticipatePct"] for r in decline], "#d1242f"),
                  ("衰退·产出%", [r["contributionPctSmooth"] for r in decline], "#1f6feb")],
                 "健康 vs 衰退:参与率领先、流失率滞后")
    print(f"{trials} 次试验平均;衰退剧本阈值越线天(第 {ds} 天起衰退):")
    for k, v in breaches.items():
        print(f"  {k:>20}: {'第 %d 天' % v if v else '未越线'}")
    print(f"第 60 天  健康:活跃 {health[-1]['activeRatePct']}% 参与 {health[-1]['gvgParticipatePct']}% 流失 {health[-1]['churnPct']}%")
    print(f"第 60 天  衰退:活跃 {decline[-1]['activeRatePct']}% 参与 {decline[-1]['gvgParticipatePct']}% "
          f"产出 {decline[-1]['contributionPctSmooth']}% 流失 {decline[-1]['churnPct']}%")


if __name__ == "__main__":
    main()

本机实跑输出:

20 次试验平均;衰退剧本阈值越线天(第 20 天起衰退):
     gvgParticipatePct: 第 29 天
  contributionPctSmooth: 第 27 天
         activeRatePct: 第 40 天
              churnPct: 第 52 天
第 60 天  健康:活跃 85.5% 参与 74.6% 流失 0.5%
第 60 天  衰退:活跃 37.1% 参与 11.4% 产出 13.8% 流失 24.4%

谁先亮红灯:领先指标与滞后指标

把四个越线天排开,规律很清楚:

指标越线天类型
贡献产出 contributionPctSmooth第 27 天领先
GVG 参与率 gvgParticipatePct第 29 天领先
活跃率 activeRatePct第 40 天中间
流失率 churnPct第 52 天滞后

产出和参与率是最早的报警器,比流失率早了整整 23~25 天。 第 20 天开始衰退,第 27 天产出就跌破 85%——因为「参与」概率先掉,贡献自然跟着掉,而人还在(在线率还没跌破 60%)。等到第 40 天活跃率才跌破 60%,第 52 天流失率才超 10%。

这条时间线就是预警设计的全部依据:

  • 盯流失率 = 事后验尸。 流失率亮红灯时,联盟已经跑了四分之一的人(第 60 天 24.4%),救不回来了。
  • 盯参与率 / 贡献率 = 提前 3 周预警。 一个还在线但不再打 GVG 的成员,是「心已经走了、人还没走」。这正是 15.1 里说的社交压力的负面——人在盟里,但不再参与。
  • 活跃率是中间指标,它开始掉的时候,参与率已经掉了 11 天了。

所以健康的联盟看板应该是倒着排的:把参与率和贡献率放在最上面,活跃率其次,流失率放在最下面(它只是结果,不是信号)。

阈值怎么定

上面的阈值(参与率 50%、贡献 85%、活跃 60%、流失 10%)不是拍脑袋,它们要和健康剧本的稳态值拉开距离:

  • 健康剧本第 60 天参与率 74.6%,阈值 50%——留了 24 个百分点的缓冲,正常波动不会误报。
  • 健康剧本流失率 0.5%,阈值 10%——留了 20 倍缓冲。
  • 贡献用 7 天滚动平滑,阈值 85%——单日贡献波动很大,不平滑的话第 20 天当天就可能误触发。

阈值定得离稳态太近,看板天天报警、运营就麻木了;定得太远,等触发时又太晚。健康的阈值应该让健康联盟在整个赛季里一次都不报警——这也是本节 S 层脚本顺带验证的事:健康剧本 60 天里,四个阈值一个都没越线。

小结

联盟健康度不是一个数,是一组有先后的数。设计预警时的三个要点:

  1. 参与率和贡献率是领先指标,比流失率早约 3 周报警——看板要盯这两个。
  2. 流失率是滞后指标,等它超标,联盟已经跑掉四分之一的人。
  3. 阈值必须相对健康稳态留足缓冲,并配合滚动平滑,否则天天误报。

这三个指标(参与、贡献、活跃)和 15.2 的 GVG 匹配是同一套玩法的两面:匹配决定 GVG 打得公不公平,健康度决定玩家还愿不愿意来打 GVG。

阅读导航:上一节:15.2 GVG 匹配与战场设计 · 下一节:16.1 把两卷装配成完整设计案 。

继续阅读

探索更多技术文章

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

全部文章 返回首页

「slg」更多文章

  1. 12.3 留存与付费的联合分析
  2. 12.2 流失预警与召回
  3. 附录 C:全书公式、参数与数据集索引