13.1 通胀症状与成因

经济生病了,第一步是能诊断。本节把「通胀 / 通缩 / 健康」从口头判断变成可判定的量:读一段存量时序,算出存量、价格、交易三组日均变化率,再归因到产出侧、消耗侧或外部注入;用《王朝纪元》五个场景验证判定与归因,并给出判定稳定性与余量的 B 层基准。

如果你搜「游戏经济通胀」,会先撞到站内那篇产业向的游戏经济系统运营清单 ——它把「盯货币存量、控产出、补回收」列成一张 checklist。清单有用,但它不回答一个更靠前的问题:我手上这条曲线,现在到底是不是通胀? 本节补的正是这一环:把诊断写成一段能跑出数字、能给出归因的代码。站内的经济平衡长文 给的是通用框架、举例用放置类,而 SLG 的存量由地块产出与玩家行为驱动,口径不同。

第 12 章我们已经能用 config/ 跑出 180 天的存量曲线,也能判定它有没有稳态。但「曲线在涨」和「曲线病了」是两件事。涨多少算病?涨的原因在哪一侧? 这两个问题不解决,后面的沉淀、回收、赛季重置全都是盲调。

症状不是「物价高」,是一组同时出现的信号

新手容易把通胀理解成「东西变贵了」。在 SLG 里,物价甚至不一定是显式的——很多品类没有公开市场,玩家感受到的是另外四件事同时发生:

  • 存量飞涨:老玩家的资源/货币越攒越多,仓库上限形同虚设;
  • 资源囤积:囤积玩家占比上升,市场上可流通的部分反而变少;
  • 交易停滞:人人都觉得「明天更便宜」,没人愿意现在卖,交易量下滑;
  • 活跃流失:新玩家追赶无望,老玩家觉得「攒了也没用」,两头都在掉。

单看任何一个指标都会误判:存量在涨,可能是赛季末的正常积累;交易在跌,可能是版本内容空窗。只有四个信号同时出现,才构成一次值得动手的通胀。 通缩是镜像——存量萎缩、价格阴跌、玩家「舍不得花」、交易冻结。通缩往往更致命:玩家不是不想花,是没有可花的量,产出端一旦砍过头,整条循环会停摆。

三个成因,对应三种完全不同的处方

症状告诉你有病,成因告诉你怎么治。把存量变化写成一条守恒式 Δ存量 = 产出 − 消耗 + 外部注入,右边三项正好是三种成因:

成因守恒式里的表现处方方向
产出侧产出增长快过消耗增长压产出曲线,或让消耗跟着产出一起涨
消耗侧消耗能力萎缩(玩家进度封顶、消耗槽不够)补消耗槽,别只想着砍产出
外部注入有一份「产出、消耗都不解释」的净流入查打金、查跨服转移、查活动超发

归因错了,处方就反了:明明是消耗侧萎缩,你却去砍产出,结果是玩家既攒不下资源又打不动仗——把通胀治成了通缩。

先把症状与成因写成配置

下面这份是 D 层(静态层)配置:五个场景各自给出时序的生成参数(base / trendBp / noiseBp),以及判定用的整数带。所有阈值都是基点(bp,万分之一/日),没有浮点。

