01 - Pass-rate with Wilson 95% CI¶

import hashlib
import json
import os
from datetime import datetime, timedelta, timezone
from importlib.metadata import version, PackageNotFoundError
from pathlib import Path

import jsonschema
from statsmodels.stats.proportion import proportion_confint

REPO_ROOT = Path.cwd()
if (REPO_ROOT / 'notebooks').exists() is False and (REPO_ROOT.parent / 'notebooks').exists():
    REPO_ROOT = REPO_ROOT.parent
RESULTS_DIR = REPO_ROOT / 'docs' / 'results'
DERIVED_DIR = REPO_ROOT / 'docs' / 'data' / 'derived'
SCHEMA_PATH = REPO_ROOT / 'schemas' / 'v2' / 'derived-metric.schema.json'
DERIVED_DIR.mkdir(parents=True, exist_ok=True)
print('repo root:', REPO_ROOT)

EXCLUDE_PREFIXES = ('bluefin-aurora-', 'bluefin-bazzite-')

raw_inputs = []  # list of (path, bytes, sha256_hex, parsed)
for path in sorted(RESULTS_DIR.glob('*.json')):
    if any(path.name.startswith(p) for p in EXCLUDE_PREFIXES):
        continue
    data = path.read_bytes()
    h = hashlib.sha256(data).hexdigest()
    raw_inputs.append((path, data, h, json.loads(data)))

concat = ''.join(sorted(h for _, _, h, _ in raw_inputs)).encode('ascii')
inputs_sha256 = hashlib.sha256(concat).hexdigest()
print('loaded', len(raw_inputs), 'input files')
print('inputs_sha256:', inputs_sha256)

AUDIENCE_WINDOWS = [
    ('contributor', 24),
    ('user', 24 * 7),
    ('downstream', 24 * 30),
]

def wilson_ci(passes: int, n: int):
    if n <= 0:
        return None, None, None
    lo, hi = proportion_confint(count=passes, nobs=n, alpha=0.05, method='wilson')
    return passes / n, float(lo), float(hi)

def parse_dt(s: str) -> datetime:
    return datetime.fromisoformat(s.replace('Z', '+00:00')).astimezone(timezone.utc)

def in_window(run_dt: datetime, now: datetime, hours: int) -> bool:
    return (now - run_dt) <= timedelta(hours=hours) and run_dt <= now

def history_date(h):
    return h.get('run_date') or h.get('run')

all_dates = []
for _, _, _, parsed in raw_inputs:
    for h in parsed.get('history', []) or []:
        d = history_date(h)
        if d:
            all_dates.append(parse_dt(d))
now_ref = max(all_dates)
print('reference now (max run_date):', now_ref.isoformat())

def split_variant_branch(s: str):
    if s.endswith('-testing'):
        return s[:-len('-testing')], 'testing'
    if s.endswith('-stable'):
        return s[:-len('-stable')], 'stable'
    return s, 'unknown'

RAW_INPUT_URL_PREFIX = 'https://github.com/projectbluefin/testing-lab/blob/main/'

metrics = []
for path, _, file_sha, parsed in raw_inputs:
    variant_str = parsed.get('variant')
    suite = parsed.get('suite')
    if not variant_str or not suite:
        continue
    variant, branch = split_variant_branch(variant_str)
    history = parsed.get('history', []) or []
    rel_path = f'docs/results/{path.name}'
    evidence = [{
        'kind': 'raw-input',
        'url': RAW_INPUT_URL_PREFIX + rel_path,
        'path': rel_path,
        'sha256': file_sha,
    }]

    for audience, hours in AUDIENCE_WINDOWS:
        in_win = [h for h in history if history_date(h) and in_window(parse_dt(history_date(h)), now_ref, hours)]
        total_scenarios = sum(int(h.get('scenarios', 0)) for h in in_win)
        total_failed = sum(int(h.get('failed', 0)) for h in in_win)
        passes = total_scenarios - total_failed
        p_hat, lo, hi = wilson_ci(passes, total_scenarios)
        n = total_scenarios
        confidence = 'high' if n >= 30 else 'medium' if n >= 10 else 'low'
        state = 'fresh' if len(in_win) > 0 else 'unknown'

        record = {
            'id': f'trust_window_pass_rate.{variant}.{branch}.{suite}.{audience}',
            'label': f'T pass-rate ({audience} window) - {variant}/{branch}/{suite}',
            'value': p_hat,
            'ci_lower': lo,
            'ci_upper': hi,
            'ci_level': 0.95,
            'n': n,
            'unit': 'proportion',
            'method': 'wilson_score_95',
            'method_citation': 'Wilson, E. B. (1927). Probable inference, the law of succession, and statistical inference. JASA 22(158):209-212.',
            'window': {'type': 'rolling_hours', 'size': hours, 'audience': audience},
            'confidence': confidence,
            'state': state,
            'context': {
                'variant': variant,
                'branch': branch,
                'suite': suite,
                'audience': audience,
            },
            'formula': 'p_hat = passes/n; CI = Wilson score interval at 95%',
            'numerator': passes,
            'denominator': n,
            'failure_modes': [],  # filled in next cell
            'evidence': evidence,
        }
        # Hero metric contract: skip emissions that would be confidence=low (n<10).
        # See tests/unit/test_derived_contract.py::test_confidence_low_metrics_not_in_hero_set.
        if n < 10:
            continue
        metrics.append(record)

print('built', len(metrics), 'metric records')

for r in metrics:
    n = r['n']
    r['failure_modes'] = [
        {
            'id': 'small_sample',
            'description': 'n<30; CI width may exceed 10 percentage points',
            'active': n < 30,
            'threshold': 'n<30',
            'evidence_url': None,
        },
        {
            'id': 'noniid_scenarios',
            'description': (
                'Wilson assumes independent Bernoulli trials. Scenarios in the same '
                'run share infrastructure (same VM, same kernel, same network) and '
                'are not independent. CI may undercount.'
            ),
            'active': True,
            'threshold': None,
            'evidence_url': None,
        },
        {
            'id': 'flake_masking_regression',
            'description': (
                'A scenario that passes intermittently inflates pass-rate even when a '
                'real regression is present'
            ),
            'active': False,
            'threshold': 'future: KM survival AUC<0.7',
            'evidence_url': None,
        },
    ]
print('failure modes attached to', len(metrics), 'records')

LIB_NAMES = ['statsmodels', 'scipy', 'pandas', 'jsonschema', 'nbformat', 'nbconvert', 'papermill', 'jupyter']
library_versions = {}
for name in LIB_NAMES:
    try:
        library_versions[name] = version(name)
    except PackageNotFoundError:
        pass

output = {
    'schema_version': 'v2',
    'generated_at': datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ'),
    'generator': {
        'notebook': 'notebooks/01_pass_rate_wilson.ipynb',
        'notebook_url': 'https://factory.projectbluefin.io/methods/01_pass_rate_wilson.html',
        'workflow_run_url': os.environ.get('GITHUB_WORKFLOW_RUN_URL') or None,
        'commit_sha': os.environ.get('GITHUB_SHA') or None,
        'inputs_sha256': inputs_sha256,
        'library_versions': library_versions,
    },
    'metrics': metrics,
}

schema = json.loads(SCHEMA_PATH.read_text())
jsonschema.validate(instance=output, schema=schema)

out_path = DERIVED_DIR / 'trust_window_pass_rate.json'
out_path.write_text(json.dumps(output, indent=2, sort_keys=False) + '\n')
print('wrote', out_path, 'with', len(metrics), 'metrics')