指标树（8.1）规定了算什么，埋点 schema（8.2）规定了采什么。最后一步是把这些数字摆上看板——但看板真正的价值不是那条漂亮的曲线，而是给曲线装上闸门：什么时候该有人被叫醒。这一节把「告警」这件事从一句「跌了就报警」变成一套可仿真、可量化的规则。
先划边界。站内的游戏商业核心指标详解 告诉你 DAU、留存、付费率分别是什么；经济系统运营 checklist 给的是产业侧的监控清单。本节不重复它们，本节回答的是清单里那句最含糊的话：「跌到多少才算跌」——用可跑的规则与基准把这个阈值算出来。
看板不是画曲线，是给曲线装闸门
一条 DAU 曲线摆在那里，人眼能看出的只有「大起大落」。但真实的告警要在变化还很小的时候就发出信号，同时又不能被日常噪声吵醒。这两件事互相拉扯，靠拍脑袋定阈值必然顾此失彼。解法是把告警拆成三类可计算的规则：
- 阈值（threshold）：绝对水位跌破某条线，比如 DAU 低于基线 90%；
- 环比（wow）：跟上一周期比，比如比上周同星期下滑超过 12%；
- 同比（yoy）：跟去年同期比，比如比 52 周前同星期下滑超过 12%。
三类规则各管一种场景：阈值管「跌破生死线」，环比管「突然掉」，同比管「今年不如去年」。再加一个 streak（连续天数）——条件必须连续成立 N 天才报警——用来把单日噪声挡在门外。
一份告警规则配置
下面这份配置是 D 层（静态层）产物：《王朝纪元》的告警规则唯一真源。它声明了时序怎么造（趋势、星期形状、噪声、事故）以及每条规则怎么判。
{
"version": 1,
"game": "王朝纪元",
"days": 420,
"seed": 20261012,
"note": "看板告警规则唯一真源：时序由确定性 PRNG(mulberry32) + 整数基点的趋势/星期/噪声合成，规则支持阈值(threshold)、环比(wow, 滞后 7 天)、同比(yoy, 滞后 364 天)。streak 表示条件需连续成立的天数，用于过滤单日噪声。",
"weekdayShapeBp": [50, 0, -30, -40, -20, 40, 100],
"metrics": [
{ "id": "dau", "name": "日活跃领主数", "base": 120000, "trendPerDayBp": 2, "noiseBp": 400,
"incidents": [
{ "fromDay": 370, "toDay": 376, "dropBp": 2200, "label": "赛季末期集中流失" },
{ "fromDay": 200, "toDay": 200, "dropBp": 1800, "label": "单日服务故障" }
] }
],
"rules": [
{ "id": "dau_floor", "metric": "dau", "kind": "threshold", "direction": "below", "floorRatioBp": 9000, "streak": 2, "level": "P2", "desc": "DAU 跌破基线 90% 且连续 2 天" },
{ "id": "dau_wow", "metric": "dau", "kind": "wow", "direction": "below", "pctBp": 1200, "streak": 2, "level": "P1", "desc": "DAU 环比上周同星期下滑超 12% 且连续 2 天" },
{ "id": "dau_yoy", "metric": "dau", "kind": "yoy", "direction": "below", "pctBp": 1200, "streak": 1, "level": "P1", "desc": "DAU 同比 52 周前同星期下滑超 12%" }
]
}
这里有两个工程决定值得说清楚。第一，所有比例都用整数基点（万分之一）：趋势 trendPerDayBp: 2 就是每天涨 0.02%，噪声 noiseBp: 400 就是 ±4%，这样时序的生成参数全程是整数，可复现。第二，时序用自实现的确定性 PRNG（mulberry32），不依赖任何外部库——同一个 seed 永远产出同一条曲线，这与附录 B
的整数口径一脉相承。
S 层：造时序 + 判告警
下面这段是 S 层仿真脚本：用 mulberry32 造 420 天的 DAU 时序（趋势 × 星期形状 × 噪声，再叠加两段事故），用三类规则逐日判定，产出 CSV（逐日）、SVG（曲线与告警点）和 JSON（告警集）。脚本里的 mulberry32 与 Node 侧逐位一致（前 6 个随机数在 Python 与 Node 上完全相同）。
# file: sim/ch08_alerting.py
# tier: S
# verify: python3 sim/ch08_alerting.py --config config/
"""看板告警规则仿真：造确定性时序（mulberry32 自实现 PRNG，零依赖），用阈值 / 环比 /
同比三类规则判定告警，产出 CSV（逐日）+ SVG（曲线与告警点）+ JSON（告警集）。
时序模型：value[d] = base × 趋势^d × (1 + 星期形状 + 噪声) × 事故衰减。"""
from __future__ import annotations
import argparse, csv, json, os
import xml.etree.ElementTree as ET
def mulberry32(seed: int):
"""与 Node 侧逐位一致的确定性 PRNG，返回 [0,1) 均匀分布。"""
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_series(m: dict, days: int, seed: int, shape: list[int], with_incidents: bool = True) -> list[int]:
rnd, trend, noise, base = mulberry32(seed), 1.0 + m["trendPerDayBp"] / 10000.0, m["noiseBp"] / 10000.0, float(m["base"])
incs = m.get("incidents", []) if with_incidents else []
out = []
for d in range(days):
seasonal = shape[d % len(shape)] / 10000.0
v = base * (trend ** d) * (1.0 + seasonal + (rnd() * 2.0 - 1.0) * noise)
for inc in incs:
if inc["fromDay"] <= d <= inc["toDay"]:
v *= 1.0 - inc["dropBp"] / 10000.0
out.append(int(round(v)))
return out
def pct_change_bp(cur: int, prev: int) -> int:
"""(cur - prev) / prev，单位基点（万分之一），整数地板除。"""
return 0 if prev == 0 else (cur - prev) * 10000 // prev
def condition(rule: dict, series: list[int], base: int, d: int) -> bool:
if rule["kind"] == "threshold":
return series[d] < base * rule["floorRatioBp"] // 10000
lag = 7 if rule["kind"] == "wow" else 364
return d >= lag and pct_change_bp(series[d], series[d - lag]) < -rule["pctBp"]
def eval_rule(rule: dict, series: list[int], base: int) -> list[bool]:
"""streak 表示条件需连续成立的天数，用于过滤单日噪声。"""
streak = rule.get("streak", 1)
cond = [condition(rule, series, base, d) for d in range(len(series))]
return [bool(cond[d]) and d >= streak - 1 and all(cond[d - i] for i in range(streak)) for d in range(len(series))]
def episodes(firing: list[bool]) -> list[dict]:
eps, start = [], None
for d, f in enumerate(firing + [False]):
if f and start is None:
start = d
if not f and start is not None:
eps.append({"from": start, "to": d - 1, "len": d - start}); start = None
return eps
def run(cfg: dict, with_incidents: bool = True) -> dict:
days, shape = cfg["days"], cfg["weekdayShapeBp"]
metrics = {m["id"]: m for m in cfg["metrics"]}
series = {mid: gen_series(m, days, cfg["seed"], shape, with_incidents) for mid, m in metrics.items()}
rules = []
for r in cfg["rules"]:
firing = eval_rule(r, series[r["metric"]], metrics[r["metric"]]["base"])
rules.append({"id": r["id"], "metric": r["metric"], "kind": r["kind"], "level": r.get("level", "P2"),
"streak": r.get("streak", 1), "firingDays": sum(firing),
"episodes": episodes(firing), "firing": firing})
return {"days": days, "series": series, "rules": rules}
def write_csv(cfg: dict, res: dict, path: str) -> None:
mid = cfg["metrics"][0]["id"]; s = res["series"][mid]
flags = {r["id"]: r["firing"] for r in res["rules"]}
floor = cfg["metrics"][0]["base"] * cfg["rules"][0]["floorRatioBp"] // 10000
with open(path, "w", encoding="utf-8", newline="") as f:
w = csv.writer(f)
w.writerow(["day", "value", "floor", "wow_pctBp", "yoy_pctBp"] + [f"{r}_fire" for r in flags])
for d in range(res["days"]):
wow = pct_change_bp(s[d], s[d - 7]) if d >= 7 else ""
yoy = pct_change_bp(s[d], s[d - 364]) if d >= 364 else ""
w.writerow([d, s[d], floor, wow, yoy] + [int(flags[r][d]) for r in flags])
def render_svg(cfg: dict, res: dict, path: str) -> None:
s, days = res["series"][cfg["metrics"][0]["id"]], res["days"]
W, H, pl, pb, pt = 1000, 420, 60, 40, 30
pw, ph = W - pl - 20, H - pb - pt
lo, hi = min(s), max(s); span = max(1, hi - lo)
floor = cfg["metrics"][0]["base"] * cfg["rules"][0]["floorRatioBp"] // 10000
px = lambda d: pl + d / (days - 1) * pw
py = lambda v: pt + (hi - v) / span * ph
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": "#0f1115"})
ET.SubElement(svg, "line", {"x1": str(pl), "y1": f"{py(floor):.1f}", "x2": str(W - 20), "y2": f"{py(floor):.1f}",
"stroke": "#f0b429", "stroke-width": "1", "stroke-dasharray": "6 4"})
ET.SubElement(svg, "polyline", {"points": " ".join(f"{px(d):.1f},{py(s[d]):.1f}" for d in range(days)),
"fill": "none", "stroke": "#4f9cf9", "stroke-width": "1.6"})
colors = {"threshold": "#ef4444", "wow": "#f97316", "yoy": "#eab308"}
for r in res["rules"]:
for d, f in enumerate(r["firing"]):
if f:
ET.SubElement(svg, "circle", {"cx": f"{px(d):.1f}", "cy": f"{py(s[d]):.1f}", "r": "2.6",
"fill": colors.get(r["kind"], "#ef4444")})
for inc in cfg["metrics"][0].get("incidents", []):
ET.SubElement(svg, "rect", {"x": f"{px(inc['fromDay']):.1f}", "y": str(pt),
"width": f"{px(inc['toDay']) - px(inc['fromDay']) + 3:.1f}", "height": str(ph),
"fill": "#ef4444", "opacity": "0.08"})
lx = pl
for label, color in [("序列", "#4f9cf9"), ("地板", "#f0b429"), ("阈值", "#ef4444"), ("环比", "#f97316"), ("同比", "#eab308")]:
ET.SubElement(svg, "rect", {"x": str(lx), "y": "8", "width": "10", "height": "10", "fill": color})
t = ET.SubElement(svg, "text", {"x": str(lx + 14), "y": "17", "fill": "#c8d2e0", "font-family": "sans-serif", "font-size": "11"})
t.text = label; lx += 72
ET.indent(ET.ElementTree(svg))
ET.ElementTree(svg).write(path, encoding="utf-8", xml_declaration=True)
def main() -> None:
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, "ch08_alerting.json"), encoding="utf-8") as f:
cfg = json.load(f)
res = run(cfg, with_incidents=True)
report = {"version": cfg["version"], "days": cfg["days"], "seed": cfg["seed"],
"incidents": cfg["metrics"][0].get("incidents", []),
"rules": [{k: v for k, v in r.items() if k != "firing"} for r in res["rules"]]}
os.makedirs(args.out, exist_ok=True)
with open(os.path.join(args.out, "ch08_alerting.json"), "w", encoding="utf-8") as f:
json.dump(report, f, ensure_ascii=False, indent=2)
write_csv(cfg, res, os.path.join(args.out, "ch08_alerting.csv"))
render_svg(cfg, res, os.path.join(args.out, "ch08_alerting.svg"))
print(json.dumps(report, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()
本机跑它，三条规则各抓到什么（同机相对量级，绝对数值不可移植）：
| 规则 | 类型 | 触发天数 | 告警段 | 说明 |
|---|---|---|---|---|
dau_floor | 阈值 | 6 | 371–376 | 跌破基线 90%（108000） |
dau_wow | 环比 | 6 | 371–376 | 比上周同星期低超 12% |
dau_yoy | 同比 | 7 | 370–376 | 比 52 周前同星期低超 12% |
三条规则都锁定了第 370–376 天那段「赛季末期集中流失」。有意思的是 dau_yoy 比另外两条早一天报警：环比规则要等到事故第二天才能和「上周同一天」比出落差，而同比规则在事故首日当天就比出了下滑。
单日噪声为什么没触发
配置里还埋了一段第 200 天的单日故障（跌 18%）。从 CSV 看，那天的值 103770 确实跌破了地板 108000——按条件它是成立的。但 dau_floor 和 dau_wow 的 streak 都是 2，要求连续两天成立，于是这一天被挡住了，没有报警。这正是 streak 的取舍：它滤掉了单日噪声，代价是延迟一天——对集中流失这种持续性事故无所谓，但对真正的秒级故障，你需要的是另一条更灵敏的规则。「阈值定多少」和「连续几天算数」是两个独立的设计旋钮，本节把它们拆开，就是为了让你能分别调。
B 层：这些规则到底会不会乱叫
一条规则「看起来合理」不算数，得看它在大量随机种子上的表现。下面这段是 B 层基准脚本，复用 ch08_alerting 的时序生成与判定，跑 200 个种子，量化三件事：无事故时的误报、有事故时的检出率、检出延迟。它是软告警——指标越界只打印提示，不返回非零退出码。
# file: sim/ch08_alert_baseline.py
# tier: B
# verify: python3 sim/ch08_alert_baseline.py --config config/ --seeds 200
"""告警规则的稳态基准（B 层，软告警不硬失败）。复用 ch08_alerting 的时序生成与规则
判定，跑一批随机种子，量化三件事：无事故时的误报、有事故时的检出率、检出延迟。
产出 JSON；越界只打印软告警，不返回非零退出码。"""
from __future__ import annotations
import argparse, json, os, sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import ch08_alerting as A # noqa: E402
def first_firing(firing: list[bool]) -> int | None:
return next((d for d, f in enumerate(firing) if f), None)
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--config", default="config/")
ap.add_argument("--out", default="out/")
ap.add_argument("--seeds", type=int, default=200)
args = ap.parse_args()
with open(os.path.join(args.config, "ch08_alerting.json"), encoding="utf-8") as f:
cfg = json.load(f)
metric, days, shape = cfg["metrics"][0], cfg["days"], cfg["weekdayShapeBp"]
main_inc = max(metric["incidents"], key=lambda i: i["dropBp"]) # 主事故 = 跌幅最大的一段
start, end = main_inc["fromDay"], main_inc["toDay"]
agg = {r["id"]: {"kind": r["kind"], "falseEpisodes": 0, "seedsWithFP": 0, "detected": 0,
"latSum": 0, "latN": 0} for r in cfg["rules"]}
for s in range(args.seeds):
seed = cfg["seed"] + s * 7919
clean = A.gen_series(metric, days, seed, shape, with_incidents=False)
inc = A.gen_series(metric, days, seed, shape, with_incidents=True)
for rule in cfg["rules"]:
a = agg[rule["id"]]
fp = A.eval_rule(rule, clean, metric["base"])
a["falseEpisodes"] += len(A.episodes(fp))
a["seedsWithFP"] += int(any(fp))
hits = [d for d, f in enumerate(A.eval_rule(rule, inc, metric["base"])) if f and start - 1 <= d <= end + 2]
if hits:
a["detected"] += 1
a["latSum"] += hits[0] - start
a["latN"] += 1
n = args.seeds
report = {"seeds": n, "incidentStart": start, "incidentEnd": end, "rules": []}
soft = []
for rid, a in agg.items():
far, det = a["seedsWithFP"] / n, a["detected"] / n
report["rules"].append({"id": rid, "kind": a["kind"],
"meanFalseEpisodesPerSeed": round(a["falseEpisodes"] / n, 4),
"falseAlarmRate": round(far, 4), "detectionRate": round(det, 4),
"meanLatencyDays": None if not a["latN"] else round(a["latSum"] / a["latN"], 2)})
if far > 0.05:
soft.append(f"SOFT 误报率偏高 {rid}={far:.3f}")
if det < 0.95:
soft.append(f"SOFT 检出率偏低 {rid}={det:.3f}")
report["softWarnings"] = soft
os.makedirs(args.out, exist_ok=True)
with open(os.path.join(args.out, "ch08_alert_baseline.json"), "w", encoding="utf-8") as f:
json.dump(report, f, ensure_ascii=False, indent=2)
print(json.dumps(report, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()
跑 200 个种子，结果是这样的：
| 规则 | 误报率 | 检出率 | 平均检出延迟（天） |
|---|---|---|---|
dau_floor | 0.0 | 1.0 | 1.0 |
dau_wow | 0.0 | 1.0 | 1.0 |
dau_yoy | 0.0 | 1.0 | 0.07 |
三条规则在 200 个种子上零误报（噪声 ±4% 够不到 12% 的环比/同比门槛），集中流失100% 被检出，延迟 1 天（streak 的代价）到 0 天。这才是「阈值定多少」的答案：不是拍一个数，而是让规则在一批种子上同时满足低误报与高检出。如果哪天有人把 noiseBp 调大到 1200（±12%），误报率会立刻上升，这个基准脚本会在 softWarnings 里直接喊出来。
三节合起来是一条链
8.1 定义算什么（指标树 + 口径唯一），8.2 定义采什么（事件 schema + 命名规范），8.3 定义什么时候响（阈值 / 环比 / 同比 + 基准）。三节共用同一套 bastion 配置口径：指标树的每个叶子指向一个事件，事件的字段决定看板的每个数，看板的每条规则决定谁被叫醒。下一节开始进入本卷的第二部分——经济系统，先给资源和货币分层，那是所有经济闭环的地基。
阅读导航：上一节：埋点规范与口径统一 · 下一节：资源分类与货币分层 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。