7.2 内容产能与排期

上一节定下了版本列车的到站日期,本节回答「能不能做完」。用每个任务的人天估计与各职能每日产能,算出确定性完成日与瓶颈职能,再用蒙特卡洛把点估计变成区间,量化排期的系统性乐观偏差,并用《王朝纪元》版本的真实跑数,给出留缓冲与砍需求两类可执行决策。

7.2 内容产能与排期

上一节把「什么时候发」钉成了版本列车。列车到站日期一旦确定,下一个问题就变成「能不能按期做完」。产业侧那篇 SLG 经济系统运营 2026 清单 把「内容排期」列为运营动作之一,但它停在 checklist 层面;本节把它做成模型——用人天需求与每日产能算完成日,再用蒙特卡洛把「一个数」变成「一个区间」。人天估计本身怎么来(拆解、类比、三点估计)是项目管理常识,这里不展开,重点是把估计喂进可跑的产能模型。

需求、产能与瓶颈

一个版本的内容 backlog 可以拆成若干任务,每个任务给出各职能(策划 design、美术 art、客户端 client、服务端 server、测试 qa)的人天估计。三个量决定了排期:

  • 需求:某职能的总人天 = 该职能下所有任务估计之和。
  • 每日产能:headcount × availabilityPct // 100(人天/天)。availabilityPct 是有效工时占比,扣掉会议、支持、请假。
  • 完成日:ceil(需求 / 每日产能)。取所有职能完成日的最大值即瓶颈——排期由最慢的职能决定,跟 12 章闭环里的「限速步」是同一个道理。

这套算术之所以要写成脚本,是因为它要反复重算:需求一改、有人休假、加一个任务,瓶颈可能从策划跳到服务端,人工算容易漏。

把 backlog 写成配置

下面这段是 D 层配置节选(完整文件 config/ch07_capacity.json),给出计划窗口与五个职能的产能,backlog 里每个任务是一组人天估计。注意《王朝纪元》这版刻意把策划需求堆到瓶颈位:

<!-- snippet: config/ch07_capacity.json(节选) -->
{
  "plan": { "startDay": 0, "launchDay": 42, "variancePct": 40 },
  "capacity": [
    { "discipline": "design", "headcount": 3, "availabilityPct": 80 },
    { "discipline": "art", "headcount": 4, "availabilityPct": 75 },
    { "discipline": "server", "headcount": 2, "availabilityPct": 70 }
  ],
  "backlog": [
    { "id": "T-103", "title": "联盟远征玩法",
      "estimates": { "design": 12, "art": 8, "client": 10, "server": 8, "qa": 6 } }
  ]
}

variancePct 是留给下一段蒙特卡洛的估计波动幅度;确定性排期先不看它。

确定性排期:先算点估计

下面这段是 S 层仿真脚本(完整文件 sim/ch07_capacity.py),把每个职能的完成日算出来,取瓶颈,判定是否赶得上 launchDay,并产出 JSON、燃尽 CSV 与 SVG。

# file: sim/ch07_capacity.py
# tier: S
# verify: python3 sim/ch07_capacity.py --config config/
"""内容产能与排期:人天需求 vs 每日产能,判定能否按期交付。

读 config/ch07_capacity.json。每日产能 = headcount × availabilityPct // 100(人天)。
完成日 = ceil(需求 / 每日产能),全程整数;瓶颈 = 完成日最大的职能。
"""

from __future__ import annotations

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

HERE = os.path.dirname(os.path.abspath(__file__))
OUT = os.path.join(HERE, "..", "out")


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


def daily_capacity(entry: dict) -> int:
    return entry["headcount"] * entry["availabilityPct"] // 100


def demand_per_discipline(cfg: dict) -> dict[str, int]:
    demand: dict[str, int] = {c["discipline"]: 0 for c in cfg["capacity"]}
    for task in cfg["backlog"]:
        for d, md in task["estimates"].items():
            demand[d] = demand.get(d, 0) + md
    return demand


