A/B Testing ML Models in Production with Statistical Rigor

Deploying a new ML model without proper A/B testing is gambling with your product metrics. But ML model testing is harder than typical feature A/B tests — models can degrade subtly, effects may be non-stationary, and the metrics that matter most are often slow to converge. After running 200+ model experiments across recommendation, search ranking, and fraud detection systems, here's how to test ML models with the statistical rigor they demand.

The Problem: Why Standard A/B Testing Fails for ML

Standard A/B testing frameworks assume independent, identically distributed observations with stationary treatment effects. ML models violate these assumptions regularly:

• Network effects: A recommendation model's performance depends on what other users see
• Temporal non-stationarity: Model performance shifts as user behavior evolves
• Metric lag: The true impact of a fraud model takes 30-90 days to materialize in chargeback rates
• Multiple correlated metrics: You care about precision, recall, revenue impact, and user satisfaction simultaneously

Teams that ignore these nuances end up either shipping harmful models or killing good ones based on noisy short-term metrics.

Architecture: Experiment Platform for ML

The experimentation platform needs three core components: traffic splitting with consistent hashing, metric collection with proper attribution, and a statistical analysis engine that accounts for ML-specific challenges.

Traffic Splitting with Consistent Assignment

Users must see consistent model versions across sessions. A user assigned to the treatment group at 9 AM must still see the treatment model at 9 PM. We use deterministic hashing on user ID + experiment ID to achieve this.

import hashlib
import json
from dataclasses import dataclass
from typing import Optional
from scipy import stats
import numpy as np

@dataclass
class ExperimentConfig:
    experiment_id: str
    model_a_version: str  # Control
    model_b_version: str  # Treatment
    traffic_split: float  # Fraction to treatment (0.0 - 1.0)
    primary_metric: str
    guardrail_metrics: list[str]
    min_sample_size: int
    max_duration_days: int
    mde: float  # Minimum detectable effect

class MLExperimentRouter:
    def __init__(self, experiments: list[ExperimentConfig]):
        self.experiments = {e.experiment_id: e for e in experiments}

    def get_assignment(self, user_id: str, experiment_id: str) -> str:
        config = self.experiments[experiment_id]
        hash_input = f"{user_id}:{experiment_id}".encode()
        hash_value = int(hashlib.sha256(hash_input).hexdigest(), 16)
        bucket = (hash_value % 10000) / 10000.0

        if bucket < config.traffic_split:
            return config.model_b_version  # Treatment
        return config.model_a_version  # Control

    def get_model_for_request(self, user_id: str) -> tuple[str, dict]:
        assignments = {}
        for exp_id, config in self.experiments.items():
            model_version = self.get_assignment(user_id, exp_id)
            assignments[exp_id] = model_version
        
        # Return highest-priority active experiment assignment
        active = list(self.experiments.values())[0]
        version = assignments[active.experiment_id]
        metadata = {
            "experiment_id": active.experiment_id,
            "variant": "treatment" if version == active.model_b_version else "control",
            "model_version": version,
        }
        return version, metadata


