Ab Testing Specialist

Population:  [Who are we testing?]
Treatment:   [What change are we making?]
Metric:      [What are we measuring?]
Direction:   [Do we expect increase or decrease?]
Magnitude:   [What is the minimum detectable effect?]
Timeline:    [How long will we run the test?]

Example:
Population:  All logged-in users on the checkout page
Treatment:   Single-page checkout vs. current multi-step checkout
Metric:      Checkout completion rate (primary), revenue per visitor (secondary)
Direction:   Increase
Magnitude:   2 percentage points (from 35% to 37%)
Timeline:    14 days minimum

from statsmodels.stats.power import NormalIndPower
import numpy as np

def calculate_sample_size(
    baseline_rate: float,
    minimum_detectable_effect: float,  # Absolute difference
    alpha: float = 0.05,
    power: float = 0.80,
    two_sided: bool = True,
):
    """Calculate required sample size per group for a proportion test."""
    effect_size = minimum_detectable_effect / np.sqrt(
        baseline_rate * (1 - baseline_rate)
    )
    analysis = NormalIndPower()
    n = analysis.solve_power(
        effect_size=effect_size,
        alpha=alpha,
        power=power,
        alternative='two-sided' if two_sided else 'larger',
    )
    return int(np.ceil(n))

# Example: Detect 2pp lift from 10% baseline
n = calculate_sample_size(baseline_rate=0.10, minimum_detectable_effect=0.02)
print(f"Sample size per group: {n:,}")
# With 100k visitors/day, test duration = 2 * n / 100_000

from statsmodels.stats.power import TTestIndPower

def sample_size_continuous(
    baseline_mean: float,
    baseline_std: float,
    minimum_detectable_effect: float,  # Absolute difference in means
    alpha: float = 0.05,
    power: float = 0.80,
):
    """Sample size for continuous metric (e.g., revenue per user)."""
    cohens_d = minimum_detectable_effect / baseline_std
    analysis = TTestIndPower()
    n = analysis.solve_power(effect_size=cohens_d, alpha=alpha, power=power)
    return int(np.ceil(n))

# Detect $2 lift in average order value (mean=$50, std=$30)
n = sample_size_continuous(50, 30, 2.0)
print(f"Sample size per group: {n:,}")

from scipy import stats
import numpy as np

def analyze_ab_test(
    control_visitors: int,
    control_conversions: int,
    treatment_visitors: int,
    treatment_conversions: int,
    alpha: float = 0.05,
):
    """Complete frequentist analysis of an A/B test."""
    # Conversion rates
    p_control = control_conversions / control_visitors
    p_treatment = treatment_conversions / treatment_visitors
    lift = (p_treatment - p_control) / p_control

    # Pooled proportion for z-test
    p_pool = (control_conversions + treatment_conversions) / (
        control_visitors + treatment_visitors
    )
    se = np.sqrt(p_pool * (1 - p_pool) * (1/control_visitors + 1/treatment_visitors))
    z_stat = (p_treatment - p_control) / se
    p_value = 2 * (1 - stats.norm.cdf(abs(z_stat)))

    # Confidence interval for the difference
    se_diff = np.sqrt(
        p_control * (1 - p_control) / control_visitors +
        p_treatment * (1 - p_treatment) / treatment_visitors
    )
    z_crit = stats.norm.ppf(1 - alpha / 2)
    ci_low = (p_treatment - p_control) - z_crit * se_diff
    ci_high = (p_treatment - p_control) + z_crit * se_diff

    return {
        'control_rate': p_control,
        'treatment_rate': p_treatment,
        'absolute_lift': p_treatment - p_control,
        'relative_lift': lift,
        'z_statistic': z_stat,
        'p_value': p_value,
        'significant': p_value < alpha,
        'ci_low': ci_low,
        'ci_high': ci_high,
    }

results = analyze_ab_test(
    control_visitors=50000,
    control_conversions=5000,
    treatment_visitors=50000,
    treatment_conversions=5400,
)

from scipy import stats
import numpy as np

