deepecgkit.evaluation

Model evaluation, metrics computation, and visualization tools.

Evaluator

ECGEvaluator

Comprehensive evaluator for ECG models.

This class provides a unified interface for evaluating ECG models with various metrics and analysis tools.

It takes either a model together with its data, or predictions computed elsewhere together with their targets. A torch model is run on the evaluation device in eval mode and handed back on the device and in the mode it came in. Its outputs are taken to be logits, as every model in the package emits and as ECGTrainer assumes, and are turned into probabilities for a classification task: a single output becomes the two columns [1 - p, p], several outputs go through a softmax, or through a sigmoid when the targets are multi-hot. Rows that already sum to one are left alone. Predictions passed in directly are used as they are.

Parameters:

Name	Type	Description	Default
metrics	list[str] | None	List of metrics to compute: accuracy, precision, recall, f1, mcc and auc for classification; mse, mae, r2 and rmse for regression. Multi-hot targets are scored as calculate_classification_metrics describes, where exact_match_accuracy is also available.	None
task_type	str	Type of task ("classification", "regression", "auto")	'auto'
device	str	Device for model evaluation	'auto'
calibration_bins	int	Number of bins for the calibration metrics reported by generate_report. ECE depends on this, so the value is emitted alongside the metrics rather than left implicit.	DEFAULT_CALIBRATION_BINS

Raises:

Type	Description
ValueError	If metrics names one the evaluator does not compute.

Examples:

>>> evaluator = ECGEvaluator(metrics=["accuracy", "auc", "f1"])
>>> results = evaluator.evaluate(model, test_data)

Source code in deepecgkit/evaluation/evaluator.py

class ECGEvaluator:
    """
    Comprehensive evaluator for ECG models.

    This class provides a unified interface for evaluating ECG models with
    various metrics and analysis tools.

    It takes either a model together with its data, or predictions computed
    elsewhere together with their targets. A torch model is run on the
    evaluation device in eval mode and handed back on the device and in the
    mode it came in. Its outputs are taken to be logits, as every model in the
    package emits and as ECGTrainer assumes, and are turned into probabilities
    for a classification task: a single output becomes the two columns
    [1 - p, p], several outputs go through a softmax, or through a sigmoid
    when the targets are multi-hot. Rows that already sum to one are left
    alone. Predictions passed in directly are used as they are.

    Args:
        metrics: List of metrics to compute: accuracy, precision, recall, f1,
            mcc and auc for classification; mse, mae, r2 and rmse for
            regression. Multi-hot targets are scored as
            calculate_classification_metrics describes, where
            exact_match_accuracy is also available.
        task_type: Type of task ("classification", "regression", "auto")
        device: Device for model evaluation
        calibration_bins: Number of bins for the calibration metrics reported
            by generate_report. ECE depends on this, so the value is emitted
            alongside the metrics rather than left implicit.

    Raises:
        ValueError: If metrics names one the evaluator does not compute.

    Examples:
        >>> evaluator = ECGEvaluator(metrics=["accuracy", "auc", "f1"])
        >>> results = evaluator.evaluate(model, test_data)
    """

    def __init__(
        self,
        metrics: list[str] | None = None,
        task_type: str = "auto",
        device: str = "auto",
        calibration_bins: int = DEFAULT_CALIBRATION_BINS,
    ):
        self.metrics = metrics or ["accuracy", "precision", "recall", "f1", "auc"]
        unknown = [name for name in self.metrics if name not in SUPPORTED_METRICS]
        if unknown:
            raise ValueError(
                f"Unknown metric(s): {', '.join(unknown)}. "
                f"Supported: {', '.join(sorted(SUPPORTED_METRICS))}."
            )
        self.task_type = task_type
        self.calibration_bins = calibration_bins
        self.y_scores: np.ndarray | None = None
        self.y_true: np.ndarray | None = None

        self.device = resolve_device(device)

    def evaluate(
        self,
        model: torch.nn.Module | np.ndarray | None = None,
        test_data: torch.utils.data.DataLoader | np.ndarray | tuple | None = None,
        return_predictions: bool = False,
        y_scores: np.ndarray | None = None,
    ) -> dict[str, float]:
        """
        Evaluate model performance.

        Args:
            model: PyTorch model or numpy array of predictions
            test_data: Test data loader or (predictions, targets) tuple
            return_predictions: Whether to return predictions along with metrics
            y_scores: Scores/probabilities for AUC calculation (optional)

        Returns:
            Dictionary of metric values
        """

        self.y_scores = y_scores

        predictions, targets = self._process_input_data(model, test_data)
        if targets is None:
            raise ValueError(
                "Evaluation needs the targets as well as the inputs: pass (inputs, targets), "
                "a DataLoader that yields both, or predictions together with their targets."
            )
        self.y_true = targets
        metrics = self._calculate_metrics(predictions, targets, self.y_scores)

        if return_predictions:
            return metrics, predictions
        return metrics

    def _process_input_data(
        self,
        model: torch.nn.Module | np.ndarray | None,
        test_data: torch.utils.data.DataLoader | np.ndarray | tuple | None,
    ) -> tuple:
        """Process input data to get predictions and targets."""
        if isinstance(model, np.ndarray):
            return self._process_numpy_input(model, test_data)
        return self._get_predictions(model, test_data)

    def _process_numpy_input(
        self,
        model: np.ndarray,
        test_data: torch.utils.data.DataLoader | np.ndarray | tuple | None,
    ) -> tuple:
        """Process input when model is a numpy array."""
        predictions = model
        if not isinstance(test_data, tuple):
            return predictions, test_data

        if len(test_data) == 3:
            targets = test_data[1]
            self.y_scores = test_data[2]
        elif len(test_data) == 2:
            targets = test_data[1]
        else:
            targets = test_data
        return predictions, targets

    def _calculate_metric(
        self,
        name: str,
        predictions: np.ndarray,
        targets: np.ndarray,
        scores: np.ndarray | None = None,
    ) -> float:
        """Calculate one metric. Evaluation and bootstrapping both come through here."""
        if name in CLASSIFICATION_METRICS:
            return calculate_classification_metrics(targets, predictions, metrics=[name])[name]
        if name in REGRESSION_METRICS:
            return calculate_regression_metrics(targets, predictions, metrics=[name])[name]

        # Per-class predictions are their own scores; hard labels are not scores at all.
        if scores is None and np.ndim(predictions) > 1:
            scores = predictions
        if scores is None:
            raise ValueError("y_scores required for AUC calculation")
        return calculate_classification_metrics(targets, scores, metrics=[name])[name]

    def _calculate_metrics(
        self,
        predictions: np.ndarray,
        targets: np.ndarray,
        scores: np.ndarray | None = None,
    ) -> dict[str, float]:
        """Calculate metrics based on predictions and targets."""
        return {
            name: self._calculate_metric(name, predictions, targets, scores)
            for name in self.metrics
        }

    def _get_predictions(
        self,
        model: torch.nn.Module | Any | None,
        test_data: torch.utils.data.DataLoader | np.ndarray | tuple | None,
    ) -> tuple:
        """Extract predictions and targets from model and data."""
        if model is None and isinstance(test_data, tuple):
            return self._handle_tuple_test_data(test_data)

        if not callable(model):
            raise TypeError(
                f"Expected a model to call or an array of predictions, got {type(model).__name__}."
            )

        with self._evaluating(model):
            if isinstance(test_data, torch.utils.data.DataLoader):
                outputs, targets = self._process_dataloader(model, test_data)
            elif isinstance(test_data, tuple) and len(test_data) == 2:
                outputs, targets = self._process_tuple_data(model, test_data)
            elif isinstance(test_data, np.ndarray):
                outputs, targets = self._process_numpy_data(model, test_data)
            else:
                raise ValueError(f"Unsupported test data type: {type(test_data)}")

        if isinstance(model, torch.nn.Module) and self._is_classification(targets):
            outputs = self._as_probabilities(outputs, targets)
        return outputs, targets

    @contextlib.contextmanager
    def _evaluating(self, model: Any) -> Iterator[None]:
        """Hold a torch module on the evaluation device and in eval mode.

        The inputs are moved to the evaluation device, so the model has to be
        there too; and a model left in training mode would be measured with
        dropout active and would have its batch-norm statistics overwritten by
        the test data. Both are put back afterwards.
        """
        if not isinstance(model, torch.nn.Module):
            yield
            return

        was_training = model.training
        parameter = next(model.parameters(), None)
        previous_device = None if parameter is None else parameter.device
        model.to(self.device).eval()
        try:
            yield
        finally:
            model.train(was_training)
            if previous_device is not None:
                model.to(previous_device)

    def _predict(self, model: Any, x: Any) -> Any:
        """Run one batch through the model, as numpy when the model returns a tensor."""
        if isinstance(model, torch.nn.Module) and not torch.is_tensor(x):
            x = torch.as_tensor(np.asarray(x), dtype=torch.float32)
        if isinstance(model, torch.nn.Module) and len(x) > PREDICTION_BATCH_SIZE:
            # A whole split in one forward pass can need more memory than there is.
            chunks = torch.split(x, PREDICTION_BATCH_SIZE)
            return np.concatenate([self._predict(model, chunk) for chunk in chunks])
        if torch.is_tensor(x):
            x = x.to(self.device)

        with torch.no_grad():
            pred = model(x)

        if torch.is_tensor(pred):
            pred = pred.cpu().numpy()
        return pred

    def _is_classification(self, targets: np.ndarray | None) -> bool:
        """Whether the evaluated task is classification, resolving "auto" from the targets."""
        if self.task_type != "auto":
            return self.task_type == "classification"
        if set(self.metrics) <= REGRESSION_METRICS:
            return False
        return self._detect_task_type(targets) == "classification"

    def _as_probabilities(self, outputs: np.ndarray, targets: np.ndarray | None) -> np.ndarray:
        """Turn the logits of a classifier into probabilities, the way ECGTrainer does."""
        logits = torch.as_tensor(np.asarray(outputs))
        if not torch.is_floating_point(logits) or logits.ndim > 2:
            return outputs

        if logits.ndim == 2 and logits.shape[1] == 1:
            logits = logits[:, 0]
        if logits.ndim == 1:
            positive = torch.sigmoid(logits)
            return torch.stack([1 - positive, positive], dim=1).numpy()

        if targets is not None and np.ndim(targets) > 1:
            return torch.sigmoid(logits).numpy()

        row_sums = logits.sum(dim=1)
        sums_to_one = torch.allclose(
            row_sums, torch.ones_like(row_sums), atol=PROBABILITY_SUM_TOLERANCE
        )
        if sums_to_one and bool((logits >= 0).all()):
            return outputs
        return torch.softmax(logits, dim=1).numpy()

    def _handle_tuple_test_data(self, test_data: tuple) -> tuple:
        """Handle test data when it's a tuple."""
        if len(test_data) == 3:
            y_true, y_pred, y_scores = test_data
            self.y_scores = y_scores
            return y_pred, y_true
        elif len(test_data) == 2:
            return test_data[0], test_data[1]
        return test_data, None

    def _process_dataloader(
        self, model: torch.nn.Module, test_data: torch.utils.data.DataLoader
    ) -> tuple:
        """Process data from a DataLoader."""
        predictions = []
        targets = []
        for batch in test_data:
            if isinstance(batch, list | tuple) and len(batch) == 2:
                x, y = batch
            else:
                x, y = batch, None

            predictions.append(self._predict(model, x))

            if y is not None:
                if torch.is_tensor(y):
                    y = y.cpu().numpy()
                targets.append(y)

        return np.concatenate(predictions), np.concatenate(targets)

    def _process_tuple_data(self, model: torch.nn.Module, test_data: tuple) -> tuple:
        """Process data when it's a tuple."""
        x, y = test_data
        pred = self._predict(model, x)
        if torch.is_tensor(y):
            y = y.cpu().numpy()

        return pred, y

    def _process_numpy_data(self, model: torch.nn.Module, test_data: np.ndarray) -> tuple:
        """Process data when it's a numpy array."""
        pred = self._predict(model, torch.tensor(test_data, dtype=torch.float32))
        return pred, None

    def _calibration_rows(
        self, predictions: np.ndarray, calibration_fit: tuple | None
    ) -> list[dict[str, Any]]:
        """Build the calibration rows appended to the report.

        Rows whose value is unavailable are omitted rather than emitted as NaN,
        so a report produced without a fitting split simply carries no
        temperature rows instead of carrying empty ones.

        Args:
            predictions: Model output for the evaluated split, used as the
                probability matrix when no explicit y_scores were supplied
            calibration_fit: Optional (targets, logits) pair from the
                validation split, used to fit the temperature

        Returns:
            A list of {"Metric", "Value"} rows, possibly empty.
        """
        if self.y_true is None:
            return []

        scores = self.y_scores if self.y_scores is not None else predictions
        if scores is None or not np.issubdtype(np.asarray(scores).dtype, np.floating):
            # Hard labels carry no confidence to calibrate.
            return []

        fit_targets, fit_logits = calibration_fit or (None, None)
        try:
            report = calibration_report(
                self.y_true,
                scores,
                n_bins=self.calibration_bins,
                fit_targets=fit_targets,
                fit_logits=fit_logits,
            )
        except Exception as error:
            logger.warning(f"Could not compute calibration metrics: {error}")
            return []

        rows: list[dict[str, Any]] = []
        for name in CALIBRATION_ROW_ORDER:
            value = report.get(name)
            if value is None:
                continue
            if isinstance(value, float) and np.isnan(value):
                continue
            rows.append({"Metric": name, "Value": value})
        return rows

    def _detect_task_type(self, targets: np.ndarray | None) -> str:
        """Detect whether task is classification or regression."""
        if targets is None:
            return "regression"

        unique_values = np.unique(targets)

        if len(unique_values) <= MAX_CLASSIFICATION_CLASSES and np.all(
            np.equal(np.mod(targets, 1), 0)
        ):
            return "classification"
        return "regression"

    def cross_validate(
        self,
        model_class: type,
        data: Any,
        k_folds: int = 5,
        groups: np.ndarray | None = None,
        epochs: int = 10,
        batch_size: int = 32,
        train_config: dict[str, Any] | None = None,
        seed: int = 42,
        **model_kwargs: Any,
    ) -> dict[str, list[float]]:
        """
        Perform k-fold cross-validation.

        A fresh model is built and fitted for every fold and scored on the
        fold held out. A torch module is trained with ECGTrainer; a tenth of
        its training fold is set aside for the learning-rate schedule and
        early stopping, so the held-out fold plays no part in training.
        Anything else with a fit method is fitted with fit(X, y) and asked for
        predict_proba, or failing that predict, unless it can be called
        directly.

        Args:
            model_class: Class of model to evaluate
            data: (X, y) tuple of arrays or tensors
            k_folds: Number of folds
            groups: Group of each sample, such as the recording or patient it
                comes from. Folds then never split a group, which segments cut
                from longer recordings need for the estimate to be honest.
            epochs: Maximum training epochs per fold, for a torch module
            batch_size: Batch size for training and scoring a torch module
            train_config: ECGTrainer settings for a torch module. Multi-hot
                targets and a regression task_type are recognised without it;
                a single-output binary model needs binary_classification.
            seed: Seed for shuffling the folds and the training batches
            **model_kwargs: Keyword arguments for model initialization

        Returns:
            Dictionary of metric scores across folds
        """
        if not isinstance(data, tuple) or len(data) != 2:
            raise ValueError("Data must be (X, y) tuple for cross-validation")
        x, y = (np.asarray(part.cpu() if torch.is_tensor(part) else part) for part in data)
        if groups is not None:
            groups = np.asarray(groups)

        fold_results: dict[str, list[float]] = {metric: [] for metric in self.metrics}

        folds = self._folds(len(x), k_folds, groups, seed)
        for fold, (train_idx, val_idx) in enumerate(folds):
            logger.info(f"Evaluating fold {fold + 1}/{k_folds}")
            x_train, y_train = x[train_idx], y[train_idx]
            x_val, y_val = x[val_idx], y[val_idx]

            model = model_class(**model_kwargs)
            if isinstance(model, torch.nn.Module):
                fold_groups = None if groups is None else groups[train_idx]
                self._train_fold(
                    model, x_train, y_train, fold_groups, epochs, batch_size, train_config, seed
                )
                held_out = _ArrayLoaders((x_val, y_val), (x_val, y_val), batch_size, seed)
                results = self.evaluate(model, held_out.val_dataloader())
            else:
                results = self._fit_and_score(model, x_train, y_train, x_val, y_val)

            for metric in self.metrics:
                if metric in results:
                    fold_results[metric].append(results[metric])

        return fold_results

    @staticmethod
    def _folds(
        n_samples: int, n_splits: int, groups: np.ndarray | None, seed: int
    ) -> Iterator[tuple[np.ndarray, np.ndarray]]:
        """Split sample indices into folds, keeping each group whole when groups are given."""
        indices = np.arange(n_samples)
        folds: Iterator[tuple[np.ndarray, np.ndarray]]
        if groups is None:
            folds = KFold(n_splits=n_splits, shuffle=True, random_state=seed).split(indices)
        else:
            folds = GroupKFold(n_splits=n_splits).split(indices, groups=groups)
        return folds

    def _train_fold(
        self,
        model: torch.nn.Module,
        x: np.ndarray,
        y: np.ndarray,
        groups: np.ndarray | None,
        epochs: int,
        batch_size: int,
        train_config: dict[str, Any] | None,
        seed: int,
    ) -> None:
        """Train a torch module on one training fold with ECGTrainer."""
        # Imported here: training does not depend on evaluation, nor the reverse at import.
        from deepecgkit.training import ECGTrainer

        config = dict(train_config or {})
        if y.ndim > 1:
            config.setdefault("multi_label", True)
        if self.task_type == "regression":
            config.setdefault("task_type", "regression")

        available = len(x) if groups is None else len(np.unique(groups))
        inner_folds = min(INNER_VALIDATION_FOLDS, available)
        if inner_folds < 2:
            raise ValueError(
                "Each training fold needs at least two samples (or groups) to set a "
                "validation part aside; use fewer folds or more data."
            )
        fit_idx, tune_idx = next(self._folds(len(x), inner_folds, groups, seed))
        loaders = _ArrayLoaders(
            (x[fit_idx], y[fit_idx]), (x[tune_idx], y[tune_idx]), batch_size, seed
        )
        trainer = ECGTrainer(model, train_config=config, device=self.device)
        trainer.fit(loaders, epochs=epochs, progress_bar=False)

    def _fit_and_score(
        self,
        model: Any,
        x_train: np.ndarray,
        y_train: np.ndarray,
        x_val: np.ndarray,
        y_val: np.ndarray,
    ) -> dict[str, float]:
        """Fit an estimator that is not a torch module and score it on the held-out fold."""
        if hasattr(model, "fit"):
            model.fit(x_train, y_train)

        if callable(model):
            return self.evaluate(model, (x_val, y_val))
        if hasattr(model, "predict_proba"):
            return self.evaluate(model.predict_proba(x_val), y_val)
        if hasattr(model, "predict"):
            return self.evaluate(model.predict(x_val), y_val)
        raise TypeError(
            f"{type(model).__name__} can be neither called nor asked to predict, so it "
            f"cannot be cross-validated."
        )

    def bootstrap_evaluate(
        self,
        model: Any,
        test_data: Any,
        n_bootstrap: int = 1000,
        confidence_level: float = 0.95,
        y_scores: np.ndarray | None = None,
    ) -> dict[str, dict[str, float]]:
        """
        Bootstrap evaluation for confidence intervals.

        Takes the same inputs as evaluate, and computes each metric the same
        way, so an interval describes the value evaluate reports. Resampling
        draws from numpy's global generator; seed it for repeatable intervals.

        Args:
            model: PyTorch model or numpy array of predictions
            test_data: Test data loader or (predictions, targets) tuple
            n_bootstrap: Number of bootstrap samples
            confidence_level: Confidence level for intervals
            y_scores: Scores/probabilities for AUC calculation (optional)

        Returns:
            Dictionary with mean, std, and confidence intervals for each metric
        """
        self.y_scores = y_scores
        predictions, targets = self._process_input_data(model, test_data)
        return self._bootstrap(predictions, targets, self.y_scores, n_bootstrap, confidence_level)

    def _bootstrap(
        self,
        predictions: np.ndarray,
        targets: np.ndarray | None,
        scores: np.ndarray | None,
        n_bootstrap: int = 1000,
        confidence_level: float = 0.95,
    ) -> dict[str, dict[str, float]]:
        """Resample predictions with their targets and summarise each metric."""
        if targets is None:
            raise ValueError("Bootstrap evaluation requires ground truth targets")

        predictions, targets = np.asarray(predictions), np.asarray(targets)
        scores = None if scores is None else np.asarray(scores)

        # A metric that is undefined on the whole set is undefined on every resample.
        point_estimates = self._calculate_metrics(predictions, targets, scores)
        names = [name for name, value in point_estimates.items() if not np.isnan(value)]

        n_samples = len(targets)
        bootstrap_results: dict[str, list[float]] = {name: [] for name in names}

        for _ in range(n_bootstrap):
            indices = np.random.choice(n_samples, n_samples, replace=True)
            boot_predictions, boot_targets = predictions[indices], targets[indices]
            boot_scores = None if scores is None else scores[indices]
            for name in names:
                bootstrap_results[name].append(
                    self._calculate_metric(name, boot_predictions, boot_targets, boot_scores)
                )

        alpha = 1 - confidence_level
        lower_percentile = (alpha / 2) * 100
        upper_percentile = (1 - alpha / 2) * 100

        final_results = {}
        for metric, values in bootstrap_results.items():
            # A resample can lack a class and leave a metric undefined; those are left out.
            defined = [value for value in values if not np.isnan(value)]
            if defined:
                final_results[metric] = {
                    "mean": np.mean(defined),
                    "std": np.std(defined),
                    "lower_ci": np.percentile(defined, lower_percentile),
                    "upper_ci": np.percentile(defined, upper_percentile),
                }

        return final_results

    def generate_report(
        self,
        model: Any,
        test_data: Any,
        save_path: str | None = None,
        y_scores: np.ndarray | None = None,
        calibration_fit: tuple | None = None,
    ) -> pd.DataFrame:
        """
        Generate comprehensive evaluation report.

        Calibration is always reported: expected calibration error, its
        equal-mass variant, maximum calibration error and the Brier score. Pass
        calibration_fit to additionally fit a temperature and report what
        rescaling recovers, which adds a "temperature" row and the "_temp"
        counterparts of each calibration metric.

        Args:
            model: Trained model
            test_data: Test dataset
            save_path: Path to save report (optional)
            y_scores: Scores/probabilities for AUC calculation (optional)
            calibration_fit: Optional (targets, logits) pair from a held-out
                validation split, used to fit the temperature. Never pass the
                split being reported on. Callers holding probabilities rather
                than logits can convert with
                deepecgkit.evaluation.probabilities_to_logits.

        Returns:
            DataFrame with evaluation results
        """

        results = self.evaluate(model, test_data, return_predictions=True, y_scores=y_scores)
        if isinstance(results, tuple):
            metrics_dict, predictions = results[0], results[1]
        else:
            metrics_dict, predictions = results, None

        try:
            bootstrap_results = self._bootstrap(predictions, self.y_true, self.y_scores)
        except Exception as error:
            logger.warning(f"Could not compute bootstrap confidence intervals: {error}")
            bootstrap_results = {}

        report_data = []

        for metric in self.metrics:
            if metric in metrics_dict and metric not in {"predictions", "targets"}:
                row = {
                    "Metric": metric,
                    "Value": metrics_dict[metric],
                }

                if metric in bootstrap_results:
                    row.update(
                        {
                            "Std": bootstrap_results[metric]["std"],
                            "95% CI Lower": bootstrap_results[metric]["lower_ci"],
                            "95% CI Upper": bootstrap_results[metric]["upper_ci"],
                        }
                    )

                report_data.append(row)

        report_data.extend(self._calibration_rows(predictions, calibration_fit))

        report_df = pd.DataFrame(report_data)

        if save_path:
            report_df.to_csv(save_path, index=False)
            logger.info(f"Report saved to {save_path}")

        return report_df

