「你们这款 SLG 的北极星指标是什么？」——如果答案是一个孤零零的数字，那它大概率撑不过一次版本复盘。北极星指标的价值不在于它是哪个数，而在于它能不能被拆成一组彼此不打架的下级指标：拆得开，团队才知道自己改的东西到底动没动那个数；拆不开，北极星就只是一句口号。
先划边界。站内的游戏商业核心指标详解 是一本指标词典：它回答「DAU、ARPU、留存、付费率分别是什么意思」。本章不重复那本词典，本章回答的是词典不回答的三个工程问题：这些指标怎么被组织成一棵树、每个指标的口径怎么唯一、这些口径怎么被采集与呈现。换句话说，那篇给你词条，本节给你词条之间的父子关系和校验规则。站内的经济系统运营 checklist 给的是产业侧的检查清单，本节给的是可跑出结论的结构校验脚本。
北极星指标不是「一个数」，是一棵可校验的树
把「周活跃领主数」当成北极星，最容易犯的错是把它当成一个原子数字：运营说它涨了、策划说它跌了，两边用的是两套算法。要避免这件事，只有一个办法——把北极星展开成一棵树，并让这棵树满足三条硬约束：
- 父子关系完整：每个下级指标都挂在某个上级下，根节点唯一，没有环，没有悬空子树；
- 口径唯一：任意两个指标的「怎么算」这句话不能一样，否则它们就是同一个指标的两个名字；
- 无重复定义：指标的 id 与中文名都不能撞车。
只要有一条不满足，这棵树就是坏的。好消息是，这三条都可以被一个脚本机械地判定——和第 4 章的系统依赖图 一样，把「共识」变成「可判定的对象」。
《王朝纪元》的指标树
下面这份配置是 D 层（静态层）产物：《王朝纪元》的指标树唯一真源。它本身不产生数字，但规定了哪个指标挂在哪、用什么方式合成、口径怎么写。
{
"version": 1,
"game": "王朝纪元",
"root": "wau",
"note": "指标树唯一真源：内部节点用 combine 声明子节点如何合成自己，叶子节点必须给出 source（埋点事件 id）。口径文本不得重复。",
"metrics": [
{ "id": "wau", "name": "周活跃领主数", "level": "north_star", "parent": null, "kind": "count", "combine": "additive", "owner": "数据", "caliber": "自然周内至少登录 1 次且完成 1 次主循环动作（出征/建造/招募）的去重领主数；按服务器时区 04:00 切天、周一 04:00 切周" },
{ "id": "new_engaged", "name": "新增深度活跃", "level": "l1", "parent": "wau", "kind": "count", "combine": "multiplicative", "owner": "数据", "caliber": "本周新注册且在注册当日完成过至少 1 次出征的领主数" },
{ "id": "new_register", "name": "新增注册领主", "level": "l2", "parent": "new_engaged", "kind": "count", "owner": "运营", "caliber": "本周内首次完成注册流程并进入主城的去重账号数", "source": "account_register_success" },
{ "id": "d1_engage_rate", "name": "注册次日出征率", "level": "l2", "parent": "new_engaged", "kind": "rate", "owner": "数据", "caliber": "注册次日完成至少 1 次出征的领主占当日新增注册领主的比例", "source": "march_start" },
{ "id": "retained_engaged", "name": "留存深度活跃", "level": "l1", "parent": "wau", "kind": "count", "combine": "multiplicative", "owner": "数据", "caliber": "上周活跃、本周仍完成至少 1 次主循环动作的领主数" },
{ "id": "prev_engaged", "name": "上周深度活跃领主", "level": "l2", "parent": "retained_engaged", "kind": "count", "owner": "数据", "caliber": "上一自然周口径下的周活跃领主数，用于本周留存基数", "source": "wau_snapshot" },
{ "id": "w1_retention", "name": "周留存率", "level": "l2", "parent": "retained_engaged", "kind": "rate", "owner": "数据", "caliber": "上周活跃领主中本周仍活跃的比例，分子分母同口径、同去重规则", "source": "wau_snapshot" },
{ "id": "returning_engaged", "name": "回流深度活跃", "level": "l1", "parent": "wau", "kind": "count", "combine": "multiplicative", "owner": "运营", "caliber": "连续沉默满 14 天后在本周重新完成至少 1 次主循环动作的领主数" },
{ "id": "dormant_pool", "name": "沉默领主池", "level": "l2", "parent": "returning_engaged", "kind": "count", "owner": "运营", "caliber": "截至本周一 04:00 已连续沉默满 14 天、尚未流失标记的领主总数", "source": "dormant_snapshot" },
{ "id": "reactivation_rate", "name": "回流率", "level": "l2", "parent": "returning_engaged", "kind": "rate", "owner": "运营", "caliber": "沉默领主池中在本周被唤回并完成至少 1 次主循环动作的比例", "source": "march_start" }
]
}
这棵树只有三层，却把北极星拆成了三个互不重叠的来源：新增、留存、回流。根节点用 additive 声明「我 = 三个子节点之和」；每个 l1 节点用 multiplicative 声明「我 = 基数 × 转化率」——比如「新增深度活跃 = 新增注册领主 × 注册次日出征率」。注意每个叶子都写了一个 source，那是采集用的埋点事件 id（下一节 8.2 会把它们展开成完整的事件 schema）。这一列是整棵树的「接地线」：任何算不出 source 的指标，都不是真指标，只是概念。
四个校验维度，一个脚本
判定这棵树是否合法，就四个维度：根唯一 + 无环可达（父子关系）、口径唯一（归一化后判等）、id 与 name 不重复（无重复定义）、合成合法（内部节点声明 combine、乘法节点必须含比例型子节点、叶子必须有 source）。下面这段是 S 层仿真脚本，读 config/ 里的树，一次算完四个维度，并画出指标树图。
# file: sim/ch08_metric_tree.py
# tier: S
# verify: python3 sim/ch08_metric_tree.py --config config/
"""指标树校验与可视化：读 config/ch08_metric_tree.json，校验父子关系、口径唯一、
无重复定义与合成合法性，产出 JSON + CSV + SVG。纯标准库，零依赖。"""
from __future__ import annotations
import argparse, copy, csv, json, os, re
import xml.etree.ElementTree as ET
VALID_COMBINE, VALID_KIND = {"additive", "multiplicative"}, {"count", "rate"}
def norm(s: str) -> str:
"""口径归一化：折叠空白后判等。'口径唯一'即按归一化文本不得重复。"""
return re.sub(r"\s+", " ", (s or "").strip())
def validate(tree: dict) -> tuple[list[str], dict]:
ms = tree["metrics"]
by_id = {m["id"]: m for m in ms}
errs: list[str] = []
seen: set[str] = set()
for m in ms:
if m["id"] in seen:
errs.append(f"DUPLICATE_ID {m['id']}")
seen.add(m["id"])
names: dict[str, list[str]] = {}
for m in ms:
names.setdefault(m["name"], []).append(m["id"])
for nm, ids in names.items():
if len(ids) > 1:
errs.append(f"DUPLICATE_NAME {nm} -> {ids}")
cal: dict[str, list[str]] = {}
for m in ms:
cal.setdefault(norm(m.get("caliber", "")), []).append(m["id"])
for key, ids in cal.items():
if len(ids) > 1:
errs.append(f"DUPLICATE_CALIBER {ids} -> {key[:20]}…")
roots = [m["id"] for m in ms if m.get("parent") is None]
if len(roots) != 1:
errs.append(f"ROOT_COUNT {roots}")
idset = set(by_id)
for m in ms:
p = m.get("parent")
if p is not None and p not in idset:
errs.append(f"ORPHAN {m['id']} -> {p}")
chain, cur, step = [], m["id"], 0
while cur is not None and step <= len(ms):
if cur in chain:
errs.append(f"CYCLE {m['id']}")
break
chain.append(cur)
cur = (by_id.get(cur) or {}).get("parent")
step += 1
else:
if cur is not None:
errs.append(f"UNREACHABLE {m['id']}")
kids: dict = {}
for m in ms:
kids.setdefault(m.get("parent"), []).append(m["id"])
for m in ms:
ch = kids.get(m["id"], [])
if ch:
if m.get("combine") not in VALID_COMBINE:
errs.append(f"BAD_COMBINE {m['id']}")
kinds = [by_id[k].get("kind") for k in ch if k in by_id]
if m.get("combine") == "multiplicative" and "rate" not in kinds:
errs.append(f"MUL_NO_RATE {m['id']}")
if m.get("combine") == "additive" and "rate" in kinds:
errs.append(f"ADD_HAS_RATE {m['id']}")
else:
if not m.get("source"):
errs.append(f"LEAF_NO_SOURCE {m['id']}")
if m.get("kind") not in VALID_KIND:
errs.append(f"BAD_KIND {m['id']}")
stats = {"nodeCount": len(ms), "leafCount": sum(1 for m in ms if not kids.get(m["id"])),
"internalCount": sum(1 for m in ms if kids.get(m["id"])), "maxDepth": _max_depth(tree),
"root": roots[0] if len(roots) == 1 else None,
"ownerCount": len({m.get("owner") for m in ms})}
return errs, stats
def _paths(tree: dict) -> dict[str, list[str]]:
ms, by_id, out = tree["metrics"], {m["id"]: m for m in tree["metrics"]}, {}
for m in ms:
chain, cur, step = [], m["id"], 0
while cur is not None and step <= len(ms):
chain.append(cur)
cur = (by_id.get(cur) or {}).get("parent")
step += 1
out[m["id"]] = list(reversed(chain))
return out
def _max_depth(tree: dict) -> int:
return max(len(p) - 1 for p in _paths(tree).values())
def _layout(tree: dict):
ms = tree["metrics"]
kids: dict = {}
for m in ms:
kids.setdefault(m.get("parent"), []).append(m["id"])
xs, counter = {}, [0]
def dfs(mid: str) -> None:
ch = kids.get(mid, [])
if not ch:
xs[mid] = counter[0]; counter[0] += 1; return
for k in ch:
dfs(k)
xs[mid] = sum(xs[k] for k in ch) / len(ch)
root = next(m["id"] for m in ms if m.get("parent") is None)
dfs(root)
depth = {m["id"]: len(_paths(tree)[m["id"]]) - 1 for m in ms}
return xs, depth, kids
def render_svg(tree: dict, path: str) -> None:
xs, depth, kids = _layout(tree)
ms, by_id = tree["metrics"], {m["id"]: m for m in tree["metrics"]}
cw, rh, bw, bh = 210, 96, 176, 52
nleaf = sum(1 for m in ms if not kids.get(m["id"]))
W, H = max(600, nleaf * cw + 40), (max(depth.values()) + 1) * rh + 40
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"})
pos = {m["id"]: (20 + xs[m["id"]] * cw + bw / 2, 20 + depth[m["id"]] * rh + bh / 2) for m in ms}
for m in ms:
p = m.get("parent")
if p is None:
continue
(x1, y1), (x2, y2) = pos[p], pos[m["id"]]
ET.SubElement(svg, "line", {"x1": f"{x1:.1f}", "y1": f"{y1 + bh / 2:.1f}", "x2": f"{x2:.1f}",
"y2": f"{y2 - bh / 2:.1f}", "stroke": "#3d4759", "stroke-width": "1.5"})
pal = {"north_star": "#f0b429", "l1": "#4f9cf9", "l2": "#3fbf7f"}
for m in ms:
cx, cy = pos[m["id"]]
ET.SubElement(svg, "rect", {"x": f"{cx - bw / 2:.1f}", "y": f"{cy - bh / 2:.1f}", "width": str(bw),
"height": str(bh), "rx": "8", "fill": "#1a1f2b",
"stroke": pal.get(m.get("level"), "#3fbf7f"), "stroke-width": "2"})
t = ET.SubElement(svg, "text", {"x": f"{cx:.1f}", "y": f"{cy - 4:.1f}", "fill": "#e6ebf5",
"font-family": "sans-serif", "font-size": "13", "text-anchor": "middle"})
t.text = m["name"]
s = ET.SubElement(svg, "text", {"x": f"{cx:.1f}", "y": f"{cy + 14:.1f}", "fill": "#8b97ab",
"font-family": "monospace", "font-size": "10", "text-anchor": "middle"})
s.text = m["id"] if not m.get("combine") else f"{m['combine']} · {m.get('kind')}"
el = ET.ElementTree(svg)
ET.indent(el)
el.write(path, encoding="utf-8", xml_declaration=True)
def write_csv(tree: dict, path: str) -> None:
paths = _paths(tree)
rows = [{"id": m["id"], "name": m["name"], "level": m.get("level", ""), "parent": m.get("parent") or "",
"kind": m.get("kind", ""), "combine": m.get("combine", ""), "owner": m.get("owner", ""),
"source": m.get("source", ""), "depth": len(paths[m["id"]]) - 1,
"path": " / ".join(paths[m["id"]]), "caliber": norm(m.get("caliber", ""))} for m in tree["metrics"]]
rows.sort(key=lambda r: (r["depth"], r["id"]))
with open(path, "w", encoding="utf-8", newline="") as f:
w = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
w.writeheader(); w.writerows(rows)
def mutation_demo(tree: dict) -> dict:
"""故意注入四类缺陷，证明校验器确实会报错（不写入任何配置文件）。"""
bad = copy.deepcopy(tree); ms = bad["metrics"]
ms[1]["id"] = ms[0]["id"] # id 重复
ms[3]["caliber"] = ms[0]["caliber"] # 口径重复
ms[5]["parent"] = "does_not_exist" # 孤儿父节点
ms[0]["parent"] = ms[2]["id"] # 成环
errs, _ = validate(bad)
return {"injected": ["DUPLICATE_ID", "DUPLICATE_CALIBER", "ORPHAN", "CYCLE"],
"caught": sorted({e.split()[0] for e in errs}), "rawCount": len(errs)}
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_metric_tree.json"), encoding="utf-8") as f:
tree = json.load(f)
errs, stats = validate(tree)
report = {"version": tree["version"], "game": tree["game"], "ok": not errs, "errors": errs,
"stats": stats, "mutationDemo": mutation_demo(tree)}
os.makedirs(args.out, exist_ok=True)
with open(os.path.join(args.out, "ch08_metric_tree.json"), "w", encoding="utf-8") as f:
json.dump(report, f, ensure_ascii=False, indent=2)
write_csv(tree, os.path.join(args.out, "ch08_metric_tree.csv"))
render_svg(tree, os.path.join(args.out, "ch08_metric_tree.svg"))
print(json.dumps(report, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()
在本机跑它，输出是这样的（同机相对量级，绝对数值不可移植）：
{"version": 1, "game": "王朝纪元", "ok": true, "errors": [],
"stats": {"nodeCount": 10, "leafCount": 6, "internalCount": 4, "maxDepth": 2,
"root": "wau", "ownerCount": 2},
"mutationDemo": {"injected": ["DUPLICATE_ID", "DUPLICATE_CALIBER", "ORPHAN", "CYCLE"],
"caught": ["BAD_COMBINE", "CYCLE", "DUPLICATE_CALIBER", "DUPLICATE_ID",
"MUL_NO_RATE", "ORPHAN", "ROOT_COUNT"], "rawCount": 15}}
ok: true、errors: [] 说明这棵树通过了全部四个维度：10 个指标、6 个叶子、4 个内部节点、深度 2、根为 wau、归 2 个 owner。脚本还会顺手落出 out/ch08_metric_tree.csv（拍平的指标清单，含每个指标的 path 与 depth）和 out/ch08_metric_tree.svg（指标树图，根节点金色、l1 蓝色、l2 绿色）。
校验器必须能证明自己会报错
一个只会说 ok: true 的校验器是没有说服力的——你怎么知道它不是永远返回 true？所以脚本里塞了一段 mutation_demo：故意把一份副本改坏（id 撞车、口径抄根节点、父节点指向不存在的 id、让根节点指回自己的孙节点成环），再让校验器去判。结果它抓出了 7 类错误、共 15 条：injected 里列的四类全部命中（DUPLICATE_ID / DUPLICATE_CALIBER / ORPHAN / CYCLE），另外还连带发现了 ROOT_COUNT（根不唯一）、BAD_COMBINE（成环后某个内部节点丢了自己的 combine）、MUL_NO_RATE（乘法节点缺比例子节点）。
这就是「口径唯一」这条约束的实战意义。假设两个团队各写了一个指标，一个叫「活跃领主」、一个叫「在线领主」，但口径那句话写的是同一句——DUPLICATE_CALIBER 会立刻把它们标成同一个指标的两个名字。反过来，只要两个指标的口径文本不同，它们就是两个独立的度量。口径不打架，靠的不是约定，是校验。
把它接进回归
和第 3 章的核心循环
一样，这棵树的真正价值不在跑一次，而在每次改指标都跑。把 ok: true 当成一条 CI 断言：任何人往树里加一个新指标，只要忘了写 source、或者和已有指标撞了名、撞了口径，脚本立刻变红。这样「指标口径」就从一份会被改乱的 wiki 文档，变成了一个有守卫的不变量。
下一节，我们把这棵树上每个叶子指向的 source 展开成真正的事件 schema：march_start 到底要带哪些字段、参数是什么类型、命名怎么统一——把「口径」从文字落到可校验的埋点契约上。
阅读导航：上一节：灰度、放量与回滚 · 下一节：埋点规范与口径统一 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。