随机效应模型

随机效应面板数据模型。

核心代码

    def _fit_random_effects(
        self,
        df: pd.DataFrame,
        entity_col: str,
        time_col: str
    ) -> Any:
        """
        执行随机效应回归

        参数:
            df: 清洗后的数据框
            entity_col: 个体标识列
            time_col: 时间标识列

        返回:
            RandomEffects 拟合结果
        """
        if not entity_col or not time_col:
            raise ValueError("RE Model needs Entity and Time ID")

        df, entity_alias, time_alias, _dup_info, notes = self._prepare_panel_dimensions(
            df,
            entity_col,
            time_col,
            include_other_effects=False
        )
        self._emit_preparation_notes(notes)

        # 剔除 singleton
        df = drop_singletons_func(df, entity_alias)

        # 设置面板索引
        df_panel = df.set_index([entity_alias, time_alias])
        y = df_panel[self.y_var]

        # 准备自变量（随机效应模型需要常数项）
        exog_cols = [v for v in self.x_vars if v != self.y_var]
        if not exog_cols:
            df_panel['const'] = 1
            x = df_panel[['const']]
        else:
            x = add_constant(df_panel[exog_cols])

        # 创建随机效应模型
        try:
            mod = RandomEffects(y, x)
        except ValueError as exc:
            if self._is_rank_error(exc):
                raise ValueError(self._build_rank_error_message(_('随机效应回归'), x)) from exc
            raise

        # 配置标准误
        se_type = self.se_options.get('type')
        fit_kwargs = {}

        if se_type == 'robust':
            fit_kwargs['cov_type'] = 'robust'
        elif se_type == 'cluster':
            fit_kwargs['cov_type'] = 'clustered'
            c_var = self.se_options.get('cluster_var')
            if c_var == entity_col:
                fit_kwargs['cluster_entity'] = True
            elif c_var == time_col:
                fit_kwargs['cluster_time'] = True
            else:
                fit_kwargs['clusters'] = df_panel[c_var]
        else:
            fit_kwargs['cov_type'] = 'unadjusted'

        # 拟合模型
        self.result = mod.fit(**fit_kwargs)

        # 计算统计量
        r2_overall = self.result.rsquared_overall
        nobs = self.result.nobs
        df_resid = self.result.df_resid
        adj_r2 = 1 - (1 - r2_overall) * (nobs - 1) / df_resid if df_resid > 0 else np.nan

        self.model_stats = {
            'N': int(nobs),
            'R2': r2_overall,
            'Adj-R2': adj_r2,
            'F': self.result.f_statistic.stat
        }

        return self.result