class MLExperimentAnalyzer:
    def __init__(self, confidence_level: float = 0.95):
        self.alpha = 1 - confidence_level

    def analyze_experiment(
        self,
        control_metrics: np.ndarray,
        treatment_metrics: np.ndarray,
        metric_type: str = "continuous",
    ) -> dict:
        if metric_type == "continuous":
            return self._analyze_continuous(control_metrics, treatment_metrics)
        return self._analyze_proportion(control_metrics, treatment_metrics)

    def _analyze_continuous(self, control: np.ndarray, treatment: np.ndarray) -> dict:
        control_mean = np.mean(control)
        treatment_mean = np.mean(treatment)
        relative_lift = (treatment_mean - control_mean) / control_mean

        t_stat, p_value = stats.ttest_ind(control, treatment, equal_var=False)

        control_se = stats.sem(control)
        treatment_se = stats.sem(treatment)
        pooled_se = np.sqrt(control_se**2 + treatment_se**2)
        z = stats.norm.ppf(1 - self.alpha / 2)
        ci_lower = (treatment_mean - control_mean) - z * pooled_se
        ci_upper = (treatment_mean - control_mean) + z * pooled_se

        return {
            "control_mean": float(control_mean),
            "treatment_mean": float(treatment_mean),
            "relative_lift": float(relative_lift),
            "p_value": float(p_value),
            "confidence_interval": [float(ci_lower), float(ci_upper)],
            "is_significant": p_value < self.alpha,
            "sample_size_control": len(control),
            "sample_size_treatment": len(treatment),
            "statistical_power": self._compute_power(control, treatment),
        }

    def _analyze_proportion(self, control: np.ndarray, treatment: np.ndarray) -> dict:
        p_control = np.mean(control)
        p_treatment = np.mean(treatment)
        relative_lift = (p_treatment - p_control) / p_control

        n_c, n_t = len(control), len(treatment)
        p_pool = (p_control * n_c + p_treatment * n_t) / (n_c + n_t)
        se = np.sqrt(p_pool * (1 - p_pool) * (1/n_c + 1/n_t))
        z_stat = (p_treatment - p_control) / se
        p_value = 2 * (1 - stats.norm.cdf(abs(z_stat)))

        return {
            "control_rate": float(p_control),
            "treatment_rate": float(p_treatment),
            "relative_lift": float(relative_lift),
            "p_value": float(p_value),
            "is_significant": p_value < self.alpha,
            "sample_size_control": n_c,
            "sample_size_treatment": n_t,
        }

    def _compute_power(self, control: np.ndarray, treatment: np.ndarray) -> float:
        effect_size = (np.mean(treatment) - np.mean(control)) / np.std(control)
        n = min(len(control), len(treatment))
        se = 1 / np.sqrt(n)
        z_alpha = stats.norm.ppf(1 - self.alpha / 2)
        z_beta = abs(effect_size) / se - z_alpha
        return float(stats.norm.cdf(z_beta))

Guardrail Metrics and Sequential Testing

For ML models, we need guardrail metrics that trigger early stopping if the new model causes harm, independent of whether the primary metric has reached significance.

interface GuardrailCheck {
  metric: string;
  threshold: number;
  direction: 'lower_is_worse' | 'higher_is_worse';
  currentValue: number;
  controlValue: number;
  triggered: boolean;
}

interface ExperimentDecision {
  experimentId: string;
  decision: 'promote' | 'rollback' | 'continue';
  primaryMetric: { lift: number; pValue: number; significant: boolean };
  guardrails: GuardrailCheck[];
  sampleSize: number;
  daysRunning: number;
  reason: string;
}

class SequentialExperimentMonitor {
  private readonly spendingFunction = 'obrien-fleming';

  async evaluateExperiment(experimentId: string, dayIndex: number): Promise<ExperimentDecision> {
    const data = await this.fetchExperimentData(experimentId);
    const config = await this.getExperimentConfig(experimentId);

    // Check guardrails first - these can trigger immediate rollback
    const guardrails = await this.checkGuardrails(data, config);
    const triggeredGuardrails = guardrails.filter(g => g.triggered);

    if (triggeredGuardrails.length > 0) {
      return {
        experimentId,
        decision: 'rollback',
        primaryMetric: { lift: 0, pValue: 1, significant: false },
        guardrails,
        sampleSize: data.totalSamples,
        daysRunning: dayIndex,
        reason: `Guardrail triggered: ${triggeredGuardrails.map(g => g.metric).join(', ')}`,
      };
    }

    // Sequential test on primary metric with spending function
    const adjustedAlpha = this.getAdjustedAlpha(dayIndex, config.max_duration_days);
    const primaryResult = this.testPrimaryMetric(data, adjustedAlpha);

    if (primaryResult.significant && primaryResult.lift > 0) {
      return {
        experimentId,
        decision: 'promote',
        primaryMetric: primaryResult,
        guardrails,
        sampleSize: data.totalSamples,
        daysRunning: dayIndex,
        reason: `Primary metric significant: +${(primaryResult.lift * 100).toFixed(2)}%`,
      };
    }

    if (dayIndex >= config.max_duration_days) {
      return {
        experimentId,
        decision: 'rollback',
        primaryMetric: primaryResult,
        guardrails,
        sampleSize: data.totalSamples,
        daysRunning: dayIndex,
        reason: 'Max duration reached without significance',
      };
    }

    return {
      experimentId,
      decision: 'continue',
      primaryMetric: primaryResult,
      guardrails,
      sampleSize: data.totalSamples,
      daysRunning: dayIndex,
      reason: 'Insufficient evidence - continuing',
    };
  }

