协整检验

EG 与 Johansen 协整检验。

核心代码

    def _fit_cointegration(self, decimals: int, title: str) -> None:
        """
        执行协整检验。
        支持两种方法：
        1. Engle-Granger 两步法（适用于两变量）
        2. Johansen 检验（适用于多变量）
        用户可选择执行其中一种或同时执行两种。
        """
        target_cols = self.x_vars
        if not target_cols or len(target_cols) < 2:
            raise ValueError(_('协整检验至少需要选择 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 行）。'))

        coint_method = self.ts_options.get('coint_method', 'johansen')  # 'eg', 'johansen', 'both'
        det_order = int(self.ts_options.get('coint_det_order', 0))  # Johansen 确定性趋势项
        # 用户输入的是 VAR 滞后阶数（与 Stata 的 lags() 一致），后端转换为 k_ar_diff = lags - 1
        var_lags = int(self.ts_options.get('coint_var_lags', 2))
        k_ar_diff = max(var_lags - 1, 1)  # VECM 差分滞后阶数 = VAR 滞后阶数 - 1

        overall_title = title if title else _('协整检验')
        html = f'<div class="table-editable-container">'
        html += render_table_title_editor(overall_title)

        # 检验信息概要
        n_obs = df_clean.shape[0] - var_lags  # 有效观测数（与 Stata 一致）
        det_labels_info = {-1: _('无确定性趋势'), 0: _('含截距项（无趋势）'), 1: _('含截距项和线性趋势')}
        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>'
        method_label = {'eg': 'Engle-Granger', 'johansen': 'Johansen', 'both': 'Engle-Granger + Johansen'}
        info_rows = [
            (_('观测数'), str(n_obs), _('变量数'), str(len(valid_cols))),
            (_('检验方法'), method_label.get(coint_method, coint_method), _('滞后阶数'), str(var_lags)),
            (_('趋势项'), det_labels_info.get(det_order, str(det_order)), _('变量'), ', '.join(valid_cols)),
        ]
        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>'

        # Engle-Granger 检验
        if coint_method in ('eg', 'both'):
            html += self._build_eg_cointegration(df_clean, valid_cols, decimals)

        # Johansen 检验
        if coint_method in ('johansen', 'both'):
            html += self._build_johansen_cointegration(df_clean, valid_cols, det_order, k_ar_diff, var_lags, decimals)

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