oversampleqa.advanced_benchmark

oversampleqa.advanced_benchmark

Advanced benchmarking utilities with statistical analysis.

BenchmarkResult dataclass

Structured benchmark summary for a single (dataset, oversampler, metric).

Source code in src/oversampleqa/advanced_benchmark.py

@dataclass
class BenchmarkResult:
    """Structured benchmark summary for a single (dataset, oversampler, metric)."""

    dataset_name: str
    oversampler_name: str
    metric: str
    hidden_ratio: float
    reference: str
    minority_label: int
    random_state: int | None
    n_folds: int
    n_repeats: int
    oversampleqa_version: str
    error_rates: list[float]
    mean_error: float
    std_error: float
    confidence_interval: tuple[float, float]
    effect_size: float | None = None
    p_value: float | None = None

FoldRecord dataclass

One evaluated fold, including the ones that produced no measurement.

The summary frame reports a mean and an interval per (dataset, oversampler, metric), which cannot be re-aggregated, plotted as a distribution, or given a different interval. It also cannot answer the question that decides whether a mean is trustworthy: how many folds actually contributed, and why the others did not.

Skipped folds are kept as rows with error_rate of nan and a stated reason rather than dropped. A mean over three surviving folds out of twenty-five looks identical to a mean over twenty-five once the skips are gone.

Source code in src/oversampleqa/advanced_benchmark.py

@dataclass
class FoldRecord:
    """One evaluated fold, including the ones that produced no measurement.

    The summary frame reports a mean and an interval per
    (dataset, oversampler, metric), which cannot be re-aggregated, plotted as a
    distribution, or given a different interval. It also cannot answer the
    question that decides whether a mean is trustworthy: how many folds
    actually contributed, and why the others did not.

    Skipped folds are kept as rows with ``error_rate`` of ``nan`` and a stated
    reason rather than dropped. A mean over three surviving folds out of
    twenty-five looks identical to a mean over twenty-five once the skips are
    gone.
    """

    dataset_name: str
    oversampler_name: str
    metric: str
    repeat: int
    fold: int
    split_seed: int | None
    hidden_ratio: float
    reference: str
    minority_label: int
    random_state: int | None
    n_folds: int
    n_repeats: int
    oversampleqa_version: str
    error_rate: float
    skipped: bool
    skip_reason: str

StatisticalBenchmark

Advanced benchmarking engine with statistical analysis.

Source code in src/oversampleqa/advanced_benchmark.py

