"""ROTR metric hierarchy + baseline-vs-candidate regression. Metrics (architecture doc §6): * ROTR recall / precision proxy (false-accusation rate on planted NON-violations) — the scenario bank supplies a true denominator. * Safety-critical ROTR recall — EXPOSURE-DERIVED weights: the weight of a stratum is its measured fraction of SAFETY_CRITICAL consequence outcomes on the evaluated bank, normalized by the overall fraction (documented calibration, monotone in measured harm — NOT arbitrary class multipliers). ILLUSTRATIVE on synthetic data by construction. * Behavioral Consequence Rate (BCR), Critical Failure Rate (CFR, with a Wilson interval — REUSE rca.stats.wilson_ci). * Regression: claims are delegated to seqeval's sequential engine (REUSE seqeval.evaluate_regression over a megaeval population, rare-event cluster handling included) and forced into the SIX-outcome distinction: observed / statistically significant / practically significant / safety-critical / insufficient evidence / distribution-driven (the last via cohort-mix divergence, megaeval-shift + rca comparison-validity concepts). """ from __future__ import annotations from typing import Dict, List, Optional # REUSE (landed, read-only). from sensorflow.rca.stats import js_divergence_categorical, wilson_ci from sensorflow.rotr import SOFTWARE_VERSION from sensorflow.rotr.models import Provenance, RegressionResult CALIBRATION_VERSION = "rotr-weights-1.0.0" CONSEQUENCE_RANK = {"NO_MATERIAL_CONSEQUENCE": 0, "DEGRADED_COMFORT": 1, "PLANNER_INTERVENTION": 2, "SAFETY_CRITICAL": 3} # ILLUSTRATIVE practical-significance thresholds. PRACTICAL_ROTR_RECALL_PP = 0.05 PRACTICAL_SC_RECALL_PP = 0.02 DISTRIBUTION_JS_THRESHOLD = 0.05 # ------------------------------------------------------------ run metrics def _stratum(scenario_summary: Dict) -> str: return f"{scenario_summary['kind']}|{scenario_summary['vulnerability']}" def compute_metrics(scenarios: List[Dict], violations: List[Dict], consequences: Dict[str, Dict]) -> Dict: """scenarios: summaries with planted truth; violations: detected records; consequences: violation_id -> {consequence_class}.""" by_scenario: Dict[str, List[Dict]] = {} for v in violations: by_scenario.setdefault(v["scenario_id"], []).append(v) committed = [s for s in scenarios if s["committed"]] negatives = [s for s in scenarios if not s["committed"]] def _detected(s: Dict) -> bool: return any(v["rule_id"] == s["expected_rule_id"] for v in by_scenario.get(s["scenario_id"], [])) n_detected = sum(1 for s in committed if _detected(s)) rotr_recall = n_detected / len(committed) if committed else None false_acc = sum(1 for s in negatives if by_scenario.get(s["scenario_id"])) false_accusation_rate = false_acc / len(negatives) if negatives else None # --- exposure-derived weight calibration (documented, monotone) ------ strata: Dict[str, Dict] = {} for s in committed: st = strata.setdefault(_stratum(s), {"n": 0, "detected": 0, "sc": 0}) st["n"] += 1 if _detected(s): st["detected"] += 1 cons = [consequences.get(v["violation_id"], {}).get("consequence_class") for v in by_scenario.get(s["scenario_id"], [])] if "SAFETY_CRITICAL" in cons or s.get("planted_sc_hint"): st["sc"] += 1 total_n = sum(st["n"] for st in strata.values()) or 1 total_sc = sum(st["sc"] for st in strata.values()) overall_harm = total_sc / total_n if total_n else 0.0 calibration = {} for name, st in sorted(strata.items()): harm = st["sc"] / st["n"] if st["n"] else 0.0 # weight = measured harm exposure relative to overall harm; floor at # a small epsilon so zero-harm strata still count, never dominate. weight = (harm / overall_harm) if overall_harm > 0 else 1.0 weight = max(weight, 0.1) calibration[name] = {"n": st["n"], "harm_fraction": round(harm, 4), "weight": round(weight, 4), "recall": round(st["detected"] / st["n"], 4) if st["n"] else None} sc_strata = {k: v for k, v in calibration.items() if v["harm_fraction"] > 0 or k.endswith("|VRU")} num = sum(v["weight"] * v["recall"] * v["n"] for v in sc_strata.values() if v["recall"] is not None) den = sum(v["weight"] * v["n"] for v in sc_strata.values()) sc_rotr_recall = (num / den) if den else None # --- consequence-level rates ---------------------------------------- detected_vids = [v["violation_id"] for v in violations] consequential = sum( 1 for vid in detected_vids if CONSEQUENCE_RANK.get( consequences.get(vid, {}).get("consequence_class", ""), 0) >= 1) bcr = consequential / len(detected_vids) if detected_vids else None sc_count = sum(1 for vid in detected_vids if consequences.get(vid, {}).get("consequence_class") == "SAFETY_CRITICAL") n_scen = len(scenarios) cfr = sc_count / n_scen if n_scen else None cfr_ci = wilson_ci(sc_count, n_scen) if n_scen else (0.0, 1.0) # --- cohort breakdowns ---------------------------------------------- cohorts: Dict[str, Dict] = {} for s in committed: for key in (f"rule:{s['expected_rule_id']}", f"visibility:{s['visibility']}", f"lighting:{s['lighting']}", f"vulnerability:{s['vulnerability']}"): c = cohorts.setdefault(key, {"n": 0, "detected": 0}) c["n"] += 1 c["detected"] += 1 if _detected(s) else 0 for c in cohorts.values(): c["recall"] = round(c["detected"] / c["n"], 4) if c["n"] else None # --- stack-behavior rates (what regresses between model versions) ---- # ROTR recall measures the EVALUATOR; violation rates measure the STACK: # committed violations per planted opportunity, per vulnerability cohort. opportunities = [s for s in scenarios if s["is_violation_opportunity"]] behavior_rates: Dict[str, Dict] = {} for coh in ("VRU", "NON_VRU", "ALL"): opp = [s for s in opportunities if coh == "ALL" or s["vulnerability"] == coh] com = [s for s in opp if s["committed"]] behavior_rates[coh] = { "opportunities": len(opp), "committed": len(com), "violation_rate": round(len(com) / len(opp), 4) if opp else None} return { "n_scenarios": n_scen, "n_committed_violations": len(committed), "n_planted_non_violations": len(negatives), "n_detected": n_detected, "n_violation_records": len(violations), "rotr_recall": _r(rotr_recall), "false_accusation_rate": _r(false_accusation_rate), "sc_rotr_recall": _r(sc_rotr_recall), "bcr": _r(bcr), "cfr": _r(cfr), "cfr_wilson_95": [round(cfr_ci[0], 4), round(cfr_ci[1], 4)], "weight_calibration": { "version": CALIBRATION_VERSION, "method": "stratum weight = measured SAFETY_CRITICAL harm " "fraction / overall harm fraction (exposure-derived, " "monotone in measured harm; floored at 0.1). " "ILLUSTRATIVE: calibrated on the synthetic bank itself.", "strata": calibration, }, "cohorts": cohorts, "behavior_rates": behavior_rates, "surrogate_caveat": "TTC/PET are surrogate conflict measures; they " "prioritize triage and never gate alone.", } def _r(v: Optional[float]) -> Optional[float]: return None if v is None else round(float(v), 4) # ------------------------------------------------------------ regression def _seqeval_effects(baseline: Dict, candidate: Dict) -> Dict[str, float]: """Translate stack-behavior deltas into seqeval stratum effects. The regressing quantity between MODEL versions is the committed violation RATE per cohort (ROTR recall measures the evaluator and is invariant across stack profiles). A rate INCREASE is a performance DECREASE, so the effect is the negated rate delta in probability points. seqeval strata are megaeval "class|lighting" strata: the VRU cohort maps onto the pedestrian strata, NON_VRU onto vehicle strata. """ effects: Dict[str, float] = {} b = baseline.get("behavior_rates", {}) c = candidate.get("behavior_rates", {}) mapping = {"VRU": "pedestrian|night", "NON_VRU": "vehicle|day", "ALL": "__global__"} for coh, stratum in mapping.items(): rb = (b.get(coh) or {}).get("violation_rate") rc = (c.get(coh) or {}).get("violation_rate") if rb is None or rc is None: continue delta = -(rc - rb) if abs(delta) > 1e-9: effects[stratum] = round(delta, 4) return effects def evaluate_candidate(regression_id: str, baseline_run: Dict, candidate_run: Dict, seed: int = 11) -> RegressionResult: bm, cm = baseline_run["metrics"], candidate_run["metrics"] deltas = {} for k in ("rotr_recall", "sc_rotr_recall", "bcr", "cfr", "false_accusation_rate"): if bm.get(k) is not None and cm.get(k) is not None: deltas[k] = round(cm[k] - bm[k], 4) for coh in ("VRU", "NON_VRU", "ALL"): rb = (bm.get("behavior_rates", {}).get(coh) or {}).get("violation_rate") rc = (cm.get("behavior_rates", {}).get(coh) or {}).get("violation_rate") if rb is not None and rc is not None: deltas[f"violation_rate_{coh}"] = round(rc - rb, 4) # --- delegation: statistical claims go through seqeval --------------- seq_payload = None seq_decision = None try: from sensorflow.megaeval import population as pop_mod from sensorflow.seqeval.controller import evaluate_regression pop_name = "rotr-regression-pop" pop_id = next((p["population_id"] for p in pop_mod.list_populations() if p.get("name") == pop_name), None) if pop_id is None: pop_id = pop_mod.generate_population( pop_name, num_objects=60_000, seed=17)["population_id"] effects = _seqeval_effects(bm, cm) seq_payload = evaluate_regression( pop_id, {"model_version": baseline_run.get("model_version", "baseline")}, {"model_version": candidate_run.get("model_version", "candidate"), "effects": effects}, policy={"target_n": 6000, "safety_floor": 800, "min_per_stratum": 120, "escalation": {"enabled": True, "max_extra_per_stratum": 600, "batch_objects": 300}}, seed=seed) seq_payload["translated_effects"] = effects seq_decision = seq_payload.get("decision") except Exception as e: # honest degradation seq_payload = {"error": f"seqeval delegation unavailable: {e}"} # --- distribution-driven check (megaeval shift + rca concepts) ------- # Compare the OPPORTUNITY mix (bank composition), not the committed mix: # committed-mix changes are model behavior, which is the signal, not a # confound. Same-seed banks have identical opportunity mixes (JS = 0). def _mix(run: Dict) -> List[str]: return [f"{s['kind']}|{s['visibility']}|{s['lighting']}" for s in run["scenario_summaries"]] js = js_divergence_categorical(_mix(baseline_run), _mix(candidate_run)) \ if baseline_run.get("scenario_summaries") and \ candidate_run.get("scenario_summaries") else 0.0 distribution_driven = js > DISTRIBUTION_JS_THRESHOLD dist_note = (f"cohort-mix JS divergence {js:.4f} " + ("EXCEEDS" if distribution_driven else "below") + f" {DISTRIBUTION_JS_THRESHOLD} — " + ("observed deltas may reflect a different committed-" "violation mix, not model behavior; verify before acting" if distribution_driven else "committed-violation mix comparable between runs")) # --- six-outcome distinction ----------------------------------------- six = { "observed_difference": any(abs(d) > 0 for d in deltas.values()), "statistically_significant": seq_decision == "REGRESSION", "practically_significant": ( abs(deltas.get("violation_rate_ALL", 0.0)) >= PRACTICAL_ROTR_RECALL_PP or abs(deltas.get("violation_rate_VRU", 0.0)) >= PRACTICAL_SC_RECALL_PP or abs(deltas.get("rotr_recall", 0.0)) >= PRACTICAL_ROTR_RECALL_PP), "safety_critical": (deltas.get("violation_rate_VRU", 0.0) >= PRACTICAL_SC_RECALL_PP or deltas.get("sc_rotr_recall", 0.0) <= -PRACTICAL_SC_RECALL_PP or deltas.get("cfr", 0.0) > 0), "insufficient_evidence": seq_decision == "INSUFFICIENT" or seq_decision is None, "distribution_driven": distribution_driven, } if six["safety_critical"] and six["statistically_significant"]: primary = "SAFETY_CRITICAL_REGRESSION" elif six["statistically_significant"]: primary = "STATISTICALLY_SIGNIFICANT_REGRESSION" elif six["insufficient_evidence"]: primary = "INSUFFICIENT_EVIDENCE" elif six["distribution_driven"]: primary = "DISTRIBUTION_DRIVEN" elif six["practically_significant"]: primary = "PRACTICALLY_SIGNIFICANT_DIFFERENCE" elif six["observed_difference"]: primary = "OBSERVED_DIFFERENCE_ONLY" else: primary = "NO_DIFFERENCE" return RegressionResult( regression_id=regression_id, baseline_run_id=baseline_run["run_id"], candidate_run_id=candidate_run["run_id"], baseline_model=baseline_run.get("model_version", ""), candidate_model=candidate_run.get("model_version", ""), metric_deltas=deltas, six_outcomes=six, primary_outcome=primary, seqeval=seq_payload, distribution_note=dist_note, provenance=Provenance( software_version=f"{SOFTWARE_VERSION}/metrics", calibration_version=CALIBRATION_VERSION, source="SYNTHETIC"))