[数据分析]A/B测试实战：一个产品UI优化的完整案例复盘

// 分流服务核心代码
function assignUserToGroup(userId, experimentId) {
  // 使用murmurHash3确保一致的用户分组
  const hash = murmurHash3(userId + experimentId);
  const bucket = hash % 100;
  
  // 50%用户分配到控制组，50%到实验组
  if (bucket < 50) {
    return 'control';
  } else {
    return 'treatment';
  }
}

// 前端集成代码
const userId = getUserId(); // 获取稳定的用户标识
const experimentId = 'ui_optimization_2024';

const group = await fetchAssignment(userId, experimentId);
if (group === 'treatment') {
  applyNewUI(); // 应用新UI样式和功能
}

// 数据收集代码示例
function trackEvent(eventName, properties) {
  // 确保包含实验分组信息
  const eventData = {
    event: eventName,
    properties: {
      ...properties,
      userId: getUserId(),
      experimentId: 'ui_optimization_2024',
      group: getCurrentGroup(),
      timestamp: Date.now()
    }
  };
  
  // 发送到数据收集端点
  sendToAnalytics(eventData);
}

// 跟踪特定事件
trackEvent('product_page_view', {
  productId: '12345',
  category: 'electronics'
});

trackEvent('add_to_cart', {
  productId: '12345',
  price: 99.99,
  quantity: 1
});

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

# 加载数据
df = pd.read_csv('ab_test_data.csv')

# 数据清洗
print("原始数据形状:", df.shape)
print("缺失值检查:")
print(df.isnull().sum())

# 移除异常值和无效记录
df = df[df['session_duration'] <= 3600]  # 移除停留时间超过1小时的异常值
df = df[df['page_views'] <= 50]  # 移除页面浏览超过50的异常值

# 添加必要计算字段
df['conversion'] = df['add_to_cart'] > 0  # 转化标志

print("清洗后数据形状:", df.shape)

# 检查两组在关键特征上的平衡性
control_group = df[df['group'] == 'control']
treatment_group = df[df['group'] == 'treatment']

print("组大小:")
print("控制组:", len(control_group))
print("实验组:", len(treatment_group))

# 检查人口统计学特征平衡
features_to_check = ['age', 'gender', 'new_user', 'previous_purchases']
balance_results = []

for feature in features_to_check:
    t_stat, p_value = stats.ttest_ind(control_group[feature], treatment_group[feature], nan_policy='omit')
    balance_results.append({'feature': feature, 'p_value': p_value, 'balanced': p_value > 0.05})

balance_df = pd.DataFrame(balance_results)
print("随机化检查结果:")
print(balance_df)

# 计算转化率
control_conversion = control_group['conversion'].mean()
treatment_conversion = treatment_group['conversion'].mean()

print("转化率结果:")
print(f"控制组: {control_conversion:.4f} ({len(control_group[control_group['conversion']])}/{len(control_group)})")
print(f"实验组: {treatment_conversion:.4f} ({len(treatment_group[treatment_group['conversion']])}/{len(treatment_group)})")
print(f"绝对提升: {treatment_conversion - control_conversion:.4f}")
print(f"相对提升: {(treatment_conversion - control_conversion) / control_conversion * 100:.2f}%")

# 执行统计显著性检验
from statsmodels.stats.proportion import proportions_ztest

count = np.array([len(treatment_group[treatment_group['conversion']]), 
                  len(control_group[control_group['conversion']])])
nobs = np.array([len(treatment_group), len(control_group)])

z_stat, p_value = proportions_ztest(count, nobs)
print(f"Z统计量: {z_stat:.4f}")
print(f"P值: {p_value:.4f}")
print(f"结果显著: {p_value < 0.05}")

# 分析不同用户群体的效果异质性
dimensions = ['new_user', 'device_type', 'traffic_source']

results = []
for dim in dimensions:
    for value in df[dim].unique():
        subset = df[df[dim] == value]
        control_conv = subset[subset['group'] == 'control']['conversion'].mean()
        treatment_conv = subset[subset['group'] == 'treatment']['conversion'].mean()
        
        if np.isnan(control_conv) or np.isnan(treatment_conv):
            continue
            
        lift = (treatment_conv - control_conv) / control_conv
        results.append({
            'dimension': dim,
            'value': value,
            'control_rate': control_conv,
            'treatment_rate': treatment_conv,
            'lift': lift
        })

results_df = pd.DataFrame(results)
print("多维度分析结果:")
print(results_df.sort_values('lift', ascending=False))

# 创建结果可视化
plt.figure(figsize=(15, 10))

# 1. 主要转化率对比
plt.subplot(2, 2, 1)
rates = [control_conversion, treatment_conversion]
groups = ['控制组', '实验组']
colors = ['#1f77b4', '#ff7f0e']
bars = plt.bar(groups, rates, color=colors, alpha=0.7)
plt.ylabel('转化率')
plt.title('加入购物车转化率对比')
for bar, rate in zip(bars, rates):
    plt.text(bar.get_x() + bar.get_width()/2, bar.get_height() + 0.001, 
             f'{rate:.3%}', ha='center', va='bottom')

# 2. 每日转化率趋势
plt.subplot(2, 2, 2)
daily_rates = df.groupby(['date', 'group'])['conversion'].mean().unstack()
daily_rates.plot(ax=plt.gca(), color=['#1f77b4', '#ff7f0e'])
plt.title('每日转化率趋势')
plt.ylabel('转化率')
plt.legend(['控制组', '实验组'])

# 3. 不同维度提升幅度
plt.subplot(2, 2, 3)
dimension_lift = results_df.pivot(index='value', columns='dimension', values='lift')
dimension_lift.plot(kind='bar', ax=plt.gca())
plt.title('不同维度下的提升幅度')
plt.ylabel('相对提升')
plt.xticks(rotation=45)

