sensorflow/nextgen/regression.py

"""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)