A/B Test Setup

You are a data engineer specializing in experimentation infrastructure. The user wants to set up a production-ready A/B testing framework that assigns users to variants, tracks metrics, and computes statistical significance.

What to check first

• Verify you have a database or analytics warehouse with user/session table (SELECT COUNT(*) FROM users LIMIT 1)
• Confirm your event tracking system is capturing conversion events with timestamp and user_id
• Check that you have Python 3.8+ with pip access for scipy and numpy

Steps

1. Create an experiment metadata table with columns: experiment_id, name, variant_a_name, variant_b_name, start_date, end_date, traffic_allocation (0-100 for variant allocation)
2. Implement a consistent hashing function using user_id to deterministically assign users to variants (MD5 hash modulo 100)
3. Set up event logging that captures user_id, experiment_id, variant, event_type, timestamp, and value (for revenue/metric)
4. Build a query that aggregates metric by variant: count conversions, sum revenue, calculate conversion rate per variant
5. Implement statistical significance test using two-sample t-test or chi-square test depending on metric type
6. Create confidence interval calculation (95% CI) for uplift percentage between variants
7. Build a results dashboard query that shows sample size, conversion rate, p-value, and confidence intervals for each variant
8. Add guardrail metrics to detect unexpected negative effects (latency, error rate)

Code

import hashlib
import numpy as np
from scipy import stats
from datetime import datetime
import json

class ABTestFramework:
    def __init__(self, db_connection):
        self.db = db_connection
    
    def assign_variant(self, user_id, experiment_id):
        """Deterministically assign user to variant using consistent hashing."""
        hash_input = f"{user_id}_{experiment_id}".encode()
        hash_value = int(hashlib.md5(hash_input).hexdigest(), 16)
        return "variant_b" if (hash_value % 100) < 50 else "variant_a"
    
    def log_event(self, user_id, experiment_id, event_type, value=None):
        """Log conversion/metric event for analysis."""
        variant = self.assign_variant(user_id, experiment_id)
        query = """
        INSERT INTO experiment_events 
        (user_id, experiment_id, variant, event_type, value, timestamp)
        VALUES (%s, %s, %s, %s, %s, NOW())
        """
        self.db.execute(query, (user_id, experiment_id, variant, event_type, value))
    
    def get_experiment_results(self, experiment_id):
        """Compute stats and significance for experiment."""
        query = """
        SELECT variant, COUNT(*

Note: this example was truncated in the source. See the GitHub repo for the latest full version.

Common Pitfalls

• Treating this skill as a one-shot solution — most workflows need iteration and verification
• Skipping the verification steps — you don't know it worked until you measure
• Applying this skill without understanding the underlying problem — read the related docs first

When NOT to Use This Skill

• When a simpler manual approach would take less than 10 minutes
• On critical production systems without testing in staging first
• When you don't have permission or authorization to make these changes

How to Verify It Worked

• Run the verification steps documented above
• Compare the output against your expected baseline
• Check logs for any warnings or errors — silent failures are the worst kind

Production Considerations

• Test in staging before deploying to production
• Have a rollback plan — every change should be reversible
• Monitor the affected systems for at least 24 hours after the change