ab-test-ml

name	ab-test-ml
description	Design and analyze A/B tests for ML models in production, including traffic splitting, metric selection, statistical significance testing, and safe rollout strategies.
argument-hint	["model type","business metric","traffic volume","acceptable risk level"]
allowed-tools	Read, Write, Bash

A/B Testing for ML Models

Testing ML models in production is fundamentally different from testing features — model behavior is probabilistic, metrics are delayed, and interactions between models complicate attribution. Rigorous A/B testing is the only way to know if a new model actually moves business metrics.

Process

1. Define the hypothesis — what improvement does the challenger model provide and how?
2. Choose the primary metric — one business metric (revenue, conversion, retention), not ML metrics.
3. Power analysis — calculate required sample size for statistical significance.
4. Design traffic split — percentage, assignment unit (user/session/item), exclusions.
5. Implement model routing — shadow, canary, or full A/B.
6. Run until significance — don't stop early; use sequential testing if needed.
7. Analyze results — primary + guardrail metrics, segment breakdowns.
8. Make deployment decision — ship, iterate, or rollback with documented rationale.

Output Format

Experiment Design

# experiment_design.py
import numpy as np
from scipy import stats
from dataclasses import dataclass
from typing import Optional

@dataclass
class ExperimentDesign:
    name: str
    hypothesis: str
    primary_metric: str
    guardrail_metrics: list[str]
    baseline_value: float
    minimum_detectable_effect: float  # MDE — smallest meaningful difference
    statistical_power: float = 0.80
    significance_level: float = 0.05
    traffic_pct: float = 0.50          # Fraction of traffic in experiment
    
    def required_sample_size(self) -> dict:
        """Calculate required sample size per arm for two-sample test."""
        # Effect size (Cohen's d for continuous, relative risk for proportions)
        if self.baseline_value < 1.0:  # Proportion metric (conversion rate)
            effect = self.minimum_detectable_effect
            p1 = self.baseline_value
            p2 = p1 * (1 + effect)
            
            # Two-proportion z-test
            from statsmodels.stats.power import NormalIndPower
            pooled = (p1 + p2) / 
            h =  * np.arcsin(np.sqrt(p2)) -  * np.arcsin(np.sqrt(p1))  
            
            analysis = NormalIndPower()
            n = analysis.solve_power(
                effect_size=h,
                alpha=.significance_level,
                power=.statistical_power,
                alternative=
            )
        :  
            
            std = .baseline_value * 
            delta = .baseline_value * .minimum_detectable_effect
            
             statsmodels.stats.power  TTestIndPower
            analysis = TTestIndPower()
            n = analysis.solve_power(
                effect_size=delta / std,
                alpha=.significance_level,
                power=.statistical_power,
                alternative=
            )
        
        n_per_arm = (np.ceil(n))
        total = n_per_arm * 
        
         {
            : n_per_arm,
            : total,
            : .traffic_pct,
        }
    
     () -> :
        
        size = .required_sample_size()
        daily_in_experiment = daily_eligible_users * .traffic_pct
         size[] / daily_in_experiment



design = ExperimentDesign(
    name=,
    hypothesis=,
    primary_metric=,
    guardrail_metrics=[, , ],
    baseline_value=,              
    minimum_detectable_effect=,   
    traffic_pct=,                 
)

sizing = design.required_sample_size()
runtime = design.estimated_runtime_days(daily_eligible_users=)

()
()

Traffic Splitting & Model Routing

# routing/model_router.py
import hashlib
import mlflow
from typing import Optional
from enum import Enum

class VariantAssignment(Enum):
    CONTROL = "control"
    TREATMENT = "treatment"
    SHADOW = "shadow"       # Runs both, uses control output

class ModelRouter:
    """
    Routes requests to control (champion) or treatment (challenger) model.
    Uses deterministic hashing for sticky assignment.
    """
    
    def __init__(
        self,
        experiment_id: str,
        treatment_pct: float = 0.10,   # 10% to challenger
        assignment_unit: str = "user_id",
        shadow_mode: bool = False
    ):
        self.experiment_id = experiment_id
        self.treatment_pct = treatment_pct
        self.assignment_unit = assignment_unit
        self.shadow_mode = shadow_mode
        
        # Load both models
        self.control_model = mlflow.pyfunc.load_model("models:/rec-model/Production")
        self.treatment_model = mlflow.pyfunc.load_model("models:/rec-model/Staging")
    
    def assign_variant(self, unit_id: str) -> VariantAssignment:
        
        hash_input = .encode()
        hash_value = (hashlib.md5(hash_input).hexdigest(), )
        bucket = (hash_value % ) /   
        
         bucket < .treatment_pct:
             VariantAssignment.TREATMENT
         VariantAssignment.CONTROL
    
      () -> :
        variant = .assign_variant(user_id)
        
         .shadow_mode:
            
            control_result =  ._run_model(.control_model, features)
            treatment_result =  ._run_model(.treatment_model, features)
            
            ._log_shadow_comparison(user_id, control_result, treatment_result)
             {**control_result, : }
        
         variant == VariantAssignment.TREATMENT:
            result =  ._run_model(.treatment_model, features)
        :
            result =  ._run_model(.control_model, features)
        
        
        ._log_assignment(user_id, variant.value, result)
        
         {**result, : variant.value, : .experiment_id}
    
      () -> :
         asyncio
        loop = asyncio.get_event_loop()
          loop.run_in_executor(, model.predict, features)
    
     ():
        
        event_bus.publish(, {
            : .experiment_id,
            : user_id,
            : variant,
            : datetime.now(timezone.utc).isoformat(),
            : hashlib.md5(
                (predictions).encode()
            ).hexdigest()[:],
        })
    
     ():
        
        metrics.histogram(
            ,
            value=compute_output_distance(control, treatment),
            tags={: .experiment_id}
        )

