4、流量分配策略设计：按用户ID哈希、按时间片轮转、按分层抽样

流量分配策略是授信引擎A/B测试的核心环节，决定了实验组与对照组能否在统计意义上保持同质性。不合理的分配策略会导致样本偏差、时间效应混淆或组间不均衡，最终使实验结论失效。本节详细阐述三种主流分配策略的原理、适用场景及实现要点。

4.1 按用户ID哈希分配

按用户ID哈希是最常用的确定性分配方法。通过对用户唯一标识（如用户ID、手机号、设备ID）进行哈希运算，将用户均匀映射到不同的实验组中。同一用户在不同时间进入实验时，始终被分配到同一组，保证了用户体验的一致性。

4.1.1 实现步骤

1. 选择哈希函数：推荐使用MurmurHash3、MD5或SHA-256，避免使用简单的取模运算（如user_id % N），防止用户ID规律性导致分配偏斜。
2. 确定分桶数：通常将流量分为100或1000个桶，再按比例映射到实验组。例如：实验组分配30%流量，则取哈希值 mod 1000，结果在[0, 299]范围内的用户进入实验组。
3. 实现分配逻辑：在授信引擎入口处，对每个请求的用户ID执行哈希取模，根据结果路由到对应的实验版本。

4.1.2 优缺点对比

4.1.3 代码示例

import hashlib

def assign_group(user_id: str, total_buckets: int = 1000, experiment_ratio: float = 0.3) -> str:
    # 使用MD5哈希，取前8位转换为整数
    hash_val = int(hashlib.md5(user_id.encode()).hexdigest()[:8], 16)
    bucket = hash_val % total_buckets
    if bucket < int(total_buckets * experiment_ratio):
        return "experiment"
    else:
        return "control"

4.2 按时间片轮转分配

时间片轮转分配以固定时间窗口（如每小时、每天）为单位，将整个时间片内的所有流量全部分配给某一实验组。这种方式适用于无法实时获取用户ID的场景，或需要模拟“自然实验”效果的情况。

4.2.1 分配规则

• 固定轮转：按预定义的时间表切换，例如：周一、三、五为实验组，周二、四、六为对照组。
• 随机轮转：每个时间片开始时，随机决定该片分配给哪个组，但需保证长期比例符合设定。
• 周期长度：建议以“天”为单位，避免小时级轮转导致昼夜效应混淆。

4.2.2 适用场景

4.2.3 代码示例

import datetime

def assign_by_time_slice(experiment_start: datetime.date, cycle_days: int = 7) -> str:
    today = datetime.date.today()
    days_since_start = (today - experiment_start).days
    # 每cycle_days为一个轮转周期，前70%天数走对照组，后30%走实验组
    position_in_cycle = days_since_start % cycle_days
    if position_in_cycle < int(cycle_days * 0.7):
        return "control"
    else:
        return "experiment"

4.3 按分层抽样分配

分层抽样分配先按关键特征（如用户风险等级、新老客、渠道来源）将用户划分为若干层，然后在每一层内独立进行随机分配。这种方法能保证各实验组在关键特征上的分布与总体一致，减少组间差异，提升统计灵敏度。

4.3.1 分层变量选择

4.3.2 实现流程

1. 定义分层变量：选择与核心指标（如授信通过率、逾期率）相关的特征。
2. 划分层级：每个分层变量取若干离散值，组合成多个层。例如：风险等级(3) × 新老客(2) = 6个层。
3. 层内分配：在每个层内，使用哈希或随机数将用户分配到实验组或对照组，比例保持一致。
4. 汇总统计：各层独立计算指标后，按层权重加权汇总得到整体效果。

4.3.3 代码示例

import hashlib

def stratified_assign(user_id: str, risk_level: str, is_new_user: bool) -> str:
    # 构建分层键
    layer_key = f"{risk_level}_{'new' if is_new_user else 'old'}"
    # 在层内使用哈希分配
    hash_val = int(hashlib.md5((user_id + layer_key).encode()).hexdigest()[:8], 16)
    bucket = hash_val % 100
    if bucket < 30:  # 每层30%进入实验组
        return "experiment"
    else:
        return "control"

4.4 三种策略对比与选型建议