{
  "version": 1,
  "note": "通胀/通缩诊断。每个场景给出时序生成参数,脚本用自实现确定性 PRNG 物化存量/价格/交易/流量序列,诊断器只消费物化后的序列。bands 全部为整数基点(bp,1bp=万分之一/日)。",
  "days": 30,
  "bands": {
    "inflationStockBp": 250, "inflationPriceBp": 150, "hoardingBp": 5500,
    "deflationStockBp": -80, "deflationTradeBp": -150,
    "externalShareBp": 2500, "sinkDecayBp": -100, "productionMarginBp": 100
  },
  "scenarios": [
    { "id": "prod_overdrive", "name": "产出侧过载", "expectClass": "inflation", "expectCause": "production", "seed": 1301,
      "stock": { "base": 1000000, "trendBp": 400, "noiseBp": 30 }, "price": { "base": 100, "trendBp": 260, "noiseBp": 6 },
      "trade": { "base": 52000, "trendBp": 150, "noiseBp": 50 }, "hoardingBp": 6300,
      "flows": { "output": { "base": 96000, "trendBp": 300 }, "sink": { "base": 82000, "trendBp": 20 }, "external": { "base": 0, "trendBp": 0 } } },
    { "id": "external_goldfarm", "name": "外部注入(打金)", "expectClass": "inflation", "expectCause": "external", "seed": 1302,
      "stock": { "base": 1000000, "trendBp": 360, "noiseBp": 35 }, "price": { "base": 100, "trendBp": 240, "noiseBp": 6 },
      "trade": { "base": 48000, "trendBp": 120, "noiseBp": 50 }, "hoardingBp": 6000,
      "flows": { "output": { "base": 96000, "trendBp": 0 }, "sink": { "base": 82000, "trendBp": 0 }, "external": { "base": 42000, "trendBp": 0 } } },
    { "id": "sink_decay", "name": "消耗侧萎缩", "expectClass": "inflation", "expectCause": "sink", "seed": 1303,
      "stock": { "base": 1000000, "trendBp": 330, "noiseBp": 30 }, "price": { "base": 100, "trendBp": 200, "noiseBp": 6 },
      "trade": { "base": 50000, "trendBp": 60, "noiseBp": 50 }, "hoardingBp": 5800,
      "flows": { "output": { "base": 96000, "trendBp": 40 }, "sink": { "base": 90000, "trendBp": -320 }, "external": { "base": 0, "trendBp": 0 } } },
    { "id": "sink_overdrive", "name": "消耗侧过强(通缩)", "expectClass": "deflation", "expectCause": "sink", "seed": 1304,
      "stock": { "base": 1500000, "trendBp": -150, "noiseBp": 20 }, "price": { "base": 100, "trendBp": -60, "noiseBp": 4 },
      "trade": { "base": 60000, "trendBp": -350, "noiseBp": 60 }, "hoardingBp": 7200,
      "flows": { "output": { "base": 90000, "trendBp": -60 }, "sink": { "base": 105000, "trendBp": 180 }, "external": { "base": 0, "trendBp": 0 } } },
    { "id": "healthy_baseline", "name": "健康基线", "expectClass": "healthy", "expectCause": "none", "seed": 1305,
      "stock": { "base": 1000000, "trendBp": 40, "noiseBp": 25 }, "price": { "base": 100, "trendBp": 20, "noiseBp": 4 },
      "trade": { "base": 50000, "trendBp": 30, "noiseBp": 40 }, "hoardingBp": 4200,
      "flows": { "output": { "base": 90000, "trendBp": 30 }, "sink": { "base": 88000, "trendBp": 30 }, "external": { "base": 0, "trendBp": 0 } } }
  ]
}

每个场景带 expectClass / expectCause 两个期望标签,是给 B 层基准做对照用的;诊断器本身不看这两个字段。

S 层:把一段时序判成三类并归因

下面这段是 S 层仿真脚本,读上面那份配置,用自实现确定性 PRNG(mulberry32)把每个场景物化成五条整数序列,再判定类别、归因成因,落成 JSON 与 CSV。

# file: sim/ch13_diagnosis.py
# tier: S
# verify: python3 sim/ch13_diagnosis.py --config config/ --out out/
"""通胀 / 通缩诊断器:给定一段存量时序,自动判定 健康 / 通胀 / 通缩,并归因到产出侧 / 消耗侧 / 外部注入。

时序由 config 的生成参数用自实现确定性 PRNG(mulberry32)物化,诊断器只消费物化后的序列;
真实后台导出的序列可直接替换,那部分属 X 层。
产出:out/ch13_diagnosis.json + out/ch13_diagnosis.csv
"""

from __future__ import annotations

import argparse
import csv
import json
import os


