11.1 指标树与北极星指标

把「有效活跃」定为北极星,向下拆成留存 / 变现 / 增长 / 体验四条驱动、十二条叶子指标,每条边带权重,同层权重和必须为 1000。本节给出一张可被程序消费的指标树配置、一段做一致性校验与加权回算的 S 层脚本,以及一套与 Python 逐位一致的 TS 求值内核,让「北极星现在多少分」从口号变成能跑出来的数字。

站内有一篇 H5 游戏产业全景 ,那是产业综述口径,讲这条赛道整体怎么长、钱从哪儿流;本节给的是一款游戏内部的数据分析体系——同样谈数据,那篇看的是行业大盘,本节看的是《合成小岛》自己的指标怎么组织、怎么算。颗粒度完全不同:那篇回答「这个市场有多大」,本节回答「我这款游戏现在健康吗」。还要先说清边界:第 7 章讲留存漏斗、第 8 章讲 A/B 实验,那两章是「怎么把某个指标做上去」的方法,本章不碰它们,只解决更前置的一环——先把指标本身定义清楚、组织成树、算得出来。指标定义错了,后面所有实验与漏斗都是在错误的地基上盖楼。

北极星不是「最重要」,而是「唯一」

团队里最常听到的一句话是「我们要把留存做上去」。问题在于「留存」不是一个指标,是一族指标:次日、7 日、30 日、次留付费留存……每个人嘴里的「留存」都可能不是同一个数。北极星指标的作用就是先终结这场混战:它是全队在同一时刻盯着的唯一一个数,所有人做的改动都要能解释成「它让这个数往好还是往坏走了」。

选北极星有四条判据:反映用户价值(用户真的得到了好处,而不只是公司收到了钱)、领先于收入(先动它,收入后动,它才有指导性)、可被团队影响(团队一使劲它就该动,否则盯着也没用)、单一可记(一个词能说清,不能是「留存和收入的加权」这种绕口令)。

按这四条,收入是最差的选择——它滞后、被定价与买量搅动、且不直接等于用户价值。《合成小岛》把北极星定为**「有效活跃」:当日登录且**完成至少 1 次合成的去重用户数。它同时满足四条——用户合成了才叫真的玩到了(价值),它领先于付费(先有活跃才有钱),策划改关卡、改引导都能推动它(可影响),一句话说得清(单一)。

指标树:把北极星拆成能动手的叶子

北极星是「我们要去哪」,指标树是「怎么走过去」。树只有三层:L0 北极星、L1 驱动(把北极星拆成几条互相独立的支柱)、L2 叶子(真正能从埋点里直接算出来的数)。《合成小岛》的 L1 是四条——留存 / 变现 / 增长 / 体验。

每条边带一个权重(千分比),表示这个子节点对父节点的贡献占比。规矩只有一条:同一个父节点下,所有子权重之和必须等于 1000‰。这条规矩不是形式主义——它保证任何一个内部节点的分数都能由子节点加权回算出来,不需要单独定义;权重和不为 1000,回算就没有意义。叶子还带两样东西:direction(up 越高越好、down 越低越好,比如「首包加载耗时」是 down)和 event(指向埋点事件名)。叶子必须能映射到埋点——算不出来的指标不配当叶子,那只是愿望。

D 层:把指标树写成配置

下面这份是 D 层(静态层)配置,全整数。17 个节点、12 条叶子,每个节点声明层级、父节点、权重、方向;叶子额外给 raw(当前值)、target(目标值)和 event(埋点事件)。

