A/B测试核心原理与实验设计完全指南

I. A/B测试基础概念与价值

1.1 什么是A/B测试

A/B测试，也称为拆分测试，是一种比较两个或多个版本（A版本和B版本）以确定哪个版本在预定义指标上表现更好的统计方法。在数字产品环境中，这通常意味着将用户流量随机分配到不同的体验中，然后通过统计分析来确定哪个版本更能实现业务目标。

1.2 A/B测试的商业价值

1.3 A/B测试的基本流程

class ABTestFramework:
    """A/B测试框架基础类"""
    
    def __init__(self):
        self.experiments = {}
        self.results = {}
    
    def define_experiment(self, experiment_id, hypothesis, metrics, variants):
        """定义实验"""
        experiment = {
            'id': experiment_id,
            'hypothesis': hypothesis,
            'metrics': metrics,  # 主要指标和辅助指标
            'variants': variants,  # 变体配置
            'status': 'draft',
            'start_time': None,
            'end_time': None
        }
        self.experiments[experiment_id] = experiment
        return experiment
    
    def calculate_sample_size(self, baseline_rate, mde, alpha=0.05, power=0.8):
        """
        计算所需样本量
        baseline_rate: 基线转化率
        mde: 最小可检测效应 (Minimum Detectable Effect)
        alpha: 显著性水平
        power: 统计功效
        """
        from scipy import stats
        import math
        
        z_alpha = stats.norm.ppf(1 - alpha/2)
        z_beta = stats.norm.ppf(power)
        
        pooled_prob = (baseline_rate + baseline_rate * (1 + mde)) / 2
        se_pooled = math.sqrt(pooled_prob * (1 - pooled_prob) * 2)
        se_alternative = math.sqrt(
            baseline_rate * (1 - baseline_rate) + 
            baseline_rate * (1 + mde) * (1 - baseline_rate * (1 + mde))
        )
        
        effect_size = baseline_rate * mde
        n = ((z_alpha * se_pooled + z_beta * se_alternative) / effect_size) ** 2
        
        return math.ceil(n)

# 示例：计算检测转化率从5%提升到5.5%所需的样本量
framework = ABTestFramework()
sample_size = framework.calculate_sample_size(
    baseline_rate=0.05, 
    mde=0.10,  # 10%的相对提升
    alpha=0.05, 
    power=0.8
)
print(f"每组需要的样本量: {sample_size}")
print(f"总样本量: {sample_size * 2}")

1.4 A/B测试的演进历程

A/B测试已经从简单的网页版本测试发展为复杂的多变量实验系统。现代A/B测试平台支持：

• 多变量测试(MVT)：同时测试多个因素的不同组合
• 渐进式展开：逐步增加新版本的流量比例
• 定向测试：针对特定用户群体进行测试
• 长期影响评估：跟踪实验的长期效果

II. 统计学基础与核心原理

2.1 假设检验框架

A/B测试的核心是统计假设检验。我们建立两个相互竞争的假设：

• 零假设(H₀)：实验组和对照组没有显著差异
• 备择假设(H₁)：实验组和对照组存在显著差异

2.2 关键统计概念

2.3 中心极限定理的应用

中心极限定理告诉我们，无论总体分布如何，样本均值的抽样分布都近似正态分布。这是A/B测试中许多统计检验的基础。

import numpy as np
import matplotlib.pyplot as plt
from scipy import stats
import seaborn as sns

class StatisticalFoundation:
    """统计学基础演示类"""
    
    def demonstrate_clt(self, population_dist, sample_sizes, n_samples=1000):
        """
        演示中心极限定理
        population_dist: 总体分布函数
        sample_sizes: 不同的样本量列表
        n_samples: 生成的样本数量
        """
        plt.figure(figsize=(15, 10))
        
        for i, sample_size in enumerate(sample_sizes, 1):
            sample_means = []
            for _ in range(n_samples):
                sample = population_dist(sample_size)
                sample_means.append(np.mean(sample))
            
            plt.subplot(2, 2, i)
            sns.histplot(sample_means, kde=True)
            plt.title(f'样本量 = {sample_size}, 样本均值标准差 = {np.std(sample_means):.4f}')
            plt.xlabel('样本均值')
            plt.ylabel('频数')
        
        plt.tight_layout()
        plt.show()
    
    def calculate_p_value(self, control_data, treatment_data, test_type='proportion'):
        """
        计算p值
        """
        if test_type == 'proportion':
            # 比例检验
            from statsmodels.stats.proportion import proportions_ztest
            
            count = [np.sum(control_data), np.sum(treatment_data)]
            nobs = [len(control_data), len(treatment_data)]
            z_stat, p_value = proportions_ztest(count, nobs)
            return z_stat, p_value
        
        elif test_type == 'means':
            # 均值t检验
            t_stat, p_value = stats.ttest_ind(treatment_data, control_data)
            return t_stat, p_value
        
        else:
            raise ValueError("不支持的检验类型")
    
    def calculate_confidence_interval(self, data, confidence=0.95):
        """计算置信区间"""
        mean = np.mean(data)
        sem = stats.sem(data)  # 标准误
        ci = stats.t.interval(confidence, len(data)-1, loc=mean, scale=sem)
        return ci

# 演示中心极限定理
stats_demo = StatisticalFoundation()

# 生成偏态分布数据
def skewed_distribution(n):
    return np.random.exponential(2, n) + 5  # 指数分布+偏移

sample_sizes = [10, 30, 50, 100]
stats_demo.demonstrate_clt(skewed_distribution, sample_sizes)

# 模拟A/B测试数据并计算p值
np.random.seed(42)
control_conversions = np.random.binomial(1, 0.10, 1000)  # 10%转化率
treatment_conversions = np.random.binomial(1, 0.12, 1000)  # 12%转化率

z_stat, p_value = stats_demo.calculate_p_value(
    control_conversions, treatment_conversions, 'proportion'
)
print(f"Z统计量: {z_stat:.4f}")
print(f"P值: {p_value:.4f}")

# 计算置信区间
control_ci = stats_demo.calculate_confidence_interval(control_conversions)
treatment_ci = stats_demo.calculate_confidence_interval(treatment_conversions)
print(f"对照组转化率置信区间: {control_ci}")
print(f"实验组转化率置信区间: {treatment_ci}")

2.4 贝叶斯A/B测试

除了传统的频率主义方法，贝叶斯统计也为A/B测试提供了强大的工具。