def plan(cfg: dict) -> dict:
    launch = cfg["plan"]["launchDay"]
    demand = demand_per_discipline(cfg)
    rows = []
    for c in cfg["capacity"]:
        d = c["discipline"]
        cap = daily_capacity(c)
        need = demand.get(d, 0)
        days = (need + cap - 1) // cap if cap > 0 else 10 ** 9
        rows.append({"discipline": d, "demand": need, "dailyCapacity": cap,
                     "completionDay": days, "slackDays": launch - days})
    bottleneck = max(rows, key=lambda r: r["completionDay"])
    verdict = "OK" if bottleneck["completionDay"] <= launch else "FAIL"
    return {
        "launchDay": launch,
        "totalManDays": sum(r["demand"] for r in rows),
        "perDiscipline": rows,
        "bottleneck": bottleneck["discipline"],
        "bottleneckCompletionDay": bottleneck["completionDay"],
        "verdict": verdict,
        "shortByDays": max(0, bottleneck["completionDay"] - launch),
    }


def write_csv(cfg: dict, rep: dict, path: str) -> None:
    launch = rep["launchDay"]
    with open(path, "w", encoding="utf-8") as f:
        f.write("day," + ",".join(f"remaining_{r['discipline']}" for r in rep["perDiscipline"]) + "\n")
        for day in range(0, launch + 1):
            cells = []
            for r in rep["perDiscipline"]:
                remaining = max(0, r["demand"] - day * r["dailyCapacity"])
                cells.append(str(remaining))
            f.write(f"{day}," + ",".join(cells) + "\n")


def write_svg(rep: dict, path: str) -> None:
    W, H = 720, 320
    PAD_L, PAD_R, PAD_T, PAD_B = 56, 24, 24, 40
    launch = rep["launchDay"]
    bn = next(r for r in rep["perDiscipline"] if r["discipline"] == rep["bottleneck"])
    hi = max(r["demand"] for r in rep["perDiscipline"])
    inner_w, inner_h = W - PAD_L - PAD_R, H - PAD_T - PAD_B

    def x_of(day: int) -> float:
        return PAD_L + inner_w * day / launch

    def y_of(v: int) -> float:
        return PAD_T + inner_h * (1 - v / hi)

    svg = ET.Element("svg", {"xmlns": "http://www.w3.org/2000/svg",
                             "viewBox": f"0 0 {W} {H}", "width": str(W), "height": str(H)})
    ET.SubElement(svg, "rect", {"x": "0", "y": "0", "width": str(W), "height": str(H), "fill": "#ffffff"})
    ET.SubElement(svg, "line", {"x1": str(PAD_L), "y1": str(PAD_T), "x2": str(PAD_L), "y2": str(H - PAD_B),
                                "stroke": "#333", "stroke-width": "1"})
    ET.SubElement(svg, "line", {"x1": str(PAD_L), "y1": str(H - PAD_B), "x2": str(W - PAD_R),
                                "y2": str(H - PAD_B), "stroke": "#333", "stroke-width": "1"})
    colors = {"design": "#1f6feb", "art": "#d29922", "client": "#3fb950",
              "server": "#d1242f", "qa": "#8250df"}
    for r in rep["perDiscipline"]:
        pts = []
        for day in range(0, launch + 1):
            remaining = max(0, r["demand"] - day * r["dailyCapacity"])
            pts.append((x_of(day), y_of(remaining)))
        ET.SubElement(svg, "polyline", {"points": " ".join(f"{x:.1f},{y:.1f}" for x, y in pts),
                                        "fill": "none", "stroke": colors.get(r["discipline"], "#555"),
                                        "stroke-width": "2" if r["discipline"] == rep["bottleneck"] else "1"})
    # 交付线
    xl = x_of(launch)
    ET.SubElement(svg, "line", {"x1": f"{xl:.1f}", "y1": str(PAD_T), "x2": f"{xl:.1f}",
                                "y2": str(H - PAD_B), "stroke": "#d1242f", "stroke-dasharray": "4 3"})
    t = ET.SubElement(svg, "text", {"x": f"{xl - 4:.1f}", "y": str(PAD_T + 12),
                                    "font-size": "11", "text-anchor": "end", "fill": "#d1242f"})
    t.text = f"launch D{launch}"
    ttl = ET.SubElement(svg, "text", {"x": str(PAD_L), "y": "16", "font-size": "13", "fill": "#111"})
    ttl.text = f"内容燃尽曲线(瓶颈={rep['bottleneck']},完成 D{rep['bottleneckCompletionDay']})"
    os.makedirs(os.path.dirname(path), exist_ok=True)
    ET.ElementTree(svg).write(path, encoding="utf-8", xml_declaration=True)


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--config", default=os.path.join(HERE, "..", "config"))
    args = ap.parse_args()
    cfg = load_config(args.config, "ch07_capacity")
    rep = plan(cfg)
    os.makedirs(OUT, exist_ok=True)
    with open(os.path.join(OUT, "ch07_capacity_report.json"), "w", encoding="utf-8") as f:
        json.dump(rep, f, ensure_ascii=False, indent=2)
    write_csv(cfg, rep, os.path.join(OUT, "ch07_capacity_burndown.csv"))
    write_svg(rep, os.path.join(OUT, "ch07_capacity.svg"))

    print(f"verdict={rep['verdict']} launch=D{rep['launchDay']} bottleneck={rep['bottleneck']} "
          f"completion=D{rep['bottleneckCompletionDay']} shortBy={rep['shortByDays']} totalManDays={rep['totalManDays']}")
    for r in rep["perDiscipline"]:
        print(f"  {r['discipline']:<7} demand={r['demand']:>3} cap/day={r['dailyCapacity']} "
              f"completion=D{r['completionDay']} slack={r['slackDays']}")
    return 0 if rep["verdict"] == "OK" else 2


