Causal Cohort Analysis — 因果队列分析：促销干预的长期用户行为效应

1. 解决的问题

传统队列分析（Cohort Analysis）追踪同一时期加入的用户群体的行为轨迹，但无法剥离选择偏差：高价值用户本来就会复购，促销只是"锦上添花"而非真正驱动因素。直接比较"接受促销的队列 vs 未接受促销的队列"，会高估促销效果 30-60%。

2. 核心算法逻辑

核心问题：传统队列分析的混淆陷阱

3. 业务应用场景

业务问题：平台推出"购买3件免费送1件"促销，GMV 短期上涨 22%。但运营团队对 6 个月后的真实 LTV 效果存疑：这批用户是真的因为促销被激活了，还是本来就会复购的高价值用户？

应用流程： 1. 队列划分：将2025年10月接受促销的用户定为处理队列，选择2025年9月（同期购买但未接受促销）作为对照队列 2. 结果变量：月度 LTV（月均消费金额），追踪促销后6个月 3. 平行趋势检验：检查两个队列在2025年7-9月的 LTV 趋势是否平行 4. DiD 估计：计算 ATT，按月分解动态效应

预期产出： - 真实因果效应：ATT ≈ 促销使 6 月 LTV 提升 12%（而非表观的 22%） - 动态效应图：第1月效应最强（+28%），第6月衰减至+8% - 识别 sleeping dogs：约 15% 的高价值老用户接受促销后 LTV 反而下降（价格锚点降低）

4. 输入数据要求

请查看原始代码模板获取输入规格。

5. 输出结果

请查看原始代码模板获取输出规格。

6. 业务价值 / ROI

• 易处：纯 Python 实现，无外部依赖；DiD 逻辑清晰
• 难处：平行趋势假设验证需要业务判断；小样本 SCM 权重求解需调参
• 前提：至少 2 个月的干预前历史数据；对照组选择需业务知识

7. 代码模板

代码块数量：1 · 路径：paper2skills-code/causal_inference/causal_cohort_analysis

"""
Causal Cohort Analysis — 因果队列分析
用 DiD + Synthetic Control 评估促销干预对长期 LTV/留存的因果效应

纯 Python 标准库 + math/statistics，无 sklearn/pandas 依赖
Python 3.14 兼容
"""
from __future__ import annotations

import math
import statistics
from dataclasses import dataclass, field


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

@dataclass
class CohortRecord:
    """单个用户的队列观测记录"""
    user_id: str
    cohort_month: str          # e.g. "2025-10"
    treated: bool              # 是否接受促销干预
    pre_metric: float          # 干预前指标（如月均 LTV）
    post_metrics: list[float]  # 干预后各月指标，post_metrics[0]=第1月，依此类推


@dataclass
class ATTResult:
    """ATT 估计结果"""
    period: int                # 干预后第几个月
    att: float                 # Average Treatment Effect on the Treated
    se: float                  # 标准误（Bootstrap 估计）
    ci_lower: float
    ci_upper: float
    n_treated: int
    n_control: int


# ─── DiD 队列分析器 ──────────────────────────────────────────────────────────

class DiDCohortAnalyzer:
    """
    双重差分因果队列分析器

    假设：平行趋势（pre-period 趋势相同）
    估计量：ATT = E[Y(1) - Y(0) | Treated=1]
    """

    def __init__(self, n_bootstrap: int = 200, alpha: float = 0.05):
        self.n_bootstrap = n_bootstrap
        self.alpha = alpha
        self._treated: list[CohortRecord] = []
        self._control: list[CohortRecord] = []
        self._n_periods: int = 0

    def fit(self, records: list[CohortRecord]) -> "DiDCohortAnalyzer":
        """加载数据，分离处理组/对照组"""
        self._treated = [r for r in records if r.treated]
        self._control = [r for r in records if not r.treated]
        if not self._treated or not self._control:

8. 论文来源

未自动抽取；请查看原始 Skill 卡片。