主成分分析（PCA）

PCA 降维与综合得分计算。
核心代码

def build_pca_index(
    *,
    df: pd.DataFrame,
    params: dict[str, Any],
    column_descriptions: dict[str, str],
) -> tuple[pd.DataFrame, dict[str, Any], list[str], dict[str, str]]:
    """PCA 主成分综合指标。"""
    indicators = require_numeric_columns(df, normalize_string_list(params.get('columns')), label=_('指标变量'))
    new_name = ensure_new_column_absent(df, params.get('new_name') or 'pca_score')
    add_component_columns = normalize_bool(params.get('add_component_columns'), default=True)
    threshold = float(params.get('variance_threshold') or 0.8)
    if threshold <= 0 or threshold > 1:
        raise DataProcessingValidationError(_('累计解释方差阈值必须在 0 到 1 之间'))

    working, valid_columns, warnings = _prepare_indicator_frame(
        df=df,
        indicators=indicators,
        missing_strategy=str(params.get('missing_strategy') or 'drop').strip().lower(),
        standardize=normalize_bool(params.get('standardize'), default=True),
    )
    if len(valid_columns) < 1:
        raise DataProcessingValidationError(_('有效指标不足，无法执行 PCA'))

    pca_full = PCA()
    pca_full.fit(working[valid_columns])
    cumulative = np.cumsum(pca_full.explained_variance_ratio_)
    recommended_components = int(np.searchsorted(cumulative, threshold) + 1)
    n_components = params.get('n_components')
    if n_components in (None, ''):
        n_components = recommended_components
    else:
        n_components = parse_positive_int(n_components, _('主成分数量'))
    n_components = min(n_components, len(valid_columns))

    pca = PCA(n_components=n_components)
    component_scores = pca.fit_transform(working[valid_columns])
    variance_weights = pca.explained_variance_ratio_
    composite_score = component_scores @ variance_weights

    result_df = df.copy()
    result_df[new_name] = np.nan
    result_df.loc[working.index, new_name] = composite_score

    descriptions = dict(column_descriptions)
    descriptions[new_name] = _('PCA 综合得分')
    created_component_columns = []
    if add_component_columns:
        for component_index in range(n_components):
            component_name = ensure_new_column_absent(result_df, f'{new_name}_pc{component_index + 1}')
            result_df[component_name] = np.nan
            result_df.loc[working.index, component_name] = component_scores[:, component_index]
            descriptions[component_name] = _('PCA 第 %(index)s 主成分得分') % {'index': component_index + 1}
            created_component_columns.append(component_name)

    loadings = pd.DataFrame(
        pca.components_.T,
        index=valid_columns,
        columns=[f'PC{index + 1}' for index in range(n_components)],
    )
    details = {
        'created_column': new_name,
        'component_columns': created_component_columns,
        'recommended_components': recommended_components,
        'explained_variance': [round(float(value), 6) for value in pca.explained_variance_.tolist()],
        'explained_variance_ratio': [round(float(value), 6) for value in variance_weights.tolist()],
        'cumulative_variance_ratio': [round(float(value), 6) for value in cumulative[:n_components].tolist()],
        'loadings': loadings.round(6).to_dict(),
        'score_summary': _build_score_summary(pd.Series(composite_score, index=working.index)),
        'used_columns': valid_columns,
    }
    return result_df, details, warnings, descriptions
主成分分析（PCA） ​

核心代码 ​

主成分分析（PCA）

核心代码
