Ever watched a dashboard flash “drift detected” every few minutes only to discover the model was fine? The culprit is usually a naïve per‑batch p‑value check that forgets the multiple‑testing penalty. If you’re a data scientist or MLOps engineer who needs to monitor a live categorical feed (e.g., gender distribution from the Random User API) without drowning in false alarms, this post shows exactly how I built a production‑ready sequential hypothesis‑testing pipeline that only alerts when the drift is statistically convincing.
Key Takeaways
- Rolling counts let you test a distribution over a moving window instead of a single batch.
- One‑proportion Z‑tests are cheap enough for sub‑second latency on streaming data.
- O’Brien‑Fleming alpha‑spending caps the family‑wise error rate across an indefinite number of looks.
- Persisting state and exponential back‑off make the loop robust to API hiccups.
- All of this fits in a single, 40‑line Python script you can drop into Airflow, Prefect, or a simple cron job.
Data and Sources
The pipeline consumes real user profiles from the Random User API. Each request returns a JSON payload with an array of user objects; we only care about the gender field. Data accessed on 2026-10-31.
Step 1 — Set Up a Rate‑Limited Streaming Ingest
We don’t want to hammer the free API, so we throttle calls to one request per second. A simple time.sleep loop does the trick and keeps the code easy to read.
import time, requests
def fetch_batch(results=100):
"""Fetch a batch of users, respecting the 1‑sec rate limit."""
time.sleep(1) # rate limit
resp = requests.get(
f"https://randomuser.me/api/?results={results}",
timeout=5
)
resp.raise_for_status()
return resp.json()
Step 2 — Validate and Parse JSON Safely
API contracts change. We guard against missing keys and unexpected gender values by filtering only the known categories.
def extract_genders(payload):
"""Return a list of 'male'/'female' strings, dropping anything else."""
genders = []
for user in payload.get("results", []):
gender = user.get("gender")
if gender in ("male", "female"):
genders.append(gender)
return genders
Step 3 — Maintain Rolling Counts with a Fixed Window
We keep a deque of the last WINDOW_SIZE observations. When the window slides, we update the male count in O(1) time.
from collections import deque
WINDOW_SIZE = 5000 # total observations we care about
class RollingCounter:
def __init__(self, size):
self.size = size
self.buffer = deque(maxlen=size)
self.male_cnt = 0
def add_batch(self, genders):
for g in genders:
if len(self.buffer) == self.size:
# pop oldest
old = self.buffer.popleft()
if old == "male":
self.male_cnt -= 1
self.buffer.append(g)
if g == "male":
self.male_cnt += 1
@property
def total(self):
return len(self.buffer)
@property
def male_proportion(self):
return self.male_cnt / self.total if self.total else 0.0
Step 4 — Compute One‑Proportion Z‑Test on the Fly
The null hypothesis is a perfect 50/50 split. The Z‑statistic and two‑sided p‑value are cheap to compute.
import math
from scipy import stats
def z_test(p_hat, n, p0=0.5):
"""Return two‑sided p‑value for proportion test."""
if n == 0:
return 1.0
se = math.sqrt(p0 * (1 - p0) / n)
z = (p_hat - p0) / se
return 2 * (1 - stats.norm.cdf(abs(z)))
Step 5 — Apply an O’Brien‑Fleming Alpha‑Spending Function
Each look consumes a slice of the overall 5 % error budget. The O’Brien‑Fleming bound is very conservative early on, which matches our “don’t alert on noise” philosophy.
def obf_alpha(look, max_looks, overall_alpha=0.05):
"""Alpha spent at look i (1‑indexed) using O’Brien‑Fleming."""
t = look / max_looks
# Critical Z for OBF: z_{α/2} / sqrt(t)
z_alpha = stats.norm.ppf(1 - overall_alpha / 2)
crit_z = z_alpha / math.sqrt(t)
# Convert back to alpha for this look
return 2 * (1 - stats.norm.cdf(crit_z))
Step 6 — Persist State and Emit Alerts
State is written to state.json after every successful batch. If the p‑value drops below the current OBF threshold we fire an alert (here we simply print, but in production you’d push to Slack, PagerDuty, etc.).
import json
import os
STATE_FILE = "state.json"
def load_state():
if os.path.exists(STATE_FILE):
with open(STATE_FILE) as f:
return json.load(f)
return {"look": 0, "counter": {"buffer": [], "male_cnt": 0}}
def save_state(look, counter):
state = {
"look": look,
"counter": {
"buffer": list(counter.buffer),
"male_cnt": counter.male_cnt
}
}
with open(STATE_FILE, "w") as f:
json.dump(state, f)
def maybe_alert(p_val, alpha, look):
if p_val < alpha:
print(f"[ALERT] Drift detected at look {look}! p={p_val:.4g} < α={alpha:.4g}")
Step 7 — Graceful Error Handling and Back‑off
Network glitches happen. We catch requests.exceptions.RequestException and retry with exponential back‑off up to three attempts before giving up on that batch.
def safe_fetch(max_retries=3):
delay = 1
for attempt in range(1, max_retries + 1):
try:
return fetch_batch()
except requests.exceptions.RequestException as e:
print(f"Fetch failed (attempt {attempt}/{max_retries}): {e}")
if attempt == max_retries:
raise
time.sleep(delay)
delay *= 2 # exponential back‑off
Limitations and Tradeoffs
This pipeline assumes a binary categorical variable and a fixed baseline of 0.5. If the true baseline drifts slowly, the O’Brien‑Fleming bound may be too strict, causing missed alerts. The rolling window size trades latency for statistical power: a larger window smooths noise but delays detection. Finally, persisting state to a local JSON file works for a single‑node demo but must be replaced with a durable store (e.g., Redis or a DB) for a true distributed production system.
Frequently Asked Questions
Why not just use a fixed α=0.05 for every batch?
Each batch is a repeated hypothesis test on overlapping data. Without alpha‑spending the family‑wise error rate inflates dramatically, leading to many false alarms.
Can I replace the Random User API with a Kafka topic?
Yes. Swap safe_fetch for a consumer poll, keep the same parsing, rolling‑counter, and testing logic. Remember to commit offsets only after state is persisted.
What if the API returns genders other than “male” or “female”?
The extract_genders function silently drops unknown values. In production you’d log the anomaly and optionally treat it as a third category.
Is O’Brien‑Fleming the only spending function I can use?
No. Pocock or Lan‑DeMets are alternatives. O’Brien‑Fleming is conservative early on, which aligns with a “don’t alert on noise” stance.
How do I scale this to monitor dozens of categorical streams?
Wrap the core logic in a class, instantiate one per stream, and orchestrate them with an executor (ThreadPoolExecutor, Airflow, Prefect). State files become per‑stream keys in a shared KV store.
Complete Script
The full runnable script combining all steps. Save it as sequential_drift.py and run python sequential_drift.py.
#!/usr/bin/env python3
"""
Sequential hypothesis‑testing pipeline for gender drift.
Fetches batches from Random User API, maintains a rolling window,
runs a one‑proportion Z‑test, and applies an O’Brien‑Fleming
alpha‑spending rule before emitting alerts.
"""
import time
import json
import os
import math
import requests
from collections import deque
from scipy import stats
# -------------------- Configuration --------------------
WINDOW_SIZE = 5000 # total observations in rolling window
BATCH_SIZE = 100 # users per API call
MAX_LOOKS = 1000 # upper bound for alpha‑spending (practical)
STATE_FILE = "state.json"
OVERALL_ALPHA = 0.05 # family‑wise error budget
# -------------------------------------------------------
def fetch_batch(results=BATCH_SIZE):
"""Rate‑limited request to Random User API."""
time.sleep(1) # simple rate limit
resp = requests.get(
f"https://randomuser.me/api/?results={results}",
timeout=5
)
resp.raise_for_status()
return resp.json()
def safe_fetch(max_retries=3):
delay = 1
for attempt in range(1, max_retries + 1):
try:
return fetch_batch()
except requests.exceptions.RequestException as e:
print(f"Fetch failed (attempt {attempt}/{max_retries}): {e}")
if attempt == max_retries:
raise
time.sleep(delay)
delay *= 2
def extract_genders(payload):
"""Return list of 'male'/'female' strings; ignore others."""
genders = []
for user in payload.get("results", []):
gender = user.get("gender")
if gender in ("male", "female"):
genders.append(gender)
return genders
class RollingCounter:
def __init__(self, size):
self.size = size
self.buffer = deque(maxlen=size)
self.male_cnt = 0
def add_batch(self, genders):
for g in genders:
if len(self.buffer) == self.size:
old = self.buffer.popleft()
if old == "male":
self.male_cnt -= 1
self.buffer.append(g)
if g == "male":
self.male_cnt += 1
@property
def total(self):
return len(self.buffer)
@property
def male_proportion(self):
return self.male_cnt / self.total if self.total else 0.0
def z_test(p_hat, n, p0=0.5):
"""Two‑sided p‑value for one‑proportion Z‑test."""
if n == 0:
return 1.0
se = math.sqrt(p0 * (1 - p0) / n)
z = (p_hat - p0) / se
return 2 * (1 - stats.norm.cdf(abs(z)))
def obf_alpha(look, max_looks, overall_alpha=OVERALL_ALPHA):
"""O’Brien‑Fleming spending for look i (1‑indexed)."""
t = look / max_looks
z_alpha = stats.norm.ppf(1 - overall_alpha / 2)
crit_z = z_alpha / math.sqrt(t)
return 2 * (1 - stats.norm.cdf(crit_z))
def load_state():
if os.path.exists(STATE_FILE):
with open(STATE_FILE) as f:
raw = json.load(f)
counter = RollingCounter(WINDOW_SIZE)
counter.buffer = deque(raw["counter"]["buffer"], maxlen=WINDOW_SIZE)
counter.male_cnt = raw["counter"]["male_cnt"]
return raw["look"], counter
return 0, RollingCounter(WINDOW_SIZE)
def save_state(look, counter):
state = {
"look": look,
"counter": {
"buffer": list(counter.buffer),
"male_cnt": counter.male_cnt
}
}
with open(STATE_FILE, "w") as f:
json.dump(state, f)
def maybe_alert(p_val, alpha, look):
if p_val < alpha:
print(f"[ALERT] Drift detected at look {look}! p={p_val:.4g} < α={alpha:.4g}")
def main():
look, counter = load_state()
while True:
# 1️⃣ Fetch & parse
payload = safe_fetch()
genders = extract_genders(payload)
if not genders:
print("No valid gender records – skipping batch.")
continue
# 2️⃣ Update rolling window
counter.add_batch(genders)
look += 1
# 3️⃣ Statistical test
p_hat = counter.male_proportion
n = counter.total
p_val = z_test(p_hat, n)
# 4️⃣ Alpha spending
cur_alpha = obf_alpha(look, MAX_LOOKS)
# 5️⃣ Alert if needed
maybe_alert(p_val, cur_alpha, look)
# 6️⃣ Persist state
save_state(look, counter)
# Stop condition for demo purposes (remove in prod)
if look >= MAX_LOOKS:
print("Reached max looks – exiting demo.")
break
if __name__ == "__main__":
main()
Expected Output
When you run the script you’ll see a steady stream of logs like:
Fetch failed (attempt 1/3): HTTPConnectionPool(host='randomuser.me', port=80): Max retries exceeded with url: /api/?results=100 (Caused by NewConnectionError(': Failed to establish a new connection'))
[ALERT] Drift detected at look 42! p=0.018 < α=0.023
...
Reached max looks – exiting demo.
The alert line appears only when the rolling proportion of “male” deviates enough from 0.5 that the O’Brien‑Fleming bound is breached.
What I'd Change
If I were to ship this at scale, I’d replace the local JSON checkpoint with a transactional store (Postgres or Redis) and move the rate‑limited fetch into a proper scheduler (Airflow or Prefect). The O’Brien‑Fleming function could be abstracted to accept any spending curve, letting product owners tune early‑ vs‑late‑alert aggressiveness without touching the core test logic. Finally, I’d enrich the alert payload with confidence intervals and recent trend plots so stakeholders get context, not just a binary “drift!” flag.