class BayesianABTest:
    """贝叶斯A/B测试实现"""
    
    def __init__(self, alpha_prior=1, beta_prior=1):
        # 使用Beta分布作为先验分布
        self.alpha_prior = alpha_prior
        self.beta_prior = beta_prior
        
        # 后验分布参数
        self.alpha_posterior_a = alpha_prior
        self.beta_posterior_a = beta_prior
        self.alpha_posterior_b = alpha_prior
        self.beta_posterior_b = beta_prior
    
    def update_posterior(self, variant, successes, failures):
        """更新后验分布"""
        if variant == 'A':
            self.alpha_posterior_a += successes
            self.beta_posterior_a += failures
        elif variant == 'B':
            self.alpha_posterior_b += successes
            self.beta_posterior_b += failures
        else:
            raise ValueError("变体必须是'A'或'B'")
    
    def probability_b_beats_a(self, n_simulations=100000):
        """计算B优于A的概率"""
        from scipy import stats
        
        # 从后验分布中采样
        samples_a = stats.beta.rvs(
            self.alpha_posterior_a, 
            self.beta_posterior_a, 
            size=n_simulations
        )
        samples_b = stats.beta.rvs(
            self.alpha_posterior_b, 
            self.beta_posterior_b, 
            size=n_simulations
        )
        
        # 计算B > A的比例
        prob = np.mean(samples_b > samples_a)
        return prob
    
    def expected_loss(self, variant, n_simulations=100000):
        """计算预期损失"""
        from scipy import stats
        
        if variant == 'A':
            alpha_self = self.alpha_posterior_a
            beta_self = self.beta_posterior_a
            alpha_other = self.alpha_posterior_b
            beta_other = self.beta_posterior_b
        else:
            alpha_self = self.alpha_posterior_b
            beta_self = self.beta_posterior_b
            alpha_other = self.alpha_posterior_a
            beta_other = self.beta_posterior_a
        
        samples_self = stats.beta.rvs(alpha_self, beta_self, size=n_simulations)
        samples_other = stats.beta.rvs(alpha_other, beta_other, size=n_simulations)
        
        loss = np.maximum(samples_other - samples_self, 0)
        return np.mean(loss)
    
    def plot_posterior_distributions(self):
        """绘制后验分布"""
        import matplotlib.pyplot as plt
        
        x = np.linspace(0, 1, 1000)
        pdf_a = stats.beta.pdf(x, self.alpha_posterior_a, self.beta_posterior_a)
        pdf_b = stats.beta.pdf(x, self.alpha_posterior_b, self.beta_posterior_b)
        
        plt.figure(figsize=(10, 6))
        plt.plot(x, pdf_a, label='变体A后验分布', linewidth=2)
        plt.plot(x, pdf_b, label='变体B后验分布', linewidth=2)
        plt.fill_between(x, pdf_a, alpha=0.3)
        plt.fill_between(x, pdf_b, alpha=0.3)
        plt.xlabel('转化率')
        plt.ylabel('概率密度')
        plt.title('A/B测试后验分布')
        plt.legend()
        plt.grid(True, alpha=0.3)
        plt.show()

# 使用贝叶斯A/B测试
bayesian_test = BayesianABTest()

# 模拟数据
np.random.seed(42)
successes_a = np.random.binomial(1000, 0.10)  # A组成功次数
failures_a = 1000 - successes_a               # A组失败次数

successes_b = np.random.binomial(1000, 0.12)  # B组成功次数
failures_b = 1000 - successes_b               # B组失败次数

# 更新后验分布
bayesian_test.update_posterior('A', successes_a, failures_a)
bayesian_test.update_posterior('B', successes_b, failures_b)

# 计算B优于A的概率
prob_b_beats_a = bayesian_test.probability_b_beats_a()
print(f"变体B优于变体A的概率: {prob_b_beats_a:.4f}")

# 计算预期损失
loss_a = bayesian_test.expected_loss('A')
loss_b = bayesian_test.expected_loss('B')
print(f"选择A的预期损失: {loss_a:.4f}")
print(f"选择B的预期损失: {loss_b:.4f}")

# 绘制后验分布
bayesian_test.plot_posterior_distributions()

III. 实验设计最佳实践

3.1 明确的实验假设

成功的A/B测试始于清晰的假设。一个好的假设应该具体、可测试，并与业务目标直接相关。

3.2 指标选择与定义

选择正确的指标是A/B测试成功的关键。指标应该分为不同层级：

3.3 样本量计算与实验时长

正确的样本量确保实验有足够的统计功效，而合理的实验时长则能避免各种偏见。

class ExperimentDesign:
    """实验设计工具类"""
    
    def __init__(self):
        self.metric_definitions = {}
    
    def define_metric(self, name, metric_type, calculation_func, 
                     direction='increase', target_value=None):
        """定义指标"""
        metric = {
            'name': name,
            'type': metric_type,  # 'primary', 'secondary', 'guardrail'
            'calculation': calculation_func,
            'direction': direction,  # 'increase' or 'decrease'
            'target_value': target_value
        }
        self.metric_definitions[name] = metric
        return metric
    
    def calculate_experiment_duration(self, daily_traffic, sample_size_per_variant, 
                                    variants=2, traffic_allocation=1.0):
        """
        计算实验所需时长
        daily_traffic: 日流量
        sample_size_per_variant: 每个变体所需样本量
        variants: 变体数量
        traffic_allocation: 流量分配比例
        """
        total_sample_size = sample_size_per_variant * variants
        available_daily_traffic = daily_traffic * traffic_allocation
        duration_days = total_sample_size / available_daily_traffic
        return max(1, math.ceil(duration_days))
    
    def check_seasonality(self, historical_data, test_duration_weeks):
        """检查季节性影响"""
        # 确保实验覆盖完整的业务周期
        if test_duration_weeks < 2:
            print("警告: 实验时长可能不足以捕捉周度模式")
        if test_duration_weeks < 4:
            print("警告: 实验时长可能不足以捕捉月度模式")
        
        # 分析历史数据的周期性
        # 这里可以添加时间序列分析代码
        
        return True
    
    def power_analysis(self, baseline_rate, effect_sizes, alpha=0.05, power=0.8):
        """功效分析 - 计算不同效应大小所需的样本量"""
        sample_sizes = {}
        for effect_size in effect_sizes:
            mde = effect_size  # 相对效应大小
            n = self.calculate_sample_size(baseline_rate, mde, alpha, power)
            sample_sizes[effect_size] = n
        
        return sample_sizes
    
    def calculate_sample_size(self, baseline_rate, mde, alpha=0.05, power=0.8):
        """计算比例检验的样本量"""
        from scipy import stats
        import math
        
        z_alpha = stats.norm.ppf(1 - alpha/2)
        z_beta = stats.norm.ppf(power)
        
        p1 = baseline_rate
        p2 = baseline_rate * (1 + mde)
        p_pool = (p1 + p2) / 2
        
        numerator = (z_alpha * math.sqrt(2 * p_pool * (1 - p_pool)) + 
                    z_beta * math.sqrt(p1 * (1 - p1) + p2 * (1 - p2)))
        denominator = abs(p2 - p1)
        
        n = (numerator / denominator) ** 2
        return math.ceil(n)

# 实验设计示例
design = ExperimentDesign()

# 定义指标
design.define_metric(
    name='conversion_rate',
    metric_type='primary',
    calculation_func=lambda data: np.mean(data),
    direction='increase',
    target_value=0.15
)

design.define_metric(
    name='average_order_value',
    metric_type='secondary', 
    calculation_func=lambda data: np.mean([d['value'] for d in data]),
    direction='increase'
)

# 计算实验参数
baseline_conversion = 0.10
daily_users = 5000
effect_sizes = [0.05, 0.10, 0.15, 0.20]  # 5%, 10%, 15%, 20%的相对提升