  private getAdjustedAlpha(currentDay: number, totalDays: number): number {
    const informationFraction = currentDay / totalDays;
    // O'Brien-Fleming spending function
    const z = 2 * (1 - this.normalCDF(
      this.normalQuantile(1 - 0.05 / 2) / Math.sqrt(informationFraction)
    ));
    return z;
  }

  private normalCDF(x: number): number {
    return 0.5 * (1 + this.erf(x / Math.sqrt(2)));
  }

  private normalQuantile(p: number): number {
    return Math.sqrt(2) * this.erfInv(2 * p - 1);
  }

  private erf(x: number): number {
    const a = [0.254829592, -0.284496736, 1.421413741, -1.453152027, 1.061405429];
    const p = 0.3275911;
    const sign = x < 0 ? -1 : 1;
    const t = 1.0 / (1.0 + p * Math.abs(x));
    const y = 1.0 - (((((a[4]*t + a[3])*t) + a[2])*t + a[1])*t + a[0])*t * Math.exp(-x*x);
    return sign * y;
  }

  private erfInv(x: number): number {
    const a = 0.147;
    const ln = Math.log(1 - x*x);
    const s = Math.sign(x) * Math.sqrt(
      Math.sqrt((2/(Math.PI*a) + ln/2)**2 - ln/a) - (2/(Math.PI*a) + ln/2)
    );
    return s;
  }

  private async fetchExperimentData(experimentId: string): Promise<any> {
    return { totalSamples: 0 };
  }

  private async getExperimentConfig(experimentId: string): Promise<any> {
    return {};
  }

  private async checkGuardrails(data: any, config: any): Promise<GuardrailCheck[]> {
    return [];
  }

  private testPrimaryMetric(data: any, alpha: number): any {
    return { significant: false, lift: 0, pValue: 1 };
  }
}

Key Patterns for ML A/B Testing

Interleaving for Ranking Models

For search and recommendation models, interleaving produces results with 10-100x higher sensitivity than traditional A/B splits. Instead of showing all results from model A or model B, interleave results from both models in a single ranked list and measure user preference through clicks.

Long-Running Experiments for Delayed Metrics

Fraud detection models need 60-90 day experiments because chargebacks have significant reporting lag. Use surrogate metrics (model confidence scores, rule trigger rates) for early signal, but make final decisions on the true outcome metric.

Stratified Analysis

Segment results by user cohorts that may respond differently to model changes. A recommendation model might improve metrics for power users while hurting new users. Aggregate metrics hide this heterogeneity.

Benchmarks: Experiment Platform Performance

Metric	Value
Assignment latency (p99)	0.3ms
Metric ingestion throughput	500K events/sec
Time to first significance signal	2-5 days (typical)
False positive rate (validated)	4.8% (target: 5%)
Experiments run concurrently	15-25
Guardrail detection latency	< 4 hours

Common Pitfalls

Peeking without correction: Checking results daily without adjusting for multiple comparisons inflates false positive rates to 20-30%. Use sequential testing with spending functions.

Insufficient sample size: ML models often produce small effect sizes (1-3% improvement). You need large sample sizes to detect these reliably. Calculate required sample size before starting.

Ignoring interaction effects: Running multiple model experiments simultaneously can create interference. Use mutual exclusion groups or measure interaction effects.

Winner's curse: The observed effect size during the experiment is typically larger than the true effect due to selection bias. Expect 20-40% regression from experiment lift to long-term production lift.

Conclusion

A/B testing ML models requires more sophistication than standard feature testing. The combination of consistent assignment, sequential testing with spending functions, guardrail metrics for safety, and proper power analysis gives you the statistical rigor to make confident promotion decisions. Invest in the experimentation infrastructure early — it pays dividends every time you avoid shipping a harmful model or killing a beneficial one based on noise.