缺失值处理

删除、均值 / 中位数 / 众数 / 常数 / 前后向及分组填充。

核心代码

def handle_missing_values(
    *,
    df: pd.DataFrame,
    params: dict[str, Any],
    column_descriptions: dict[str, str],
) -> tuple[pd.DataFrame, dict[str, Any], list[str], dict[str, str]]:
    """缺失值处理。"""
    action = str(params.get('action') or '').strip().lower()
    columns = require_columns(df, normalize_string_list(params.get('columns')), label=_('缺失值变量'))
    result_df = df.copy()

    if action == 'drop_rows':
        mode = str(params.get('mode') or 'any').strip().lower()
        subset = result_df[columns]
        if mode == 'any':
            mask = subset.isna().any(axis=1)
        elif mode == 'all':
            mask = subset.isna().all(axis=1)
        elif mode == 'proportion':
            threshold = parse_fraction(params.get('threshold'), _('缺失比例阈值'))
            mask = subset.isna().mean(axis=1) >= threshold
        else:
            raise DataProcessingValidationError(_('缺失值删除模式仅支持 any、all、proportion'))
        removed_rows = int(mask.sum())
        result_df = result_df.loc[~mask].reset_index(drop=True)
        details = {
            'action': action,
            'mode': mode,
            'removed_rows': removed_rows,
            'columns': columns,
        }
        return result_df, details, [], column_descriptions

    if action == 'fill':
        strategy = str(params.get('strategy') or '').strip().lower()
        fill_value = params.get('fill_value')
        group_by = normalize_string_list(params.get('group_by'))
        if group_by:
            require_columns(df, group_by, label=_('分组变量'))

        for column in columns:
            series = result_df[column]
            if strategy == 'mean':
                value = pd.to_numeric(series, errors='coerce').mean()
                result_df[column] = pd.to_numeric(series, errors='coerce').fillna(value)
            elif strategy == 'median':
                value = pd.to_numeric(series, errors='coerce').median()
                result_df[column] = pd.to_numeric(series, errors='coerce').fillna(value)
            elif strategy == 'mode':
                mode_series = series.mode(dropna=True)
                value = mode_series.iloc[0] if not mode_series.empty else None
                result_df[column] = series.fillna(value)
            elif strategy == 'constant':
                result_df[column] = series.fillna(fill_value)
            elif strategy == 'ffill':
                result_df[column] = series.ffill()
            elif strategy == 'bfill':
                result_df[column] = series.bfill()
            elif strategy in {'group_mean', 'group_median'}:
                if not group_by:
                    raise DataProcessingValidationError(_('分组填充需要选择分组变量'))
                numeric_series = pd.to_numeric(series, errors='coerce')
                agg_func = 'mean' if strategy == 'group_mean' else 'median'
                group_values = result_df.groupby(group_by)[column].transform(
                    lambda current: pd.to_numeric(current, errors='coerce').agg(agg_func)
                )
                result_df[column] = numeric_series.fillna(group_values)
            else:
                raise DataProcessingValidationError(_('填充策略仅支持 mean、median、mode、constant、ffill、bfill、group_mean、group_median'))

        details = {
            'action': action,
            'strategy': strategy,
            'columns': columns,
            'group_by': group_by,
        }
        warnings = []
        if strategy.startswith('group_') and result_df[columns].isna().sum().sum() > 0:
            warnings.append(_('部分分组可能全部缺失，处理后仍保留缺失值'))
        return result_df, details, warnings, column_descriptions

    if action == 'indicator':
        created_columns = []
        for column in columns:
            new_name = ensure_new_column_absent(result_df, f'{column}_missing')
            result_df[new_name] = result_df[column].isna().astype(int)
            created_columns.append(new_name)
        details = {
            'action': action,
            'columns': columns,
            'created_columns': created_columns,
        }
        return result_df, details, [], column_descriptions

    raise DataProcessingValidationError(_('缺失值操作仅支持 drop_rows、fill、indicator'))