Skip to content

deepecgkit.analysis

Multi-seed analysis over finished training runs: discovery and aggregation of run directories, ranking stability, the statistics behind both, and LaTeX table emitters. These modules back the deepecg aggregate and deepecg rank commands.

Aggregation

Discovers run directories (any directory holding classification_metrics.csv), groups them by model, dataset and seed, and summarises every metric across seeds.

aggregation

Discovery and aggregation of finished training runs.

A run directory is usable when it holds classification_metrics.csv. Two schemas are written by :mod:deepecgkit.evaluation.report depending on whether the task is multi-label, and they share only precision, recall, f1, auroc and the calibration columns, so the schema and the averaging mode travel with every record.

Every metric name is looked up defensively, so runs produced before a metric existed stay readable and simply contribute nothing to it. The calibration_bins column is deliberately not a metric: averaging a bin count across seeds would be meaningless.

RunRecord

Bases: NamedTuple

One finished training run and the metrics it produced.

Source code in deepecgkit/analysis/aggregation.py
class RunRecord(NamedTuple):
    """One finished training run and the metrics it produced."""

    run_dir: Path
    model: str
    dataset: str
    seed: int | None
    multi_label: bool
    averaging: str
    metrics: dict[str, float]
    n_train: int | None
    n_val: int | None
    n_test: int | None
    n_test_groups: int | None
    mtime: float
    returncode: int | None = None
    fingerprint: str | None = None

read_run_metrics

read_run_metrics(run_dir: Path) -> RunRecord | None

Read one run directory into a record.

Model and dataset come from the metrics CSV, which is authoritative. The seed comes from run_metadata.json when present and falls back to a seed<N> suffix on the directory name, so runs produced before seeds were recorded remain readable.

Parameters:

Name Type Description Default
run_dir Path

Directory holding classification_metrics.csv.

required

Returns:

Type Description
RunRecord | None

A populated record, or None when the directory holds no usable metrics.

Source code in deepecgkit/analysis/aggregation.py
def read_run_metrics(run_dir: Path) -> RunRecord | None:
    """Read one run directory into a record.

    Model and dataset come from the metrics CSV, which is authoritative. The
    seed comes from ``run_metadata.json`` when present and falls back to a
    ``seed<N>`` suffix on the directory name, so runs produced before seeds
    were recorded remain readable.

    Args:
        run_dir: Directory holding classification_metrics.csv.

    Returns:
        A populated record, or None when the directory holds no usable metrics.
    """
    metrics_path = run_dir / METRICS_FILENAME
    if not metrics_path.is_file():
        return None
    try:
        frame = pd.read_csv(metrics_path)
    except (ValueError, OSError, pd.errors.ParserError):
        return None
    if frame.empty:
        return None

    row = frame.iloc[0]
    columns = set(frame.columns)
    if "model" not in columns or "dataset" not in columns:
        return None

    multi_label = "exact_match_accuracy" in columns
    metric_names = MULTI_LABEL_METRICS if multi_label else SINGLE_LABEL_METRICS
    metrics: dict[str, float] = {}
    for name in metric_names:
        if name in columns:
            metrics[name] = float(row[name])

    if multi_label:
        averaging = AVERAGING_MACRO_LABEL
    elif "average" in columns and isinstance(row["average"], str):
        averaging = str(row["average"])
    else:
        averaging = AVERAGING_UNKNOWN

    metadata = _read_json(run_dir / METADATA_FILENAME) or {}
    seed = metadata.get("seed")
    if not isinstance(seed, int) or isinstance(seed, bool):
        seed = _seed_from_directory(run_dir)
    returncode = metadata.get("returncode")
    if not isinstance(returncode, int) or isinstance(returncode, bool):
        returncode = None
    fingerprint = metadata.get("params_fingerprint")
    if not isinstance(fingerprint, str):
        fingerprint = None

    split_info = _read_json(run_dir / SPLIT_INFO_FILENAME) or {}

    def _size(key: str) -> int | None:
        value = split_info.get(key)
        return int(value) if isinstance(value, int) and not isinstance(value, bool) else None

    return RunRecord(
        run_dir=run_dir,
        model=str(row["model"]),
        dataset=str(row["dataset"]),
        seed=seed,
        multi_label=multi_label,
        averaging=averaging,
        metrics=metrics,
        n_train=_size("n_train"),
        n_val=_size("n_val"),
        n_test=_size("n_test"),
        n_test_groups=_size("n_test_groups"),
        mtime=metrics_path.stat().st_mtime,
        returncode=returncode,
        fingerprint=fingerprint,
    )

discover_runs

discover_runs(
    roots: Sequence[Path],
    recursive: bool = False,
    include_legacy: bool = False,
    on_duplicate: str = ON_DUPLICATE_NEWEST,
    include_failed: bool = False,
) -> tuple[list[RunRecord], dict[str, list[str]]]

Collect usable runs beneath the given roots.