class StatisticalBenchmark:
    """Advanced benchmarking engine with statistical analysis."""

    def __init__(
        self,
        n_folds: int = 5,
        n_repeats: int = 5,
        confidence_level: float = 0.95,
        correction_method: str = "holm",
        random_state: int | None = 42,
    ) -> None:
        if n_folds < 2:
            raise ValueError("n_folds must be at least 2")
        self.n_folds = n_folds
        self.n_repeats = n_repeats
        self.confidence_level = confidence_level
        self.correction_method = correction_method
        self.random_state = random_state
        self._skipped: list[str] = []
        self._fold_records_all: list[FoldRecord] = []

    def run_comprehensive_benchmark(
        self,
        datasets: Sequence[dict[str, Any]],
        oversamplers: Sequence[Any],
        metrics: Sequence[str] | None = None,
    ) -> pd.DataFrame:
        """Run repeated stratified benchmarking across datasets.

        Parameters
        ----------
        datasets:
            Sequence of dataset descriptors. Each entry should provide ``data``,
            ``target`` and optionally ``name`` and ``minority_label``.
        oversamplers:
            Sequence of initialised oversampler instances (will be cloned).
        metrics:
            Distance metrics to evaluate. Defaults to Hassanat, Euclidean, Mahalanobis.
        """

        # mahalanobis is not a default. Without a covariance inverse it is
        # Euclidean, so it produced a third set of rows identical to the
        # euclidean ones -- inflating euclidean's weight in the rankings and
        # splitting one comparison into two for the p-value correction. Pass it
        # explicitly, with cov_inv in metric_kwargs, if you want it.
        metrics = tuple(metrics or ("hassanat", "euclidean"))
        # Reset per run: a reused engine must not accumulate skips or folds.
        self._skipped = []
        self._fold_records_all = []

        all_results: list[BenchmarkResult] = []
        for dataset in datasets:
            all_results.extend(
                self._benchmark_single_dataset(dataset, oversamplers, metrics)
            )

        frame = pd.DataFrame([self._result_to_dict(r) for r in all_results])

        if self._skipped:
            warnings.warn(
                f"{len(self._skipped)} of "
                f"{len(datasets) * len(oversamplers) * len(metrics)} "
                "dataset/oversampler/metric combinations produced no usable "
                "folds and are absent from the results: "
                + ", ".join(self._skipped[:5])
                + (" ..." if len(self._skipped) > 5 else "")
                + ". The most common cause is a minority class too small to "
                "hold out from once it has been split into folds -- try fewer "
                "folds or a larger hidden_ratio.",
                UserWarning,
                stacklevel=2,
            )

        if frame.empty:
            # Return the expected columns rather than a (0, 0) frame. An empty
            # frame with no columns raises KeyError on any column access, so a
            # caller that handles "no results" still breaks.
            return pd.DataFrame(columns=list(_RESULT_COLUMNS))

        frame = self._add_statistical_analysis(frame)
        return frame

    def _benchmark_single_dataset(
        self,
        dataset: dict[str, Any],
        oversamplers: Sequence[Any],
        metrics: Sequence[str],
    ) -> list[BenchmarkResult]:
        """Run benchmark for a single dataset across oversamplers and metrics.

        Args:
            dataset: Dataset descriptor with ``data`` and ``target`` arrays.
            oversamplers: Oversamplers to evaluate.
            metrics: Distance metrics to test.

        Returns:
            List of BenchmarkResult entries for the dataset.
        """
        X, y = dataset["data"], dataset["target"]
        dataset_name = dataset.get("name", "dataset")
        minority_label = dataset.get("minority_label", 1)

        # basic preprocessing to avoid scaling issues for distance metrics
        scaler = StandardScaler()
        X_scaled = scaler.fit_transform(X)

        results: list[BenchmarkResult] = []
        for oversampler in oversamplers:
            for metric in metrics:
                fold_records = self._fold_records(
                    X_scaled,
                    y,
                    minority_label=minority_label,
                    oversampler=oversampler,
                    metric=metric,
                    dataset_name=dataset_name,
                )
                self._fold_records_all.extend(fold_records)
                error_rates = [
                    record.error_rate for record in fold_records if not record.skipped
                ]
                if not error_rates:
                    # Every fold failed for this combination. The per-fold
                    # warnings above explain why, but on a real sweep there are
                    # hundreds of them; record the combination so the caller
                    # gets one summary rather than a silently missing row.
                    self._skipped.append(
                        f"{dataset_name} / {oversampler.__class__.__name__} / {metric}"
                    )
                    continue
                mean_error = float(np.mean(error_rates))
                std_error = (
                    float(np.std(error_rates, ddof=1)) if len(error_rates) > 1 else 0.0
                )
                ci_lower, ci_upper = self._confidence_interval(error_rates)
                results.append(
                    BenchmarkResult(
                        dataset_name=dataset_name,
                        oversampler_name=oversampler.__class__.__name__,
                        metric=metric,
                        hidden_ratio=_FOLD_HIDDEN_RATIO,
                        reference="hidden_minority",
                        minority_label=minority_label,
                        random_state=self.random_state,
                        n_folds=self.n_folds,
                        n_repeats=self.n_repeats,
                        oversampleqa_version=_PACKAGE_VERSION,
                        error_rates=error_rates,
                        mean_error=mean_error,
                        std_error=std_error,
                        confidence_interval=(ci_lower, ci_upper),
                    )
                )
        return results

    def fold_results(self) -> pd.DataFrame:
        """Return one row per attempted fold from the most recent run.

        The summary frame reports a mean and interval per
        (dataset, oversampler, metric). That is enough to read a ranking and
        not enough to check one: it cannot be re-aggregated, plotted as a
        distribution, or given a different interval, and it does not say how
        many folds actually contributed.

        This frame answers those. Skipped folds are present with ``error_rate``
        of ``nan``, ``skipped`` true and a stated ``skip_reason``, because a
        mean over three surviving folds of twenty-five is indistinguishable
        from a mean over twenty-five once the skips are dropped.

        ``split_seed`` is the seed given to the fold splitter for that repeat,
        so a single repeat can be reproduced without rerunning the sweep.

        Returns:
            A long-format frame with :data:`_FOLD_COLUMNS`. Empty of rows but
            not of columns when no run has happened yet, so column access works
            either way.
        """
        frame = pd.DataFrame([asdict(record) for record in self._fold_records_all])
        # Built from the records first, then aliased: FoldRecord stays the one
        # place a fold's identity is defined, rather than carrying each name
        # twice and letting the two drift.
        if frame.empty:
            frame = pd.DataFrame(columns=list(_FOLD_COLUMNS))
        else:
            frame["dataset"] = frame["dataset_name"]
            frame["oversampler"] = frame["oversampler_name"]
        return frame.reindex(columns=list(_FOLD_COLUMNS))

    def _fold_records(
        self,
        X: np.ndarray,
        y: np.ndarray,
        minority_label: int,
        oversampler: Any,
        metric: str,
        dataset_name: str,
    ) -> list[FoldRecord]:
        """Evaluate every repeat/fold, recording skips as well as measurements.

        Args:
            X: Feature matrix.
            y: Target labels.
            minority_label: Minority class label.
            oversampler: Oversampler instance to clone per fold.
            metric: Distance metric name.
            dataset_name: Name recorded on each row.

        Returns:
            One :class:`FoldRecord` per fold attempted, including skipped ones.
        """
        records: list[FoldRecord] = []
        rng = np.random.default_rng(self.random_state)
        name = oversampler.__class__.__name__

        def record(
            repeat: int, fold: int, seed: int | None, error: float, reason: str
        ) -> None:
            records.append(
                FoldRecord(
                    dataset_name=dataset_name,
                    oversampler_name=name,
                    metric=metric,
                    repeat=repeat,
                    fold=fold,
                    split_seed=seed,
                    hidden_ratio=_FOLD_HIDDEN_RATIO,
                    reference="hidden_minority",
                    minority_label=minority_label,
                    random_state=self.random_state,
                    n_folds=self.n_folds,
                    n_repeats=self.n_repeats,
                    oversampleqa_version=_PACKAGE_VERSION,
                    error_rate=error,
                    skipped=bool(reason),
                    skip_reason=reason,
                )
            )

        for repeat in range(self.n_repeats):
            split_seed = (
                None if self.random_state is None else int(rng.integers(0, 1_000_000))
            )
            cv = StratifiedKFold(
                n_splits=self.n_folds,
                shuffle=True,
                random_state=split_seed,
            )
            for fold, (train_idx, _val_idx) in enumerate(cv.split(X, y)):
                # Only the training fold is used. validate_oversampling performs
                # its own hold-out internally, so the CV validation fold has no
                # role here -- the splitter is effectively a stratified
                # subsampler, and each "fold" is one subsample of the data
                # rather than a held-out evaluation. See docs/benchmarking.rst
                # for what the resulting intervals therefore describe.
                X_train = X[train_idx]
                y_train = y[train_idx]
                if len(np.unique(y_train)) < 2:
                    warnings.warn(
                        "Training fold lacks class diversity; skipping fold.",
                        stacklevel=2,
                    )
                    record(
                        repeat,
                        fold,
                        split_seed,
                        float("nan"),
                        "training fold lacks class diversity",
                    )
                    continue
                sampler = clone(oversampler)
                try:
                    error = validate_oversampling(
                        X_train,
                        y_train,
                        minority_label=minority_label,
                        oversampler=sampler,
                        hidden_ratio=_FOLD_HIDDEN_RATIO,
                        metric=metric,
                    )
                except Exception as exc:  # pragma: no cover - defensive
                    warnings.warn(
                        f"Validation failed for fold {fold}: {exc}", stacklevel=2
                    )
                    record(repeat, fold, split_seed, float("nan"), str(exc))
                    continue
                if np.isnan(error):
                    # validate_oversampling returns nan when the sampler made
                    # no synthetic points. This used to `continue` with no
                    # warning at all, so the fold vanished from both the mean
                    # and the count backing the interval.
                    record(
                        repeat,
                        fold,
                        split_seed,
                        float("nan"),
                        "no synthetic samples generated",
                    )
                    continue
                record(repeat, fold, split_seed, float(error), "")
        return records

    def _confidence_interval(self, values: Sequence[float]) -> tuple[float, float]:
        """Return a confidence interval for the **mean** of the provided values.

        Always a confidence interval for the mean, at every sample size.

        This previously switched formula at n = 30: a Student-t interval for the
        mean below, and the 2.5th-97.5th percentiles of the observations at or
        above. Those are different quantities. The t-interval narrows as
        1/sqrt(n); the percentile range describes the spread of individual
        observations and does not narrow at all. Both were written into the same
        ``ci_lower`` / ``ci_upper`` columns, so on sigma = 0.05 data the reported
        width jumped from 0.036 at n = 29 to 0.172 at n = 30 -- 4.7x wider from
        one extra observation -- and intervals could not be compared across
        configurations with different fold counts.

        The t-interval is used throughout. It is exact for normally distributed
        values and asymptotically valid otherwise, and it is what a reader
        assumes a "confidence interval" means.

        Args:
            values: Sample values.

        Returns:
            Lower and upper bounds for the mean, at the configured level.
        """
        if len(values) < 2:
            return (
                float(values[0]) if values else 0.0,
                float(values[0]) if values else 0.0,
            )
        arr = np.asarray(values, dtype=float)
        mean = float(arr.mean())
        alpha = 1 - self.confidence_level
        standard_error = float(stats.sem(arr))
        if standard_error == 0.0:
            # Every observation identical: the mean is known exactly.
            return (mean, mean)
        margin = float(stats.t.ppf(1 - alpha / 2, len(arr) - 1)) * standard_error
        return (mean - margin, mean + margin)

    def _add_statistical_analysis(self, frame: pd.DataFrame) -> pd.DataFrame:
        """Add pairwise p-values and effect sizes per dataset.

        Args:
            frame: Benchmark results dataframe.

        Returns:
            Dataframe with pairwise statistics columns populated.
        """
        frame = frame.copy()
        frame["pairwise_p_values"] = None
        frame["pairwise_effect_sizes"] = None

        # Grouped by (dataset, metric), not dataset alone. Grouping by dataset
        # put several metrics in one slice, and the lookups below take .iloc[0]
        # -- so the tests ran on whichever metric happened to sort first and the
        # result was stamped onto every row, including rows for the other
        # metrics. Error rates are not comparable across metrics, so those
        # p-values described a comparison the row did not represent.
        for _key, group in frame.groupby(
            ["dataset_name", "metric"], sort=False
        ):
            if len(group) < 2:
                continue
            pvals = self._pairwise_statistical_tests(group)
            effects = self._calculate_effect_sizes(group)
            for idx in group.index:
                # strict_json_dumps, not json.dumps: a skipped fold leaves a
                # nan in error_rates, Wilcoxon then returns nan, and the plain
                # encoder writes a bare `NaN` token. That is not JSON -- a
                # strict parser rejects the column outright -- and these two
                # columns are the only export path that had bypassed the
                # helper. nan becomes null, which parsers accept and which says
                # the same thing: no comparison was available.
                frame.at[idx, "pairwise_p_values"] = strict_json_dumps(
                    pvals, indent=None
                )
                frame.at[idx, "pairwise_effect_sizes"] = strict_json_dumps(
                    effects, indent=None
                )
        return frame

    def _pairwise_statistical_tests(
        self, dataset_slice: pd.DataFrame
    ) -> dict[str, float]:
        """Compute pairwise Wilcoxon tests across oversamplers.

        Args:
            dataset_slice: Subset of results for a single dataset.

        Returns:
            Mapping of ``oversampler_a_vs_b`` to corrected p-values.
        """
        p_values: dict[str, float] = {}
        oversamplers = dataset_slice["oversampler_name"].unique()
        for i, os1 in enumerate(oversamplers):
            for os2 in oversamplers[i + 1 :]:
                errors1 = np.asarray(
                    dataset_slice.loc[
                        dataset_slice["oversampler_name"] == os1, "error_rates"
                    ].iloc[0],
                    dtype=float,
                )
                errors2 = np.asarray(
                    dataset_slice.loc[
                        dataset_slice["oversampler_name"] == os2, "error_rates"
                    ].iloc[0],
                    dtype=float,
                )
                try:
                    if (
                        len(errors1) == 0
                        or len(errors2) == 0
                        or len(errors1) != len(errors2)
                        or np.allclose(errors1, errors2)
                    ):
                        p_val = 1.0
                    else:
                        with warnings.catch_warnings():
                            warnings.simplefilter("ignore", RuntimeWarning)
                            _, p_val = stats.wilcoxon(errors1, errors2)
                except Exception:
                    p_val = 1.0
                key = f"{os1}_vs_{os2}"
                p_values[key] = p_val
        return self._apply_correction(p_values)

    def _calculate_effect_sizes(self, dataset_slice: pd.DataFrame) -> dict[str, float]:
        """Compute pairwise matched-pairs rank-biserial correlations.

        Paired, to match the design the p-value comes from. The tests are
        Wilcoxon signed-rank, which pairs the two samplers fold by fold; the
        effect size was independent-samples Cohen's d, which pools the two
        standard deviations and throws that pairing away. On fold errors that
        move together -- an awkward fold is awkward for both samplers -- the
        pooled deviation is dominated by between-fold variation that the paired
        test has already removed, so the effect looks smaller than the test
        says it is.

        Rank-biserial is the natural companion to a rank test: it is computed
        from the same signed ranks Wilcoxon uses.

        Args:
            dataset_slice: Results for one (dataset, metric).

        Returns:
            Mapping of ``a_vs_b`` to a correlation in ``[-1, 1]``. Positive
            means the first sampler had the higher error rate on more folds,
            weighted by how much higher -- so positive favours the second.
        """
        effect_sizes: dict[str, float] = {}
        oversamplers = dataset_slice["oversampler_name"].unique()
        for i, os1 in enumerate(oversamplers):
            for os2 in oversamplers[i + 1 :]:
                errors1 = np.asarray(
                    dataset_slice.loc[
                        dataset_slice["oversampler_name"] == os1, "error_rates"
                    ].iloc[0],
                    dtype=float,
                )
                errors2 = np.asarray(
                    dataset_slice.loc[
                        dataset_slice["oversampler_name"] == os2, "error_rates"
                    ].iloc[0],
                    dtype=float,
                )
                effect = self._rank_biserial(errors1, errors2)
                if effect is not None:
                    effect_sizes[f"{os1}_vs_{os2}"] = effect
        return effect_sizes

    @staticmethod
    def _rank_biserial(x: np.ndarray, y: np.ndarray) -> float | None:
        """Matched-pairs rank-biserial correlation for two paired samples.

        ``(W+ - W-) / (W+ + W-)`` over the signed ranks of the differences,
        which is exactly what Wilcoxon signed-rank sums. Zero differences are
        dropped, as the test drops them.

        Returns ``None`` when the samples cannot be paired or every difference
        is zero -- there is no effect to report, and 0.0 would claim there is
        one and that it is exactly nil.
        """
        if len(x) != len(y) or len(x) == 0:
            return None
        diff = x - y
        nonzero = diff[diff != 0]
        if nonzero.size == 0:
            return None
        ranks = stats.rankdata(np.abs(nonzero))
        positive = float(ranks[nonzero > 0].sum())
        negative = float(ranks[nonzero < 0].sum())
        total = positive + negative
        if total == 0:
            return None
        return (positive - negative) / total

    @staticmethod
    def _pooled_std(x: np.ndarray, y: np.ndarray) -> float:
        """Return pooled standard deviation for two samples.

        Args:
            x: Sample 1.
            y: Sample 2.

        Returns:
            Pooled standard deviation.
        """
        if len(x) < 2 or len(y) < 2:
            return 0.0
        n1, n2 = len(x), len(y)
        s1, s2 = x.var(ddof=1), y.var(ddof=1)
        pooled = ((n1 - 1) * s1 + (n2 - 1) * s2) / (n1 + n2 - 2)
        return math.sqrt(max(pooled, 0.0))

    def _apply_correction(self, p_values: dict[str, float]) -> dict[str, float]:
        """Apply multiple-comparison correction to p-values.

        With 8 oversamplers there are 28 pairwise comparisons, and uncorrected
        p-values manufacture significance at that many looks.

        Args:
            p_values: Raw p-values keyed by comparison.

        Returns:
            Corrected p-values using the configured method: ``"holm"``
            (family-wise error rate, the default and the right choice for small
            families), ``"bh"`` / ``"fdr"`` (false discovery rate, better when
            the family is large and some false positives are tolerable), or
            ``"bonferroni"``.

        Notes:
            Both step procedures enforce **monotonicity**, which the previous
            Holm implementation omitted: it scaled each p-value by its rank
            without taking a running maximum, so raw p-values of
            ``[0.01, 0.02, 0.03]`` corrected to ``[0.03, 0.04, 0.03]``. The
            least significant comparison came out *more* significant than the
            middle one, which is incoherent and can flip a decision at a fixed
            alpha.
        """
        if not p_values:
            return p_values

        keys = list(p_values)
        raw = np.asarray([p_values[k] for k in keys], dtype=float)
        m = len(raw)
        method = self.correction_method.lower()

        if method == "bonferroni":
            adjusted = np.minimum(raw * m, 1.0)
            return dict(zip(keys, (float(v) for v in adjusted), strict=True))

        order = np.argsort(raw, kind="stable")

        if method in {"bh", "fdr", "benjamini-hochberg"}:
            # Step-up: scale by m / rank, then enforce monotonicity from the
            # largest p-value downward.
            ranks = np.arange(1, m + 1)
            scaled = raw[order] * m / ranks
            adjusted_sorted = np.minimum.accumulate(scaled[::-1])[::-1]
        else:
            # Holm step-down: scale by the number of remaining hypotheses, then
            # enforce monotonicity from the smallest p-value upward.
            scaled = raw[order] * (m - np.arange(m))
            adjusted_sorted = np.maximum.accumulate(scaled)

        adjusted_sorted = np.minimum(adjusted_sorted, 1.0)
        adjusted = np.empty_like(adjusted_sorted)
        adjusted[order] = adjusted_sorted
        return dict(zip(keys, (float(v) for v in adjusted), strict=True))

    @staticmethod
    def _result_to_dict(result: BenchmarkResult) -> dict[str, Any]:
        """Convert BenchmarkResult to a serializable dictionary.

        Args:
            result: Benchmark result structure.

        Returns:
            Dict suitable for DataFrame construction or serialization.
        """
        return {
            # Canonical names first; the `_name` spellings below are this
            # frame's originals and are kept for existing readers.
            "dataset": result.dataset_name,
            "oversampler": result.oversampler_name,
            "dataset_name": result.dataset_name,
            "oversampler_name": result.oversampler_name,
            "metric": result.metric,
            "hidden_ratio": result.hidden_ratio,
            "reference": result.reference,
            "minority_label": result.minority_label,
            "random_state": result.random_state,
            "n_folds": result.n_folds,
            "n_repeats": result.n_repeats,
            "oversampleqa_version": result.oversampleqa_version,
            "mean_error": result.mean_error,
            "std_error": result.std_error,
            "ci_lower": result.confidence_interval[0],
            "ci_upper": result.confidence_interval[1],
            "n_observations": len(result.error_rates),
            "error_rates": result.error_rates,
            "effect_size": result.effect_size,
            "p_value": result.p_value,
        }

