PSM-DID

倾向得分匹配双重差分。

核心代码

    def _fit_psm_did(self, decimals: int, title: str) -> Any:
        if not self.y_var:
            raise ValueError(_('PSM 匹配需要选择因变量'))

        treat_var = self.did_options.get('treat_var')
        if not treat_var:
            raise ValueError(_('PSM 匹配需要选择处理组变量'))

        covariates = self._get_control_vars(exclude=[treat_var])
        if not covariates:
            raise ValueError(_('PSM 匹配至少需要选择一个匹配协变量'))

        regression_spec = self._get_psm_regression_spec(treat_var, covariates)
        psm_df = self._prepare_psm_frame(treat_var, covariates, regression_spec)
        psm_df[treat_var] = (pd.to_numeric(psm_df[treat_var], errors='coerce') > 0).astype(int)
        if psm_df[treat_var].nunique() != 2:
            raise ValueError(_('处理组变量必须同时包含处理组和控制组'))

        psm_df, score_result, support_note = self._estimate_propensity_scores(psm_df, treat_var, covariates)
        match_result = self._run_psm_matching(psm_df, treat_var, self.y_var)
        psm_df['__psm_support__'] = match_result['support'].astype(int)
        psm_df['__psm_weight__'] = match_result['weights']
        psm_df['__psm_matched_outcome__'] = match_result['matched_outcome']
        psm_df['__psm_regression_sample__'] = match_result['regression_sample'].astype(int)

        effect_df, effect_summary = self._build_psmatch2_effect_table(psm_df, treat_var, self.y_var, match_result)
        balance_df, balance_summary_df = self._build_pstest_tables(psm_df, treat_var, covariates, match_result)

        regression_result, regression_df, regression_x_vars = self._fit_psm_regression(psm_df, regression_spec, match_result)
        self.result = regression_result
        self._capture_panel_result(regression_result)
        self._set_model_stats(regression_result)
        self.model_stats['N'] = int(len(regression_df))
        self.display_x_vars = regression_x_vars

        method = self.did_options.get('psm_matching_method', 'neighbor')
        header = title or _('PSM 匹配与回归')
        summary_df = pd.DataFrame([
            {'项目': _('匹配方法'), '取值': self._matching_method_label(method)},
            {'项目': _('匹配效应口径'), '取值': self._psm_estimand_label()},
            {'项目': _('倾向得分模型'), '取值': self._psm_score_model_label()},
            {'项目': _('匹配尺度'), '取值': self._psm_score_scale_label()},
            {'项目': _('原始样本量'), '取值': int(len(psm_df))},
            {'项目': _('最终共同支撑域样本量'), '取值': int(match_result['support'].sum())},
            {'项目': _('匹配后原始回归样本量'), '取值': int(match_result['regression_sample'].sum())},
            {'项目': _('匹配后回归估计样本量'), '取值': int(self.model_stats.get('N', 0) or 0)},
            {'项目': _('匹配后处理组样本量'), '取值': int(((psm_df[treat_var] == 1) & match_result['regression_sample']).sum())},
            {'项目': _('匹配后控制组样本量'), '取值': int(((psm_df[treat_var] == 0) & match_result['regression_sample']).sum())},
            {'项目': _('共同支撑域区间'), '取值': support_note},
        ])

        self.diagnostic_html = (
            self._render_dataframe_table(summary_df, _('%(header)s - 匹配摘要') % {'header': header}, decimals=decimals)
            + self._render_dataframe_table(effect_df, _('%(header)s - psmatch2 效果摘要') % {'header': header}, decimals=decimals)
            + self._render_dataframe_table(balance_summary_df, _('%(header)s - pstest 总体摘要') % {'header': header}, decimals=decimals)
            + self._render_dataframe_table(balance_df, _('%(header)s - pstest 协变量平衡') % {'header': header}, decimals=decimals)
        )

        self.chart_images = []
        selected_charts = []
        if self._as_bool(self.did_options.get('psm_export_balance_chart', True)):
            selected_charts.append(self._build_pstest_bias_chart(balance_df, header))
        if self._as_bool(self.did_options.get('psm_export_support_chart', True)):
            selected_charts.append(self._build_support_chart(psm_df, treat_var, header))
        if self._as_bool(self.did_options.get('psm_export_density_chart', True)):
            selected_charts.extend([
                self._build_density_chart(psm_df, treat_var, f'{header} - ?????????', matched=False),
                self._build_density_chart(psm_df, treat_var, f'{header} - ?????????', matched=True),
            ])
        for chart in selected_charts:
            if chart:
                self.chart_images.append(chart)

        fixed_effects_label = self._format_fixed_effects_label(regression_spec['fe_vars'])
        self.table_custom_rows = [
            {'label': _('匹配方法'), 'value': self._matching_method_label(method)},
            {'label': _('匹配效应'), 'value': self._psm_estimand_label()},
            {'label': _('固定效应'), 'value': fixed_effects_label},
            {'label': _('共同支撑域'), 'value': _('是') if self._as_bool(self.did_options.get('psm_common_support', True)) else _('否')},
        ]

        score_summary_text = self._result_summary_text(score_result['result'])
        regression_summary_text = self._result_summary_text(regression_result)
        self.raw_output = (
            _('PSM 倾向得分模型\n')
            + score_summary_text
            + _('\n\npsmatch2 效果摘要\n')
            + effect_df.to_string(index=False)
            + _('\n\npstest 总体摘要\n')
            + balance_summary_df.to_string(index=False)
            + _('\n\npstest 协变量平衡\n')
            + balance_df.to_string(index=False)
            + _('\n\n匹配后回归\n')
            + regression_summary_text
        )

        self.model_stats['Matched-N'] = int(match_result['regression_sample'].sum())
        return self.result