if __name__ == "__main__":
    sys.exit(main())

跑一遍,design 职能需求 84 人天、每日产能 3×80//100=2,完成日 ceil(84/2)=42,正好卡在 launchDay=42,slack=0:

$ python3 sim/ch07_capacity.py --config config/
verdict=OK launch=D42 bottleneck=design completion=D42 shortBy=0 totalManDays=324
  design  demand= 84 cap/day=2 completion=D42 slack=0
  art     demand= 96 cap/day=3 completion=D32 slack=10
  client  demand= 66 cap/day=2 completion=D33 slack=9
  server  demand= 38 cap/day=1 completion=D38 slack=4
  qa      demand= 40 cap/day=1 completion=D40 slack=2

判定是 OK,但 slack=0 意味着一天都不能出意外。这正是点估计最危险的地方:它把「估计值 = 真实值」当成前提。

点估计的乐观偏差:蒙特卡洛基准

人天估计从来不是确定的数。下面这段是 B 层基准脚本(完整文件 sim/ch07_capacity_bench.py),把每个任务的人天按 variancePct 做整数抖动,重复 trials 次,得到瓶颈完成日的分布,与点估计对照。它用固定种子的 LCG,结果可复现;B 层只出 JSON 与软告警,不硬失败。

# file: sim/ch07_capacity_bench.py
# tier: B
# verify: python3 sim/ch07_capacity_bench.py --config config/ --trials 2000
"""内容产能的蒙特卡洛基准(B 层,软告警不硬失败)。

把每个任务的人天按 variancePct 做整数抖动,重复 trials 次,得到瓶颈完成日的分布,
与确定性点估计对照,量化「乐观偏差」。LCG 固定种子,结果可复现。
"""

from __future__ import annotations

import argparse
import json
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from ch07_capacity import load_config, daily_capacity, demand_per_discipline, plan  # noqa: E402

OUT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "out")


def lcg(seed: int):
    state = seed & 0xFFFFFFFF

    def nxt() -> int:
        nonlocal state
        state = (state * 1103515245 + 12345) % 2147483648
        return state

    return nxt


def percentile(sorted_vals: list[int], p: int) -> int:
    if not sorted_vals:
        return 0
    idx = (p * (len(sorted_vals) - 1)) // 100
    return sorted_vals[idx]


