15.2 投放、结算与资质的典型失败

triggered   = (comparator == "gt") ? observed > threshold : observed < threshold
gap         = |observed − threshold|            # 越过阈值的绝对量
gapPermille = gap × 1000 // max(threshold, 1)   # 越界幅度（千分比）

#!/usr/bin/env python3
# file: sim/o15_failure.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/o15_failure.py --config config/ --out out/
"""投放、结算与资质三类典型失败的判定仿真：把「失败模式清单」变成可复算的布尔判定。

读 config/o15_failure.json 的每条场景，按比较方向判定是否触发失败：

    triggered(s)   = (comparator == "gt") ? observed > threshold : observed < threshold
    gap(s)         = |observed - threshold|                        # 越过阈值的绝对量
    gapPermille(s) = gap * 1000 // max(threshold, 1)               # 相对阈值的越界幅度
    severity(s)    = triggered ? "HIGH" : "OK"

再按 domain 汇总：某域内任一条触发，则该域判为 RISK，否则 OK。给出一份「失败体检报告」——
它不预测未来，只把复盘时填进去的观测值逐条对照阈值，把「拍脑袋觉得有问题」变成「这一条越界了多少」。

口径：observed 是模型侧填值，不是任何真实后台 / 结算单 / 审批结果的实测；真实结算差异与
真实审批结果需真实业务环境，命令行不可复现（X）。产出 out/o15_failure.json + .csv + .svg。
全整数、地板除，无浮点。
"""
from __future__ import annotations

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


def judge(s: dict) -> dict:
    obs, th, cmp = s["observed"], s["threshold"], s["comparator"]
    triggered = (obs > th) if cmp == "gt" else (obs < th)
    gap = abs(obs - th)
    gap_permille = gap * 1000 // max(th, 1)
    return dict(s, triggered=triggered, gap=gap, gapPermille=gap_permille,
                severity="HIGH" if triggered else "OK")


def draw_svg(path: str, rows: list, domains: list) -> int:
    """纯 Python SVG：每个域一根横条——该域触发的失败条数 / 该域场景总数。"""
    W, H, PAD, ROW = 760, 90 + 48 * len(domains), 200, 48
    root = ET.Element("svg", {"xmlns": "http://www.w3.org/2000/svg", "width": str(W),
                              "height": str(H), "viewBox": f"0 0 {W} {H}"})
    ET.SubElement(root, "rect", {"width": str(W), "height": str(H), "fill": "#ffffff"})
    ET.SubElement(root, "text", {"x": "16", "y": "28", "font-size": "15",
                                 "fill": "#111827"}).text = "三类失败的体检：各域触发条数 / 场景总数"
    span = W - PAD - 160
    for i, d in enumerate(domains):
        total = sum(1 for r in rows if r["domain"] == d["id"])
        hit = sum(1 for r in rows if r["domain"] == d["id"] and r["triggered"])
        y = 56 + i * ROW
        w = span * hit // max(total, 1)
        color = "#ef4444" if hit else "#16a34a"
        ET.SubElement(root, "text", {"x": "16", "y": str(y + 20), "font-size": "12",
                                     "fill": "#374151"}).text = d["name"]
        ET.SubElement(root, "rect", {"x": str(PAD), "y": str(y + 6), "width": str(max(w, 1)),
                                     "height": "22", "rx": "4", "fill": color})
        ET.SubElement(root, "text", {"x": str(PAD + max(w, 1) + 8), "y": str(y + 22),
                                     "font-size": "12", "fill": "#6b7280"}).text = f"{hit} / {total} 触发"
    os.makedirs(os.path.dirname(path), exist_ok=True)
    ET.ElementTree(root).write(path, encoding="utf-8", xml_declaration=True)
    return len(ET.tostring(root))


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, "o15_failure.json"), encoding="utf-8") as f:
        cfg = json.load(f)
    rows = [judge(s) for s in cfg["scenarios"]]

    domain_summary = []
    for d in cfg["domains"]:
        in_domain = [r for r in rows if r["domain"] == d["id"]]
        hit = sum(1 for r in in_domain if r["triggered"])
        domain_summary.append({"id": d["id"], "name": d["name"], "total": len(in_domain),
                               "triggered": hit, "verdict": "RISK" if hit else "OK"})
    triggered_ids = [r["id"] for r in rows if r["triggered"]]
    digest = ",".join(f'{r["id"]}:{"1" if r["triggered"] else "0"}:{r["gapPermille"]}'
                      for r in rows) + f';hit={len(triggered_ids)}'

    os.makedirs(a.out, exist_ok=True)
    svg = draw_svg(os.path.join(a.out, "o15_failure.svg"), rows, cfg["domains"])
    with open(os.path.join(a.out, "o15_failure.csv"), "w", encoding="utf-8", newline="") as f:
        w = csv.writer(f)
        cols = ("id", "domain", "name", "metric", "observed", "threshold", "comparator",
                "unit", "triggered", "gap", "gapPermille", "severity", "note")
        w.writerow(cols)
        for r in rows:
            w.writerow([r[k] for k in cols])
    rep = {"version": cfg["version"], "game": cfg["game"], "observedAt": cfg["observedAt"],
           "scenarioCount": len(rows), "triggeredCount": len(triggered_ids),
           "triggeredIds": triggered_ids, "domainSummary": domain_summary, "scenarios": rows,
           "worstScenario": max(rows, key=lambda r: (r["triggered"], r["gapPermille"]))["id"],
           "digest": digest, "csv": os.path.join(a.out, "o15_failure.csv"),
           "svg": os.path.join(a.out, "o15_failure.svg"), "svgBytes": svg}
    with open(os.path.join(a.out, "o15_failure.json"), "w", encoding="utf-8") as f:
        json.dump(rep, f, ensure_ascii=False, indent=2)
    print(json.dumps({"domainSummary": domain_summary, "triggeredCount": len(triggered_ids),
                      "triggeredIds": triggered_ids,
                      "scenarios": [(r["id"], "FAIL" if r["triggered"] else "OK", r["gapPermille"])
                                    for r in rows],
                      "worstScenario": rep["worstScenario"], "digest": digest},
                     ensure_ascii=False, indent=2))
    return 0


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