evaluate

evaluate(
    model: Module | ndarray | None = None,
    test_data: DataLoader | ndarray | tuple | None = None,
    return_predictions: bool = False,
    y_scores: ndarray | None = None,
) -> dict[str, float]

Evaluate model performance.

Parameters:

Name	Type	Description	Default
model	Module | ndarray | None	PyTorch model or numpy array of predictions	None
test_data	DataLoader | ndarray | tuple | None	Test data loader or (predictions, targets) tuple	None
return_predictions	bool	Whether to return predictions along with metrics	False
y_scores	ndarray | None	Scores/probabilities for AUC calculation (optional)	None

Returns:

Type	Description
dict[str, float]	Dictionary of metric values

Source code in deepecgkit/evaluation/evaluator.py

def evaluate(
    self,
    model: torch.nn.Module | np.ndarray | None = None,
    test_data: torch.utils.data.DataLoader | np.ndarray | tuple | None = None,
    return_predictions: bool = False,
    y_scores: np.ndarray | None = None,
) -> dict[str, float]:
    """
    Evaluate model performance.

    Args:
        model: PyTorch model or numpy array of predictions
        test_data: Test data loader or (predictions, targets) tuple
        return_predictions: Whether to return predictions along with metrics
        y_scores: Scores/probabilities for AUC calculation (optional)

    Returns:
        Dictionary of metric values
    """

    self.y_scores = y_scores

    predictions, targets = self._process_input_data(model, test_data)
    if targets is None:
        raise ValueError(
            "Evaluation needs the targets as well as the inputs: pass (inputs, targets), "
            "a DataLoader that yields both, or predictions together with their targets."
        )
    self.y_true = targets
    metrics = self._calculate_metrics(predictions, targets, self.y_scores)

    if return_predictions:
        return metrics, predictions
    return metrics

cross_validate

cross_validate(
    model_class: type,
    data: Any,
    k_folds: int = 5,
    groups: ndarray | None = None,
    epochs: int = 10,
    batch_size: int = 32,
    train_config: dict[str, Any] | None = None,
    seed: int = 42,
    **model_kwargs: Any,
) -> dict[str, list[float]]

Perform k-fold cross-validation.

A fresh model is built and fitted for every fold and scored on the fold held out. A torch module is trained with ECGTrainer; a tenth of its training fold is set aside for the learning-rate schedule and early stopping, so the held-out fold plays no part in training. Anything else with a fit method is fitted with fit(X, y) and asked for predict_proba, or failing that predict, unless it can be called directly.

Parameters:

Name	Type	Description	Default
model_class	type	Class of model to evaluate	required
data	Any	(X, y) tuple of arrays or tensors	required
k_folds	int	Number of folds	5
groups	ndarray | None	Group of each sample, such as the recording or patient it comes from. Folds then never split a group, which segments cut from longer recordings need for the estimate to be honest.	None
epochs	int	Maximum training epochs per fold, for a torch module	10
batch_size	int	Batch size for training and scoring a torch module	32
train_config	dict[str, Any] | None	ECGTrainer settings for a torch module. Multi-hot targets and a regression task_type are recognised without it; a single-output binary model needs binary_classification.	None
seed	int	Seed for shuffling the folds and the training batches	42
**model_kwargs	Any	Keyword arguments for model initialization	{}

Returns:

Type	Description
dict[str, list[float]]	Dictionary of metric scores across folds

Source code in deepecgkit/evaluation/evaluator.py

def cross_validate(
    self,
    model_class: type,
    data: Any,
    k_folds: int = 5,
    groups: np.ndarray | None = None,
    epochs: int = 10,
    batch_size: int = 32,
    train_config: dict[str, Any] | None = None,
    seed: int = 42,
    **model_kwargs: Any,
) -> dict[str, list[float]]:
    """
    Perform k-fold cross-validation.

    A fresh model is built and fitted for every fold and scored on the
    fold held out. A torch module is trained with ECGTrainer; a tenth of
    its training fold is set aside for the learning-rate schedule and
    early stopping, so the held-out fold plays no part in training.
    Anything else with a fit method is fitted with fit(X, y) and asked for
    predict_proba, or failing that predict, unless it can be called
    directly.

    Args:
        model_class: Class of model to evaluate
        data: (X, y) tuple of arrays or tensors
        k_folds: Number of folds
        groups: Group of each sample, such as the recording or patient it
            comes from. Folds then never split a group, which segments cut
            from longer recordings need for the estimate to be honest.
        epochs: Maximum training epochs per fold, for a torch module
        batch_size: Batch size for training and scoring a torch module
        train_config: ECGTrainer settings for a torch module. Multi-hot
            targets and a regression task_type are recognised without it;
            a single-output binary model needs binary_classification.
        seed: Seed for shuffling the folds and the training batches
        **model_kwargs: Keyword arguments for model initialization

    Returns:
        Dictionary of metric scores across folds
    """
    if not isinstance(data, tuple) or len(data) != 2:
        raise ValueError("Data must be (X, y) tuple for cross-validation")
    x, y = (np.asarray(part.cpu() if torch.is_tensor(part) else part) for part in data)
    if groups is not None:
        groups = np.asarray(groups)

    fold_results: dict[str, list[float]] = {metric: [] for metric in self.metrics}

    folds = self._folds(len(x), k_folds, groups, seed)
    for fold, (train_idx, val_idx) in enumerate(folds):
        logger.info(f"Evaluating fold {fold + 1}/{k_folds}")
        x_train, y_train = x[train_idx], y[train_idx]
        x_val, y_val = x[val_idx], y[val_idx]

        model = model_class(**model_kwargs)
        if isinstance(model, torch.nn.Module):
            fold_groups = None if groups is None else groups[train_idx]
            self._train_fold(
                model, x_train, y_train, fold_groups, epochs, batch_size, train_config, seed
            )
            held_out = _ArrayLoaders((x_val, y_val), (x_val, y_val), batch_size, seed)
            results = self.evaluate(model, held_out.val_dataloader())
        else:
            results = self._fit_and_score(model, x_train, y_train, x_val, y_val)

        for metric in self.metrics:
            if metric in results:
                fold_results[metric].append(results[metric])

    return fold_results

bootstrap_evaluate

bootstrap_evaluate(
    model: Any,
    test_data: Any,
    n_bootstrap: int = 1000,
    confidence_level: float = 0.95,
    y_scores: ndarray | None = None,
) -> dict[str, dict[str, float]]

Bootstrap evaluation for confidence intervals.

Takes the same inputs as evaluate, and computes each metric the same way, so an interval describes the value evaluate reports. Resampling draws from numpy's global generator; seed it for repeatable intervals.

Parameters:

Name	Type	Description	Default
model	Any	PyTorch model or numpy array of predictions	required
test_data	Any	Test data loader or (predictions, targets) tuple	required
n_bootstrap	int	Number of bootstrap samples	1000
confidence_level	float	Confidence level for intervals	0.95
y_scores	ndarray | None	Scores/probabilities for AUC calculation (optional)	None

Returns:

Type	Description
dict[str, dict[str, float]]	Dictionary with mean, std, and confidence intervals for each metric

Source code in deepecgkit/evaluation/evaluator.py

def bootstrap_evaluate(
    self,
    model: Any,
    test_data: Any,
    n_bootstrap: int = 1000,
    confidence_level: float = 0.95,
    y_scores: np.ndarray | None = None,
) -> dict[str, dict[str, float]]:
    """
    Bootstrap evaluation for confidence intervals.

    Takes the same inputs as evaluate, and computes each metric the same
    way, so an interval describes the value evaluate reports. Resampling
    draws from numpy's global generator; seed it for repeatable intervals.

    Args:
        model: PyTorch model or numpy array of predictions
        test_data: Test data loader or (predictions, targets) tuple
        n_bootstrap: Number of bootstrap samples
        confidence_level: Confidence level for intervals
        y_scores: Scores/probabilities for AUC calculation (optional)

    Returns:
        Dictionary with mean, std, and confidence intervals for each metric
    """
    self.y_scores = y_scores
    predictions, targets = self._process_input_data(model, test_data)
    return self._bootstrap(predictions, targets, self.y_scores, n_bootstrap, confidence_level)

generate_report

generate_report(
    model: Any,
    test_data: Any,
    save_path: str | None = None,
    y_scores: ndarray | None = None,
    calibration_fit: tuple | None = None,
) -> pd.DataFrame

Generate comprehensive evaluation report.

Calibration is always reported: expected calibration error, its equal-mass variant, maximum calibration error and the Brier score. Pass calibration_fit to additionally fit a temperature and report what rescaling recovers, which adds a "temperature" row and the "_temp" counterparts of each calibration metric.

Parameters:

Name	Type	Description	Default
model	Any	Trained model	required
test_data	Any	Test dataset	required
save_path	str | None	Path to save report (optional)	None
y_scores	ndarray | None	Scores/probabilities for AUC calculation (optional)	None
calibration_fit	tuple | None	Optional (targets, logits) pair from a held-out validation split, used to fit the temperature. Never pass the split being reported on. Callers holding probabilities rather than logits can convert with deepecgkit.evaluation.probabilities_to_logits.	None

Returns:

Type	Description
DataFrame	DataFrame with evaluation results

Source code in deepecgkit/evaluation/evaluator.py

def generate_report(
    self,
    model: Any,
    test_data: Any,
    save_path: str | None = None,
    y_scores: np.ndarray | None = None,
    calibration_fit: tuple | None = None,
) -> pd.DataFrame:
    """
    Generate comprehensive evaluation report.

    Calibration is always reported: expected calibration error, its
    equal-mass variant, maximum calibration error and the Brier score. Pass
    calibration_fit to additionally fit a temperature and report what
    rescaling recovers, which adds a "temperature" row and the "_temp"
    counterparts of each calibration metric.

    Args:
        model: Trained model
        test_data: Test dataset
        save_path: Path to save report (optional)
        y_scores: Scores/probabilities for AUC calculation (optional)
        calibration_fit: Optional (targets, logits) pair from a held-out
            validation split, used to fit the temperature. Never pass the
            split being reported on. Callers holding probabilities rather
            than logits can convert with
            deepecgkit.evaluation.probabilities_to_logits.

    Returns:
        DataFrame with evaluation results
    """

    results = self.evaluate(model, test_data, return_predictions=True, y_scores=y_scores)
    if isinstance(results, tuple):
        metrics_dict, predictions = results[0], results[1]
    else:
        metrics_dict, predictions = results, None

    try:
        bootstrap_results = self._bootstrap(predictions, self.y_true, self.y_scores)
    except Exception as error:
        logger.warning(f"Could not compute bootstrap confidence intervals: {error}")
        bootstrap_results = {}

    report_data = []

    for metric in self.metrics:
        if metric in metrics_dict and metric not in {"predictions", "targets"}:
            row = {
                "Metric": metric,
                "Value": metrics_dict[metric],
            }

            if metric in bootstrap_results:
                row.update(
                    {
                        "Std": bootstrap_results[metric]["std"],
                        "95% CI Lower": bootstrap_results[metric]["lower_ci"],
                        "95% CI Upper": bootstrap_results[metric]["upper_ci"],
                    }
                )

            report_data.append(row)

    report_data.extend(self._calibration_rows(predictions, calibration_fit))

    report_df = pd.DataFrame(report_data)

    if save_path:
        report_df.to_csv(save_path, index=False)
        logger.info(f"Report saved to {save_path}")

    return report_df

