When I first started pulling data from the Random User API for a demographic study, I kept getting “significant” results even after the sample size was tiny. It turned out I was peeking at the data after every batch and ignoring the fact that I was running dozens of tests at once. This post walks you through building a production‑ready pipeline that fetches paginated user profiles, flattens nested JSON, runs segmented chi‑square and Mann‑Whitney U tests, applies Benjamini‑Hochberg FDR correction, and uses O’Brien‑Fleming alpha‑spending to decide whether to stop or keep collecting data.
Key Takeaways
- Peeking at streaming data inflates false positives; sequential boundaries keep the family‑wise error rate in check.
- Benjamini‑Hochberg FDR is preferable to Bonferroni when you have many concurrent hypotheses in production.
- Robust ingestion—exponential backoff, deduplication, and JSON flattening—prevents downstream statistical artifacts.
- Integrating alpha‑spending into your reporting loop lets you make data‑driven decisions about stopping data collection.
The Problem
Data scientists often run standard t‑tests or Mann‑Whitney U tests on data that arrives in real time. When the data is streamed, each new batch feels like a fresh experiment, but statistically it’s a repeated look at the same population. Without adjusting for multiple comparisons or peeking, the cumulative type I error can skyrocket, leading to misguided conclusions. I needed a pipeline that could ingest live API data, perform many hypothesis tests, control the false discovery rate, and decide when enough data has been collected.
Data and Sources
Random User API: https://randomuser.me/documentation SciPy Stats: https://docs.scipy.org/doc/scipy/reference/stats.html Statsmodels Multiple Testing: https://www.statsmodels.org/stable/stats.mats.html Data accessed on 2026-10-29.
Loading the Data
We pull five batches of 100 users each from the Random User API, respecting the service’s rate limit with exponential backoff. Duplicate users are removed by their unique login.uuid field. The nested JSON (e.g., location.coordinates, name.title) is flattened into a tidy DataFrame.
import requests
import time
import pandas as pd
def fetch_batch(batch_size=100, max_retries=5):
url = f"https://randomuser.me/api/?results={batch_size}"
backoff = 1
for attempt in range(max_retries):
try:
resp = requests.get(url, timeout=10)
resp.raise_for_status()
return resp.json()["results"]
except requests.HTTPError as e:
if resp.status_code == 429: # rate limited
time.sleep(backoff)
backoff *= 2
else:
raise e
raise RuntimeError("Max retries exceeded")
The Core Logic
Below are the key functions that drive the pipeline: flattening the raw JSON, segmenting the data by country, running chi‑square on gender counts and Mann‑Whitney U on age distributions, applying Benjamini‑Hochberg correction, and computing the O’Brien‑Fleming alpha‑spending boundary.
from scipy import stats
from statsmodels.stats.multitest import multipletests
import numpy as np
def flatten_user(user):
flat = {
"uuid": user["login"]["uuid"],
"gender": user["gender"],
"age": user["dob"]["age"],
"country": user["location"]["country"],
"lat": user["location"]["coordinates"]["latitude"],
"lon": user["location"]["coordinates"]["longitude"],
}
return flat
def segment_tests(df):
results = []
for country, grp in df.groupby("country"):
# Chi‑square on gender counts
contingency = pd.crosstab(df["country"], df["gender"]).loc[[country]]
chi2, p_chi, _, _ = stats.chi2_contingency(contingency)
# Mann‑Whitney U on age vs overall
u, p_mwu = stats.mannwhitneyu(grp["age"], df["age"], alternative="two-sided")
results.append({
"country": country,
"p_chi": p_chi,
"p_mwu": p_mwu
})
return pd.DataFrame(results)
def bh_adjust(df, alpha=0.05):
pvals = df[["p_chi", "p_mwu"]].values.flatten()
reject, pvals_adj, _, _ = multipletests(pvals, alpha=alpha, method="fdr_bh")
df["p_chi_adj"] = pvals_adj[:len(df)]
df["p_mwu_adj"] = pvals_adj[len(df):]
df["reject"] = reject
return df
def obrian_fleming_alpha(i, k, alpha=0.05):
# i: current look, k: total looks
z = np.sqrt(i / k)
alpha_i = alpha / (2 * (1 - stats.norm.cdf(z)))
return alpha_i
Putting It Together
The main loop fetches each batch, deduplicates, concatenates, runs the tests, adjusts p‑values, and checks against the sequential alpha boundary. If any adjusted p‑value falls below the boundary, the pipeline stops early and prints a recommendation.
import matplotlib.pyplot as plt
def main():
all_users = []
seen = set()
total_batches = 5
for batch_num in range(1, total_batches + 1):
batch = fetch_batch()
for u in batch:
uid = u["login"]["uuid"]
if uid not in seen:
seen.add(uid)
all_users.append(flatten_user(u))
df = pd.DataFrame(all_users)
test_df = segment_tests(df)
test_df = bh_adjust(test_df)
# Sequential boundary
alpha_i = obrian_fleming_alpha(batch_num, total_batches)
if any(test_df[["p_chi_adj", "p_mwu_adj"]].values.flatten() < alpha_i):
print(f"Stopping early at batch {batch_num} with alpha boundary {alpha_i:.4f}")
break
else:
print(f"Continuing to next batch (current alpha boundary {alpha_i:.4f})")
# Plot adjusted p-values
test_df.set_index("country")[["p_chi_adj", "p_mwu_adj"]].plot(kind="bar")
plt.title("Adjusted p‑values by country")
plt.tight_layout()
plt.savefig("output.png")
print("\nFinal report:")
print(test_df)
if __name__ == "__main__":
main()
Complete Script
The full runnable script combining all steps:
#!/usr/bin/env python3
import requests
import time
import pandas as pd
import numpy as np
from scipy import stats
from statsmodels.stats.multitest import multipletests
import matplotlib.pyplot as plt
def fetch_batch(batch_size=100, max_retries=5):
url = f"https://randomuser.me/api/?results={batch_size}"
backoff = 1
for attempt in range(max_retries):
try:
resp = requests.get(url, timeout=10)
resp.raise_for_status()
return resp.json()["results"]
except requests.HTTPError as e:
if resp.status_code == 429:
time.sleep(backoff)
backoff *= 2
else:
raise e
raise RuntimeError("Max retries exceeded")
def flatten_user(user):
flat = {
"uuid": user["login"]["uuid"],
"gender": user["gender"],
"age": user["dob"]["age"],
"country": user["location"]["country"],
"lat": user["location"]["coordinates"]["latitude"],
"lon": user["location"]["coordinates"]["longitude"],
}
return flat
def segment_tests(df):
results = []
for country, grp in df.groupby("country"):
contingency = pd.crosstab(df["country"], df["gender"]).loc[[country]]
chi2, p_chi, _, _ = stats.chi2_contingency(contingency)
u, p_mwu = stats.mannwhitneyu(grp["age"], df["age"], alternative="two-sided")
results.append({"country": country, "p_chi": p_chi, "p_mwu": p_mwu})
return pd.DataFrame(results)
def bh_adjust(df, alpha=0.05):
pvals = df[["p_chi", "p_mwu"]].values.flatten()
reject, pvals_adj, _, _ = multipletests(pvals, alpha=alpha, method="fdr_bh")
df["p_chi_adj"] = pvals_adj[:len(df)]
df["p_mwu_adj"] = pvals_adj[len(df):]
df["reject"] = reject
return df
def obrian_fleming_alpha(i, k, alpha=0.05):
z = np.sqrt(i / k)
alpha_i = alpha / (2 * (1 - stats.norm.cdf(z)))
return alpha_i
def main():
all_users = []
seen = set()
total_batches = 5
for batch_num in range(1, total_batches + 1):
try:
batch = fetch_batch()
except Exception as e:
print(f"Error fetching batch {batch_num}: {e}")
break
for u in batch:
uid = u["login"]["uuid"]
if uid not in seen:
seen.add(uid)
all_users.append(flatten_user(u))
df = pd.DataFrame(all_users)
test_df = segment_tests(df)
test_df = bh_adjust(test_df)
alpha_i = obrian_fleming_alpha(batch_num, total_batches)
if any(test_df[["p_chi_adj", "p_mwu_adj"]].values.flatten() < alpha_i):
print(f"Stopping early at batch {batch_num} with alpha boundary {alpha_i:.4f}")
break
else:
print(f"Continuing to next batch (current alpha boundary {alpha_i:.4f})")
test_df.set_index("country")[["p_chi_adj", "p_mwu_adj"]].plot(kind="bar")
plt.title("Adjusted p‑values by country")
plt.tight_layout()
plt.savefig("output.png")
print("\nFinal report:")
print(test_df)
if __name__ == "__main__":
main()
Expected Output
Running the script will print progress after each batch, a decision to stop early or continue, and a final DataFrame showing adjusted p‑values and whether each null hypothesis was rejected. A bar chart is saved as output.png showing the adjusted p‑values per country.
Limitations and Tradeoffs
• The O’Brien‑Fleming boundary assumes equal batch sizes and independent observations; real API streams may have duplicates or varying batch sizes.
• Mann‑Whitney U and chi‑square tests assume independent samples; demographic data from the same region may violate this.
• The BH procedure controls the false discovery rate but does not guard against type‑II errors; in low‑power scenarios, you may miss true effects.
• The script uses a simple exponential backoff; more sophisticated rate‑limit handling (e.g., parsing Retry‑After headers) would be needed for production.
Frequently Asked Questions
Why can't I just run a t‑test every time new data arrives?
Each look at the data inflates the cumulative type‑I error rate. Even if you set alpha to 5%, after 20 peeks the real false‑positive rate can exceed 30%. Sequential boundaries keep the overall error rate under control.
How does the Benjamini‑Hochberg procedure differ from Bonferroni correction in production pipelines?
Bonferroni divides alpha by the number of tests, which is overly conservative and reduces power. Benjamini‑Hochberg controls the expected proportion of false discoveries, allowing more true positives while still limiting false positives.
What happens if the Random User API returns rate‑limited or cached payloads?
Rate limits trigger HTTP 429 responses; the backoff logic handles this. Cached payloads can introduce duplicate UUIDs, which we deduplicate. If duplicates slip through, they artificially inflate sample sizes and deflate p‑values.
How do I handle missing demographic fields without introducing selection bias?
Drop rows with missing critical fields (e.g., age, gender) before testing, or impute using domain‑specific logic. Avoid removing entire countries unless you can justify that the missingness is random.
What I'd Change
In a real deployment I’d replace the simple backoff with a token‑bucket rate‑limit manager, add a persistent deduplication store (e.g., Redis), and switch to Bayesian sequential tests that naturally incorporate prior knowledge. I’d also move the statistical engine into a lightweight microservice so that the ingestion layer can scale independently. Finally, I’d automate the alpha‑spending schedule based on actual information fraction rather than assuming equal batch sizes.