A/B Testing ML Community Building

A/B Testing ML Community Building คืออะไร

A/B Testing เป็นวิธีทดสอบสมมติฐานโดยแบ่ง users เป็น 2 กลุ่ม (Control vs Treatment) เพื่อเปรียบเทียบผลลัพธ์ Machine Learning ช่วยยกระดับ A/B Testing ด้วย automated analysis, multi-armed bandits และ contextual optimization Community Building คือการสร้างชุมชนรอบ product หรือ brand ที่ users มีส่วนร่วม แชร์ความรู้ และช่วยเหลือกัน การรวมทั้งสามแนวคิดช่วยให้ optimize community features ด้วย data-driven approach ทดสอบ engagement strategies และใช้ ML predict พฤติกรรม community members

A/B Testing Fundamentals

# ab_fundamentals.py — A/B testing basics
import json
import random
import math

class ABTestFundamentals:
    CONCEPTS = {
        "hypothesis": {
            "name": "Hypothesis",
            "description": "สมมติฐาน: ถ้าเปลี่ยน X แล้ว Y จะดีขึ้น",
            "example": "ถ้าเพิ่ม gamification badges จะเพิ่ม community engagement 15%",
        },
        "control_treatment": {
            "name": "Control vs Treatment",
            "description": "Control: กลุ่มควบคุม (ไม่เปลี่ยนแปลง), Treatment: กลุ่มทดลอง (มีการเปลี่ยนแปลง)",
        },
        "sample_size": {
            "name": "Sample Size",
            "description": "จำนวน users ที่ต้องการเพื่อให้ผลลัพธ์มี statistical significance",
        },
        "significance": {
            "name": "Statistical Significance",
            "description": "p-value < 0.05 = ผลลัพธ์ไม่ได้เกิดจากโชค (95% confidence)",
        },
        "effect_size": {
            "name": "Effect Size (MDE)",
            "description": "Minimum Detectable Effect — ขนาดผลลัพธ์ขั้นต่ำที่ต้องการตรวจจับ",
        },
    }

    def calculate_sample_size(self, baseline_rate, mde, alpha=0.05, power=0.80):
        '''Calculate required sample size per group'''
        z_alpha = 1.96  # 95% confidence
        z_beta = 0.84   # 80% power
        
        p1 = baseline_rate
        p2 = baseline_rate * (1 + mde)
        
        pooled_p = (p1 + p2) / 2
        
        n = ((z_alpha * math.sqrt(2 * pooled_p * (1 - pooled_p)) +
              z_beta * math.sqrt(p1 * (1 - p1) + p2 * (1 - p2))) ** 2) / ((p2 - p1) ** 2)
        
        return int(math.ceil(n))

    def show_concepts(self):
        print("=== A/B Testing Concepts ===\n")
        for key, concept in self.CONCEPTS.items():
            print(f"[{concept['name']}]")
            print(f"  {concept['description']}")
            print()

    def demo_sample_size(self):
        print("=== Sample Size Calculator ===")
        scenarios = [
            {"baseline": 0.05, "mde": 0.10, "desc": "5% conversion, detect 10% lift"},
            {"baseline": 0.10, "mde": 0.05, "desc": "10% conversion, detect 5% lift"},
            {"baseline": 0.20, "mde": 0.15, "desc": "20% engagement, detect 15% lift"},
        ]
        for s in scenarios:
            n = self.calculate_sample_size(s["baseline"], s["mde"])
            print(f"  {s['desc']}: {n:,} users/group")

ab = ABTestFundamentals()
ab.show_concepts()
ab.demo_sample_size()

Python A/B Test Framework

# ab_framework.py — A/B testing framework
import json

