本节目标：用复式账本把每一次资源变动记成一组借贷分录，并证明「总量守恒」——产出与消耗单独记账，玩家之间的转移不改变总量。
站内那篇经济平衡长文 讲了「产出 ≈ 消耗」的平衡思想，但它是平衡框架，没给记账机制。本节给的是 SLG 大地图口径的复式账本：它能把「凭空造物」「重复发奖」「转移算错」这类 bug 在记账层就抓住——这是本卷经济系统的地基，也是后续产出、消耗、闭环各章共同依赖的底层约束。
9.3 账本模型与守恒约束
9.1 和 9.2 把资源的「身份」定清楚了：它属于哪一层、归谁、能流到哪。但身份是静态的，经济是动态的——每天有无数笔资源变动在发生。怎么保证资源不会凭空产生、也不会凭空消失？
答案是把每一次变动都当成一笔会计意义上的交易来记。
9.3.1 复式记账：每一次变动都是借贷
会计里的复式记账有一条铁律：有借必有贷，借贷必相等。经济系统直接借用它：
一笔资源变动不是一个孤立的「银币 +100」，而是一组借贷分录（postings）。同一资源在所有分录上的增量（delta）之和必须为 0。这样一来，任何一次「凭空造物」都会立刻让分录不平衡，被账本层当场拦下。
产出与消耗不能凭空进出，所以给它们配两个系统账户：@source（经济体外，产出从这里流出）与 @sink（消耗流进这里）。玩家之间的转移则不涉及系统账户。
9.3.2 三类交易与它们的记账
| 交易类型 | 分录 | 对玩家总量的影响 |
|---|---|---|
produce 产出 | 玩家 +N，@source −N | +N |
consume 消耗 | 玩家 −N，@sink +N | −N |
transfer 转移 | 玩家 A −N，玩家 B +N | 0（只搬存量） |
这三类交易合起来给出守恒方程（transfer 不在右边，它只搬存量）：
玩家期末总量 = 玩家期初总量 + 累计产出 − 累计消耗
9.3.3 账本脚本（S 层）
本章的账本配置含 3 个玩家、2 个系统账户（@source / @sink）、2 种资源（wood / gold）和 8 笔交易。交易清单如下：
| 交易 | 类型 | 分录 |
|---|---|---|
| tx1 | produce | p1 wood +200 / @source wood −200 |
| tx2 | produce | p2 wood +150 / @source wood −150 |
| tx3 | transfer | p1 wood −100 / p2 wood +100 |
| tx4 | transfer | p2 wood −80 / p3 wood +80 |
| tx5 | consume | p3 wood −50 / @sink wood +50 |
| tx6 | produce | p1 gold +300 / @source gold −300 |
| tx7 | transfer | p1 gold −120 / p3 gold +120 |
| tx8 | consume | p2 gold −70 / @sink gold +70 |
下面这段是 S 层仿真脚本，它按顺序应用这 8 笔交易，每笔都做复式校验，最后断言守恒方程成立、且 transfer 交易对玩家总量净贡献为 0。它同样带反例自检：注入一笔凭空造物的交易，确认复式校验能抓到它。
# file: sim/ch09_ledger.py
# tier: S
# verify: python3 sim/ch09_ledger.py --config config/ --out out/
"""复式账本与守恒约束。
每条交易是一组借贷分录（postings）：同一资源在所有分录上的 delta 之和必须为 0
（有借必有贷）。产出记到 @source、消耗记到 @sink，玩家之间的转移只搬存量、不改
玩家总量。判定：1) 复式平衡；2) 玩家期末 == 期初 + 累计产出 − 累计消耗；
3) transfer 类交易对玩家总量净贡献为 0。
产出 out/ch09_ledger.json + out/ch09_ledger_log.csv + out/ch09_ledger.svg。
"""
from __future__ import annotations
import argparse, csv, json, os, sys, xml.etree.ElementTree as ET
def load(cfg_dir, name):
with open(os.path.join(cfg_dir, f"{name}.json"), encoding="utf-8") as f:
return json.load(f)
def apply_transaction(ledger, tx, resources):
"""应用一笔交易，返回它的复式校验结果。"""
per_res = {r: 0 for r in resources}
for p in tx["postings"]:
acct = ledger.setdefault(p["account"], {})
acct[p["resource"]] = acct.get(p["resource"], 0) + p["delta"]
per_res[p["resource"]] += p["delta"]
residual = {r: v for r, v in per_res.items() if v != 0}
return {"id": tx["id"], "kind": tx["kind"], "balanced": not residual, "residual": residual}
def player_total(ledger, players, resources):
return {r: sum(ledger[a].get(r, 0) for a in players) for r in resources}
def run(cfg):
resources = cfg["resources"]
players = [a["id"] for a in cfg["accounts"] if a["type"] == "player"]
system = {a["id"] for a in cfg["accounts"] if a["type"] == "system"}
ledger = {a["id"]: {r: 0 for r in resources} for a in cfg["accounts"]}
for a, bal in cfg["initial"].items():
for r in resources:
ledger[a][r] = bal.get(r, 0)
start = player_total(ledger, players, resources)
produced = {r: 0 for r in resources}
consumed = {r: 0 for r in resources}
transferred = {r: 0 for r in resources}
checks, errors = [], []
for tx in cfg["transactions"]:
chk = apply_transaction(ledger, tx, resources)
checks.append(chk)
if not chk["balanced"]:
errors.append(f"交易 {tx['id']} 复式不平衡：{chk['residual']}")
for p in tx["postings"]:
if p["account"] in system:
if tx["kind"] == "produce":
produced[p["resource"]] += abs(p["delta"])
elif tx["kind"] == "consume":
consumed[p["resource"]] += abs(p["delta"])
if tx["kind"] == "transfer" and p["delta"] < 0:
transferred[p["resource"]] += -p["delta"]
end = player_total(ledger, players, resources)
conservation = {r: end[r] == start[r] + produced[r] - consumed[r] for r in resources}
errors += [f"资源 {r} 不守恒：{end[r]} != {start[r]}+{produced[r]}-{consumed[r]}"
for r in resources if not conservation[r]]
transfer_neutral = all(
sum(p["delta"] for p in tx["postings"] if p["account"] in players) == 0
for tx in cfg["transactions"] if tx["kind"] == "transfer")
if not transfer_neutral:
errors.append("存在改变玩家总量的 transfer 交易")
return {"version": 1, "players": players, "resources": resources, "start": start, "end": end,
"produced": produced, "consumed": consumed, "transferred": transferred,
"doubleEntryBalanced": all(c["balanced"] for c in checks), "conservation": conservation,
"transferNeutral": transfer_neutral, "transactionCount": len(cfg["transactions"]),
"checks": checks, "errors": errors, "ok": not errors}
def self_test(cfg):
"""反例自检：注入一笔凭空造物的交易，确认复式校验能抓到它。"""
bad = json.loads(json.dumps(cfg))
bad["transactions"].append({"id": "txL", "kind": "transfer",
"postings": [{"account": "p1", "resource": "wood", "delta": 999}]})
rep = run(bad)
return {"detectsLeak": not rep["doubleEntryBalanced"], "ok": rep["ok"]}
def draw_svg(path, rep):
svg = ET.Element("svg", {"xmlns": "http://www.w3.org/2000/svg", "viewBox": "0 0 660 300"})
for i, r in enumerate(rep["resources"]):
y = 60 + i * 90
lo, hi, span = rep["start"][r], rep["end"][r], max(1, max(rep["start"][r], rep["end"][r]))
ET.SubElement(svg, "text", {"x": "30", "y": f"{y + 5}", "font-size": "12", "fill": "#1f2937"}).text = r
ET.SubElement(svg, "rect", {"x": "90", "y": f"{y - 10}", "width": f"{400 * lo // span}",
"height": "20", "fill": "#cbd5e1"})
ET.SubElement(svg, "rect", {"x": "90", "y": f"{y + 12}", "width": f"{400 * hi // span}",
"height": "20", "fill": "#1f6feb"})
ET.SubElement(svg, "text", {"x": f"{100 + 400 * hi // span}", "y": f"{y + 27}",
"font-size": "11", "fill": "#1f2937"}).text = f"{lo} → {hi}"
os.makedirs(os.path.dirname(path), exist_ok=True)
ET.ElementTree(svg).write(path, encoding="utf-8", xml_declaration=True)
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--config", default="config/")
ap.add_argument("--out", default="out/")
args = ap.parse_args()
cfg = load(args.config, "ch09_ledger")
rep, st = run(cfg), self_test(cfg)
os.makedirs(args.out, exist_ok=True)
with open(os.path.join(args.out, "ch09_ledger.json"), "w", encoding="utf-8") as f:
json.dump({**rep, "selfTest": st}, f, ensure_ascii=False, indent=2)
with open(os.path.join(args.out, "ch09_ledger_log.csv"), "w", encoding="utf-8", newline="") as f:
w = csv.writer(f)
w.writerow(["txId", "kind", "balanced", "postings"])
w.writerows([[c["id"], c["kind"], c["balanced"], len(tx["postings"])]
for c, tx in zip(rep["checks"], cfg["transactions"])])
draw_svg(os.path.join(args.out, "ch09_ledger.svg"), rep)
print(f"transactions={rep['transactionCount']} doubleEntryBalanced={rep['doubleEntryBalanced']}")
for r in rep["resources"]:
print(f"{r}: start={rep['start'][r]} produced={rep['produced'][r]} consumed={rep['consumed'][r]} "
f"end={rep['end'][r]} transferred={rep['transferred'][r]} conservation={rep['conservation'][r]}")
print(f"transferNeutral={rep['transferNeutral']} selfTest(detectsLeak)={st['detectsLeak']}")
print("== 账本守恒：OK" if rep["ok"] and st["detectsLeak"] else "== 账本守恒：FAIL")
return 0 if rep["ok"] and st["detectsLeak"] else 1
if __name__ == "__main__":
sys.exit(main())
本机实跑输出：
transactions=8 doubleEntryBalanced=True
wood: start=2400 produced=350 consumed=50 end=2700 transferred=180 conservation=True
gold: start=1000 produced=300 consumed=70 end=1230 transferred=120 conservation=True
transferNeutral=True selfTest(detectsLeak)=True
== 账本守恒：OK
验证一遍：木材期初 2400、累计产出 350、累计消耗 50，期末 2700 = 2400 + 350 − 50，对得上；期间有 180 木材被转移，但它没有进入方程右边，证明转移只搬存量。银币同理：1000 + 300 − 70 = 1230。
9.3.4 守恒的压力基准（B 层）
8 笔交易证明不了记账路径在高压下不出错。下面这段是 B 层基准（软告警不硬失败）：用固定种子的 LCG 生成 5000 笔随机交易（产出 / 消耗 / 转移），每笔都走复式记账，逐步核对守恒方程。全程整数运算，理论漂移恒为 0；一旦出现非零漂移就打印 WARN。
# file: sim/ch09_ledger_bench.py
# tier: B
# verify: python3 sim/ch09_ledger_bench.py --config config/ --trials 5000
"""账本守恒的压力基准（B 层，软告警不硬失败）。
用固定种子的 LCG 生成一大批随机交易（产出/消耗/转移），每笔都走复式记账，逐步核对
「玩家总量 == 期初 + 累计产出 − 累计消耗」。全程整数运算，理论漂移恒为 0；一旦出现
非零漂移就打印 WARN，提示记账路径有漏。产出 out/ch09_ledger_bench.json。
"""
from __future__ import annotations
import argparse, json, os, sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from ch09_ledger import load # noqa: E402
OUT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "out")
def lcg(seed):
state = seed & 0xFFFFFFFF
def nxt():
nonlocal state
state = (state * 1103515245 + 12345) % 2147483648
return state
return nxt
def run(cfg, trials, seed):
resources = cfg["resources"]
players = [a["id"] for a in cfg["accounts"] if a["type"] == "player"]
ledger = {a["id"]: {r: 0 for r in resources} for a in cfg["accounts"]}
for a, bal in cfg["initial"].items():
for r in resources:
ledger[a][r] = bal.get(r, 0)
start = {r: sum(ledger[a][r] for a in players) for r in resources}
produced = {r: 0 for r in resources}
consumed = {r: 0 for r in resources}
rng, postings, transfers, max_drift, violations = lcg(seed), 0, 0, 0, 0
for _ in range(trials):
kind, res, amt = rng() % 3, resources[rng() % len(resources)], 1 + rng() % 99
src, dst = players[rng() % len(players)], players[rng() % len(players)]
if kind == 0:
ledger[src][res] += amt
ledger["@source"][res] -= amt
produced[res] += amt
elif kind == 1:
ledger[src][res] -= amt
ledger["@sink"][res] += amt
consumed[res] += amt
else:
if dst == src:
continue
ledger[src][res] -= amt
ledger[dst][res] += amt
transfers += 1
postings += 2
drift = abs(sum(ledger[a][res] for a in players) - (start[res] + produced[res] - consumed[res]))
max_drift = max(max_drift, drift)
violations += 1 if drift != 0 else 0
end = {r: sum(ledger[a][r] for a in players) for r in resources}
return {"trials": trials, "seed": seed, "postings": postings, "transfers": transfers,
"transferSharePct": transfers * 100 // max(1, trials), "maxDrift": max_drift,
"conservationViolations": violations, "start": start, "produced": produced,
"consumed": consumed, "end": end}
def main():
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=5000)
ap.add_argument("--seed", type=int, default=20261012)
args = ap.parse_args()
rep = run(load(args.config, "ch09_ledger"), args.trials, args.seed)
os.makedirs(OUT, exist_ok=True)
with open(os.path.join(OUT, "ch09_ledger_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']} postings={rep['postings']} "
f"transfers={rep['transfers']} ({rep['transferSharePct']}%)")
print(f"maxDrift={rep['maxDrift']} conservationViolations={rep['conservationViolations']}")
for r in rep["end"]:
print(f"{r}: start={rep['start'][r]} +produced={rep['produced'][r]} "
f"-consumed={rep['consumed'][r]} = end={rep['end'][r]}")
if rep["maxDrift"] != 0:
print(f"WARN 出现非零漂移 {rep['maxDrift']}：记账路径存在漏记")
return 0
if __name__ == "__main__":
sys.exit(main())
本机实跑输出：
trials=5000 seed=20261012 postings=8850 transfers=1059 (21%)
maxDrift=0 conservationViolations=0
wood: start=2400 +produced=42712 -consumed=42537 = end=2575
gold: start=1000 +produced=41900 -consumed=41264 = end=1636
5000 笔交易、8850 条分录、1059 次转移，最大漂移 0、守恒违例 0。这里的所有数字都是同机相对量级，绝对数值不可移植——它证明的是记账路径本身不漏记，不是某个具体的产出速率。B 层是软告警：真出现非零漂移时会打印 WARN，而不是让脚本失败。
小结
- 复式账本把每次资源变动记成一组借贷分录，同一资源的分录 delta 之和必须为 0（有借必有贷）。
- 产出记
@source、消耗记@sink、玩家间转移只搬存量，守恒方程是 期末 = 期初 + 产出 − 消耗，transfer 不在右边。 - S 层脚本用 8 笔交易证明守恒成立、transfer 净贡献为 0；反例自检证明「凭空造物」会被当场抓住。
- B 层基准用 5000 笔随机交易压测记账路径，最大漂移 0，确认路径不漏记。
账本是本卷经济系统的地基。接下来第 10 章进入产出端：资源从哪来、以什么速率进来——而所有产出都会记成 @source 流出的分录，落进本节这套账本。
阅读导航：上一节：9.2 绑定与非绑定、账号与角色 · 下一节：10.1 产出源清单与产出速率 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。