{
  "version": 1,
  "series": "《小游戏开发实战》",
  "volume": "卷三《上线变现与长线运营》",
  "game": "merge-island",
  "observedAt": "2026-10",
  "note": "指标树配置(D 层)。北极星在根,向下分解为 L1 驱动与 L2 叶子指标;每条边带权重 contributionPermille,同一父节点下子权重之和必须为 1000。叶子给 raw / target / direction,归一到 0–1000 的 scorePermille;内部节点由子节点加权回算。direction: up=越高越好,down=越低越好(归一取 target/raw)。叶子必须带 event 字段,指向 11.2 的埋点事件名。全整数,禁浮点。raw / target 均为示例假设值,真实数值需真实业务环境,命令行不可复现(X)。",
  "units": {
    "scorePermille": "归一分,0–1000",
    "weightPermille": "该节点对父节点的贡献权重,千分比",
    "raw": "叶子指标原始值(单位见 unit)",
    "target": "叶子指标目标值"
  },
  "northStar": "efficient-active",
  "nodes": [
    { "id": "efficient-active", "name": "有效活跃(北极星)", "level": 0, "parent": null, "weightPermille": 1000, "direction": "up" },

    { "id": "retention", "name": "留存", "level": 1, "parent": "efficient-active", "weightPermille": 400, "direction": "up" },
    { "id": "monetization", "name": "变现", "level": 1, "parent": "efficient-active", "weightPermille": 300, "direction": "up" },
    { "id": "growth", "name": "增长", "level": 1, "parent": "efficient-active", "weightPermille": 200, "direction": "up" },
    { "id": "experience", "name": "体验", "level": 1, "parent": "efficient-active", "weightPermille": 100, "direction": "up" },

    { "id": "d1-retention", "name": "次日留存", "level": 2, "parent": "retention", "weightPermille": 500, "direction": "up", "unit": "%", "raw": 38, "target": 45, "event": "session_start" },
    { "id": "d7-retention", "name": "7 日留存", "level": 2, "parent": "retention", "weightPermille": 300, "direction": "up", "unit": "%", "raw": 12, "target": 18, "event": "session_start" },
    { "id": "session-per-dau", "name": "人均日启动", "level": 2, "parent": "retention", "weightPermille": 200, "direction": "up", "unit": "次", "raw": 4, "target": 6, "event": "session_start" },

    { "id": "arpdau", "name": "ARPDAU", "level": 2, "parent": "monetization", "weightPermille": 500, "direction": "up", "unit": "分", "raw": 12, "target": 20, "event": "ad_impression" },
    { "id": "pay-rate", "name": "付费率", "level": 2, "parent": "monetization", "weightPermille": 300, "direction": "up", "unit": "‰", "raw": 18, "target": 30, "event": "iap_purchase" },
    { "id": "ad-imp-per-dau", "name": "人均广告展示", "level": 2, "parent": "monetization", "weightPermille": 200, "direction": "up", "unit": "次", "raw": 7, "target": 10, "event": "ad_impression" },

    { "id": "organic-k", "name": "自然裂变 K 值", "level": 2, "parent": "growth", "weightPermille": 400, "direction": "up", "unit": "‰", "raw": 35, "target": 60, "event": "invite_sent" },
    { "id": "ua-roas", "name": "买量 ROAS", "level": 2, "parent": "growth", "weightPermille": 400, "direction": "up", "unit": "%", "raw": 85, "target": 100, "event": "attribution_install" },
    { "id": "invite-per-user", "name": "人均邀请", "level": 2, "parent": "growth", "weightPermille": 200, "direction": "up", "unit": "‰", "raw": 12, "target": 20, "event": "invite_sent" },

    { "id": "crash-free-rate", "name": "无崩溃会话率", "level": 2, "parent": "experience", "weightPermille": 400, "direction": "up", "unit": "‰", "raw": 996, "target": 1000, "event": "session_end" },
    { "id": "first-load-sec", "name": "首包加载耗时", "level": 2, "parent": "experience", "weightPermille": 300, "direction": "down", "unit": "秒", "raw": 4, "target": 3, "event": "first_frame" },
    { "id": "tutorial-complete", "name": "引导完成率", "level": 2, "parent": "experience", "weightPermille": 300, "direction": "up", "unit": "%", "raw": 78, "target": 90, "event": "tutorial_complete" }
  ]
}

S 层:一致性校验 + 加权回算

配置写完,第一件事不是看数字,而是校验这棵树自不自洽。下面这段是 S 层脚本:先做一致性校验(单根、parent 存在、无环、同父权重和为 1000、方向合法、叶子带 event),再做加权回算——叶子归一到 0–1000 分,内部节点由子节点加权求和,最终得到北极星复合分。产出 JSON + CSV + SVG。

#!/usr/bin/env python3
# file: sim/l11_metric_tree.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/l11_metric_tree.py --config config/ --out out/
"""指标树一致性校验 + 加权回算:读 config/l11_metric_tree.json。

校验:单根且等于 northStar;parent 存在;无环;同父子权重和 = 1000;direction ∈ {up,down};
叶子带 event。回算:叶子归一到 0–1000 分(up: raw*1000//target;down: target*1000//raw,截顶 1000),
内部节点 score = Σ(child.weightPermille * child.score // 1000),得北极星复合分。全整数地板除。

产出:out/l11_metric_tree.json + .csv + .svg。raw / target 为示例假设值,真实数值需真实业务环境,
命令行不可复现(X)。
"""