def bayesian_ab_test(
    control_conversions: int,
    control_visitors: int,
    treatment_conversions: int,
    treatment_visitors: int,
    n_simulations: int = 100_000,
    prior_alpha: float = 1,
    prior_beta: float = 1,
):
    """Bayesian analysis using Beta-Binomial model."""
    # Posterior distributions (Beta)
    control_posterior = stats.beta(
        prior_alpha + control_conversions,
        prior_beta + control_visitors - control_conversions,
    )
    treatment_posterior = stats.beta(
        prior_alpha + treatment_conversions,
        prior_beta + treatment_visitors - treatment_conversions,
    )

    # Monte Carlo simulation
    control_samples = control_posterior.rvs(n_simulations)
    treatment_samples = treatment_posterior.rvs(n_simulations)

    # Probability that treatment is better
    prob_treatment_better = np.mean(treatment_samples > control_samples)

    # Expected lift distribution
    lift_samples = (treatment_samples - control_samples) / control_samples
    expected_lift = np.mean(lift_samples)
    lift_ci = np.percentile(lift_samples, [2.5, 97.5])

    # Expected loss (risk of choosing treatment if it is worse)
    loss_if_treatment = np.mean(np.maximum(control_samples - treatment_samples, 0))
    loss_if_control = np.mean(np.maximum(treatment_samples - control_samples, 0))

    return {
        'prob_treatment_better': prob_treatment_better,
        'expected_lift': expected_lift,
        'lift_ci_95': lift_ci,
        'expected_loss_treatment': loss_if_treatment,
        'expected_loss_control': loss_if_control,
    }

def sequential_test_boundary(n_looks: int, alpha: float = 0.05):
    """Calculate adjusted significance thresholds for sequential testing."""
    # O'Brien-Fleming spending function
    from scipy.stats import norm
    import numpy as np

    info_fractions = np.linspace(1/n_looks, 1.0, n_looks)
    boundaries = []

    for t in info_fractions:
        # O'Brien-Fleming boundary
        z_boundary = norm.ppf(1 - alpha / 2) / np.sqrt(t)
        p_boundary = 2 * (1 - norm.cdf(z_boundary))
        boundaries.append({
            'look': int(t * n_looks),
            'info_fraction': t,
            'z_boundary': z_boundary,
            'p_threshold': p_boundary,
        })

    return pd.DataFrame(boundaries)

# Plan 5 interim analyses
boundaries = sequential_test_boundary(n_looks=5)
print(boundaries)
# Early looks require very strong evidence; final look is near alpha=0.05

Problem: Checking results daily and stopping when significant
Impact:  Inflated false positive rate (up to 30% instead of 5%)
Fix:     Pre-commit to sample size, or use sequential testing methods

Problem: Testing 20 metrics and highlighting the one that is significant
Impact:  1 - (1 - 0.05)^20 = 64% chance of at least one false positive
Fix:     Designate one primary metric; apply Bonferroni or FDR correction

Problem: Overall result differs from every segment's result
Example: Treatment wins overall but loses in mobile AND desktop
         (because treatment got more high-converting desktop traffic)
Fix:     Check results across key segments; use stratified analysis

Problem: New UI gets more clicks initially due to curiosity
Impact:  Overstated lift that decays over time
Fix:     Run test for 2+ weeks; analyze new-user cohort separately
         from those who switched mid-experiment

Problem: Control users are affected by treatment users (e.g., social features)
Impact:  Understated or biased treatment effect
Fix:     Cluster randomization (randomize by region, team, or network cluster)

Problem: Test cannot detect realistic effect sizes
Impact:  Many "no result" tests that waste time
Fix:     Calculate sample size beforehand; accept larger MDE or run longer

def segment_analysis(df, metric_col, treatment_col, segment_col, alpha=0.05):
    """Analyze A/B test results across segments."""
    results = []

    for segment in df[segment_col].unique():
        seg_data = df[df[segment_col] == segment]
        control = seg_data[seg_data[treatment_col] == 'control'][metric_col]
        treatment = seg_data[seg_data[treatment_col] == 'treatment'][metric_col]

        t_stat, p_val = stats.ttest_ind(control, treatment)
        lift = treatment.mean() - control.mean()

        results.append({
            'segment': segment,
            'n_control': len(control),
            'n_treatment': len(treatment),
            'control_mean': control.mean(),
            'treatment_mean': treatment.mean(),
            'absolute_lift': lift,
            'relative_lift': lift / control.mean(),
            'p_value': p_val,
            'significant': p_val < alpha,
        })

    return pd.DataFrame(results).sort_values('relative_lift', ascending=False)

def thompson_sampling_step(arms_data):
    """One step of Thompson Sampling for multi-armed bandit."""
    samples = {}
    for arm_name, data in arms_data.items():
        alpha = 1 + data['successes']
        beta = 1 + data['failures']
        samples[arm_name] = np.random.beta(alpha, beta)

    chosen_arm = max(samples, key=samples.get)
    return chosen_arm, samples

# When to use Bandits vs. A/B Tests
# Bandits: Optimizing during the test (minimize regret)
# A/B:     Need clean causal measurement (maximize learning)