Ab Test Setup

# test_results.csv — A/B test results for the ab-test-setup analyzer
# Experiment: Homepage CTA button redesign (March 2026)
# Fields: variant, visitor_id, timestamp, converted, revenue_cents, device, source
# Control = existing blue button, Variant A = green button, Variant B = larger red button
variant,visitor_id,timestamp,converted,revenue_cents,device,source
control,V00101,2026-03-01 08:12:33,0,0,desktop,organic
control,V00102,2026-03-01 08:45:19,1,4900,desktop,organic
control,V00103,2026-03-01 09:03:41,0,0,mobile,paid
control,V00104,2026-03-01 09:22:08,0,0,mobile,organic
control,V00105,2026-03-01 10:15:50,1,2900,desktop,referral
control,V00106,2026-03-01 11:30:22,0,0,tablet,organic
control,V00107,2026-03-01 12:08:14,1,4900,desktop,paid
control,V00108,2026-03-01 13:44:59,0,0,mobile,organic
control,V00109,2026-03-01 14:20:33,0,0,mobile,paid
control,V00110,2026-03-01 15:55:18,0,0,desktop,organic
control,V00111,2026-03-02 07:33:41,1,2900,desktop,organic
control,V00112,2026-03-02 08:19:05,0,0,mobile,social
control,V00113,2026-03-02 09:48:22,0,0,mobile,organic
control,V00114,2026-03-02 10:30:17,1,9900,desktop,paid
control,V00115,2026-03-02 11:15:44,0,0,tablet,organic
control,V00116,2026-03-02 12:42:30,0,0,mobile,referral
control,V00117,2026-03-02 14:05:19,0,0,desktop,organic
control,V00118,2026-03-02 15:33:28,1,4900,desktop,paid
control,V00119,2026-03-03 08:10:42,0,0,mobile,organic
control,V00120,2026-03-03 09:25:51,0,0,mobile,social
variant_a,V00201,2026-03-01 08:14:20,1,4900,desktop,organic
variant_a,V00202,2026-03-01 08:50:33,0,0,mobile,organic
variant_a,V00203,2026-03-01 09:18:45,1,2900,desktop,paid
variant_a,V00204,2026-03-01 10:05:12,0,0,mobile,organic
variant_a,V00205,2026-03-01 10:42:38,1,4900,desktop,referral
variant_a,V00206,2026-03-01 11:28:55,0,0,tablet,organic
variant_a,V00207,2026-03-01 12:15:10,1,9900,desktop,paid
variant_a,V00208,2026-03-01 13:50:44,1,2900,mobile,organic
variant_a,V00209,2026-03-01 14:33:21,0,0,mobile,paid
variant_a,V00210,2026-03-01 16:05:18,0,0,desktop,organic
variant_a,V00211,2026-03-02 07:45:33,1,4900,desktop,organic
variant_a,V00212,2026-03-02 08:30:19,0,0,mobile,social
variant_a,V00213,2026-03-02 09:55:41,1,2900,desktop,organic
variant_a,V00214,2026-03-02 10:48:05,0,0,mobile,paid
variant_a,V00215,2026-03-02 11:22:38,1,4900,desktop,paid
variant_a,V00216,2026-03-02 12:50:14,0,0,tablet,referral
variant_a,V00217,2026-03-02 14:18:33,1,9900,desktop,organic
variant_a,V00218,2026-03-02 15:45:20,0,0,mobile,organic
variant_a,V00219,2026-03-03 08:22:15,0,0,mobile,social
variant_a,V00220,2026-03-03 09:40:30,1,2900,desktop,paid
variant_b,V00301,2026-03-01 08:18:42,1,4900,desktop,organic
variant_b,V00302,2026-03-01 08:55:10,1,2900,mobile,organic
variant_b,V00303,2026-03-01 09:25:33,0,0,desktop,paid
variant_b,V00304,2026-03-01 10:10:48,1,4900,mobile,organic
variant_b,V00305,2026-03-01 10:50:15,1,9900,desktop,referral
variant_b,V00306,2026-03-01 11:35:22,0,0,tablet,organic
variant_b,V00307,2026-03-01 12:20:40,1,4900,desktop,paid
variant_b,V00308,2026-03-01 13:58:18,1,2900,mobile,organic
variant_b,V00309,2026-03-01 14:40:55,0,0,mobile,paid
variant_b,V00310,2026-03-01 16:12:30,1,4900,desktop,organic
variant_b,V00311,2026-03-02 07:50:18,1,9900,desktop,organic
variant_b,V00312,2026-03-02 08:38:44,0,0,mobile,social
variant_b,V00313,2026-03-02 10:02:19,1,2900,desktop,organic
variant_b,V00314,2026-03-02 10:55:33,1,4900,mobile,paid
variant_b,V00315,2026-03-02 11:30:50,0,0,desktop,paid
variant_b,V00316,2026-03-02 12:58:42,1,4900,tablet,referral
variant_b,V00317,2026-03-02 14:25:10,0,0,desktop,organic
variant_b,V00318,2026-03-02 15:52:38,1,9900,mobile,organic
variant_b,V00319,2026-03-03 08:30:25,0,0,mobile,social
variant_b,V00320,2026-03-03 09:48:12,1,2900,desktop,paid

