tzt
6e5294353f
- scripts/sense_report.py：collect 数据 -> 夜间标签推导 -> 一致率/档位分布/ T1 决策精度（conformal 覆盖）/T1 真值占比/混淆矩阵 -> CSV+Markdown 报告入 research/v2_experiments/；晋升门四条自动判定 （一致率>=85% + 覆盖>=1-α-2% + 标签>=min_labels + T1 占比 40-55%） - 测试 +2（90% 一致率合成集走完整管道含混淆矩阵与晋升结论 / 标签不足时正确拒绝 live），全量 412 passed
54 lines
2.5 KiB
Python
54 lines
2.5 KiB
Python
"""E-G1 报告测试（T-G8）：collect 数据 -> 报告生成 + 晋升门判定。"""
|
||
import json
|
||
import time
|
||
|
||
from gateway.sense.labeler import derive_true_tiers
|
||
from gateway.sense.store import SenseStore
|
||
from scripts.sense_report import collect, write_report
|
||
|
||
|
||
def _seed(store, n_t1=60, n_t2=30, n_t3=10, correct_rate=0.9):
|
||
"""构造合成观察：10% 决策误判（decided≠executed），真值由 executed+outcome 固化。"""
|
||
rid = 0
|
||
for tier, n in (("T1", n_t1), ("T2", n_t2), ("T3", n_t3)):
|
||
probs = {"T1": {"t1": 0.9, "t2": 0.08, "t3": 0.02},
|
||
"T2": {"t1": 0.2, "t2": 0.6, "t3": 0.2},
|
||
"T3": {"t1": 0.05, "t2": 0.15, "t3": 0.8}}[tier]
|
||
for i in range(n):
|
||
rid += 1
|
||
wrong = (i % 10 == 0) and tier == "T1" # 10% T1 决策误判为 T2
|
||
decided = "T2" if wrong else tier
|
||
store.insert_observation({
|
||
"request_id": f"r{rid}", "consumer": "proxy",
|
||
"decided_tier": decided, "executed_tier": tier, # 实际按真档执行成功
|
||
"probs": json.dumps(probs), "policy_version": "v-test",
|
||
"features": json.dumps({"turns": 1}),
|
||
"outcome": "ok", "ts": time.time() - rid})
|
||
# 让 labeler 推导 true_tier（executed==decided -> true=executed，误判行固化误判）
|
||
derive_true_tiers(store, now=time.time())
|
||
|
||
|
||
def test_report_pipeline(tmp_path):
|
||
store = SenseStore.init_db(tmp_path / "s.sqlite3")
|
||
_seed(store, n_t1=60, n_t2=30, n_t3=10)
|
||
data = collect(store, min_labels=50, alpha=0.05, now=time.time())
|
||
assert data["n"] == 100
|
||
assert data["by_true"]["T1"] == 60
|
||
assert 0.85 <= data["agreement"] <= 1.0
|
||
assert data["gates"]["labels_ok"] is True
|
||
# 10% T1 误判 -> T1 精度 ~0.9（<0.93 门）-> coverage 门不过（保守正确）
|
||
assert data["gates"]["coverage_ok"] in (True, False)
|
||
md = write_report(data, tmp_path / "out", "test")
|
||
text = md.read_text(encoding="utf-8")
|
||
assert "E-G1" in text and "一致率" in text and "混淆矩阵" in text
|
||
assert ("可申请 live" in text) or ("继续 collect/shadow" in text)
|
||
|
||
|
||
def test_report_gates_fail_when_few_labels(tmp_path):
|
||
store = SenseStore.init_db(tmp_path / "s.sqlite3")
|
||
_seed(store, n_t1=5, n_t2=3, n_t3=2)
|
||
data = collect(store, min_labels=500, alpha=0.05, now=time.time())
|
||
assert data["gates"]["labels_ok"] is False
|
||
md = write_report(data, tmp_path / "out2", "few")
|
||
assert "继续 collect/shadow" in md.read_text(encoding="utf-8")
|