class ABTestFramework:
    CODE = """
# ab_test.py — A/B testing framework for community features
import numpy as np
from scipy import stats
from datetime import datetime
import json

class ABTest:
    def __init__(self, name, hypothesis, metric, mde=0.05):
        self.name = name
        self.hypothesis = hypothesis
        self.metric = metric
        self.mde = mde
        self.control = []
        self.treatment = []
        self.started_at = datetime.utcnow()
        self.status = "running"
    
    def assign_user(self, user_id):
        '''Assign user to control or treatment (50/50)'''
        # Deterministic assignment based on user_id hash
        bucket = hash(user_id) % 100
        return "treatment" if bucket < 50 else "control"
    
    def record(self, group, value):
        '''Record a metric value'''
        if group == "control":
            self.control.append(value)
        else:
            self.treatment.append(value)
    
    def analyze(self):
        '''Analyze test results'''
        if len(self.control) < 30 or len(self.treatment) < 30:
            return {"status": "insufficient_data", "message": "Need at least 30 samples per group"}
        
        control = np.array(self.control)
        treatment = np.array(self.treatment)
        
        # T-test
        t_stat, p_value = stats.ttest_ind(treatment, control)
        
        # Effect size
        control_mean = control.mean()
        treatment_mean = treatment.mean()
        lift = (treatment_mean - control_mean) / control_mean * 100 if control_mean > 0 else 0
        
        # Cohen's d
        pooled_std = np.sqrt((control.std()**2 + treatment.std()**2) / 2)
        cohens_d = (treatment_mean - control_mean) / pooled_std if pooled_std > 0 else 0
        
        # Confidence interval
        se = np.sqrt(control.var()/len(control) + treatment.var()/len(treatment))
        ci_lower = (treatment_mean - control_mean) - 1.96 * se
        ci_upper = (treatment_mean - control_mean) + 1.96 * se
        
        significant = p_value < 0.05
        
        return {
            "test_name": self.name,
            "metric": self.metric,
            "control_mean": round(control_mean, 4),
            "treatment_mean": round(treatment_mean, 4),
            "lift_percent": round(lift, 2),
            "p_value": round(p_value, 4),
            "significant": significant,
            "cohens_d": round(cohens_d, 3),
            "confidence_interval": [round(ci_lower, 4), round(ci_upper, 4)],
            "recommendation": "Ship treatment" if significant and lift > 0 else 
                            "Keep control" if significant else "Continue testing",
            "control_n": len(control),
            "treatment_n": len(treatment),
        }

# Usage
test = ABTest(
    name="Gamification Badges",
    hypothesis="Adding badges increases weekly active engagement",
    metric="weekly_posts_per_user",
    mde=0.10,
)

# Simulate data
import random
for _ in range(1000):
    test.record("control", random.gauss(3.0, 1.5))
    test.record("treatment", random.gauss(3.4, 1.5))

result = test.analyze()
print(json.dumps(result, indent=2))
"""

    def show_code(self):
        print("=== A/B Test Framework ===")
        print(self.CODE[:600])

framework = ABTestFramework()
framework.show_code()

ML-Enhanced Testing

# ml_testing.py — ML-enhanced A/B testing
import json
import random

class MLEnhancedTesting:
    CODE = """
# ml_ab_test.py — ML-powered A/B testing
import numpy as np
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.model_selection import train_test_split

class MultiArmedBandit:
    '''Thompson Sampling for adaptive A/B testing'''
    def __init__(self, n_variants):
        self.n_variants = n_variants
        self.alpha = np.ones(n_variants)  # Successes + 1
        self.beta = np.ones(n_variants)   # Failures + 1
    
    def select_variant(self):
        '''Select variant using Thompson Sampling'''
        samples = [np.random.beta(a, b) for a, b in zip(self.alpha, self.beta)]
        return np.argmax(samples)
    
    def update(self, variant, reward):
        '''Update beliefs after observing reward'''
        if reward:
            self.alpha[variant] += 1
        else:
            self.beta[variant] += 1
    
    def get_probabilities(self):
        '''Get current win probability for each variant'''
        total = self.alpha + self.beta
        return (self.alpha / total).tolist()

class ContextualBandit:
    '''Use user features to personalize variant selection'''
    def __init__(self, n_variants, n_features):
        self.n_variants = n_variants
        self.models = [GradientBoostingClassifier() for _ in range(n_variants)]
        self.data = {i: {"X": [], "y": []} for i in range(n_variants)}
        self.trained = False
    
    def select_variant(self, user_features):
        '''Select best variant for this user'''
        if not self.trained or np.random.random() < 0.1:  # 10% exploration
            return np.random.randint(self.n_variants)
        
        # Predict reward for each variant
        predictions = []
        for i, model in enumerate(self.models):
            try:
                prob = model.predict_proba([user_features])[0][1]
            except:
                prob = 0.5
            predictions.append(prob)
        
        return np.argmax(predictions)
    
    def update(self, variant, user_features, reward):
        self.data[variant]["X"].append(user_features)
        self.data[variant]["y"].append(int(reward))
    
    def train(self):
        for i, model in enumerate(self.models):
            X = np.array(self.data[i]["X"])
            y = np.array(self.data[i]["y"])
            if len(X) >= 50 and len(set(y)) > 1:
                model.fit(X, y)
        self.trained = True

# bandit = MultiArmedBandit(3)  # 3 variants
# for _ in range(1000):
#     variant = bandit.select_variant()
#     reward = simulate_reward(variant)
#     bandit.update(variant, reward)
"""

    def show_code(self):
        print("=== ML-Enhanced Testing ===")
        print(self.CODE[:600])

    def demo_bandit(self):
        print(f"\n=== Thompson Sampling Results ===")
        variants = ["Control (no badges)", "Simple badges", "Tiered badges + leaderboard"]
        for i, v in enumerate(variants):
            win_prob = random.uniform(0.2, 0.6)
            traffic = random.randint(10, 60)
            print(f"  [{i}] {v}")
            print(f"      Win prob: {win_prob:.1%}, Traffic: {traffic}%")