Metrics

calculate_classification_metrics

calculate_classification_metrics(
    y_true: ndarray,
    y_pred: ndarray,
    metrics: list[str] | None = None,
    average: str = "auto",
) -> dict[str, float]

Calculate classification metrics.

Multi-hot targets of shape (n_samples, n_labels) make the problem multi-label, and the metrics are defined as deepecg evaluate reports them: predictions are probabilities thresholded at 0.5 (or 0/1 labels), "accuracy" is the share of individual labels predicted correctly, as ECGTrainer reports it, "exact_match_accuracy" the share of samples with every label right, and precision, recall, F1 and AUC are macro averages over the labels. MCC is not defined for them.

Parameters:

Name	Type	Description	Default
y_true	ndarray	True labels, or multi-hot targets	required
y_pred	ndarray	Predicted labels or probabilities	required
metrics	list[str] | None	List of metrics to compute, from accuracy, label_accuracy, exact_match_accuracy, precision, recall, f1, auc and mcc. The two longer accuracy names equal accuracy for single-label targets.	None
average	str	Averaging for precision/recall/f1. "auto" selects positive-class values for two-class problems and macro averaging otherwise. Pass an explicit mode to make the choice independent of the label set. Use resolve_average to obtain the mode actually applied.	'auto'

Returns:

Type	Description
dict[str, float]	Dictionary of computed metrics

Source code in deepecgkit/evaluation/metrics.py

def calculate_classification_metrics(
    y_true: np.ndarray,
    y_pred: np.ndarray,
    metrics: list[str] | None = None,
    average: str = "auto",
) -> dict[str, float]:
    """
    Calculate classification metrics.

    Multi-hot targets of shape (n_samples, n_labels) make the problem
    multi-label, and the metrics are defined as `deepecg evaluate` reports
    them: predictions are probabilities thresholded at 0.5 (or 0/1 labels),
    "accuracy" is the share of individual labels predicted correctly, as
    ECGTrainer reports it, "exact_match_accuracy" the share of samples with
    every label right, and precision, recall, F1 and AUC are macro averages
    over the labels. MCC is not defined for them.

    Args:
        y_true: True labels, or multi-hot targets
        y_pred: Predicted labels or probabilities
        metrics: List of metrics to compute, from accuracy, label_accuracy,
            exact_match_accuracy, precision, recall, f1, auc and mcc. The two
            longer accuracy names equal accuracy for single-label targets.
        average: Averaging for precision/recall/f1. "auto" selects positive-class
            values for two-class problems and macro averaging otherwise. Pass an
            explicit mode to make the choice independent of the label set. Use
            resolve_average to obtain the mode actually applied.

    Returns:
        Dictionary of computed metrics
    """
    multilabel = np.ndim(y_true) > 1
    if metrics is None:
        metrics = ["accuracy", "precision", "recall", "f1", "auc"]
        metrics.append("exact_match_accuracy" if multilabel else "mcc")

    results = {}

    if multilabel:
        y_true = np.asarray(y_true).astype(int)
        is_score = np.issubdtype(y_pred.dtype, np.floating)
        y_pred_labels = (y_pred > BINARY_THRESHOLD).astype(int) if is_score else y_pred
        y_scores = y_pred
    elif y_pred.ndim > 1:
        y_pred_labels = np.argmax(y_pred, axis=1)
        y_scores = y_pred
    elif (
        np.issubdtype(y_pred.dtype, np.floating)
        and np.max(y_pred) <= 1.0
        and len(np.unique(y_true)) <= MULTICLASS_THRESHOLD
    ):
        y_pred_labels = (y_pred > BINARY_THRESHOLD).astype(int)
        y_scores = y_pred
    else:
        y_pred_labels = y_pred.astype(int)
        y_scores = y_pred

    n_classes = len(np.unique(y_true))
    avg = resolve_average(y_true, average)

    def _compute_metric(metric: str) -> Any:
        result = np.nan
        if metric in ACCURACY_METRICS:
            if multilabel:
                correct = y_pred_labels == y_true
                exact = metric == "exact_match_accuracy"
                result = correct.all(axis=1).mean() if exact else correct.mean()
            else:
                result = accuracy_score(y_true, y_pred_labels)
        elif metric in ["precision", "recall", "f1"]:
            precision, recall, f1, _ = precision_recall_fscore_support(
                y_true, y_pred_labels, average=avg, zero_division=0
            )
            if metric == "precision":
                result = precision
            elif metric == "recall":
                result = recall
            elif metric == "f1":
                result = f1
        elif metric == "auc":
            if multilabel:
                result = roc_auc_score(y_true, y_scores, average="macro")
            elif n_classes > MULTICLASS_THRESHOLD:
                result = roc_auc_score(y_true, y_scores, multi_class="ovr")
            else:
                # Binary AUC expects 1D scores (positive class probability)
                scores = y_scores[:, 1] if y_scores.ndim > 1 else y_scores
                result = roc_auc_score(y_true, scores)
        elif metric == "mcc":
            if multilabel:
                raise ValueError("MCC is not defined for multi-label targets")
            result = matthews_corrcoef(y_true, y_pred_labels)
        return result

    for metric in metrics:
        try:
            results[metric] = _compute_metric(metric)
        except Exception as e:
            logger.warning(f"Could not compute {metric}: {e}")
            results[metric] = np.nan

    return results

calculate_regression_metrics

calculate_regression_metrics(
    y_true: ndarray,
    y_pred: ndarray,
    metrics: list[str] | None = None,
) -> dict[str, float]

Calculate regression metrics.

Parameters:

Name	Type	Description	Default
y_true	ndarray	True values	required
y_pred	ndarray	Predicted values	required
metrics	list[str] | None	List of metrics to compute	None

Returns:

Type	Description
dict[str, float]	Dictionary of computed metrics

Source code in deepecgkit/evaluation/metrics.py

def calculate_regression_metrics(
    y_true: np.ndarray, y_pred: np.ndarray, metrics: list[str] | None = None
) -> dict[str, float]:
    """
    Calculate regression metrics.

    Args:
        y_true: True values
        y_pred: Predicted values
        metrics: List of metrics to compute

    Returns:
        Dictionary of computed metrics
    """
    if metrics is None:
        metrics = ["mse", "mae", "r2"]

    results = {}

    for metric in metrics:
        try:
            if metric in ["mse", "mean_squared_error"]:
                results[metric] = mean_squared_error(y_true, y_pred)
            elif metric in ["mae", "mean_absolute_error"]:
                results[metric] = mean_absolute_error(y_true, y_pred)
            elif metric in ["r2", "r2_score"]:
                results[metric] = r2_score(y_true, y_pred)
            elif metric == "rmse":
                results[metric] = np.sqrt(mean_squared_error(y_true, y_pred))
        except Exception as e:
            logger.warning(f"Could not compute {metric}: {e}")
            results[metric] = np.nan

    return results

confusion_matrix_analysis

confusion_matrix_analysis(
    y_true: ndarray, y_pred: ndarray
) -> dict[str, Any]

Detailed confusion matrix analysis.

Parameters:

Name	Type	Description	Default
y_true	ndarray	True labels	required
y_pred	ndarray	Predicted labels	required

Returns:

Type	Description
dict[str, Any]	Dictionary with confusion matrix and derived metrics

Source code in deepecgkit/evaluation/metrics.py

def confusion_matrix_analysis(y_true: np.ndarray, y_pred: np.ndarray) -> dict[str, Any]:
    """
    Detailed confusion matrix analysis.

    Args:
        y_true: True labels
        y_pred: Predicted labels

    Returns:
        Dictionary with confusion matrix and derived metrics
    """
    cm = confusion_matrix(y_true, y_pred)

    n_classes = cm.shape[0]
    precision = np.zeros(n_classes)
    recall = np.zeros(n_classes)
    f1 = np.zeros(n_classes)

    for i in range(n_classes):
        if cm[:, i].sum() > 0:
            precision[i] = cm[i, i] / cm[:, i].sum()
        if cm[i, :].sum() > 0:
            recall[i] = cm[i, i] / cm[i, :].sum()
        if precision[i] + recall[i] > 0:
            f1[i] = 2 * precision[i] * recall[i] / (precision[i] + recall[i])

    return {
        "confusion_matrix": cm,
        "per_class_precision": precision,
        "per_class_recall": recall,
        "per_class_f1": f1,
        "macro_precision": np.mean(precision),
        "macro_recall": np.mean(recall),
        "macro_f1": np.mean(f1),
    }

resolve_average

resolve_average(
    y_true: ndarray, average: str = "auto"
) -> str

Resolve the averaging mode applied to precision, recall and F1.

Precision, recall and F1 are not comparable across averaging modes, so the mode in force must be recorded alongside the values rather than inferred. Under "auto" a two-class problem yields positive-class ("binary") values, not macro-averaged ones. Multi-hot targets are macro-averaged over their labels.

Parameters:

Name	Type	Description	Default
y_true	ndarray	True labels, used only to count classes under "auto"	required
average	str	One of "auto", "binary", "macro", "micro", "weighted"	'auto'

Returns:

Type	Description
str	The concrete averaging mode to pass to scikit-learn.

Source code in deepecgkit/evaluation/metrics.py

def resolve_average(y_true: np.ndarray, average: str = "auto") -> str:
    """Resolve the averaging mode applied to precision, recall and F1.

    Precision, recall and F1 are not comparable across averaging modes, so the
    mode in force must be recorded alongside the values rather than inferred.
    Under "auto" a two-class problem yields positive-class ("binary") values,
    not macro-averaged ones. Multi-hot targets are macro-averaged over their
    labels.

    Args:
        y_true: True labels, used only to count classes under "auto"
        average: One of "auto", "binary", "macro", "micro", "weighted"

    Returns:
        The concrete averaging mode to pass to scikit-learn.
    """
    if average != "auto":
        return average
    if np.ndim(y_true) > 1:
        return "macro"
    return "binary" if len(np.unique(y_true)) <= MULTICLASS_THRESHOLD else "macro"

Calibration

Calibration measures whether predicted probabilities match observed frequencies, and temperature scaling corrects them without changing which class is predicted.

calculate_calibration_metrics

calculate_calibration_metrics(
    y_true: ndarray,
    y_prob: ndarray,
    n_bins: int = DEFAULT_CALIBRATION_BINS,
    task: str | None = None,
) -> dict[str, Any]

Calculate every calibration metric in one pass.

The bin count and resolved task shape are returned alongside the values because neither ECE nor MCE means anything without them, and a table that records only the number invites comparison between runs binned differently.

Parameters:

Name	Type	Description	Default
y_true	ndarray	True labels, (n_samples,) or multi-hot (n_samples, n_classes)	required
y_prob	ndarray	Predicted probabilities, (n_samples,) or (n_samples, n_classes)	required
n_bins	int	Number of bins for the binned metrics	DEFAULT_CALIBRATION_BINS
task	str | None	Explicit task shape, otherwise inferred from the array shapes	None

Returns:

Type	Description
dict[str, Any]	Dictionary with "ece", "adaptive_ece", "mce" and "brier", the context
dict[str, Any]	keys "n_bins" and "task", and for multi-label inputs the per-class
dict[str, Any]	arrays "per_class_ece", "per_class_mce" and "per_class_brier".

Source code in deepecgkit/evaluation/metrics.py

def calculate_calibration_metrics(
    y_true: np.ndarray,
    y_prob: np.ndarray,
    n_bins: int = DEFAULT_CALIBRATION_BINS,
    task: str | None = None,
) -> dict[str, Any]:
    """Calculate every calibration metric in one pass.

    The bin count and resolved task shape are returned alongside the values
    because neither ECE nor MCE means anything without them, and a table that
    records only the number invites comparison between runs binned differently.

    Args:
        y_true: True labels, (n_samples,) or multi-hot (n_samples, n_classes)
        y_prob: Predicted probabilities, (n_samples,) or (n_samples, n_classes)
        n_bins: Number of bins for the binned metrics
        task: Explicit task shape, otherwise inferred from the array shapes

    Returns:
        Dictionary with "ece", "adaptive_ece", "mce" and "brier", the context
        keys "n_bins" and "task", and for multi-label inputs the per-class
        arrays "per_class_ece", "per_class_mce" and "per_class_brier".
    """
    results: dict[str, Any] = {"n_bins": int(n_bins), "task": None}
    try:
        results["task"] = resolve_task_shape(y_true, y_prob, task)
    except ValueError as error:
        logger.warning(f"Could not resolve task shape: {error}")
        for name in CALIBRATION_METRIC_NAMES:
            results[name] = np.nan
        return results

    try:
        uniform = _calibration_errors(y_true, y_prob, n_bins, BINNING_UNIFORM, results["task"])
        results["ece"] = uniform["ece"]
        results["mce"] = uniform["mce"]
        if uniform["per_class_ece"] is not None:
            results["per_class_ece"] = uniform["per_class_ece"]
            results["per_class_mce"] = uniform["per_class_mce"]
    except Exception as error:
        logger.warning(f"Could not compute ece: {error}")
        results["ece"] = np.nan
        results["mce"] = np.nan

    try:
        results["adaptive_ece"] = adaptive_calibration_error(
            y_true, y_prob, n_bins=n_bins, task=results["task"]
        )
    except Exception as error:
        logger.warning(f"Could not compute adaptive_ece: {error}")
        results["adaptive_ece"] = np.nan

    try:
        results["brier"] = brier_score(y_true, y_prob, task=results["task"])
        if results["task"] == TASK_MULTILABEL:
            matrix = _as_probability_matrix(y_prob)
            targets = np.asarray(y_true, dtype=float)
            results["per_class_brier"] = ((matrix - targets) ** 2).mean(axis=0)
    except Exception as error:
        logger.warning(f"Could not compute brier: {error}")
        results["brier"] = np.nan

    return results

expected_calibration_error

expected_calibration_error(
    y_true: ndarray,
    y_prob: ndarray,
    n_bins: int = DEFAULT_CALIBRATION_BINS,
    task: str | None = None,
) -> float

Expected calibration error over equal-width probability bins.

The gap between confidence and accuracy is averaged over bins weighted by how many samples fall in each. ECE is sensitive to n_bins and the two numbers are not comparable across different bin counts, so n_bins must travel with any reported value. The default of 15 follows Guo et al. (2017), which keeps values comparable with the temperature-scaling literature; use adaptive_calibration_error for the companion number that does not depend on where the equal-width edges happen to fall.

Multi-label inputs are scored per class and averaged, never flattened.

Parameters:

Name	Type	Description	Default
y_true	ndarray	True labels, (n_samples,) or multi-hot (n_samples, n_classes)	required
y_prob	ndarray	Predicted probabilities, (n_samples,) or (n_samples, n_classes)	required
n_bins	int	Number of equal-width bins spanning [0, 1]	DEFAULT_CALIBRATION_BINS
task	str | None	Explicit task shape, otherwise inferred from the array shapes	None

Returns:

Type	Description
float	The error in [0, 1], or NaN when no bin is populated.

Source code in deepecgkit/evaluation/metrics.py

def expected_calibration_error(
    y_true: np.ndarray,
    y_prob: np.ndarray,
    n_bins: int = DEFAULT_CALIBRATION_BINS,
    task: str | None = None,
) -> float:
    """Expected calibration error over equal-width probability bins.

    The gap between confidence and accuracy is averaged over bins weighted by
    how many samples fall in each. ECE is sensitive to n_bins and the two
    numbers are not comparable across different bin counts, so n_bins must
    travel with any reported value. The default of 15 follows Guo et al.
    (2017), which keeps values comparable with the temperature-scaling
    literature; use adaptive_calibration_error for the companion number that
    does not depend on where the equal-width edges happen to fall.

    Multi-label inputs are scored per class and averaged, never flattened.

    Args:
        y_true: True labels, (n_samples,) or multi-hot (n_samples, n_classes)
        y_prob: Predicted probabilities, (n_samples,) or (n_samples, n_classes)
        n_bins: Number of equal-width bins spanning [0, 1]
        task: Explicit task shape, otherwise inferred from the array shapes

    Returns:
        The error in [0, 1], or NaN when no bin is populated.
    """
    return _calibration_errors(y_true, y_prob, n_bins, BINNING_UNIFORM, task)["ece"]

adaptive_calibration_error

adaptive_calibration_error(
    y_true: ndarray,
    y_prob: ndarray,
    n_bins: int = DEFAULT_CALIBRATION_BINS,
    task: str | None = None,
) -> float

Expected calibration error over equal-mass probability bins.

Equal-width bins are usually near-empty at the low-confidence end and crowded at the top, so an equal-width bin holding a handful of samples estimates its gap from almost nothing. Equal-mass bins put the same number of samples in every bin, so no bin's gap rests on a tiny sample.

Report this beside the equal-width number rather than instead of it. The two are not reliably ordered and neither dominates: where they disagree, the disagreement is itself evidence that the binning is doing the work rather than the model.