#!/usr/bin/env python3
"""
A/B Test Results Analyzer

Analyzes A/B test results with statistical significance testing, confidence
intervals, effect size calculation, and actionable recommendations.

Supports two-proportion z-tests for conversion rate experiments.

Expected JSON input: {"test_name", "variants": {"control": {"visitors", "conversions"},
  "treatment": {"visitors", "conversions"}}, "significance_level"}

Usage:
    python results_analyzer.py results.json
    python results_analyzer.py results.json --format json
    python results_analyzer.py results.json --batch
    python results_analyzer.py results.json --bayesian
"""

import argparse
import json
import math
import sys
from typing import Any, Dict, List, Optional, Tuple


def norm_ppf(p: float) -> float:
    """Inverse normal CDF using rational approximation."""
    if p <= 0 or p >= 1:
        raise ValueError("p must be between 0 and 1")
    if p < 0.5:
        return -norm_ppf(1 - p)
    t = math.sqrt(-2 * math.log(1 - p))
    c0, c1, c2 = 2.515517, 0.802853, 0.010328
    d1, d2, d3 = 1.432788, 0.189269, 0.001308
    return t - (c0 + c1 * t + c2 * t**2) / (1 + d1 * t + d2 * t**2 + d3 * t**3)


def norm_cdf(x: float) -> float:
    """Standard normal CDF."""
    return 0.5 * (1 + math.erf(x / math.sqrt(2)))


def two_proportion_z_test(
    n1: int, x1: int, n2: int, x2: int
) -> Dict[str, float]:
    """
    Perform a two-proportion z-test.

    Args:
        n1: Sample size of control
        x1: Conversions in control
        n2: Sample size of treatment
        x2: Conversions in treatment

    Returns:
        z_statistic, p_value (two-sided)
    """
    p1 = x1 / n1 if n1 > 0 else 0
    p2 = x2 / n2 if n2 > 0 else 0
    p_pooled = (x1 + x2) / (n1 + n2) if (n1 + n2) > 0 else 0

    se = math.sqrt(p_pooled * (1 - p_pooled) * (1/n1 + 1/n2)) if n1 > 0 and n2 > 0 else 0

    if se == 0:
        return {"z_statistic": 0, "p_value": 1.0, "se": 0}

    z = (p2 - p1) / se
    p_value = 2 * (1 - norm_cdf(abs(z)))

    return {"z_statistic": round(z, 4), "p_value": round(p_value, 6), "se": round(se, 6)}


def confidence_interval(
    n1: int, x1: int, n2: int, x2: int, alpha: float = 0.05
) -> Dict[str, float]:
    """Calculate confidence interval for difference in proportions."""
    p1 = x1 / n1 if n1 > 0 else 0
    p2 = x2 / n2 if n2 > 0 else 0
    diff = p2 - p1

    se = math.sqrt(p1 * (1 - p1) / n1 + p2 * (1 - p2) / n2) if n1 > 0 and n2 > 0 else 0
    z = norm_ppf(1 - alpha / 2)

    margin = z * se
    return {
        "point_estimate": round(diff, 6),
        "lower_bound": round(diff - margin, 6),
        "upper_bound": round(diff + margin, 6),
        "margin_of_error": round(margin, 6),
        "confidence_level": round((1 - alpha) * 100, 0),
    }


def cohens_h(p1: float, p2: float) -> float:
    """Calculate Cohen's h effect size for proportions."""
    return 2 * math.asin(math.sqrt(p2)) - 2 * math.asin(math.sqrt(p1))


def bayesian_analysis(n1: int, x1: int, n2: int, x2: int,
                      n_samples: int = 100000) -> Dict[str, Any]:
    """Simple Bayesian analysis using Beta-Binomial model with grid approximation.
    Uses a uniform Beta(1,1) prior."""
    # Beta posterior parameters
    alpha1, beta1 = x1 + 1, n1 - x1 + 1
    alpha2, beta2 = x2 + 1, n2 - x2 + 1

    # Compute probability that treatment > control using numerical integration
    # Grid-based approximation
    grid_size = 1000
    prob_b_better = 0.0
    step = 1.0 / grid_size

    for i in range(grid_size):
        p = (i + 0.5) * step
        # Beta PDF approximation using Stirling's for large factorials
        # Use log-space for numerical stability
        # For simplicity, use the analytical result for Beta distributions
        pass

    # Simpler approximation: normal approximation to beta posterior
    mean1 = alpha1 / (alpha1 + beta1)
    var1 = (alpha1 * beta1) / ((alpha1 + beta1) ** 2 * (alpha1 + beta1 + 1))
    mean2 = alpha2 / (alpha2 + beta2)
    var2 = (alpha2 * beta2) / ((alpha2 + beta2) ** 2 * (alpha2 + beta2 + 1))

    diff_mean = mean2 - mean1
    diff_std = math.sqrt(var1 + var2)

    if diff_std > 0:
        prob_b_better = norm_cdf(diff_mean / diff_std)
    else:
        prob_b_better = 0.5

    # Credible interval for the difference
    z95 = 1.96
    ci_lower = diff_mean - z95 * diff_std
    ci_upper = diff_mean + z95 * diff_std

    return {
        "probability_treatment_better": round(prob_b_better, 4),
        "expected_difference": round(diff_mean, 6),
        "credible_interval_95": {
            "lower": round(ci_lower, 6),
            "upper": round(ci_upper, 6),
        },
        "control_posterior_mean": round(mean1, 6),
        "treatment_posterior_mean": round(mean2, 6),
    }


