[数据分析]A/B测试中的"坑"：常见误区与规避指南

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

# 设置随机种子保证可重现
np.random.seed(42)  

# 真实转化率：A和B完全相同，均为10%
true_rate_a = 0.10
true_rate_b = 0.10

# 模拟100次完全相同的A/B测试
n_simulations = 100
n_samples = 1000  # 每组样本量
p_values = []

for i in range(n_simulations):
    # 模拟A组数据：生成1000个二元变量，1的概率为true_rate_a
    group_a = np.random.binomial(1, true_rate_a, n_samples)
    # 模拟B组数据：生成1000个二元变量，1的概率为true_rate_b
    group_b = np.random.binomial(1, true_rate_b, n_samples)
    
    # 执行双比例Z检验，计算p值（这里用卡方检验代替，结果类似）
    _, p_value = stats.chisquare([group_b.sum(), group_a.sum()],
                                 f_exp=[group_b.mean() * (n_samples*2), group_a.mean() * (n_samples*2)])
    p_values.append(p_value)

# 可视化100次实验的p值分布
plt.figure(figsize=(10, 6))
plt.hist(p_values, bins=20, edgecolor='black', alpha=0.7)
plt.axvline(x=0.05, color='red', linestyle='--', label='显著性阈值 (α=0.05)')
plt.xlabel('P值')
plt.ylabel('频次')
plt.title('100次模拟A/B测试的P值分布 (真实效应为0)')
plt.legend()
plt.show()

# 计算有多少次实验出现了“假阳性”
false_positives = sum(np.array(p_values) < 0.05)
print(f"在{false_positives}次实验中，我们错误地得出了‘统计显著’的结论（假阳性）。")
print(f"假阳性率: {false_positives/n_simulations:.2%}")

# 模拟实时窥探过程
true_rate_a = 0.10
true_rate_b = 0.10  # 依然没有真实效应

n_samples_peek = 100  # 每收集100个样本窥探一次
max_samples = 5000    # 实验最大样本量
peek_points = range(n_samples_peek, max_samples + 1, n_samples_peek)
p_value_history = []
observed_diff_history = []

# 模拟数据流，逐步增加样本
for n in peek_points:
    # 生成至今为止的所有数据
    group_a_so_far = np.random.binomial(1, true_rate_a, n)
    group_b_so_far = np.random.binomial(1, true_rate_b, n)
    
    # 计算当前的p值
    _, p_value = stats.chisquare([group_b_so_far.sum(), group_a_so_far.sum()],
                                 f_exp=[group_b_so_far.mean() * (n*2), group_a_so_far.mean() * (n*2)])
    current_diff = group_b_so_far.mean() - group_a_so_far.mean()
    
    p_value_history.append(p_value)
    observed_diff_history.append(current_diff)

# 绘制p值随时间（样本量）的变化
plt.figure(figsize=(12, 6))
plt.subplot(1, 2, 1)
plt.plot(peek_points, p_value_history, 'b-', marker='o')
plt.axhline(y=0.05, color='r', linestyle='--', label='α=0.05')
plt.yscale('log')  # 使用对数坐标更清晰地显示p值变化
plt.xlabel('总样本量（每组）')
plt.ylabel('P值 (对数尺度)')
plt.title('P值在窥探过程中的波动')
plt.legend()

plt.subplot(1, 2, 2)
plt.plot(peek_points, observed_diff_history, 'g-', marker='s')
plt.axhline(y=0, color='r', linestyle='--', label='真实效应=0')
plt.xlabel('总样本量（每组）')
plt.ylabel('观察到的差异 (B - A)')
plt.title('观察效应在窥探过程中的波动')
plt.legend()
plt.tight_layout()
plt.show()

# 检查是否在某个窥探点p值曾小于0.05
false_positive_peek = any(np.array(p_value_history) < 0.05)
print(f"在窥探过程中，是否曾出现过假阳性（p < 0.05）? {false_positive_peek}")

# 设置参数：用户类型是混杂变量
# 新用户更喜欢新内容，但也更难转化
rate_new_user_a = 0.15  # 新用户用A的转化率
rate_new_user_b = 0.25  # 新用户用B的转化率 (B更好!)
rate_old_user_a = 0.30  # 老用户用A的转化率
rate_old_user_b = 0.25  # 老用户用B的转化率 (A更好!)

# 但是！新算法B被更多地展示给了老用户（基于历史行为），而老用户本身转化率高
# 模拟数据分布
n_new_users = 1000  # 新用户数量较少
n_old_users = 4000  # 老用户数量较多

# 分配：假设B组中老用户占比更高（80%），A组中新老用户各半
# 生成数据
np.random.seed(123)
# A组
a_new_users = np.random.binomial(1, rate_new_user_a, int(n_new_users*0.5))
a_old_users = np.random.binomial(1, rate_old_user_a, int(n_old_users*0.5))
group_a = np.concatenate([a_new_users, a_old_users])

# B组 (大部分流量给了老用户)
b_new_users = np.random.binomial(1, rate_new_user_b, int(n_new_users*0.5))
b_old_users = np.random.binomial(1, rate_old_user_b, int(n_old_users*0.8)) # B组老用户更多！
group_b = np.concatenate([b_new_users, b_old_users])

# 计算整体转化率
cr_a_total = group_a.mean()
cr_b_total = group_b.mean()

# 分用户类型计算转化率
cr_a_new = a_new_users.mean()
cr_a_old = a_old_users.mean()
cr_b_new = b_new_users.mean()
cr_b_old = b_old_users.mean()

# 创建展示结果的DataFrame
results_df = pd.DataFrame({
    '': ['整体', '新用户', '老用户'],
    '样本量_A': [len(group_a), len(a_new_users), len(a_old_users)],
    '转化率_A': [cr_a_total, cr_a_new, cr_a_old],
    '样本量_B': [len(group_b), len(b_new_users), len(b_old_users)],
    '转化率_B': [cr_b_total, cr_b_new, cr_b_old],
    '绝对差异_B-A': [cr_b_total - cr_a_total, cr_b_new - cr_a_new, cr_b_old - cr_a_old]
})
print(results_df.to_string(index=False))

# 进行统计检验
from statsmodels.stats.proportion import proportions_ztest
# 整体检验
count_total = [group_b.sum(), group_a.sum()]
nobs_total = [len(group_b), len(group_a)]
_, p_value_total = proportions_ztest(count_total, nobs_total, alternative='larger')
print(f"\n整体检验 P值: {p_value_total:.4f}")

# 新用户检验
count_new = [b_new_users.sum(), a_new_users.sum()]
nobs_new = [len(b_new_users), len(a_new_users)]
_, p_value_new = proportions_ztest(count_new, nobs_new, alternative='larger')
print(f"新用户检验 P值: {p_value_new:.4f}")

# 老用户检验
count_old = [b_old_users.sum(), a_old_users.sum()]
nobs_old = [len(b_old_users), len(a_old_users)]
_, p_value_old = proportions_ztest(count_old, nobs_old, alternative='larger')
print(f"老用户检验 P值: {p_value_old:.4f}")