run_comprehensive_benchmark(datasets, oversamplers, metrics=None)

Run repeated stratified benchmarking across datasets.

Parameters

datasets: Sequence of dataset descriptors. Each entry should provide data, target and optionally name and minority_label. oversamplers: Sequence of initialised oversampler instances (will be cloned). metrics: Distance metrics to evaluate. Defaults to Hassanat, Euclidean, Mahalanobis.

Source code in src/oversampleqa/advanced_benchmark.py

def run_comprehensive_benchmark(
    self,
    datasets: Sequence[dict[str, Any]],
    oversamplers: Sequence[Any],
    metrics: Sequence[str] | None = None,
) -> pd.DataFrame:
    """Run repeated stratified benchmarking across datasets.

    Parameters
    ----------
    datasets:
        Sequence of dataset descriptors. Each entry should provide ``data``,
        ``target`` and optionally ``name`` and ``minority_label``.
    oversamplers:
        Sequence of initialised oversampler instances (will be cloned).
    metrics:
        Distance metrics to evaluate. Defaults to Hassanat, Euclidean, Mahalanobis.
    """

    # mahalanobis is not a default. Without a covariance inverse it is
    # Euclidean, so it produced a third set of rows identical to the
    # euclidean ones -- inflating euclidean's weight in the rankings and
    # splitting one comparison into two for the p-value correction. Pass it
    # explicitly, with cov_inv in metric_kwargs, if you want it.
    metrics = tuple(metrics or ("hassanat", "euclidean"))
    # Reset per run: a reused engine must not accumulate skips or folds.
    self._skipped = []
    self._fold_records_all = []

    all_results: list[BenchmarkResult] = []
    for dataset in datasets:
        all_results.extend(
            self._benchmark_single_dataset(dataset, oversamplers, metrics)
        )

    frame = pd.DataFrame([self._result_to_dict(r) for r in all_results])

    if self._skipped:
        warnings.warn(
            f"{len(self._skipped)} of "
            f"{len(datasets) * len(oversamplers) * len(metrics)} "
            "dataset/oversampler/metric combinations produced no usable "
            "folds and are absent from the results: "
            + ", ".join(self._skipped[:5])
            + (" ..." if len(self._skipped) > 5 else "")
            + ". The most common cause is a minority class too small to "
            "hold out from once it has been split into folds -- try fewer "
            "folds or a larger hidden_ratio.",
            UserWarning,
            stacklevel=2,
        )

    if frame.empty:
        # Return the expected columns rather than a (0, 0) frame. An empty
        # frame with no columns raises KeyError on any column access, so a
        # caller that handles "no results" still breaks.
        return pd.DataFrame(columns=list(_RESULT_COLUMNS))

    frame = self._add_statistical_analysis(frame)
    return frame

