Data Collection Causal Debiasing — 采集偏差因果修正：爬虫选择性采集对因果分析的去污染

1. 解决的问题

某母婴品牌通过爬虫采集 Amazon 评论，分析"打折 coupon 是否提升复购率"

2. 核心算法逻辑

跨境电商数据采集面临一个根本性困境：你能采到的数据，恰恰不是随机样本。Amazon 评论爬虫只能采集到购买且写了评论的买家（约 25% 购买者），而这批人的特征（高参与度、使用经验较强、投诉阈值较低）与全体买家截然不同。当你用这批有偏数据做因果推断（如"折扣是否提升复购率"），估计结果会被选择偏差（Selection Bias）严重扭曲。

3. 业务应用场景

业务背景：某母婴品牌通过爬虫采集 Amazon 评论，分析"打折 coupon 是否提升复购率"。原始爬虫样本显示打折组复购率高出对照组 31%，但市场部基于这一结果激进投放 coupon，实际复购提升仅 8%，造成严重的 coupon 成本浪费。

量化收益： - 避免基于虚高 CATE（+0.31）多投放 coupon 预算 $82,000/季 - 实际 CATE（+0.09）仍显著 → coupon 策略保留但降低力度 - 净节省 coupon 预算 ≈ $65,000/年，同时维持真实复购提升

业务背景：为了解目标买家对价格敏感度，通过 Amazon Vine + 站内消息发放价格调查问卷，但高收入买家（对价格不敏感）应答率仅 12%，低收入买家（对价格高度敏感）应答率 31%，导致调查结果严重低估价格弹性。

4. 输入数据要求

• 平台官方人口统计报告（抽样模拟）
• 内部 CRM 全量用户数据
• 随机对照实验（A/B test）中的控制组

5. 输出结果

• 平台官方人口统计报告（抽样模拟）
• 内部 CRM 全量用户数据
• 随机对照实验（A/B test）中的控制组

6. 业务价值 / ROI

180 万

7. 代码模板

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

"""
Data Collection Causal Debiasing
整合 IPW 选择偏差修正 + 双重稳健估计 + 爬虫截断检测
CausalDebiasWeb (arXiv:2407.15392) + SurveyBiasCorrect (arXiv:2501.08734) + CrawlerBiasAudit (arXiv:2503.12640)
"""

import numpy as np
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
from scipy import stats
from sklearn.linear_model import LogisticRegression, LinearRegression
from sklearn.preprocessing import StandardScaler
import warnings
warnings.filterwarnings("ignore")


@dataclass
class BiasAuditResult:
    """偏差审计结果"""
    feature_name: str
    sample_mean: float
    population_mean: float
    bias_ratio: float          # sample_mean / population_mean
    is_biased: bool            # |bias_ratio - 1| > 0.2 视为显著偏差


@dataclass
class TruncationTestResult:
    """爬虫截断检验结果"""
    ks_stat: float
    p_value: float
    has_truncation: bool
    estimated_truncation_point: float
    extrapolation_factor: float        # 估计的样本外推因子


class CrawlerBiasAuditor:
    """
    爬虫偏差审计器
    arXiv:2503.12640 CrawlerBiasAudit
    检测爬虫系统性截断与样本分布偏差
    """

    def __init__(self, truncation_p_threshold: float = 0.05):
        self.truncation_p_threshold = truncation_p_threshold

    def audit_feature_bias(
        self,
        sample_features: np.ndarray,           # 爬虫样本特征
        population_means: Dict[str, float],     # 已知总体均值（来自平台统计）
        feature_names: List[str],
    ) -> List[BiasAuditResult]:
        """检测特征分布偏差"""
        results = []
        for i, name in enumerate(feature_names):
            if name not in population_means:
                continue
            sample_mean = float(np.mean(sample_features[:, i]))
            pop_mean = population_means[name]
            bias_ratio = sample_mean / max(abs(pop_mean), 1e-6)

8. 论文来源

• 2407.15392
• 2501.08734
• 2503.12640