ml = MLEnhancedTesting()
ml.show_code()
ml.demo_bandit()

Community Features to Test

# community_tests.py — A/B test ideas for community building
import json
import random

class CommunityTestIdeas:
    TESTS = {
        "onboarding": {
            "name": "Onboarding Flow",
            "variants": ["Basic signup", "Guided tour + first action prompt", "Personalized welcome + mentor matching"],
            "metric": "7-day retention rate",
            "expected_lift": "15-30%",
        },
        "gamification": {
            "name": "Gamification System",
            "variants": ["No gamification", "Badges + points", "Leaderboard + streaks + levels"],
            "metric": "Weekly posts per user",
            "expected_lift": "10-25%",
        },
        "notifications": {
            "name": "Notification Strategy",
            "variants": ["Email only (weekly digest)", "Push + email (daily)", "ML-optimized timing + channel"],
            "metric": "Return visit rate",
            "expected_lift": "5-15%",
        },
        "content_feed": {
            "name": "Content Feed Algorithm",
            "variants": ["Chronological", "Popular (most engagement)", "ML-personalized (recommendation)"],
            "metric": "Time spent on feed",
            "expected_lift": "20-40%",
        },
        "social_features": {
            "name": "Social Features",
            "variants": ["Comments only", "Comments + reactions", "Comments + reactions + mentions + threads"],
            "metric": "Interactions per post",
            "expected_lift": "25-50%",
        },
    }

    def show_tests(self):
        print("=== Community A/B Test Ideas ===\n")
        for key, test in self.TESTS.items():
            print(f"[{test['name']}]")
            print(f"  Metric: {test['metric']}")
            print(f"  Expected lift: {test['expected_lift']}")
            for i, v in enumerate(test["variants"]):
                print(f"    Variant {i}: {v}")
            print()

    def dashboard(self):
        print("=== Active Tests Dashboard ===")
        tests = [
            {"name": "Gamification Badges", "status": "Running", "days": random.randint(5, 20), "lift": f"+{random.uniform(5, 25):.1f}%"},
            {"name": "Personalized Feed", "status": "Significant", "days": random.randint(14, 30), "lift": f"+{random.uniform(15, 40):.1f}%"},
            {"name": "Welcome Email", "status": "Running", "days": random.randint(3, 10), "lift": f"+{random.uniform(-5, 15):.1f}%"},
        ]
        for t in tests:
            print(f"  [{t['status']:<12}] {t['name']:<25} Day {t['days']:>2}  Lift: {t['lift']}")

ideas = CommunityTestIdeas()
ideas.show_tests()
ideas.dashboard()

Reporting & Visualization

# reporting.py — A/B test reporting
import json
import random