def analyze_test(data: Dict[str, Any], include_bayesian: bool = False) -> Dict[str, Any]:
    """Analyze a single A/B test."""
    test_name = data.get("test_name", "Unnamed Test")
    metric = data.get("metric", "conversion_rate")
    alpha = data.get("significance_level", 0.05)
    variants = data.get("variants", {})

    # Support both dict and list format for variants
    if isinstance(variants, list):
        if len(variants) >= 2:
            control = variants[0]
            treatment = variants[1]
        else:
            return {"error": "Need at least 2 variants"}
    else:
        control = variants.get("control", {})
        treatment = variants.get("treatment", {})

    n1 = control.get("visitors", 0)
    x1 = control.get("conversions", 0)
    n2 = treatment.get("visitors", 0)
    x2 = treatment.get("conversions", 0)

    if n1 == 0 or n2 == 0:
        return {"error": "Both variants must have visitors > 0", "test_name": test_name}

    p1 = x1 / n1
    p2 = x2 / n2
    relative_change = (p2 - p1) / p1 if p1 > 0 else 0

    # Statistical test
    z_test = two_proportion_z_test(n1, x1, n2, x2)
    ci = confidence_interval(n1, x1, n2, x2, alpha)
    effect_size = cohens_h(p1, p2)

    # Significance determination
    is_significant = z_test["p_value"] < alpha
    direction = "positive" if p2 > p1 else "negative" if p2 < p1 else "neutral"

    # Effect size interpretation
    abs_h = abs(effect_size)
    if abs_h < 0.2:
        effect_label = "negligible"
    elif abs_h < 0.5:
        effect_label = "small"
    elif abs_h < 0.8:
        effect_label = "medium"
    else:
        effect_label = "large"

    # Recommendation
    if is_significant and direction == "positive":
        recommendation = "SHIP - Statistically significant positive result"
        confidence = "high"
    elif is_significant and direction == "negative":
        recommendation = "REVERT - Statistically significant negative result"
        confidence = "high"
    elif not is_significant and abs(relative_change) > 0.05:
        recommendation = "EXTEND - Trending but not yet significant. Consider running longer."
        confidence = "low"
    else:
        recommendation = "NO EFFECT - No meaningful difference detected"
        confidence = "medium"

    result = {
        "test_name": test_name,
        "metric": metric,
        "control": {
            "visitors": n1,
            "conversions": x1,
            "rate": round(p1, 6),
            "rate_pct": round(p1 * 100, 3),
        },
        "treatment": {
            "visitors": n2,
            "conversions": x2,
            "rate": round(p2, 6),
            "rate_pct": round(p2 * 100, 3),
        },
        "difference": {
            "absolute": round(p2 - p1, 6),
            "absolute_pct": round((p2 - p1) * 100, 3),
            "relative_pct": round(relative_change * 100, 2),
            "direction": direction,
        },
        "statistical_test": {
            "method": "Two-proportion z-test (two-sided)",
            "z_statistic": z_test["z_statistic"],
            "p_value": z_test["p_value"],
            "significance_level": alpha,
            "is_significant": is_significant,
        },
        "confidence_interval": ci,
        "effect_size": {
            "cohens_h": round(effect_size, 4),
            "interpretation": effect_label,
        },
        "recommendation": recommendation,
        "confidence": confidence,
    }

    if include_bayesian:
        result["bayesian"] = bayesian_analysis(n1, x1, n2, x2)

    return result


def print_human(results: List[Dict[str, Any]]) -> None:
    """Print analysis in human-readable format."""
    for r in results:
        if "error" in r:
            print(f"\n  Error in {r.get('test_name', 'unknown')}: {r['error']}")
            continue

        print("=" * 65)
        print(f"  A/B Test Results: {r['test_name']}")
        print("=" * 65)

        c = r["control"]
        t = r["treatment"]
        print(f"\n  --- Variant Performance ({r['metric']}) ---")
        print(f"  {'Variant':<12} {'Visitors':>10} {'Conversions':>12} {'Rate':>10}")
        print(f"  {'-'*12} {'-'*10} {'-'*12} {'-'*10}")
        print(f"  {'Control':<12} {c['visitors']:>10,} {c['conversions']:>12,} {c['rate_pct']:>9.3f}%")
        print(f"  {'Treatment':<12} {t['visitors']:>10,} {t['conversions']:>12,} {t['rate_pct']:>9.3f}%")

        d = r["difference"]
        sign = "+" if d["relative_pct"] >= 0 else ""
        print(f"\n  --- Effect ---")
        print(f"  Absolute Change:  {sign}{d['absolute_pct']:.3f}pp")
        print(f"  Relative Change:  {sign}{d['relative_pct']:.2f}%")
        print(f"  Direction:        {d['direction']}")

        s = r["statistical_test"]
        print(f"\n  --- Statistical Significance ---")
        print(f"  Z-statistic:  {s['z_statistic']:.4f}")
        print(f"  P-value:      {s['p_value']:.6f}")
        print(f"  Alpha:        {s['significance_level']}")
        sig_label = "YES - Statistically significant" if s["is_significant"] else "NO - Not statistically significant"
        print(f"  Significant:  {sig_label}")

        ci = r["confidence_interval"]
        print(f"\n  --- {ci['confidence_level']:.0f}% Confidence Interval ---")
        print(f"  [{ci['lower_bound']*100:+.3f}pp, {ci['upper_bound']*100:+.3f}pp]")
        print(f"  Margin of Error: +/-{ci['margin_of_error']*100:.3f}pp")

        e = r["effect_size"]
        print(f"\n  --- Effect Size ---")
        print(f"  Cohen's h: {e['cohens_h']:.4f} ({e['interpretation']})")

        if "bayesian" in r:
            b = r["bayesian"]
            print(f"\n  --- Bayesian Analysis ---")
            print(f"  P(Treatment > Control): {b['probability_treatment_better']*100:.1f}%")
            print(f"  Expected Difference:    {b['expected_difference']*100:.3f}pp")
            ci_b = b["credible_interval_95"]
            print(f"  95% Credible Interval:  [{ci_b['lower']*100:.3f}pp, {ci_b['upper']*100:.3f}pp]")

        print(f"\n  {'=' * 61}")
        print(f"  RECOMMENDATION: {r['recommendation']}")
        print(f"  Confidence: {r['confidence']}")
        print()


