raw_to_processed_results

aggregate_df(res_df, metric_name, metric_type, columns=ALL_REGIMES, error_type='std')

Aggregates metrics data by computing mean and error statistics.

Parameters:

Name	Type	Description	Default
res_df	DataFrame	DataFrame containing lists of metric values for each regime	required
metric_name	str	Name of the metric (e.g., 'accuracy', 'balanced_accuracy')	required
columns	list[str]	List of evaluation regimes to process	ALL_REGIMES
error_type	str	Type of error to calculate ('std' or 'sem')	'std'

Returns:

Type	Description
DataFrame	DataFrame with aggregated metrics and formatted results

Source code in src/run/multi_run/raw_to_processed_results.py

def aggregate_df(
    res_df: pd.DataFrame,
    metric_name: str,
    metric_type: str,
    columns: list[str] = ALL_REGIMES,
    error_type: str = 'std',
) -> pd.DataFrame:
    """
    Aggregates metrics data by computing mean and error statistics.

    Args:
        res_df: DataFrame containing lists of metric values for each regime
        metric_name: Name of the metric (e.g., 'accuracy', 'balanced_accuracy')
        columns: List of evaluation regimes to process
        error_type: Type of error to calculate ('std' or 'sem')

    Returns:
        DataFrame with aggregated metrics and formatted results
    """
    res_df = res_df.copy()

    for col in columns:
        # Calculate mean values and convert to percentage rounded to 1 decimal place
        avg_col = f'Avg {metric_name} {col}'
        res_df[avg_col] = res_df[col].apply(np.nanmean)

        # if metric_type is not 'Regression':
        if metric_type == 'Regression':
            res_df[avg_col] = res_df[avg_col].round(2)
        else:
            res_df[avg_col] = (100 * res_df[avg_col]).round(1)

        def handle_missing(x):
            if isinstance(x, list):
                if len(x) > 0:
                    return np.std(x) / np.sqrt(len(x))
            return 0

        # Calculate error statistics based on specified type
        if error_type == 'std':
            err_col = f'Std {metric_name} {col}'
            res_df[err_col] = res_df[col].apply(np.nanstd)
        elif error_type == 'sem':
            err_col = f'SEM {metric_name} {col}'
            res_df[err_col] = res_df[col].apply(lambda x: handle_missing(x))
        else:
            raise ValueError(f'Invalid error type: {error_type}')

        # Convert error to percentage and round
        if metric_type == 'Regression':
            res_df[err_col] = res_df[err_col].round(1)
        else:
            res_df[err_col] = (100 * res_df[err_col]).round(1)

        # Format the final result string with mean ± error
        formatted_col = f'{col}'.replace('_', ' ').capitalize()
        res_df[formatted_col] = res_df.apply(
            lambda x: f'{x[avg_col]} ± {x[err_col]}',
            axis=1,
        )

        # Remove intermediate columns
        res_df = res_df.drop([col, avg_col, err_col], axis=1)

    return res_df

get_metric_from_raw_res(res, metric_type, metric_name, data_task, model, eval_type)

Converts raw results dataframe into accuracy metrics by evaluation regime.

Parameters:

Name	Type	Description	Default
res	DataFrame	DataFrame containing predictions and labels	required
metric_name	str	Metric to compute ('accuracy' or 'balanced_accuracy')	required

Returns:

Type	Description
DataFrame	DataFrame with metrics for each evaluation regime and overall

Source code in src/run/multi_run/raw_to_processed_results.py