fold_results()

Return one row per attempted fold from the most recent run.

The summary frame reports a mean and interval per (dataset, oversampler, metric). That is enough to read a ranking and not enough to check one: it cannot be re-aggregated, plotted as a distribution, or given a different interval, and it does not say how many folds actually contributed.

This frame answers those. Skipped folds are present with error_rate of nan, skipped true and a stated skip_reason, because a mean over three surviving folds of twenty-five is indistinguishable from a mean over twenty-five once the skips are dropped.

split_seed is the seed given to the fold splitter for that repeat, so a single repeat can be reproduced without rerunning the sweep.

Returns:

Type	Description
DataFrame	A long-format frame with :data:_FOLD_COLUMNS. Empty of rows but
DataFrame	not of columns when no run has happened yet, so column access works
DataFrame	either way.

Source code in src/oversampleqa/advanced_benchmark.py

def fold_results(self) -> pd.DataFrame:
    """Return one row per attempted fold from the most recent run.

    The summary frame reports a mean and interval per
    (dataset, oversampler, metric). That is enough to read a ranking and
    not enough to check one: it cannot be re-aggregated, plotted as a
    distribution, or given a different interval, and it does not say how
    many folds actually contributed.

    This frame answers those. Skipped folds are present with ``error_rate``
    of ``nan``, ``skipped`` true and a stated ``skip_reason``, because a
    mean over three surviving folds of twenty-five is indistinguishable
    from a mean over twenty-five once the skips are dropped.

    ``split_seed`` is the seed given to the fold splitter for that repeat,
    so a single repeat can be reproduced without rerunning the sweep.

    Returns:
        A long-format frame with :data:`_FOLD_COLUMNS`. Empty of rows but
        not of columns when no run has happened yet, so column access works
        either way.
    """
    frame = pd.DataFrame([asdict(record) for record in self._fold_records_all])
    # Built from the records first, then aliased: FoldRecord stays the one
    # place a fold's identity is defined, rather than carrying each name
    # twice and letting the two drift.
    if frame.empty:
        frame = pd.DataFrame(columns=list(_FOLD_COLUMNS))
    else:
        frame["dataset"] = frame["dataset_name"]
        frame["oversampler"] = frame["oversampler_name"]
    return frame.reindex(columns=list(_FOLD_COLUMNS))

