Skip to content

单位根检验

ADF / PP 平稳性检验。

核心代码

py
    def _fit_stationarity(self, decimals: int, title: str) -> None:
        """
        执行平稳性检验(ADF、可选PP),按 Stata 风格输出 (c, t, k) 结果。
        所有变量的结果合并在同一张表中。
        若原序列不平稳,自动差分后在下一行继续输出。
        """
        target_cols = self.x_vars
        if not target_cols:
            raise ValueError(_('未选择任何变量。'))

        valid_cols = [c for c in target_cols if c in self.data.columns]
        df_clean = self.data[valid_cols].select_dtypes(include=[np.number]).dropna()

        if df_clean.empty:
            raise ValueError(_('有效数据为空,请确保选择了数值型变量。'))

        max_diff = self.ts_options.get('max_diff', 3)
        max_lags = self.ts_options.get('max_lags')
        if max_lags is not None:
            max_lags = int(max_lags)

        include_pp = self.ts_options.get('include_pp', True)
        # 用户选择输出哪些回归类型组合
        # 'all' = 三种都输出, 'ct' = 仅c=1,t=1, 'c' = 仅c=1,t=0, 'n' = 仅c=0,t=0
        output_specs = self.ts_options.get('output_specs', 'all')

        # 确定要输出的回归类型
        all_specs = [
            ('ct', '1', '1'),  # c=1, t=1
            ('c', '1', '0'),   # c=1, t=0
            ('n', '0', '0'),   # c=0, t=0
        ]
        if output_specs == 'ct':
            specs = [all_specs[0]]
        elif output_specs == 'c':
            specs = [all_specs[1]]
        elif output_specs == 'n':
            specs = [all_specs[2]]
        else:
            specs = all_specs

        # 收集所有行数据
        adf_rows = []
        pp_rows = []

        for col in df_clean.columns:
            series = df_clean[col].values
            diff_order = 0

            while diff_order <= max_diff:
                if diff_order > 0:
                    series = np.diff(series)
                    series = series[~np.isnan(series)]

                if len(series) < 10:
                    break

                var_label = col if diff_order == 0 else f"D{diff_order}.{col}"

                # ADF 检验
                adf_spec_rows = self._run_adf_specs(series, specs, max_lags, decimals)
                for row in adf_spec_rows:
                    row['var_label'] = var_label
                adf_rows.extend(adf_spec_rows)

                # PP 检验
                if include_pp:
                    pp_spec_rows = self._run_pp_specs(series, specs, decimals)
                    for row in pp_spec_rows:
                        row['var_label'] = var_label
                    pp_rows.extend(pp_spec_rows)

                # 判断是否平稳
                is_stationary = any(r['pvalue_num'] < 0.05 for r in adf_spec_rows)
                if include_pp:
                    is_stationary = is_stationary or any(r['pvalue_num'] < 0.05 for r in pp_spec_rows)

                # 在最后一行标记平稳性
                if adf_spec_rows:
                    adf_spec_rows[-1]['is_last'] = True
                    adf_spec_rows[-1]['is_stationary'] = is_stationary
                if include_pp and pp_spec_rows:
                    pp_spec_rows[-1]['is_last'] = True
                    pp_spec_rows[-1]['is_stationary'] = is_stationary

                if is_stationary:
                    break

                diff_order += 1

        # 生成 HTML
        overall_title = title if title else _('平稳性检验 (Stationarity Tests)')
        html = f'<div class="table-editable-container">'
        html += render_table_title_editor(overall_title)

        # ADF 表
        html += self._build_merged_table(_('ADF 检验'), adf_rows, specs, decimals, is_adf=True)

        # PP 表
        if include_pp and pp_rows:
            html += self._build_merged_table(_('PP 检验'), pp_rows, specs, decimals, is_adf=False)

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

Released under the AGPL-3.0 License.