projectAIpopular/scripts/sense_report.py

"""E-G1 shadow 报告（T-G8）：一致率 / 档位分布 / conformal 覆盖率 / 成本延迟 delta。

数据源：sense.sqlite3 tier_observations（collect/shadow 期积累的观察）。
输出：CSV + Markdown 报告入 research/v2_experiments/（E-G1_sense_report.*）。
晋升门核对（§6）：一致率 >=85%、覆盖 <= α+2%、T1 占比 40–55%、标签 >= min_labels。

用法：
    python scripts/sense_report.py --db data/sense.sqlite3
"""
from __future__ import annotations

import argparse
import json
import sys
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))

if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(encoding="utf-8")
    sys.stderr.reconfigure(encoding="utf-8")

from gateway.sense.labeler import derive_true_tiers  # noqa: E402
from gateway.sense.store import SenseStore  # noqa: E402

TIERS = ("T1", "T2", "T3")


def collect(store, min_labels: int, alpha: float, now: float) -> dict:
    """从观察表推导标签并聚合报告数据。"""
    derived = derive_true_tiers(store, now=now)
    rows = store.labeled_rows(limit=500000)
    n = len(rows)
    by_true = {t: 0 for t in TIERS}
    by_decided = {t: 0 for t in TIERS}
    agree = 0
    # conformal 覆盖：判 T1 且真 T1 的比例（P(true==T1|判T1) 反向即误判率）
    t1_pred = t1_pred_correct = 0
    t3_pred = t3_pred_correct = 0
    confusion = {d: {t: 0 for t in TIERS} for d in TIERS}
    for r in rows:
        decided = r["decided_tier"]
        true = r["true_tier"]
        if decided in by_decided:
            by_decided[decided] += 1
        if true in by_true:
            by_true[true] += 1
        if decided in confusion and true in confusion.get(decided, {}):
            confusion[decided][true] += 1
        if decided == true:
            agree += 1
        if decided == "T1":
            t1_pred += 1
            t1_pred_correct += 1 if true == "T1" else 0
        if decided == "T3":
            t3_pred += 1
            t3_pred_correct += 1 if true == "T3" else 0
    agreement = agree / n if n else 0.0
    t1_precision = t1_pred_correct / t1_pred if t1_pred else 1.0
    t3_precision = t3_pred_correct / t3_pred if t3_pred else 1.0
    # T1 占比（真值口径，§12 总验收 40–55%）
    t1_share = by_true["T1"] / n if n else 0.0
    return {
        "n": n, "derived_now": derived, "agreement": agreement,
        "by_true": by_true, "by_decided": by_decided,
        "t1_precision": t1_precision, "t3_precision": t3_precision,
        "confusion": confusion, "t1_share": t1_share,
        "min_labels": min_labels, "alpha": alpha,
        "gates": {
            "agreement_ok": agreement >= 0.85,
            "coverage_ok": t1_precision >= 1.0 - alpha - 0.02,
            "labels_ok": n >= min_labels,
            "t1_share_ok": 0.40 <= t1_share <= 0.55,
        },
    }


def write_report(data: dict, out_dir: Path, since_label: str) -> Path:
    out_dir.mkdir(parents=True, exist_ok=True)
    csv_path = out_dir / "E-G1_sense_report.csv"
    lines = ["metric,value"]
    lines.append(f"observations,{data['n']}")
    lines.append(f"derived_now,{data['derived_now']}")
    lines.append(f"agreement,{data['agreement']:.4f}")
    lines.append(f"t1_precision,{data['t1_precision']:.4f}")
    lines.append(f"t3_precision,{data['t3_precision']:.4f}")
    lines.append(f"t1_share,{data['t1_share']:.4f}")
    for t in TIERS:
        lines.append(f"by_true_{t},{data['by_true'][t]}")
        lines.append(f"by_decided_{t},{data['by_decided'][t]}")
    csv_path.write_text("\n".join(lines), encoding="utf-8")

    g = data["gates"]
    md = out_dir / "E-G1_sense_report.md"
    gates_pass = all(g.values())
    md_lines = [
        "# E-G1 语义分级 shadow 报告",
        "",
        f"- 生成时间：{time.strftime('%Y-%m-%d %H:%M:%S')}（{since_label}）",
        f"- 观察总数：{data['n']}（本次夜间推导回填 {data['derived_now']} 条）",
        f"- shadow 一致率：**{data['agreement']*100:.1f}%**（门：≥85% "
        f"{'✅' if g['agreement_ok'] else '❌'}）",
        f"- T1 决策精度（conformal 覆盖）：**{data['t1_precision']*100:.1f}%**"
        f"（门：≥{(1-data['alpha']-0.02)*100:.0f}% "
        f"{'✅' if g['coverage_ok'] else '❌'}）",
        f"- T1 真值占比：**{data['t1_share']*100:.1f}%**（目标区间 40–55% "
        f"{'✅' if g['t1_share_ok'] else '❌'}）",
        f"- 标签量：{data['n']} / min_labels {data['min_labels']} "
        f"{'✅' if g['labels_ok'] else '❌'}",
        "",
        "## 档位分布（真值 / 决策）",
        "",
        "| 档 | 真值 | 决策 |",
        "|---|---|---|",
    ]
    for t in TIERS:
        md_lines.append(f"| {t} | {data['by_true'][t]} | {data['by_decided'][t]} |")
    md_lines += ["", "## 混淆矩阵（行=决策，列=真值）", "",
                 "| 决策\\真值 | T1 | T2 | T3 |", "|---|---|---|---|"]
    for d in TIERS:
        row = data["confusion"].get(d, {})
        md_lines.append(f"| {d} | {row.get('T1',0)} | {row.get('T2',0)} "
                        f"| {row.get('T3',0)} |")
    md_lines += ["", f"## 晋升结论：{'**可申请 live**' if gates_pass else '**继续 collect/shadow**'}",
                 ""]
    md.write_text("\n".join(md_lines), encoding="utf-8")
    return md


def main() -> int:
    ap = argparse.ArgumentParser(description="E-G1 shadow 报告（T-G8）")
    ap.add_argument("--db", default="data/sense.sqlite3")
    ap.add_argument("--out", default="research/v2_experiments")
    ap.add_argument("--alpha", type=float, default=0.05)
    ap.add_argument("--min-labels", type=int, default=500)
    args = ap.parse_args()

    if not Path(args.db).exists():
        print(f"[report] 无观察库 {args.db}——先以 collect 模式积累观察。")
        return 1
    store = SenseStore.init_db(args.db)
    data = collect(store, args.min_labels, args.alpha, time.time())
    md = write_report(data, Path(args.out), time.strftime("%Y%m%d-%H%M"))
    print(f"[report] n={data['n']} 一致率={data['agreement']*100:.1f}% "
          f"T1 精度={data['t1_precision']*100:.1f}% T1 占比={data['t1_share']*100:.1f}%")
    print(f"[report] 报告: {md}")
    gates = data["gates"]
    print(f"[report] 晋升门：一致率{'✅' if gates['agreement_ok'] else '❌'} "
          f"覆盖{'✅' if gates['coverage_ok'] else '❌'} "
          f"标签{'✅' if gates['labels_ok'] else '❌'} "
          f"T1占比{'✅' if gates['t1_share_ok'] else '❌'}")
    return 0


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