# 功效分析
power_results = design.power_analysis(baseline_conversion, effect_sizes)
print("不同效应大小所需的样本量:")
for effect_size, sample_size in power_results.items():
    print(f"{effect_size:.0%} 提升: {sample_size} 用户每组")

# 选择10%提升作为MDE
mde = 0.10
required_sample_size = power_results[mde]
duration = design.calculate_experiment_duration(
    daily_traffic=daily_users,
    sample_size_per_variant=required_sample_size,
    variants=2,
    traffic_allocation=0.5  # 50%流量用于实验
)

print(f"\n实验参数总结:")
print(f"基线转化率: {baseline_conversion:.1%}")
print(f"目标提升: {mde:.0%}")
print(f"每组所需样本量: {required_sample_size}")
print(f"预计实验时长: {duration} 天")
print(f"总样本量: {required_sample_size * 2}")
print(f"所需总流量: {daily_users * duration}")

# 检查季节性
design.check_seasonality(None, duration/7)

3.4 随机化与偏差控制

确保实验组和对照组的可比性是A/B测试有效性的基础。

class RandomizationValidator:
    """随机化验证工具"""
    
    def __init__(self):
        self.covariates = {}
    
    def add_covariate(self, name, data_type, importance='medium'):
        """添加协变量"""
        self.covariates[name] = {
            'data_type': data_type,  # 'continuous', 'categorical', 'binary'
            'importance': importance  # 'high', 'medium', 'low'
        }
    
    def check_balance(self, group_assignments, user_data):
        """检查组间平衡性"""
        results = {}
        
        for covariate, info in self.covariates.items():
            if covariate not in user_data.columns:
                continue
                
            if info['data_type'] == 'continuous':
                result = self._check_continuous_balance(group_assignments, user_data[covariate])
            elif info['data_type'] in ['categorical', 'binary']:
                result = self._check_categorical_balance(group_assignments, user_data[covariate])
            else:
                continue
            
            results[covariate] = result
        
        return results
    
    def _check_continuous_balance(self, groups, values):
        """检查连续变量的平衡性"""
        from scipy import stats
        
        unique_groups = np.unique(groups)
        if len(unique_groups) != 2:
            raise ValueError("只支持两组比较")
        
        group_a_values = values[groups == unique_groups[0]]
        group_b_values = values[groups == unique_groups[1]]
        
        # t检验
        t_stat, p_value = stats.ttest_ind(group_a_values, group_b_values)
        
        # 标准化均值差异
        mean_a, mean_b = np.mean(group_a_values), np.mean(group_b_values)
        std_pooled = np.sqrt((np.var(group_a_values) + np.var(group_b_values)) / 2)
        smd = abs(mean_a - mean_b) / std_pooled
        
        return {
            'test_type': 't_test',
            'p_value': p_value,
            'mean_difference': mean_a - mean_b,
            'standardized_mean_difference': smd,
            'balanced': p_value > 0.05 and smd < 0.1
        }
    
    def _check_categorical_balance(self, groups, values):
        """检查分类变量的平衡性"""
        from scipy.stats import chi2_contingency
        
        contingency_table = pd.crosstab(groups, values)
        chi2, p_value, dof, expected = chi2_contingency(contingency_table)
        
        return {
            'test_type': 'chi_square',
            'p_value': p_value, 
            'chi2_statistic': chi2,
            'balanced': p_value > 0.05
        }
    
    def generate_balance_report(self, balance_results):
        """生成平衡性报告"""
        report = ["随机化平衡性检查报告", "=" * 50]
        
        balanced_count = 0
        total_count = len(balance_results)
        
        for covariate, result in balance_results.items():
            status = "平衡" if result['balanced'] else "不平衡"
            report.append(f"\n{covariate}: {status}")
            report.append(f"  检验类型: {result['test_type']}")
            report.append(f"  P值: {result['p_value']:.4f}")
            
            if result['test_type'] == 't_test':
                report.append(f"  标准化均值差异: {result['standardized_mean_difference']:.4f}")
            elif result['test_type'] == 'chi_square':
                report.append(f"  卡方统计量: {result['chi2_statistic']:.4f}")
            
            if result['balanced']:
                balanced_count += 1
        
        report.append(f"\n总结: {balanced_count}/{total_count} 个变量平衡")
        
        if balanced_count == total_count:
            report.append("✅ 随机化成功: 所有变量在组间平衡")
        else:
            report.append("⚠️ 警告: 部分变量在组间不平衡，考虑重新随机化或统计校正")
        
        return "\n".join(report)

# 随机化验证示例
validator = RandomizationValidator()

# 添加需要检查的协变量
validator.add_covariate('age', 'continuous', 'high')
validator.add_covariate('gender', 'categorical', 'medium') 
validator.add_covariate('new_user', 'binary', 'high')
validator.add_covariate('previous_purchases', 'continuous', 'medium')

# 生成模拟用户数据
np.random.seed(42)
n_users = 2000

user_data = pd.DataFrame({
    'user_id': range(n_users),
    'age': np.random.normal(35, 10, n_users),
    'gender': np.random.choice(['Male', 'Female', 'Other'], n_users, p=[0.48, 0.50, 0.02]),
    'new_user': np.random.binomial(1, 0.3, n_users),
    'previous_purchases': np.random.poisson(3, n_users)
})

# 模拟随机分配（这里使用简单的随机分配）
group_assignments = np.random.choice(['control', 'treatment'], n_users)

# 检查平衡性
balance_results = validator.check_balance(group_assignments, user_data)
report = validator.generate_balance_report(balance_results)
print(report)

IV. 实施与执行策略

4.1 流量分配与用户一致性

确保用户在整个实验期间被一致地分配到同一个组别是A/B测试可靠性的关键。

4.2 实验基础设施

健壮的A/B测试系统需要可靠的基础设施支持。

import hashlib
import json
from datetime import datetime, timedelta
import redis
from typing import Dict, List, Optional

