Spaces:
Paused
Paused
File size: 7,627 Bytes
5d10875 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 | """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))
|