Skip to content

安慰剂检验

时间安慰剂检验(亦支持随机安慰剂)。

核心代码

py
    def _fit_time_placebo(self, decimals: int, title: str) -> Any:
        if not self.y_var:
            raise ValueError(_('时间安慰剂检验需要选择因变量'))

        treat_var = self.did_options.get('treat_var')
        post_var = self.did_options.get('post_var')
        entity_col, time_col = self._get_panel_columns()
        if not treat_var:
            raise ValueError(_('时间安慰剂检验需要选择处理变量或 DID 变量'))
        if not post_var:
            raise ValueError(_('时间安慰剂检验需要选择政策后虚拟变量 post'))

        shift = max(int(self.did_options.get('placebo_shift', 1)), 1)
        controls = self._get_control_vars(exclude=[treat_var, post_var, entity_col, time_col])
        absorb_vars = self._get_placebo_absorb_vars(entity_col, time_col)
        cols = [self.y_var, treat_var, post_var, entity_col, time_col] + controls + self._cluster_cols()
        df = self._numeric_clean_frame(cols, numeric_cols=[self.y_var, treat_var, post_var] + controls)
        df = self._add_panel_aliases(df, entity_col, time_col)
        df['__did_event_time_base__'], note = self._event_time_numeric(df[time_col])
        if note:
            warnings.warn(note, UserWarning)

        post_series = pd.to_numeric(df[post_var], errors='coerce').fillna(0.0) > 0
        if not post_series.any():
            raise ValueError(_('post 变量中没有识别到政策后时期(取值应包含 1 或正数)'))
        policy_time = float(df.loc[post_series, '__did_event_time_base__'].min())
        placebo_policy_time = policy_time - shift

        treat_numeric = pd.to_numeric(df[treat_var], errors='coerce').fillna(0.0)
        ever_treated = df.groupby('__did_entity__')[treat_var].transform(lambda s: float(pd.to_numeric(s, errors='coerce').fillna(0.0).max() > 0))
        treat_varies = df.groupby('__did_entity__')[treat_var].transform(lambda s: pd.to_numeric(s, errors='coerce').fillna(0.0).nunique()).max() > 1

        if treat_varies:
            df['__did_real__'] = treat_numeric
        else:
            df['__did_real__'] = ever_treated * post_series.astype(float)

        df['__did_placebo_time__'] = (
            ever_treated
            * (df['__did_event_time_base__'] >= placebo_policy_time).astype(float)
            * (df['__did_event_time_base__'] < policy_time).astype(float)
        )
        pre_df = df[df['__did_event_time_base__'] < policy_time].copy()
        if pre_df['__did_placebo_time__'].sum() <= 0:
            raise ValueError(_('虚构政策期没有生成有效的安慰剂处理变量,请减小前移期数或检查时间变量'))

        real_result, _unused_a, _unused_b = self._fit_placebo_regression(
            df=df,
            y_col=self.y_var,
            x_cols=['__did_real__'] + controls,
            entity_col=entity_col,
            time_col=time_col,
            absorb_vars=absorb_vars,
        )
        placebo_result, _unused_a, _unused_b = self._fit_placebo_regression(
            df=pre_df,
            y_col=self.y_var,
            x_cols=['__did_placebo_time__'] + controls,
            entity_col=entity_col,
            time_col=time_col,
            absorb_vars=absorb_vars,
        )

        self.result = placebo_result
        self._param_label_map['__did_placebo_time__'] = 'Placebo DID'
        self._capture_panel_result(placebo_result)
        self._set_model_stats(placebo_result)
        self.display_x_vars = ['Placebo DID'] + controls
        self.table_custom_rows = [
            {'label': 'Fixed effects', 'value': self._format_fixed_effects_label(absorb_vars)},
            {'label': 'Placebo type', 'value': _('时间前移 %(shift)s 期') % {'shift': shift}},
        ]

        diagnostic_df = pd.DataFrame([
            self._coef_summary_row(_('真实 DID'), real_result, '__did_real__', decimals),
            self._coef_summary_row(_('时间安慰剂 DID'), placebo_result, '__did_placebo_time__', decimals),
        ])
        placebo_p = self._safe_series_get(placebo_result.pvalues, '__did_placebo_time__')
        verdict = _('通过:安慰剂 DID 不显著') if np.isfinite(placebo_p) and placebo_p >= 0.1 else _('需关注:安慰剂 DID 显著')
        summary_df = pd.DataFrame([
            {'项目': _('真实政策期'), '取值': policy_time},
            {'项目': _('虚构政策期'), '取值': placebo_policy_time},
            {'项目': _('真实政策前样本量'), '取值': int(len(pre_df))},
            {'项目': _('判定'), '取值': verdict},
        ])
        header = title or _('时间安慰剂检验')
        self.diagnostic_html = (
            self._render_dataframe_table(summary_df, _('%(header)s - 设定摘要') % {'header': header}, decimals=decimals)
            + self._render_dataframe_table(diagnostic_df, _('%(header)s - 系数对比') % {'header': header}, decimals=decimals)
        )
        self.raw_output = (
            _('真实 DID 回归\n')
            + str(real_result.summary)
            + _('\n\n时间安慰剂回归\n')
            + str(placebo_result.summary)
        )
        return self.result

Released under the AGPL-3.0 License.