def main():
    parser = argparse.ArgumentParser(
        description="Analyze A/B test results with statistical significance testing"
    )
    parser.add_argument("file", help="JSON file with test results")
    parser.add_argument("--format", choices=["human", "json"], default="human", help="Output format")
    parser.add_argument("--batch", action="store_true",
                        help="Process multiple tests (expects 'tests' array)")
    parser.add_argument("--bayesian", action="store_true",
                        help="Include Bayesian analysis alongside frequentist")
    args = parser.parse_args()

    with open(args.file, "r", encoding="utf-8") as f:
        data = json.load(f)

    if args.batch:
        tests = data.get("tests", [])
    else:
        tests = [data]

    results = [analyze_test(t, args.bayesian) for t in tests]

    if args.format == "json":
        output = {"results": results} if len(results) > 1 else results[0]
        print(json.dumps(output, indent=2))
    else:
        print_human(results)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
A/B Test Sample Size Calculator

Calculates required sample sizes for statistical significance in A/B tests.
Supports conversion rate (proportion) tests with configurable significance
level, power, and minimum detectable effect.

Usage:
    python sample_size_calculator.py --baseline 0.05 --mde 0.10
    python sample_size_calculator.py --baseline 0.05 --mde 0.10 --power 0.90
    python sample_size_calculator.py --baseline 0.05 --mde 0.10 --daily-traffic 5000
    python sample_size_calculator.py --baseline 0.05 --mde 0.10 --format json