def run(cfg: dict, trials: int, seed: int) -> dict:
    launch = cfg["plan"]["launchDay"]
    var = cfg["plan"]["variancePct"]
    caps = {c["discipline"]: daily_capacity(c) for c in cfg["capacity"]}
    disciplines = list(caps)
    rng = lcg(seed)
    days = []
    for _ in range(trials):
        sums = {d: 0 for d in disciplines}
        for task in cfg["backlog"]:
            for d, md in task["estimates"].items():
                jitter = rng() % (var + 1)          # [0, var]
                sums[d] += md * (100 + jitter) // 100
        completion = max((sums[d] + caps[d] - 1) // caps[d] for d in disciplines)
        days.append(completion)
    days.sort()
    det = plan(cfg)
    miss = sum(1 for d in days if d > launch)
    return {
        "trials": trials, "seed": seed, "variancePct": var,
        "deterministicCompletionDay": det["bottleneckCompletionDay"],
        "p10": percentile(days, 10), "p50": percentile(days, 50),
        "p80": percentile(days, 80), "p95": percentile(days, 95),
        "meanDay": sum(days) // len(days),
        "minDay": days[0], "maxDay": days[-1],
        "biasDays": percentile(days, 50) - det["bottleneckCompletionDay"],
        "missProbabilityPct": miss * 100 // trials,
        "launchDay": launch,
    }


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--config", default=os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "config"))
    ap.add_argument("--trials", type=int, default=2000)
    ap.add_argument("--seed", type=int, default=20261012)
    args = ap.parse_args()

    cfg = load_config(args.config, "ch07_capacity")
    rep = run(cfg, args.trials, args.seed)
    os.makedirs(OUT, exist_ok=True)
    with open(os.path.join(OUT, "ch07_capacity_bench.json"), "w", encoding="utf-8") as f:
        json.dump(rep, f, ensure_ascii=False, indent=2)

    print(f"trials={rep['trials']} seed={rep['seed']} variance={rep['variancePct']}%")
    print(f"deterministic=D{rep['deterministicCompletionDay']}  p10=D{rep['p10']} p50=D{rep['p50']} "
          f"p80=D{rep['p80']} p95=D{rep['p95']} mean=D{rep['meanDay']}")
    print(f"bias(P50-det)=+{rep['biasDays']}d  P(overrun>D{rep['launchDay']})={rep['missProbabilityPct']}%")
    if rep["biasDays"] > 0:
        print(f"WARN 点估计乐观 {rep['biasDays']} 天:确定性排期未计入估计方差")
    return 0


if __name__ == "__main__":
    sys.exit(main())

2000 次试验的输出(同机相对量级):

$ python3 sim/ch07_capacity_bench.py --config config/ --trials 2000
trials=2000 seed=20261012 variance=40%
deterministic=D42  p10=D46 p50=D48 p80=D50 p95=D51 mean=D48
bias(P50-det)=+6d  P(overrun>D42)=100%
WARN 点估计乐观 6 天:确定性排期未计入估计方差

结论很刺眼:确定性排期说「刚好赶上 D42」,蒙特卡洛却说中位数完成日在 D48,P(超期)=100%。原因是双重的——每个任务的人天被向上抖动,而瓶颈又是若干职能的最大值,max 运算对右尾尤其敏感。这不是模型太悲观,而是点估计太乐观:它把一堆右偏分布各自取均值再相加,系统性低估。

从点估计到区间:排期要给区间

实务上的动作有三个:

  • 报区间而非点:对外承诺用 P80(此例 D50),对内排期盯 P50(D48),别再拿 D42 当计划。
  • 留缓冲:launchDay 与 P80 之间的 8 天就是该版本的内容缓冲;缓冲不是浪费,是吸收方差的成本。
  • 砍需求或加产能:若不能推迟上线,就回到 backlog 砍任务或提 availabilityPct,重跑脚本看瓶颈是否移位。这与 7.1 的冲突修复是同一种「改配置、重跑、看报告」的闭环。

产能模型的下一步是把它接到闭环经济仿真上(12 章),因为内容上线会改变产出与消耗;本节只负责「人够不够」。

阅读导航:上一节:7.1 版本列车与赛季结构 · 下一节:7.3 灰度、放量与回滚 。

继续阅读

探索更多技术文章

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

全部文章 返回首页

「slg」更多文章

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