"""StatisticalRegressionAgent — mostly deterministic. Establishes, before anyone treats "~0.01%" as real: the denominator, the sample size, confidence intervals, statistical significance, power/MDE and small-sample instability. This agent is a THIN ORCHESTRATOR over sensorflow.seqeval — the anytime-valid sequential machinery (PairedSequentialTest: empirical-Bernstein confidence sequence + betting e-processes) and the MDE approximation are imported from seqeval, not reimplemented. Wilson intervals come from megaeval.sampling (single source of truth). Rare-event handling: when event counts are small, the practical-margin sequential test is honest but low-powered, so significance for the rate DELTA is additionally established with the exact conditional binomial (McNemar-exact) test over discordant pairs; both results are reported with their methods, and the seqeval three-outcome decision is never coerced. """ from __future__ import annotations from typing import Any, Dict, Tuple import numpy as np from scipy import stats as sstats from sensorflow.agentic import data as data_mod from sensorflow.agentic.agents.base import BaseAgent, compact, no_escalation from sensorflow.agentic.models import (AgentEscalation, RateEstimate, StatisticalAssessment) from sensorflow.megaeval.sampling import wilson_interval from sensorflow.seqeval.sequential import PairedSequentialTest, approx_mde ALPHA = 0.05 SEQ_DELTA_MARGIN = 5.0e-5 # practical-significance margin for the flip rate SMALL_SAMPLE_EVENT_FLOOR = 50 class StatisticalRegressionAgent(BaseAgent): name = "statistical_regression" version = "1.0.0" epistemic_status = "OBSERVED" # this agent reports measurements def assess(self, seed: int = data_mod.DEFAULT_SEED) -> StatisticalAssessment: arrays = data_mod.get_rate_arrays(seed) n = arrays.n b_events = int(arrays.baseline_flip.sum()) c_events = int(arrays.candidate_flip.sum()) b_rate, c_rate = b_events / n, c_events / n b_lo, b_hi = wilson_interval(b_events, n) c_lo, c_hi = wilson_interval(c_events, n) # ---- exact rare-event significance (McNemar-exact / conditional # binomial over discordant pairs) -------------------------------- n01 = int((arrays.baseline_flip & ~arrays.candidate_flip).sum()) n10 = int((~arrays.baseline_flip & arrays.candidate_flip).sum()) if n01 + n10 > 0: exact_p = float(sstats.binomtest(n10, n01 + n10, 0.5, alternative="greater").pvalue) else: exact_p = None significant = exact_p is not None and exact_p < ALPHA # ---- seqeval delegation: anytime-valid paired sequential test ---- # success = "no flip"; d = candidate - baseline (negative = worse) b_success = (~arrays.baseline_flip).astype(np.float64) c_success = (~arrays.candidate_flip).astype(np.float64) d = c_success - b_success cluster_sums = np.bincount(arrays.cluster_id, weights=d) cluster_counts = np.bincount(arrays.cluster_id) cluster_means = cluster_sums / np.maximum(cluster_counts, 1) seq_test = PairedSequentialTest(delta=SEQ_DELTA_MARGIN, alpha=ALPHA) seq_test.update_clusters(cluster_means.tolist()) seq_test.record_objects(b_success.astype(bool), c_success.astype(bool)) seq_decision = seq_test.evaluate() seq_snapshot = seq_test.snapshot() # ---- power / MDE at the achieved sample size --------------------- nu = (seq_test.n01 + seq_test.n10) / max(seq_test.n_objects, 1) est = seq_test.delta_estimate() or 0.0 var_d = max(nu - est ** 2, 1e-12) mde = approx_mde(var_d, n_eff=float(len(cluster_means)), delta=SEQ_DELTA_MARGIN, alpha=ALPHA) observed_abs_delta = abs(c_rate - b_rate) power_note = ( "sequential-margin test is UNDERPOWERED at this n for the " "observed delta" if (mde is not None and mde > observed_abs_delta) else "sequential-margin test adequately powered") # ---- small-sample instability flags ------------------------------- flags = [] if c_events < SMALL_SAMPLE_EVENT_FLOOR: flags.append(f"candidate event count {c_events} < " f"{SMALL_SAMPLE_EVENT_FLOOR}: rate estimate unstable") if b_events < 10: flags.append(f"baseline event count {b_events} < 10: relative " "risk denominator unstable") if (c_hi - c_lo) > c_rate: flags.append("candidate CI width exceeds the point estimate") return StatisticalAssessment( baseline=RateEstimate(events=b_events, denominator=n, rate=round(b_rate, 8), wilson_ci=[round(b_lo, 8), round(b_hi, 8)]), candidate=RateEstimate(events=c_events, denominator=n, rate=round(c_rate, 8), wilson_ci=[round(c_lo, 8), round(c_hi, 8)]), absolute_delta=round(c_rate - b_rate, 8), relative_delta=(round(c_rate / b_rate, 3) if b_rate > 0 else None), significant=significant, significance_method=( "exact conditional binomial (McNemar-exact) over discordant " f"pairs: n10={n10} candidate-only vs n01={n01} baseline-only, " f"H0 p=0.5, one-sided"), exact_binomial_p=exact_p, seqeval={ "delegated_to": "sensorflow.seqeval.sequential.PairedSequentialTest", "test_method": ("paired cluster-mean empirical-Bernstein " "confidence sequence + one-sided betting " "e-process (anytime-valid)"), "delta_margin": SEQ_DELTA_MARGIN, "alpha": ALPHA, "decision": seq_decision, "snapshot": seq_snapshot, "clusters_fed": int(len(cluster_means)), "note": ("three-outcome decision reported as-is; " "INSUFFICIENT_EVIDENCE is never coerced to PASS"), }, power_mde={ "mde_abs": mde, "observed_abs_delta": round(observed_abs_delta, 8), "assessment": power_note, "method": "sensorflow.seqeval.sequential.approx_mde", }, small_sample_flags=flags, rare_event_handling=( "event counts are small, so exact binomial inference is used " "for the significance claim; the seqeval sequential test is " "reported alongside with its own (unforced) decision"), ) def _analyze(self, failure_id: str, seed: int = data_mod.DEFAULT_SEED, **inputs) -> Tuple[Dict[str, Any], float, str, AgentEscalation]: assessment = self.assess(seed) conf = 0.9 if assessment.significant and not assessment.small_sample_flags \ else (0.7 if assessment.significant else 0.4) return (assessment.model_dump(), conf, "deterministic statistics; confidence reflects significance " "and sample stability", no_escalation()) def _llm_prompt(self, failure_id, output, **inputs): return ("Explain these regression statistics to a launch-review " "audience without changing any number: " + compact(output))