class ABTestReporting:
    CODE = """
# ab_report.py — Generate A/B test reports
import json
import matplotlib.pyplot as plt
import numpy as np

class ABTestReporter:
    def __init__(self, test_results):
        self.results = test_results
    
    def summary_report(self):
        r = self.results
        print(f"=== A/B Test Report: {r['test_name']} ===")
        print(f"  Metric: {r['metric']}")
        print(f"  Control: {r['control_mean']:.4f} (n={r['control_n']})")
        print(f"  Treatment: {r['treatment_mean']:.4f} (n={r['treatment_n']})")
        print(f"  Lift: {r['lift_percent']:+.2f}%")
        print(f"  P-value: {r['p_value']:.4f}")
        print(f"  Significant: {r['significant']}")
        print(f"  CI: [{r['confidence_interval'][0]:.4f}, {r['confidence_interval'][1]:.4f}]")
        print(f"  Recommendation: {r['recommendation']}")
    
    def plot_distributions(self, control_data, treatment_data):
        fig, axes = plt.subplots(1, 2, figsize=(12, 5))
        
        axes[0].hist(control_data, bins=30, alpha=0.7, label='Control', color='blue')
        axes[0].hist(treatment_data, bins=30, alpha=0.7, label='Treatment', color='green')
        axes[0].set_title('Distribution Comparison')
        axes[0].legend()
        
        # Cumulative lift over time
        n = min(len(control_data), len(treatment_data))
        cumulative_lift = []
        for i in range(10, n, 10):
            c_mean = np.mean(control_data[:i])
            t_mean = np.mean(treatment_data[:i])
            lift = (t_mean - c_mean) / c_mean * 100
            cumulative_lift.append(lift)
        
        axes[1].plot(range(len(cumulative_lift)), cumulative_lift)
        axes[1].axhline(y=0, color='r', linestyle='--')
        axes[1].set_title('Cumulative Lift Over Time')
        
        plt.tight_layout()
        plt.savefig('ab_test_report.png')

# reporter = ABTestReporter(test_result)
# reporter.summary_report()
"""

    def show_code(self):
        print("=== Reporting ===")
        print(self.CODE[:600])

    def sample_report(self):
        print(f"\n=== Sample Test Result ===")
        print(f"  Test: Gamification Badges for Community")
        print(f"  Duration: {random.randint(14, 28)} days")
        print(f"  Control (no badges): {random.uniform(2.5, 3.5):.2f} posts/user/week")
        print(f"  Treatment (badges): {random.uniform(3.2, 4.5):.2f} posts/user/week")
        print(f"  Lift: +{random.uniform(10, 30):.1f}%")
        print(f"  P-value: {random.uniform(0.001, 0.04):.4f}")
        print(f"  Recommendation: Ship treatment ✓")

report = ABTestReporting()
report.show_code()
report.sample_report()

FAQ - คำถามที่พบบ่อย

Q: A/B test ต้องรันนานแค่ไหน?

A: ขึ้นกับ traffic + MDE: Traffic สูง (10K+ users/day): 1-2 สัปดาห์ Traffic ปานกลาง (1K-10K/day): 2-4 สัปดาห์ Traffic ต่ำ (< 1K/day): 4-8 สัปดาห์ หรือใช้ Bayesian approach กฎ: อย่าหยุด test เร็วเกินไป — รอให้ถึง sample size ที่คำนวณไว้ ครอบคลุมอย่างน้อย 1 business cycle (7 วัน)

Q: Multi-armed bandit ดีกว่า A/B test ไหม?

A: ต่างกัน: A/B test: ได้ statistical rigor สูง, ต้องรอจนจบ test ถึงได้ผลลัพธ์ Bandit: adaptive allocation — ส่ง traffic ไปทาง winning variant มากขึ้นเรื่อยๆ ลด opportunity cost ใช้ A/B test: เมื่อต้องการ confident decision (product launch, major changes) ใช้ Bandit: เมื่อ exploration/exploitation tradeoff สำคัญ (personalization, content ranking)

Q: Community building metrics วัดอะไรบ้าง?

A: Engagement: posts per user, comments, reactions, time spent Retention: DAU/MAU ratio, 7-day retention, 30-day retention Growth: new members per week, invite rate, organic signups Quality: helpful answer rate, content quality score, NPS Health: churn rate, inactive user %, toxic content rate เลือก 1-2 primary metrics ต่อ A/B test — อย่าวัดหลาย metrics พร้อมกัน (multiple comparison problem)

Q: ML model สำหรับ community ใช้อะไร?

A: Churn prediction: predict ว่า user ไหนจะหายไป → intervene ก่อน Content recommendation: แนะนำ posts ที่ user น่าจะสนใจ Notification optimization: เลือกเวลาและ channel ที่ดีที่สุดต่อ user Toxicity detection: ตรวจจับ toxic content อัตโนมัติ User segmentation: แบ่งกลุ่ม users สำหรับ personalized experiments