Instrumental Variables (IV) for Causal Inference with Endogeneity

1. 解决的问题

Momcozy 想知道自己产品的价格弹性——价格下降10%，销量会增加多少

2. 核心算法逻辑

核心思想：当解释变量（如价格）与误差项相关时（内生性），直接回归会产生偏误。工具变量法是找一个"只通过影响解释变量来影响结果"的外生变量（工具变量），用它的变异来剥离出解释变量的"干净"部分，从而估计真实的因果效应。

3. 业务应用场景

业务问题：Momcozy 想知道自己产品的价格弹性——价格下降10%，销量会增加多少？但简单回归价格对销量的系数不是弹性，因为价格本身由需求决定（内生性）。

工具变量选择： - 竞争对手价格：竞品涨价会迫使本品调整价格，但竞品价格不直接影响本品销量（排他性） - 汇率波动：目标市场货币对美元的汇率波动影响采购成本，进而影响定价，但汇率不直接影响消费者需求 - 原材料成本冲击：如芯片短缺导致吸奶器电子元件成本上涨

应用流程： 1. 第一阶段：用竞争对手价格 + 汇率预测本品实际售价 $$Price_i = \pi_0 + \pi_1 \cdot CompetitorPrice_i + \pi_2 \cdot ExchangeRate_i + \epsilon_i$$ 2. 检验工具变量强度：计算第一阶段F统计量 3. 第二阶段：用预测价格预测销量 $$Sales_i = \beta_0 + \beta_1 \cdot \hat{Price}_i + \gamma \cdot Controls_i + u_i$$ 4. 计算价格弹性：$\epsilon = \beta_1 \cdot \frac{\bar

4. 输入数据要求

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

5. 输出结果

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

6. 业务价值 / ROI

• 数据要求：需要找到合理的工具变量（最大的挑战）
• 技术门槛：中等，2SLS逻辑清晰但工具变量选择需要领域知识
• 主要挑战：论证排他性约束（说服团队为什么工具变量不直接影响结果）
• 刚需场景：价格弹性是定价策略的核心输入，但没有IV就无法得到因果弹性
• 方法成熟：2SLS有100年历史，实现简单
• 与现有技能互补：DiD处理时间维度，IV处理截面维度，两者覆盖电商因果推断的两大场景

7. 代码模板

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

"""
Instrumental Variables (IV) — 工具变量法
用于处理内生性问题的因果推断

支持：经典2SLS、弱工具变量检验、异质性IV-DML
"""

import numpy as np
import pandas as pd
from sklearn.linear_model import LinearRegression
from sklearn.ensemble import GradientBoostingRegressor
import warnings
warnings.filterwarnings('ignore')


# ==================== IV 核心实现 ====================

class InstrumentalVariables:
    """工具变量估计器 (2SLS)"""

    def __init__(self):
        self.first_stage_model = None
        self.second_stage_model = None
        self.iv_coef = None
        self.first_stage_f = None
        self.first_stage_r2 = None

    def fit(self, X_endog, Z_iv, Y_outcome, X_controls=None):
        """
        拟合2SLS

        Args:
            X_endog: 内生解释变量 (n_samples,)
            Z_iv: 工具变量 (n_samples, n_iv)
            Y_outcome: 结果变量 (n_samples,)
            X_controls: 控制变量 (n_samples, n_controls) — 可选
        """
        n = len(X_endog)
        X_endog = np.array(X_endog).reshape(-1, 1)
        Z_iv = np.array(Z_iv).reshape(n, -1) if Z_iv.ndim == 1 else np.array(Z_iv)
        Y_outcome = np.array(Y_outcome)

        # ========== 第一阶段：Z → X ==========
        # 特征 = [Z, controls]
        if X_controls is not None:
            X_controls = np.array(X_controls)
            if X_controls.ndim == 1:
                X_controls = X_controls.reshape(-1, 1)
            Z_first = np.column_stack([Z_iv, X_controls])
        else:
            Z_first = Z_iv

        self.first_stage_model = LinearRegression()
        self.first_stage_model.fit(Z_first, X_endog.ravel())

        # 预测 X
        X_pred = self.first_stage_model.predict(Z_first)

        # 第一阶段统计量
        self.first_stage_r2 = self.first_stage_model.score(Z_first, X_endog.ravel())

8. 论文来源

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