# 4. 置信区间可视化
plt.subplot(2, 2, 4)
lift = treatment_conversion - control_conversion
se = np.sqrt(control_conversion*(1-control_conversion)/len(control_group) + 
             treatment_conversion*(1-treatment_conversion)/len(treatment_group))
ci_lower = lift - 1.96 * se
ci_upper = lift + 1.96 * se
plt.errorbar(0, lift, yerr=[[lift - ci_lower], [ci_upper - lift]], 
             fmt='o', color='black', capsize=5)
plt.xlim(-0.5, 0.5)
plt.axhline(y=0, color='red', linestyle='--')
plt.title('提升幅度与95%置信区间')
plt.ylabel('绝对提升')
plt.xticks([])

plt.tight_layout()
plt.savefig('ab_test_results.png', dpi=300, bbox_inches='tight')
plt.show()

# 决策矩阵分析
decision_matrix = pd.DataFrame({
    ' criterion': ['统计显著性', '效应大小', '业务影响', '实施成本', '用户体验', '风险水平'],
    'weight': [0.25, 0.20, 0.25, 0.15, 0.10, 0.05],
    'score_control': [0, 0, 0, 10, 6, 10],  # 保持现状的评分
    'score_treatment': [10, 8, 9, 7, 8, 8]   # 采用新UI的评分
})

# 计算加权得分
decision_matrix['control_weighted'] = decision_matrix['weight'] * decision_matrix['score_control']
decision_matrix['treatment_weighted'] = decision_matrix['weight'] * decision_matrix['score_treatment']

total_control = decision_matrix['control_weighted'].sum()
total_treatment = decision_matrix['treatment_weighted'].sum()

print("决策矩阵分析结果:")
print(decision_matrix)
print(f"\n保持现状总分: {total_control:.2f}")
print(f"采用新UI总分: {total_treatment:.2f}")
print(f"推荐决策: {'采用新UI' if total_treatment > total_control else '保持现状'}")

# 监控仪表板代码示例
def create_monitoring_dashboard():
    # 实时监控关键指标
    metrics_to_monitor = [
        'add_to_cart_rate',
        'checkout_conversion_rate',
        'revenue_per_user',
        'bounce_rate'
    ]
    
    # 设置警报阈值
    alert_thresholds = {
        'add_to_cart_rate': -0.10,  # 下降超过10%触发警报
        'checkout_conversion_rate': -0.15,
        'revenue_per_user': -0.10,
        'bounce_rate': +0.20  # 跳出率上升超过20%触发警报
    }
    
    print("监控仪表板配置完成")
    print("监控指标:", metrics_to_monitor)
    print("警报阈值:", alert_thresholds)
    
    return True

# 实施监控
create_monitoring_dashboard()

from statsmodels.stats.power import NormalIndPower
from statsmodels.stats.proportion import proportion_effectsize

# 精确计算所需样本量
def calculate_sample_size(base_rate, mde, alpha=0.05, power=0.8):
    """计算每组所需样本量"""
    effect_size = proportion_effectsize(base_rate, base_rate * (1 + mde))
    power_analysis = NormalIndPower()
    sample_size = power_analysis.solve_power(
        effect_size=effect_size,
        alpha=alpha,
        power=power,
        ratio=1.0
    )
    return int(sample_size)

# 示例计算
base_conversion = 0.087  # 8.7%
mde = 0.15  # 15%相对提升

required_sample_size = calculate_sample_size(base_conversion, mde)
print(f"每组所需样本量: {required_sample_size}")
print(f"总所需样本量: {required_sample_size * 2}")

# 多重比较校正示例
from statsmodels.stats.multitest import multipletests

# 模拟测试多个指标
p_values = [0.03, 0.05, 0.01, 0.10, 0.25, 0.04, 0.02, 0.08, 0.15, 0.06]

# 应用各种校正方法
bonferroni_corrected = multipletests(p_values, alpha=0.05, method='bonferroni')
holm_corrected = multipletests(p_values, alpha=0.05, method='holm')
fdr_corrected = multipletests(p_values, alpha=0.05, method='fdr_bh')

print("原始p值:", p_values)
print("Bonferroni校正后显著:", bonferroni_corrected[0])
print("Holm校正后显著:", holm_corrected[0])
print("FDR校正后显著:", fdr_corrected[0])

# 模拟多变量测试(MVT)设计
def generate_factorial_design(factors):
    """生成全因子实验设计"""
    from itertools import product
    return list(product(*factors.values()))

# 示例：测试3个因素，每个因素2个水平
factors = {
    'button_color': ['blue', 'green'],
    'button_size': ['small', 'large'],
    'button_text': ['Add to Cart', 'Buy Now']
}

design = generate_factorial_design(factors)
print("全因子实验设计:")
for i, combination in enumerate(design):
    print(f"组合 {i+1}: {combination}")

# 贝叶斯A/B测试示例（概念代码）
def bayesian_ab_test(control_success, control_total, treatment_success, treatment_total):
    """使用贝叶斯方法分析A/B测试"""
    # 使用Beta先验分布
    from scipy.stats import beta
    
    # 假设使用均匀先验Beta(1,1)
    control_posterior = beta(1 + control_success, 1 + control_total - control_success)
    treatment_posterior = beta(1 + treatment_success, 1 + treatment_total - treatment_success)
    
    # 计算treatment优于control的概率
    samples = 100000
    treatment_samples = treatment_posterior.rvs(samples)
    control_samples = control_posterior.rvs(samples)
    probability = (treatment_samples > control_samples).mean()
    
    return probability

# 示例计算
prob = bayesian_ab_test(550, 6300, 650, 6400)
print(f"新UI优于旧UI的概率: {prob:.3f}")