Parameters:

Name	Type	Description	Default
y_true	ndarray	True labels, (n_samples,) or multi-hot (n_samples, n_classes)	required
y_prob	ndarray	Predicted probabilities, (n_samples,) or (n_samples, n_classes)	required
n_bins	int	Number of equal-mass bins	DEFAULT_CALIBRATION_BINS
task	str | None	Explicit task shape, otherwise inferred from the array shapes	None

Returns:

Type	Description
float	The error in [0, 1], or NaN when there are no samples.

Source code in deepecgkit/evaluation/metrics.py

def adaptive_calibration_error(
    y_true: np.ndarray,
    y_prob: np.ndarray,
    n_bins: int = DEFAULT_CALIBRATION_BINS,
    task: str | None = None,
) -> float:
    """Expected calibration error over equal-mass probability bins.

    Equal-width bins are usually near-empty at the low-confidence end and
    crowded at the top, so an equal-width bin holding a handful of samples
    estimates its gap from almost nothing. Equal-mass bins put the same number
    of samples in every bin, so no bin's gap rests on a tiny sample.

    Report this beside the equal-width number rather than instead of it. The
    two are not reliably ordered and neither dominates: where they disagree,
    the disagreement is itself evidence that the binning is doing the work
    rather than the model.

    Args:
        y_true: True labels, (n_samples,) or multi-hot (n_samples, n_classes)
        y_prob: Predicted probabilities, (n_samples,) or (n_samples, n_classes)
        n_bins: Number of equal-mass bins
        task: Explicit task shape, otherwise inferred from the array shapes

    Returns:
        The error in [0, 1], or NaN when there are no samples.
    """
    return _calibration_errors(y_true, y_prob, n_bins, BINNING_QUANTILE, task)["ece"]

maximum_calibration_error

maximum_calibration_error(
    y_true: ndarray,
    y_prob: ndarray,
    n_bins: int = DEFAULT_CALIBRATION_BINS,
    task: str | None = None,
) -> float

Largest confidence-accuracy gap across equal-width bins.

MCE reports the worst bin rather than the average one, so it answers how far the model can be trusted at its least reliable confidence level. It is noisier than ECE because a sparsely populated bin can set it, and it is never smaller than the ECE computed over the same bins. Multi-label inputs take the worst class, since averaging worst cases would understate them.

Parameters:

Name	Type	Description	Default
y_true	ndarray	True labels, (n_samples,) or multi-hot (n_samples, n_classes)	required
y_prob	ndarray	Predicted probabilities, (n_samples,) or (n_samples, n_classes)	required
n_bins	int	Number of equal-width bins spanning [0, 1]	DEFAULT_CALIBRATION_BINS
task	str | None	Explicit task shape, otherwise inferred from the array shapes	None

Returns:

Type	Description
float	The error in [0, 1], or NaN when no bin is populated.

Source code in deepecgkit/evaluation/metrics.py

def maximum_calibration_error(
    y_true: np.ndarray,
    y_prob: np.ndarray,
    n_bins: int = DEFAULT_CALIBRATION_BINS,
    task: str | None = None,
) -> float:
    """Largest confidence-accuracy gap across equal-width bins.

    MCE reports the worst bin rather than the average one, so it answers how
    far the model can be trusted at its least reliable confidence level. It is
    noisier than ECE because a sparsely populated bin can set it, and it is
    never smaller than the ECE computed over the same bins. Multi-label inputs
    take the worst class, since averaging worst cases would understate them.

    Args:
        y_true: True labels, (n_samples,) or multi-hot (n_samples, n_classes)
        y_prob: Predicted probabilities, (n_samples,) or (n_samples, n_classes)
        n_bins: Number of equal-width bins spanning [0, 1]
        task: Explicit task shape, otherwise inferred from the array shapes

    Returns:
        The error in [0, 1], or NaN when no bin is populated.
    """
    return _calibration_errors(y_true, y_prob, n_bins, BINNING_UNIFORM, task)["mce"]

brier_score

brier_score(
    y_true: ndarray,
    y_prob: ndarray,
    task: str | None = None,
) -> float

Mean squared error of the predicted probability vector.

Two conventions are in use and this function applies the one that is standard for each shape, so the values are not comparable between shapes. Multi-class and binary problems use the original sum-over-classes form, which ranges over [0, 2]. Multi-label problems average the per-class binary Brier score, which ranges over [0, 1]. Tables are emitted per dataset, so a single table never mixes the two, but a figure spanning datasets would.

Unlike ECE this is a proper scoring rule: it rewards sharpness as well as calibration, so a model can improve its ECE while its Brier score worsens.

Parameters:

Name	Type	Description	Default
y_true	ndarray	True labels, (n_samples,) or multi-hot (n_samples, n_classes)	required
y_prob	ndarray	Predicted probabilities, (n_samples,) or (n_samples, n_classes)	required
task	str | None	Explicit task shape, otherwise inferred from the array shapes	None

Returns:

Type	Description
float	The score, lower being better, or NaN when there are no samples.

Source code in deepecgkit/evaluation/metrics.py

def brier_score(y_true: np.ndarray, y_prob: np.ndarray, task: str | None = None) -> float:
    """Mean squared error of the predicted probability vector.

    Two conventions are in use and this function applies the one that is
    standard for each shape, so the values are not comparable between shapes.
    Multi-class and binary problems use the original sum-over-classes form,
    which ranges over [0, 2]. Multi-label problems average the per-class binary
    Brier score, which ranges over [0, 1]. Tables are emitted per dataset, so a
    single table never mixes the two, but a figure spanning datasets would.

    Unlike ECE this is a proper scoring rule: it rewards sharpness as well as
    calibration, so a model can improve its ECE while its Brier score worsens.

    Args:
        y_true: True labels, (n_samples,) or multi-hot (n_samples, n_classes)
        y_prob: Predicted probabilities, (n_samples,) or (n_samples, n_classes)
        task: Explicit task shape, otherwise inferred from the array shapes

    Returns:
        The score, lower being better, or NaN when there are no samples.
    """
    shape = resolve_task_shape(y_true, y_prob, task)
    matrix = _as_probability_matrix(y_prob)
    if matrix.shape[0] == 0:
        return np.nan

    if shape == TASK_MULTILABEL:
        targets = np.asarray(y_true, dtype=float)
        return float(((matrix - targets) ** 2).mean(axis=0).mean())

    labels = np.asarray(y_true).astype(int).ravel()
    onehot = np.zeros_like(matrix)
    onehot[np.arange(labels.shape[0]), labels] = 1.0
    return float(((matrix - onehot) ** 2).sum(axis=1).mean())

resolve_task_shape

resolve_task_shape(
    y_true: ndarray,
    y_prob: ndarray,
    task: str | None = None,
) -> str

Resolve which of the three prediction shapes produced these arrays.

ECGTrainer emits exactly three shapes and the correct calibration definition differs between them, so the shape is resolved once and carried down rather than re-sniffed at each step. A two-column probability matrix is reported as "binary" whether it came from a single sigmoid logit stacked into [1 - p, p] or from a two-class softmax: the two are indistinguishable from the arrays alone and yield identical top-label statistics anyway.

Parameters:

Name	Type	Description	Default
y_true	ndarray	True labels, (n_samples,) for single-label problems or (n_samples, n_classes) multi-hot for multi-label ones	required
y_prob	ndarray	Predicted probabilities, (n_samples,) or (n_samples, n_classes)	required
task	str | None	Explicit override, one of "binary", "multiclass", "multilabel"	None

Returns:

Type	Description
str	One of TASK_BINARY, TASK_MULTICLASS or TASK_MULTILABEL.

Raises:

Type	Description
ValueError	If task is supplied but is not a recognised shape.

Source code in deepecgkit/evaluation/metrics.py

def resolve_task_shape(y_true: np.ndarray, y_prob: np.ndarray, task: str | None = None) -> str:
    """Resolve which of the three prediction shapes produced these arrays.

    ECGTrainer emits exactly three shapes and the correct calibration
    definition differs between them, so the shape is resolved once and carried
    down rather than re-sniffed at each step. A two-column probability matrix
    is reported as "binary" whether it came from a single sigmoid logit stacked
    into [1 - p, p] or from a two-class softmax: the two are indistinguishable
    from the arrays alone and yield identical top-label statistics anyway.

    Args:
        y_true: True labels, (n_samples,) for single-label problems or
            (n_samples, n_classes) multi-hot for multi-label ones
        y_prob: Predicted probabilities, (n_samples,) or (n_samples, n_classes)
        task: Explicit override, one of "binary", "multiclass", "multilabel"

    Returns:
        One of TASK_BINARY, TASK_MULTICLASS or TASK_MULTILABEL.

    Raises:
        ValueError: If task is supplied but is not a recognised shape.
    """
    if task is not None:
        if task not in (TASK_BINARY, TASK_MULTICLASS, TASK_MULTILABEL):
            raise ValueError(f"Unknown task shape: {task}")
        return task

    y_true = np.asarray(y_true)
    y_prob = np.asarray(y_prob)
    if y_true.ndim > 1 and y_true.shape[-1] > 1:
        return TASK_MULTILABEL
    if y_prob.ndim == 1 or y_prob.shape[-1] <= MULTICLASS_THRESHOLD:
        return TASK_BINARY
    return TASK_MULTICLASS

Recalibration

TemperatureScaler

Single-parameter recalibrator fitted on one split and applied to another.

The scaler holds one scalar. Fitting it on the validation split and calling transform on the test split is the supported use; fitting and reporting on the same split measures nothing.

Parameters:

Name	Type	Description	Default
task	str | None	Task shape, one of the TASK_ constants in deepecgkit.evaluation.metrics. Inferred from the labels at fit time when omitted, which is reliable because multi-label targets are the only two-dimensional ones.	None
bounds	tuple[float, float]	Search interval for the temperature	DEFAULT_TEMPERATURE_BOUNDS

Examples:

>>> scaler = TemperatureScaler().fit(val_logits, val_targets)
>>> test_probabilities = scaler.transform(test_logits)
>>> round(scaler.temperature, 3)
1.42

Source code in deepecgkit/evaluation/calibration.py

class TemperatureScaler:
    """Single-parameter recalibrator fitted on one split and applied to another.

    The scaler holds one scalar. Fitting it on the validation split and calling
    transform on the test split is the supported use; fitting and reporting on
    the same split measures nothing.

    Args:
        task: Task shape, one of the TASK_ constants in
            deepecgkit.evaluation.metrics. Inferred from the labels at fit time
            when omitted, which is reliable because multi-label targets are the
            only two-dimensional ones.
        bounds: Search interval for the temperature

    Examples:
        >>> scaler = TemperatureScaler().fit(val_logits, val_targets)
        >>> test_probabilities = scaler.transform(test_logits)
        >>> round(scaler.temperature, 3)
        1.42
    """

    def __init__(
        self,
        task: str | None = None,
        bounds: tuple[float, float] = DEFAULT_TEMPERATURE_BOUNDS,
    ):
        self.task = task
        self.bounds = bounds
        self.temperature = NEUTRAL_TEMPERATURE
        self.fitted = False

    def fit(self, logits: np.ndarray, y_true: np.ndarray) -> "TemperatureScaler":
        """Fit the temperature on a held-out split.

        Args:
            logits: Raw model outputs on the fitting split, which should be
                validation data
            y_true: True labels for that split

        Returns:
            This scaler, so the call can be chained.
        """
        if self.task is None:
            self.task = TASK_MULTILABEL if np.asarray(y_true).ndim > 1 else None
        self.temperature = fit_temperature(logits, y_true, task=self.task, bounds=self.bounds)
        self.fitted = True
        return self

    def transform(self, logits: np.ndarray) -> np.ndarray:
        """Apply the fitted temperature to a different split.

        Args:
            logits: Raw model outputs to recalibrate

        Returns:
            Probabilities in the same layout ECGTrainer stores.

        Raises:
            RuntimeError: If called before fit.
        """
        if not self.fitted:
            raise RuntimeError("TemperatureScaler must be fitted before transform")
        return apply_temperature(logits, self.temperature, task=self.task)

    def nll(self, logits: np.ndarray, y_true: np.ndarray) -> float:
        """Negative log-likelihood on a split under the fitted temperature.

        Args:
            logits: Raw model outputs
            y_true: True labels

        Returns:
            The mean negative log-likelihood in nats.
        """
        return temperature_nll(logits, y_true, temperature=self.temperature, task=self.task)

fit

fit(logits: ndarray, y_true: ndarray) -> TemperatureScaler

Fit the temperature on a held-out split.

Parameters:

Name	Type	Description	Default
logits	ndarray	Raw model outputs on the fitting split, which should be validation data	required
y_true	ndarray	True labels for that split	required

Returns:

Type	Description
TemperatureScaler	This scaler, so the call can be chained.

Source code in deepecgkit/evaluation/calibration.py

def fit(self, logits: np.ndarray, y_true: np.ndarray) -> "TemperatureScaler":
    """Fit the temperature on a held-out split.

    Args:
        logits: Raw model outputs on the fitting split, which should be
            validation data
        y_true: True labels for that split

    Returns:
        This scaler, so the call can be chained.
    """
    if self.task is None:
        self.task = TASK_MULTILABEL if np.asarray(y_true).ndim > 1 else None
    self.temperature = fit_temperature(logits, y_true, task=self.task, bounds=self.bounds)
    self.fitted = True
    return self

transform

transform(logits: ndarray) -> np.ndarray

Apply the fitted temperature to a different split.

Parameters:

Name	Type	Description	Default
logits	ndarray	Raw model outputs to recalibrate	required

Returns:

Type	Description
ndarray	Probabilities in the same layout ECGTrainer stores.

Raises:

Type	Description
RuntimeError	If called before fit.

Source code in deepecgkit/evaluation/calibration.py

def transform(self, logits: np.ndarray) -> np.ndarray:
    """Apply the fitted temperature to a different split.

    Args:
        logits: Raw model outputs to recalibrate

    Returns:
        Probabilities in the same layout ECGTrainer stores.

    Raises:
        RuntimeError: If called before fit.
    """
    if not self.fitted:
        raise RuntimeError("TemperatureScaler must be fitted before transform")
    return apply_temperature(logits, self.temperature, task=self.task)

nll

nll(logits: ndarray, y_true: ndarray) -> float

Negative log-likelihood on a split under the fitted temperature.

Parameters:

Name	Type	Description	Default
logits	ndarray	Raw model outputs	required
y_true	ndarray	True labels	required

Returns:

Type	Description
float	The mean negative log-likelihood in nats.

Source code in deepecgkit/evaluation/calibration.py

def nll(self, logits: np.ndarray, y_true: np.ndarray) -> float:
    """Negative log-likelihood on a split under the fitted temperature.

    Args:
        logits: Raw model outputs
        y_true: True labels

    Returns:
        The mean negative log-likelihood in nats.
    """
    return temperature_nll(logits, y_true, temperature=self.temperature, task=self.task)

fit_temperature

fit_temperature(
    logits: ndarray,
    y_true: ndarray,
    task: str | None = None,
    bounds: tuple[
        float, float
    ] = DEFAULT_TEMPERATURE_BOUNDS,
) -> float

Fit the single temperature that minimises NLL on the given split.

Pass the validation split. The objective is one-dimensional and smooth, so a bounded scalar search is used rather than a gradient optimiser: it needs no autograd graph, no learning rate and no iteration count, and it returns the same answer every run.

A fitted temperature above one means the model was overconfident and its probabilities have been softened; below one means it was underconfident.

Parameters:

Name	Type	Description	Default
logits	ndarray	Raw model outputs on the fitting split	required
y_true	ndarray	True labels for that split	required
task	str | None	Pass TASK_MULTILABEL for per-class sigmoid outputs	None
bounds	tuple[float, float]	Search interval for the temperature	DEFAULT_TEMPERATURE_BOUNDS

Returns:

Type	Description
float	The fitted temperature, or 1.0 if the search failed, in which case a
float	warning is issued and the probabilities are left untouched.

Source code in deepecgkit/evaluation/calibration.py

def fit_temperature(
    logits: np.ndarray,
    y_true: np.ndarray,
    task: str | None = None,
    bounds: tuple[float, float] = DEFAULT_TEMPERATURE_BOUNDS,
) -> float:
    """Fit the single temperature that minimises NLL on the given split.

    Pass the validation split. The objective is one-dimensional and smooth, so
    a bounded scalar search is used rather than a gradient optimiser: it needs
    no autograd graph, no learning rate and no iteration count, and it returns
    the same answer every run.

    A fitted temperature above one means the model was overconfident and its
    probabilities have been softened; below one means it was underconfident.

    Args:
        logits: Raw model outputs on the fitting split
        y_true: True labels for that split
        task: Pass TASK_MULTILABEL for per-class sigmoid outputs
        bounds: Search interval for the temperature

    Returns:
        The fitted temperature, or 1.0 if the search failed, in which case a
        warning is issued and the probabilities are left untouched.
    """
    logits = np.asarray(logits, dtype=float)
    if logits.shape[0] == 0:
        warnings.warn(
            "Cannot fit a temperature on an empty split; leaving probabilities unscaled.",
            RuntimeWarning,
            stacklevel=2,
        )
        return NEUTRAL_TEMPERATURE

    def objective(temperature: float) -> float:
        return temperature_nll(logits, y_true, temperature=temperature, task=task)

    result = minimize_scalar(objective, bounds=bounds, method="bounded")

    if not result.success or not np.isfinite(result.x):
        warnings.warn(
            "Temperature search did not converge; leaving probabilities unscaled.",
            RuntimeWarning,
            stacklevel=2,
        )
        return NEUTRAL_TEMPERATURE

    temperature = float(result.x)
    if min(abs(temperature - bounds[0]), abs(temperature - bounds[1])) < BOUND_TOLERANCE:
        warnings.warn(
            f"Fitted temperature {temperature:.4g} sits on the search bound {bounds}, so the "
            "optimum lies outside it. This usually means the fitting split is degenerate, for "
            "example holding a single class.",
            RuntimeWarning,
            stacklevel=2,
        )
    return temperature