"""

import argparse
import json
import math
import sys
from typing import Any, Dict, List, Optional, Tuple


# Standard normal distribution quantiles (using rational approximation)
def norm_ppf(p: float) -> float:
    """Inverse normal CDF (percent point function) using rational approximation.
    Abramowitz and Stegun approximation 26.2.23. Accurate to ~4.5e-4."""
    if p <= 0 or p >= 1:
        raise ValueError("p must be between 0 and 1 exclusive")
    if p < 0.5:
        return -norm_ppf(1 - p)

    t = math.sqrt(-2 * math.log(1 - p))
    # Coefficients for rational approximation
    c0, c1, c2 = 2.515517, 0.802853, 0.010328
    d1, d2, d3 = 1.432788, 0.189269, 0.001308
    return t - (c0 + c1 * t + c2 * t**2) / (1 + d1 * t + d2 * t**2 + d3 * t**3)


def norm_cdf(x: float) -> float:
    """Standard normal CDF using error function approximation."""
    return 0.5 * (1 + math.erf(x / math.sqrt(2)))


def calculate_sample_size(
    baseline: float,
    mde_relative: float,
    alpha: float = 0.05,
    power: float = 0.80,
    two_sided: bool = True,
    variants: int = 2,
) -> Dict[str, Any]:
    """
    Calculate required sample size per variant for a proportion test.

    Args:
        baseline: Baseline conversion rate (e.g., 0.05 for 5%)
        mde_relative: Minimum detectable effect as relative change (e.g., 0.10 for 10% lift)
        alpha: Significance level (default 0.05)
        power: Statistical power (default 0.80)
        two_sided: Whether to use two-sided test (default True)
        variants: Number of variants including control (default 2)
    """
    # Calculate treatment rate
    absolute_effect = baseline * mde_relative
    treatment_rate = baseline + absolute_effect

    if treatment_rate <= 0 or treatment_rate >= 1:
        return {"error": f"Treatment rate ({treatment_rate}) must be between 0 and 1"}

    # Z-scores
    if two_sided:
        z_alpha = norm_ppf(1 - alpha / 2)
    else:
        z_alpha = norm_ppf(1 - alpha)
    z_beta = norm_ppf(power)

    # Sample size formula for two-proportion z-test
    p1 = baseline
    p2 = treatment_rate
    p_bar = (p1 + p2) / 2

    numerator = (z_alpha * math.sqrt(2 * p_bar * (1 - p_bar)) +
                 z_beta * math.sqrt(p1 * (1 - p1) + p2 * (1 - p2))) ** 2
    denominator = (p1 - p2) ** 2

    n_per_variant = math.ceil(numerator / denominator)
    n_total = n_per_variant * variants

    return {
        "baseline_rate": baseline,
        "treatment_rate": round(treatment_rate, 6),
        "absolute_effect": round(absolute_effect, 6),
        "relative_effect_pct": round(mde_relative * 100, 2),
        "significance_level": alpha,
        "power": power,
        "two_sided": two_sided,
        "variants": variants,
        "sample_per_variant": n_per_variant,
        "total_sample": n_total,
        "z_alpha": round(z_alpha, 4),
        "z_beta": round(z_beta, 4),
    }


def calculate_duration(total_sample: int, daily_traffic: int,
                       allocation_pct: float = 1.0) -> Dict[str, Any]:
    """Calculate test duration from sample size and traffic."""
    effective_daily = daily_traffic * allocation_pct
    if effective_daily <= 0:
        return {"error": "Effective daily traffic must be positive"}

    days_needed = math.ceil(total_sample / effective_daily)
    weeks_needed = math.ceil(days_needed / 7)

    # Minimum recommended duration (2 full weeks for day-of-week effects)
    min_days = 14
    recommended_days = max(days_needed, min_days)

    return {
        "daily_traffic": daily_traffic,
        "allocation_pct": allocation_pct,
        "effective_daily_traffic": int(effective_daily),
        "days_needed": days_needed,
        "weeks_needed": weeks_needed,
        "recommended_days": recommended_days,
        "recommended_weeks": math.ceil(recommended_days / 7),
    }


def sensitivity_table(baseline: float, alpha: float, power: float,
                      mde_values: Optional[List[float]] = None) -> List[Dict[str, Any]]:
    """Generate sensitivity table across different MDE values."""
    if mde_values is None:
        mde_values = [0.05, 0.08, 0.10, 0.15, 0.20, 0.25, 0.30]

    rows = []
    for mde in mde_values:
        result = calculate_sample_size(baseline, mde, alpha, power)
        if "error" not in result:
            rows.append({
                "mde_pct": round(mde * 100, 1),
                "absolute_effect": round(baseline * mde, 6),
                "treatment_rate": round(baseline * (1 + mde), 6),
                "sample_per_variant": result["sample_per_variant"],
                "total_sample": result["total_sample"],
            })
    return rows


def power_table(baseline: float, mde_relative: float, alpha: float,
                power_values: Optional[List[float]] = None) -> List[Dict[str, Any]]:
    """Generate table across different power levels."""
    if power_values is None:
        power_values = [0.70, 0.75, 0.80, 0.85, 0.90, 0.95]

    rows = []
    for pwr in power_values:
        result = calculate_sample_size(baseline, mde_relative, alpha, pwr)
        if "error" not in result:
            rows.append({
                "power_pct": round(pwr * 100, 0),
                "sample_per_variant": result["sample_per_variant"],
                "total_sample": result["total_sample"],
            })
    return rows


def print_human(result: Dict, duration: Optional[Dict], sens_table: List[Dict],
                pwr_table: List[Dict]) -> None:
    """Print results in human-readable format."""
    print("=" * 60)
    print("  A/B Test Sample Size Calculator")
    print("=" * 60)

    if "error" in result:
        print(f"\n  Error: {result['error']}")
        return

    print(f"\n  --- Test Parameters ---")
    print(f"  Baseline Rate:       {result['baseline_rate']*100:.2f}%")
    print(f"  Treatment Rate:      {result['treatment_rate']*100:.2f}%")
    print(f"  Relative MDE:        {result['relative_effect_pct']:.1f}%")
    print(f"  Absolute Effect:     {result['absolute_effect']*100:.3f}pp")
    print(f"  Significance Level:  {result['significance_level']*100:.0f}% (alpha)")
    print(f"  Statistical Power:   {result['power']*100:.0f}%")
    print(f"  Test Type:           {'Two-sided' if result['two_sided'] else 'One-sided'}")
    print(f"  Variants:            {result['variants']}")

    print(f"\n  --- Required Sample Size ---")
    print(f"  Per Variant:         {result['sample_per_variant']:,}")
    print(f"  Total:               {result['total_sample']:,}")

    if duration and "error" not in duration:
        print(f"\n  --- Duration Estimate ---")
        print(f"  Daily Traffic:       {duration['daily_traffic']:,}")
        if duration["allocation_pct"] < 1:
            print(f"  Traffic Allocation:  {duration['allocation_pct']*100:.0f}%")
            print(f"  Effective Daily:     {duration['effective_daily_traffic']:,}")
        print(f"  Days Needed:         {duration['days_needed']}")
        print(f"  Recommended:         {duration['recommended_days']} days ({duration['recommended_weeks']} weeks)")
        if duration['recommended_days'] > duration['days_needed']:
            print(f"  (Extended to {duration['recommended_days']} days minimum for day-of-week coverage)")

    # Sensitivity table
    if sens_table:
        print(f"\n  --- Sensitivity: Sample Size by MDE ---")
        print(f"  {'MDE':>6} {'Abs Effect':>11} {'Treatment':>10} {'Per Variant':>12} {'Total':>10}")
        print(f"  {'-'*6} {'-'*11} {'-'*10} {'-'*12} {'-'*10}")
        for row in sens_table:
            current = " <--" if abs(row["mde_pct"] - result["relative_effect_pct"]) < 0.01 else ""
            print(f"  {row['mde_pct']:>5.1f}% {row['absolute_effect']*100:>10.3f}pp "
                  f"{row['treatment_rate']*100:>9.2f}% {row['sample_per_variant']:>11,} "
                  f"{row['total_sample']:>9,}{current}")

    # Power table
    if pwr_table:
        print(f"\n  --- Sensitivity: Sample Size by Power ---")
        print(f"  {'Power':>6} {'Per Variant':>12} {'Total':>10}")
        print(f"  {'-'*6} {'-'*12} {'-'*10}")
        for row in pwr_table:
            current = " <--" if abs(row["power_pct"] - result["power"] * 100) < 0.01 else ""
            print(f"  {row['power_pct']:>5.0f}% {row['sample_per_variant']:>11,} "
                  f"{row['total_sample']:>9,}{current}")

    print()


def main():
    parser = argparse.ArgumentParser(
        description="Calculate required sample sizes for A/B test statistical significance"
    )
    parser.add_argument("--baseline", type=float, required=True,
                        help="Baseline conversion rate (e.g., 0.05 for 5%%)")
    parser.add_argument("--mde", type=float, required=True,
                        help="Minimum detectable effect as relative change (e.g., 0.10 for 10%% lift)")
    parser.add_argument("--alpha", type=float, default=0.05,
                        help="Significance level (default: 0.05)")
    parser.add_argument("--power", type=float, default=0.80,
                        help="Statistical power (default: 0.80)")
    parser.add_argument("--daily-traffic", type=int, help="Daily traffic for duration estimation")
    parser.add_argument("--allocation", type=float, default=1.0,
                        help="Fraction of traffic allocated to test (default: 1.0)")
    parser.add_argument("--one-sided", action="store_true", help="Use one-sided test")
    parser.add_argument("--variants", type=int, default=2, help="Number of variants (default: 2)")
    parser.add_argument("--format", choices=["human", "json"], default="human", help="Output format")
    args = parser.parse_args()

    result = calculate_sample_size(
        baseline=args.baseline,
        mde_relative=args.mde,
        alpha=args.alpha,
        power=args.power,
        two_sided=not args.one_sided,
        variants=args.variants,
    )

    duration = None
    if args.daily_traffic:
        duration = calculate_duration(result["total_sample"], args.daily_traffic, args.allocation)

    sens = sensitivity_table(args.baseline, args.alpha, args.power)
    pwr = power_table(args.baseline, args.mde, args.alpha)

    if args.format == "json":
        output = {"sample_size": result}
        if duration:
            output["duration"] = duration
        output["sensitivity_by_mde"] = sens
        output["sensitivity_by_power"] = pwr
        print(json.dumps(output, indent=2))
    else:
        print_human(result, duration, sens, pwr)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
A/B Test Plan Designer

Designs comprehensive A/B test plans with hypothesis documentation, metric
definitions, sample size calculations, duration estimates, and risk assessment.

Expected JSON config with: test_name, hypothesis, metric_primary, baseline_rate,
  minimum_detectable_effect, daily_traffic, variants

Usage:
    python test_designer.py test_config.json
    python test_designer.py test_config.json --format json
    python test_designer.py test_config.json --template minimal
"""

