在电商行业中，AB测试是优化产品决策的重要手段。然而在实际落地过程中，我们常常会遇到流量分配不均、数据统计偏差、实验污染等问题。这些问题如果处理不当，可能会导致实验结果不可靠，甚至误导产品决策。今天，我就结合一个真实电商项目案例，和大家分享如何设计高可信度的AB测试系统。

背景与痛点

AB测试看似简单，但在实际应用中却存在诸多陷阱。以下是几个典型的痛点：

1. 流量分配不均：可能导致实验组和对照组的用户特征不匹配
2. 数据一致性：埋点丢失或重复上报会影响数据准确性
3. 实验隔离：新老用户混杂可能导致辛普森悖论

技术方案设计

分流算法选择

常见的分流算法主要有两种：

1. 哈希分流：简单高效，适合大多数场景
2. 分层抽样：能保证某些关键特征的均匀分布

对于电商场景，我们最终选择了加权哈希分流算法，因为它既能保证随机性，又能灵活调整流量分配比例。

Python实现加权分流算法

def weighted_allocation(user_id: str, experiment_id: str, weights: dict) -> str:
    """
    加权分流算法实现
    :param user_id: 用户唯一标识
    :param experiment_id: 实验唯一标识
    :param weights: 分组权重，如{"control": 0.5, "test": 0.5}
    :return: 分配到的组别
    """
    try:
        # 生成分桶键
        bucket_key = f"{user_id}_{experiment_id}"
        # 计算哈希值并归一化
        hash_val = hash(bucket_key) % 10000 / 10000.0

        # 根据权重分配组别
        cumulative_weight = 0.0
        for group, weight in weights.items():
            cumulative_weight += weight
            if hash_val <= cumulative_weight:
                return group

        # 默认返回权重最大的组
        return max(weights.items(), key=lambda x: x[1])[0]
    except Exception as e:
        # 异常处理
        print(f"Allocation error: {str(e)}")
        return "control"  # 默认返回对照组

分桶键生成技巧

为了避免同一个用户在不同实验中分配到不同的组，我们使用UserID+ExperimentID作为分桶键。这样可以确保：

1. 同一个用户在同一实验中始终分配到同一组
2. 不同实验之间互不干扰

生产实践要点

流量比例设置

在Redis中使用HyperLogLog进行去重计数，可以高效统计各组的独立用户数：

import redis

r = redis.Redis()

def track_user(group: str, user_id: str):
    """
    使用HyperLogLog记录用户分组
    :param group: 组别名称
    :param user_id: 用户ID
    """
    r.pfadd(f"ab_test:{experiment_id}:{group}", user_id)

数据收集保障

为了防止埋点丢失，我们设计了补发机制：

1. 客户端缓存未发送的埋点数据
2. 定时检查网络状态，自动重试发送
3. 服务端记录接收时间，过滤过期数据

显著性检验

使用PySpark进行大规模的显著性检验：

from pyspark.sql import functions as F
from scipy import stats

# 计算p值
def calculate_p_value(df, metric_col, group_col):
    """
    计算两组指标的p值
    :param df: 包含实验数据的DataFrame
    :param metric_col: 需要比较的指标列名
    :param group_col: 分组列名
    """
    # 分组统计数据
    groups = df.groupBy(group_col).agg(
        F.mean(metric_col).alias("mean"),
        F.stddev(metric_col).alias("std"),
        F.count(metric_col).alias("count")
    ).collect()

    # 提取两组数据
    group1 = [g for g in groups if g[group_col] == "control"][0]
    group2 = [g for g in groups if g[group_col] == "test"][0]

    # 计算t检验
    t_stat, p_value = stats.ttest_ind_from_stats(
        group1["mean"], group1["std"], group1["count"],
        group2["mean"], group2["std"], group2["count"]
    )

    return p_value

避坑指南

1. 实验期间不要修改分流规则：这会破坏实验的随机性
2. 确保最小样本量：使用以下公式计算

   样本量 = 16 * (标准差/最小可检测效应)^2

3. 多实验叠加时的正交分层：确保不同实验的用户分布独立

开放性问题

在实际业务中，我们可能会遇到这样的情况：实验组的CTR(点击率)提升了，但GMV(总交易额)却下降了。这种情况下，我们应该如何决策？是相信CTR的提升最终会带来GMV增长，还是立即停止实验？这需要结合业务场景和长期数据来综合判断。

希望通过这篇分享，能帮助大家在AB测试实践中少走弯路。如果你有其他好的经验或想法，欢迎一起交流讨论！