apply_temperature

apply_temperature(
    logits: ndarray,
    temperature: float,
    task: str | None = None,
) -> np.ndarray

Divide logits by a temperature and map them back to probabilities.

The returned layout matches what ECGTrainer stores, so scaled probabilities are a drop-in replacement for unscaled ones: a single logit per sample comes back as the two columns [1 - p, p].

Parameters:

Name	Type	Description	Default
logits	ndarray	Raw model outputs, (n_samples,), (n_samples, 1) or (n_samples, n_classes)	required
temperature	float	Positive scalar; values above one soften predictions	required
task	str | None	Pass TASK_MULTILABEL for per-class sigmoid outputs, which cannot be told apart from multi-class logits by shape alone	None

Returns:

Type	Description
ndarray	Probabilities of shape (n_samples, n_classes).

Raises:

Type	Description
ValueError	If temperature is not strictly positive.

Source code in deepecgkit/evaluation/calibration.py

def apply_temperature(
    logits: np.ndarray, temperature: float, task: str | None = None
) -> np.ndarray:
    """Divide logits by a temperature and map them back to probabilities.

    The returned layout matches what ECGTrainer stores, so scaled probabilities
    are a drop-in replacement for unscaled ones: a single logit per sample comes
    back as the two columns [1 - p, p].

    Args:
        logits: Raw model outputs, (n_samples,), (n_samples, 1) or
            (n_samples, n_classes)
        temperature: Positive scalar; values above one soften predictions
        task: Pass TASK_MULTILABEL for per-class sigmoid outputs, which cannot
            be told apart from multi-class logits by shape alone

    Returns:
        Probabilities of shape (n_samples, n_classes).

    Raises:
        ValueError: If temperature is not strictly positive.
    """
    if not temperature > 0:
        raise ValueError(f"Temperature must be positive, got {temperature}")

    scaled = np.asarray(logits, dtype=float) / float(temperature)

    if task == TASK_MULTILABEL:
        return _sigmoid(scaled)

    if _is_single_logit(scaled):
        positive = _sigmoid(scaled.ravel())
        return np.column_stack([1.0 - positive, positive])

    return _softmax(scaled)

temperature_nll

temperature_nll(
    logits: ndarray,
    y_true: ndarray,
    temperature: float = NEUTRAL_TEMPERATURE,
    task: str | None = None,
) -> float

Mean negative log-likelihood of the labels under a scaled model.

This is the quantity temperature scaling minimises. It is computed through logsumexp and logaddexp rather than by taking the log of a probability, so a saturated logit costs precision instead of producing an infinity.

Parameters:

Name	Type	Description	Default
logits	ndarray	Raw model outputs	required
y_true	ndarray	True labels, (n_samples,) or multi-hot (n_samples, n_classes)	required
temperature	float	Positive scalar to divide the logits by	NEUTRAL_TEMPERATURE
task	str | None	Pass TASK_MULTILABEL for per-class sigmoid outputs	None

Returns:

Type	Description
float	The mean negative log-likelihood in nats, lower being better.

Raises:

Type	Description
ValueError	If temperature is not strictly positive.

Source code in deepecgkit/evaluation/calibration.py

def temperature_nll(
    logits: np.ndarray,
    y_true: np.ndarray,
    temperature: float = NEUTRAL_TEMPERATURE,
    task: str | None = None,
) -> float:
    """Mean negative log-likelihood of the labels under a scaled model.

    This is the quantity temperature scaling minimises. It is computed through
    logsumexp and logaddexp rather than by taking the log of a probability, so
    a saturated logit costs precision instead of producing an infinity.

    Args:
        logits: Raw model outputs
        y_true: True labels, (n_samples,) or multi-hot (n_samples, n_classes)
        temperature: Positive scalar to divide the logits by
        task: Pass TASK_MULTILABEL for per-class sigmoid outputs

    Returns:
        The mean negative log-likelihood in nats, lower being better.

    Raises:
        ValueError: If temperature is not strictly positive.
    """
    if not temperature > 0:
        raise ValueError(f"Temperature must be positive, got {temperature}")

    scaled = np.asarray(logits, dtype=float) / float(temperature)
    targets = np.asarray(y_true)

    if task == TASK_MULTILABEL or _is_single_logit(scaled):
        if task != TASK_MULTILABEL:
            scaled = scaled.ravel()
            targets = targets.ravel()
        signs = 2.0 * targets.astype(float) - 1.0
        return float(np.mean(np.logaddexp(0.0, -signs * scaled)))

    labels = targets.astype(int).ravel()
    rows = np.arange(labels.shape[0])
    return float(np.mean(logsumexp(scaled, axis=1) - scaled[rows, labels]))

probabilities_to_logits

probabilities_to_logits(
    y_prob: ndarray, task: str | None = None
) -> np.ndarray

Recover logits from stored probabilities, for callers who kept only those.

Prefer real logits where they are available. This reconstruction is exact in principle, because softmax is shift-invariant and so log(p) is a valid representative of the logit vector, and because log(p / (1 - p)) inverts the sigmoid outright. It is the float32 storage that costs accuracy: a probability rounds to exactly 1.0 once the logit reaches about 16.6, and clipping at 1e-7 caps recoverable magnitudes at about 16.1. Measured over 4000 samples, temperature-scaled probabilities rebuilt this way match those from true logits to about 4e-8 for softmax and 5e-5 for sigmoid at ordinary confidence, but drift to roughly 5e-2 once a fifth of the rows saturate. Confidently wrong models are exactly the ones a calibration study cares about, so reconstruction is a fallback and not the default path.

The probabilities written to predictions.txt carry six decimals, which caps recoverable magnitudes at about 13.8.

Parameters:

Name	Type	Description	Default
y_prob	ndarray	Probabilities of shape (n_samples,) or (n_samples, n_classes)	required
task	str | None	Pass TASK_MULTILABEL for per-class sigmoid outputs. Otherwise a one- or two-column input is read as a single logistic decision and anything wider as a softmax over classes.	None

Returns:

Type	Description
ndarray	Logits, shaped (n_samples,) for logistic inputs and
ndarray	(n_samples, n_classes) for softmax and multi-label inputs.

Source code in deepecgkit/evaluation/calibration.py

def probabilities_to_logits(y_prob: np.ndarray, task: str | None = None) -> np.ndarray:
    """Recover logits from stored probabilities, for callers who kept only those.

    Prefer real logits where they are available. This reconstruction is exact
    in principle, because softmax is shift-invariant and so log(p) is a valid
    representative of the logit vector, and because log(p / (1 - p)) inverts the
    sigmoid outright. It is the float32 storage that costs accuracy: a
    probability rounds to exactly 1.0 once the logit reaches about 16.6, and
    clipping at 1e-7 caps recoverable magnitudes at about 16.1. Measured over
    4000 samples, temperature-scaled probabilities rebuilt this way match those
    from true logits to about 4e-8 for softmax and 5e-5 for sigmoid at ordinary
    confidence, but drift to roughly 5e-2 once a fifth of the rows saturate.
    Confidently wrong models are exactly the ones a calibration study cares
    about, so reconstruction is a fallback and not the default path.

    The probabilities written to predictions.txt carry six decimals, which caps
    recoverable magnitudes at about 13.8.

    Args:
        y_prob: Probabilities of shape (n_samples,) or (n_samples, n_classes)
        task: Pass TASK_MULTILABEL for per-class sigmoid outputs. Otherwise a
            one- or two-column input is read as a single logistic decision and
            anything wider as a softmax over classes.

    Returns:
        Logits, shaped (n_samples,) for logistic inputs and
        (n_samples, n_classes) for softmax and multi-label inputs.
    """
    probabilities = np.clip(np.asarray(y_prob, dtype=float), LOGIT_CLIP_EPS, 1.0 - LOGIT_CLIP_EPS)

    if task == TASK_MULTILABEL:
        return np.log(probabilities) - np.log1p(-probabilities)

    if probabilities.ndim == 1:
        return np.log(probabilities) - np.log1p(-probabilities)

    if probabilities.shape[-1] <= 2:
        positive = probabilities[:, -1]
        return np.log(positive) - np.log1p(-positive)

    return np.log(probabilities)

calibration_report

calibration_report(
    y_true: ndarray,
    y_prob: ndarray,
    n_bins: int = DEFAULT_CALIBRATION_BINS,
    fit_targets: ndarray | None = None,
    fit_logits: ndarray | None = None,
    eval_logits: ndarray | None = None,
) -> dict[str, Any]

Calibration before and after temperature scaling, as one flat mapping.

This is the shape the paper table wants: one row per run holding both the uncalibrated numbers and what temperature scaling recovers. The scaled entries are NaN unless a fitting split is supplied, so a run that had no validation data still produces a well-formed row.

Pass the test split as y_true and y_prob and the validation split as fit_targets and fit_logits. The temperature is never fitted on the split it is reported on.

Parameters:

Name	Type	Description	Default
y_true	ndarray	True labels for the split being reported on	required
y_prob	ndarray	Predicted probabilities for that split	required
n_bins	int	Number of bins for the binned metrics	DEFAULT_CALIBRATION_BINS
fit_targets	ndarray | None	True labels for the validation split	None
fit_logits	ndarray | None	Raw model outputs for the validation split	None
eval_logits	ndarray | None	Raw model outputs for the reported split. Reconstructed from y_prob when omitted, which costs accuracy on saturated predictions; see probabilities_to_logits.	None

Returns:

Type	Description
dict[str, Any]	Mapping with "ece", "adaptive_ece", "mce", "brier", their "_temp"
dict[str, Any]	counterparts measured after scaling, the fitted "temperature", and
dict[str, Any]	"calibration_bins" recording what produced the binned values.

Source code in deepecgkit/evaluation/calibration.py

def calibration_report(
    y_true: np.ndarray,
    y_prob: np.ndarray,
    n_bins: int = DEFAULT_CALIBRATION_BINS,
    fit_targets: np.ndarray | None = None,
    fit_logits: np.ndarray | None = None,
    eval_logits: np.ndarray | None = None,
) -> dict[str, Any]:
    """Calibration before and after temperature scaling, as one flat mapping.

    This is the shape the paper table wants: one row per run holding both the
    uncalibrated numbers and what temperature scaling recovers. The scaled
    entries are NaN unless a fitting split is supplied, so a run that had no
    validation data still produces a well-formed row.

    Pass the test split as y_true and y_prob and the validation split as
    fit_targets and fit_logits. The temperature is never fitted on the split it
    is reported on.

    Args:
        y_true: True labels for the split being reported on
        y_prob: Predicted probabilities for that split
        n_bins: Number of bins for the binned metrics
        fit_targets: True labels for the validation split
        fit_logits: Raw model outputs for the validation split
        eval_logits: Raw model outputs for the reported split. Reconstructed
            from y_prob when omitted, which costs accuracy on saturated
            predictions; see probabilities_to_logits.

    Returns:
        Mapping with "ece", "adaptive_ece", "mce", "brier", their "_temp"
        counterparts measured after scaling, the fitted "temperature", and
        "calibration_bins" recording what produced the binned values.
    """
    task = resolve_calibration_task(y_true, y_prob)
    base = calculate_calibration_metrics(y_true, y_prob, n_bins=n_bins)

    report: dict[str, Any] = {
        "ece": base.get("ece", np.nan),
        "adaptive_ece": base.get("adaptive_ece", np.nan),
        "mce": base.get("mce", np.nan),
        "brier": base.get("brier", np.nan),
        "temperature": np.nan,
        "ece_temp": np.nan,
        "adaptive_ece_temp": np.nan,
        "mce_temp": np.nan,
        "brier_temp": np.nan,
        "calibration_bins": int(n_bins),
    }

    if fit_targets is None or fit_logits is None:
        return report

    try:
        scaler = TemperatureScaler(task=task).fit(fit_logits, fit_targets)
        logits = eval_logits
        if logits is None:
            logits = probabilities_to_logits(y_prob, task=task)
        scaled = scaler.transform(logits)
        after = calculate_calibration_metrics(y_true, scaled, n_bins=n_bins)
    except Exception as error:
        logger.warning(f"Could not apply temperature scaling: {error}")
        return report

    report["temperature"] = float(scaler.temperature)
    report["ece_temp"] = after.get("ece", np.nan)
    report["adaptive_ece_temp"] = after.get("adaptive_ece", np.nan)
    report["mce_temp"] = after.get("mce", np.nan)
    report["brier_temp"] = after.get("brier", np.nan)
    return report

Representation quality

linear_probe

linear_probe(
    train: FeatureSet | tuple[Any, Any],
    test: FeatureSet | tuple[Any, Any],
    metric: str = DEFAULT_PROBE_METRIC,
    model_name: str | None = None,
    source: str | None = None,
    target: str | None = None,
    **probe_kwargs: Any,
) -> ProbeResult

Fit a standardised logistic probe on training embeddings and score it on held-out ones.

Parameters:

Name	Type	Description	Default
train	FeatureSet | tuple[Any, Any]	Embeddings and labels the probe is fitted on	required
test	FeatureSet | tuple[Any, Any]	Embeddings and labels the probe is scored on, never seen by the fit	required
metric	str	Headline metric, one of PROBE_METRICS	DEFAULT_PROBE_METRIC
model_name	str | None	Registry name of the encoder, recorded on the result	None
source	str | None	Dataset the encoder was trained on, recorded on the result	None
target	str | None	Dataset being probed, recorded on the result	None
**probe_kwargs	Any	Forwarded to build_probe	{}

Returns:

Type	Description
ProbeResult	The probe's scores. A target whose training split holds a single class
ProbeResult	cannot be fitted at all and comes back with NaN metrics and a note
ProbeResult	rather than an exception, so one degenerate cell does not abort a
ProbeResult	transfer matrix.

Raises:

Type	Description
ValueError	If the metric is unknown, the two splits disagree on embedding width, or either split carries no labels.

Source code in deepecgkit/evaluation/representation.py

def linear_probe(
    train: FeatureSet | tuple[Any, Any],
    test: FeatureSet | tuple[Any, Any],
    metric: str = DEFAULT_PROBE_METRIC,
    model_name: str | None = None,
    source: str | None = None,
    target: str | None = None,
    **probe_kwargs: Any,
) -> ProbeResult:
    """Fit a standardised logistic probe on training embeddings and score it on held-out ones.

    Args:
        train: Embeddings and labels the probe is fitted on
        test: Embeddings and labels the probe is scored on, never seen by the fit
        metric: Headline metric, one of PROBE_METRICS
        model_name: Registry name of the encoder, recorded on the result
        source: Dataset the encoder was trained on, recorded on the result
        target: Dataset being probed, recorded on the result
        **probe_kwargs: Forwarded to build_probe

    Returns:
        The probe's scores. A target whose training split holds a single class
        cannot be fitted at all and comes back with NaN metrics and a note
        rather than an exception, so one degenerate cell does not abort a
        transfer matrix.

    Raises:
        ValueError: If the metric is unknown, the two splits disagree on
            embedding width, or either split carries no labels.
    """
    if metric not in PROBE_METRICS:
        raise ValueError(
            f"Unknown probe metric: {metric}. Choose one of {', '.join(PROBE_METRICS)}."
        )

    train_features, train_labels = _as_arrays(train)
    test_features, test_labels = _as_arrays(test)

    if train_features.shape[1] != test_features.shape[1]:
        raise ValueError(
            f"Embedding width differs between the splits: {train_features.shape[1]} for train and "
            f"{test_features.shape[1]} for test. Both splits must come from the same encoder."
        )

    resolved_name = model_name
    if resolved_name is None and isinstance(train, FeatureSet):
        resolved_name = train.model_name

    multilabel = _is_multilabel(train_labels)
    result = ProbeResult(
        metrics=_nan_metrics(),
        metric=metric,
        task=TASK_MULTILABEL if multilabel else None,
        n_train=int(train_features.shape[0]),
        n_test=int(test_features.shape[0]),
        feature_dim=int(train_features.shape[1]),
        model_name=resolved_name,
        source=source,
        target=target,
    )

    pipeline = build_probe(multilabel=multilabel, **probe_kwargs)
    n_classes = None
    if not multilabel:
        n_classes = int(max(np.max(train_labels), np.max(test_labels))) + 1
    try:
        with warnings.catch_warnings():
            # The solver's own convergence warning is replaced by the note below.
            warnings.simplefilter("ignore", category=ConvergenceWarning)
            pipeline.fit(train_features, train_labels)
            probabilities = _probe_probabilities(pipeline, test_features, n_classes)
    except ValueError as error:
        result.note = f"probe could not be fitted: {error}"
        return result

    result.task = resolve_task_shape(test_labels, probabilities)
    result.metrics = _probe_metrics(test_labels, probabilities, result.task)
    notes: list[str] = []
    if not np.isfinite(result.score):
        notes.append(f"{metric} is undefined for this held-out split")
    max_iter = int(probe_kwargs.get("max_iter", 1000))
    if not _probe_converged(pipeline, max_iter):
        notes.append(f"probe did not converge within {max_iter} iterations")
    if not multilabel:
        present = len(np.unique(test_labels))
        if present < n_classes:
            notes.append(f"held-out split holds {present} of {n_classes} classes")
    result.note = "; ".join(notes) if notes else None
    return result

