ML Experimentation

import json, hashlib, pathlib, datetime

def log_run(name, hypothesis, params, metrics, data_version, seed):
    record = {
        'name': name,
        'hypothesis': hypothesis,
        'params': params,
        'metrics': metrics,
        'data_version': data_version,
        'seed': seed,
        'created_at': datetime.datetime.now(datetime.timezone.utc).isoformat(),
    }
    record['id'] = hashlib.sha1(json.dumps(record, sort_keys=True).encode()).hexdigest()[:10]
    pathlib.Path('runs.jsonl').open('a').write(json.dumps(record) + '\n')
    return record['id']

log_run(
    name='exp-012-group-split',
    hypothesis='the random split leaked users; '
               'a group split will lower AUC to the honest value',
    params={'cv': 'StratifiedGroupKFold(5)', 'model': 'hgb', 'lr': 0.05},
    metrics={'cv_auc_mean': 0.791, 'cv_auc_std': 0.014},
    data_version='snap-2026-09-01', seed=0,
)

import subprocess, sys, sklearn, numpy as np, platform

def environment():
    out = subprocess.run(['git', 'rev-parse', 'HEAD'], capture_output=True, text=True)
    sha = out.stdout.strip()
    status = subprocess.run(['git', 'status', '--porcelain'], capture_output=True, text=True)
    dirty = bool(status.stdout)
    return {
        'git_sha': sha,
        'git_dirty': dirty,
        'python': sys.version.split()[0],
        'platform': platform.platform(),
        'numpy': np.__version__,
        'sklearn': sklearn.__version__,
    }

meta = {'env': environment(), 'seed': 0, 'data_version': 'snap-2026-09-01',
        'params': pipe.get_params(deep=True), 'protocol': 'StratifiedGroupKFold(5), roc_auc'}

# Offline: one number per candidate, on a frozen sample.
offline = {m: roc_auc_score(y_val, p.predict_proba(X_val)[:, 1]) for m, p in candidates.items()}

# Online: the metric the business actually cares about, on live traffic.
def assign_variant(user_id, salt='exp-014', variants=('control', 'challenger')):
    import hashlib
    h = hashlib.md5(f'{salt}:{user_id}'.encode()).hexdigest()
    return variants[int(h, 16) % len(variants)]     # stable, so a user never flips

# online_metrics: conversion rate per variant, over the assignment period.

import hashlib
import numpy as np

def bucket(user_id, salt, buckets=100):
    h = hashlib.sha256(f'{salt}:{user_id}'.encode()).digest()
    return int.from_bytes(h[:8], 'big') % buckets

def variant(user_id, salt='exp-014', treatment_share=0.5):
    return 'treatment' if bucket(user_id, salt) < treatment_share * 100 else 'control'

# Sanity checks before reading any result:
users = [f'u-{i}' for i in range(20000)]
assign = np.array([variant(u) for u in users])
print((assign == 'treatment').mean())          # ~0.50: the split is balanced
print(all(variant(u) == variant(u) for u in users[:100]))   # stable across calls

import numpy as np
from scipy import stats

n_c, x_c = 40000, 2000      # control:   5.00%
n_t, x_t = 40000, 2120      # treatment: 5.30%
p_c, p_t = x_c / n_c, x_t / n_t

diff = p_t - p_c
se = np.sqrt(p_c * (1 - p_c) / n_c + p_t * (1 - p_t) / n_t)
ci = (diff - 1.96 * se, diff + 1.96 * se)
z = diff / se
p_value = 2 * (1 - stats.norm.cdf(abs(z)))
print(f'diff {diff:+.4f}  95% CI [{ci[0]:+.4f}, {ci[1]:+.4f}]  z {z:.2f}  p {p_value:.4f}')
# diff +0.0030  95% CI [+0.0000, +0.0060]  z 1.96  p 0.0498

import numpy as np
from scipy import stats

def n_per_arm(baseline, rel_lift, alpha=0.05, power=0.8):
    p1 = baseline
    p2 = baseline * (1 + rel_lift)
    pbar = (p1 + p2) / 2
    z_a = stats.norm.ppf(1 - alpha / 2)
    z_b = stats.norm.ppf(power)
    return int(np.ceil(2 * pbar * (1 - pbar) * (z_a + z_b) ** 2 / (p2 - p1) ** 2))

for lift in (0.01, 0.02, 0.05, 0.10):
    n = n_per_arm(0.05, lift)
    print(f'{lift:.0%} relative lift -> {n:>9,} users per arm '
          f'-> {n * 2 / 120000:.1f} days at 120k/day')

import numpy as np

CONFIG = {'seed': 20260919, 'data_version': 'snap-2026-09-01', 'n_folds': 5}

root = np.random.default_rng(CONFIG['seed'])
split_rng, init_rng, aug_rng = root.spawn(3)      # independent, reproducible streams

# Each consumer gets its own generator; none share hidden global state.
folds = make_folds(n=len(X), k=CONFIG['n_folds'], rng=split_rng)
model = build_model(random_state=int(init_rng.integers(0, 2**31 - 1)))

import numpy as np
from scipy import stats
from sklearn.model_selection import StratifiedKFold, cross_val_score

cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=0)   # identical folds for both
a = cross_val_score(pipe_a, X, y, cv=cv, scoring='roc_auc')
b = cross_val_score(pipe_b, X, y, cv=cv, scoring='roc_auc')

d = b - a
print(f'A {a.mean():.3f} ± {a.std():.3f}   B {b.mean():.3f} ± {b.std():.3f}')
print(f'paired diff {d.mean():+.4f} ± {d.std():.4f}')
print(stats.ttest_rel(b, a))

def handle(request, champion, challenger, log):
    features = build_features(request)

    decision = champion.predict_proba(features)[0, 1]          # this is what ships

    try:
        shadow = float(challenger.predict_proba(features)[0, 1])   # observed, never acted on
        error = None
    except Exception as exc:
        shadow, error = None, repr(exc)

    log.write({'request_id': request['id'], 'champion': float(decision),
               'challenger': shadow, 'challenger_error': error})
    return decision

import numpy as np

rng = np.random.default_rng(0)
false_positives = 0
trials = 2000
for _ in range(trials):
    a = rng.random(4000) < 0.05      # identical true rates: no effect exists
    b = rng.random(4000) < 0.05
    crossed = False
    for n in range(500, 4001, 250):  # peek every 250 users, 15 looks
        pa, pb = a[:n].mean(), b[:n].mean()
        se = np.sqrt(pa * (1 - pa) / n + pb * (1 - pb) / n) + 1e-12
        if abs((pb - pa) / se) > 1.96:
            crossed = True
            break
    false_positives += crossed
print(false_positives / trials)      # ~0.25, not 0.05

import numpy as np
from statsmodels.stats.multitest import multipletests

p_values = np.array([0.003, 0.011, 0.028, 0.041, 0.046, 0.19, 0.31, 0.52, 0.68, 0.84,
                     0.02, 0.033, 0.049, 0.07, 0.12, 0.25, 0.44, 0.61, 0.77, 0.91])

for method in ('bonferroni', 'fdr_bh'):
    reject, adjusted, _, _ = multipletests(p_values, alpha=0.05, method=method)
    print(f'{method:<11} survivors: {reject.sum():>2}  '
          f'smallest adjusted p: {adjusted.min():.4f}')
# bonferroni  survivors:  1   smallest adjusted p: 0.0600
# fdr_bh      survivors:  5   smallest adjusted p: 0.0600