import argparse
import json
import math
import sys
from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional


def norm_ppf(p: float) -> float:
    """Inverse normal CDF using rational approximation."""
    if p <= 0 or p >= 1:
        raise ValueError("p must be between 0 and 1")
    if p < 0.5:
        return -norm_ppf(1 - p)
    t = math.sqrt(-2 * math.log(1 - p))
    c0, c1, c2 = 2.515517, 0.802853, 0.010328
    d1, d2, d3 = 1.432788, 0.189269, 0.001308
    return t - (c0 + c1 * t + c2 * t**2) / (1 + d1 * t + d2 * t**2 + d3 * t**3)


def calculate_sample_size(baseline: float, mde_relative: float,
                          alpha: float, power: float) -> int:
    """Calculate required sample size per variant."""
    treatment = baseline * (1 + mde_relative)
    if treatment <= 0 or treatment >= 1:
        return 0
    z_alpha = norm_ppf(1 - alpha / 2)
    z_beta = norm_ppf(power)
    p_bar = (baseline + treatment) / 2
    numerator = (z_alpha * math.sqrt(2 * p_bar * (1 - p_bar)) +
                 z_beta * math.sqrt(baseline * (1 - baseline) + treatment * (1 - treatment))) ** 2
    denominator = (baseline - treatment) ** 2
    return math.ceil(numerator / denominator)