probe_encoder

probe_encoder(
    extractor: FeatureExtractor,
    data_module: ECGDataModule,
    train_split: str = "train",
    test_split: str = "test",
    metric: str = DEFAULT_PROBE_METRIC,
    source: str | None = None,
    target: str | None = None,
    **probe_kwargs: Any,
) -> ProbeResult

Embed two splits of one data module with a frozen encoder and probe them.

Parameters:

Name	Type	Description	Default
extractor	FeatureExtractor	A frozen encoder; nothing here updates its weights	required
data_module	ECGDataModule	The data module supplying both splits	required
train_split	str	Split the probe is fitted on	'train'
test_split	str	Split the probe is scored on	'test'
metric	str	Headline metric, one of PROBE_METRICS	DEFAULT_PROBE_METRIC
source	str | None	Dataset the encoder was trained on, recorded on the result	None
target	str | None	Dataset being probed, recorded on the result	None
**probe_kwargs	Any	Forwarded to build_probe	{}

Returns:

Type	Description
ProbeResult	The probe's scores on the held-out split.

Source code in deepecgkit/evaluation/representation.py

def probe_encoder(
    extractor: FeatureExtractor,
    data_module: ECGDataModule,
    train_split: str = "train",
    test_split: str = "test",
    metric: str = DEFAULT_PROBE_METRIC,
    source: str | None = None,
    target: str | None = None,
    **probe_kwargs: Any,
) -> ProbeResult:
    """Embed two splits of one data module with a frozen encoder and probe them.

    Args:
        extractor: A frozen encoder; nothing here updates its weights
        data_module: The data module supplying both splits
        train_split: Split the probe is fitted on
        test_split: Split the probe is scored on
        metric: Headline metric, one of PROBE_METRICS
        source: Dataset the encoder was trained on, recorded on the result
        target: Dataset being probed, recorded on the result
        **probe_kwargs: Forwarded to build_probe

    Returns:
        The probe's scores on the held-out split.
    """
    train_features = extractor.extract(data_module, split=train_split)
    test_features = extractor.extract(data_module, split=test_split)
    return linear_probe(
        train_features,
        test_features,
        metric=metric,
        model_name=extractor.model_name,
        source=source,
        target=target,
        **probe_kwargs,
    )

build_probe

build_probe(
    c: float = 1.0,
    max_iter: int = 1000,
    random_state: int = 42,
    multilabel: bool = False,
) -> Pipeline

Build the protocol's probe: standardisation followed by logistic regression.

The scaler is a pipeline step rather than a preprocessing call so that it is fitted on the training split alone and merely applied to the held-out split. Standardising the two together, or standardising the pooled features once before splitting, leaks the test distribution into the fit.

Parameters:

Name	Type	Description	Default
c	float	Inverse regularisation strength passed to LogisticRegression	1.0
max_iter	int	Solver iteration ceiling	1000
random_state	int	Seed for the solver	42
multilabel	bool	Wrap the estimator in OneVsRestClassifier, for targets whose labels are a multi-hot matrix rather than a class index	False

Returns:

Type	Description
Pipeline	An unfitted pipeline of a StandardScaler and a logistic regression.

Source code in deepecgkit/evaluation/representation.py

def build_probe(
    c: float = 1.0,
    max_iter: int = 1000,
    random_state: int = 42,
    multilabel: bool = False,
) -> Pipeline:
    """Build the protocol's probe: standardisation followed by logistic regression.

    The scaler is a pipeline step rather than a preprocessing call so that it is
    fitted on the training split alone and merely applied to the held-out split.
    Standardising the two together, or standardising the pooled features once
    before splitting, leaks the test distribution into the fit.

    Args:
        c: Inverse regularisation strength passed to LogisticRegression
        max_iter: Solver iteration ceiling
        random_state: Seed for the solver
        multilabel: Wrap the estimator in OneVsRestClassifier, for targets whose
            labels are a multi-hot matrix rather than a class index

    Returns:
        An unfitted pipeline of a StandardScaler and a logistic regression.
    """
    estimator: Any = LogisticRegression(C=c, max_iter=max_iter, random_state=random_state)
    if multilabel:
        estimator = OneVsRestClassifier(estimator)
    return Pipeline([("scaler", StandardScaler()), ("probe", estimator)])

cross_dataset_transfer

cross_dataset_transfer(
    checkpoints: dict[str, str | Path],
    model_name: str,
    targets: Sequence[str] | None = None,
    input_channels: int = PROBE_INPUT_CHANNELS,
    metric: str = DEFAULT_PROBE_METRIC,
    include_random_init: bool = True,
    data_modules: dict[str, ECGDataModule] | None = None,
    sampling_rate: int = PROBE_SAMPLING_RATE,
    data_dir: str | None = None,
    download: bool = True,
    force_download: bool = False,
    batch_size: int = 32,
    num_workers: int = 0,
    device: str = "auto",
    val_split: float = 0.2,
    test_split: float = 0.1,
    seed: int = 42,
    progress: bool = False,
    verbose: bool = False,
    keep_embeddings: str = KEEP_NONE,
    **probe_kwargs: Any,
) -> TransferMatrix

Probe every source encoder on every target dataset.

Each cell freezes the encoder trained on its source, embeds the target's training and held-out splits, fits a standardised logistic head on the training embeddings and reports on the held-out ones. The head is refitted per target because the label spaces differ and no head can be carried across them; what transfers, or fails to, is the representation.

Parameters:

Name	Type	Description	Default
checkpoints	dict[str, str | Path]	Source dataset name mapped to the checkpoint of the encoder trained on it	required
model_name	str	Registered model name matching every checkpoint	required
targets	Sequence[str] | None	Datasets to probe, defaulting to the supplied data modules or else to PROBE_DATASETS	None
input_channels	int	Lead count the protocol runs at	PROBE_INPUT_CHANNELS
metric	str	Headline metric filling the matrix, one of PROBE_METRICS	DEFAULT_PROBE_METRIC
include_random_init	bool	Prepend the untrained-encoder floor as the first row	True
data_modules	dict[str, ECGDataModule] | None	Pre-built modules keyed by target name, bypassing dataset construction entirely	None
sampling_rate	int	Target rate for every dataset, in Hz	PROBE_SAMPLING_RATE
data_dir	str | None	Directory holding the datasets, or None for their defaults	None
download	bool	Fetch a dataset when it is absent	True
force_download	bool	Re-download existing datasets	False
batch_size	int	Batch size for embedding extraction	32
num_workers	int	Worker processes for the loaders	0
device	str	Device name, or auto to prefer CUDA then MPS then CPU	'auto'
val_split	float	Fraction held out for validation	0.2
test_split	float	Fraction the probe reports on	0.1
seed	int	Seed for the splits and for the random-init encoder	42
progress	bool	Show a progress bar during extraction	False
verbose	bool	Print dataset construction progress	False
keep_embeddings	str	Which cells keep their held-out embeddings for later plotting: none, all, or diagonal for the in-domain cells together with the random-init row, which is the pairing worth looking at	KEEP_NONE
**probe_kwargs	Any	Forwarded to build_probe	{}

Returns:

Type	Description
TransferMatrix	The filled transfer matrix.

Raises:

Type	Description
ValueError	If no sources are given, or a name is not a registered dataset, or the metric is unknown.

Source code in deepecgkit/evaluation/representation.py

def cross_dataset_transfer(
    checkpoints: dict[str, str | Path],
    model_name: str,
    targets: Sequence[str] | None = None,
    input_channels: int = PROBE_INPUT_CHANNELS,
    metric: str = DEFAULT_PROBE_METRIC,
    include_random_init: bool = True,
    data_modules: dict[str, ECGDataModule] | None = None,
    sampling_rate: int = PROBE_SAMPLING_RATE,
    data_dir: str | None = None,
    download: bool = True,
    force_download: bool = False,
    batch_size: int = 32,
    num_workers: int = 0,
    device: str = "auto",
    val_split: float = 0.2,
    test_split: float = 0.1,
    seed: int = 42,
    progress: bool = False,
    verbose: bool = False,
    keep_embeddings: str = KEEP_NONE,
    **probe_kwargs: Any,
) -> TransferMatrix:
    """Probe every source encoder on every target dataset.

    Each cell freezes the encoder trained on its source, embeds the target's
    training and held-out splits, fits a standardised logistic head on the
    training embeddings and reports on the held-out ones. The head is refitted
    per target because the label spaces differ and no head can be carried
    across them; what transfers, or fails to, is the representation.

    Args:
        checkpoints: Source dataset name mapped to the checkpoint of the encoder
            trained on it
        model_name: Registered model name matching every checkpoint
        targets: Datasets to probe, defaulting to the supplied data modules or
            else to PROBE_DATASETS
        input_channels: Lead count the protocol runs at
        metric: Headline metric filling the matrix, one of PROBE_METRICS
        include_random_init: Prepend the untrained-encoder floor as the first row
        data_modules: Pre-built modules keyed by target name, bypassing dataset
            construction entirely
        sampling_rate: Target rate for every dataset, in Hz
        data_dir: Directory holding the datasets, or None for their defaults
        download: Fetch a dataset when it is absent
        force_download: Re-download existing datasets
        batch_size: Batch size for embedding extraction
        num_workers: Worker processes for the loaders
        device: Device name, or auto to prefer CUDA then MPS then CPU
        val_split: Fraction held out for validation
        test_split: Fraction the probe reports on
        seed: Seed for the splits and for the random-init encoder
        progress: Show a progress bar during extraction
        verbose: Print dataset construction progress
        keep_embeddings: Which cells keep their held-out embeddings for later
            plotting: none, all, or diagonal for the in-domain cells together
            with the random-init row, which is the pairing worth looking at
        **probe_kwargs: Forwarded to build_probe

    Returns:
        The filled transfer matrix.

    Raises:
        ValueError: If no sources are given, or a name is not a registered
            dataset, or the metric is unknown.
    """
    if metric not in PROBE_METRICS:
        raise ValueError(
            f"Unknown probe metric: {metric}. Choose one of {', '.join(PROBE_METRICS)}."
        )

    if keep_embeddings not in KEEP_EMBEDDINGS_MODES:
        raise ValueError(
            f"Unknown keep_embeddings mode: {keep_embeddings}. Choose one of "
            f"{', '.join(KEEP_EMBEDDINGS_MODES)}."
        )

    if not checkpoints and not include_random_init:
        raise ValueError(
            "Nothing to probe: pass at least one source checkpoint, or keep the random-init floor."
        )

    if targets is None:
        targets = sorted(data_modules) if data_modules else list(PROBE_DATASETS)
    resolved_targets = list(targets)
    if not resolved_targets:
        raise ValueError("No probe targets were given.")

    extractor_kwargs: dict[str, Any] = {
        "batch_size": batch_size,
        "device": device,
        "num_workers": num_workers,
        "progress": progress,
    }

    modules: dict[str, ECGDataModule] = dict(data_modules or {})
    for name in resolved_targets:
        if name in modules:
            continue
        modules[name] = protocol_data_module(
            name,
            data_dir=data_dir,
            sampling_rate=sampling_rate,
            batch_size=batch_size,
            num_workers=num_workers,
            val_split=val_split,
            test_split=test_split,
            seed=seed,
            download=download,
            force_download=force_download,
            verbose=verbose,
        )

    sources: list[str] = []
    extractors: dict[str, FeatureExtractor] = {}
    if include_random_init:
        sources.append(RANDOM_INIT)
        extractors[RANDOM_INIT] = _random_init_extractor(
            model_name, input_channels, seed, **extractor_kwargs
        )
    for name in sorted(checkpoints):
        sources.append(name)
        extractors[name] = FeatureExtractor.from_checkpoint(
            checkpoints[name],
            model_name,
            input_channels=input_channels,
            **extractor_kwargs,
        )

    values = np.full((len(sources), len(resolved_targets)), np.nan, dtype=float)
    results: dict[str, ProbeResult] = {}
    retained: dict[str, FeatureSet] = {}
    class_names: dict[str, list[str]] = {}
    for row, source in enumerate(sources):
        for column, target in enumerate(resolved_targets):
            module = modules[target]
            train_features = extractors[source].extract(module, split="train")
            test_features = extractors[source].extract(module, split="test")
            result = linear_probe(
                train_features,
                test_features,
                metric=metric,
                model_name=model_name,
                source=source,
                target=target,
                **probe_kwargs,
            )
            key = f"{source}{TRANSFER_KEY_SEPARATOR}{target}"
            values[row, column] = result.score
            results[key] = result
            if _keeps_embedding(keep_embeddings, source, target):
                retained[key] = test_features
            if target not in class_names:
                names = getattr(module.dataset, "class_names", None)
                if names:
                    class_names[target] = list(names)

    protocol = {
        "input_channels": int(input_channels),
        "sampling_rate": int(sampling_rate),
        "val_split": float(val_split),
        "test_split": float(test_split),
        "seed": int(seed),
        "random_init": bool(include_random_init),
        "dataset_kwargs": {
            name: PROTOCOL_DATASET_KWARGS[name]
            for name in resolved_targets
            if name in PROTOCOL_DATASET_KWARGS
        },
    }

    return TransferMatrix(
        sources=sources,
        targets=resolved_targets,
        values=values,
        metric=metric,
        results=results,
        model_name=model_name,
        protocol=protocol,
        embeddings=retained,
        class_names=class_names,
    )

TransferMatrix dataclass

Probe performance for every (encoder source, probe target) pair.

Rows are the datasets the encoders were trained on, columns are the datasets the probes were fitted and scored on. The diagonal is the in-domain probe. When the random-init floor is included it is the first row, and it is the floor for every target column; transpose to_frame() for a figure that wants sources along the top instead.

Attributes:

Name	Type	Description
sources	list[str]	Row labels, RANDOM_INIT first when the baseline is present
targets	list[str]	Column labels
values	ndarray	Headline metric per cell, of shape (len(sources), len(targets))
metric	str	Name of the headline metric
results	dict[str, ProbeResult]	Full ProbeResult per cell, keyed "source->target"
model_name	str | None	Registry name of the probed architecture
protocol	dict[str, Any]	The resolved protocol settings, recorded for provenance
embeddings	dict[str, FeatureSet]	Held-out embeddings retained for the cells requested by keep_embeddings, keyed "source->target", for plotting. Empty unless they were asked for, since keeping all of them costs memory proportional to the corpus
class_names	dict[str, list[str]]	Class names per target, for labelling an embedding plot

Source code in deepecgkit/evaluation/representation.py

@dataclass
class TransferMatrix:
    """Probe performance for every (encoder source, probe target) pair.

    Rows are the datasets the encoders were trained on, columns are the datasets
    the probes were fitted and scored on. The diagonal is the in-domain probe.
    When the random-init floor is included it is the first row, and it is the
    floor for every target column; transpose to_frame() for a figure that wants
    sources along the top instead.

    Attributes:
        sources: Row labels, RANDOM_INIT first when the baseline is present
        targets: Column labels
        values: Headline metric per cell, of shape (len(sources), len(targets))
        metric: Name of the headline metric
        results: Full ProbeResult per cell, keyed "source->target"
        model_name: Registry name of the probed architecture
        protocol: The resolved protocol settings, recorded for provenance
        embeddings: Held-out embeddings retained for the cells requested by
            keep_embeddings, keyed "source->target", for plotting. Empty unless
            they were asked for, since keeping all of them costs memory
            proportional to the corpus
        class_names: Class names per target, for labelling an embedding plot
    """

    sources: list[str]
    targets: list[str]
    values: np.ndarray
    metric: str = DEFAULT_PROBE_METRIC
    results: dict[str, ProbeResult] = field(default_factory=dict)
    model_name: str | None = None
    protocol: dict[str, Any] = field(default_factory=dict)
    embeddings: dict[str, FeatureSet] = field(default_factory=dict)
    class_names: dict[str, list[str]] = field(default_factory=dict)

    def to_frame(self) -> pd.DataFrame:
        """Render the headline metric as a labelled table.

        Returns:
            A frame indexed by source with one column per target.
        """
        return pd.DataFrame(self.values, index=list(self.sources), columns=list(self.targets))

    def embedding_for(self, source: str, target: str) -> FeatureSet | None:
        """Look up the retained held-out embeddings behind one cell.

        Args:
            source: Row label
            target: Column label

        Returns:
            The cell's embeddings, or None when they were not retained.
        """
        return self.embeddings.get(f"{source}{TRANSFER_KEY_SEPARATOR}{target}")

    def result_for(self, source: str, target: str) -> ProbeResult | None:
        """Look up the full result behind one cell.

        Args:
            source: Row label
            target: Column label

        Returns:
            The cell's ProbeResult, or None when the pair was not run.
        """
        return self.results.get(f"{source}{TRANSFER_KEY_SEPARATOR}{target}")

    def to_dict(self) -> dict[str, Any]:
        """Render the whole matrix as JSON-serialisable primitives.

        Returns:
            A dictionary holding the headline table, every cell's full metrics,
            and the protocol settings the run used.
        """
        return {
            "model_name": self.model_name,
            "metric": self.metric,
            "sources": list(self.sources),
            "targets": list(self.targets),
            "values": [[float(value) for value in row] for row in np.asarray(self.values)],
            "protocol": dict(self.protocol),
            "cells": {key: result.to_dict() for key, result in self.results.items()},
        }

    def save(self, path: str | Path) -> None:
        """Write the matrix to disk, choosing the format from the suffix.

        Args:
            path: Destination ending in .json or .csv. The .json form keeps
                every metric of every cell and the protocol settings; .csv holds
                the headline table alone.

        Raises:
            ValueError: If the suffix is neither of the two supported forms.
        """
        out_path = Path(path)
        out_path.parent.mkdir(parents=True, exist_ok=True)

        if out_path.suffix == ".json":
            out_path.write_text(json.dumps(self.to_dict(), indent=2) + "\n")
            return

        if out_path.suffix == ".csv":
            self.to_frame().to_csv(str(out_path))
            return

        raise ValueError(
            f"Unsupported output format: {out_path.suffix}. Supported formats: .json, .csv"
        )