def mulberry32(seed: int):
    """确定性 PRNG,返回 [0,1) 取值函数;与 proto 侧同构。"""
    a = seed & 0xFFFFFFFF

    def rnd() -> float:
        nonlocal a
        a = (a + 0x6D2B79F5) & 0xFFFFFFFF
        t = (a ^ (a >> 15)) & 0xFFFFFFFF
        t = (t * (1 | a)) & 0xFFFFFFFF
        t = ((t + ((t ^ (t >> 7)) * (61 | t))) ^ t) & 0xFFFFFFFF
        return ((t ^ (t >> 14)) & 0xFFFFFFFF) / 4294967296

    return rnd


def gen(spec: dict, days: int, rnd) -> list[int]:
    """base × (1+trendBp/10000)^d × (1±noiseBp),取整。"""
    v = float(spec["base"])
    trend = 1.0 + spec["trendBp"] / 10000.0
    noise = spec.get("noiseBp", 0) / 10000.0
    out = []
    for _ in range(days):
        v *= trend
        out.append(int(round(v * (1.0 + (rnd() * 2.0 - 1.0) * noise))))
    return out


def trend_bp(s: list[int]) -> int:
    """日均变化率(bp)。整数地板除,与 TS 侧 Math.floor 一致。"""
    return 0 if len(s) < 2 or s[0] == 0 else (s[-1] - s[0]) * 10000 // (s[0] * (len(s) - 1))


def diagnose(sc: dict, bands: dict, days: int) -> dict:
    rnd = mulberry32(sc["seed"])
    stock, price, trade = gen(sc["stock"], days, rnd), gen(sc["price"], days, rnd), gen(sc["trade"], days, rnd)
    f = sc["flows"]
    output, sink, ext = gen(f["output"], days, rnd), gen(f["sink"], days, rnd), gen(f["external"], days, rnd)
    sb, pb, tb, hoard = trend_bp(stock), trend_bp(price), trend_bp(trade), sc["hoardingBp"]
    if sb >= bands["inflationStockBp"] and (pb >= bands["inflationPriceBp"] or hoard >= bands["hoardingBp"]):
        cls = "inflation"
    elif sb <= bands["deflationStockBp"] or tb <= bands["deflationTradeBp"]:
        cls = "deflation"
    else:
        cls = "healthy"
    inflow = sum(output) + sum(ext)
    share = sum(ext) * 10000 // inflow if inflow > 0 else 0
    og, sg = trend_bp(output), trend_bp(sink)
    if cls == "healthy":
        cause = "none"
    elif share >= bands["externalShareBp"]:
        cause = "external"
    elif sg <= bands["sinkDecayBp"]:
        cause = "sink"
    elif og - sg >= bands["productionMarginBp"]:
        cause = "production"
    else:
        cause = "sink"
    return {"id": sc["id"], "name": sc["name"], "class": cls, "cause": cause,
            "stockTrendBp": sb, "priceTrendBp": pb, "tradeTrendBp": tb, "hoardingBp": hoard,
            "externalShareBp": share, "outputTrendBp": og, "sinkTrendBp": sg, "stockEnd": stock[-1],
            "expectClass": sc["expectClass"], "expectCause": sc["expectCause"],
            "classMatch": cls == sc["expectClass"], "causeMatch": cause == sc["expectCause"]}


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, "ch13_diagnosis.json"), encoding="utf-8") as fh:
        cfg = json.load(fh)
    res = [diagnose(s, cfg["bands"], cfg["days"]) for s in cfg["scenarios"]]
    summary = {"total": len(res), "classMatched": sum(r["classMatch"] for r in res),
               "causeMatched": sum(r["causeMatch"] for r in res)}
    os.makedirs(a.out, exist_ok=True)
    with open(os.path.join(a.out, "ch13_diagnosis.json"), "w", encoding="utf-8") as fh:
        json.dump({"version": 1, "days": cfg["days"], "results": res, "summary": summary}, fh, ensure_ascii=False, indent=2)
    cols = ["id", "class", "cause", "stockTrendBp", "priceTrendBp", "tradeTrendBp", "hoardingBp", "externalShareBp"]
    with open(os.path.join(a.out, "ch13_diagnosis.csv"), "w", encoding="utf-8", newline="") as fh:
        w = csv.writer(fh)
        w.writerow(cols)
        for r in res:
            w.writerow([r[c] for c in cols])
    print(json.dumps({"summary": summary, "verdicts": {r["id"]: r["class"] + "/" + r["cause"] for r in res}}, ensure_ascii=False))
    return 0


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