class ExperimentAssignmentSystem:
    """实验分配系统"""
    
    def __init__(self, redis_client, salt="experiment_salt_2024"):
        self.redis = redis_client
        self.salt = salt
        self.assignment_cache = {}
    
    def assign_user_to_variant(self, user_id: str, experiment_id: str, 
                             variants: List[str], weights: List[float]) -> str:
        """
        将用户分配到实验变体
        使用确定性哈希确保一致性
        """
        # 检查缓存
        cache_key = f"{user_id}_{experiment_id}"
        if cache_key in self.assignment_cache:
            return self.assignment_cache[cache_key]
        
        # 检查持久化存储
        stored_assignment = self._get_stored_assignment(user_id, experiment_id)
        if stored_assignment:
            self.assignment_cache[cache_key] = stored_assignment
            return stored_assignment
        
        # 新分配
        assignment = self._calculate_assignment(user_id, experiment_id, variants, weights)
        
        # 存储分配结果
        self._store_assignment(user_id, experiment_id, assignment)
        self.assignment_cache[cache_key] = assignment
        
        return assignment
    
    def _calculate_assignment(self, user_id: str, experiment_id: str,
                            variants: List[str], weights: List[float]) -> str:
        """计算用户分配"""
        # 归一化权重
        total_weight = sum(weights)
        normalized_weights = [w / total_weight for w in weights]
        
        # 计算哈希值
        hash_input = f"{user_id}_{experiment_id}_{self.salt}".encode('utf-8')
        hash_value = hashlib.md5(hash_input).hexdigest()
        hash_int = int(hash_value[:8], 16) % 10000
        
        # 根据权重分配
        cumulative = 0
        for variant, weight in zip(variants, normalized_weights):
            cumulative += weight
            if hash_int < cumulative * 10000:
                return variant
        
        return variants[0]  # 兜底
    
    def _get_stored_assignment(self, user_id: str, experiment_id: str) -> Optional[str]:
        """从存储中获取分配结果"""
        key = f"assignment:{experiment_id}:{user_id}"
        assignment = self.redis.get(key)
        return assignment.decode() if assignment else None
    
    def _store_assignment(self, user_id: str, experiment_id: str, assignment: str):
        """存储分配结果"""
        key = f"assignment:{experiment_id}:{user_id}"
        # 存储30天
        self.redis.setex(key, timedelta(days=30), assignment)
    
    def get_user_assignments(self, user_id: str) -> Dict[str, str]:
        """获取用户的所有实验分配"""
        # 使用模式匹配获取所有相关键
        pattern = f"assignment:*:{user_id}"
        keys = self.redis.keys(pattern)
        
        assignments = {}
        for key in keys:
            experiment_id = key.decode().split(':')[1]
            assignment = self.redis.get(key)
            if assignment:
                assignments[experiment_id] = assignment.decode()
        
        return assignments

class EventTrackingSystem:
    """事件跟踪系统"""
    
    def __init__(self, redis_client, kafka_client=None):
        self.redis = redis_client
        self.kafka_client = kafka_client
    
    def track_event(self, user_id: str, experiment_id: str, variant: str,
                   event_type: str, event_properties: Dict, timestamp: datetime = None):
        """跟踪事件"""
        if timestamp is None:
            timestamp = datetime.now()
        
        event = {
            'user_id': user_id,
            'experiment_id': experiment_id,
            'variant': variant,
            'event_type': event_type,
            'properties': event_properties,
            'timestamp': timestamp.isoformat(),
            'version': '1.0'
        }
        
        # 存储到Redis（用于实时分析）
        event_key = f"event:{experiment_id}:{timestamp.strftime('%Y%m%d')}"
        self.redis.lpush(event_key, json.dumps(event))
        
        # 发送到Kafka（用于离线分析）
        if self.kafka_client:
            self.kafka_client.send('ab_test_events', event)
        
        # 更新实时计数器
        self._update_realtime_counters(experiment_id, variant, event_type)
    
    def _update_realtime_counters(self, experiment_id: str, variant: str, event_type: str):
        """更新实时计数器"""
        # 日计数器
        day_key = f"counts:{experiment_id}:{variant}:{event_type}:{datetime.now().strftime('%Y%m%d')}"
        self.redis.incr(day_key)
        
        # 总计数器
        total_key = f"counts:{experiment_id}:{variant}:{event_type}:total"
        self.redis.incr(total_key)
    
    def get_event_counts(self, experiment_id: str, start_date: str, end_date: str) -> Dict:
        """获取事件计数"""
        counts = {}
        current_date = datetime.strptime(start_date, '%Y-%m-%d')
        end_date = datetime.strptime(end_date, '%Y-%m-%d')
        
        while current_date <= end_date:
            date_str = current_date.strftime('%Y%m%d')
            for variant in ['control', 'treatment']:
                for event_type in ['pageview', 'conversion']:
                    key = f"counts:{experiment_id}:{variant}:{event_type}:{date_str}"
                    count = self.redis.get(key)
                    if count:
                        if variant not in counts:
                            counts[variant] = {}
                        if event_type not in counts[variant]:
                            counts[variant][event_type] = 0
                        counts[variant][event_type] += int(count)
            
            current_date += timedelta(days=1)
        
        return counts

# 初始化系统
redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
assignment_system = ExperimentAssignmentSystem(redis_client)
tracking_system = EventTrackingSystem(redis_client)

# 实验配置
experiment_config = {
    'button_color_test': {
        'variants': ['control', 'red_button', 'blue_button'],
        'weights': [0.33, 0.33, 0.34],
        'metrics': ['conversion_rate', 'click_through_rate']
    }
}

# 模拟用户交互
def simulate_user_journey(user_id, experiment_id):
    """模拟用户旅程"""
    # 分配变体
    config = experiment_config[experiment_id]
    variant = assignment_system.assign_user_to_variant(
        user_id, experiment_id, config['variants'], config['weights']
    )
    
    # 跟踪页面浏览
    tracking_system.track_event(
        user_id, experiment_id, variant,
        'pageview', {'page': 'homepage'}
    )
    
    # 模拟转化（基于变体有不同的转化率）
    conversion_rates = {
        'control': 0.10,
        'red_button': 0.12, 
        'blue_button': 0.11
    }
    
    converted = np.random.random() < conversion_rates[variant]
    if converted:
        tracking_system.track_event(
            user_id, experiment_id, variant,
            'conversion', {'value': 100, 'product': 'premium'}
        )
    
    return variant, converted

# 运行模拟
np.random.seed(42)
n_users = 1000
conversions_by_variant = {}

for user_id in range(n_users):
    variant, converted = simulate_user_journey(f"user_{user_id}", "button_color_test")
    
    if variant not in conversions_by_variant:
        conversions_by_variant[variant] = {'conversions': 0, 'users': 0}
    
    conversions_by_variant[variant]['users'] += 1
    if converted:
        conversions_by_variant[variant]['conversions'] += 1

# 输出结果
print("实验结果:")
for variant, data in conversions_by_variant.items():
    rate = data['conversions'] / data['users']
    print(f"{variant}: {data['conversions']}/{data['users']} = {rate:.3f}")

4.3 质量监控与警报

实时监控实验质量，及时发现问题。

