Spaces:
Paused
Paused
| """Baseline-vs-candidate regression analysis across strata, with statistical | |
| and safety significance kept STRICTLY separate. | |
| Strata: global / class / scenario-type / safety / ODD. For each stratum we | |
| report absolute + relative delta, a cluster-robust confidence interval, | |
| sample size, and two independent verdicts: | |
| * statistical significance — is the delta distinguishable from noise? | |
| Delegated to sensorflow.seqeval (anytime-valid PairedSequentialTest on | |
| cluster means) — REUSE, not reimplementation. The descriptive CI uses | |
| seqeval's cluster machinery (units.py) as well. | |
| * safety significance — does the delta matter for safety? A deterministic | |
| policy question: the stratum's safety margin (tighter for safety strata) | |
| and the direction of the effect. A tiny-but-certain regression can be | |
| statistically significant yet safety-insignificant; a large regression on | |
| a small safety stratum can be safety-significant (blocking further | |
| evaluation, demanding more data) while statistically unresolved. The two | |
| are NEVER combined into one number. | |
| Launch recommendation policy (deterministic): | |
| * lineage invalid -> INVALID (never launchable) | |
| * any safety-significant regression that | |
| is also statistically confirmed -> DO_NOT_LAUNCH | |
| * safety-significant but statistically | |
| unresolved -> INSUFFICIENT_EVIDENCE | |
| * otherwise -> LAUNCH | |
| """ | |
| from __future__ import annotations | |
| from typing import Dict, List, Optional | |
| import numpy as np | |
| # REUSE: all sequential statistics from seqeval. | |
| from sensorflow.seqeval.sequential import ( | |
| DECISION_INSUFFICIENT, DECISION_PASS, DECISION_REGRESSION, | |
| PairedSequentialTest, | |
| ) | |
| from sensorflow.seqeval.units import cluster_summary, cluster_units | |
| from sensorflow.nextgen.lineage import validate_lineage | |
| from sensorflow.nextgen.models import LaunchRecommendation, LineageRecord | |
| # Safety margins per stratum kind (absolute metric points). | |
| SAFETY_MARGINS = {"safety": 0.005, "odd": 0.010, "scenario": 0.010, | |
| "class": 0.015, "global": 0.010} | |
| STATISTICAL_MARGIN = 0.005 # practical-significance margin for the e-process | |
| ALPHA = 0.05 | |
| def analyze_stratum(name: str, kind: str, | |
| baseline: np.ndarray, candidate: np.ndarray, | |
| cluster_ids: np.ndarray, | |
| data_label: str = "SIMULATED") -> Dict: | |
| """Full per-stratum record from paired per-unit outcomes.""" | |
| b = np.asarray(baseline, dtype=np.float64) | |
| c = np.asarray(candidate, dtype=np.float64) | |
| d = c - b | |
| test = PairedSequentialTest(delta=STATISTICAL_MARGIN, alpha=ALPHA) | |
| means, _sizes = cluster_units(d, cluster_ids) | |
| test.update_clusters(means) | |
| test.record_objects(b.astype(bool), c.astype(bool)) | |
| decision = test.evaluate() | |
| lo, hi = test.delta_interval() | |
| base_rate = float(b.mean()) if b.size else None | |
| cand_rate = float(c.mean()) if c.size else None | |
| delta_abs = None if base_rate is None else cand_rate - base_rate | |
| delta_rel = (None if not base_rate else delta_abs / base_rate) | |
| margin = SAFETY_MARGINS.get(kind, 0.010) | |
| # Safety significance: the point estimate breaches the stratum's safety | |
| # margin in the harmful direction. Deliberately a POLICY statement about | |
| # magnitude, independent of the statistical verdict. | |
| safety_significant = delta_abs is not None and delta_abs < -margin | |
| statistically_significant = decision == DECISION_REGRESSION | |
| return { | |
| "stratum": name, "kind": kind, "data_label": data_label, | |
| "n": int(b.size), | |
| "clustering": cluster_summary(d, cluster_ids), | |
| "baseline_rate": _r(base_rate), "candidate_rate": _r(cand_rate), | |
| "delta_abs": _r(delta_abs), "delta_rel": _r(delta_rel), | |
| "delta_ci": [_r(lo), _r(hi)], | |
| "ci_method": "anytime-valid empirical-Bernstein confidence sequence " | |
| "on cluster means (seqeval)", | |
| "statistical": { | |
| "decision": decision, | |
| "significant_regression": statistically_significant, | |
| "e_regression": round(test.e_reg.e_value, 4), | |
| "e_pass": round(test.e_pass.e_value, 4), | |
| "margin": STATISTICAL_MARGIN, "alpha": ALPHA, | |
| }, | |
| "safety": { | |
| "margin": margin, | |
| "significant": bool(safety_significant), | |
| "note": "safety significance is a policy magnitude judgment; it " | |
| "is never derived from, nor equated with, the " | |
| "statistical verdict", | |
| }, | |
| } | |
| def launch_recommendation(run_id: str, strata: List[Dict], | |
| lineage: Optional[LineageRecord], | |
| data_labels: List[str]) -> LaunchRecommendation: | |
| lineage_ok, lineage_reasons = validate_lineage(lineage) | |
| blockers: List[str] = [] | |
| stat_summary: Dict = {"regressions": [], "passes": [], "unresolved": []} | |
| safety_summary: Dict = {"significant_regressions": [], "margins": SAFETY_MARGINS} | |
| for s in strata: | |
| dec = s["statistical"]["decision"] | |
| if dec == DECISION_REGRESSION: | |
| stat_summary["regressions"].append(s["stratum"]) | |
| elif dec == DECISION_PASS: | |
| stat_summary["passes"].append(s["stratum"]) | |
| else: | |
| stat_summary["unresolved"].append(s["stratum"]) | |
| if s["safety"]["significant"]: | |
| safety_summary["significant_regressions"].append( | |
| {"stratum": s["stratum"], "delta_abs": s["delta_abs"], | |
| "statistical_decision": dec}) | |
| if not lineage_ok: | |
| recommendation = "INVALID" | |
| blockers.extend(lineage_reasons) | |
| blockers.append("run is INVALID for launch purposes: incomplete lineage") | |
| else: | |
| confirmed = [r for r in safety_summary["significant_regressions"] | |
| if r["statistical_decision"] == DECISION_REGRESSION] | |
| unresolved = [r for r in safety_summary["significant_regressions"] | |
| if r["statistical_decision"] == DECISION_INSUFFICIENT] | |
| if confirmed: | |
| recommendation = "DO_NOT_LAUNCH" | |
| for r in confirmed: | |
| blockers.append( | |
| f"safety-significant regression in {r['stratum']} " | |
| f"(delta {r['delta_abs']:+.4f}), statistically confirmed") | |
| elif unresolved: | |
| recommendation = "INSUFFICIENT_EVIDENCE" | |
| for r in unresolved: | |
| blockers.append( | |
| f"safety-significant point estimate in {r['stratum']} " | |
| f"(delta {r['delta_abs']:+.4f}) not yet statistically " | |
| f"resolved — needs more evidence, not a launch") | |
| else: | |
| recommendation = "LAUNCH" | |
| return LaunchRecommendation( | |
| run_id=run_id, recommendation=recommendation, blockers=blockers, | |
| statistical_significance=stat_summary, | |
| safety_significance=safety_summary, | |
| lineage_valid=lineage_ok, | |
| data_labels=list(dict.fromkeys(data_labels))) # type: ignore[arg-type] | |
| def _r(v: Optional[float], nd: int = 5) -> Optional[float]: | |
| return None if v is None else round(float(v), nd) | |