在本机跑它(同机相对量级,绝对数值不可移植):

{"summary": {"total": 5, "classMatched": 5, "causeMatched": 5},
 "verdicts": {"prod_overdrive": "inflation/production", "external_goldfarm": "inflation/external",
              "sink_decay": "inflation/sink", "sink_overdrive": "deflation/sink",
              "healthy_baseline": "healthy/none"}}

五个场景的判定与归因全部命中期望标签。落下来的 out/ch13_diagnosis.csv 长这样:

id,class,cause,stockTrendBp,priceTrendBp,tradeTrendBp,hoardingBp,externalShareBp
prod_overdrive,inflation,production,725,378,187,6300,0
external_goldfarm,inflation,external,618,344,140,6000,3043
sink_decay,inflation,sink,538,267,66,5800,0
sink_overdrive,deflation,sink,-124,-56,-222,7200,0
healthy_baseline,healthy,none,42,20,29,4200,0

五个场景,三条判据

把上面那张表读一遍,诊断器其实只用了三个数在干活:

  • 存量日均变化率:725 / 618 / 538 三个场景都在 250 bp 以上,越过通胀线;-124 的场景越过通缩线;健康基线只有 42。
  • 价格与囤积的二选一:通胀要求「价格也在涨」或「囤积率越线」。prod_overdrive 的价格涨到 378 bp,external_goldfarm 涨到 344 bp,两个都靠价格这一条触发;如果某天价格被官方锚定住了,囤积率就是备用判据。
  • 交易量的独立通缩信号:sink_overdrive 的交易量是 -222 bp,即使不看存量也知道市场在冻结。

这三条判据是并列的,不是加权求和:加权求和需要一个谁也说不清的权重,而并列判据的每一环都能被单独复盘——「为什么判成通胀?因为存量 +618 bp 且价格 +344 bp」,这句话在复盘会上比一个总分有用得多。

归因:三个证据,一个优先级

归因的规则同样刻意做得很直白,优先级从高到低:

  1. 外部注入优先:只要外部净流入占「总流入」的比例越过 2500 bp(25%),直接判 external。external_goldfarm 场景实测 3043 bp,一眼就是外部注水。
  2. 消耗侧萎缩次之:消耗的日均变化率跌破 -100 bp,说明消耗槽在塌,判 sink。sink_decay 场景的消耗是 -320 bp/日,资源无处可去。
  3. 产出侧最后:产出增长比消耗增长高出 100 bp 以上,判 production。prod_overdrive 场景产出 +300 bp、消耗 +20 bp,差 280 bp,产出侧无疑。

优先级之所以是「外部 → 消耗 → 产出」,是因为外部注入是唯一一种「运营没做错、但数据就是脏」的情况,必须最先排除;把它排掉之后,再看是自己的消耗槽塌了,还是产出曲线自己跑飞了。

B 层:判定稳不稳,余量够不够

一个诊断器最怕的不是判错,而是在噪声里来回跳——今天判通胀、明天判健康,运营根本没法据此决策。下面这段是 B 层基准,把每个场景的 PRNG 种子扰动 60 次重生成时序,看结论是否稳定、距判定边界还剩多少余量。

# file: sim/ch13_diagnosis_baseline.py
# tier: B
# verify: python3 sim/ch13_diagnosis_baseline.py --config config/ --seeds 60
"""诊断器的 B 层基准:扰动 PRNG 种子重生成 60 条时序,看判定结论稳不稳,
并量出「观测趋势距最近的判定边界还有多少 bp」的余量。软告警,不改变退出码。
"""

from __future__ import annotations

import argparse
import json
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import ch13_diagnosis as D  # noqa: E402


