缩尾样本回归

按指定分位点截尾本次估计副本中的数值变量，再调用所选基础模型。

核心代码

    def _fit_robustness_analysis(self, decimals: int = 3, title: str = "") -> Any:
        """在隔离的数据副本上执行单一稳健性规格，并复用现有回归估计器。"""
        robustness_type = str(self.robustness_options.get('type') or '').strip().lower()
        if robustness_type not in self.ROBUSTNESS_TYPES:
            raise ValueError(_('稳健性检验类型仅支持条件回归、缩尾样本回归或交互固定效应回归。'))

        base_model = str(self.advanced_model or 'ols').strip().lower()
        if base_model not in self.ROBUSTNESS_BASE_MODEL_LABELS:
            raise ValueError(_('稳健性检验的基础模型仅支持 OLS、固定效应、随机效应、混合效应、Logit 或 Probit。'))
        if not self.y_var or self.y_var not in self.data.columns:
            raise ValueError(_('稳健性检验必须选择有效的因变量。'))

        missing_x = [column for column in self.x_vars if column not in self.data.columns]
        if missing_x:
            raise ValueError(_('以下解释变量不存在: %(columns)s') % {'columns': ', '.join(missing_x)})

        source_rows = int(len(self.data))
        work_df = self.data.copy(deep=True)
        hidden_params: List[str] = []
        extra_absorb_vars: List[str] = []
        child_x_vars = list(self.x_vars)
        type_labels = {
            'conditional': _('条件回归'),
            'winsor': _('缩尾样本回归'),
            'interaction_fe': _('交互固定效应回归'),
        }
        metadata: Dict[str, Any] = {
            'type': robustness_type,
            'type_label': type_labels[robustness_type],
            'base_model': base_model,
            'source_sample_size': source_rows,
        }
        if robustness_type == 'conditional':
            work_df, condition_metadata, _condition_rows = self._prepare_conditional_robustness(work_df)
            metadata.update(condition_metadata)
        elif robustness_type == 'winsor':
            work_df, winsor_metadata, _winsor_rows = self._prepare_winsor_robustness(work_df)
            metadata.update(winsor_metadata)
        else:
            (
                work_df,
                interaction_metadata,
                _interaction_rows,
                extra_absorb_vars,
                interaction_dummies,
            ) = self._prepare_interaction_fe_robustness(work_df, base_model)
            metadata.update(interaction_metadata)
            if interaction_dummies:
                child_x_vars = self._unique_preserve_order(child_x_vars + interaction_dummies)
                hidden_params.extend(interaction_dummies)

        child_method = base_model
        child_fe_vars = list(self.fe_vars)
        if robustness_type == 'interaction_fe' and base_model in {'ols', 'fe'}:
            child_method = 'fe'
            child_fe_vars = self._unique_preserve_order(
                ([] if base_model == 'ols' else child_fe_vars) + extra_absorb_vars
            )
        elif robustness_type == 'interaction_fe' and base_model in {'logit', 'probit'}:
            child_fe_vars = self._unique_preserve_order(child_fe_vars + extra_absorb_vars)

        if base_model in {'logit', 'probit'}:
            numeric_y = pd.to_numeric(work_df[self.y_var], errors='coerce')
            valid_y = numeric_y.dropna()
            if valid_y.empty or numeric_y.notna().sum() != work_df[self.y_var].notna().sum():
                raise ValueError(_('Logit/Probit 的因变量必须是 0/1 数值变量。'))
            if not valid_y.isin([0, 1]).all():
                raise ValueError(_('Logit/Probit 的因变量必须是 0/1 二元变量。'))

        child = RegressionAnalysis(
            work_df,
            y_var=self.y_var,
            x_vars=child_x_vars,
            method=child_method,
            panel_ids=dict(self.panel_ids),
            fe_vars=child_fe_vars,
            se_options=dict(self.se_options),
            margeff_options=dict(self.margeff_options),
            control_vars=list(self.options.get('control_vars', [])),
        )
        result = child.fit(decimals=decimals, title=title)

        params = self._result_series(result, 'params')
        std_errors = self._result_series(result, 'std_errors', 'bse')
        test_stats = self._result_series(result, 'tstats', 'tvalues', 'zvalues')
        pvalues = self._result_series(result, 'pvalues')

        actual_sample_size = int(child.model_stats.get('N', getattr(result, 'nobs', 0)) or 0)
        metadata['actual_sample_size'] = actual_sample_size
        child_custom_rows = list(child.table_custom_rows)
        if extra_absorb_vars:
            child_custom_rows = [
                row for row in child_custom_rows
                if not any(
                    absorb_var in str(row.get('label', ''))
                    for absorb_var in extra_absorb_vars
                )
            ]
        # 表格底栏与普通回归保持一致：模型细节继续保存在 metadata 中，
        # 仅展示基础模型实际生成的固定效应标记，避免重复展示规格和样本量。
        custom_rows = child_custom_rows

        result_label = _('%(robustness)s（%(model)s）') % {
            'robustness': type_labels[robustness_type],
            'model': str(self.ROBUSTNESS_BASE_MODEL_LABELS.get(base_model, base_model.upper())),
        }
        self._result_model_snapshots = []
        self._store_result_model_snapshot(
            method='robustness',
            method_label=result_label,
            y_name=self.y_var,
            x_vars=list(self.x_vars),
            params=params,
            std_errors=std_errors,
            test_stats=test_stats,
            pvalues=pvalues,
            stats=dict(child.model_stats),
            custom_rows=custom_rows,
            hidden_params=hidden_params,
        )

        self.result = result
        self.model_stats = dict(child.model_stats)
        self.marginal_effects_result = child.marginal_effects_result
        self.raw_output = child.raw_output or self._result_raw_output(result)
        self.display_x_vars = list(self.x_vars)
        self.table_custom_rows = custom_rows
        self.method_label = result_label
        self.robustness_metadata = metadata
        return result