Skip to content

Binary classification curves

BinaryClassificationCurves

Class for computing and storing ROC & Precision-Recall curves in Polars.

Source code in src/pheval/analyse/binary_classification_curves.py 
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
class BinaryClassificationCurves:
    """Class for computing and storing ROC & Precision-Recall curves in Polars."""

    @staticmethod
    def _compute_finite_bounds(result_scan: pl.LazyFrame) -> tuple[float, float]:
        """
        Compute min and max finite values in the 'score' column to handle NaN and Inf values.
        Args:
            result_scan (pl.LazyFrame): The LazyFrame containing the results for the directory.

        Returns:
            Tuple[float, float]: The (max_finite, min_finite) values for normalising scores.
        """
        return (
            result_scan.select(
                [
                    pl.col("score").filter(pl.col("score").is_finite()).max().alias("max_finite"),
                    pl.col("score").filter(pl.col("score").is_finite()).min().alias("min_finite"),
                ]
            )
            .collect()
            .row(0)
        )

    @staticmethod
    def _clean_and_extract_data(result_scan: pl.LazyFrame, max_finite: float, min_finite: float) -> pl.LazyFrame:
        """
        Normalise the 'score' column (handling NaNs and Inf values) and extract 'true_positive' labels.

        Args:
            result_scan (pl.LazyFrame): The LazyFrame containing the results for the directory.
            max_finite (float): The maximum finite score value.
            min_finite (float): The minimum finite score value.

        Returns:
            pl.LazyFrame: A LazyFrame with cleaned 'score' and binary 'true_positive' columns.
        """
        return result_scan.with_columns(
            [
                pl.when(pl.col("score").is_nan())
                .then(0.0)
                .when(pl.col("score").is_infinite() & (pl.col("score") > 0))
                .then(max_finite)
                .when(pl.col("score").is_infinite() & (pl.col("score") < 0))
                .then(min_finite)
                .otherwise(pl.col("score"))
                .alias("score"),
                pl.when(pl.col("true_positive").is_null())
                .then(0)
                .otherwise(pl.col("true_positive").cast(pl.Int8))
                .alias("true_positive"),
            ]
        )

    @staticmethod
    def _compute_roc_pr_curves(run_identifier: str, labels: np.ndarray, scores: np.ndarray) -> pl.LazyFrame:
        """
        Compute ROC and Precision-Recall curves.

        Args:
            labels (np.ndarray): Binary ground truth labels (0 or 1).
            scores (np.ndarray): Prediction scores.

        Returns:
            pl.LazyFrame: A LazyFrame containing the computed FPR, TPR, Precision, Recall, and Thresholds.
        """
        fpr, tpr, roc_thresholds = roc_curve(labels, scores, pos_label=1)
        precision, recall, pr_thresholds = precision_recall_curve(labels, scores, pos_label=1)

        return pl.LazyFrame(
            {
                "run_identifier": [run_identifier],
                "fpr": [fpr.tolist()],
                "tpr": [tpr.tolist()],
                "threshold_roc": [roc_thresholds.tolist()],
                "precision": [precision.tolist()],
                "recall": [recall.tolist()],
                "threshold_pr": [pr_thresholds.tolist()],
            }
        )

    @classmethod
    def process(cls, result_scan: pl.LazyFrame, run_identifier: str) -> pl.LazyFrame:
        """
        Process scores, extract true labels, compute ROC and Precision-Recall curves,
        and store results in a Polars LazyFrame with NumPy arrays.

        Args:
            result_scan (pl.LazyFrame): The LazyFrame containing the results for the directory.
            run_identifier (str): Identifier for this run.

        Returns:
            pl.LazyFrame: A LazyFrame containing ROC & PR curve data with NumPy arrays.
        """
        max_finite, min_finite = cls._compute_finite_bounds(result_scan)
        cleaned_data = (
            cls._clean_and_extract_data(result_scan, max_finite, min_finite)
            .select(["true_positive", "score"])
            .collect()
        )
        return cls._compute_roc_pr_curves(
            run_identifier,
            cleaned_data["true_positive"].to_numpy().flatten(),
            cleaned_data["score"].to_numpy().flatten(),
        )

process(result_scan, run_identifier) classmethod

Process scores, extract true labels, compute ROC and Precision-Recall curves, and store results in a Polars LazyFrame with NumPy arrays.

Parameters:

Name Type Description Default
result_scan LazyFrame

The LazyFrame containing the results for the directory.

 required
run_identifier str

Identifier for this run.

 required

Returns:

Type Description
LazyFrame

pl.LazyFrame: A LazyFrame containing ROC & PR curve data with NumPy arrays.
Source code in src/pheval/analyse/binary_classification_curves.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
@classmethod
def process(cls, result_scan: pl.LazyFrame, run_identifier: str) -> pl.LazyFrame:
    """
    Process scores, extract true labels, compute ROC and Precision-Recall curves,
    and store results in a Polars LazyFrame with NumPy arrays.

    Args:
        result_scan (pl.LazyFrame): The LazyFrame containing the results for the directory.
        run_identifier (str): Identifier for this run.

    Returns:
        pl.LazyFrame: A LazyFrame containing ROC & PR curve data with NumPy arrays.
    """
    max_finite, min_finite = cls._compute_finite_bounds(result_scan)
    cleaned_data = (
        cls._clean_and_extract_data(result_scan, max_finite, min_finite)
        .select(["true_positive", "score"])
        .collect()
    )
    return cls._compute_roc_pr_curves(
        run_identifier,
        cleaned_data["true_positive"].to_numpy().flatten(),
        cleaned_data["score"].to_numpy().flatten(),
    )

compute_curves(run_identifier, result_scan)

Compute ROC and Precision-Recall curves. Args: result_scan (pl.LazyFrame): The LazyFrame containing the results for the directory. run_identifier (str): Identifier for this run. Returns: pl.LazyFrame: LazyFrame containing the ROC & Precision-Recall curve data with NumPy arrays.

Source code in src/pheval/analyse/binary_classification_curves.py 
115
116
117
118
119
120
121
122
123
124
125
126
def compute_curves(run_identifier: str, result_scan: pl.LazyFrame) -> pl.LazyFrame:
    """
    Compute ROC and Precision-Recall curves.
    Args:
        result_scan (pl.LazyFrame): The LazyFrame containing the results for the directory.
        run_identifier (str): Identifier for this run.
    Returns:
        pl.LazyFrame: LazyFrame containing the ROC & Precision-Recall curve data with NumPy arrays.
    """
    logger = get_logger()
    logger.info("Calculating ROC and Precision-Recall metrics")
    return BinaryClassificationCurves.process(result_scan, run_identifier)