def margin(stock_bp: int, cls: str, b: dict) -> int:
    """距最近的判定边界还有多少 bp;正数=有余量。"""
    if cls == "inflation":
        return min(stock_bp - b["inflationStockBp"], stock_bp - b["deflationStockBp"])
    if cls == "deflation":
        return min(b["inflationStockBp"] - stock_bp, b["deflationStockBp"] - stock_bp)
    return min(b["inflationStockBp"] - stock_bp, stock_bp - b["deflationStockBp"])


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--config", default="config/")
    ap.add_argument("--seeds", type=int, default=60)
    ap.add_argument("--out", default="out/")
    a = ap.parse_args()
    with open(os.path.join(a.config, "ch13_diagnosis.json"), encoding="utf-8") as fh:
        cfg = json.load(fh)
    rows, warns = [], []
    for s in cfg["scenarios"]:
        cls, cau, tr = [], [], []
        for k in range(a.seeds):
            r = D.diagnose(dict(s, seed=s["seed"] + k * 7919), cfg["bands"], cfg["days"])
            cls.append(r["class"])
            cau.append(r["cause"])
            tr.append(r["stockTrendBp"])
        ca = cls.count(s["expectClass"]) * 100 // a.seeds
        cc = cau.count(s["expectCause"]) * 100 // a.seeds
        row = {"id": s["id"], "classAccPct": ca, "causeAccPct": cc, "stable": len(set(cls)) == 1,
               "trendMinBp": min(tr), "trendMaxBp": max(tr),
               "marginBp": min(margin(min(tr), s["expectClass"], cfg["bands"]),
                               margin(max(tr), s["expectClass"], cfg["bands"]))}
        rows.append(row)
        if ca < 90 or cc < 90:
            warns.append(s["id"] + " 准确率偏低")
        if not row["stable"]:
            warns.append(s["id"] + " 结论随噪声漂移")
        if row["marginBp"] <= 0:
            warns.append(s["id"] + " 判定余量不足")
    rep = {"version": 1, "seeds": a.seeds, "scenarios": rows,
           "overallClassAccPct": sum(r["classAccPct"] for r in rows) // len(rows),
           "overallCauseAccPct": sum(r["causeAccPct"] for r in rows) // len(rows), "warnings": warns}
    os.makedirs(a.out, exist_ok=True)
    with open(os.path.join(a.out, "ch13_diagnosis_baseline.json"), "w", encoding="utf-8") as fh:
        json.dump(rep, fh, ensure_ascii=False, indent=2)
    print(json.dumps({k: rep[k] for k in ("seeds", "overallClassAccPct", "overallCauseAccPct", "warnings")}, ensure_ascii=False))
    return 0


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

跑 60 个种子,结论是:

{"seeds": 60, "overallClassAccPct": 100, "overallCauseAccPct": 100, "warnings": []}

准确率 100%、零告警,但真正有用的是每个场景的判定余量(out/ch13_diagnosis_baseline.json):

场景存量趋势区间(bp)距边界余量(bp)
产出侧过载725 ~ 736475
外部注入611 ~ 622361
消耗侧萎缩535 ~ 544285
消耗侧过强-124 ~ -12242
健康基线40 ~ 43120

消耗侧过强 这个场景的余量只有 42 bp:它的存量趋势在 -124 ~ -122 之间,而通缩线在 -80,只差 42 bp 就要掉回「健康」区。这不是 bug,而是提示:通缩线定在 -80 bp/日 时,一个「轻度通缩」的经济会在判定边界附近反复横跳。 处理办法是给判定加迟滞(连续 N 天越线才报警),或者把通缩线往下调。B 层基准的作用就是把这种「边界脆弱性」量化出来,让你在调阈值前就知道要付出多少稳定性代价。

从诊断到动作

这一节交付的是一个能说清「为什么」的诊断器:它不只会说「这条曲线在涨」,还会说「涨在产出侧」「涨在消耗侧」还是「有人在外面注水」。三者处方完全不同,误判的代价是——把通胀治成通缩。下一节给的是处方本身:沉淀机制与回收活动各自能把存量曲线压到多低,以及为什么「只回收不沉淀」根本拦不住。

阅读导航:上一节:参数敏感性与漂移诊断 · 下一节:沉淀机制与回收活动 。

继续阅读

探索更多技术文章

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

全部文章 返回首页

「slg」更多文章

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