Experiment Data Quality Guard — A/B 实验数据采集质量保障：爬虫/日志污染检测与因果实验完整性

1. 解决的问题

A/B 实验的因果推断依赖随机化的完整性：处理组（Treatment）与对照组（Control）的差异必须仅来自实验干预，而非数据采集过程的污染。

2. 核心算法逻辑

A/B 实验的因果推断依赖随机化的完整性：处理组（Treatment）与对照组（Control）的差异必须仅来自实验干预，而非数据采集过程的污染。母婴跨境电商场景中常见的污染源包括：

3. 业务应用场景

业务背景：针对婴儿推车详情页（PDP）新版信任徽章（Trust Badge）的 A/B 实验，实验周期 14 天，发现实验组 CTR 显著高于对照组（+8.3%），但转化率无显著差异，怀疑爬虫污染导致点击数虚高。

量化 ROI：避免因虚假显著性结果而错误全量上线（该改版涉及开发成本 ~15 万元 + 页面跳转改造），节省无效投入 15 万元。

业务背景：针对母乳储奶袋的首购优惠券 Push 实验，发现实验组与对照组的实验前 GMV 协变量 SMD = 0.34（远超 0.1 阈值），怀疑 SDK 批量上报导致日志乱序，部分高价值用户被错误分配。

4. 输入数据要求

• `bot_stats.bot_rate_treatment` vs `bot_stats.bot_rate_control`：差距 > 3pp 表示不均匀污染，需高度警惕
• `balance.status == "critical"`：立即停止实验，排查日志管道
• `effect.adjusted_p_value` vs `effect.raw_p_value`：对比校正前后结论
• [ ] 部署 A/A 测试（实验前 3 天），验证 SMD < 0.1
• [ ] 检查实验组/对照组 bot_rate 差值 < 2pp

5. 输出结果

• `bot_stats.bot_rate_treatment` vs `bot_stats.bot_rate_control`：差距 > 3pp 表示不均匀污染，需高度警惕
• `balance.status == "critical"`：立即停止实验，排查日志管道
• `effect.adjusted_p_value` vs `effect.raw_p_value`：对比校正前后结论
• [ ] 部署 A/A 测试（实验前 3 天），验证 SMD < 0.1
• [ ] 检查实验组/对照组 bot_rate 差值 < 2pp

6. 业务价值 / ROI

10-50 万元

7. 代码模板

代码块数量：2 · 路径：未检测到

"""
A/B 实验数据采集质量保障系统
整合爬虫过滤 + 分组均衡检验 + CUPED 偏差校正
arXiv 参考: 2309.12215 (ExP: Scalable Experimentation), 
           2405.01817 (Causal Testing with Contaminated Data)
"""

import numpy as np
import pandas as pd
from dataclasses import dataclass
from typing import Dict, List, Optional, Tuple
from scipy import stats


# ── 数据结构 ─────────────────────────────────────────────────────────────

@dataclass
class ExperimentRecord:
    user_id: str
    group: str          # "treatment" or "control"
    ua_string: str
    event_count: int
    session_duration: float   # 秒
    click_count: int
    pre_exp_gmv: float        # 实验前 GMV（CUPED 协变量）
    outcome_metric: float     # 实验期间指标（如 CTR, GMV）
    timestamp_gaps: List[float]  # 操作间隔时间列表（秒）


# ── Layer 1：爬虫/机器人检测 ───────────────────────────────────────────────

class BotDetector:
    """
    基于行为特征的爬虫检测
    三维评分：UA异常 + 时序异常 + 行为熵
    """

    KNOWN_BOT_PATTERNS = [
        "bot", "crawler", "spider", "scraper", "python-requests",
        "curl", "wget", "scrapy", "selenium", "headless",
    ]

    def __init__(self, bot_score_threshold: float = 0.6):
        self.threshold = bot_score_threshold

    def ua_score(self, ua_string: str) -> float:
        """UA 异常得分：匹配已知爬虫特征"""
        ua_lower = ua_string.lower()
        if any(pat in ua_lower for pat in self.KNOWN_BOT_PATTERNS):
            return 1.0
        # 缺少常见浏览器标识
        if not any(b in ua_lower for b in ["mozilla", "chrome", "safari", "firefox"]):
            return 0.7
        return 0.0

    def timing_score(self, timestamp_gaps: List[float]) -> float:
        """时序异常得分：机器行为 → 操作间隔方差极小"""
        if len(timestamp_gaps) < 3:
            return 0.0
        gaps = np.array(timestamp_gaps)

8. 论文来源

• 2309.12215
• 2405.01817