12.2 回答的是「这个地区能不能用」，本节换一个维度：同一批流量里，哪些是真玩家、哪些是刷量。刷量是发行端最常见的作弊形态——渠道刷激活、玩家刷广告奖励、脚本刷签到，直接污染留存与 eCPM 口径。本节不写风控策略，只把「怎么把两类流量分开」做成一条可复现的工程闭环：D 层定义可观测信号与阈值，P 层把判定内核写成可播种的确定性代码，S 层用同一条随机流跑仿真、量化精确率与召回率的权衡。
边界先划清：本节不给风控策略建议，也不复现任何真实风控结论。真实刷量流量与真实风控阈值需要真实业务环境与真实样本，命令行不可复现，标记为 X。下面所有数字都是建模假设（观察时点 2026-10）在可播种仿真下的结果，用来演示「降阈值多抓、代价是误伤」的机制，不代表线上真实分布。这条纪律与出海合规与税务风险清单 里「不把假设当结论」的写法一致。
可观测信号与阈值（D 层）
判据只用四个公开可观测的特征：每分钟事件数 eventsPerMinute（突发速率）、事件间隔的变异系数千分比 intervalCvPermille（越低越机械，正常人手速抖动大）、每千次事件领奖数 claimPermille（脚本倾向于无脑领满）、同一设备指纹下的账号数 accountsPerDevice（设备复用）。把「正常 / 重度正常 / 快刷 / 慢刷」四类账号的特征区间写成可机读配置：
{
"version": 1,
"game": "merge-island",
"observedAt": "2026-10",
"seed": 20261013,
"note": "刷量与作弊流量的可播种检测配置。口径是「公开可观察机制」的建模假设（观察时点 2026-10，非实测）：把「正常 / 重度正常 / 快刷 / 慢刷」四类账号的行为特征区间与检测阈值整理成可机读配置。cohorts 按顺序消费同一个 PCG32 流（每实体 4 次抽签：rate→cv→claim→devices），故实体序列完全可复现。label=1 为刷量账号（用于算检测率），label=0 为正常账号（用于算误伤率）。eventsPerMinute=每分钟事件数、intervalCvPermille=事件间隔的变异系数千分比（越低越机械）、claimPermille=每千次事件领取奖励数、accountsPerDevice=同一设备指纹下的账号数。detector 四条规则各记 1 分，score ≥ scoreThreshold 即判为刷量。真实刷量流量与真实风控结论需真实业务环境，命令行不可复现（X）。全整数，无浮点。",
"cohorts": [
{ "id": "normal", "label": 0, "count": 40, "rateMin": 1, "rateMax": 6, "cvMin": 400, "cvMax": 900, "claimMin": 0, "claimMax": 250, "devicesMin": 1, "devicesMax": 2 },
{ "id": "heavy_legit", "label": 0, "count": 15, "rateMin": 10, "rateMax": 20, "cvMin": 400, "cvMax": 900, "claimMin": 0, "claimMax": 250, "devicesMin": 1, "devicesMax": 2 },
{ "id": "bot_fast", "label": 1, "count": 25, "rateMin": 20, "rateMax": 60, "cvMin": 0, "cvMax": 60, "claimMin": 700, "claimMax": 990, "devicesMin": 1, "devicesMax": 2 },
{ "id": "bot_stealth", "label": 1, "count": 25, "rateMin": 2, "rateMax": 8, "cvMin": 450, "cvMax": 850, "claimMin": 100, "claimMax": 300, "devicesMin": 6, "devicesMax": 40 }
],
"detector": {
"burstPerMinuteMin": 12,
"regularityMaxPermille": 150,
"claimRatePermilleMin": 500,
"accountsPerDeviceMin": 5,
"scoreThreshold": 2
},
"sweepThresholds": [1, 2, 3]
}
四类里最值得留意的是 bot_stealth（慢刷）：它把速率压到 2–8、间隔抖动做到 450–850‰、领奖率降到 100–300‰，单看任何一条都像正常号，唯一暴露点是设备复用 6–40。detector 四条规则各记 1 分，score ≥ scoreThreshold 判为刷量，主阈值取 2。全整数、无浮点。
检测内核：可播种、跨语言对拍（P 层）
内核的关键是确定性：不能用 Math.random()，随机数用可播种 PCG32（BigInt 64 位状态），Python 侧同款实现逐位等价。同一份配置喂给两侧，落盘的混淆摘要必须逐字符一致——这就是「一份配置、两处消费」的验收。
// file: core/src/g12/fraud.ts
// tier: P
// verify: node --test 'core/test/g12_*.test.ts'
//
// 刷量与作弊流量的检测内核：与 sim/g12_fraud_sim.py 逐位等价的 TypeScript 对偶实现。
// 读同一份 config/g12_fraud.json；全整数、无浮点、无 Math.random()，随机数用可播种 PCG32。
export const MULT = 6364136223846793005n;
export const MASK64 = (1n << 64n) - 1n;
export const MASK32 = (1n << 32n) - 1n;
/** PCG32 伪随机数发生器：与 sim/g12_fraud_sim.py 的 Pcg32 逐位等价。 */
export class Pcg32 {
private state: bigint;
private readonly inc: bigint;
constructor(seed: number, seq = 54) {
this.state = 0n;
this.inc = ((BigInt(seq) << 1n) | 1n) & MASK64;
this.u32();
this.state = (this.state + BigInt(seed)) & MASK64;
this.u32();
}
private u32(): number {
const old = this.state;
this.state = (old * MULT + this.inc) & MASK64;
const xs = (((old >> 18n) ^ old) >> 27n) & MASK32;
const rot = old >> 59n;
return Number(((xs >> rot) | (xs << ((32n - rot) & 31n))) & MASK32);
}
span(lo: number, hi: number): number { return lo + this.u32() % (hi - lo + 1); }
}
export interface Cohort { readonly id: string; readonly label: number; readonly count: number;
readonly rateMin: number; readonly rateMax: number; readonly cvMin: number; readonly cvMax: number;
readonly claimMin: number; readonly claimMax: number; readonly devicesMin: number; readonly devicesMax: number; }
export interface Detector { readonly burstPerMinuteMin: number; readonly regularityMaxPermille: number;
readonly claimRatePermilleMin: number; readonly accountsPerDeviceMin: number; readonly scoreThreshold: number; }
export interface FraudConfig { readonly seed: number; readonly cohorts: readonly Cohort[];
readonly detector: Detector; readonly sweepThresholds: readonly number[]; }
export interface Entity { readonly id: string; readonly cohort: string; readonly label: number;
readonly eventsPerMinute: number; readonly intervalCvPermille: number;
readonly claimPermille: number; readonly accountsPerDevice: number; score: number; }
export interface Confusion { readonly threshold: number; readonly tp: number; readonly fp: number;
readonly fn: number; readonly tn: number; readonly precisionPermille: number; readonly recallPermille: number; }
/** 按 cohort 顺序、每实体 4 次抽签（rate→cv→claim→devices），生成可复现的实体序列。 */
export function genEntities(cfg: FraudConfig): Entity[] {
const rng = new Pcg32(cfg.seed);
const out: Entity[] = [];
for (const c of cfg.cohorts) {
for (let j = 0; j < c.count; j += 1) {
out.push({ id: `${c.id}-${String(j).padStart(2, "0")}`, cohort: c.id, label: c.label,
eventsPerMinute: rng.span(c.rateMin, c.rateMax), intervalCvPermille: rng.span(c.cvMin, c.cvMax),
claimPermille: rng.span(c.claimMin, c.claimMax),
accountsPerDevice: rng.span(c.devicesMin, c.devicesMax), score: 0 });
}
}
return out;
}
/** 四条规则各记 1 分：突发速率、间隔规律、领取率、设备复用。 */
export function score(e: Entity, det: Detector): number {
return (e.eventsPerMinute >= det.burstPerMinuteMin ? 1 : 0)
+ (e.intervalCvPermille <= det.regularityMaxPermille ? 1 : 0)
+ (e.claimPermille >= det.claimRatePermilleMin ? 1 : 0)
+ (e.accountsPerDevice >= det.accountsPerDeviceMin ? 1 : 0);
}
/** 在给定阈值下统计混淆矩阵与精确率 / 召回率（千分比，整数地板除）。 */
export function confusion(entities: readonly Entity[], det: Detector, threshold: number): Confusion {
let tp = 0; let fp = 0; let fn = 0; let tn = 0;
for (const e of entities) {
const flagged = score(e, det) >= threshold;
if (e.label === 1 && flagged) tp += 1; else if (e.label === 1) fn += 1;
else if (flagged) fp += 1; else tn += 1;
}
return { threshold, tp, fp, fn, tn,
precisionPermille: tp + fp === 0 ? 1000 : Math.floor((tp * 1000) / (tp + fp)),
recallPermille: tp + fn === 0 ? 1000 : Math.floor((tp * 1000) / (tp + fn)) };
}
/** 跨语言对拍摘要：与 sim/g12_fraud_sim.py 的 digest 逐字符一致（用主阈值）。 */
export function digest(cfg: FraudConfig): string {
const c = confusion(genEntities(cfg), cfg.detector, cfg.detector.scoreThreshold);
return `flagged=${c.tp + c.fp},tp=${c.tp},fp=${c.fp},fn=${c.fn},tn=${c.tn},`
+ `precPermille=${c.precisionPermille},recallPermille=${c.recallPermille}`;
}
/** 阈值扫描：展示「降阈值多抓慢刷、代价是误伤重度正常号」的权衡。 */
export function sweep(cfg: FraudConfig): Confusion[] {
const entities = genEntities(cfg);
return cfg.sweepThresholds.map((th) => confusion(entities, cfg.detector, th));
}
score 的四条规则与配置一一对应；confusion 用整数地板除算 precisionPermille / recallPermille（千分比），避免浮点。同一个内核在浏览器端与 Node 端跑出的数字必须一样，这正是引擎侧「一份逻辑两处宿主」的延续。
仿真与阈值权衡（S 层）
S 层脚本用同一个 seed 生成 105 个实体（40 正常 + 15 重度正常 + 25 快刷 + 25 慢刷），跑检测器，再对 sweepThresholds 扫描阈值：
#!/usr/bin/env python3
# file: sim/g12_fraud_sim.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/g12_fraud_sim.py --config config/ --out out/
"""刷量与作弊流量的可播种检测仿真（S 层）。读 config/g12_fraud.json，用 PCG32（seed=config.seed）
按 cohort 顺序生成四类账号行为特征，四条规则各记 1 分，score ≥ threshold 即判为刷量，统计
tp/fp/fn/tn 与 precision‰/recall‰ 并扫描阈值。PCG32 与 core/src/g12/fraud.ts 逐位等价，实体序列
与摘要完全可复现。真实刷量流量与真实风控结论需真实业务环境，命令行不可复现（X）。
产出 .json/.csv/.svg，全整数无浮点。
"""
from __future__ import annotations
import argparse
import csv
import json
import os
import xml.etree.ElementTree as ET
MULT = 6364136223846793005
MASK64 = (1 << 64) - 1
MASK32 = (1 << 32) - 1
class Pcg32:
"""与 core/src/g12/fraud.ts 的 Pcg32 逐位等价的 PCG32（可播种、确定性）。"""
def __init__(self, seed: int, seq: int = 54) -> None:
self.state, self.inc = 0, ((seq << 1) | 1) & MASK64
self._u32()
self.state = (self.state + seed) & MASK64
self._u32()
def _u32(self) -> int:
old = self.state
self.state = (old * MULT + self.inc) & MASK64
xs = (((old >> 18) ^ old) >> 27) & MASK32
rot = old >> 59
return ((xs >> rot) | (xs << ((32 - rot) & 31))) & MASK32
def span(self, lo: int, hi: int) -> int:
return lo + self._u32() % (hi - lo + 1)
def gen_entities(cfg: dict) -> list[dict]:
"""按 cohort 顺序、每实体 4 次抽签（rate→cv→claim→devices），生成可复现的实体序列。"""
rng, out = Pcg32(cfg["seed"]), []
for c in cfg["cohorts"]:
for j in range(c["count"]):
out.append({"id": f'{c["id"]}-{j:02d}', "cohort": c["id"], "label": c["label"],
"eventsPerMinute": rng.span(c["rateMin"], c["rateMax"]),
"intervalCvPermille": rng.span(c["cvMin"], c["cvMax"]),
"claimPermille": rng.span(c["claimMin"], c["claimMax"]),
"accountsPerDevice": rng.span(c["devicesMin"], c["devicesMax"])})
return out
def score(e: dict, det: dict) -> int:
"""与 core/src/g12/fraud.ts 的 score 逐位等价的四条规则打分。"""
return ((e["eventsPerMinute"] >= det["burstPerMinuteMin"])
+ (e["intervalCvPermille"] <= det["regularityMaxPermille"])
+ (e["claimPermille"] >= det["claimRatePermilleMin"])
+ (e["accountsPerDevice"] >= det["accountsPerDeviceMin"]))
def confusion(entities: list[dict], det: dict, threshold: int) -> dict:
tp = fp = fn = tn = 0
for e in entities:
flagged = score(e, det) >= threshold
if e["label"] == 1 and flagged:
tp += 1
elif e["label"] == 1:
fn += 1
elif flagged:
fp += 1
else:
tn += 1
return {"threshold": threshold, "tp": tp, "fp": fp, "fn": fn, "tn": tn,
"precisionPermille": (tp * 1000) // (tp + fp) if (tp + fp) else 1000,
"recallPermille": (tp * 1000) // (tp + fn) if (tp + fn) else 1000}
def draw_svg(path: str, sweep: list[dict]) -> int:
"""纯 Python SVG：阈值扫描的召回（蓝）与误伤（红）双条。"""
W, ROW, PAD, BAR = 720, 44, 150, 400
H = 70 + ROW * len(sweep)
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"})
for i, r in enumerate(sweep):
y = 30 + i * ROW
ET.SubElement(svg, "text", {"x": "12", "y": str(y + 20), "font-size": "12",
"fill": "#374151"}).text = f'阈值 ≥{r["threshold"]}'
ET.SubElement(svg, "rect", {"x": str(PAD), "y": str(y + 4), "height": "14", "rx": "3",
"width": str(BAR * r["recallPermille"] // 1000), "fill": "#2563eb"})
ET.SubElement(svg, "rect", {"x": str(PAD), "y": str(y + 22), "height": "14", "rx": "3",
"width": str(BAR * r["fp"] // max(1, r["tp"] + r["fp"])), "fill": "#dc2626"})
ET.SubElement(svg, "text", {"x": str(PAD + BAR + 8), "y": str(y + 20), "font-size": "11",
"fill": "#6b7280"}).text = f'召回 {r["recallPermille"]}‰ TP{r["tp"]} · 误伤 {r["fp"]}'
os.makedirs(os.path.dirname(path), exist_ok=True)
ET.ElementTree(svg).write(path, encoding="utf-8", xml_declaration=True)
return len(ET.tostring(svg))
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--config", default="config/")
ap.add_argument("--out", default="out/")
a = ap.parse_args()
with open(os.path.join(a.config, "g12_fraud.json"), encoding="utf-8") as f:
cfg = json.load(f)
det = cfg["detector"]
entities = gen_entities(cfg)
for e in entities:
e["score"] = score(e, det)
sweep = [confusion(entities, det, th) for th in cfg["sweepThresholds"]]
primary = next(r for r in sweep if r["threshold"] == det["scoreThreshold"])
cohorts = [{"cohort": c["id"], "label": c["label"],
"count": sum(1 for e in entities if e["cohort"] == c["id"]),
"flaggedAtPrimary": sum(1 for e in entities if e["cohort"] == c["id"]
and e["score"] >= det["scoreThreshold"]),
"scoreMin": min(e["score"] for e in entities if e["cohort"] == c["id"]),
"scoreMax": max(e["score"] for e in entities if e["cohort"] == c["id"])}
for c in cfg["cohorts"]]
digest = (f'flagged={primary["tp"] + primary["fp"]},tp={primary["tp"]},fp={primary["fp"]},'
f'fn={primary["fn"]},tn={primary["tn"]},precPermille={primary["precisionPermille"]},'
f'recallPermille={primary["recallPermille"]}')
os.makedirs(a.out, exist_ok=True)
svg = draw_svg(os.path.join(a.out, "g12_fraud_sim.svg"), sweep)
with open(os.path.join(a.out, "g12_fraud_sim.csv"), "w", encoding="utf-8", newline="") as f:
w = csv.writer(f)
w.writerow(["id", "cohort", "label", "eventsPerMinute", "intervalCvPermille",
"claimPermille", "accountsPerDevice", "score", "flagged"])
for e in entities:
w.writerow([e["id"], e["cohort"], e["label"], e["eventsPerMinute"],
e["intervalCvPermille"], e["claimPermille"], e["accountsPerDevice"],
e["score"], int(e["score"] >= det["scoreThreshold"])])
rep = {"version": cfg["version"], "game": cfg["game"], "observedAt": cfg["observedAt"],
"seed": cfg["seed"], "entityCount": len(entities), "detector": det, "primary": primary,
"sweep": sweep, "cohorts": cohorts, "digest": digest,
"svg": os.path.join(a.out, "g12_fraud_sim.svg"), "svgBytes": svg}
with open(os.path.join(a.out, "g12_fraud_sim.json"), "w", encoding="utf-8") as f:
json.dump(rep, f, ensure_ascii=False, indent=2)
print(json.dumps({"digest": digest, "entityCount": len(entities), "primary": primary,
"sweep": sweep, "cohorts": cohorts}, ensure_ascii=False, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())
同机实跑（/opt/homebrew/bin/python3，3.14.6）的关键数字如下，全部来自命令行真实输出：
digest = flagged=25,tp=25,fp=0,fn=25,tn=55,precPermille=1000,recallPermille=500
阈值扫描 ≥1 → tp=50 fp=13 recall=1000‰ prec=793‰
阈值扫描 ≥2 → tp=25 fp=0 recall=500‰ prec=1000‰
阈值扫描 ≥3 → tp=25 fp=0 recall=500‰ prec=1000‰
cohort score 区间 normal 0–0 / heavy_legit 0–1 / bot_fast 3–3 / bot_stealth 1–1
这张表就是本节要传达的核心权衡：主阈值 2 下精确率 1000‰（零误伤），但召回只有 500‰——慢刷全漏（25 个 bot_stealth 每个 score 只有 1，只命中设备复用一条规则）。把阈值降到 1，召回拉满 1000‰，代价是误伤 13 个 heavy_legit（重度正常玩家被当成刷量）。这正是反作弊的日常：没有免费的全中，阈值就是在「漏掉多少刷量」和「冤枉多少真玩家」之间选一个可接受的交点。
TypeScript 内核跑 node --test 'core/test/g12_*.test.ts' 6/6 通过，其中「跨语言一致」用例直接断言 digest(cfg) 等于 Python 侧落盘的金标准字符串——两侧判定逻辑等价、可复现。这就是本节的验收标准：不是某条规则多聪明，而是整条链路可重算——换个 seed、加一条规则、调一次阈值，结论都能重新算出来。
阅读导航：上一节：12.2 制裁与地区可用性 · 下一节：13.1 海外渠道 UA 口径 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。