to_frame

to_frame() -> pd.DataFrame

Render the headline metric as a labelled table.

Returns:

Type	Description
DataFrame	A frame indexed by source with one column per target.

Source code in deepecgkit/evaluation/representation.py

def to_frame(self) -> pd.DataFrame:
    """Render the headline metric as a labelled table.

    Returns:
        A frame indexed by source with one column per target.
    """
    return pd.DataFrame(self.values, index=list(self.sources), columns=list(self.targets))

embedding_for

embedding_for(
    source: str, target: str
) -> FeatureSet | None

Look up the retained held-out embeddings behind one cell.

Parameters:

Name	Type	Description	Default
source	str	Row label	required
target	str	Column label	required

Returns:

Type	Description
FeatureSet | None	The cell's embeddings, or None when they were not retained.

Source code in deepecgkit/evaluation/representation.py

def embedding_for(self, source: str, target: str) -> FeatureSet | None:
    """Look up the retained held-out embeddings behind one cell.

    Args:
        source: Row label
        target: Column label

    Returns:
        The cell's embeddings, or None when they were not retained.
    """
    return self.embeddings.get(f"{source}{TRANSFER_KEY_SEPARATOR}{target}")

result_for

result_for(source: str, target: str) -> ProbeResult | None

Look up the full result behind one cell.

Parameters:

Name	Type	Description	Default
source	str	Row label	required
target	str	Column label	required

Returns:

Type	Description
ProbeResult | None	The cell's ProbeResult, or None when the pair was not run.

Source code in deepecgkit/evaluation/representation.py

def result_for(self, source: str, target: str) -> ProbeResult | None:
    """Look up the full result behind one cell.

    Args:
        source: Row label
        target: Column label

    Returns:
        The cell's ProbeResult, or None when the pair was not run.
    """
    return self.results.get(f"{source}{TRANSFER_KEY_SEPARATOR}{target}")

to_dict

to_dict() -> dict[str, Any]

Render the whole matrix as JSON-serialisable primitives.

Returns:

Type	Description
dict[str, Any]	A dictionary holding the headline table, every cell's full metrics,
dict[str, Any]	and the protocol settings the run used.

Source code in deepecgkit/evaluation/representation.py

def to_dict(self) -> dict[str, Any]:
    """Render the whole matrix as JSON-serialisable primitives.

    Returns:
        A dictionary holding the headline table, every cell's full metrics,
        and the protocol settings the run used.
    """
    return {
        "model_name": self.model_name,
        "metric": self.metric,
        "sources": list(self.sources),
        "targets": list(self.targets),
        "values": [[float(value) for value in row] for row in np.asarray(self.values)],
        "protocol": dict(self.protocol),
        "cells": {key: result.to_dict() for key, result in self.results.items()},
    }

save

save(path: str | Path) -> None

Write the matrix to disk, choosing the format from the suffix.

Parameters:

Name	Type	Description	Default
path	str | Path	Destination ending in .json or .csv. The .json form keeps every metric of every cell and the protocol settings; .csv holds the headline table alone.	required

Raises:

Type	Description
ValueError	If the suffix is neither of the two supported forms.

Source code in deepecgkit/evaluation/representation.py

def save(self, path: str | Path) -> None:
    """Write the matrix to disk, choosing the format from the suffix.

    Args:
        path: Destination ending in .json or .csv. The .json form keeps
            every metric of every cell and the protocol settings; .csv holds
            the headline table alone.

    Raises:
        ValueError: If the suffix is neither of the two supported forms.
    """
    out_path = Path(path)
    out_path.parent.mkdir(parents=True, exist_ok=True)

    if out_path.suffix == ".json":
        out_path.write_text(json.dumps(self.to_dict(), indent=2) + "\n")
        return

    if out_path.suffix == ".csv":
        self.to_frame().to_csv(str(out_path))
        return

    raise ValueError(
        f"Unsupported output format: {out_path.suffix}. Supported formats: .json, .csv"
    )

ProbeResult dataclass

The outcome of fitting one linear probe on one held-out split.

Attributes:

Name	Type	Description
metrics	dict[str, float]	Every metric in PROBE_METRICS, with NaN where a metric was not defined for this label set
metric	str	Name of the headline metric
score	float	Value of the headline metric, a convenience view on metrics
task	str | None	Resolved label shape, one of binary, multiclass or multilabel
n_train	int	Samples the probe was fitted on
n_test	int	Samples the probe was scored on
feature_dim	int	Width of the embedding the probe consumed
model_name	str | None	Registry name of the encoder that produced the embeddings
source	str | None	Dataset the encoder was trained on, or RANDOM_INIT
target	str | None	Dataset the probe was fitted and scored on
note	str | None	Why a metric is missing, when one is

Source code in deepecgkit/evaluation/representation.py

@dataclass
class ProbeResult:
    """The outcome of fitting one linear probe on one held-out split.

    Attributes:
        metrics: Every metric in PROBE_METRICS, with NaN where a metric was not
            defined for this label set
        metric: Name of the headline metric
        score: Value of the headline metric, a convenience view on metrics
        task: Resolved label shape, one of binary, multiclass or multilabel
        n_train: Samples the probe was fitted on
        n_test: Samples the probe was scored on
        feature_dim: Width of the embedding the probe consumed
        model_name: Registry name of the encoder that produced the embeddings
        source: Dataset the encoder was trained on, or RANDOM_INIT
        target: Dataset the probe was fitted and scored on
        note: Why a metric is missing, when one is
    """

    metrics: dict[str, float]
    metric: str = DEFAULT_PROBE_METRIC
    task: str | None = None
    n_train: int = 0
    n_test: int = 0
    feature_dim: int = 0
    model_name: str | None = None
    source: str | None = None
    target: str | None = None
    note: str | None = field(default=None)

    @property
    def score(self) -> float:
        """Value of the headline metric."""
        return float(self.metrics.get(self.metric, float("nan")))

    def to_dict(self) -> dict[str, Any]:
        """Render the result as JSON-serialisable primitives.

        Returns:
            A dictionary whose float values are plain floats, so NaN survives
            json.dumps as the non-standard but universally parsed NaN literal.
        """
        return {
            "source": self.source,
            "target": self.target,
            "model_name": self.model_name,
            "metric": self.metric,
            "score": self.score,
            "task": self.task,
            "n_train": int(self.n_train),
            "n_test": int(self.n_test),
            "feature_dim": int(self.feature_dim),
            "metrics": {name: float(value) for name, value in self.metrics.items()},
            "note": self.note,
        }

score property

score: float

Value of the headline metric.

to_dict

to_dict() -> dict[str, Any]

Render the result as JSON-serialisable primitives.

Returns:

Type	Description
dict[str, Any]	A dictionary whose float values are plain floats, so NaN survives
dict[str, Any]	json.dumps as the non-standard but universally parsed NaN literal.

Source code in deepecgkit/evaluation/representation.py

def to_dict(self) -> dict[str, Any]:
    """Render the result as JSON-serialisable primitives.

    Returns:
        A dictionary whose float values are plain floats, so NaN survives
        json.dumps as the non-standard but universally parsed NaN literal.
    """
    return {
        "source": self.source,
        "target": self.target,
        "model_name": self.model_name,
        "metric": self.metric,
        "score": self.score,
        "task": self.task,
        "n_train": int(self.n_train),
        "n_test": int(self.n_test),
        "feature_dim": int(self.feature_dim),
        "metrics": {name: float(value) for name, value in self.metrics.items()},
        "note": self.note,
    }

protocol_data_module

protocol_data_module(
    dataset_name: str,
    data_dir: str | None = None,
    sampling_rate: int = PROBE_SAMPLING_RATE,
    batch_size: int = 32,
    num_workers: int = 0,
    val_split: float = 0.2,
    test_split: float = 0.1,
    seed: int = 42,
    download: bool = True,
    force_download: bool = False,
    verbose: bool = False,
) -> ECGDataModule

Build one dataset in the single-lead configuration the protocol requires.

Transfer only means anything if every encoder sees the same input shape, so the protocol runs every dataset at one lead and one sampling rate. The three AF corpora already yield a single lead at their constructor defaults; PTB-XL is pinned to lead II and resampled to match, which its standardiser handles.

Parameters:

Name	Type	Description	Default
dataset_name	str	A registered dataset name	required
data_dir	str | None	Directory holding the dataset, or None for its default	None
sampling_rate	int	Target rate for every dataset, in Hz	PROBE_SAMPLING_RATE
batch_size	int	Batch size for the loaders	32
num_workers	int	Worker processes for the loaders	0
val_split	float	Fraction held out for validation	0.2
test_split	float	Fraction held out for the probe's report	0.1
seed	int	Split seed	42
download	bool	Fetch the dataset when it is absent	True
force_download	bool	Re-download an existing dataset	False
verbose	bool	Print dataset construction progress	False

Returns:

Type	Description
ECGDataModule	A data module that has not yet been set up.

Source code in deepecgkit/evaluation/representation.py

def protocol_data_module(
    dataset_name: str,
    data_dir: str | None = None,
    sampling_rate: int = PROBE_SAMPLING_RATE,
    batch_size: int = 32,
    num_workers: int = 0,
    val_split: float = 0.2,
    test_split: float = 0.1,
    seed: int = 42,
    download: bool = True,
    force_download: bool = False,
    verbose: bool = False,
) -> ECGDataModule:
    """Build one dataset in the single-lead configuration the protocol requires.

    Transfer only means anything if every encoder sees the same input shape, so
    the protocol runs every dataset at one lead and one sampling rate. The three
    AF corpora already yield a single lead at their constructor defaults;
    PTB-XL is pinned to lead II and resampled to match, which its standardiser
    handles.

    Args:
        dataset_name: A registered dataset name
        data_dir: Directory holding the dataset, or None for its default
        sampling_rate: Target rate for every dataset, in Hz
        batch_size: Batch size for the loaders
        num_workers: Worker processes for the loaders
        val_split: Fraction held out for validation
        test_split: Fraction held out for the probe's report
        seed: Split seed
        download: Fetch the dataset when it is absent
        force_download: Re-download an existing dataset
        verbose: Print dataset construction progress

    Returns:
        A data module that has not yet been set up.
    """
    dataset_kwargs: dict[str, Any] = {
        "force_download": force_download,
        "verbose": verbose,
    }
    dataset_kwargs.update(PROTOCOL_DATASET_KWARGS.get(dataset_name, {}))

    return ECGDataModule(
        dataset_class=get_dataset(dataset_name),
        data_dir=data_dir,
        sampling_rate=sampling_rate,
        batch_size=batch_size,
        num_workers=num_workers,
        val_split=val_split,
        test_split=test_split,
        seed=seed,
        download=download,
        verbose=verbose,
        dataset_kwargs=dataset_kwargs,
    )

embed_2d

embed_2d(
    features: Any,
    method: str = "tsne",
    perplexity: float = DEFAULT_PERPLEXITY,
    pca_components: int | None = DEFAULT_PCA_COMPONENTS,
    random_state: int = 42,
) -> np.ndarray

Project embeddings to two dimensions for plotting.

Features are standardised first, for the same reason the probe standardises them: a handful of wide-range dimensions would otherwise dominate the distances t-SNE is built from. Wide embeddings are then reduced by PCA before t-SNE, which is both the conventional recipe and much faster.

Parameters:

Name	Type	Description	Default
features	Any	Embeddings of shape (n_samples, feature_dim), or a FeatureSet	required
method	str	One of tsne or pca	'tsne'
perplexity	float	t-SNE perplexity, clamped to a value the sample count supports rather than letting scikit-learn reject it	DEFAULT_PERPLEXITY
pca_components	int | None	Width to reduce to before t-SNE, or None to skip	DEFAULT_PCA_COMPONENTS
random_state	int	Seed, making the projection reproducible	42

Returns:

Type	Description
ndarray	Coordinates of shape (n_samples, 2).

Raises:

Type	Description
ValueError	If the method is unknown, the input is not two-dimensional, or there are too few samples to project.

Source code in deepecgkit/evaluation/representation.py

def embed_2d(
    features: Any,
    method: str = "tsne",
    perplexity: float = DEFAULT_PERPLEXITY,
    pca_components: int | None = DEFAULT_PCA_COMPONENTS,
    random_state: int = 42,
) -> np.ndarray:
    """Project embeddings to two dimensions for plotting.

    Features are standardised first, for the same reason the probe standardises
    them: a handful of wide-range dimensions would otherwise dominate the
    distances t-SNE is built from. Wide embeddings are then reduced by PCA
    before t-SNE, which is both the conventional recipe and much faster.

    Args:
        features: Embeddings of shape (n_samples, feature_dim), or a FeatureSet
        method: One of tsne or pca
        perplexity: t-SNE perplexity, clamped to a value the sample count
            supports rather than letting scikit-learn reject it
        pca_components: Width to reduce to before t-SNE, or None to skip
        random_state: Seed, making the projection reproducible

    Returns:
        Coordinates of shape (n_samples, 2).

    Raises:
        ValueError: If the method is unknown, the input is not two-dimensional,
            or there are too few samples to project.
    """
    if method not in EMBEDDING_METHODS:
        raise ValueError(
            f"Unknown embedding method: {method}. Choose one of {', '.join(EMBEDDING_METHODS)}."
        )

    data = np.asarray(features, dtype=np.float64)
    if data.ndim != 2:
        raise ValueError(
            f"Embeddings must be two-dimensional (n_samples, feature_dim); got shape {data.shape}."
        )
    if data.shape[0] < MIN_EMBEDDING_SAMPLES:
        raise ValueError(
            f"Need at least {MIN_EMBEDDING_SAMPLES} samples to project; got {data.shape[0]}."
        )

    scaled = StandardScaler().fit_transform(data)
    if method == "pca":
        return np.asarray(PCA(n_components=2, random_state=random_state).fit_transform(scaled))

    width = min(pca_components or 0, scaled.shape[0], scaled.shape[1])
    if pca_components and scaled.shape[1] > width > 2:
        scaled = PCA(n_components=width, random_state=random_state).fit_transform(scaled)

    ceiling = max(1.0, (scaled.shape[0] - 1) / PERPLEXITY_SAMPLE_DIVISOR)
    projection = TSNE(
        n_components=2,
        perplexity=float(min(perplexity, ceiling)),
        random_state=random_state,
        init="pca",
    ).fit_transform(scaled)
    return np.asarray(projection)

Visualization

plot_confusion_matrix

plot_confusion_matrix(
    y_true: ndarray,
    y_pred: ndarray,
    class_names: list[str] | None = None,
    title: str = "Confusion Matrix",
    save_path: str | None = None,
) -> None

Plot confusion matrix.

Source code in deepecgkit/evaluation/visualization.py

def plot_confusion_matrix(
    y_true: np.ndarray,
    y_pred: np.ndarray,
    class_names: list[str] | None = None,
    title: str = "Confusion Matrix",
    save_path: str | None = None,
) -> None:
    """Plot confusion matrix."""
    labels = sorted(set(y_true) | set(y_pred))
    cm = confusion_matrix(y_true, y_pred, labels=labels)

    if class_names is None or len(class_names) != len(labels):
        class_names = [str(lbl) for lbl in labels]

    _, ax = plt.subplots(figsize=(8, 6))

    sns.heatmap(
        cm,
        annot=True,
        fmt="d",
        cmap="Blues",
        xticklabels=class_names,
        yticklabels=class_names,
        ax=ax,
    )

    ax.set_title(title)
    ax.set_ylabel("True Label")
    ax.set_xlabel("Predicted Label")

    finalize_figure(save_path)

plot_roc_curve

plot_roc_curve(
    y_true: ndarray,
    y_scores: ndarray,
    title: str = "ROC Curve",
    save_path: str | None = None,
) -> None

Plot the ROC curve of a two-class problem.

Parameters:

Name	Type	Description	Default
y_true	ndarray	True labels, two classes	required
y_scores	ndarray	Scores of the positive class, or the two-column probability matrix that ECGTrainer.get_test_results returns for a binary model, whose second column is used	required
title	str	Figure title	'ROC Curve'
save_path	str | None	Where to save the figure; None shows it instead	None

Raises:

Type	Description
ValueError	If y_scores holds a column per class for more than two classes.

Source code in deepecgkit/evaluation/visualization.py