Statistical Analysis

# analysis/experiment_analyzer.py
import pandas as pd
import numpy as np
from scipy import stats
from dataclasses import dataclass
from typing import Optional

@dataclass
class ExperimentResult:
    metric: str
    control_mean: float
    treatment_mean: float
    relative_lift: float
    p_value: float
    confidence_interval_lower: float
    confidence_interval_upper: float
    is_significant: bool
    is_practically_significant: bool  # Meets MDE threshold
    sample_size_control: int
    sample_size_treatment: int
    recommendation: str

class ExperimentAnalyzer:
    def __init__(
        self,
        significance_level: float = 0.05,
        mde: float = 0.05
    ):
        self.alpha = significance_level
        self.mde = mde
    
    def analyze(
        self,
        df: pd.DataFrame,
        metric_col: str,
        variant_col: str = "variant",
        control_label: str = "control",
        treatment_label:  = 
    ) -> ExperimentResult:
        
        
        control = df[df[variant_col] == control_label][metric_col].dropna()
        treatment = df[df[variant_col] == treatment_label][metric_col].dropna()
        
        
        statistic, p_value = stats.ttest_ind(control, treatment, equal_var=)
        
        
        diff = treatment.mean() - control.mean()
        se = np.sqrt(control.std()** / (control) + treatment.std()** / (treatment))
        ci_margin = stats.t.ppf( - .alpha/, df=(control) + (treatment) - ) * se
        
        relative_lift = (treatment.mean() - control.mean()) / control.mean()
        
         ExperimentResult(
            metric=metric_col,
            control_mean=control.mean(),
            treatment_mean=treatment.mean(),
            relative_lift=relative_lift,
            p_value=p_value,
            confidence_interval_lower=diff - ci_margin,
            confidence_interval_upper=diff + ci_margin,
            is_significant=p_value < .alpha,
            is_practically_significant=(relative_lift) >= .mde,
            sample_size_control=(control),
            sample_size_treatment=(treatment),
            recommendation=._recommend(p_value, relative_lift)
        )
    
     () -> :
         p_value < .alpha  lift >= .mde:
             
         p_value < .alpha  lift < :
             
         p_value >= .alpha:
             
        :
             
    
     () -> :
        
        primary = .analyze(df, primary_metric)
        guardrails = [.analyze(df, m)  m  guardrail_metrics]
        
        
        guardrail_failures = [
            g  g  guardrails
             g.is_significant  g.relative_lift < -  
        ]
        
        overall_recommendation = primary.recommendation
         guardrail_failures:
            overall_recommendation = 
        
         {
            : (primary),
            : [(g)  g  guardrails],
            : [g.metric  g  guardrail_failures],
            : overall_recommendation,
            : (df[].() - df[].()).days,
        }



df = pd.read_parquet()
analyzer = ExperimentAnalyzer(significance_level=, mde=)

report = analyzer.full_report(
    df=df,
    primary_metric=,
    guardrail_metrics=[, , ]
)

()
()
()

Experiment Registry

# experiments/rec-model-v2.yaml — version-controlled experiment spec
name: rec-model-v2-ab-test
description: "Test new collaborative filtering model vs. baseline matrix factorization"
created_by: [e-mail]
created_at: 2024-01-15

hypothesis: "New model improves CTR by 5% through better user embedding representation"

model:
  control: "rec-model/Production (v8)"
  treatment: "rec-model/Staging (v9)"

traffic:
  total_pct: 20              # 20% of eligible users in experiment
  control_split: 50          # 50/50 within experiment
  treatment_split: 50
  assignment_unit: user_id   # Sticky by user
  eligible_filter: "registered_users_30d_active"
  excluded: ["new_users", "enterprise_accounts"]

metrics:
  primary: click_through_rate
  guardrails:
    - revenue_per_user         # No revenue regression
    - session_duration         # No engagement drop
    - error_rate               # No reliability issues
    - p99_latency_ms           


        
   
     


   
       
   
  
 

Rules

• One primary metric — multiple primary metrics cause multiple testing problems.
• Pre-register the hypothesis — define success criteria before looking at data.
• Never stop early based on significance — p-values fluctuate; run the full planned duration.
• Guardrail metrics block shipping — significant degradation in revenue or reliability blocks even a successful primary metric.
• Sticky assignment — users must always see the same variant; session-based assignment causes flip-flopping.
• Exclude novelty effects — first-week results often reflect novelty bias; analyze week 2+ separately.
• Sample ratio mismatch check — if treatment has 10% but you get 12%, the experiment is broken.
• Segment analysis after significance — don't hunt for winning segments before the primary metric is significant.
• Document the decision — the decision rationale (ship/rollback/iterate) is as valuable as the data.
• Power analysis before starting — never start an experiment you can't power within your timeline.