15.1 异常检测与打金识别
前十四章都在设计一个「正常玩家会喜欢、也不会崩」的经济。但经济还有一个它自己管不了的对手：从外部来薅它的人。打金工作室不关心你的产出曲线平不平滑，他们只关心单位时间里能从你的系统里抽走多少资源，然后换成钱。站内的经济系统运营 checklist 把风控列进了上线前必查项，但它给的是「该建风控」这个结论；这一节给的是可跑的检测规则——四条能用一行比较写出来、能在一份配置里调阈值、能在 Python 和 TypeScript 两侧跑出同一个混淆矩阵的规则。它接在 14.3 讲完的「套利空间与工作室边界」之后：14.3 说清楚工作室靠什么赚钱，这一节说清楚怎么把他们从正常玩家里面挑出来。
为什么是统计规则，不是黑名单
最直觉的做法是维护一份黑名单。但黑名单有三个死穴：新号天天在注册、特征随时在变、误封一个付费玩家的代价远大于放过十个打金号。所以工业界的做法不是「认人」，而是「认行为」——不看你叫什么是谁，只看你的行为像不像在薅羊毛。
打金行为有几个稳定的统计学特征，跟具体玩法无关：他们要么产出速率显著偏离同层玩家，要么把资源集中转移到一个出口账号，要么多个小号的行为高度同步（因为是同一台脚本在操作），要么资源的最终流向单一（都汇到一个卖金账号）。这四条特征，每一条都能写成一个可以比较的阈值，四条合起来就是一个可解释、可调参、可复现的检测器。
四条规则
| 规则 | 看什么 | 判据 | 直觉 |
|---|---|---|---|
R1_PROD_OUTLIER | 单位时间产出 | prodPerHour ≥ 阈值 | 产出高到不像人 |
R2_STAR_TOPOLOGY | 转移对端数 + 集中度 | transferPeers ≥ 8 且 topPeerSharePct ≥ 65 | 一个中心账号连着很多小号 |
R3_BEHAVIOR_SYNC | 行为同步率 | syncPct ≥ 阈值 | 动作和另一个号卡在同一拍 |
R4_SINGLE_EXIT | 出口集中度 | exitSharePct ≥ 阈值 | 资源都流向同一个出口 |
四条规则各判一个布尔，任意一条命中就标记为可疑。这里刻意不做加权评分——评分会把「为什么可疑」藏进一个数字里，而运营需要看到的是「它命中了哪条」，好去人工复核。评分只用来做阈值扫描（后面会看到）。
把规则写成配置
下面这段是 D 层配置：三类账号的特征区间，加四条规则的阈值。它是唯一真源——改一个阈值，Python 和 TypeScript 两侧的混淆矩阵会一起变。
{
"version": 1,
"game": "王朝纪元",
"note": "打金识别唯一真源：scenarios 定义三类账号的整数特征区间，rules 定义四条检测规则的阈值。seed + mulberry32 决定整批账号，Python 与 TypeScript 逐位一致。",
"seed": 20261012,
"windowHours": 24,
"scenarios": [
{
"id": "normal",
"label": 0,
"count": 300,
"prodPerHour": [40, 150],
"transferPeers": [0, 9],
"topPeerSharePct": [0, 70],
"syncPct": [0, 85],
"exitSharePct": [0, 72]
},
{
"id": "gold_farmer",
"label": 1,
"count": 40,
"prodPerHour": [110, 320],
"transferPeers": [6, 30],
"topPeerSharePct": [55, 99],
"syncPct": [30, 92],
"exitSharePct": [60, 99]
},
{
"id": "bot_farm",
"label": 2,
"count": 30,
"prodPerHour": [45, 130],
"transferPeers": [1, 4],
"topPeerSharePct": [0, 45],
"syncPct": [68, 100],
"exitSharePct": [0, 45]
}
],
"rules": {
"R1_PROD_OUTLIER": { "field": "prodPerHour", "op": ">=", "threshold": 135 },
"R2_STAR_TOPOLOGY": { "field": "transferPeers", "op": ">=", "threshold": 8, "and": { "field": "topPeerSharePct", "op": ">=", "threshold": 65 } },
"R3_BEHAVIOR_SYNC": { "field": "syncPct", "op": ">=", "threshold": 80 },
"R4_SINGLE_EXIT": { "field": "exitSharePct", "op": ">=", "threshold": 70 }
}
}
三类账号的区间是故意重叠的：正常玩家的产出可以高到 150（越过 135 的阈值），脚本农场的行为同步率可以低到 68（掉到 80 的阈值之下）。重叠是故意的——真实数据里没有一刀切干净的边界，检测器的价值恰恰在于在重叠区里权衡。
跑一批模拟账号
下面这段是 S 层仿真脚本。它用自己实现的确定性 PRNG（mulberry32）按配置造出 370 个账号，跑四条规则，算出混淆矩阵、逐规则统计和阈值扫描，产出 JSON 与 CSV。不用 random 模块，因为默认随机数在不同运行时序列不同，那样 Python 和 TypeScript 就对不上了。
# file: sim/ch15_detection.py
# tier: S
# verify: python3 sim/ch15_detection.py --config config/
"""打金识别：用四条统计规则跑一批模拟账号，输出混淆矩阵。
读 config/ch15_detection.json：scenarios 给三类账号的整数特征区间，rules 给四条规则的阈值。
用自己实现的确定性 PRNG（mulberry32）生成整批账号，四条规则各判一个布尔，任一条命中即标记可疑。
地面真值是 scenario 的 label（0=正常，1=打金，2=脚本农场）。
产出 out/ch15_detection.json（混淆矩阵 + 逐规则统计 + 阈值扫描）与 out/ch15_detection.csv（逐账号）。
纯标准库，零依赖。
"""
from __future__ import annotations
import argparse
import csv
import json
import os
M = 0xFFFFFFFF
def mulberry32(seed: int):
"""确定性 PRNG：与 proto/src/ch15_detection.ts 的 mulberry32 逐位一致。"""
state = seed & M
def rnd() -> float:
nonlocal state
state = (state + 0x6D2B79F5) & M
a = state
t = ((a ^ (a >> 15)) * (1 | a)) & M
t0 = t
t = (t0 + (((t0 ^ (t0 >> 7)) * (61 | t0)) & M)) & M
t = (t ^ t0) & M
return ((t ^ (t >> 14)) & M) / 4294967296.0
return rnd
FEATURE_KEYS = ["prodPerHour", "transferPeers", "topPeerSharePct", "syncPct", "exitSharePct"]
def build_population(cfg: dict) -> list[dict]:
rnd = mulberry32(cfg["seed"])
rows: list[dict] = []
idx = 0
for sc in cfg["scenarios"]:
for _ in range(sc["count"]):
rec = {"id": f"acct-{idx:04d}", "label": sc["label"]}
for k in FEATURE_KEYS:
lo, hi = sc[k]
rec[k] = lo + int(rnd() * (hi - lo + 1)) # int() 截断与 TS 的 Math.floor 对正数一致
rows.append(rec)
idx += 1
return rows
def rule_fires(rule: dict, rec: dict) -> bool:
def cond(c: dict) -> bool:
v, t = rec[c["field"]], c["threshold"]
return v >= t if c["op"] == ">=" else v <= t
return cond(rule) and (cond(rule["and"]) if "and" in rule else True)
def detect(cfg: dict, rows: list[dict]) -> list[dict]:
for rec in rows:
rec["fired"] = {name: rule_fires(rule, rec) for name, rule in cfg["rules"].items()}
rec["score"] = sum(1 for v in rec["fired"].values() if v)
rec["flagged"] = rec["score"] >= 1
return rows
def confusion(rows: list[dict], pred) -> dict:
tp = fp = fn = tn = 0
for r in rows:
truth, guess = r["label"] != 0, pred(r)
if truth and guess:
tp += 1
elif truth:
fn += 1
elif guess:
fp += 1
else:
tn += 1
precision = tp / (tp + fp) if (tp + fp) else 0.0
recall = tp / (tp + fn) if (tp + fn) else 0.0
f1 = 2 * precision * recall / (precision + recall) if (precision + recall) else 0.0
return {"tp": tp, "fp": fp, "fn": fn, "tn": tn,
"precision": round(precision, 4), "recall": round(recall, 4), "f1": round(f1, 4)}
def threshold_sweep(rows: list[dict]) -> list[dict]:
return [{"minRules": m, **confusion(rows, lambda r, m=m: r["score"] >= m)} for m in (1, 2, 3, 4)]
def write_csv(rows: list[dict], names: list[str], path: str) -> None:
with open(path, "w", encoding="utf-8", newline="") as f:
w = csv.writer(f)
w.writerow(["id", "label", *FEATURE_KEYS, *[f"r{i+1}" for i in range(len(names))], "score", "flagged"])
for r in rows:
w.writerow([r["id"], r["label"], *[r[k] for k in FEATURE_KEYS],
*[int(r["fired"][n]) for n in names], r["score"], int(r["flagged"])])
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, "ch15_detection.json"), encoding="utf-8") as f:
cfg = json.load(f)
rows = detect(cfg, build_population(cfg))
report = {
"version": cfg["version"], "game": cfg["game"], "seed": cfg["seed"],
"population": len(rows), "positives": sum(1 for r in rows if r["label"] != 0),
"overall": confusion(rows, lambda r: r["flagged"]),
"perRule": {n: confusion(rows, lambda r, n=n: r["fired"][n]) for n in cfg["rules"]},
"thresholdSweep": threshold_sweep(rows),
}
os.makedirs(args.out, exist_ok=True)
with open(os.path.join(args.out, "ch15_detection.json"), "w", encoding="utf-8") as f:
json.dump(report, f, ensure_ascii=False, indent=2)
write_csv(rows, list(cfg["rules"]), os.path.join(args.out, "ch15_detection.csv"))
print(json.dumps(report, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()
本机实跑（python3 sim/ch15_detection.py --config config/）的关键数字：
population=370 positives=70
overall TP=55 FP=80 FN=15 TN=220 precision=0.4074 recall=0.7857 f1=0.5366
R1_PROD_OUTLIER TP=34 FP=48 FN=36 precision=0.4146 recall=0.4857
R2_STAR_TOPOLOGY TP=30 FP= 9 FN=40 precision=0.7692 recall=0.4286
R3_BEHAVIOR_SYNC TP=22 FP=25 FN=48 precision=0.4681 recall=0.3143
R4_SINGLE_EXIT TP=26 FP= 9 FN=44 precision=0.7429 recall=0.3714
score>=1 precision=0.4074 recall=0.7857 tp=55 fp=80 fn=15
score>=2 precision=0.766 recall=0.5143 tp=36 fp=11 fn=34
score>=3 precision=1.0 recall=0.2857 tp=20 fp=0 fn=50
score>=4 precision=1.0 recall=0.0143 tp=1 fp=0 fn=69
（同机相对量级，绝对数值不可移植；换 seed 或改区间，矩阵会变。）
怎么读这个混淆矩阵
先看逐规则那一栏，四条规则的性格完全不同：R2_STAR_TOPOLOGY 和 R4_SINGLE_EXIT 精确率高（0.77、0.74）但召回率低——它们只在打金行为「成型」时才命中，是保守规则，几乎不冤枉人；R1_PROD_OUTLIER 精确率只有 0.41，因为它把「产出高的正常玩家」也算进来了，是激进规则。所以四条规则合起来（score>=1）的召回率冲到 0.79，代价是精确率掉到 0.41——每抓 5 个可疑号，有 3 个是误报。
阈值扫描那一栏回答的就是「这个代价能不能调」：把标记线从「命中 1 条」抬到「命中 2 条」，精确率从 0.41 升到 0.77，召回率从 0.79 降到 0.51；抬到 3 条，精确率到 1.0（零误报），但召回率只剩 0.29，漏掉七成。没有免费的阈值，运营要选的不是「哪个数最好」，而是「误封和漏抓，哪个更疼」——对一款已经有付费玩家的 SLG，通常宁可漏抓也不误封，于是会把阈值往「命中 2 条」这一档挪。
跨语言一致：同一份配置，两处消费
这套工程的地基是「一份配置、两处消费」。下面这段是 P 层断言，它读同一份 config/ch15_detection.json，用 TypeScript 重写一遍 mulberry32 与四条规则，然后断言混淆矩阵和 Python 侧逐位一致。模块本体在 proto/src/ch15_detection.ts。
// file: proto/test/ch15.test.ts
// tier: P
// verify: node --test 'proto/**/*.test.ts'
// 打金识别的断言：PRNG 与 Python 逐位一致、混淆矩阵对得上、阈值扫描单调。
import { test } from "node:test";
import assert from "node:assert/strict";
import {
loadJson, mulberry32, buildPopulation, detect, confusion, thresholdSweep,
type DetectionConfig,
} from "../src/ch15_detection.ts";
const cfg = loadJson<DetectionConfig>("ch15_detection");
test("mulberry32 前三个数固定（与 Python 侧逐位一致）", () => {
const r = mulberry32(20261012);
const got = [r(), r(), r()].map((x) => x.toFixed(10));
assert.deepEqual(got, ["0.3411238657", "0.0178032639", "0.3645789088"]);
});
test("账号总数与正样本数", () => {
const rows = buildPopulation(cfg);
assert.equal(rows.length, 370);
assert.equal(rows.filter((r) => r.label !== 0).length, 70);
});
test("任意规则命中（score>=1）的混淆矩阵", () => {
const rows = detect(cfg, buildPopulation(cfg));
const c = confusion(rows, (r) => r.flagged);
assert.deepEqual(c, { tp: 55, fp: 80, fn: 15, tn: 220, precision: 0.4074, recall: 0.7857, f1: 0.5366 });
});
test("阈值扫描：召回率随 minRules 单调不增", () => {
const rows = detect(cfg, buildPopulation(cfg));
const sweep = thresholdSweep(rows);
assert.deepEqual(sweep.map((s) => s.minRules), [1, 2, 3, 4]);
for (let i = 1; i < sweep.length; i++) {
assert.ok(sweep[i].recall <= sweep[i - 1].recall, `recall 应在 >=${sweep[i].minRules} 时不增`);
}
});
test("收紧到 score>=3 时零误报", () => {
const rows = detect(cfg, buildPopulation(cfg));
const c = confusion(rows, (r) => r.score >= 3);
assert.equal(c.fp, 0);
assert.equal(c.precision, 1);
});
test("检测器是纯函数：同输入两次结果逐位一致", () => {
const a = confusion(detect(cfg, buildPopulation(cfg)), (r) => r.flagged);
const b = confusion(detect(cfg, buildPopulation(cfg)), (r) => r.flagged);
assert.deepEqual(a, b);
});
node --test proto/test/ch15.test.ts 六条断言全绿。这里最有价值的是第一条——它把「Python 和 TypeScript 的随机数序列相同」这件事钉成了一条可以失败的测试。哪天有人在 Python 侧把 int() 换成 round()，或者 TS 侧忘了 >>> 0，这条测试会立刻变红，而不是等到某个赛季的封号名单里出现一群无辜玩家。
本节小结
这一节把打金识别落成了一条可跑的链路：
- 认行为，不认人：四条统计规则——产出离群、星型拓扑、行为同步、单一出口——每条都能写成一行比较，合起来是一个可解释、可调参、可复现的检测器。
- 阈值是一笔交易：
score>=1时召回 0.79 / 精确 0.41，score>=2时召回 0.51 / 精确 0.77，score>=3时零误报但只抓到三成。选哪档取决于「误封贵还是漏抓贵」。 - 跨语言一致是地基：同一份配置、自己实现的 PRNG，Python 与 TypeScript 跑出同一个混淆矩阵，并被一条测试锁住。
检测器只能告诉你「谁可疑」。一旦可疑变成了确认的事故——多发了资源、复制了物品、算错了重置——真正的考验才开始：怎么止血、怎么定位、怎么决定回不回滚。下一节先把 SLG 常见的经济事故归成几类，每一类都给出症状、根因、影响面与处置。
阅读导航：上一节：14.3 套利、工作室与风控边界 · 下一节：15.2 经济事故案例库 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。