{
  "version": 1,
  "game": "merge-island",
  "observedAt": "2026-10",
  "note": "投放、结算与资质三类典型失败的判定清单。口径是「公开可观察机制」的建模假设：每条失败模式给出一个可观测指标 observed、一条阈值 threshold 与比较方向 comparator（gt=observed 超过阈值即判失败；lt=observed 低于阈值即判失败）。observed 是「复盘时在模型侧填进去的值」，不是任何真实后台、结算单或审批结果的实测；真实结算差异、真实审批结果需真实业务环境，命令行不可复现（X）。全整数，无浮点。",
  "domains": [
    { "id": "ua", "name": "投放失败", "owner": "投放操盘手" },
    { "id": "settle", "name": "结算失败", "owner": "财务 / 商务" },
    { "id": "qual", "name": "资质失败", "owner": "法务 / 发行" }
  ],
  "scenarios": [
    { "id": "ua-cac-over-ltv", "domain": "ua", "name": "单用户成本高于生命周期价值", "metric": "cacFen", "observed": 520, "threshold": 430, "comparator": "gt", "unit": "分", "note": "CAC 高于 LTV30，越投越亏；应关停或降量" },
    { "id": "ua-ramp-too-fast", "domain": "ua", "name": "放量节奏过猛", "metric": "rampPct", "observed": 200, "threshold": 130, "comparator": "gt", "unit": "%", "note": "单次预算爬坡超过学习期承载，易触发重新学习" },
    { "id": "ua-creative-fatigue", "domain": "ua", "name": "素材疲劳未及时替换", "metric": "creativeAgeDays", "observed": 21, "threshold": 14, "comparator": "gt", "unit": "天", "note": "同组素材在线超过衰减窗口，CPI 抬头" },
    { "id": "ua-funnel-drop", "domain": "ua", "name": "点击到安装漏斗异常", "metric": "ctiPermille", "observed": 145, "threshold": 180, "comparator": "lt", "unit": "‰", "note": "点击到安装转化率偏低，落地页或包体有问题" },
    { "id": "settle-long-cash-cycle", "domain": "settle", "name": "账期超过现金缓冲", "metric": "cashDays", "observed": 75, "threshold": 45, "comparator": "gt", "unit": "天", "note": "结算周期加冻结期超过可垫付天数，现金流断裂风险" },
    { "id": "settle-recon-gap", "domain": "settle", "name": "三方对账差异超阈", "metric": "reconDiffPermille", "observed": 85, "threshold": 30, "comparator": "gt", "unit": "‰", "note": "归因口径差异过大，结算金额对不上" },
    { "id": "settle-refund-spike", "domain": "settle", "name": "退款率异常", "metric": "refundPermille", "observed": 45, "threshold": 80, "comparator": "gt", "unit": "‰", "note": "退款率在阈值内，本项不判失败" },
    { "id": "qual-lead-short", "domain": "qual", "name": "资质倒排时间不足", "metric": "qualLeadDays", "observed": 45, "threshold": 90, "comparator": "lt", "unit": "天", "note": "距上线预留的资质办理天数不足，会卡上线" },
    { "id": "qual-missing-material", "domain": "qual", "name": "必填材料缺项", "metric": "missingMandatory", "observed": 2, "threshold": 0, "comparator": "gt", "unit": "项", "note": "版号/文网文/软著必填材料缺项，提审必被退回" },
    { "id": "qual-ready", "domain": "qual", "name": "资质齐备", "metric": "missingMandatory", "observed": 0, "threshold": 0, "comparator": "gt", "unit": "项", "note": "材料齐备，本项不判失败" }
  ]
}