def get_metric_from_raw_res(
    res: pd.DataFrame,
    metric_type: object,
    metric_name: str,
    data_task: str,
    model: str,
    eval_type: str,
) -> pd.DataFrame:
    """
    Converts raw results dataframe into accuracy metrics by evaluation regime.

    Args:
        res: DataFrame containing predictions and labels
        metric_name: Metric to compute ('accuracy' or 'balanced_accuracy')

    Returns:
        DataFrame with metrics for each evaluation regime and overall
    """
    res_df = defaultdict(list)
    stats = []

    # Calculate metrics for each evaluation regime and fold
    for (eval_regime, fold), group in res.groupby(['eval_regime', 'fold_index']):
        y_true = group['label']
        prediction_prob = group['prediction_prob']
        n_folds = group['fold_index'].nunique()
        fold_list = group['fold_index'].unique().tolist()

        new_stats = validate_results(
            eval_type,
            data_task,
            model,
            eval_regime,
            metric_type,
            metric_name,
            y_true,
            prediction_prob,
            fold,
            n_folds,
            fold_list,
        )

        score = get_scores(
            y_true=y_true,
            prediction_prob=prediction_prob,
            metric_name=metric_name,
        )
        res_df[eval_regime].append(score)
        new_stats['score'] = score
        stats.append(new_stats)

    # Calculate overall metrics for each fold
    for fold, group in res.groupby('fold_index'):
        y_true = group['label']
        prediction_prob = group['prediction_prob']
        eval_regime = 'all'
        n_folds = group['fold_index'].nunique()
        fold_list = group['fold_index'].unique().tolist()

        new_stats = validate_results(
            eval_type,
            data_task,
            model,
            eval_regime,
            metric_type,
            metric_name,
            y_true,
            prediction_prob,
            fold,
            n_folds,
            fold_list,
        )

        score = get_scores(
            y_true=y_true,
            prediction_prob=prediction_prob,
            metric_name=metric_name,
        )
        res_df[eval_regime].append(score)
        new_stats['score'] = score
        stats.append(new_stats)

    # Calculate overall metrics for each evaluation regime
    for eval_regime, group in res.groupby('eval_regime'):
        y_true = group['label']
        prediction_prob = group['prediction_prob']
        fold = 'all'
        n_folds = group['fold_index'].nunique()
        fold_list = group['fold_index'].unique().tolist()

        new_stats = validate_results(
            eval_type,
            data_task,
            model,
            eval_regime,
            metric_type,
            metric_name,
            y_true,
            prediction_prob,
            fold,
            n_folds,
            fold_list,
        )

        score = get_scores(
            y_true=y_true,
            prediction_prob=prediction_prob,
            metric_name=metric_name,
        )
        new_stats['score'] = score
        stats.append(new_stats)

    # Create DataFrame and ensure all regimes are included
    df = pd.DataFrame(
        res_df.items(),
        columns=['Eval Regime', metric_name],
    ).set_index('Eval Regime')
    # Concatenate all stats DataFrames
    stats_df = pd.DataFrame(stats)
    # Reindex to ensure all regimes are present, even if empty
    return df.reindex(ALL_REGIMES), stats_df

validate_results(eval_type, data_task, model, eval_regime, metric_type, metric_name, y_true, prediction_prob, fold, n_folds, fold_list)

Validate the results before computing statistics.

Source code in src/run/multi_run/raw_to_processed_results.py

def validate_results(
    eval_type: str,
    data_task: str,
    model: str,
    eval_regime,
    metric_type: str,
    metric_name,
    y_true: pd.Series,
    prediction_prob: pd.Series,
    fold,
    n_folds: int,
    fold_list,
) -> dict:
    """Validate the results before computing statistics."""

    # Process prediction probabilities to get predicted classes/values
    y_pred = _process_prediction_prob(metric_name, prediction_prob)

    unique_y_true = np.unique(y_true)
    unique_y_pred = np.unique(y_pred)

    # If classification, ensure predicted classes are a subset of true classes
    if metric_type != 'Regression':
        extra_classes = set(unique_y_pred) - set(unique_y_true)
        # convert to float and round for readability
        extra_classes = [round(float(x), 2) for x in extra_classes]
        if len(extra_classes) > 5:
            extra_classes = (
                list(extra_classes)[:5]
                + ['...']
                + [f'N extra classes: {len(extra_classes)}']
            )

    # get distribution if n_unique < 10 else ['Too many classes: {n_unique}']
    y_true_distribution = (
        y_true.value_counts().to_dict()
        if len(unique_y_true) < 10
        else {f'N unique: {len(unique_y_true)}'}
    )
    y_pred_distribution = (
        y_pred.value_counts().to_dict()
        if len(unique_y_pred) < 10
        else {f'N unique: {len(unique_y_pred)}'}
    )

    stats = {
        'metric_type': metric_type,
        'metric_name': metric_name.value,
        'eval_type': eval_type,
        'data_task': data_task,
        'model': model,
        'eval_regime': eval_regime,
        'fold': fold,
        'n_folds': n_folds,
        'fold_list': fold_list,
        'n_samples': len(y_true),
        'n_unique_y_true': len(np.unique(y_true)),
        'n_unique_y_pred': len(np.unique(y_pred)),
        'y_true_single': len(np.unique(y_true)) == 1,
        'y_pred_single': len(np.unique(y_pred)) == 1,
        'y_true_distribution': y_true_distribution,
        'y_pred_distribution': y_pred_distribution,
        'extra_classes_in_y_pred': list(extra_classes)
        if metric_type != 'Regression' and extra_classes
        else None,
    }

    return stats