异质性分析

分组回归的异质性分析。

核心代码

    def _fit_heterogeneity_analysis(
        self,
        df: pd.DataFrame,
        entity_col: str,
        time_col: str,
        decimals: int,
        title: str,
    ) -> Any:
        x_var, controls = self._resolve_core_x_and_controls()
        model_kind, absorb_vars, panel_dims, base_rows = self._resolve_advanced_model_spec()
        group_slices = self._build_group_slices(df)

        self._result_model_snapshots = []
        raw_outputs: List[str] = []
        summary_rows: List[Dict[str, Any]] = []
        fitted_groups: List[Tuple[str, Dict[str, Any]]] = []

        for label, mask in group_slices:
            subset = df.loc[mask].copy()
            if len(subset) < max(len(controls) + 12, 20):
                warnings.warn(f"组别“{label}”样本量过小，已自动跳过。", UserWarning)
                continue
            try:
                fit_info = self._fit_linear_model(
                    subset,
                    self.y_var,
                    [x_var] + controls,
                    model_kind=model_kind,
                    absorb_vars=absorb_vars,
                    panel_dims=panel_dims,
                    header=_('异质性分析 - 组别 %(label)s') % {'label': label},
                )
            except ZeroDivisionError as exc:
                if model_kind == 're':
                    raise ValueError(_('异质性分析中，组别“%(label)s”的有效个体数过少，无法估计随机效应。') % {'label': label}) from exc
                raise
            self._store_result_model_snapshot(
                method='heterogeneity',
                method_label=_('组别：%(label)s') % {'label': label},
                y_name=self.y_var,
                x_vars=[x_var] + controls,
                params=fit_info['params'],
                std_errors=fit_info['std_errors'],
                test_stats=fit_info['test_stats'],
                pvalues=fit_info['pvalues'],
                stats=fit_info['stats'],
                custom_rows=base_rows + [{'label': _('分组变量'), 'value': str(self.group_var)}],
            )
            summary_rows.append({
                '组别': label,
                '样本量': fit_info['stats'].get('N', np.nan),
                _('%(x_var)s 系数') % {'x_var': x_var}: fit_info['params'].get(x_var, np.nan),
                '标准误': fit_info['std_errors'].get(x_var, np.nan),
                '统计量': fit_info['test_stats'].get(x_var, np.nan),
                'P-value': fit_info['pvalues'].get(x_var, np.nan),
            })
            fitted_groups.append((label, fit_info))
            raw_outputs.append(fit_info['raw_output'])

        if len(fitted_groups) < 2:
            raise ValueError(_('可用组别不足 2 个，无法完成异质性分析。'))

        diff_rows: List[Dict[str, Any]] = []
        if len(fitted_groups) == 2:
            left_label, left_fit = fitted_groups[0]
            right_label, right_fit = fitted_groups[1]
            left_coef = float(left_fit['params'].get(x_var, np.nan))
            right_coef = float(right_fit['params'].get(x_var, np.nan))
            left_se = float(left_fit['std_errors'].get(x_var, np.nan))
            right_se = float(right_fit['std_errors'].get(x_var, np.nan))
            diff = left_coef - right_coef
            diff_se = np.sqrt(left_se ** 2 + right_se ** 2) if np.isfinite(left_se) and np.isfinite(right_se) else np.nan
            diff_z = diff / diff_se if np.isfinite(diff_se) and diff_se > 0 else np.nan
            diff_p = 2 * (1 - norm.cdf(abs(diff_z))) if np.isfinite(diff_z) else np.nan
            diff_rows.append({
                '比较': f'{left_label} vs {right_label}',
                '系数差': diff,
                '近似标准误': diff_se,
                '近似统计量': diff_z,
                '近似 P-value': diff_p,
            })

        self.result = fitted_groups[0][1]['result']
        self.model_stats = dict(fitted_groups[0][1]['stats'])
        self.display_x_vars = [x_var] + controls
        self.table_custom_rows = base_rows
        self.diagnostic_html = (
            '<div style="margin-top: 30px;"></div>'
            + self._render_dataframe_table(pd.DataFrame(summary_rows), _('%(title)s - 分组摘要') % {'title': title or _('异质性分析')}, decimals=decimals)
            + (self._render_dataframe_table(pd.DataFrame(diff_rows), _('%(title)s - 组间系数差（近似）') % {'title': title or _('异质性分析')}, decimals=decimals) if diff_rows else '')
        )
        self.raw_output = "\n\n".join(raw_outputs)
        return self.result