A run whose run_metadata.json records a non-zero returncode (the batch runner writes the training subprocess's exit status there) failed after writing its metrics and is set aside under the "failed" diagnostic unless include_failed is given. Runs without a recorded status are kept.

Two runs count as duplicates when they share dataset, model, seed and, when recorded, the batch runner's params_fingerprint; runs of one pairing and seed with different hyperparameters are therefore kept apart rather than resolved by modification time.

Parameters:

Name Type Description Default
roots Sequence[Path]

Directories to search.

required
recursive bool

Search at any depth rather than only immediate children.

False
include_legacy bool

Keep runs whose seed could not be determined, tagging them with a sentinel seed so they are visible but never silently pooled with seeded runs.

False
on_duplicate str

One of "newest", "error" or "all", controlling what happens when the same model, dataset and seed appear twice.

ON_DUPLICATE_NEWEST
include_failed bool

Keep runs whose recorded exit status is non-zero.

False

Returns:

Type Description
tuple[list[RunRecord], dict[str, list[str]]]

The records and a diagnostics mapping from reason to directory names.

Raises:

Type Description
ValueError

If on_duplicate is unrecognised, or is "error" and a duplicate is found.

Source code in deepecgkit/analysis/aggregation.py
def discover_runs(
    roots: Sequence[Path],
    recursive: bool = False,
    include_legacy: bool = False,
    on_duplicate: str = ON_DUPLICATE_NEWEST,
    include_failed: bool = False,
) -> tuple[list[RunRecord], dict[str, list[str]]]:
    """Collect usable runs beneath the given roots.

    A run whose ``run_metadata.json`` records a non-zero ``returncode`` (the
    batch runner writes the training subprocess's exit status there) failed
    after writing its metrics and is set aside under the "failed" diagnostic
    unless ``include_failed`` is given. Runs without a recorded status are
    kept.

    Two runs count as duplicates when they share dataset, model, seed and, when
    recorded, the batch runner's ``params_fingerprint``; runs of one pairing
    and seed with different hyperparameters are therefore kept apart rather
    than resolved by modification time.

    Args:
        roots: Directories to search.
        recursive: Search at any depth rather than only immediate children.
        include_legacy: Keep runs whose seed could not be determined, tagging
            them with a sentinel seed so they are visible but never silently
            pooled with seeded runs.
        on_duplicate: One of "newest", "error" or "all", controlling what
            happens when the same model, dataset and seed appear twice.
        include_failed: Keep runs whose recorded exit status is non-zero.

    Returns:
        The records and a diagnostics mapping from reason to directory names.

    Raises:
        ValueError: If on_duplicate is unrecognised, or is "error" and a
            duplicate is found.
    """
    if on_duplicate not in (ON_DUPLICATE_NEWEST, ON_DUPLICATE_ERROR, ON_DUPLICATE_ALL):
        raise ValueError(f"Unknown on_duplicate: {on_duplicate}")

    diagnostics: dict[str, list[str]] = {
        "unreadable": [],
        "no_seed": [],
        "duplicate": [],
        "failed": [],
    }
    candidates: list[Path] = []
    for root in roots:
        root_path = Path(root)
        if not root_path.is_dir():
            continue
        pattern = f"**/{METRICS_FILENAME}" if recursive else f"*/{METRICS_FILENAME}"
        candidates.extend(sorted(path.parent for path in root_path.glob(pattern)))

    records: list[RunRecord] = []
    for run_dir in candidates:
        record = read_run_metrics(run_dir)
        if record is None:
            diagnostics["unreadable"].append(str(run_dir))
            continue
        if record.returncode not in (None, 0):
            diagnostics["failed"].append(f"{run_dir} (returncode {record.returncode})")
            if not include_failed:
                continue
        if record.seed is None:
            diagnostics["no_seed"].append(str(run_dir))
            if not include_legacy:
                continue
            record = record._replace(seed=LEGACY_SEED)
        records.append(record)

    if on_duplicate == ON_DUPLICATE_ALL:
        return records, diagnostics

    kept: dict[tuple[str, str, int, str | None], RunRecord] = {}
    for record in records:
        key = (record.dataset, record.model, int(record.seed), record.fingerprint)
        previous = kept.get(key)
        if previous is None:
            kept[key] = record
            continue
        if on_duplicate == ON_DUPLICATE_ERROR:
            raise ValueError(
                f"Duplicate run for {record.model} on {record.dataset} seed {record.seed}: "
                f"{previous.run_dir} and {record.run_dir}"
            )
        newer, older = (record, previous) if record.mtime > previous.mtime else (previous, record)
        diagnostics["duplicate"].append(f"{older.run_dir} (dropped; kept {newer.run_dir})")
        kept[key] = newer

    ordered = sorted(
        kept.values(), key=lambda r: (r.dataset, r.model, int(r.seed), r.fingerprint or "")
    )
    return ordered, diagnostics

runs_to_frame

runs_to_frame(records: Sequence[RunRecord]) -> pd.DataFrame

Flatten records into one row per run and metric.

Parameters:

Name Type Description Default
records Sequence[RunRecord]

Discovered runs.

required

Returns:

Type Description
DataFrame

A long frame with columns run_dir, model, dataset, seed, averaging,

DataFrame

multi_label, n_train, n_val, n_test, n_test_groups, metric and value.

Source code in deepecgkit/analysis/aggregation.py
def runs_to_frame(records: Sequence[RunRecord]) -> pd.DataFrame:
    """Flatten records into one row per run and metric.

    Args:
        records: Discovered runs.

    Returns:
        A long frame with columns run_dir, model, dataset, seed, averaging,
        multi_label, n_train, n_val, n_test, n_test_groups, metric and value.
    """
    rows: list[dict[str, Any]] = []
    for record in records:
        for metric, value in record.metrics.items():
            rows.append(
                {
                    "run_dir": str(record.run_dir),
                    "model": record.model,
                    "dataset": record.dataset,
                    "seed": record.seed,
                    "averaging": record.averaging,
                    "multi_label": record.multi_label,
                    "n_train": record.n_train,
                    "n_val": record.n_val,
                    "n_test": record.n_test,
                    "n_test_groups": record.n_test_groups,
                    "metric": metric,
                    "value": value,
                }
            )
    columns = [
        "run_dir",
        "model",
        "dataset",
        "seed",
        "averaging",
        "multi_label",
        "n_train",
        "n_val",
        "n_test",
        "n_test_groups",
        "metric",
        "value",
    ]
    return pd.DataFrame(rows, columns=columns)

reconcile_averaging

reconcile_averaging(
    runs: DataFrame,
) -> tuple[pd.DataFrame, pd.DataFrame]

Drop runs whose averaging mode disagrees with their group's majority.

resolve_average returns "binary" whenever the test split holds two or fewer classes, so precision, recall and F1 are positive-class values there and macro elsewhere. Pooling the two would average incommensurable quantities. The minority is dropped rather than blanking the whole cell, because one unlucky split should not discard four sound runs.

Parameters:

Name Type Description Default
runs DataFrame

Long frame from :func:runs_to_frame.

required

Returns:

Type Description
tuple[DataFrame, DataFrame]

The surviving rows and a frame describing what was dropped.

Source code in deepecgkit/analysis/aggregation.py
def reconcile_averaging(runs: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]:
    """Drop runs whose averaging mode disagrees with their group's majority.

    ``resolve_average`` returns "binary" whenever the test split holds two or
    fewer classes, so precision, recall and F1 are positive-class values there
    and macro elsewhere. Pooling the two would average incommensurable
    quantities. The minority is dropped rather than blanking the whole cell,
    because one unlucky split should not discard four sound runs.

    Args:
        runs: Long frame from :func:`runs_to_frame`.

    Returns:
        The surviving rows and a frame describing what was dropped.
    """
    if runs.empty:
        return runs, pd.DataFrame(columns=["dataset", "model", "kept_averaging", "dropped_runs"])

    keep_mask = pd.Series(True, index=runs.index)
    conflicts: list[dict[str, Any]] = []
    for (dataset, model), group in runs.groupby(["dataset", "model"], sort=True):
        modes = group.drop_duplicates("run_dir")["averaging"]
        if modes.nunique() <= 1:
            continue
        majority = modes.value_counts().idxmax()
        offending = group["averaging"] != majority
        keep_mask.loc[group.index[offending]] = False
        dropped = sorted(set(group.loc[offending, "run_dir"]))
        conflicts.append(
            {
                "dataset": dataset,
                "model": model,
                "kept_averaging": majority,
                "dropped_runs": len(dropped),
            }
        )
        warnings.warn(
            f"{model} on {dataset} mixes averaging modes {sorted(set(modes))}; keeping "
            f"'{majority}' and dropping {len(dropped)} run(s)",
            RuntimeWarning,
            stacklevel=2,
        )

    conflict_frame = pd.DataFrame(
        conflicts, columns=["dataset", "model", "kept_averaging", "dropped_runs"]
    )
    return runs.loc[keep_mask].reset_index(drop=True), conflict_frame

aggregate_runs

aggregate_runs(
    runs: DataFrame,
    confidence_level: float = CONFIDENCE_LEVEL,
) -> pd.DataFrame

Summarise every metric across seeds for each dataset and model.

Parameters:

Name Type Description Default
runs DataFrame

Long frame from :func:runs_to_frame, already reconciled.

required
confidence_level float

Two-sided coverage for the Student-t interval.

CONFIDENCE_LEVEL

Returns:

Type Description
DataFrame

One row per dataset, model and metric with mean, std, sem, interval

DataFrame

bounds, the usable count and the contributing seeds.

Source code in deepecgkit/analysis/aggregation.py
def aggregate_runs(runs: pd.DataFrame, confidence_level: float = CONFIDENCE_LEVEL) -> pd.DataFrame:
    """Summarise every metric across seeds for each dataset and model.

    Args:
        runs: Long frame from :func:`runs_to_frame`, already reconciled.
        confidence_level: Two-sided coverage for the Student-t interval.

    Returns:
        One row per dataset, model and metric with mean, std, sem, interval
        bounds, the usable count and the contributing seeds.
    """
    columns = [
        "dataset",
        "model",
        "metric",
        "averaging",
        "n_runs",
        "n",
        "mean",
        "std",
        "sem",
        "lower_ci",
        "upper_ci",
        "seeds",
    ]
    if runs.empty:
        return pd.DataFrame(columns=columns)

    rows: list[dict[str, Any]] = []
    for (dataset, model, metric), group in runs.groupby(["dataset", "model", "metric"], sort=True):
        summary = summarize_scores(group["value"].tolist(), confidence_level=confidence_level)
        finite = group.loc[np.isfinite(group["value"]), "seed"]
        rows.append(
            {
                "dataset": dataset,
                "model": model,
                "metric": metric,
                "averaging": group["averaging"].iloc[0],
                "n_runs": len(group),
                "n": summary["n"],
                "mean": summary["mean"],
                "std": summary["std"],
                "sem": summary["sem"],
                "lower_ci": summary["lower_ci"],
                "upper_ci": summary["upper_ci"],
                "seeds": ",".join(str(int(s)) for s in sorted(finite.tolist())),
            }
        )
    return pd.DataFrame(rows, columns=columns)

Ranking

Rank stability and cross-dataset flip tests across random seeds: per-seed ranks, rank distributions, Kendall's tau between seeds, and paired comparisons between models.

ranking

Ranking stability and cross-dataset flip tests across random seeds.

Ranks are formed inside a single dataset and seed, so the averaging differences between datasets never contaminate them. Comparisons between two models on one dataset are paired: within a seed both models see the same split, so the split effect cancels in the difference.

rank_within_seed

rank_within_seed(
    runs: DataFrame, metric: str = DEFAULT_RANK_METRIC
) -> pd.DataFrame

Rank models inside each dataset and seed, best first.

Parameters:

Name Type Description Default
runs DataFrame

Long frame of per-run metric values.

required
metric str

Metric to rank on.

DEFAULT_RANK_METRIC

Returns:

Type Description
DataFrame

A frame with dataset, seed, model, value, rank and rank_average. Ties

DataFrame

share the better rank under "min" for display, while rank_average is

DataFrame

used for Kendall's tau.

Source code in deepecgkit/analysis/ranking.py
def rank_within_seed(runs: pd.DataFrame, metric: str = DEFAULT_RANK_METRIC) -> pd.DataFrame:
    """Rank models inside each dataset and seed, best first.

    Args:
        runs: Long frame of per-run metric values.
        metric: Metric to rank on.

    Returns:
        A frame with dataset, seed, model, value, rank and rank_average. Ties
        share the better rank under "min" for display, while rank_average is
        used for Kendall's tau.
    """
    selected = _metric_slice(runs, metric)
    frames: list[pd.DataFrame] = []
    for (dataset, seed), group in selected.groupby(["dataset", "seed"], sort=True):
        usable = group[np.isfinite(group["value"])]
        if usable.empty:
            continue
        values = usable["value"].to_numpy(dtype=float)
        frames.append(
            pd.DataFrame(
                {
                    "dataset": dataset,
                    "seed": seed,
                    "model": usable["model"].tolist(),
                    "value": values,
                    "rank": stats.rankdata(-values, method="min"),
                    "rank_average": stats.rankdata(-values, method="average"),
                }
            )
        )
    if not frames:
        return pd.DataFrame(columns=["dataset", "seed", "model", "value", "rank", "rank_average"])
    return pd.concat(frames, ignore_index=True)

rank_distribution

rank_distribution(
    runs: DataFrame, metric: str = DEFAULT_RANK_METRIC
) -> pd.DataFrame

Count how often each model attained each rank.

Parameters:

Name Type Description Default
runs DataFrame

Long frame of per-run metric values.

required
metric str

Metric to rank on.

DEFAULT_RANK_METRIC

Returns:

Type Description
DataFrame

One row per dataset and model with a rank_ count column per

DataFrame

attainable rank, the modal rank, the fraction of seeds at that rank,

DataFrame

and the fraction of seeds spent at rank one.

Source code in deepecgkit/analysis/ranking.py
def rank_distribution(runs: pd.DataFrame, metric: str = DEFAULT_RANK_METRIC) -> pd.DataFrame:
    """Count how often each model attained each rank.

    Args:
        runs: Long frame of per-run metric values.
        metric: Metric to rank on.

    Returns:
        One row per dataset and model with a rank_<k> count column per
        attainable rank, the modal rank, the fraction of seeds at that rank,
        and the fraction of seeds spent at rank one.
    """
    ranked = rank_within_seed(runs, metric)
    if ranked.empty:
        return pd.DataFrame()

    rows: list[dict[str, Any]] = []
    for dataset, group in ranked.groupby("dataset", sort=True):
        n_models = int(group["model"].nunique())
        for model, sub in group.groupby("model", sort=True):
            counts = sub["rank"].value_counts()
            n_seeds = len(sub)
            row: dict[str, Any] = {
                "dataset": dataset,
                "model": model,
                "n_seeds": n_seeds,
                "n_models": n_models,
            }
            for position in range(1, n_models + 1):
                row[f"rank_{position}"] = int(counts.get(position, 0))
            modal_rank = int(counts.idxmax())
            row["mode_rank"] = modal_rank
            row["mode_fraction"] = float(counts.max()) / n_seeds if n_seeds else float("nan")
            row["top1_fraction"] = float(counts.get(1, 0)) / n_seeds if n_seeds else float("nan")
            row["mean_value"] = float(np.mean(sub["value"]))
            row["std_value"] = float(np.std(sub["value"], ddof=1)) if n_seeds > 1 else float("nan")
            rows.append(row)
    return pd.DataFrame(rows)

tau_summary

tau_summary(
    runs: DataFrame, metric: str = DEFAULT_RANK_METRIC
) -> pd.DataFrame

Kendall's tau-b between every pair of per-seed rankings.

Parameters:

Name Type Description Default
runs DataFrame

Long frame of per-run metric values.

required
metric str

Metric to rank on.

DEFAULT_RANK_METRIC

Returns:

Name Type Description
DataFrame

One row per dataset with the mean, minimum and maximum tau over all

DataFrame

seed pairs, the number of pairs, and the smallest two-sided p-value

DataFrame

the coefficient could attain given the number of models. No p-value is

computed DataFrame

at four models that floor is 0.083, so significance is

DataFrame

unreachable and the statistic is descriptive only.

Source code in deepecgkit/analysis/ranking.py
def tau_summary(runs: pd.DataFrame, metric: str = DEFAULT_RANK_METRIC) -> pd.DataFrame:
    """Kendall's tau-b between every pair of per-seed rankings.

    Args:
        runs: Long frame of per-run metric values.
        metric: Metric to rank on.

    Returns:
        One row per dataset with the mean, minimum and maximum tau over all
        seed pairs, the number of pairs, and the smallest two-sided p-value
        the coefficient could attain given the number of models. No p-value is
        computed: at four models that floor is 0.083, so significance is
        unreachable and the statistic is descriptive only.
    """
    ranked = rank_within_seed(runs, metric)
    rows: list[dict[str, Any]] = []
    if ranked.empty:
        return pd.DataFrame(
            columns=[
                "dataset",
                "n_models",
                "n_seeds",
                "n_pairs",
                "tau_mean",
                "tau_min",
                "tau_max",
                "min_attainable_p",
            ]
        )

    for dataset, group in ranked.groupby("dataset", sort=True):
        by_seed: dict[int, dict[str, float]] = {}
        for seed, sub in group.groupby("seed", sort=True):
            by_seed[int(seed)] = dict(zip(sub["model"], sub["rank_average"], strict=False))

        common = set.intersection(*(set(v) for v in by_seed.values())) if by_seed else set()
        n_models = len(common)
        taus: list[float] = []
        if n_models >= MIN_MODELS_FOR_TAU:
            for first, second in itertools.combinations(sorted(by_seed), 2):
                left = {model: by_seed[first][model] for model in common}
                right = {model: by_seed[second][model] for model in common}
                taus.append(kendall_tau(left, right))
        rows.append(
            {
                "dataset": dataset,
                "n_models": n_models,
                "n_seeds": len(by_seed),
                "n_pairs": len(taus),
                "tau_mean": float(np.mean(taus)) if taus else float("nan"),
                "tau_min": float(np.min(taus)) if taus else float("nan"),
                "tau_max": float(np.max(taus)) if taus else float("nan"),
                "min_attainable_p": (
                    min_attainable_kendall_p(n_models)
                    if n_models >= MIN_MODELS_FOR_TAU
                    else float("nan")
                ),
            }
        )
    return pd.DataFrame(rows)

paired_difference

paired_difference(
    runs: DataFrame,
    model_a: str,
    model_b: str,
    dataset: str,
    metric: str = DEFAULT_RANK_METRIC,
    confidence_level: float = CONFIDENCE_LEVEL,
    correction: str = NADEAU_BENGIO,
    min_test_groups: int = MIN_TEST_GROUPS,
) -> dict[str, Any]

Paired per-seed difference between two models on one dataset.

Pairing is exact: within a seed both models train and evaluate on the same split, so the split effect cancels. Across seeds the splits overlap heavily, so the standard error carries the Nadeau-Bengio correction.

Parameters:

Name Type Description Default
runs DataFrame

Long frame of per-run metric values.

required
model_a str

Model whose score is the minuend.

required
model_b str

Model whose score is the subtrahend.

required
dataset str

Dataset to compare on.

required
metric str

Metric to compare.

DEFAULT_RANK_METRIC
confidence_level float

Two-sided coverage.

CONFIDENCE_LEVEL
correction str

Standard error correction, "nadeau-bengio" or "none".

NADEAU_BENGIO
min_test_groups int

Fewest independent test units that can support an interval.

MIN_TEST_GROUPS

Returns:

Type Description
dict[str, Any]

The paired summary augmented with seeds, averaging, the recorded split

dict[str, Any]

sizes, a bimodality flag and a reason string.

Source code in deepecgkit/analysis/ranking.py
def paired_difference(
    runs: pd.DataFrame,
    model_a: str,
    model_b: str,
    dataset: str,
    metric: str = DEFAULT_RANK_METRIC,
    confidence_level: float = CONFIDENCE_LEVEL,
    correction: str = NADEAU_BENGIO,
    min_test_groups: int = MIN_TEST_GROUPS,
) -> dict[str, Any]:
    """Paired per-seed difference between two models on one dataset.

    Pairing is exact: within a seed both models train and evaluate on the same
    split, so the split effect cancels. Across seeds the splits overlap
    heavily, so the standard error carries the Nadeau-Bengio correction.

    Args:
        runs: Long frame of per-run metric values.
        model_a: Model whose score is the minuend.
        model_b: Model whose score is the subtrahend.
        dataset: Dataset to compare on.
        metric: Metric to compare.
        confidence_level: Two-sided coverage.
        correction: Standard error correction, "nadeau-bengio" or "none".
        min_test_groups: Fewest independent test units that can support an
            interval.

    Returns:
        The paired summary augmented with seeds, averaging, the recorded split
        sizes, a bimodality flag and a reason string.
    """
    selected = _metric_slice(runs, metric)
    seeds, a_values, b_values, contributing = _paired_values(selected, model_a, model_b, dataset)
    differences = a_values - b_values

    n_train = _mean_optional(contributing, "n_train")
    n_test = _mean_optional(contributing, "n_test")
    n_test_groups = _mean_optional(contributing, "n_test_groups")

    summary = paired_summary(
        differences,
        confidence_level=confidence_level,
        correction=correction,
        n_train=n_train,
        n_test=n_test,
    )
    bimodal_a, reason_a = is_bimodal(a_values)
    bimodal_b, reason_b = is_bimodal(b_values)

    reason = ""
    if n_test_groups is not None and n_test_groups < min_test_groups:
        reason = (
            f"test partition holds {n_test_groups} independent unit(s), which cannot "
            "support an interval"
        )
    elif summary["n"] and summary["std"] == 0.0:
        reason = "zero variance across seeds; the interval is undefined, not infinitely precise"
    elif bimodal_a:
        reason = f"{model_a}: {reason_a}"
    elif bimodal_b:
        reason = f"{model_b}: {reason_b}"

    result: dict[str, Any] = dict(summary)
    result.update(
        {
            "model_a": model_a,
            "model_b": model_b,
            "dataset": dataset,
            "metric": metric,
            "seeds": seeds,
            "averaging": (
                contributing["averaging"].iloc[0] if not contributing.empty else "unknown"
            ),
            "n_train": n_train,
            "n_test": n_test,
            "n_test_groups": n_test_groups,
            "bimodal": bool(bimodal_a or bimodal_b),
            "testable": not (n_test_groups is not None and n_test_groups < min_test_groups),
            "reason": reason,
        }
    )
    return result

flip_test

flip_test(
    runs: DataFrame,
    model_a: str,
    model_b: str,
    dataset_1: str,
    dataset_2: str,
    metric: str = DEFAULT_RANK_METRIC,
    confidence_level: float = CONFIDENCE_LEVEL,
    correction: str = NADEAU_BENGIO,
    min_test_groups: int = MIN_TEST_GROUPS,
    min_seeds: int = MIN_SEEDS_FOR_INTERVAL,
    min_seeds_for_claim: int = MIN_SEEDS_FOR_CLAIM,
) -> dict[str, Any]

Intersection-union test for a cross-dataset ranking reversal.

The reversal is supported only when both paired intervals exclude zero with opposite signs. Because the decision requires both components, no multiplicity correction is applied between them, and the reported p is the larger of the two. That p is an upper bound: pairing two-sided intervals with a sign requirement is a one-sided test at half the nominal level.

Parameters:

Name Type Description Default
runs DataFrame

Long frame of per-run metric values.

required
model_a str

First model.

required
model_b str

Second model.

required
dataset_1 str

Dataset on which A is expected to lead.

required
dataset_2 str

Dataset on which B is expected to lead.

required
metric str

Metric to compare.

DEFAULT_RANK_METRIC
confidence_level float

Two-sided coverage.

CONFIDENCE_LEVEL
correction str

Standard error correction.

NADEAU_BENGIO
min_test_groups int

Fewest independent test units that can support a test.

MIN_TEST_GROUPS
min_seeds int

Fewest seeds for which an interval is computed at all.

MIN_SEEDS_FOR_INTERVAL
min_seeds_for_claim int

Fewest seeds that may yield a supported verdict.

MIN_SEEDS_FOR_CLAIM

Returns:

Type Description
dict[str, Any]

Both paired summaries, the verdict, p_flip, and the averaging modes

dict[str, Any]

with a flag for whether the two magnitudes are comparable.

Source code in deepecgkit/analysis/ranking.py
def flip_test(
    runs: pd.DataFrame,
    model_a: str,
    model_b: str,
    dataset_1: str,
    dataset_2: str,
    metric: str = DEFAULT_RANK_METRIC,
    confidence_level: float = CONFIDENCE_LEVEL,
    correction: str = NADEAU_BENGIO,
    min_test_groups: int = MIN_TEST_GROUPS,
    min_seeds: int = MIN_SEEDS_FOR_INTERVAL,
    min_seeds_for_claim: int = MIN_SEEDS_FOR_CLAIM,
) -> dict[str, Any]:
    """Intersection-union test for a cross-dataset ranking reversal.

    The reversal is supported only when both paired intervals exclude zero
    with opposite signs. Because the decision requires both components, no
    multiplicity correction is applied between them, and the reported p is the
    larger of the two. That p is an upper bound: pairing two-sided intervals
    with a sign requirement is a one-sided test at half the nominal level.

    Args:
        runs: Long frame of per-run metric values.
        model_a: First model.
        model_b: Second model.
        dataset_1: Dataset on which A is expected to lead.
        dataset_2: Dataset on which B is expected to lead.
        metric: Metric to compare.
        confidence_level: Two-sided coverage.
        correction: Standard error correction.
        min_test_groups: Fewest independent test units that can support a test.
        min_seeds: Fewest seeds for which an interval is computed at all.
        min_seeds_for_claim: Fewest seeds that may yield a supported verdict.

    Returns:
        Both paired summaries, the verdict, p_flip, and the averaging modes
        with a flag for whether the two magnitudes are comparable.
    """
    first = paired_difference(
        runs, model_a, model_b, dataset_1, metric, confidence_level, correction, min_test_groups
    )
    second = paired_difference(
        runs, model_a, model_b, dataset_2, metric, confidence_level, correction, min_test_groups
    )

    n_seeds = min(int(first["n"]), int(second["n"]))
    verdict, reason = _classify_flip(first, second, n_seeds, min_seeds, min_seeds_for_claim)

    # The intersection-union p-value exists whenever both components could
    # be tested; it is the larger component p. Which rows enter the
    # false-discovery-rate family is decided by all_flip_tests.
    if verdict in (VERDICT_UNTESTABLE, VERDICT_BIMODAL):
        p_flip = float("nan")
    else:
        p_flip = max(float(first["p_value"]), float(second["p_value"]))

    return {
        "model_a": model_a,
        "model_b": model_b,
        "dataset_1": dataset_1,
        "dataset_2": dataset_2,
        "metric": metric,
        "difference_1": first,
        "difference_2": second,
        "mean_1": first["mean"],
        "mean_2": second["mean"],
        "lower_1": first["lower_ci"],
        "upper_1": first["upper_ci"],
        "lower_2": second["lower_ci"],
        "upper_2": second["upper_ci"],
        "wins_1": first["n_wins"],
        "wins_2": second["n_wins"],
        "n_1": first["n"],
        "n_2": second["n"],
        "n_seeds": n_seeds,
        "averaging_1": first["averaging"],
        "averaging_2": second["averaging"],
        "comparable_magnitudes": first["averaging"] == second["averaging"],
        "verdict": verdict,
        "p_flip": p_flip,
        "reason": reason,
    }

all_flip_tests

all_flip_tests(
    runs: DataFrame,
    metric: str = DEFAULT_RANK_METRIC,
    confidence_level: float = CONFIDENCE_LEVEL,
    correction: str = NADEAU_BENGIO,
    models: Sequence[str] | None = None,
    datasets: Sequence[str] | None = None,
    min_test_groups: int = MIN_TEST_GROUPS,
    fdr_family: str = FDR_FAMILY_TESTABLE,
) -> pd.DataFrame

Run a flip test for every model pair shared by every dataset pair.

Only models present on both datasets of a pair are compared, because models that failed to converge were omitted from some datasets. The exploratory grid is adjusted with Benjamini-Yekutieli over the family chosen by fdr_family: every testable comparison (the default, so that the family does not depend on the outcomes), or only the comparisons whose verdict is supported, suggestive or contradicted (the historical behaviour, whose family size depends on the verdicts).

Parameters:

Name Type Description Default
runs DataFrame

Long frame of per-run metric values.

required
metric str

Metric to compare.

DEFAULT_RANK_METRIC
confidence_level float

Two-sided coverage.

CONFIDENCE_LEVEL
correction str

Standard error correction.

NADEAU_BENGIO
models Sequence[str] | None

Restrict to these models.

None
datasets Sequence[str] | None

Restrict to these datasets.

None
min_test_groups int

Fewest independent test units that can support a test.

MIN_TEST_GROUPS
fdr_family str

FDR_FAMILY_TESTABLE or FDR_FAMILY_SIGNIFICANT.

FDR_FAMILY_TESTABLE

Returns:

Type Description
DataFrame

One row per tested combination, with raw and adjusted p-values and

DataFrame

the family definition in fdr_family.

Raises:

Type Description
ValueError

If fdr_family is unknown.

Source code in deepecgkit/analysis/ranking.py
def all_flip_tests(
    runs: pd.DataFrame,
    metric: str = DEFAULT_RANK_METRIC,
    confidence_level: float = CONFIDENCE_LEVEL,
    correction: str = NADEAU_BENGIO,
    models: Sequence[str] | None = None,
    datasets: Sequence[str] | None = None,
    min_test_groups: int = MIN_TEST_GROUPS,
    fdr_family: str = FDR_FAMILY_TESTABLE,
) -> pd.DataFrame:
    """Run a flip test for every model pair shared by every dataset pair.

    Only models present on both datasets of a pair are compared, because
    models that failed to converge were omitted from some datasets. The
    exploratory grid is adjusted with Benjamini-Yekutieli over the family
    chosen by ``fdr_family``: every testable comparison (the default, so that
    the family does not depend on the outcomes), or only the comparisons
    whose verdict is supported, suggestive or contradicted (the historical
    behaviour, whose family size depends on the verdicts).

    Args:
        runs: Long frame of per-run metric values.
        metric: Metric to compare.
        confidence_level: Two-sided coverage.
        correction: Standard error correction.
        models: Restrict to these models.
        datasets: Restrict to these datasets.
        min_test_groups: Fewest independent test units that can support a test.
        fdr_family: FDR_FAMILY_TESTABLE or FDR_FAMILY_SIGNIFICANT.

    Returns:
        One row per tested combination, with raw and adjusted p-values and
        the family definition in ``fdr_family``.

    Raises:
        ValueError: If fdr_family is unknown.
    """
    if fdr_family not in (FDR_FAMILY_TESTABLE, FDR_FAMILY_SIGNIFICANT):
        raise ValueError(f"Unknown fdr_family: {fdr_family}")
    selected = _metric_slice(runs, metric)
    if models is not None:
        selected = selected[selected["model"].isin(list(models))]
    if datasets is not None:
        selected = selected[selected["dataset"].isin(list(datasets))]

    available = sorted(selected["dataset"].unique())
    rows: list[dict[str, Any]] = []
    for dataset_1, dataset_2 in itertools.combinations(available, 2):
        models_1 = set(selected[selected["dataset"] == dataset_1]["model"])
        models_2 = set(selected[selected["dataset"] == dataset_2]["model"])
        shared = sorted(models_1 & models_2)
        for model_a, model_b in itertools.combinations(shared, 2):
            result = flip_test(
                selected,
                model_a,
                model_b,
                dataset_1,
                dataset_2,
                metric=metric,
                confidence_level=confidence_level,
                correction=correction,
                min_test_groups=min_test_groups,
            )
            result.pop("difference_1", None)
            result.pop("difference_2", None)
            rows.append(result)

    frame = pd.DataFrame(rows)
    if frame.empty:
        return frame
    in_family = np.isfinite(frame["p_flip"].to_numpy(dtype=float))
    if fdr_family == FDR_FAMILY_SIGNIFICANT:
        in_family &= (
            frame["verdict"]
            .isin([VERDICT_SUPPORTED, VERDICT_SUGGESTIVE, VERDICT_CONTRADICTED])
            .to_numpy()
        )
    family_p = [
        p if keep else float("nan") for p, keep in zip(frame["p_flip"], in_family, strict=False)
    ]
    frame["p_adjusted"] = benjamini_yekutieli_adjust(family_p)
    frame["fdr_family"] = fdr_family
    frame["fdr_family_size"] = int(in_family.sum())
    return frame.sort_values(["dataset_1", "dataset_2", "p_flip"], na_position="last").reset_index(
        drop=True
    )

evaluate_rank_claim

evaluate_rank_claim(
    runs: DataFrame,
    model: str,
    dataset_best: str,
    dataset_worst: str,
    metric: str = DEFAULT_RANK_METRIC,
    opponents: Sequence[str] | None = None,
    confidence_level: float = CONFIDENCE_LEVEL,
    correction: str = NADEAU_BENGIO,
    min_test_groups: int = MIN_TEST_GROUPS,
) -> dict[str, Any]

Decompose a "best here, worst there" claim into component flip tests.

Every model present on both datasets is a component. Components that cannot be tested are reported as untestable rather than dropped, so the claim is never presented as supported on a silently reduced subset.

Parameters:

Name Type Description Default
runs DataFrame

Long frame of per-run metric values.

required
model str

The model the claim is about.

required
dataset_best str

Dataset on which the model is claimed strongest.

required
dataset_worst str

Dataset on which the model is claimed weakest.

required
metric str

Metric to compare.

DEFAULT_RANK_METRIC
opponents Sequence[str] | None

Restrict the components to these models.

None
confidence_level float

Two-sided coverage.

CONFIDENCE_LEVEL
correction str

Standard error correction.

NADEAU_BENGIO
min_test_groups int

Fewest independent test units that can support a test.

MIN_TEST_GROUPS

Returns:

Type Description
dict[str, Any]

The component results, the overall verdict and the governing p-value.

Source code in deepecgkit/analysis/ranking.py
def evaluate_rank_claim(
    runs: pd.DataFrame,
    model: str,
    dataset_best: str,
    dataset_worst: str,
    metric: str = DEFAULT_RANK_METRIC,
    opponents: Sequence[str] | None = None,
    confidence_level: float = CONFIDENCE_LEVEL,
    correction: str = NADEAU_BENGIO,
    min_test_groups: int = MIN_TEST_GROUPS,
) -> dict[str, Any]:
    """Decompose a "best here, worst there" claim into component flip tests.

    Every model present on both datasets is a component. Components that
    cannot be tested are reported as untestable rather than dropped, so the
    claim is never presented as supported on a silently reduced subset.

    Args:
        runs: Long frame of per-run metric values.
        model: The model the claim is about.
        dataset_best: Dataset on which the model is claimed strongest.
        dataset_worst: Dataset on which the model is claimed weakest.
        metric: Metric to compare.
        opponents: Restrict the components to these models.
        confidence_level: Two-sided coverage.
        correction: Standard error correction.
        min_test_groups: Fewest independent test units that can support a test.

    Returns:
        The component results, the overall verdict and the governing p-value.
    """
    selected = _metric_slice(runs, metric)
    models_best = set(selected[selected["dataset"] == dataset_best]["model"])
    models_worst = set(selected[selected["dataset"] == dataset_worst]["model"])
    candidates = sorted((models_best & models_worst) - {model})
    if opponents is not None:
        candidates = [name for name in candidates if name in set(opponents)]

    components: list[dict[str, Any]] = []
    for opponent in candidates:
        result = flip_test(
            selected,
            model,
            opponent,
            dataset_best,
            dataset_worst,
            metric=metric,
            confidence_level=confidence_level,
            correction=correction,
            min_test_groups=min_test_groups,
        )
        result.pop("difference_1", None)
        result.pop("difference_2", None)
        # The flip test is symmetric in the direction of the reversal. A claim
        # is not: the model must lead on dataset_best and trail on
        # dataset_worst. A supported reversal the other way round contradicts
        # the claim and is relabelled, with the reason recorded.
        leads_best = float(result["mean_1"]) > 0
        trails_worst = float(result["mean_2"]) < 0
        result["claimed_direction"] = bool(leads_best and trails_worst)
        if result["verdict"] in (VERDICT_SUPPORTED, VERDICT_SUGGESTIVE) and not (
            leads_best and trails_worst
        ):
            result["verdict"] = VERDICT_CONTRADICTED
            result["reason"] = "the ordering reverses, but opposite to the claimed direction"
        components.append(result)

    verdicts = [component["verdict"] for component in components]
    if not components:
        overall = VERDICT_UNTESTABLE
    elif all(verdict == VERDICT_SUPPORTED for verdict in verdicts):
        overall = VERDICT_SUPPORTED
    elif any(verdict == VERDICT_CONTRADICTED for verdict in verdicts):
        overall = VERDICT_CONTRADICTED
    elif any(verdict in (VERDICT_UNTESTABLE, VERDICT_BIMODAL) for verdict in verdicts):
        overall = VERDICT_UNTESTABLE
    elif any(verdict == VERDICT_NOT_REPRODUCED for verdict in verdicts):
        overall = VERDICT_NOT_REPRODUCED
    else:
        overall = VERDICT_UNRESOLVED

    finite = [c["p_flip"] for c in components if np.isfinite(c["p_flip"])]
    return {
        "model": model,
        "dataset_best": dataset_best,
        "dataset_worst": dataset_worst,
        "metric": metric,
        "components": components,
        "n_components": len(components),
        "n_supported": sum(1 for v in verdicts if v == VERDICT_SUPPORTED),
        "n_supported_in_direction": sum(
            1 for c in components if c["verdict"] == VERDICT_SUPPORTED and c["claimed_direction"]
        ),
        "n_untestable": sum(1 for v in verdicts if v in (VERDICT_UNTESTABLE, VERDICT_BIMODAL)),
        "verdict": overall,
        "p_claim": max(finite) if finite else float("nan"),
    }

Statistics

Summary statistics and paired tests over per-seed scalar metrics, including Student-t intervals and the Nadeau-Bengio correction.

statistics

Statistics for summarising and comparing metrics across repeated training runs.

The functions here operate on plain sequences of per-seed scalars. They are deliberately separate from :mod:deepecgkit.evaluation.evaluator, whose bootstrap_evaluate resamples individual test samples and recomputes metrics from predictions; that estimator cannot consume a handful of finished run scores. The output key names mirror it so the two read alike.

summarize_scores

summarize_scores(
    values: Sequence[float],
    confidence_level: float = CONFIDENCE_LEVEL,
) -> dict[str, float]

Summarise one metric across seeds with a two-sided Student-t interval.

Non-finite entries are dropped first, so "n" counts usable values rather than attempted runs. The t distribution is used because with three to ten seeds the variance is estimated from the same points; a percentile bootstrap over so few scalars cannot place an endpoint outside their observed range and therefore understates the interval.

Parameters:

Name Type Description Default
values Sequence[float]

Per-seed values of a single metric.

required
confidence_level float

Two-sided coverage.

CONFIDENCE_LEVEL

Returns:

Type Description
dict[str, float]

Mapping with "mean", "std" (sample standard deviation, ddof=1),

dict[str, float]

"sem", "lower_ci", "upper_ci" and "n". With a single value the

dict[str, float]

dispersion and interval keys are NaN rather than zero. With no values

dict[str, float]

every key but "n" is NaN.

Raises:

Type Description
ValueError

If confidence_level is not strictly between 0 and 1.

Source code in deepecgkit/analysis/statistics.py
def summarize_scores(
    values: Sequence[float], confidence_level: float = CONFIDENCE_LEVEL
) -> dict[str, float]:
    """Summarise one metric across seeds with a two-sided Student-t interval.

    Non-finite entries are dropped first, so "n" counts usable values rather
    than attempted runs. The t distribution is used because with three to ten
    seeds the variance is estimated from the same points; a percentile
    bootstrap over so few scalars cannot place an endpoint outside their
    observed range and therefore understates the interval.

    Args:
        values: Per-seed values of a single metric.
        confidence_level: Two-sided coverage.

    Returns:
        Mapping with "mean", "std" (sample standard deviation, ddof=1),
        "sem", "lower_ci", "upper_ci" and "n". With a single value the
        dispersion and interval keys are NaN rather than zero. With no values
        every key but "n" is NaN.

    Raises:
        ValueError: If confidence_level is not strictly between 0 and 1.
    """
    _check_confidence_level(confidence_level)
    finite = _finite(values)
    n = int(finite.size)
    if n == 0:
        return {
            "mean": float("nan"),
            "std": float("nan"),
            "sem": float("nan"),
            "lower_ci": float("nan"),
            "upper_ci": float("nan"),
            "n": 0,
        }

    mean = float(np.mean(finite))
    if n == 1:
        return {
            "mean": mean,
            "std": float("nan"),
            "sem": float("nan"),
            "lower_ci": float("nan"),
            "upper_ci": float("nan"),
            "n": 1,
        }

    std = float(np.std(finite, ddof=1))
    sem = std / math.sqrt(n)
    half_width = float(stats.t.ppf(0.5 + confidence_level / 2.0, n - 1)) * sem
    return {
        "mean": mean,
        "std": std,
        "sem": sem,
        "lower_ci": mean - half_width,
        "upper_ci": mean + half_width,
        "n": n,
    }

nadeau_bengio_factor

nadeau_bengio_factor(
    n: int, n_train: int | None, n_test: int | None
) -> float

Variance inflation factor for repeated random subsampling.

When every seed redraws the train/test partition of one fixed corpus the per-seed differences share training data and are positively correlated, so the ordinary standard error understates the spread. Nadeau and Bengio (2003), Machine Learning 52(3), replace 1 / n with 1 / n + n_test / n_train.

Parameters:

Name Type Description Default
n int

Number of usable seeds.

required
n_train int | None

Training set size, or None when it was not recorded.

required
n_test int | None

Test set size, or None when it was not recorded.

required

Returns:

Type Description
float

The multiplier applied to the sample standard deviation to obtain the

float

corrected standard error. Falls back to the uncorrected

float

1 / sqrt(n) when the split sizes are unavailable.

Source code in deepecgkit/analysis/statistics.py
def nadeau_bengio_factor(n: int, n_train: int | None, n_test: int | None) -> float:
    """Variance inflation factor for repeated random subsampling.

    When every seed redraws the train/test partition of one fixed corpus the
    per-seed differences share training data and are positively correlated, so
    the ordinary standard error understates the spread. Nadeau and Bengio
    (2003), Machine Learning 52(3), replace ``1 / n`` with
    ``1 / n + n_test / n_train``.

    Args:
        n: Number of usable seeds.
        n_train: Training set size, or None when it was not recorded.
        n_test: Test set size, or None when it was not recorded.

    Returns:
        The multiplier applied to the sample standard deviation to obtain the
        corrected standard error. Falls back to the uncorrected
        ``1 / sqrt(n)`` when the split sizes are unavailable.
    """
    if n <= 0:
        return float("nan")
    base = 1.0 / n
    if not n_train or not n_test or n_train <= 0 or n_test <= 0:
        return math.sqrt(base)
    return math.sqrt(base + float(n_test) / float(n_train))

paired_summary

paired_summary(
    differences: Sequence[float],
    confidence_level: float = CONFIDENCE_LEVEL,
    correction: str = NADEAU_BENGIO,
    n_train: int | None = None,
    n_test: int | None = None,
) -> dict[str, float]

Summarise paired per-seed differences and test them against zero.

Parameters:

Name Type Description Default
differences Sequence[float]

One difference per shared seed.

required
confidence_level float

Two-sided coverage.

CONFIDENCE_LEVEL
correction str

Either "nadeau-bengio" or "none".

NADEAU_BENGIO
n_train int | None

Mean training set size across the contributing runs.

None
n_test int | None

Mean test set size across the contributing runs.

None

Returns:

Type Description
dict[str, float]

Mapping with "mean", "std", "sem", "lower_ci", "upper_ci", "t_stat",

dict[str, float]

"p_value", "n", "n_wins" and "excludes_zero". A degenerate sample with

dict[str, float]

zero variance yields NaN for the interval and the p-value rather than a

dict[str, float]

zero-width interval, because zero observed spread is not infinite

dict[str, float]

precision.

Raises:

Type Description
ValueError

If correction is not a recognised name.

Source code in deepecgkit/analysis/statistics.py
def paired_summary(
    differences: Sequence[float],
    confidence_level: float = CONFIDENCE_LEVEL,
    correction: str = NADEAU_BENGIO,
    n_train: int | None = None,
    n_test: int | None = None,
) -> dict[str, float]:
    """Summarise paired per-seed differences and test them against zero.

    Args:
        differences: One difference per shared seed.
        confidence_level: Two-sided coverage.
        correction: Either "nadeau-bengio" or "none".
        n_train: Mean training set size across the contributing runs.
        n_test: Mean test set size across the contributing runs.

    Returns:
        Mapping with "mean", "std", "sem", "lower_ci", "upper_ci", "t_stat",
        "p_value", "n", "n_wins" and "excludes_zero". A degenerate sample with
        zero variance yields NaN for the interval and the p-value rather than a
        zero-width interval, because zero observed spread is not infinite
        precision.

    Raises:
        ValueError: If correction is not a recognised name.
    """
    if correction not in (NO_CORRECTION, NADEAU_BENGIO):
        raise ValueError(f"Unknown correction: {correction}")
    _check_confidence_level(confidence_level)

    finite = _finite(differences)
    n = int(finite.size)
    result: dict[str, float] = {
        "mean": float("nan"),
        "std": float("nan"),
        "sem": float("nan"),
        "lower_ci": float("nan"),
        "upper_ci": float("nan"),
        "t_stat": float("nan"),
        "p_value": float("nan"),
        "n": n,
        "n_wins": 0,
        "excludes_zero": False,
    }
    if n == 0:
        return result

    result["mean"] = float(np.mean(finite))
    result["n_wins"] = int(np.sum(finite > 0))
    if n < 2:
        return result

    std = float(np.std(finite, ddof=1))
    result["std"] = std
    if std == 0.0:
        return result

    factor = (
        nadeau_bengio_factor(n, n_train, n_test)
        if correction == NADEAU_BENGIO
        else (1.0 / math.sqrt(n))
    )
    sem = std * factor
    result["sem"] = sem
    half_width = float(stats.t.ppf(0.5 + confidence_level / 2.0, n - 1)) * sem
    lower = result["mean"] - half_width
    upper = result["mean"] + half_width
    t_stat = result["mean"] / sem
    result["lower_ci"] = lower
    result["upper_ci"] = upper
    result["t_stat"] = t_stat
    result["p_value"] = float(2.0 * stats.t.sf(abs(t_stat), n - 1))
    result["excludes_zero"] = bool(lower > 0.0 or upper < 0.0)
    return result

kendall_tau

kendall_tau(
    ranks_a: Mapping[str, float],
    ranks_b: Mapping[str, float],
) -> float

Kendall's tau-b between two rankings over the same labels.

Parameters:

Name Type Description Default
ranks_a Mapping[str, float]

Label to rank for the first ranking.

required
ranks_b Mapping[str, float]

Label to rank for the second ranking.

required

Returns:

Type Description
float

The tau-b coefficient. No p-value is returned: with the four or five

float

architectures compared here the smallest attainable two-sided p-value

float

is 2 / len(labels)!, so no single coefficient can reach

float

conventional significance and reporting one would invite over-reading.

Raises:

Type Description
ValueError

If the label sets differ or fewer than three labels are given.

Source code in deepecgkit/analysis/statistics.py
def kendall_tau(ranks_a: Mapping[str, float], ranks_b: Mapping[str, float]) -> float:
    """Kendall's tau-b between two rankings over the same labels.

    Args:
        ranks_a: Label to rank for the first ranking.
        ranks_b: Label to rank for the second ranking.

    Returns:
        The tau-b coefficient. No p-value is returned: with the four or five
        architectures compared here the smallest attainable two-sided p-value
        is ``2 / len(labels)!``, so no single coefficient can reach
        conventional significance and reporting one would invite over-reading.

    Raises:
        ValueError: If the label sets differ or fewer than three labels are given.
    """
    labels = sorted(ranks_a)
    if labels != sorted(ranks_b):
        raise ValueError("Both rankings must cover the same labels")
    if len(labels) < 3:
        raise ValueError(f"Kendall's tau needs at least 3 labels, got {len(labels)}")

    first = [float(ranks_a[label]) for label in labels]
    second = [float(ranks_b[label]) for label in labels]
    tau, _ = stats.kendalltau(first, second, variant="b")
    return float(tau)

min_attainable_kendall_p

min_attainable_kendall_p(n_labels: int) -> float

Smallest two-sided p-value a tie-free Kendall's tau can attain.

Parameters:

Name Type Description Default
n_labels int

Number of ranked labels.

required

Returns:

Type Description
float

2 / n_labels! under the exact permutation null, capped at 1.0. At

float

four labels this is 0.083, so significance is unattainable however

float

perfectly the rankings agree.

Raises:

Type Description
ValueError

If fewer than three labels are given.

Source code in deepecgkit/analysis/statistics.py
def min_attainable_kendall_p(n_labels: int) -> float:
    """Smallest two-sided p-value a tie-free Kendall's tau can attain.

    Args:
        n_labels: Number of ranked labels.

    Returns:
        ``2 / n_labels!`` under the exact permutation null, capped at 1.0. At
        four labels this is 0.083, so significance is unattainable however
        perfectly the rankings agree.

    Raises:
        ValueError: If fewer than three labels are given.
    """
    if n_labels < 3:
        raise ValueError(f"Kendall's tau needs at least 3 labels, got {n_labels}")
    return min(1.0, 2.0 / float(math.factorial(n_labels)))

holm_adjust

holm_adjust(pvalues: Sequence[float]) -> list[float]

Holm step-down familywise adjustment.

Parameters:

Name Type Description Default
pvalues Sequence[float]

Raw p-values, possibly containing NaN.

required

Returns:

Type Description
list[float]

Adjusted p-values in the input order.

Source code in deepecgkit/analysis/statistics.py
def holm_adjust(pvalues: Sequence[float]) -> list[float]:
    """Holm step-down familywise adjustment.

    Args:
        pvalues: Raw p-values, possibly containing NaN.

    Returns:
        Adjusted p-values in the input order.
    """
    return _adjust(pvalues, scale=1.0, step_up=False)

benjamini_yekutieli_adjust

benjamini_yekutieli_adjust(
    pvalues: Sequence[float],
) -> list[float]

Benjamini-Yekutieli false discovery rate adjustment.

Yekutieli's correction is used rather than plain Benjamini-Hochberg because the comparisons here share training runs, models and datasets, so the positive dependence that Benjamini-Hochberg assumes is not established. The price is a factor of the harmonic number of the family size.

Parameters:

Name Type Description Default
pvalues Sequence[float]

Raw p-values, possibly containing NaN.

required

Returns:

Type Description
list[float]

Adjusted p-values in the input order.

Source code in deepecgkit/analysis/statistics.py
def benjamini_yekutieli_adjust(pvalues: Sequence[float]) -> list[float]:
    """Benjamini-Yekutieli false discovery rate adjustment.

    Yekutieli's correction is used rather than plain Benjamini-Hochberg
    because the comparisons here share training runs, models and datasets, so
    the positive dependence that Benjamini-Hochberg assumes is not
    established. The price is a factor of the harmonic number of the family
    size.

    Args:
        pvalues: Raw p-values, possibly containing NaN.

    Returns:
        Adjusted p-values in the input order.
    """
    raw = [value for value in pvalues if np.isfinite(value)]
    m = len(raw)
    if m == 0:
        return [float("nan") for _ in pvalues]
    harmonic = float(sum(1.0 / (i + 1) for i in range(m)))
    return _adjust(pvalues, scale=harmonic, step_up=True)

is_bimodal

is_bimodal(
    values: Sequence[float],
    baseline: float | None = None,
    min_gap: float = MIN_COLLAPSE_GAP,
) -> tuple[bool, str]

Detect a converge-or-collapse pattern across seeds.

A mean and interval are not a summary of a bimodal sample. A model that fails to train on a subset of seeds and is competitive on the rest is a more informative finding than a wide interval, so such cells are flagged rather than averaged.

Detection keys on the largest gap between consecutive sorted values rather than on a median absolute deviation. With five seeds the deviation is far too noisy a scale estimate, and a robust-outlier rule fires on ordinary scatter; a collapse separates the sample by a wide absolute margin.

Parameters:

Name Type Description Default
values Sequence[float]

Per-seed values of a single metric.

required
baseline float | None

Optional score below which a run counts as collapsed.

None
min_gap float

Smallest separation, in metric units, that counts as a split.

MIN_COLLAPSE_GAP

Returns:

Type Description
tuple[bool, str]

A pair of the flag and a short reason, empty when not flagged.

Source code in deepecgkit/analysis/statistics.py
def is_bimodal(
    values: Sequence[float],
    baseline: float | None = None,
    min_gap: float = MIN_COLLAPSE_GAP,
) -> tuple[bool, str]:
    """Detect a converge-or-collapse pattern across seeds.

    A mean and interval are not a summary of a bimodal sample. A model that
    fails to train on a subset of seeds and is competitive on the rest is a
    more informative finding than a wide interval, so such cells are flagged
    rather than averaged.

    Detection keys on the largest gap between consecutive sorted values rather
    than on a median absolute deviation. With five seeds the deviation is far
    too noisy a scale estimate, and a robust-outlier rule fires on ordinary
    scatter; a collapse separates the sample by a wide absolute margin.

    Args:
        values: Per-seed values of a single metric.
        baseline: Optional score below which a run counts as collapsed.
        min_gap: Smallest separation, in metric units, that counts as a split.

    Returns:
        A pair of the flag and a short reason, empty when not flagged.
    """
    finite = _finite(values)
    if finite.size < 3:
        return False, ""

    if baseline is not None and np.any(finite < baseline):
        collapsed = int(np.sum(finite < baseline))
        return True, f"{collapsed}/{finite.size} seeds scored below the {baseline:.3f} baseline"

    ordered = np.sort(finite)
    gaps = np.diff(ordered)
    if gaps.size == 0:
        return False, ""
    index = int(np.argmax(gaps))
    largest = float(gaps[index])
    if largest >= min_gap:
        low = index + 1
        high = int(finite.size) - low
        return True, (f"{low} of {finite.size} seeds sit {largest:.3f} below the other {high}")
    return False, ""

LaTeX

booktabs table emitters for the seed-variance analysis.

latex

booktabs table emitters for the seed-variance analysis.

Every literal containing a backslash is a raw string. Written as an ordinary string, "\toprule" is a tab followed by "oprule" and "\bottomrule" is a backspace followed by "ottomrule"; both are valid Python escapes, so neither the configured ruff rules nor W605 reports them and the emitted table silently fails to compile.

The tables need only the booktabs package, which the manuscript already loads.

escape_latex

escape_latex(text: str) -> str

Escape the characters LaTeX treats specially.

Parameters:

Name Type Description Default
text str

Raw text.

required

Returns:

Type Description
str

Text safe to place in a table cell.

Source code in deepecgkit/analysis/latex.py
def escape_latex(text: str) -> str:
    """Escape the characters LaTeX treats specially.

    Args:
        text: Raw text.

    Returns:
        Text safe to place in a table cell.
    """
    escaped = str(text)
    for character, replacement in _ESCAPES:
        escaped = escaped.replace(character, replacement)
    return escaped

display_model

display_model(name: str) -> str

Render a registry model name for the manuscript.

Parameters:

Name Type Description Default
name str

Registry name such as "se-resnet".

required

Returns:

Type Description
str

The display name, escaped when unrecognised.

Source code in deepecgkit/analysis/latex.py
def display_model(name: str) -> str:
    """Render a registry model name for the manuscript.

    Args:
        name: Registry name such as "se-resnet".

    Returns:
        The display name, escaped when unrecognised.
    """
    return MODEL_DISPLAY_NAMES.get(name, escape_latex(name))

display_dataset

display_dataset(name: str) -> str

Render a registry dataset name for the manuscript.

Parameters:

Name Type Description Default
name str

Registry name such as "af-classification".

required

Returns:

Type Description
str

The display name, escaped when unrecognised.

Source code in deepecgkit/analysis/latex.py
def display_dataset(name: str) -> str:
    """Render a registry dataset name for the manuscript.

    Args:
        name: Registry name such as "af-classification".

    Returns:
        The display name, escaped when unrecognised.
    """
    return DATASET_DISPLAY_NAMES.get(name, escape_latex(name))

display_metric

display_metric(name: str) -> str

Render a metric column name for the manuscript.

Parameters:

Name Type Description Default
name str

Metric key such as "f1".

required

Returns:

Type Description
str

The display name, escaped when unrecognised.

Source code in deepecgkit/analysis/latex.py
def display_metric(name: str) -> str:
    """Render a metric column name for the manuscript.

    Args:
        name: Metric key such as "f1".

    Returns:
        The display name, escaped when unrecognised.
    """
    return METRIC_DISPLAY_NAMES.get(name, escape_latex(name))

format_cell

format_cell(
    mean: float,
    lower: float,
    upper: float,
    n: int = 0,
    precision: int = DEFAULT_PRECISION,
) -> str

Render a mean with its interval beneath, in the paper's style.

Parameters:

Name Type Description Default
mean float

Point estimate.

required
lower float

Interval lower bound.

required
upper float

Interval upper bound.

required
n int

Number of contributing seeds.

0
precision int

Decimal places.

DEFAULT_PRECISION

Returns:

Type Description
str

A table cell. A single seed renders bare with no interval, because one

str

run carries no dispersion; a missing value renders as a dash.

Source code in deepecgkit/analysis/latex.py
def format_cell(
    mean: float,
    lower: float,
    upper: float,
    n: int = 0,
    precision: int = DEFAULT_PRECISION,
) -> str:
    """Render a mean with its interval beneath, in the paper's style.

    Args:
        mean: Point estimate.
        lower: Interval lower bound.
        upper: Interval upper bound.
        n: Number of contributing seeds.
        precision: Decimal places.

    Returns:
        A table cell. A single seed renders bare with no interval, because one
        run carries no dispersion; a missing value renders as a dash.
    """
    if mean is None or not np.isfinite(mean):
        return MISSING_CELL
    point = f"${mean:.{precision}f}$"
    if n == 1 or not np.isfinite(lower) or not np.isfinite(upper):
        return point
    return f"{point} {{\\scriptsize [{lower:.{precision}f}, {upper:.{precision}f}]}}"

results_table

results_table(
    aggregated: DataFrame,
    dataset: str,
    metrics: Sequence[str] = ("f1", "precision", "recall"),
    label: str | None = None,
    precision: int = DEFAULT_PRECISION,
    confidence_level: float = 0.95,
    split_description: str = SPLIT_DESCRIPTION,
) -> str

Per-dataset mean and interval table, replacing a single-run table.

Parameters:

Name Type Description Default
aggregated DataFrame

Frame from :func:deepecgkit.analysis.aggregation.aggregate_runs.

required
dataset str

Dataset to render.

required
metrics Sequence[str]

Metric columns in display order.

('f1', 'precision', 'recall')
label str | None

Cross-reference label.

None
precision int

Decimal places.

DEFAULT_PRECISION

Returns:

Type Description
str

The table as a string. Nothing is bolded: marking the best mean and

str

everything whose interval overlaps it is a pairwise comparison the

str

per-model intervals cannot support. Rows are ordered by the first

str

metric, ascending when that metric is one where lower is better, so a

str

calibration table leads with the best-calibrated model rather than the

str

worst.

Source code in deepecgkit/analysis/latex.py
def results_table(
    aggregated: pd.DataFrame,
    dataset: str,
    metrics: Sequence[str] = ("f1", "precision", "recall"),
    label: str | None = None,
    precision: int = DEFAULT_PRECISION,
    confidence_level: float = 0.95,
    split_description: str = SPLIT_DESCRIPTION,
) -> str:
    """Per-dataset mean and interval table, replacing a single-run table.

    Args:
        aggregated: Frame from :func:`deepecgkit.analysis.aggregation.aggregate_runs`.
        dataset: Dataset to render.
        metrics: Metric columns in display order.
        label: Cross-reference label.
        precision: Decimal places.

    Returns:
        The table as a string. Nothing is bolded: marking the best mean and
        everything whose interval overlaps it is a pairwise comparison the
        per-model intervals cannot support. Rows are ordered by the first
        metric, ascending when that metric is one where lower is better, so a
        calibration table leads with the best-calibrated model rather than the
        worst.
    """
    scope = aggregated[aggregated["dataset"] == dataset]
    label = label or f"tab:seed_{dataset.replace('-', '_')}"
    ascending = metrics[0] in LOWER_IS_BETTER
    ranking = (
        scope[scope["metric"] == metrics[0]]
        .sort_values("mean", ascending=ascending)["model"]
        .tolist()
    )

    header = " & ".join(["Model"] + [display_metric(m) for m in metrics] + ["$n$"])
    body = [header + r" \\", r"\midrule"]
    averagings = set()
    for model in ranking:
        cells = [display_model(model)]
        seeds = 0
        for metric in metrics:
            row = scope[(scope["model"] == model) & (scope["metric"] == metric)]
            if row.empty:
                cells.append(MISSING_CELL)
                continue
            entry = row.iloc[0]
            seeds = int(entry["n"])
            averagings.add(str(entry["averaging"]))
            cells.append(
                format_cell(entry["mean"], entry["lower_ci"], entry["upper_ci"], seeds, precision)
            )
        cells.append(str(seeds))
        body.append(" & ".join(cells) + r" \\")

    notes = [
        "Intervals describe how much each model moves when the split and the initialisation",
        "are redrawn; they are not a pairwise test.",
    ]
    if averagings:
        notes.append(f"Averaging: {', '.join(sorted(escape_latex(a) for a in averagings))}.")
    body.append(r"\midrule")
    body.extend(_note_rows(notes, len(metrics) + 2))

    caption = (
        f"{display_dataset(dataset)} results, mean over $n$ random seeds. Each seed redraws the "
        f"{split_description} and reinitialises the weights, so brackets give two-sided "
        f"{_percent(confidence_level)} Student-$t$ intervals over that combined variation. The "
        "interval is a within-corpus statement and is not a generalisation interval."
    )
    return _wrap_table(body, "l" + "c" * (len(metrics) + 1), caption, label)

rank_distribution_table

rank_distribution_table(
    distribution: DataFrame,
    taus: DataFrame | None = None,
    label: str = "tab:rank_distribution",
    metric: str = "f1",
) -> str

How often each model attained each rank, per dataset.

Parameters:

Name Type Description Default
distribution DataFrame

Frame from :func:deepecgkit.analysis.ranking.rank_distribution.

required
taus DataFrame | None

Optional frame from :func:deepecgkit.analysis.ranking.tau_summary.

None
label str

Cross-reference label.

'tab:rank_distribution'
metric str

Metric that was ranked, for the caption.

'f1'

Returns:

Type Description
str

The table as a string.

Source code in deepecgkit/analysis/latex.py
def rank_distribution_table(
    distribution: pd.DataFrame,
    taus: pd.DataFrame | None = None,
    label: str = "tab:rank_distribution",
    metric: str = "f1",
) -> str:
    """How often each model attained each rank, per dataset.

    Args:
        distribution: Frame from :func:`deepecgkit.analysis.ranking.rank_distribution`.
        taus: Optional frame from :func:`deepecgkit.analysis.ranking.tau_summary`.
        label: Cross-reference label.
        metric: Metric that was ranked, for the caption.

    Returns:
        The table as a string.
    """
    max_models = int(distribution["n_models"].max()) if not distribution.empty else 0
    n_columns = max_models + 4
    header = " & ".join(
        ["Model"]
        + [f"R{position}" for position in range(1, max_models + 1)]
        + ["Mode", "Top-1", f"{display_metric(metric)} (mean $\\pm$ sd)"]
    )
    body = [header + r" \\"]

    for dataset in sorted(distribution["dataset"].unique()):
        scope = distribution[distribution["dataset"] == dataset].sort_values(
            "mean_value", ascending=False
        )
        n_seeds = int(scope["n_seeds"].max())
        n_models = int(scope["n_models"].max())
        body.append(r"\midrule")
        body.append(
            rf"\multicolumn{{{n_columns}}}{{l}}{{\textit{{{display_dataset(dataset)}}} "
            rf"({n_models} models, {n_seeds} seeds)}} \\"
        )
        for _, row in scope.iterrows():
            cells = [display_model(str(row["model"]))]
            for position in range(1, max_models + 1):
                column = f"rank_{position}"
                if column in scope.columns and np.isfinite(row.get(column, np.nan)):
                    cells.append(str(int(row[column])))
                else:
                    cells.append(MISSING_CELL)
            mode_count = round(float(row["mode_fraction"]) * row["n_seeds"])
            cells.append(f"{int(row['mode_rank'])} ({mode_count}/{int(row['n_seeds'])})")
            cells.append(f"{float(row['top1_fraction']):.2f}")
            std = row["std_value"]
            spread = f" $\\pm$ {float(std):.3f}" if np.isfinite(std) else ""
            cells.append(f"${float(row['mean_value']):.3f}$" + spread)
            body.append(" & ".join(cells) + r" \\")

    notes = ["Ties share the better rank, so a rank column may over-fill."]
    if taus is not None and not taus.empty:
        rendered = []
        for _, row in taus.iterrows():
            if not np.isfinite(row["tau_mean"]):
                continue
            rendered.append(
                f"{display_dataset(str(row['dataset']))} ${float(row['tau_mean']):.2f}$ "
                f"$[{float(row['tau_min']):.2f}, {float(row['tau_max']):.2f}]$"
            )
        if rendered:
            joined = "; ".join(rendered)
            notes.append(r"Kendall $\bar\tau_b$ over all seed-pair rankings: " + joined + ".")
            notes.extend(_tau_floor_notes(taus))
    body.append(r"\midrule")
    body.extend(_note_rows(notes, n_columns))

    caption = (
        f"Distribution of {display_metric(metric)} ranks over random seeds. Each cell counts the "
        "seeds in which the model attained that rank within its dataset (rank 1 = best); rows sum "
        "to the seed count. Dashes mark ranks that cannot occur because fewer models were trained "
        "on that dataset."
    )
    return _wrap_table(body, "l" + "c" * (n_columns - 1), caption, label, size=r"\scriptsize")

flip_survival_table

flip_survival_table(
    flips: DataFrame,
    label: str = "tab:flip_survival",
    metric: str = "f1",
    exclusions: Sequence[str] | None = None,
    confidence_level: float = 0.95,
    correction: str = "nadeau-bengio",
    split_description: str = SPLIT_DESCRIPTION,
) -> str

The headline table: which cross-task reversals survive seed variance.

Parameters:

Name Type Description Default
flips DataFrame

Frame from :func:deepecgkit.analysis.ranking.all_flip_tests.

required
label str

Cross-reference label.

'tab:flip_survival'
metric str

Metric that was compared, for the caption.

'f1'
exclusions Sequence[str] | None

Model or dataset restrictions to echo into the notes.

None

Returns:

Type Description
str

The table as a string.

Source code in deepecgkit/analysis/latex.py
def flip_survival_table(
    flips: pd.DataFrame,
    label: str = "tab:flip_survival",
    metric: str = "f1",
    exclusions: Sequence[str] | None = None,
    confidence_level: float = 0.95,
    correction: str = "nadeau-bengio",
    split_description: str = SPLIT_DESCRIPTION,
) -> str:
    """The headline table: which cross-task reversals survive seed variance.

    Args:
        flips: Frame from :func:`deepecgkit.analysis.ranking.all_flip_tests`.
        label: Cross-reference label.
        metric: Metric that was compared, for the caption.
        exclusions: Model or dataset restrictions to echo into the notes.

    Returns:
        The table as a string.
    """
    n_columns = 6
    header = " & ".join(
        [
            "$A$ vs.\\ $B$",
            r"$\Delta_1$ [95\% CI]",
            "$k/n$",
            r"$\Delta_2$ [95\% CI]",
            r"$p_{\mathrm{BY}}$",
            "Verdict",
        ]
    )
    body = [header + r" \\"]
    mixed_averaging = False

    for (dataset_1, dataset_2), scope in flips.groupby(["dataset_1", "dataset_2"], sort=True):
        body.append(r"\midrule")
        body.append(
            rf"\multicolumn{{{n_columns}}}{{l}}{{\textit{{{display_dataset(str(dataset_1))} "
            rf"$\rightarrow$ {display_dataset(str(dataset_2))}}}}} \\"
        )
        for _, row in scope.iterrows():
            comparable = bool(row.get("comparable_magnitudes", True))
            marker = "" if comparable else r"^{\dagger}"
            if not comparable:
                mixed_averaging = True
            first = _interval_cell(row["mean_1"], row["lower_1"], row["upper_1"], marker)
            second = _interval_cell(row["mean_2"], row["lower_2"], row["upper_2"], marker)
            wins = (
                f"{int(row['wins_1'])}/{int(row['n_1'])}"
                if np.isfinite(row["n_1"]) and int(row["n_1"]) > 0
                else MISSING_CELL
            )
            adjusted = row.get("p_adjusted", float("nan"))
            pair = f"{display_model(str(row['model_a']))} / {display_model(str(row['model_b']))}"
            verdict = str(row["verdict"])
            body.append(
                " & ".join(
                    [
                        pair,
                        first,
                        wins,
                        second,
                        _number(adjusted, 3) if np.isfinite(adjusted) else MISSING_CELL,
                        VERDICT_DISPLAY_NAMES.get(verdict, escape_latex(verdict)),
                    ]
                )
                + r" \\"
            )

    notes = [
        "Supported: both intervals exclude zero with opposite signs. Unresolved: the mean",
        "ordering reverses but an interval covers zero. Not reproduced: the ordering does not",
        "reverse. Bimodal: a model collapses on a subset of seeds, so a mean is not a summary.",
        "Untestable: too few independent test units or seeds to support an interval.",
    ]
    if mixed_averaging:
        notes.append(
            r"$^{\dagger}$ the two $\Delta$ are differences of differently averaged scores and are"
        )
        notes.append("not comparable in magnitude; only their signs enter the test.")
    if exclusions:
        notes.append("Restricted to: " + escape_latex(", ".join(exclusions)) + ".")
    body.append(r"\midrule")
    body.extend(_note_rows(notes, n_columns))

    if correction == "nadeau-bengio":
        se_text = (
            "whose standard error carries the Nadeau--Bengio correction for the training-set "
            "overlap between seeds"
        )
    else:
        se_text = "whose standard error is the uncorrected standard error of the mean"
    family = str(flips["fdr_family"].iloc[0]) if "fdr_family" in flips.columns else "testable"
    family_text = (
        "every testable comparison"
        if family == "testable"
        else "the comparisons whose verdict is supported, suggestive or contradicted"
    )
    caption = (
        "Cross-task ranking reversals under seed variance. Each row compares two architectures "
        f"on two datasets over $n$ shared random seeds; every seed redraws the {split_description} "
        "and reinitialises the weights. $\\Delta_i$ is the mean paired per-seed difference in "
        f"{display_metric(metric)} on dataset $D_i$, with a two-sided {_percent(confidence_level)} "
        f"Student-$t$ interval {se_text}. $k/n$ counts the seeds in which $A$ led on $D_1$. A "
        "reversal is supported only when both intervals exclude zero with opposite signs (an "
        "intersection--union test); $p$ is the larger component $p$ and is an upper bound, and "
        f"$p_{{\\mathrm{{BY}}}}$ its Benjamini--Yekutieli adjustment over {family_text}. "
        "Verdicts: supported, both intervals exclude zero with opposite signs over at least five "
        "seeds; suggestive, the same with fewer seeds; unresolved, opposite signs but an interval "
        "covers zero; not reproduced, the same sign; contradicted, the same sign with both "
        "intervals excluding zero; untestable, too few test groups, seeds or a bimodal score. "
        "All comparisons in this grid are exploratory."
    )
    return _wrap_table(body, "lccccl", caption, label)

tau_table

tau_table(
    taus: DataFrame, label: str = "tab:kendall_tau"
) -> str

Supplementary Kendall's tau table.

Parameters:

Name Type Description Default
taus DataFrame

Frame from :func:deepecgkit.analysis.ranking.tau_summary.

required
label str

Cross-reference label.

'tab:kendall_tau'

Returns:

Type Description
str

The table as a string.

Source code in deepecgkit/analysis/latex.py
def tau_table(taus: pd.DataFrame, label: str = "tab:kendall_tau") -> str:
    """Supplementary Kendall's tau table.

    Args:
        taus: Frame from :func:`deepecgkit.analysis.ranking.tau_summary`.
        label: Cross-reference label.

    Returns:
        The table as a string.
    """
    n_columns = 6
    header = " & ".join(
        ["Dataset", "$|M|$", "Seeds", "Pairs", r"$\bar\tau_b$ [min, max]", "$p_{\\min}$"]
    )
    body = [header + r" \\", r"\midrule"]
    for _, row in taus.iterrows():
        if np.isfinite(row["tau_mean"]):
            summary = (
                f"${float(row['tau_mean']):.2f}$ "
                f"$[{float(row['tau_min']):.2f}, {float(row['tau_max']):.2f}]$"
            )
        else:
            summary = MISSING_CELL
        body.append(
            " & ".join(
                [
                    display_dataset(str(row["dataset"])),
                    str(int(row["n_models"])),
                    str(int(row["n_seeds"])),
                    str(int(row["n_pairs"])),
                    summary,
                    _number(row["min_attainable_p"], 3),
                ]
            )
            + r" \\"
        )
    body.append(r"\midrule")
    body.extend(
        _note_rows(
            [
                r"$p_{\min}=2/|M|!$ is the smallest two-sided $p$ the exact permutation null can",
                "attain. No p-values are reported; the coefficients are descriptive only.",
                *_tau_floor_notes(taus),
            ],
            n_columns,
        )
    )
    caption = (
        r"Kendall $\tau_b$ between per-seed rankings, over every pair of seeds within each "
        "dataset. The coefficients are mutually dependent, so no interval or pooled test is given."
    )
    return _wrap_table(body, "lccccc", caption, label)

transfer_matrix_table

transfer_matrix_table(
    matrix: DataFrame,
    metric: str = "auroc_macro",
    baseline_row: str | None = None,
    label: str = "tab:transfer_matrix",
    precision: int = DEFAULT_PRECISION,
) -> str

Cross-dataset linear-probe transfer matrix.

Parameters:

Name Type Description Default
matrix DataFrame

Frame from :meth:deepecgkit.evaluation.TransferMatrix.to_frame, indexed by encoder source with one column per probe target.

required
metric str

Metric filling the cells, named in the caption.

'auroc_macro'
baseline_row str | None

Index entry holding the untrained-encoder floor, moved to the top and ruled off from the trained encoders.

None
label str

Cross-reference label.

'tab:transfer_matrix'
precision int

Decimal places.

DEFAULT_PRECISION

Returns:

Type Description
str

The table as a string. Nothing is bolded: a single probe score carries

str

no interval, so marking a best cell would assert a comparison the

str

numbers cannot support.

Source code in deepecgkit/analysis/latex.py
def transfer_matrix_table(
    matrix: pd.DataFrame,
    metric: str = "auroc_macro",
    baseline_row: str | None = None,
    label: str = "tab:transfer_matrix",
    precision: int = DEFAULT_PRECISION,
) -> str:
    """Cross-dataset linear-probe transfer matrix.

    Args:
        matrix: Frame from :meth:`deepecgkit.evaluation.TransferMatrix.to_frame`,
            indexed by encoder source with one column per probe target.
        metric: Metric filling the cells, named in the caption.
        baseline_row: Index entry holding the untrained-encoder floor, moved to
            the top and ruled off from the trained encoders.
        label: Cross-reference label.
        precision: Decimal places.

    Returns:
        The table as a string. Nothing is bolded: a single probe score carries
        no interval, so marking a best cell would assert a comparison the
        numbers cannot support.
    """
    targets = list(matrix.columns)
    header = " & ".join(["Encoder source"] + [display_dataset(t) for t in targets])
    body = [header + r" \\", r"\midrule"]

    ordered = list(matrix.index)
    if baseline_row is not None and baseline_row in ordered:
        ordered = [baseline_row] + [row for row in ordered if row != baseline_row]

    for source in ordered:
        name = BASELINE_DISPLAY_NAMES.get(source) or display_dataset(source)
        cells = [name] + [_number(matrix.loc[source, t], precision) for t in targets]
        body.append(" & ".join(cells) + r" \\")
        if baseline_row is not None and source == baseline_row:
            body.append(r"\midrule")

    notes = [
        "Each cell freezes the encoder trained on its row's corpus, embeds the column corpus,",
        "and fits a standardised logistic probe on that corpus's training split alone.",
    ]
    if baseline_row is not None and baseline_row in matrix.index:
        notes.append("The first row probes an untrained encoder and is the floor for every column.")
    body.append(r"\midrule")
    body.extend(_note_rows(notes, len(targets) + 1))

    caption = (
        f"Cross-dataset transfer, {display_metric(metric)} of a linear probe on a held-out "
        "split. Rows are the corpus each frozen encoder was trained on, columns the corpus it "
        "was probed on; the diagonal is the in-domain probe. Every corpus is run single-lead at "
        "a common sampling rate so that each encoder sees an identical input, and the probe head "
        "is refitted per column because the label spaces differ."
    )
    return _wrap_table(body, "l" + "c" * len(targets), caption, label)

write_tables

write_tables(
    tables: dict[str, str], output_dir: Any
) -> list[str]

Write rendered tables to .tex files.

Parameters:

Name Type Description Default
tables dict[str, str]

Mapping from stem to rendered table.

required
output_dir Any

Destination directory, created if absent.

required

Returns:

Type Description
list[str]

The written paths as strings.

Source code in deepecgkit/analysis/latex.py
def write_tables(tables: dict[str, str], output_dir: Any) -> list[str]:
    """Write rendered tables to .tex files.

    Args:
        tables: Mapping from stem to rendered table.
        output_dir: Destination directory, created if absent.

    Returns:
        The written paths as strings.
    """
    directory = Path(output_dir)
    directory.mkdir(parents=True, exist_ok=True)
    written: list[str] = []
    for stem, content in sorted(tables.items()):
        path = directory / f"{stem}.tex"
        path.write_text(content)
        written.append(str(path))
    return written