Spaces:
Paused
Paused
| """Performance regression tracking across model/dataset versions and runs. | |
| Flags (spec §18): | |
| - performance regression: precision/recall/safety-recall below baseline-tolerance | |
| - tracking regression: IDF1 down, ID switches up, fragmentation up | |
| - annotation regression: 3D IoU down, position/orientation error up | |
| """ | |
| from __future__ import annotations | |
| import math | |
| from typing import Dict, List, Optional, Tuple | |
| from sensorflow.evaluation.records import ( | |
| EvalStore, | |
| RegressionMetricDelta, | |
| RegressionResult, | |
| new_id, | |
| ) | |
| # metric -> (tolerance, direction). "up" means higher is better. | |
| # "proportion" marks metrics that are rates in [0, 1] backed by trial counts, | |
| # eligible for the confidence-interval decision path in compare_runs. | |
| METRIC_SPECS: Dict[str, Dict] = { | |
| "precision": {"tolerance": 0.03, "better": "up", "kind": "performance", "proportion": True}, | |
| "recall": {"tolerance": 0.03, "better": "up", "kind": "performance", "proportion": True}, | |
| "safety_critical_recall": {"tolerance": 0.02, "better": "up", "kind": "performance", "proportion": True}, | |
| "idf1": {"tolerance": 0.05, "better": "up", "kind": "tracking", "proportion": True}, | |
| "id_swap_rate": {"tolerance": 0.02, "better": "down", "kind": "tracking", "proportion": True}, | |
| "fragmentation_rate": {"tolerance": 0.03, "better": "down", "kind": "tracking", "proportion": True}, | |
| "mean_iou_3d": {"tolerance": 0.03, "better": "up", "kind": "annotation", "proportion": False}, | |
| "mean_position_error": {"tolerance": 0.10, "better": "down", "kind": "annotation", "proportion": False}, | |
| "mean_orientation_error_deg": {"tolerance": 3.0, "better": "down", "kind": "annotation", "proportion": False}, | |
| } | |
| _Z_95 = 1.959963984540054 # two-sided 95% | |
| def wilson_interval(p: float, n: int, z: float = _Z_95) -> Tuple[float, float]: | |
| """Wilson score interval for a binomial proportion (never degenerate at 0/1).""" | |
| if n <= 0: | |
| return (0.0, 1.0) | |
| denom = 1.0 + z * z / n | |
| center = (p + z * z / (2 * n)) / denom | |
| half = z * math.sqrt(max(p * (1 - p) / n + z * z / (4 * n * n), 0.0)) / denom | |
| return (max(0.0, center - half), min(1.0, center + half)) | |
| def newcombe_delta_ci(p_base: float, n_base: int, | |
| p_cur: float, n_cur: int, | |
| z: float = _Z_95) -> Tuple[float, float]: | |
| """Newcombe hybrid-score CI for the difference (p_cur - p_base). | |
| Combines the two Wilson intervals; well-behaved for small n and | |
| boundary proportions, unlike the Wald interval. | |
| """ | |
| lb, ub = wilson_interval(p_base, n_base, z) | |
| lc, uc = wilson_interval(p_cur, n_cur, z) | |
| return (lc - ub, uc - lb) | |
| def compare_runs( | |
| store: EvalStore, | |
| current_metrics: Dict[str, Optional[float]], | |
| model_version: str, | |
| dataset_version: str, | |
| run_id: str, | |
| baseline_metrics: Optional[Dict[str, Optional[float]]] = None, | |
| baseline_version: Optional[str] = None, | |
| affected_classes: Optional[List[str]] = None, | |
| affected_scenarios: Optional[List[str]] = None, | |
| sample_sizes: Optional[Dict[str, Tuple[int, int]]] = None, | |
| ) -> RegressionResult: | |
| """Compare current run metrics vs a baseline run; persist RegressionResult. | |
| sample_sizes (optional, additive): metric -> (n_baseline, n_current) trial | |
| counts. When provided for a proportion metric, the regression decision uses | |
| a Newcombe 95% CI on the delta with the tolerance acting as the | |
| practical-significance margin: a regression is flagged only when the | |
| ENTIRE interval lies beyond the tolerance in the bad direction. This | |
| prevents noise-flagging on small runs and adds sensitivity on large runs. | |
| Without sample_sizes the legacy point-delta rule is preserved. | |
| For repeated/sequential monitoring across many runs, delegate to | |
| sensorflow.seqeval (anytime-valid e-processes, e-BH multiplicity) instead | |
| of calling this per run: fixed-level CIs are not valid under continuous | |
| monitoring. | |
| """ | |
| deltas: List[RegressionMetricDelta] = [] | |
| kinds: List[str] = [] | |
| regressed_any = False | |
| if baseline_metrics: | |
| for metric, spec in METRIC_SPECS.items(): | |
| base = baseline_metrics.get(metric) | |
| cur = current_metrics.get(metric) | |
| if base is None or cur is None: | |
| continue | |
| delta = cur - base | |
| ns = (sample_sizes or {}).get(metric) | |
| use_ci = (spec.get("proportion") and ns is not None | |
| and ns[0] > 0 and ns[1] > 0 | |
| and 0.0 <= base <= 1.0 and 0.0 <= cur <= 1.0) | |
| if use_ci: | |
| lo, hi = newcombe_delta_ci(base, ns[0], cur, ns[1]) | |
| if spec["better"] == "up": | |
| regressed = hi < -spec["tolerance"] | |
| else: | |
| regressed = lo > spec["tolerance"] | |
| elif spec["better"] == "up": | |
| regressed = delta < -spec["tolerance"] | |
| else: | |
| regressed = delta > spec["tolerance"] | |
| deltas.append(RegressionMetricDelta( | |
| metric=metric, | |
| baseline=round(float(base), 4), | |
| current=round(float(cur), 4), | |
| delta=round(float(delta), 4), | |
| tolerance=spec["tolerance"], | |
| regressed=regressed, | |
| )) | |
| if regressed: | |
| regressed_any = True | |
| if spec["kind"] not in kinds: | |
| kinds.append(spec["kind"]) | |
| result = RegressionResult( | |
| result_id=new_id("reg"), | |
| model_version=model_version, | |
| baseline_version=baseline_version, | |
| dataset_version=dataset_version, | |
| run_id=run_id, | |
| regression_detected=regressed_any, | |
| affected_classes=affected_classes or [], | |
| affected_scenarios=affected_scenarios or [], | |
| deltas=deltas, | |
| kinds=kinds, | |
| ) | |
| store.put("regressions", result) | |
| if regressed_any: | |
| store.audit("regression_detected", "RegressionResult", result.result_id, | |
| f"{model_version} vs {baseline_version}: kinds={kinds} classes={result.affected_classes}") | |
| return result | |
| def find_affected_classes( | |
| current_per_class: Dict[str, Dict[str, Optional[float]]], | |
| baseline_per_class: Dict[str, Dict[str, Optional[float]]], | |
| ) -> List[str]: | |
| """Classes whose precision or recall regressed beyond tolerance.""" | |
| affected = [] | |
| for cls, cur in current_per_class.items(): | |
| base = baseline_per_class.get(cls) | |
| if not base: | |
| continue | |
| for metric in ("precision", "recall"): | |
| b, c = base.get(metric), cur.get(metric) | |
| if b is not None and c is not None and (c - b) < -METRIC_SPECS[metric]["tolerance"]: | |
| affected.append(cls) | |
| break | |
| return affected | |