DatasetRepository

Repository for curated real-world and synthetic benchmarking datasets.

Source code in src/oversampleqa/advanced_benchmark.py

class DatasetRepository:
    """Repository for curated real-world and synthetic benchmarking datasets."""

    def __init__(self, cache_dir: str = ".oversampleqa_datasets") -> None:
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(parents=True, exist_ok=True)

    def load_research_datasets(
        self,
        domains: Sequence[str] | None = None,
        max_samples: int = 10_000,
        include_openml: bool = False,
    ) -> list[dict[str, Any]]:
        """Load curated datasets for benchmarking.

        Args:
            domains: Domain names to load.
            max_samples: Maximum number of samples per dataset.
            include_openml: Whether to attempt OpenML downloads.

        Returns:
            List of dataset descriptors.
        """
        domains = tuple(domains or ("medical", "financial"))
        datasets: list[dict[str, Any]] = []
        for domain in domains:
            datasets.extend(self._load_domain(domain, max_samples, include_openml))
        return datasets

    def _load_domain(
        self, domain: str, max_samples: int, include_openml: bool
    ) -> list[dict[str, Any]]:
        """Load datasets for a single domain.

        Args:
            domain: Domain name.
            max_samples: Maximum number of samples per dataset.
            include_openml: Whether to attempt OpenML downloads.

        Returns:
            List of dataset descriptors.
        """
        domain = domain.lower()
        if domain == "medical":
            return self._load_medical(max_samples, include_openml)
        if domain == "financial":
            return self._load_financial(max_samples)
        return []

    def _load_medical(
        self, max_samples: int, include_openml: bool
    ) -> list[dict[str, Any]]:
        """Load medical datasets for benchmarking.

        Args:
            max_samples: Maximum number of samples per dataset.
            include_openml: Whether to attempt OpenML downloads.

        Returns:
            List of dataset descriptors.
        """
        from sklearn.datasets import load_breast_cancer

        cancer = load_breast_cancer()
        X, y = cancer.data[:max_samples], cancer.target[:max_samples]
        datasets = [
            {
                "name": "breast_cancer",
                "data": X,
                "target": y,
                # Derived, not declared. The full dataset's minority is class 0
                # (212 malignant against 357 benign) and this said 1, so the
                # benchmark oversampled the majority. max_samples then inverts
                # it again -- the first 200 rows are 104 class-0 to 96 class-1.
                "minority_label": infer_minority_label(y),
                "provenance": bundled_provenance(
                    "sklearn.datasets.load_breast_cancer",
                    url="https://archive.ics.uci.edu/dataset/17/breast+cancer+wisconsin+diagnostic",
                    license="CC BY 4.0 (UCI ML Repository); redistributed with scikit-learn.",
                    notes=(
                        f"Truncated to the first {max_samples} rows, which is a "
                        "positional slice and not a random sample."
                    ),
                    max_samples=max_samples,
                ),
            }
        ]
        if include_openml:
            try:
                from sklearn.datasets import fetch_openml

                diabetes = fetch_openml("diabetes", version=1, as_frame=False)
                Xd = diabetes.data[:max_samples]
                yd = (diabetes.target[:max_samples] == "tested_positive").astype(int)
                datasets.append(
                    {
                        "name": "diabetes_openml",
                        "data": Xd,
                        "target": yd,
                        "minority_label": infer_minority_label(yd),
                        "provenance": openml_provenance(
                            "diabetes",
                            1,
                            notes=(
                                "Version pinned to 1. Target binarised as "
                                "'tested_positive'. Truncated to the first "
                                f"{max_samples} rows."
                            ),
                        ),
                    }
                )
            except Exception as exc:  # pragma: no cover - network dependent
                warnings.warn(f"OpenML download failed: {exc}", stacklevel=2)
        return datasets

    def _load_financial(self, max_samples: int) -> list[dict[str, Any]]:
        """Load financial datasets for benchmarking.

        Args:
            max_samples: Maximum number of samples per dataset.

        Returns:
            List of dataset descriptors.
        """
        try:
            from imbalanced_datasets import creditcard  # optional
        except Exception:  # pragma: no cover
            creditcard = None

        datasets: list[dict[str, Any]] = []
        if creditcard is not None:  # pragma: no cover - optional dependency
            X, y = creditcard.load_data()
            datasets.append(
                {
                    "name": "creditcard",
                    "data": X[:max_samples],
                    "target": y[:max_samples],
                    "minority_label": infer_minority_label(y[:max_samples]),
                    "provenance": {
                        "source": "third-party",
                        "generator": "imbalanced_datasets.creditcard.load_data",
                        "params": {"max_samples": max_samples},
                        "url": "https://www.kaggle.com/datasets/mlg-ulb/creditcardfraud",
                        "license": (
                            "Database Contents License (DbCL) v1.0. Redistribution "
                            "terms are the upstream package's, not this project's."
                        ),
                        "notes": (
                            "Optional dependency; the loader pins no version, so "
                            "the contents are whatever the installed release "
                            f"ships. Truncated to the first {max_samples} rows."
                        ),
                    },
                }
            )
        return datasets

    def create_synthetic_benchmark_suite(
        self, difficulty_levels: Sequence[str] | None = None
    ) -> list[dict[str, Any]]:
        """Generate synthetic datasets for the requested difficulty levels.

        Args:
            difficulty_levels: Difficulty labels to generate.

        Returns:
            List of synthetic dataset descriptors.
        """
        difficulty_levels = tuple(
            difficulty_levels or ("easy", "medium", "hard", "extreme")
        )
        synthetic: list[dict[str, Any]] = []
        for difficulty in difficulty_levels:
            synthetic.extend(self._generate_difficulty(difficulty))
        return synthetic

    def _generate_difficulty(self, difficulty: str) -> list[dict[str, Any]]:
        """Create synthetic datasets for a single difficulty tier.

        Args:
            difficulty: Difficulty label.

        Returns:
            List of dataset descriptors for the difficulty tier.
        """
        from sklearn.datasets import make_classification

        difficulty = difficulty.lower()
        configs: list[dict[str, Any]]
        if difficulty == "easy":
            configs = [
                {
                    "n_samples": 600,
                    "n_features": 8,
                    "class_sep": 2.0,
                    "weights": [0.75, 0.25],
                },
                {
                    "n_samples": 800,
                    "n_features": 5,
                    "class_sep": 1.8,
                    "weights": [0.8, 0.2],
                },
            ]
        elif difficulty == "medium":
            configs = [
                {
                    "n_samples": 1000,
                    "n_features": 12,
                    "class_sep": 1.2,
                    "weights": [0.85, 0.15],
                },
            ]
        elif difficulty == "hard":
            configs = [
                {
                    "n_samples": 1500,
                    "n_features": 20,
                    "class_sep": 0.8,
                    "weights": [0.9, 0.1],
                },
            ]
        elif difficulty == "extreme":
            configs = [
                {
                    "n_samples": 2000,
                    "n_features": 40,
                    "class_sep": 0.4,
                    "weights": [0.97, 0.03],
                },
            ]
        else:
            configs = [
                {
                    "n_samples": 800,
                    "n_features": 10,
                    "class_sep": 1.0,
                    "weights": [0.8, 0.2],
                }
            ]

        datasets: list[dict[str, Any]] = []
        for idx, config in enumerate(configs):
            X, y = make_classification(
                random_state=42 + idx,
                n_informative=max(2, config["n_features"] // 2),
                n_redundant=config["n_features"] // 4,
                flip_y=0.02 if difficulty in {"hard", "extreme"} else 0.0,
                **config,
            )
            flip_y = 0.02 if difficulty in {"hard", "extreme"} else 0.0
            datasets.append(
                {
                    "name": f"{difficulty}_synthetic_{idx}",
                    "data": X,
                    "target": y,
                    "minority_label": infer_minority_label(y),
                    "difficulty": difficulty,
                    "provenance": synthetic_provenance(
                        "sklearn.datasets.make_classification",
                        random_state=42 + idx,
                        n_informative=max(2, config["n_features"] // 2),
                        n_redundant=config["n_features"] // 4,
                        flip_y=flip_y,
                        **config,
                    ),
                }
            )
        return datasets

load_research_datasets(domains=None, max_samples=10000, include_openml=False)

Load curated datasets for benchmarking.

Parameters:

Name	Type	Description	Default
domains	Sequence[str] | None	Domain names to load.	None
max_samples	int	Maximum number of samples per dataset.	10000
include_openml	bool	Whether to attempt OpenML downloads.	False

Returns:

Type	Description
list[dict[str, Any]]	List of dataset descriptors.

Source code in src/oversampleqa/advanced_benchmark.py

def load_research_datasets(
    self,
    domains: Sequence[str] | None = None,
    max_samples: int = 10_000,
    include_openml: bool = False,
) -> list[dict[str, Any]]:
    """Load curated datasets for benchmarking.

    Args:
        domains: Domain names to load.
        max_samples: Maximum number of samples per dataset.
        include_openml: Whether to attempt OpenML downloads.

    Returns:
        List of dataset descriptors.
    """
    domains = tuple(domains or ("medical", "financial"))
    datasets: list[dict[str, Any]] = []
    for domain in domains:
        datasets.extend(self._load_domain(domain, max_samples, include_openml))
    return datasets

create_synthetic_benchmark_suite(difficulty_levels=None)

Generate synthetic datasets for the requested difficulty levels.

Parameters:

Name	Type	Description	Default
difficulty_levels	Sequence[str] | None	Difficulty labels to generate.	None

Returns:

Type	Description
list[dict[str, Any]]	List of synthetic dataset descriptors.

Source code in src/oversampleqa/advanced_benchmark.py

def create_synthetic_benchmark_suite(
    self, difficulty_levels: Sequence[str] | None = None
) -> list[dict[str, Any]]:
    """Generate synthetic datasets for the requested difficulty levels.

    Args:
        difficulty_levels: Difficulty labels to generate.

    Returns:
        List of synthetic dataset descriptors.
    """
    difficulty_levels = tuple(
        difficulty_levels or ("easy", "medium", "hard", "extreme")
    )
    synthetic: list[dict[str, Any]] = []
    for difficulty in difficulty_levels:
        synthetic.extend(self._generate_difficulty(difficulty))
    return synthetic

format_statistical_summary(results_df, significance_level=0.05)

Render a Markdown summary of a statistical benchmark frame.

The frame is expected to come from :meth:StatisticalBenchmark.run_comprehensive_benchmark. The summary lists, per dataset, the mean error, standard deviation and confidence interval for each oversampler/metric, followed by the statistically significant pairwise comparisons (corrected p-value below significance_level).

Parameters:

Name	Type	Description	Default
results_df	DataFrame	Benchmark results dataframe.	required
significance_level	float	Threshold below which a pairwise p-value is reported.	0.05

Returns:

Type	Description
str	A Markdown-formatted string.

Source code in src/oversampleqa/advanced_benchmark.py

def format_statistical_summary(
    results_df: pd.DataFrame, significance_level: float = 0.05
) -> str:
    """Render a Markdown summary of a statistical benchmark frame.

    The frame is expected to come from
    :meth:`StatisticalBenchmark.run_comprehensive_benchmark`. The summary lists,
    per dataset, the mean error, standard deviation and confidence interval for
    each oversampler/metric, followed by the statistically significant pairwise
    comparisons (corrected p-value below ``significance_level``).

    Args:
        results_df: Benchmark results dataframe.
        significance_level: Threshold below which a pairwise p-value is reported.

    Returns:
        A Markdown-formatted string.
    """
    if results_df.empty:
        return (
            "# OversampleQA Statistical Benchmark\n\nNo benchmark results available.\n"
        )

    lines = ["# OversampleQA Statistical Benchmark", ""]
    lines.append(
        "Confidence intervals use the configured confidence level. Pairwise "
        "p-values and effect sizes (Cohen's d) compare oversamplers on the same "
        "dataset and metric, corrected by the configured method."
    )
    lines.append("")

    for dataset_name, group in results_df.groupby("dataset_name", sort=True):
        lines.append(f"## Dataset: {dataset_name}")
        lines.append("")
        lines.append("| Oversampler | Metric | Mean error | Std | CI | n |")
        lines.append("| --- | --- | --- | --- | --- | --- |")
        for _, row in group.iterrows():
            ci = f"[{row['ci_lower']:.3f}, {row['ci_upper']:.3f}]"
            lines.append(
                f"| {row['oversampler_name']} | {row['metric']} | "
                f"{row['mean_error']:.3f} | {row['std_error']:.3f} | {ci} | "
                f"{int(row['n_observations'])} |"
            )
        lines.append("")

        significant = _significant_pairwise(group, significance_level)
        if significant:
            lines.append(
                f"Significant pairwise differences (p < {significance_level}):"
            )
            for label, p_val, effect in significant:
                effect_str = f", d={effect:.2f}" if effect is not None else ""
                lines.append(f"- {label}: p={p_val:.4f}{effect_str}")
            lines.append("")

    return "\n".join(lines)

create_benchmark_report(results_df, output_path='benchmark_report.html')

Create a lightweight HTML report summarising benchmark statistics.

Parameters:

Name	Type	Description	Default
results_df	DataFrame	Benchmark results dataframe.	required
output_path	str	Output HTML path.	'benchmark_report.html'

Returns:

Type	Description
Path	Path to the generated report.

Source code in src/oversampleqa/advanced_benchmark.py

def create_benchmark_report(
    results_df: pd.DataFrame, output_path: str = "benchmark_report.html"
) -> Path:
    """Create a lightweight HTML report summarising benchmark statistics.

    Args:
        results_df: Benchmark results dataframe.
        output_path: Output HTML path.

    Returns:
        Path to the generated report.
    """

    output = Path(output_path)
    output.parent.mkdir(parents=True, exist_ok=True)

    if results_df.empty:
        html = (
            "<html><body><h1>No benchmark results available.</h1>"
            "<h2>Run metadata</h2>"
            f"{report_metadata_html(results_df)}</body></html>"
        )
        output.write_text(html, encoding="utf-8")
        write_export_metadata(output, export_kind="statistical_benchmark_report")
        return output

    summary = results_df.copy()
    summary["ci"] = summary.apply(
        lambda row: f"[{row['ci_lower']:.3f}, {row['ci_upper']:.3f}]", axis=1
    )
    summary["p_values"] = summary["pairwise_p_values"].fillna("{}")
    summary["effect_sizes"] = summary["pairwise_effect_sizes"].fillna("{}")

    table_html = summary[
        [
            "dataset_name",
            "oversampler_name",
            "metric",
            "mean_error",
            "std_error",
            "ci",
            "p_values",
            "effect_sizes",
        ]
    ].to_html(index=False, escape=False)

    html = f"""
    <html>
        <head>
            <title>OversampleQA Benchmark Report</title>
            <style>
                body {{ font-family: Arial, sans-serif; padding: 2rem; }}
                table {{ border-collapse: collapse; width: 100%; }}
                th, td {{ border: 1px solid #ccc; padding: 0.5rem; }}
                th {{ background: #f5f5f5; }}
            </style>
        </head>
        <body>
            <h1>OversampleQA Benchmark Report</h1>
            <h2>Run metadata</h2>
            {report_metadata_html(results_df)}
            <h2>Results</h2>
            {table_html}
        </body>
    </html>
    """
    output.write_text(html, encoding="utf-8")
    write_export_metadata(
        output,
        export_kind="statistical_benchmark_report",
        data=summary,
    )
    return output