[数据分析]实验设计的基石：如何科学地划分流量？

import hashlib
from typing import Dict, List, Any

class TrafficDivertor:
    """
    一个支持分层与互斥的确定性流量划分器。
    使用SHA-256哈希函数确保分配的一致性和均匀性。
    """

    def __init__(self):
        """
        初始化一个空的层配置字典。
        配置格式: {
            'layer_name': {
                'salt': 'unique_salt_for_layer',
                'buckets': [('exp1_name', 0.3), ('exp2_name', 0.3), ...]
            }
        }
        """
        self.layer_configs = {}

    def setup_layer(self, layer_name: str, salt: str, buckets: List[tuple]):
        """
        设置一个层的配置。

        :param layer_name: 层的名称（如'ui_layer'）
        :param salt: 该层使用的唯一盐值
        :param buckets: 一个包含(实验名, 流量比例)元组的列表。
                        比例总和必须小于等于1，剩余部分自动归为对照组('control')。
        """
        total_weight = sum(weight for _, weight in buckets)
        if total_weight > 1.0:
            raise ValueError(f"层 {layer_name} 的流量比例总和({total_weight})不能大于1.0")

        # 存储配置
        self.layer_configs[layer_name] = {
            'salt': salt,
            'buckets': buckets
        }

    def _get_bucket_assignment(self, unit_id: Any, salt: str) -> float:
        """
        核心方法：将单元ID和盐值结合，哈希后归一化为[0, 1)之间的浮点数。

        :param unit_id: 划分单元（如用户ID）
        :param salt: 盐值
        :return: 归一化的哈希值（桶值）
        """
        # 将单元ID转换为字符串，并与盐值拼接
        combined_str = f"{unit_id}{salt}".encode('utf-8')

        # 使用SHA-256计算哈希值
        hash_obj = hashlib.sha256(combined_str)
        hash_hex = hash_obj.hexdigest()

        # 取哈希值的前8个字符（4字节），将其转换为0到1之间的浮点数
        # 2^32 = 4294967296
        hash_int = int(hash_hex[:8], 16)
        normalized_value = hash_int / 4294967296.0

        return normalized_value

    def assign(self, unit_id: Any, layer_name: str) -> str:
        """
        为给定的单元ID在指定层中分配一个实验组。

        :param unit_id: 划分单元（如用户ID）
        :param layer_name: 层名称
        :return: 分配到的实验名称。如果落在所有定义的范围之外，则返回'control'。
        """
        if layer_name not in self.layer_configs:
            raise ValueError(f"层 {layer_name} 未被配置")

        config = self.layer_configs[layer_name]
        salt = config['salt']
        buckets = config['buckets']

        # 获取确定的桶值
        bucket_value = self._get_bucket_assignment(unit_id, salt)

        # 遍历该层的所有桶配置，判断bucket_value落在哪个区间
        current_cutoff = 0.0
        for exp_name, weight in buckets:
            if current_cutoff <= bucket_value < current_cutoff + weight:
                return exp_name
            current_cutoff += weight

        # 如果落在所有定义的实验范围之外，则分配为对照组
        return 'control'

# 1. 实例化流量划分器
divertor = TrafficDivertor()

# 2. 配置两个实验层
# 层1: UI实验层
divertor.setup_layer(
    layer_name='ui_layer',
    salt='my_ui_salt_20231027', # 唯一盐值
    buckets=[
        ('new_button_color', 0.3),   # 30%流量给新按钮颜色实验
        ('new_layout', 0.3)          # 30%流量给新布局实验
        # 剩余40%流量自动成为对照组
    ]
)

# 层2: 推荐算法层
divertor.setup_layer(
    layer_name='algo_layer',
    salt='my_algo_salt_20231027', # 另一个唯一盐值
    buckets=[
        ('new_ranking_v1', 0.2),    # 20%流量给新排序算法V1
        ('new_ranking_v2', 0.2)     # 20%流量给新排序算法V2
        # 剩余60%流量自动成为对照组
    ]
)

# 3. 模拟一批用户，测试他们的分配结果
test_user_ids = [f"user_{i}" for i in range(1, 11)] # 生成10个测试用户ID
assignments = []

for user_id in test_user_ids:
    # 为每个用户在两个层中进行分配
    ui_assignment = divertor.assign(user_id, 'ui_layer')
    algo_assignment = divertor.assign(user_id, 'algo_layer')
    assignments.append({
        'user_id': user_id,
        'ui_layer': ui_assignment,
        'algo_layer': algo_assignment
    })

# 将结果转换为DataFrame以便查看
import pandas as pd
df_assignments = pd.DataFrame(assignments)
print("10个用户的分配情况模拟:")
print(df_assignments)

# 4. 验证分配的均匀性（模拟更大样本量）
print("\n验证分配的均匀性 (模拟10000个用户):")
large_sample_ids = [f"user_{i}" for i in range(1, 10001)]

ui_assignments_large = [divertor.assign(uid, 'ui_layer') for uid in large_sample_ids]
algo_assignments_large = [divertor.assign(uid, 'algo_layer') for uid in large_sample_ids]

# 计算各实验组的比例
from collections import Counter
ui_counts = Counter(ui_assignments_large)
algo_counts = Counter(algo_assignments_large)

print("\nUI层分配比例:")
for exp, count in ui_counts.items():
    proportion = count / len(large_sample_ids)
    print(f"  {exp}: {proportion:.3%} (期望: {'30%' if exp.startswith('new') else '40%' if exp=='control' else 'N/A'})")

print("\n算法层分配比例:")
for exp, count in algo_counts.items():
    proportion = count / len(large_sample_ids)
    print(f"  {exp}: {proportion:.3%} (期望: {'20%' if exp.startswith('new') else '60%' if exp=='control' else 'N/A'})")

10个用户的分配情况模拟:
   user_id       ui_layer   algo_layer
0   user_1       control  new_ranking_v2
1   user_2  new_layout       control
2   user_3  new_button_color  new_ranking_v1
...

UI层分配比例:
  control: 39.950% (期望: 40%)
  new_button_color: 30.120% (期望: 30%)
  new_layout: 29.930% (期望: 30%)

算法层分配比例:
  control: 60.100% (期望: 60%)
  new_ranking_v1: 19.890% (期望: 20%)
  new_ranking_v2: 20.010% (期望: 20%)

def is_in_domain(unit_id: Any, domain_salt: str, domain_percentage: float) -> bool:
    """
    判断一个单元是否在指定的域内。

    :param unit_id: 单元ID
    :param domain_salt: 域的唯一盐值
    :param domain_percentage: 域的流量比例
    :return: 如果在域内返回True，否则False
    """
    # 使用相同的哈希归一化方法
    divertor = TrafficDivertor()
    bucket_value = divertor._get_bucket_assignment(unit_id, domain_salt)
    return bucket_value < domain_percentage

# 在分配前先判断域
user_id = "user_123"
domain_salt = "domain_login_users"
if is_in_domain(user_id, domain_salt, 0.8): # 80%的登录用户域
    assignment = divertor.assign(user_id, 'ui_layer')
    # ... 进行实验曝光
else:
    assignment = 'control' # 不在域内，直接返回控制组