Spaces:
Paused
Paused
File size: 3,193 Bytes
9762da0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 | """Regression attribution: turn node states into a decision-first regression map.
On detection the question changes from "is there a regression?" to "WHERE and
HOW BIG?". Every node with data yields a row carrying the affected class /
condition / difficulty band, baseline and candidate values, absolute and
relative deltas, the anytime-valid CI, raw and effective sample sizes, the
e-values and the decision. The regression map is the REGRESSION-decided subset
sorted by magnitude — the payload a release manager (or the safety Regression
Gate) acts on.
"""
from __future__ import annotations
from typing import Dict, List
from sensorflow.seqeval.hierarchy import LEVELS, HierarchyController
from sensorflow.seqeval.sequential import DECISION_REGRESSION
def node_row(node, alpha_allocated: float) -> Dict:
t = node.test
clustering = node.clustering()
base_rate = t.sum_baseline / t.n_objects if t.n_objects else None
cand_rate = t.sum_candidate / t.n_objects if t.n_objects else None
abs_delta = None if base_rate is None else cand_rate - base_rate
rel_delta = (None if not base_rate else abs_delta / base_rate)
lo, hi = t.delta_interval()
return {
"node": node.key,
"level": node.level,
"level_name": LEVELS[node.level],
"safety_primary": node.safety_primary,
"suspect": node.suspect,
"metric": "recall",
"n": t.n_objects,
"n_clusters": clustering["n_clusters"],
"n_effective": clustering["n_effective"],
"icc": clustering["icc"],
"design_effect": clustering["design_effect"],
"baseline_value": None if base_rate is None else round(base_rate, 5),
"candidate_value": None if cand_rate is None else round(cand_rate, 5),
"abs_delta": None if abs_delta is None else round(abs_delta, 5),
"rel_delta": None if rel_delta is None else round(rel_delta, 5),
"delta_ci": [round(lo, 5), round(hi, 5)],
"e_regression": round(t.e_reg.e_value, 4),
"e_pass": round(t.e_pass.e_value, 4),
"bayes_p_regression": t.bayes_p_regression(),
"alpha_allocated": round(alpha_allocated, 6),
"decision": t.decision,
"decided_at_n": t.decided_at_n,
"discordant_pairs": {"regressions_n01": t.n01, "improvements_n10": t.n10},
}
def build_attribution(hier: HierarchyController) -> Dict:
rows: List[Dict] = []
for node in hier.nodes.values():
if node.test.n_objects == 0:
continue
rows.append(node_row(node, hier.alpha_for(node)))
regressions = [r for r in rows if r["decision"] == DECISION_REGRESSION]
regressions.sort(key=lambda r: r["abs_delta"] if r["abs_delta"] is not None else 0.0)
affected_strata = [r["node"] for r in regressions if r["level"] == 3]
return {
"regression_map": regressions,
"affected_strata": affected_strata,
"all_nodes": sorted(rows, key=lambda r: (r["level"], r["node"])),
"note": ("regression_map lists nodes with a confirmed anytime-valid "
"regression, most severe first; deltas are candidate minus "
"baseline in absolute metric points"),
}
|