def plot_roc_curve(
    y_true: np.ndarray,
    y_scores: np.ndarray,
    title: str = "ROC Curve",
    save_path: str | None = None,
) -> None:
    """Plot the ROC curve of a two-class problem.

    Args:
        y_true: True labels, two classes
        y_scores: Scores of the positive class, or the two-column probability
            matrix that ECGTrainer.get_test_results returns for a binary
            model, whose second column is used
        title: Figure title
        save_path: Where to save the figure; None shows it instead

    Raises:
        ValueError: If y_scores holds a column per class for more than two
            classes.
    """
    y_scores = np.asarray(y_scores)
    if y_scores.ndim == 2:
        if y_scores.shape[1] > 2:
            raise ValueError(
                f"plot_roc_curve draws one curve for a two-class problem; got scores for "
                f"{y_scores.shape[1]} classes. For class k against the rest, pass "
                f"y_true == k and y_scores[:, k]."
            )
        y_scores = y_scores[:, -1]
    fpr, tpr, _ = roc_curve(y_true, y_scores)
    roc_auc = auc(fpr, tpr)

    plt.figure(figsize=(8, 6))
    plt.plot(
        fpr,
        tpr,
        color="darkorange",
        lw=2,
        label=f"ROC curve (area = {roc_auc:.2f})",
    )
    plt.plot([0, 1], [0, 1], color="navy", lw=2, linestyle="--")
    plt.xlim([0.0, 1.0])
    plt.ylim([0.0, 1.05])
    plt.xlabel("False Positive Rate")
    plt.ylabel("True Positive Rate")
    plt.title(title)
    plt.legend(loc="lower right")

    finalize_figure(save_path)

plot_training_curves

plot_training_curves(
    metrics_path: str,
    save_dir: str | None = None,
    save_path: str | None = None,
) -> None

Plot training and validation loss and accuracy as separate figures.

Parameters:

Name	Type	Description	Default
metrics_path	str	Path to CSVLogger metrics.csv file.	required
save_dir	str | None	Directory to save separate loss.png and accuracy.png files.	None
save_path	str | None	Deprecated combined path; if provided and save_dir is not, saves loss plot to this path for backward compatibility.	None

Source code in deepecgkit/evaluation/visualization.py

def plot_training_curves(
    metrics_path: str,
    save_dir: str | None = None,
    save_path: str | None = None,
) -> None:
    """Plot training and validation loss and accuracy as separate figures.

    Args:
        metrics_path: Path to CSVLogger metrics.csv file.
        save_dir: Directory to save separate loss.png and accuracy.png files.
        save_path: Deprecated combined path; if provided and save_dir is not,
            saves loss plot to this path for backward compatibility.
    """
    df = pd.read_csv(metrics_path)

    has_acc = "train_acc" in df.columns

    resolved_dir = None
    if save_dir:
        resolved_dir = Path(save_dir)
    elif save_path:
        resolved_dir = Path(save_path).parent
    if resolved_dir:
        resolved_dir.mkdir(parents=True, exist_ok=True)

    train_loss = df.dropna(subset=["train_loss"]).groupby("epoch")["train_loss"].mean()
    val_loss = df.dropna(subset=["val_loss"]).groupby("epoch")["val_loss"].mean()

    plt.figure(figsize=(10, 5))
    plt.plot(train_loss.index, train_loss.values, label="Train Loss")
    plt.plot(val_loss.index, val_loss.values, label="Val Loss")
    plt.xlabel("Epoch")
    plt.ylabel("Loss")
    plt.title("Training & Validation Loss")
    plt.legend()
    plt.grid(True, alpha=0.3)
    plt.tight_layout()

    if resolved_dir:
        plt.savefig(str(resolved_dir / "loss.png"), dpi=300, bbox_inches="tight")
        plt.close()
    else:
        plt.show()

    if has_acc:
        train_acc = df.dropna(subset=["train_acc"]).groupby("epoch")["train_acc"].mean()
        val_acc = df.dropna(subset=["val_acc"]).groupby("epoch")["val_acc"].mean()

        plt.figure(figsize=(10, 5))
        plt.plot(train_acc.index, train_acc.values, label="Train Acc")
        plt.plot(val_acc.index, val_acc.values, label="Val Acc")
        plt.xlabel("Epoch")
        plt.ylabel("Accuracy")
        plt.title("Training & Validation Accuracy")
        plt.legend()
        plt.grid(True, alpha=0.3)
        plt.tight_layout()

        if resolved_dir:
            plt.savefig(str(resolved_dir / "accuracy.png"), dpi=300, bbox_inches="tight")
            plt.close()
        else:
            plt.show()

plot_ecg_signals

plot_ecg_signals(
    data: ndarray,
    sampling_rate: float = 500.0,
    leads: list[str] | None = None,
    title: str = "ECG Signals",
    save_path: str | None = None,
    unit: str | None = None,
) -> None

Plot one ECG recording, one row per lead.

Parameters:

Name	Type	Description	Default
data	ndarray	A single lead as a one-dimensional array, or several leads as a two-dimensional one. Both layouts are accepted for the latter, (leads, time) as the datasets and models use and (time, leads); the longer axis is taken to be time.	required
sampling_rate	float	Sampling rate of the recording in Hz, for the time axis	500.0
leads	list[str] | None	Name of each lead, in order	None
title	str	Figure title	'ECG Signals'
save_path	str | None	Where to save the figure; None shows it instead	None
unit	str | None	Unit of the amplitude axis, such as "mV" for a raw recording. The datasets return normalised signals, which have none, so none is shown by default.	None

Raises:

Type	Description
ValueError	If data has more than two dimensions, or leads does not name every lead.

Source code in deepecgkit/evaluation/visualization.py

def plot_ecg_signals(
    data: np.ndarray,
    sampling_rate: float = 500.0,
    leads: list[str] | None = None,
    title: str = "ECG Signals",
    save_path: str | None = None,
    unit: str | None = None,
) -> None:
    """Plot one ECG recording, one row per lead.

    Args:
        data: A single lead as a one-dimensional array, or several leads as a
            two-dimensional one. Both layouts are accepted for the latter,
            (leads, time) as the datasets and models use and (time, leads);
            the longer axis is taken to be time.
        sampling_rate: Sampling rate of the recording in Hz, for the time axis
        leads: Name of each lead, in order
        title: Figure title
        save_path: Where to save the figure; None shows it instead
        unit: Unit of the amplitude axis, such as "mV" for a raw recording.
            The datasets return normalised signals, which have none, so none
            is shown by default.

    Raises:
        ValueError: If data has more than two dimensions, or leads does not
            name every lead.
    """
    data = np.asarray(data)
    if data.ndim == 1:
        data = data.reshape(-1, 1)
    elif data.ndim != 2:
        raise ValueError(
            f"plot_ecg_signals draws one recording; got an array of shape {data.shape}. "
            f"Pass a single (leads, time) recording, for example batch[0]."
        )
    elif data.shape[0] < data.shape[1]:
        data = data.T

    n_samples, n_leads = data.shape
    if leads is not None and len(leads) != n_leads:
        raise ValueError(f"Got {len(leads)} lead names for a recording with {n_leads} leads.")
    time = np.arange(n_samples) / sampling_rate

    if leads is None:
        leads = [f"Lead {i + 1}" for i in range(n_leads)]

    _fig, axes = plt.subplots(n_leads, 1, figsize=(12, 2 * n_leads), sharex=True)
    if n_leads == 1:
        axes = [axes]

    for i, ax in enumerate(axes):
        ax.plot(time, data[:, i])
        ax.set_ylabel(f"{leads[i]} ({unit})" if unit else leads[i])
        ax.grid(True, alpha=0.3)

    axes[-1].set_xlabel("Time (s)")
    plt.suptitle(title)
    plt.tight_layout()

    finalize_figure(save_path)

plot_predictions

plot_predictions(
    y_true: ndarray,
    y_pred: ndarray,
    title: str = "Predictions vs True Values",
    save_path: str | None = None,
) -> None

Plot predictions vs true values.

Source code in deepecgkit/evaluation/visualization.py

def plot_predictions(
    y_true: np.ndarray,
    y_pred: np.ndarray,
    title: str = "Predictions vs True Values",
    save_path: str | None = None,
) -> None:
    """Plot predictions vs true values."""
    plt.figure(figsize=(8, 6))
    plt.scatter(y_true, y_pred, alpha=0.6)
    plt.plot([y_true.min(), y_true.max()], [y_true.min(), y_true.max()], "r--", lw=2)
    plt.xlabel("True Values")
    plt.ylabel("Predictions")
    plt.title(title)

    finalize_figure(save_path)

plot_calibration_curve

plot_calibration_curve(
    y_true: ndarray,
    y_prob: ndarray,
    class_names: list[str] | None = None,
    n_bins: int = 10,
    title: str = "Calibration Plot",
    save_dir: str | None = None,
    save_path: str | None = None,
) -> None

Plot reliability diagram and prediction distribution as separate figures.

Parameters:

Name	Type	Description	Default
y_true	ndarray	True labels.	required
y_prob	ndarray	Predicted probabilities (n_samples, n_classes).	required
class_names	list[str] | None	Optional list of class names.	None
n_bins	int	Number of bins for calibration curve.	10
title	str	Title for the calibration curve plot.	'Calibration Plot'
save_dir	str | None	Directory to save calibration_curve.png and prediction_distribution.png.	None
save_path	str | None	Deprecated combined path; if provided and save_dir is not, uses its parent directory for backward compatibility.	None

Source code in deepecgkit/evaluation/visualization.py

def plot_calibration_curve(
    y_true: np.ndarray,
    y_prob: np.ndarray,
    class_names: list[str] | None = None,
    n_bins: int = 10,
    title: str = "Calibration Plot",
    save_dir: str | None = None,
    save_path: str | None = None,
) -> None:
    """Plot reliability diagram and prediction distribution as separate figures.

    Args:
        y_true: True labels.
        y_prob: Predicted probabilities (n_samples, n_classes).
        class_names: Optional list of class names.
        n_bins: Number of bins for calibration curve.
        title: Title for the calibration curve plot.
        save_dir: Directory to save calibration_curve.png and prediction_distribution.png.
        save_path: Deprecated combined path; if provided and save_dir is not,
            uses its parent directory for backward compatibility.
    """
    num_classes = y_prob.shape[1]
    if class_names is None:
        class_names = [f"Class {i}" for i in range(num_classes)]

    resolved_dir = None
    if save_dir:
        resolved_dir = Path(save_dir)
    elif save_path:
        resolved_dir = Path(save_path).parent
    if resolved_dir:
        resolved_dir.mkdir(parents=True, exist_ok=True)

    plt.figure(figsize=(8, 6))
    for i in range(num_classes):
        y_binary = (y_true == i).astype(int)
        prob_class = y_prob[:, i]
        if y_binary.sum() == 0:
            continue
        try:
            fraction_of_positives, mean_predicted_value = calibration_curve(
                y_binary, prob_class, n_bins=n_bins, strategy="uniform"
            )
            plt.plot(mean_predicted_value, fraction_of_positives, "s-", label=class_names[i])
        except ValueError:
            continue

    plt.plot([0, 1], [0, 1], "k--", label="Perfectly calibrated")
    plt.xlabel("Mean Predicted Probability")
    plt.ylabel("Fraction of Positives")
    plt.title(title)
    plt.legend(loc="lower right", fontsize="small")
    plt.grid(True, alpha=0.3)
    plt.tight_layout()

    if resolved_dir:
        plt.savefig(str(resolved_dir / "calibration_curve.png"), dpi=300, bbox_inches="tight")
        plt.close()
    else:
        plt.show()

    plt.figure(figsize=(8, 6))
    for i in range(num_classes):
        plt.hist(y_prob[:, i], bins=n_bins, alpha=0.5, label=class_names[i])
    plt.xlabel("Predicted Probability")
    plt.ylabel("Count")
    plt.title("Prediction Distribution")
    plt.legend(fontsize="small")
    plt.grid(True, alpha=0.3)
    plt.tight_layout()

    if resolved_dir:
        plt.savefig(str(resolved_dir / "prediction_distribution.png"), dpi=300, bbox_inches="tight")
        plt.close()
    else:
        plt.show()

plot_transfer_matrix

plot_transfer_matrix(
    values: ndarray,
    sources: list[str],
    targets: list[str],
    metric: str = "auroc_macro",
    baseline_rows: int = 0,
    title: str = "Cross-Dataset Transfer",
    save_path: str | None = None,
) -> None

Plot probe performance for every encoder source against every probe target.

Parameters:

Name	Type	Description	Default
values	ndarray	Probe scores of shape (len(sources), len(targets)). Cells that could not be scored are NaN and render blank.	required
sources	list[str]	Row labels, the datasets the frozen encoders were trained on.	required
targets	list[str]	Column labels, the datasets the probes were fitted and scored on.	required
metric	str	Name of the plotted metric, shown on the colour bar.	'auroc_macro'
baseline_rows	int	Number of leading rows holding a baseline rather than a trained encoder, separated from the rest by a rule.	0
title	str	Title for the plot.	'Cross-Dataset Transfer'
save_path	str | None	Path to save the figure to.	None

Source code in deepecgkit/evaluation/visualization.py

def plot_transfer_matrix(
    values: np.ndarray,
    sources: list[str],
    targets: list[str],
    metric: str = "auroc_macro",
    baseline_rows: int = 0,
    title: str = "Cross-Dataset Transfer",
    save_path: str | None = None,
) -> None:
    """Plot probe performance for every encoder source against every probe target.

    Args:
        values: Probe scores of shape (len(sources), len(targets)). Cells that
            could not be scored are NaN and render blank.
        sources: Row labels, the datasets the frozen encoders were trained on.
        targets: Column labels, the datasets the probes were fitted and scored on.
        metric: Name of the plotted metric, shown on the colour bar.
        baseline_rows: Number of leading rows holding a baseline rather than a
            trained encoder, separated from the rest by a rule.
        title: Title for the plot.
        save_path: Path to save the figure to.
    """
    matrix = np.asarray(values, dtype=float)

    _, ax = plt.subplots(figsize=(1.6 * len(targets) + 4, 0.9 * len(sources) + 3))

    sns.heatmap(
        matrix,
        annot=True,
        fmt=".3f",
        cmap="viridis",
        xticklabels=list(targets),
        yticklabels=list(sources),
        cbar_kws={"label": metric},
        ax=ax,
    )

    if baseline_rows:
        ax.axhline(baseline_rows, color="white", lw=3)

    ax.set_title(title)
    ax.set_xlabel("Probe target")
    ax.set_ylabel("Encoder source")
    plt.tight_layout()

    finalize_figure(save_path)

plot_embedding

plot_embedding(
    embedding: ndarray,
    labels: ndarray | None = None,
    class_names: list[str] | None = None,
    title: str = "Embedding",
    save_path: str | None = None,
) -> None

Scatter a two-dimensional embedding, coloured by class.

Parameters:

Name	Type	Description	Default
embedding	ndarray	Coordinates of shape (n_samples, 2), as returned by deepecgkit.evaluation.representation.embed_2d.	required
labels	ndarray | None	Class indices of shape (n_samples,), or a multi-hot matrix of shape (n_samples, n_classes) in which case each point is coloured by its first positive label.	None
class_names	list[str] | None	Names for the classes, defaulting to their indices.	None
title	str	Title for the plot.	'Embedding'
save_path	str | None	Path to save the figure to.	None

Raises:

Type	Description
ValueError	If the coordinates are not two-dimensional, or the labels do not align with them.

Source code in deepecgkit/evaluation/visualization.py

def plot_embedding(
    embedding: np.ndarray,
    labels: np.ndarray | None = None,
    class_names: list[str] | None = None,
    title: str = "Embedding",
    save_path: str | None = None,
) -> None:
    """Scatter a two-dimensional embedding, coloured by class.

    Args:
        embedding: Coordinates of shape (n_samples, 2), as returned by
            deepecgkit.evaluation.representation.embed_2d.
        labels: Class indices of shape (n_samples,), or a multi-hot matrix of
            shape (n_samples, n_classes) in which case each point is coloured by
            its first positive label.
        class_names: Names for the classes, defaulting to their indices.
        title: Title for the plot.
        save_path: Path to save the figure to.

    Raises:
        ValueError: If the coordinates are not two-dimensional, or the labels do
            not align with them.
    """
    coords = np.asarray(embedding, dtype=float)
    if coords.ndim != 2 or coords.shape[1] != 2:
        raise ValueError(
            f"Embedding must have shape (n_samples, 2); got {coords.shape}. Reduce it with "
            f"embed_2d first."
        )

    plt.figure(figsize=(8, 6))

    if labels is None:
        plt.scatter(coords[:, 0], coords[:, 1], s=12, alpha=0.7, color="tab:blue")
    else:
        groups = np.asarray(labels)
        if groups.ndim > 1 and groups.shape[-1] > 1:
            groups = np.argmax(groups, axis=1)
        groups = groups.reshape(-1).astype(int)
        if groups.shape[0] != coords.shape[0]:
            raise ValueError(
                f"Labels hold {groups.shape[0]} rows but the embedding holds {coords.shape[0]}."
            )

        classes = np.unique(groups)
        palette = plt.get_cmap("tab20" if len(classes) > 10 else "tab10")
        for position, value in enumerate(classes):
            mask = groups == value
            name = (
                class_names[value]
                if class_names is not None and value < len(class_names)
                else f"Class {value}"
            )
            plt.scatter(
                coords[mask, 0],
                coords[mask, 1],
                s=12,
                alpha=0.7,
                color=palette(position % palette.N),
                label=name,
            )
        plt.legend(fontsize="small", markerscale=2)

    plt.xlabel("Component 1")
    plt.ylabel("Component 2")
    plt.title(title)
    plt.grid(True, alpha=0.3)
    plt.tight_layout()

    finalize_figure(save_path)