格兰杰因果检验

格兰杰因果关系检验。

核心代码

    def _fit_granger(self, decimals: int, title: str) -> None:
        """
        执行独立的 Granger 因果检验。
        基于 VAR 模型框架，对所有变量对进行双向 Granger 因果检验。
        支持差分处理。
        """
        from statsmodels.tsa.api import VAR

        target_cols = self.x_vars
        if not target_cols or len(target_cols) < 2:
            raise ValueError(_('Granger 因果检验至少需要选择 2 个变量。'))

        valid_cols = [c for c in target_cols if c in self.data.columns]
        if len(valid_cols) < 2:
            raise ValueError(_('有效的数值型变量不足 2 个。'))

        df_clean = self.data[valid_cols].select_dtypes(include=[np.number]).dropna()
        if df_clean.shape[0] < 20:
            raise ValueError(_('有效数据行数不足（至少需要 20 行）。'))

        max_lags = self.ts_options.get('granger_max_lags', 4)
        if max_lags is not None:
            max_lags = int(max_lags)
        ic = self.ts_options.get('granger_ic', 'aic')

        # 差分处理
        diff_order = int(self.ts_options.get('granger_diff_order', 0))
        if diff_order > 0:
            for _step in range(diff_order):
                df_clean = df_clean.diff().dropna()
            if df_clean.shape[0] < 20:
                raise ValueError(_('经过 %(diff_order)s 阶差分后数据行数不足（至少需要 20 行）。') % {'diff_order': diff_order})

        # 构建 VAR 模型以确定最优滞后阶数
        model = VAR(df_clean)

        try:
            lag_order_result = model.select_order(maxlags=max_lags)
        except Exception as e:
            raise ValueError(_('滞后阶数选择失败: %(error)s') % {'error': str(e)})

        ic_map = {'aic': 'aic', 'bic': 'bic', 'hqic': 'hqic', 'fpe': 'fpe'}
        selected_ic = ic_map.get(ic, 'aic')
        optimal_lag = getattr(lag_order_result, selected_ic, None)
        if optimal_lag is None or optimal_lag == 0:
            optimal_lag = 1

        # 拟合 VAR 模型
        try:
            var_result = model.fit(maxlags=optimal_lag, trend='c')
        except Exception as e:
            raise ValueError(_('VAR 模型拟合失败: %(error)s') % {'error': str(e)})

        # 生成 HTML
        diff_note = _('（%(diff_order)s阶差分）') % {'diff_order': diff_order} if diff_order > 0 else ""
        overall_title = title if title else _('Granger 因果检验%(diff_note)s') % {'diff_note': diff_note}
        html = f'<div class="table-editable-container">'
        html += render_table_title_editor(overall_title)

        # 检验信息概要
        html += '<table class="academic-table" style="width: auto; margin: 10px auto; min-width: 60%;">'
        html += '<thead><tr>'
        html += _('<th style="border-bottom: 1px solid black; text-align: center;" colspan="4">检验信息</th>')
        html += '</tr></thead><tbody>'
        diff_label = _('%(diff_order)s阶差分') % {'diff_order': diff_order} if diff_order > 0 else _('原序列（未差分）')
        info_rows = [
            (_('观测数'), str(var_result.nobs), _('变量数'), str(len(valid_cols))),
            (_('滞后阶数'), _('%(lag)s（%(ic)s准则）') % {'lag': optimal_lag, 'ic': selected_ic.upper()}, _('差分阶数'), diff_label),
        ]
        for r in info_rows:
            html += '<tr>'
            html += f'<td style="text-align: center; font-weight: bold; width: 20%;">{r[0]}</td>'
            html += f'<td style="text-align: center; width: 30%;">{r[1]}</td>'
            html += f'<td style="text-align: center; font-weight: bold; width: 20%;">{r[2]}</td>'
            html += f'<td style="text-align: center; width: 30%;">{r[3]}</td>'
            html += '</tr>'
        html += '</tbody></table>'

        # Granger 因果检验结果表
        html += self._build_var_granger_table(var_result, valid_cols, optimal_lag, decimals)

        html += '</div>'
        self.custom_html = html