from __future__ import annotations

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

LEVEL_COLOR = {0: "#111827", 1: "#2563eb", 2: "#10b981"}


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


def normalize(raw: int, target: int, direction: str) -> int:
    if target <= 0 or raw <= 0:
        return 0
    return min(1000, target * 1000 // raw) if direction == "down" else min(1000, raw * 1000 // target)


def validate(cfg: dict) -> list[str]:
    errs, nodes = [], cfg["nodes"]
    by_id = {n["id"]: n for n in nodes}
    kids: dict[str, list] = {}
    for n in nodes:
        if n["parent"] is not None:
            kids.setdefault(n["parent"], []).append(n)
    roots = [n for n in nodes if n["parent"] is None]
    if len(roots) != 1:
        errs.append(f"根节点数={len(roots)},必须恰好 1")
    elif roots[0]["id"] != cfg["northStar"]:
        errs.append(f"根 {roots[0]['id']} != northStar {cfg['northStar']}")
    for n in nodes:
        if n["direction"] not in ("up", "down"):
            errs.append(f"{n['id']}: direction={n['direction']} 非法")
        if n["parent"] is not None and n["parent"] not in by_id:
            errs.append(f"{n['id']}: parent {n['parent']} 不存在")
        if n["id"] not in kids and not n.get("event"):
            errs.append(f"{n['id']}: 叶子缺 event 映射")
    for parent, ks in kids.items():
        s = sum(k["weightPermille"] for k in ks)
        if s != 1000:
            errs.append(f"{parent}: 子权重和={s}‰,必须 1000")
    for n in nodes:  # 无环:向上走不超过节点数
        cur, steps = n, 0
        while cur["parent"] is not None:
            cur = by_id.get(cur["parent"])
            steps += 1
            if cur is None or steps > len(nodes):
                errs.append(f"{n['id']}: 向上遍历出现环或断链")
                break
    return errs


def rollup(cfg: dict) -> dict[str, int]:
    nodes, kids, score = cfg["nodes"], {}, {}
    for n in nodes:
        if n["parent"] is not None:
            kids.setdefault(n["parent"], []).append(n)

    def compute(n: dict) -> int:
        if n["id"] in score:
            return score[n["id"]]
        if n["id"] in kids:
            total = sum(compute(k) * k["weightPermille"] // 1000 for k in kids[n["id"]])
        else:
            total = normalize(n["raw"], n["target"], n["direction"])
        score[n["id"]] = total
        return total

    for n in nodes:
        compute(n)
    return score


def draw_svg(path: str, nodes: list, score: dict, ns_id: str) -> int:
    W, PAD, ROW = 760, 240, 30
    H = 70 + ROW * len(nodes)
    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 = \
        f"《合成小岛》指标树回算(北极星复合分 {score[ns_id]}‰,示例参数)"
    for i, n in enumerate(nodes):
        y = 50 + i * ROW
        ET.SubElement(svg, "text", {"x": str(16 + n["level"] * 14), "y": str(y + 16),
                                    "font-size": "12", "fill": "#374151"}).text = n["name"]
        w = (W - PAD - 80) * score[n["id"]] // 1000
        ET.SubElement(svg, "rect", {"x": str(PAD), "y": str(y + 4), "width": str(max(w, 2)),
                                    "height": "16", "rx": "3", "fill": LEVEL_COLOR.get(n["level"], "#6b7280")})
        ET.SubElement(svg, "text", {"x": str(PAD + max(w, 2) + 8), "y": str(y + 16),
                                    "font-size": "12", "fill": "#6b7280"}).text = str(score[n["id"]])
    os.makedirs(os.path.dirname(path), exist_ok=True)
    ET.ElementTree(svg).write(path, encoding="utf-8", xml_declaration=True)
    return os.path.getsize(path)


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--config", default="config/")
    ap.add_argument("--out", default="out/")
    a = ap.parse_args()
    cfg = load(a.config)
    nodes = cfg["nodes"]
    errs, score = validate(cfg), rollup(cfg)
    leaves = [n for n in nodes if not any(c["parent"] == n["id"] for c in nodes)]
    l1 = [n for n in nodes if n["level"] == 1]
    os.makedirs(a.out, exist_ok=True)
    sb = draw_svg(os.path.join(a.out, "l11_metric_tree.svg"), nodes, score, cfg["northStar"])
    with open(os.path.join(a.out, "l11_metric_tree_nodes.csv"), "w", encoding="utf-8", newline="") as f:
        w = csv.writer(f)
        w.writerow(["id", "name", "level", "parent", "weightPermille", "direction", "raw", "target", "scorePermille", "event"])
        for n in nodes:
            w.writerow([n["id"], n["name"], n["level"], n["parent"] or "-", n["weightPermille"],
                        n["direction"], n.get("raw", ""), n.get("target", ""), score[n["id"]], n.get("event", "")])
    report = {"version": cfg["version"], "game": cfg["game"], "observedAt": cfg["observedAt"],
              "northStar": cfg["northStar"], "nodeCount": len(nodes), "leafCount": len(leaves),
              "northStarScore": score[cfg["northStar"]], "l1Scores": {n["id"]: score[n["id"]] for n in l1},
              "leafScores": {n["id"]: score[n["id"]] for n in leaves}, "violations": errs,
              "valid": not errs, "svgBytes": sb}
    with open(os.path.join(a.out, "l11_metric_tree.json"), "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False, indent=2)
    print(json.dumps({"valid": not errs, "violations": errs, "nodeCount": len(nodes),
                      "northStarScore": score[cfg["northStar"]], "l1Scores": report["l1Scores"],
                      "svgBytes": sb}, ensure_ascii=False, indent=2))
    return 0


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

在本机跑它(假设参数),输出如下(同机相对量级,绝对数值不可移植):

{
  "valid": true,
  "violations": [],
  "nodeCount": 17,
  "northStarScore": 713,
  "l1Scores": {
    "retention": 754,
    "monetization": 620,
    "growth": 693,
    "experience": 882
  },
  "svgBytes": 3754
}

这棵树是自洽的(violations 为空),北极星复合分 713‰。四条驱动里 experience(体验)最高 882,但它权重只有 100‰;monetization(变现)最低 620,权重 300‰。这组数字立刻回答了「该往哪儿使劲」——体验已经很好,问题在变现拖后腿:arpdau 归一只有 600(12 分 / 目标 20 分)、pay-rate 也只有 600。反过来看,就算把 experience 做到满分,对北极星的拉动也有限——因为权重摆在那儿。校验这一步也不能省:把 retention 的子权重从 500 改成 450(和变 950),或把某个 direction 写成 higher,脚本都会立刻报违规。指标树的错误是「静默」的——权重和是 950 时它照样能算出一个数,只是那个数是错的,唯一可靠的办法是让机器来查。

P 层:求值内核

同一份「校验 + 回算」逻辑,真到看板上要每秒重算,就不适合跑 Python 脚本了。下面这段是 P 层 TypeScript 内核,与上面的 Python 同一份逻辑:normalize 归一、validateTree 校验、rollup 回算、northStarScore 取北极星分。全整数地板除,结果与 Python 逐位一致——这正是「一份逻辑、多处消费」在数据层的体现。

// file: core/src/l11/metric_tree.ts
// tier: P
// verify: node --test 'core/test/l11_*.test.ts'
//
// 指标树求值内核:与 S 层 Python 同一份逻辑(一致性校验 + 加权回算),结果逐位一致。
// 叶子归一到 0–1000 分(up: raw*1000/target;down: target*1000/raw,截顶 1000),
// 内部节点 score = Σ(child.weightPermille * child.score / 1000)。全整数地板除(Math.floor)。
// 无 Math.random,只用可擦除语法(禁 enum / namespace / 参数属性 / 装饰器)。

export interface MetricNode {
  readonly id: string; readonly name: string; readonly level: number; readonly parent: string | null;
  readonly weightPermille: number; readonly direction: string;
  readonly raw?: number; readonly target?: number; readonly event?: string;
}
export interface MetricTreeCfg { readonly northStar: string; readonly nodes: readonly MetricNode[]; }
export interface TreeCheck { readonly valid: boolean; readonly violations: readonly string[]; }

/** 归一:up 取 raw/target,down 取 target/raw,均截顶 1000;非法输入返回 0。 */
export function normalize(raw: number, target: number, direction: string): number {
  if (target <= 0 || raw <= 0) return 0;
  if (direction === "down") return Math.min(1000, Math.floor(target * 1000 / raw));
  return Math.min(1000, Math.floor(raw * 1000 / target));
}
function kids(cfg: MetricTreeCfg, id: string): MetricNode[] {
  return cfg.nodes.filter((n) => n.parent === id);
}

/** 一致性校验:单根、parent 存在、无环、同父权重和 1000、direction 合法、叶子带 event。 */
export function validateTree(cfg: MetricTreeCfg): TreeCheck {
  const errs: string[] = [], byId = new Map(cfg.nodes.map((n) => [n.id, n]));
  const roots = cfg.nodes.filter((n) => n.parent === null);
  if (roots.length !== 1) errs.push(`根节点数=${roots.length},必须恰好 1`);
  else if (roots[0].id !== cfg.northStar) errs.push(`根 ${roots[0].id} != northStar ${cfg.northStar}`);
  for (const n of cfg.nodes) {
    if (n.direction !== "up" && n.direction !== "down") errs.push(`${n.id}: direction=${n.direction} 非法`);
    if (n.parent !== null && !byId.has(n.parent)) errs.push(`${n.id}: parent ${n.parent} 不存在`);
    if (kids(cfg, n.id).length === 0 && !n.event) errs.push(`${n.id}: 叶子缺 event 映射`);
  }
  const groups = new Map<string, MetricNode[]>();
  for (const n of cfg.nodes) if (n.parent !== null) groups.set(n.parent, [...(groups.get(n.parent) ?? []), n]);
  for (const [parent, ks] of groups) {
    const s = ks.reduce((a, k) => a + k.weightPermille, 0);
    if (s !== 1000) errs.push(`${parent}: 子权重和=${s}‰,必须 1000`);
  }
  for (const n of cfg.nodes) {
    let cur: MetricNode | undefined = n, steps = 0;
    while (cur && cur.parent !== null) {
      cur = byId.get(cur.parent); steps += 1;
      if (!cur || steps > cfg.nodes.length) { errs.push(`${n.id}: 向上遍历出现环或断链`); break; }
    }
  }
  return { valid: errs.length === 0, violations: errs };
}

/** 加权回算:返回 id -> scorePermille(自底向上,同输入必同输出)。 */
export function rollup(cfg: MetricTreeCfg): Record<string, number> {
  const score: Record<string, number> = {};
  const compute = (n: MetricNode): number => {
    if (score[n.id] !== undefined) return score[n.id];
    const ks = kids(cfg, n.id);
    const total = ks.length > 0
      ? ks.reduce((a, k) => a + Math.floor(compute(k) * k.weightPermille / 1000), 0)
      : normalize(n.raw ?? 0, n.target ?? 0, n.direction);
    score[n.id] = total;
    return total;
  };
  for (const n of cfg.nodes) compute(n);
  return score;
}

/** 北极星复合分。 */
export function northStarScore(cfg: MetricTreeCfg): number {
  return rollup(cfg)[cfg.northStar] ?? 0;
}

配套的 core/test/l11_metric_tree.test.ts 把五件事钉死:normalize 的四种边界、合法配置零违规、三种注入错误被抓出、L1 与北极星分数逐位确定、以及「同配置两次回算结果相同」。跑 node --test 'core/test/l11_*.test.ts',5 个用例全过。注意 P 层与 S 层必须给出一模一样的数——retention 754 / monetization 620 / growth 693 / experience 882、北极星 713,两边差一位就说明有一边写错了。

小结与下一步

这一节把指标树做成了一份可校验、可回算的配置:北极星唯一、L1 四条驱动、L2 十二条叶子,权重自洽、叶子落到埋点,Python 与 TS 两处算出一模一样的 713‰。三件事记住就够——**北极星要唯一且领先于收入;同父权重和必须等于 1000‰;叶子必须能落到埋点,而权重和错误是静默的,只能靠脚本抓。**下一节解决「叶子怎么落到埋点」:事件怎么命名、参数怎么定类型、口径怎么统一。

阅读导航:上一节:目录 · 下一节:11.2 埋点规范与口径统一 。

继续阅读

探索更多技术文章

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

全部文章 返回首页

「minigame」更多文章

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