class ExperimentMonitor:
    """实验监控系统"""
    
    def __init__(self, redis_client, alert_thresholds=None):
        self.redis = redis_client
        
        # 默认警报阈值
        self.alert_thresholds = alert_thresholds or {
            'traffic_imbalance': 0.15,      # 流量偏差超过15%
            'conversion_drop': 0.10,        # 转化率下降超过10%
            'sample_ratio_mismatch': 0.05,  # 样本比例偏差超过5%
            'confidence_level': 0.95        # 置信水平
        }
    
    def check_traffic_quality(self, experiment_id: str, expected_weights: Dict[str, float]) -> Dict:
        """检查流量质量"""
        total_traffic = 0
        actual_traffic = {}
        
        # 获取实际流量
        for variant in expected_weights.keys():
            key = f"counts:{experiment_id}:{variant}:pageview:total"
            count = self.redis.get(key)
            actual_traffic[variant] = int(count) if count else 0
            total_traffic += actual_traffic[variant]
        
        if total_traffic == 0:
            return {'status': 'no_data', 'message': '暂无流量数据'}
        
        # 计算偏差
        imbalances = {}
        for variant, expected_weight in expected_weights.items():
            expected_traffic = total_traffic * expected_weight
            actual_traffic_variant = actual_traffic.get(variant, 0)
            deviation = abs(actual_traffic_variant - expected_traffic) / expected_traffic
            imbalances[variant] = {
                'expected': expected_traffic,
                'actual': actual_traffic_variant,
                'deviation': deviation,
                'within_threshold': deviation <= self.alert_thresholds['traffic_imbalance']
            }
        
        # 总体评估
        max_deviation = max(imbalances[variant]['deviation'] for variant in imbalances)
        overall_balanced = max_deviation <= self.alert_thresholds['traffic_imbalance']
        
        return {
            'status': 'balanced' if overall_balanced else 'imbalanced',
            'total_traffic': total_traffic,
            'imbalances': imbalances,
            'max_deviation': max_deviation
        }
    
    def check_conversion_safety(self, experiment_id: str, baseline_rate: float) -> Dict:
        """检查转化率安全性"""
        variants_data = {}
        
        for variant in ['control', 'treatment']:  # 假设只有这两个变体
            conversion_key = f"counts:{experiment_id}:{variant}:conversion:total"
            pageview_key = f"counts:{experiment_id}:{variant}:pageview:total"
            
            conversions = int(self.redis.get(conversion_key) or 0)
            pageviews = int(self.redis.get(pageview_key) or 1)  # 避免除零
            
            rate = conversions / pageviews
            variants_data[variant] = {
                'conversions': conversions,
                'pageviews': pageviews,
                'rate': rate
            }
        
        # 检查转化率下降
        control_rate = variants_data['control']['rate']
        treatment_rate = variants_data['treatment']['rate']
        relative_drop = (control_rate - treatment_rate) / control_rate
        
        safety_issue = relative_drop > self.alert_thresholds['conversion_drop']
        
        return {
            'safety_issue': safety_issue,
            'relative_drop': relative_drop,
            'control_rate': control_rate,
            'treatment_rate': treatment_rate,
            'data': variants_data
        }
    
    def generate_monitoring_report(self, experiment_id: str, 
                                 expected_weights: Dict[str, float],
                                 baseline_conversion_rate: float) -> str:
        """生成监控报告"""
        traffic_report = self.check_traffic_quality(experiment_id, expected_weights)
        safety_report = self.check_conversion_safety(experiment_id, baseline_conversion_rate)
        
        report_lines = [
            f"实验监控报告: {experiment_id}",
            "=" * 50,
            f"生成时间: {datetime.now().isoformat()}",
            ""
        ]
        
        # 流量质量部分
        report_lines.append("流量质量检查:")
        report_lines.append(f"  总流量: {traffic_report['total_traffic']}")
        report_lines.append(f"  最大偏差: {traffic_report['max_deviation']:.3f}")
        report_lines.append(f"  状态: {traffic_report['status']}")
        
        for variant, data in traffic_report['imbalances'].items():
            status = "✅" if data['within_threshold'] else "❌"
            report_lines.append(
                f"  {variant}: {data['actual']} (期望: {data['expected']:.1f}) "
                f"- 偏差: {data['deviation']:.3f} {status}"
            )
        
        # 安全性检查部分
        report_lines.append("\n转化率安全性检查:")
        report_lines.append(f"  基线转化率: {baseline_conversion_rate:.3f}")
        report_lines.append(f"  对照组转化率: {safety_report['control_rate']:.3f}")
        report_lines.append(f"  实验组转化率: {safety_report['treatment_rate']:.3f}")
        report_lines.append(f"  相对下降: {safety_report['relative_drop']:.3f}")
        
        if safety_report['safety_issue']:
            report_lines.append("  ❌ 警报: 检测到显著转化率下降!")
        else:
            report_lines.append("  ✅ 转化率在安全范围内")
        
        # 建议
        report_lines.append("\n建议:")
        if traffic_report['status'] == 'imbalanced':
            report_lines.append("  • 检查流量分配系统")
        if safety_report['safety_issue']:
            report_lines.append("  • 考虑停止实验或减少流量")
        if traffic_report['status'] == 'balanced' and not safety_report['safety_issue']:
            report_lines.append("  • 实验运行正常，继续监控")
        
        return "\n".join(report_lines)

# 使用监控系统
monitor = ExperimentMonitor(redis_client)

# 生成监控报告
expected_weights = {'control': 0.5, 'treatment': 0.5}
baseline_rate = 0.10

report = monitor.generate_monitoring_report(
    "button_color_test", expected_weights, baseline_rate
)
print(report)

V. 数据分析与结果解读

5.1 统计显著性检验

正确理解和应用统计检验是解读A/B测试结果的关键。

5.2 效应大小与业务意义

统计显著性不等于业务重要性。我们需要同时考虑效应大小和实际影响。