def assess_risks(config: Dict[str, Any], duration_days: int) -> List[Dict[str, str]]:
    """Assess test risks and generate mitigations."""
    risks = []

    # Duration risk
    if duration_days > 30:
        risks.append({
            "risk": "Long test duration",
            "severity": "medium",
            "detail": f"Test requires {duration_days} days. External factors may contaminate results.",
            "mitigation": "Consider increasing MDE or traffic allocation to shorten duration.",
        })
    if duration_days > 60:
        risks[-1]["severity"] = "high"

    # Sample size risk
    baseline = config.get("baseline_rate", 0)
    if baseline < 0.01:
        risks.append({
            "risk": "Low baseline rate",
            "severity": "high",
            "detail": f"Baseline rate of {baseline*100:.2f}% requires very large samples.",
            "mitigation": "Consider using a broader metric or accepting larger MDE.",
        })

    # Novelty effect risk
    risks.append({
        "risk": "Novelty/primacy effects",
        "severity": "low",
        "detail": "Users may react differently to new experiences initially.",
        "mitigation": "Run test for minimum 14 days. Consider excluding first 3 days from analysis.",
    })

    # Multiple metrics risk
    secondary = config.get("metric_secondary", [])
    if len(secondary) > 3:
        risks.append({
            "risk": "Multiple comparison inflation",
            "severity": "medium",
            "detail": f"Testing {len(secondary)} secondary metrics increases false positive risk.",
            "mitigation": "Apply Bonferroni correction. Only primary metric determines ship decision.",
        })

    # Interaction risk
    concurrent = config.get("concurrent_tests", [])
    if concurrent:
        risks.append({
            "risk": "Test interaction effects",
            "severity": "medium",
            "detail": f"Running concurrently with: {', '.join(concurrent)}",
            "mitigation": "Ensure non-overlapping audiences or verify no shared metrics.",
        })

    return risks


def generate_test_plan(config: Dict[str, Any]) -> Dict[str, Any]:
    """Generate a complete test plan from configuration."""
    # Extract config values with defaults
    test_name = config.get("test_name", "Untitled Test")
    hypothesis = config.get("hypothesis", "")
    metric_primary = config.get("metric_primary", "conversion_rate")
    metric_secondary = config.get("metric_secondary", [])
    baseline = config.get("baseline_rate", 0.05)
    mde = config.get("minimum_detectable_effect", 0.10)
    alpha = config.get("significance_level", 0.05)
    power = config.get("power", 0.80)
    variants = config.get("variants", [
        {"name": "control", "description": "Current experience"},
        {"name": "treatment", "description": "New experience"},
    ])
    daily_traffic = config.get("daily_traffic", 0)
    allocation = config.get("allocation", {})

    # Default equal allocation
    if not allocation:
        n_variants = len(variants)
        allocation = {v["name"]: round(1.0 / n_variants, 2) for v in variants}

    # Calculate sample size
    n_per_variant = calculate_sample_size(baseline, mde, alpha, power)
    n_total = n_per_variant * len(variants)

    # Duration estimation
    duration_days = 0
    traffic_per_variant = 0
    if daily_traffic > 0:
        min_allocation = min(allocation.values())
        traffic_per_variant = int(daily_traffic * min_allocation)
        if traffic_per_variant > 0:
            duration_days = math.ceil(n_per_variant / traffic_per_variant)
            duration_days = max(duration_days, 14)  # Minimum 2 weeks

    # Start date (next Monday)
    today = datetime.now()
    days_to_monday = (7 - today.weekday()) % 7
    if days_to_monday == 0:
        days_to_monday = 7
    start_date = today + timedelta(days=days_to_monday)
    end_date = start_date + timedelta(days=duration_days) if duration_days > 0 else None
    analysis_date = end_date + timedelta(days=1) if end_date else None

    # Risk assessment
    risks = assess_risks(config, duration_days)

    # Guardrail metrics
    guardrails = config.get("guardrail_metrics", [])
    if not guardrails:
        guardrails = [
            {"metric": "error_rate", "threshold": "No more than 5% increase"},
            {"metric": "page_load_time", "threshold": "No more than 200ms increase"},
            {"metric": "bounce_rate", "threshold": "No more than 10% relative increase"},
        ]

    # Build plan
    plan = {
        "test_name": test_name,
        "created_date": today.strftime("%Y-%m-%d"),
        "status": "draft",
        "hypothesis": {
            "statement": hypothesis,
            "null_hypothesis": f"There is no difference in {metric_primary} between variants",
            "alternative_hypothesis": f"The treatment variant has a different {metric_primary} than control",
        },
        "metrics": {
            "primary": metric_primary,
            "secondary": metric_secondary,
            "guardrails": guardrails,
        },
        "statistical_design": {
            "test_type": "Two-proportion z-test (two-sided)",
            "baseline_rate": baseline,
            "minimum_detectable_effect": f"{mde*100:.1f}% relative",
            "absolute_effect": f"{baseline*mde*100:.3f}pp",
            "expected_treatment_rate": round(baseline * (1 + mde), 6),
            "significance_level": alpha,
            "power": power,
            "sample_per_variant": n_per_variant,
            "total_sample_required": n_total,
        },
        "variants": [
            {**v, "allocation_pct": round(allocation.get(v["name"], 1/len(variants)) * 100, 1)}
            for v in variants
        ],
        "timeline": {
            "daily_traffic": daily_traffic,
            "traffic_per_variant": traffic_per_variant,
            "estimated_duration_days": duration_days,
            "start_date": start_date.strftime("%Y-%m-%d") if start_date else None,
            "end_date": end_date.strftime("%Y-%m-%d") if end_date else None,
            "analysis_date": analysis_date.strftime("%Y-%m-%d") if analysis_date else None,
        },
        "risks": risks,
        "pre_launch_checklist": [
            {"item": "Hypothesis documented and reviewed by team", "complete": False},
            {"item": "Primary metric instrumented and validated", "complete": False},
            {"item": "Secondary metrics instrumented", "complete": False},
            {"item": "Guardrail metrics monitoring set up", "complete": False},
            {"item": "Variant implementation QA'd in staging", "complete": False},
            {"item": "Random assignment mechanism verified", "complete": False},
            {"item": "Stakeholders aligned on success criteria", "complete": False},
            {"item": "Analysis date committed (no peeking)", "complete": False},
            {"item": "Rollback plan documented", "complete": False},
        ],
        "analysis_plan": {
            "primary_analysis": f"Two-proportion z-test on {metric_primary} at alpha={alpha}",
            "secondary_analysis": f"Report point estimates and CIs for: {', '.join(metric_secondary)}" if metric_secondary else "N/A",
            "subgroup_analysis": config.get("segments", ["device_type", "new_vs_returning"]),
            "decision_framework": {
                "ship": f"Primary metric statistically significant (p < {alpha}) AND positive direction AND no guardrail violations",
                "iterate": "Primary metric trending positive but not significant. Consider extending test or larger MDE.",
                "revert": "Primary metric negative OR guardrail violation",
            },
        },
    }

    return plan


