因子分析

因子分析（KMO、Bartlett 检验与 varimax 旋转）。

核心代码

def build_factor_index(
    *,
    df: pd.DataFrame,
    params: dict[str, Any],
    column_descriptions: dict[str, str],
) -> tuple[pd.DataFrame, dict[str, Any], list[str], dict[str, str]]:
    """因子分析综合指标。"""
    indicators = require_numeric_columns(df, normalize_string_list(params.get('columns')), label=_('指标变量'))
    new_name = ensure_new_column_absent(df, params.get('new_name') or 'factor_score')
    add_factor_columns = normalize_bool(params.get('add_factor_columns'), default=True)
    rotation = str(params.get('rotation') or 'varimax').strip().lower()
    if rotation not in {'none', 'varimax'}:
        raise DataProcessingValidationError(_('因子旋转暂时只支持 none 或 varimax'))

    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) < 2:
        raise DataProcessingValidationError(_('因子分析至少需要两个有效指标'))

    kmo_value, kmo_per_item = compute_kmo(working[valid_columns])
    bartlett_stat, bartlett_p = compute_bartlett_sphericity(working[valid_columns])
    if kmo_value < 0.5:
        warnings.append(_('KMO = %(kmo)s，样本结构偏弱，因子分析结果需谨慎解释') % {'kmo': format(kmo_value, '.4f')})
    if bartlett_p >= 0.05:
        warnings.append(_('Bartlett 球形检验未显著，变量相关结构可能不足'))

    corr = np.corrcoef(working[valid_columns].to_numpy().T)
    eigenvalues = np.linalg.eigvalsh(corr)[::-1]
    recommended_factors = max(int(np.sum(eigenvalues > 1)), 1)
    n_factors = params.get('n_factors')
    if n_factors in (None, ''):
        n_factors = recommended_factors
    else:
        n_factors = parse_positive_int(n_factors, _('因子数量'))
    n_factors = min(n_factors, len(valid_columns))

    fa = FactorAnalysis(n_components=n_factors, random_state=0)
    factor_scores = fa.fit_transform(working[valid_columns])
    loadings = fa.components_.T
    if rotation == 'varimax':
        loadings = varimax(loadings)
        standardized_matrix = working[valid_columns].to_numpy()
        factor_scores = standardized_matrix @ np.linalg.pinv(loadings.T)

    factor_variance = np.sum(loadings ** 2, axis=0)
    if factor_variance.sum() == 0:
        raise DataProcessingValidationError(_('因子载荷方差为 0，无法生成综合得分'))
    factor_weights = factor_variance / factor_variance.sum()
    composite_score = factor_scores @ factor_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] = _('因子分析综合得分')
    created_factor_columns = []
    if add_factor_columns:
        for factor_index in range(n_factors):
            factor_name = ensure_new_column_absent(result_df, f'{new_name}_f{factor_index + 1}')
            result_df[factor_name] = np.nan
            result_df.loc[working.index, factor_name] = factor_scores[:, factor_index]
            descriptions[factor_name] = _('因子分析第 %(index)s 因子得分') % {'index': factor_index + 1}
            created_factor_columns.append(factor_name)

    loadings_df = pd.DataFrame(
        loadings,
        index=valid_columns,
        columns=[f'Factor{index + 1}' for index in range(n_factors)],
    )
    details = {
        'created_column': new_name,
        'factor_columns': created_factor_columns,
        'recommended_factors': recommended_factors,
        'rotation': rotation,
        'kmo': round(float(kmo_value), 6),
        'kmo_per_item': {column: round(float(value), 6) for column, value in zip(valid_columns, kmo_per_item)},
        'bartlett_stat': round(float(bartlett_stat), 6),
        'bartlett_p_value': round(float(bartlett_p), 6),
        'factor_variance': [round(float(value), 6) for value in factor_variance.tolist()],
        'factor_weight': [round(float(value), 6) for value in factor_weights.tolist()],
        'loadings': loadings_df.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