class ResultsAnalyzer:
    """结果分析器"""
    
    def __init__(self, confidence_level=0.95):
        self.confidence_level = confidence_level
    
    def analyze_proportion_test(self, control_success, control_total, 
                              treatment_success, treatment_total):
        """分析比例检验结果"""
        from statsmodels.stats.proportion import proportions_ztest
        
        # 计算比例
        p_control = control_success / control_total
        p_treatment = treatment_success / treatment_total
        
        # 执行z检验
        count = [control_success, treatment_success]
        nobs = [control_total, treatment_total]
        z_stat, p_value = proportions_ztest(count, nobs, alternative='two-sided')
        
        # 计算置信区间
        ci_control = self._proportion_ci(control_success, control_total)
        ci_treatment = self._proportion_ci(treatment_success, treatment_total)
        
        # 计算效应大小
        relative_improvement = (p_treatment - p_control) / p_control
        absolute_difference = p_treatment - p_control
        
        # 计算统计功效
        power = self._calculate_power(p_control, p_treatment, control_total, treatment_total)
        
        return {
            'p_control': p_control,
            'p_treatment': p_treatment,
            'absolute_difference': absolute_difference,
            'relative_improvement': relative_improvement,
            'z_statistic': z_stat,
            'p_value': p_value,
            'significant': p_value < (1 - self.confidence_level),
            'ci_control': ci_control,
            'ci_treatment': ci_treatment,
            'power': power,
            'n_control': control_total,
            'n_treatment': treatment_total
        }
    
    def _proportion_ci(self, successes, total):
        """计算比例置信区间"""
        from statsmodels.stats.proportion import proportion_confint
        ci_low, ci_high = proportion_confint(successes, total, alpha=1-self.confidence_level)
        return (ci_low, ci_high)
    
    def _calculate_power(self, p1, p2, n1, n2, alpha=0.05):
        """计算统计功效"""
        from statsmodels.stats.power import NormalIndPower
        import math
        
        effect_size = 2 * math.asin(math.sqrt(p2)) - 2 * math.asin(math.sqrt(p1))
        power_analysis = NormalIndPower()
        power = power_analysis.solve_power(
            effect_size=effect_size, 
            nobs1=n1, 
            alpha=alpha, 
            ratio=n2/n1
        )
        return power
    
    def analyze_revenue_data(self, control_revenues, treatment_revenues):
        """分析收入数据（非正态分布）"""
        from scipy import stats
        
        # 由于收入数据通常是非正态的，使用非参数检验
        # Mann-Whitney U检验
        u_stat, p_value = stats.mannwhitneyu(treatment_revenues, control_revenues, alternative='two-sided')
        
        # 计算中位数差异
        median_control = np.median(control_revenues)
        median_treatment = np.median(treatment_revenues)
        median_difference = median_treatment - median_control
        
        # 计算 bootstrap 置信区间
        ci_difference = self._bootstrap_ci(control_revenues, treatment_revenues)
        
        return {
            'test_type': 'mann_whitney',
            'u_statistic': u_stat,
            'p_value': p_value,
            'median_control': median_control,
            'median_treatment': median_treatment,
            'median_difference': median_difference,
            'ci_difference': ci_difference,
            'significant': p_value < (1 - self.confidence_level)
        }
    
    def _bootstrap_ci(self, control_data, treatment_data, n_bootstrap=10000):
        """使用bootstrap计算置信区间"""
        differences = []
        n_control = len(control_data)
        n_treatment = len(treatment_data)
        
        for _ in range(n_bootstrap):
            # 有放回抽样
            bootstrap_control = np.random.choice(control_data, n_control, replace=True)
            bootstrap_treatment = np.random.choice(treatment_data, n_treatment, replace=True)
            
            median_diff = np.median(bootstrap_treatment) - np.median(bootstrap_control)
            differences.append(median_diff)
        
        # 计算百分位数置信区间
        alpha = 1 - self.confidence_level
        lower = np.percentile(differences, 100 * alpha/2)
        upper = np.percentile(differences, 100 * (1 - alpha/2))
        
        return (lower, upper)
    
    def generate_decision_framework(self, analysis_result, mde, business_impact):
        """生成决策框架"""
        statistical_significant = analysis_result['significant']
        effect_size = analysis_result.get('relative_improvement', 
                         analysis_result.get('median_difference', 0))
        
        # 判断业务显著性
        business_significant = abs(effect_size) >= mde
        
        # 决策矩阵
        if statistical_significant and business_significant:
            if effect_size > 0:
                decision = "LAUNCH - 显著正向效果"
                confidence = "高"
            else:
                decision = "STOP - 显著负向效果" 
                confidence = "高"
        elif statistical_significant and not business_significant:
            decision = "HOLD - 统计显著但业务影响小"
            confidence = "中"
        elif not statistical_significant and business_significant:
            decision = "CONTINUE - 效应大有潜力但需要更多数据"
            confidence = "低"
        else:
            decision = "HOLD - 无显著效果"
            confidence = "中"
        
        return {
            'decision': decision,
            'confidence': confidence,
            'statistical_significant': statistical_significant,
            'business_significant': business_significant,
            'effect_size': effect_size,
            'recommendation': self._generate_recommendation(decision, analysis_result)
        }
    
    def _generate_recommendation(self, decision, analysis_result):
        """生成具体建议"""
        recommendations = {
            "LAUNCH - 显著正向效果": [
                "准备全量发布计划",
                "监控长期效果",
                "评估对其他指标的影响"
            ],
            "STOP - 显著负向效果": [
                "立即停止实验",
                "分析负面原因",
                "考虑回滚更改"
            ],
            "HOLD - 统计显著但业务影响小": [
                "评估实施成本",
                "考虑与其他优化结合",
                "可能不值得单独发布"
            ],
            "CONTINUE - 效应大有潜力但需要更多数据": [
                "延长实验时间",
                "增加样本量", 
                "监控其他变体表现"
            ],
            "HOLD - 无显著效果": [
                "分析实验设计",
                "考虑不同的优化方向",
                "收集用户反馈"
            ]
        }
        return recommendations.get(decision, ["需要进一步分析"])

# 结果分析示例
analyzer = ResultsAnalyzer(confidence_level=0.95)

# 模拟转化率数据
np.random.seed(42)
control_conversions = 120
control_total = 1000
treatment_conversions = 150  
treatment_total = 1000

# 分析比例检验
conversion_analysis = analyzer.analyze_proportion_test(
    control_conversions, control_total,
    treatment_conversions, treatment_total
)

print("转化率分析结果:")
print(f"对照组转化率: {conversion_analysis['p_control']:.3f}")
print(f"实验组转化率: {conversion_analysis['p_treatment']:.3f}")
print(f"绝对差异: {conversion_analysis['absolute_difference']:.3f}")
print(f"相对提升: {conversion_analysis['relative_improvement']:.3f}")
print(f"P值: {conversion_analysis['p_value']:.4f}")
print(f"统计显著性: {conversion_analysis['significant']}")
print(f"统计功效: {conversion_analysis['power']:.3f}")

# 决策框架
mde = 0.10  # 10%最小业务效应
business_impact = "中等"  # 业务影响程度

decision_framework = analyzer.generate_decision_framework(
    conversion_analysis, mde, business_impact
)