def print_human(plan: Dict[str, Any]) -> None:
    """Print test plan in human-readable format."""
    print("=" * 70)
    print(f"  A/B TEST PLAN: {plan['test_name']}")
    print("=" * 70)
    print(f"  Created: {plan['created_date']}  |  Status: {plan['status']}")

    # Hypothesis
    h = plan["hypothesis"]
    print(f"\n  --- Hypothesis ---")
    print(f"  H1: {h['statement']}")
    print(f"  H0: {h['null_hypothesis']}")

    # Metrics
    m = plan["metrics"]
    print(f"\n  --- Metrics ---")
    print(f"  Primary:    {m['primary']}")
    if m["secondary"]:
        print(f"  Secondary:  {', '.join(m['secondary'])}")
    print(f"  Guardrails:")
    for g in m["guardrails"]:
        print(f"    - {g['metric']}: {g['threshold']}")

    # Statistical design
    s = plan["statistical_design"]
    print(f"\n  --- Statistical Design ---")
    print(f"  Test Type:       {s['test_type']}")
    print(f"  Baseline Rate:   {s['baseline_rate']*100:.2f}%")
    print(f"  MDE:             {s['minimum_detectable_effect']}")
    print(f"  Expected Rate:   {s['expected_treatment_rate']*100:.2f}%")
    print(f"  Alpha:           {s['significance_level']}")
    print(f"  Power:           {s['power']*100:.0f}%")
    print(f"  Sample/Variant:  {s['sample_per_variant']:,}")
    print(f"  Total Sample:    {s['total_sample_required']:,}")

    # Variants
    print(f"\n  --- Variants ---")
    for v in plan["variants"]:
        print(f"  [{v['allocation_pct']:.0f}%] {v['name']}: {v['description']}")

    # Timeline
    t = plan["timeline"]
    print(f"\n  --- Timeline ---")
    if t["daily_traffic"] > 0:
        print(f"  Daily Traffic:    {t['daily_traffic']:,}")
        print(f"  Per Variant:      {t['traffic_per_variant']:,}/day")
        print(f"  Duration:         {t['estimated_duration_days']} days")
        print(f"  Start Date:       {t['start_date']}")
        print(f"  End Date:         {t['end_date']}")
        print(f"  Analysis Date:    {t['analysis_date']}")
    else:
        print(f"  Duration:         Provide daily_traffic to estimate")

    # Risks
    risks = plan["risks"]
    if risks:
        print(f"\n  --- Risks ({len(risks)}) ---")
        for r in risks:
            severity_icon = {"high": "!!!", "medium": " ! ", "low": "   "}.get(r["severity"], "   ")
            print(f"  [{severity_icon}] {r['risk']}: {r['detail']}")
            print(f"         Mitigation: {r['mitigation']}")

    # Decision framework
    d = plan["analysis_plan"]["decision_framework"]
    print(f"\n  --- Decision Framework ---")
    print(f"  SHIP:    {d['ship']}")
    print(f"  ITERATE: {d['iterate']}")
    print(f"  REVERT:  {d['revert']}")

    # Pre-launch checklist
    print(f"\n  --- Pre-Launch Checklist ---")
    for item in plan["pre_launch_checklist"]:
        status = "[x]" if item["complete"] else "[ ]"
        print(f"  {status} {item['item']}")

    print()


def main():
    parser = argparse.ArgumentParser(
        description="Design comprehensive A/B test plans with hypothesis, metrics, and duration"
    )
    parser.add_argument("file", help="JSON file with test configuration")
    parser.add_argument("--format", choices=["human", "json"], default="human", help="Output format")
    args = parser.parse_args()

    with open(args.file, "r", encoding="utf-8") as f:
        config = json.load(f)

    plan = generate_test_plan(config)

    if args.format == "json":
        print(json.dumps(plan, indent=2, default=str))
    else:
        print_human(plan)


if __name__ == "__main__":
    main()