Spaces:
Paused
Paused
File size: 4,054 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 | """Distribution + concentration analysis (deterministic).
Determines whether a failure pattern is uniform or concentrated across the
instrumented dimensions (construction zone, lighting, geography), computing
per-stratum failure rate vs the population baseline with relative risk, odds
ratio, Wilson CI, absolute risk difference and small-sample instability flags.
All statistics are exact deterministic computations; the Wilson interval is
reused from megaeval (single source of truth for that formula).
"""
from __future__ import annotations
from typing import Dict, List
import numpy as np
from sensorflow.agentic import data as data_mod
from sensorflow.agentic.models import ConcentrationAnalysis, StratumRisk
from sensorflow.megaeval.sampling import wilson_interval
MIN_EVENTS_FOR_STABLE_ESTIMATE = 5
RELATIVE_RISK_CONCENTRATION_THRESHOLD = 3.0
def _stratum_row(dim: str, name: str, mask: np.ndarray, flips: np.ndarray,
overall_rate: float) -> StratumRisk:
exposure = int(mask.sum())
events = int(flips[mask].sum())
rate = events / exposure if exposure else 0.0
lo, hi = wilson_interval(events, exposure)
out_events = int(flips[~mask].sum())
out_exposure = int((~mask).sum())
out_rate = out_events / out_exposure if out_exposure else 0.0
rr = (rate / out_rate) if out_rate > 0 else None
# odds ratio with Haldane-Anscombe 0.5 correction when any cell is zero
a, b = events, exposure - events
c, d = out_events, out_exposure - out_events
if min(a, b, c, d) == 0:
a, b, c, d = a + 0.5, b + 0.5, c + 0.5, d + 0.5
odds = (a / b) / (c / d) if b > 0 and d > 0 and c > 0 else None
return StratumRisk(
dimension=dim, stratum=name,
exposure=exposure,
exposure_share=round(exposure / flips.size, 4),
events=events,
stratum_rate=round(rate, 8),
baseline_rate=round(overall_rate, 8),
relative_risk=None if rr is None else round(rr, 3),
odds_ratio=None if odds is None else round(odds, 3),
risk_difference=round(rate - out_rate, 8),
rate_wilson_ci=[round(lo, 8), round(hi, 8)],
small_sample_flag=events < MIN_EVENTS_FOR_STABLE_ESTIMATE,
)
def analyze_concentration(failure_id: str, seed: int = data_mod.DEFAULT_SEED
) -> ConcentrationAnalysis:
arrays = data_mod.get_rate_arrays(seed)
flips = arrays.candidate_flip
overall_rate = float(flips.mean())
strata: List[StratumRisk] = []
dims: Dict[str, List] = {
"construction_zone": [("construction", arrays.construction),
("non_construction", ~arrays.construction)],
"lighting": [("night", arrays.night), ("day", ~arrays.night)],
"geo_bucket": [(data_mod.GEO_BUCKETS[i], arrays.geo == i)
for i in range(len(data_mod.GEO_BUCKETS))],
}
for dim, entries in dims.items():
for name, mask in entries:
strata.append(_stratum_row(dim, name, mask, flips, overall_rate))
total_events = int(flips.sum())
if total_events < MIN_EVENTS_FOR_STABLE_ESTIMATE:
determination = "insufficient_data"
concentrated = []
else:
concentrated = sorted({
s.dimension for s in strata
if s.relative_risk is not None
and s.relative_risk >= RELATIVE_RISK_CONCENTRATION_THRESHOLD
and not s.small_sample_flag})
determination = "concentrated" if concentrated else "uniform"
return ConcentrationAnalysis(
failure_id=failure_id,
determination=determination,
concentrated_dimensions=concentrated,
strata=strata,
method=(
"per-stratum rate vs complement with relative risk, "
"Haldane-Anscombe-corrected odds ratio, Wilson CI and absolute "
f"risk difference; a dimension is 'concentrated' when RR >= "
f"{RELATIVE_RISK_CONCENTRATION_THRESHOLD} on a stratum with >= "
f"{MIN_EVENTS_FOR_STABLE_ESTIMATE} events"),
)
|