标准化

Z-score / MinMax / Robust 标准化。

核心代码

def generate_standardized_variables(
    *,
    df: pd.DataFrame,
    params: dict[str, Any],
    column_descriptions: dict[str, str],
) -> tuple[pd.DataFrame, dict[str, Any], list[str], dict[str, str]]:
    """生成标准化变量。"""
    columns = require_numeric_columns(df, normalize_string_list(params.get('columns')), label=_('标准化变量'))
    method = str(params.get('method') or 'zscore').strip().lower()
    group_by = normalize_string_list(params.get('group_by'))
    if group_by:
        require_columns(df, group_by, label=_('分组变量'))
    result_df = df.copy()
    descriptions = dict(column_descriptions)
    created_columns = []
    warnings = []

    for column in columns:
        if method == 'zscore':
            new_name = ensure_new_column_absent(result_df, f'{column}_z')
            transformed = _group_transform(result_df, column, group_by, _zscore_series)
        elif method == 'minmax':
            new_name = ensure_new_column_absent(result_df, f'{column}_minmax')
            transformed = _group_transform(result_df, column, group_by, _minmax_series)
        elif method == 'robust':
            new_name = ensure_new_column_absent(result_df, f'{column}_robust')
            transformed = _group_transform(result_df, column, group_by, _robust_series)
        else:
            raise DataProcessingValidationError(_('标准化方式仅支持 zscore、minmax、robust'))
        result_df[new_name] = transformed
        descriptions[new_name] = _('%(column)s 的 %(method)s 标准化') % {'column': column, 'method': method}
        created_columns.append(new_name)
        if pd.Series(transformed).dropna().empty:
            warnings.append(_('%(column)s 标准化后全部为空，可能因为该变量没有有效波动') % {'column': column})

    details = {
        'created_columns': created_columns,
        'method': method,
        'group_by': group_by,
    }
    return result_df, details, warnings, descriptions