print(f"\n决策建议: {decision_framework['decision']}")
print(f置信度: {decision_framework['confidence']}")
print("具体建议:")
for rec in decision_framework['recommendation']:
    print(f"  • {rec}")

# 收入数据分析示例
print("\n" + "="*50)
print("收入数据分析示例")

# 模拟收入数据（典型的偏态分布）
control_revenues = np.random.exponential(50, 1000)
treatment_revenues = np.random.exponential(55, 1000)  # 稍微更高的收入

revenue_analysis = analyzer.analyze_revenue_data(control_revenues, treatment_revenues)

print(f"收入中位数 - 对照组: ${revenue_analysis['median_control']:.2f}")
print(f"收入中位数 - 实验组: ${revenue_analysis['median_treatment']:.2f}")
print(f"中位数差异: ${revenue_analysis['median_difference']:.2f}")
print(f"P值: {revenue_analysis['p_value']:.4f}")
print(f"统计显著性: {revenue_analysis['significant']}")

5.3 多重检验校正

当同时运行多个实验或检查多个指标时，需要进行多重检验校正。

class MultipleTestingCorrection:
    """多重检验校正"""
    
    def __init__(self):
        self.methods = ['bonferroni', 'fdr_bh', 'holm']
    
    def apply_correction(self, p_values, method='fdr_bh', alpha=0.05):
        """应用多重检验校正"""
        from statsmodels.stats.multitest import multipletests
        
        if method not in self.methods:
            raise ValueError(f"方法必须是: {self.methods}")
        
        rejected, corrected_pvals, _, _ = multipletests(
            p_values, alpha=alpha, method=method
        )
        
        return {
            'original_pvalues': p_values,
            'corrected_pvalues': corrected_pvals,
            'rejected': rejected,
            'method': method,
            'alpha': alpha
        }
    
    def analyze_experiment_family(self, experiments_data, family_wise_alpha=0.05):
        """分析实验族（一组相关实验）"""
        # 提取所有p值
        p_values = [exp['p_value'] for exp in experiments_data]
        experiment_names = [exp['name'] for exp in experiments_data]
        
        # 应用校正
        bonferroni_results = self.apply_correction(p_values, 'bonferroni', family_wise_alpha)
        fdr_results = self.apply_correction(p_values, 'fdr_bh', family_wise_alpha)
        
        # 生成报告
        report = self._generate_family_report(
            experiment_names, p_values, bonferroni_results, fdr_results
        )
        
        return {
            'bonferroni': bonferroni_results,
            'fdr': fdr_results,
            'report': report
        }
    
    def _generate_family_report(self, names, original_pvals, bonferroni, fdr):
        """生成实验族报告"""
        report = ["多重检验校正报告", "=" * 50]
        
        for i, name in enumerate(names):
            report.append(f"\n{name}:")
            report.append(f"  原始P值: {original_pvals[i]:.6f}")
            report.append(f"  Bonferroni校正P值: {bonferroni['corrected_pvalues'][i]:.6f}")
            report.append(f"  FDR校正P值: {fdr['corrected_pvalues'][i]:.6f}")
            
            # 显著性标记
            bonf_sig = "✅" if bonferroni['rejected'][i] else "❌"
            fdr_sig = "✅" if fdr['rejected'][i] else "❌"
            
            report.append(f"  Bonferroni显著性: {bonf_sig}")
            report.append(f"  FDR显著性: {fdr_sig}")
        
        # 总结
        n_bonf_sig = sum(bonferroni['rejected'])
        n_fdr_sig = sum(fdr['rejected'])
        
        report.append(f"\n总结:")
        report.append(f"  Bonferroni校正后显著: {n_bonf_sig}/{len(names)}")
        report.append(f"  FDR校正后显著: {n_fdr_sig}/{len(names)}")
        report.append(f"  推荐使用: {'FDR' if n_fdr_sig > n_bonf_sig else 'Bonferroni'}")
        
        return "\n".join(report)

# 多重检验校正示例
correction = MultipleTestingCorrection()

# 模拟多个相关实验
experiments = [
    {'name': '按钮颜色测试', 'p_value': 0.04},
    {'name': '标题文案测试', 'p_value': 0.03},
    {'name': '图片优化测试', 'p_value': 0.08},
    {'name': '价格显示测试', 'p_value': 0.01},
    {'name': '推荐算法测试', 'p_value': 0.06}
]

# 应用多重检验校正
family_analysis = correction.analyze_experiment_family(experiments, family_wise_alpha=0.05)
print(family_analysis['report'])

VI. 高级主题与最佳实践

6.1 常见陷阱与规避策略

A/B测试实践中存在许多常见陷阱，了解并规避这些陷阱至关重要。

6.2 组织实践与文化

成功的A/B测试不仅需要技术能力，还需要正确的组织文化和流程。

class ABTestBestPractices:
    """A/B测试最佳实践指南"""
    
    def __init__(self):
        self.pitfalls = {
            'peeking': {
                'description': '过早查看结果并停止实验',
                'impact': '增加第一类错误率',
                'solution': '预先确定样本量，避免中途查看',
                'severity': 'high'
            },
            'multiple_metrics': {
                'description': '检查过多指标而不进行校正',
                'impact': '增加假阳性风险', 
                'solution': '确定主要指标，使用多重检验校正',
                'severity': 'high'
            },
            'selection_bias': {
                'description': '样本选择偏差',
                'impact': '结果不可泛化',
                'solution': '确保随机化正确实施',
                'severity': 'medium'
            },
            'novelty_effect': {
                'description': '新奇效应影响短期结果',
                'impact': '高估长期效果',
                'solution': '运行足够长时间，分析时间趋势',
                'severity': 'medium'
            },
            'carryover_effect': {
                'description': '实验间的相互影响',
                'impact': '结果污染',
                'solution': '使用正交实验设计，控制流量重叠',
                'severity': 'medium'
            }
        }
    
    def generate_pitfall_checklist(self):
        """生成陷阱检查清单"""
        checklist = ["A/B测试陷阱检查清单", "=" * 50]
        
        for pitfall, info in self.pitfalls.items():
            severity_icon = "🔴" if info['severity'] == 'high' else "🟡"
            checklist.append(f"\n{severity_icon} {pitfall}:")
            checklist.append(f"   描述: {info['description']}")
            checklist.append(f"   影响: {info['impact']}")
            checklist.append(f"   解决方案: {info['solution']}")
        
        checklist.append("\n✅ 最佳实践总结:")
        checklist.append("   • 预先注册实验假设和指标")
        checklist.append("   • 计算足够的样本量")
        checklist.append("   • 确保正确的随机化")
        checklist.append("   • 运行完整实验周期")
        checklist.append("   • 使用适当的统计方法")
        checklist.append("   • 考虑业务意义而不仅是统计意义")
        checklist.append("   • 记录和分享学习成果")
        
        return "\n".join(checklist)
    
    def calculate_risk_score(self, experiment_design, traffic_volume, business_criticality):
        """计算实验风险分数"""
        risk_factors = {
            'sample_size_adequacy': 0.3 if experiment_design.get('pre_calculated_sample_size') else 1.0,
            'randomization_check': 0.2 if experiment_design.get('randomization_validated') else 0.8,
            'primary_metric_defined': 0.1 if experiment_design.get('primary_metric') else 0.5,
            'guardrail_metrics': 0.1 if experiment_design.get('guardrail_metrics') else 0.4,
            'duration_adequacy': 0.3 if experiment_design.get('adequate_duration') else 0.7
        }
        
        # 基础风险分数
        base_risk = sum(risk_factors.values()) / len(risk_factors)
        
        # 调整因子
        traffic_factor = 1.0 if traffic_volume == 'high' else 0.7 if traffic_volume == 'medium' else 0.4
        business_factor = 1.5 if business_criticality == 'high' else 1.0
        
        final_risk = base_risk * traffic_factor * business_factor
        
        # 风险等级
        if final_risk < 0.3:
            risk_level = "低风险"
            recommendation = "可以按计划进行"
        elif final_risk < 0.6:
            risk_level = "中等风险" 
            recommendation = "建议进行设计优化"
        else:
            risk_level = "高风险"
            recommendation = "需要重新设计实验"
        
        return {
            'risk_score': final_risk,
            'risk_level': risk_level,
            'recommendation': recommendation,
            'factors': risk_factors
        }

class ExperimentDocumentation:
    """实验文档化工具"""
    
    def __init__(self):
        self.template = {
            'experiment_id': '',
            'title': '',
            'hypothesis': '',
            'primary_metric': '',
            'secondary_metrics': [],
            'guardrail_metrics': [],
            'variants': {},
            'target_audience': '',
            'sample_size_calculation': {},
            'success_criteria': '',
            'risks': '',
            'stakeholders': [],
            'timeline': {}
        }
    
    def create_experiment_charter(self, experiment_data):
        """创建实验章程"""
        charter = [
            "实验章程",
            "=" * 50,
            f"实验ID: {experiment_data['experiment_id']}",
            f"标题: {experiment_data['title']}",
            "",
            "假设:",
            f"  {experiment_data['hypothesis']}",
            "",
            "指标定义:",
            f"  主要指标: {experiment_data['primary_metric']}",
            "  辅助指标:",
        ]
        
        for metric in experiment_data['secondary_metrics']:
            charter.append(f"    • {metric}")
        
        charter.extend([
            "  护栏指标:",
        ])
        
        for metric in experiment_data['guardrail_metrics']:
            charter.append(f"    • {metric}")
        
        charter.extend([
            "",
            "变体定义:",
        ])
        
        for variant, description in experiment_data['variants'].items():
            charter.append(f"  {variant}: {description}")
        
        charter.extend([
            "",
            "样本量计算:",
            f"  基线率: {experiment_data['sample_size_calculation']['baseline_rate']}",
            f"  MDE: {experiment_data['sample_size_calculation']['mde']}",
            f"  显著性水平: {experiment_data['sample_size_calculation']['alpha']}",
            f"  统计功效: {experiment_data['sample_size_calculation']['power']}",
            f"  每组样本量: {experiment_data['sample_size_calculation']['sample_size_per_variant']}",
            f"  总样本量: {experiment_data['sample_size_calculation']['total_sample_size']}",
            f"  预计时长: {experiment_data['sample_size_calculation']['estimated_duration']}天",
            "",
            "成功标准:",
            f"  {experiment_data['success_criteria']}",
            "",
            "相关方:",
        ])
        
        for stakeholder in experiment_data['stakeholders']:
            charter.append(f"  • {stakeholder}")
        
        return "\n".join(charter)

# 最佳实践示例
best_practices = ABTestBestPractices()
print(best_practices.generate_pitfall_checklist())

print("\n" + "="*50)
print("实验风险评估示例")

# 实验设计评估
experiment_design = {
    'pre_calculated_sample_size': True,
    'randomization_validated': False,  # 未验证随机化
    'primary_metric': True,
    'guardrail_metrics': False,  # 未定义护栏指标
    'adequate_duration': True
}

risk_assessment = best_practices.calculate_risk_score(
    experiment_design, 
    traffic_volume='high', 
    business_criticality='medium'
)

print(f"风险分数: {risk_assessment['risk_score']:.2f}")
print(f"风险等级: {risk_assessment['risk_level']}")
print(f"建议: {risk_assessment['recommendation']}")

# 实验文档化示例
print("\n" + "="*50)
print("实验文档化示例")

doc_tool = ExperimentDocumentation()

experiment_data = {
    'experiment_id': '2024-Q1-button-color',
    'title': '主要按钮颜色对转化率的影响',
    'hypothesis': '将主要按钮从蓝色改为绿色将提高注册转化率，因为绿色在心理学上与确认和前进相关',
    'primary_metric': '注册转化率',
    'secondary_metrics': ['点击率', '页面停留时间', ' bounce率'],
    'guardrail_metrics': ['页面加载时间', '错误率'],
    'variants': {
        'control': '蓝色按钮 (#007BFF)',
        'treatment': '绿色按钮 (#28A745)'
    },
    'sample_size_calculation': {
        'baseline_rate': 0.15,
        'mde': 0.10,
        'alpha': 0.05,
        'power': 0.8,
        'sample_size_per_variant': 3500,
        'total_sample_size': 7000,
        'estimated_duration': 7
    },
    'success_criteria': '绿色按钮在95%置信水平下显著提升注册转化率，且相对提升至少达到5%',
    'stakeholders': ['产品经理-张三', '设计师-李四', '工程师-王五', '数据分析师-赵六']
}

charter = doc_tool.create_experiment_charter(experiment_data)
print(charter)

6.3 持续优化与学习文化

建立从A/B测试中持续学习和改进的机制。

class LearningRepository:
    """学习知识库"""
    
    def __init__(self, db_connection):
        self.db = db_connection
        self.insight_categories = [
            'ui_ux', 'pricing', 'messaging', 'onboarding', 'feature_impact'
        ]
    
    def log_experiment_learning(self, experiment_id, success, key_learnings, 
                              unexpected_findings, next_steps):
        """记录实验学习成果"""
        learning_record = {
            'experiment_id': experiment_id,
            'timestamp': datetime.now().isoformat(),
            'success': success,
            'key_learnings': key_learnings,
            'unexpected_findings': unexpected_findings,
            'next_steps': next_steps,
            'impact_score': self._calculate_impact_score(success, key_learnings)
        }
        
        # 这里应该是数据库插入操作
        print(f"记录学习成果: {experiment_id}")
        return learning_record
    
    def _calculate_impact_score(self, success, learnings):
        """计算影响分数"""
        base_score = 10 if success else 5
        learning_bonus = min(len(learnings) * 2, 10)  # 最多加10分
        return base_score + learning_bonus
    
    def generate_quarterly_learnings_report(self, quarter):
        """生成季度学习报告"""
        # 这里应该是从数据库获取数据
        mock_learnings = [
            {
                'experiment_id': '2024-Q1-button-color',
                'success': True,
                'key_learnings': ['绿色按钮比蓝色按钮转化率高12%', '颜色对比度对可访问性很重要'],
                'category': 'ui_ux',
                'impact_score': 18
            },
            {
                'experiment_id': '2024-Q1-pricing-page',
                'success': False, 
                'key_learnings': ['价格显示方式对高端用户影响更大', '需要更好的价值传达'],
                'category': 'pricing',
                'impact_score': 12
            }
        ]
        
        report = [f"{quarter} 季度A/B测试学习报告", "=" * 50]
        
        # 按类别汇总
        by_category = {}
        for learning in mock_learnings:
            category = learning['category']
            if category not in by_category:
                by_category[category] = []
            by_category[category].append(learning)
        
        for category, learnings in by_category.items():
            report.append(f"\n{category.upper()} 类别:")
            success_rate = sum(1 for l in learnings if l['success']) / len(learnings)
            avg_impact = sum(l['impact_score'] for l in learnings) / len(learnings)
            
            report.append(f"  实验数量: {len(learnings)}")
            report.append(f"  成功率: {success_rate:.1%}")
            report.append(f"  平均影响分数: {avg_impact:.1f}")
            
            for learning in learnings:
                status = "✅" if learning['success'] else "❌"
                report.append(f"  {status} {learning['experiment_id']}: {learning['key_learnings'][0]}")
        
        # 总体洞察
        report.append("\n关键洞察:")
        report.append("  • 用户对视觉变化反应积极")
        report.append("  • 定价实验需要更精细的受众定位")
        report.append("  • 文案测试显示出较高的成功率")
        
        report.append("\n改进建议:")
        report.append("  • 增加UI/UX实验的比例")
        report.append("  • 为定价实验建立更好的用户分群")
        report.append("  • 优化实验流程，缩短从洞察到行动的时间")
        
        return "\n".join(report)

# 学习知识库示例
learning_repo = LearningRepository(None)  # 简化示例，不使用真实数据库

# 记录学习成果
learning_repo.log_experiment_learning(
    experiment_id='2024-Q1-homepage-redesign',
    success=True,
    key_learnings=[
        '简化导航提高用户参与度',
        '英雄区域的清晰价值主张对转化率至关重要'
    ],
    unexpected_findings=['移动端效果比桌面端更明显'],
    next_steps=['将成功模式应用到其他页面', '进行移动端专项优化']
)

# 生成季度报告
quarterly_report = learning_repo.generate_quarterly_learnings_report('2024年第一季度')
print(quarterly_report)