Source code for ictonyx.plotting

# ictonyx/plotting.py
import warnings
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union

import numpy as np
import pandas as pd
from scipy import stats as _scipy_stats

from . import settings

# Optional plotting dependencies
try:
    import matplotlib.pyplot as plt
    import seaborn as sns
    from matplotlib.figure import Figure

    HAS_PLOTTING = True
except ImportError:
    plt = None  # type: ignore[assignment]
    Figure = None  # type: ignore[assignment,misc]
    sns = None
    HAS_PLOTTING = False

try:
    from matplotlib.axes import Axes
except ImportError:
    Axes = None  # type: ignore[assignment,misc]

# Optional sklearn for metrics
try:
    from sklearn.metrics import auc, precision_recall_curve, roc_curve

    HAS_SKLEARN_METRICS = True
except ImportError:
    HAS_SKLEARN_METRICS = False

# Optional TensorFlow for utils
try:
    from tensorflow.keras.utils import to_categorical

    HAS_TENSORFLOW_UTILS = True
except ImportError:
    HAS_TENSORFLOW_UTILS = False

if TYPE_CHECKING:
    from .core import BaseModelWrapper
    from .runners import VariabilityStudyResults


# --- Helpers ---


def _check_plotting():
    if not HAS_PLOTTING:
        raise ImportError(
            "matplotlib and seaborn are required. Install with: pip install matplotlib seaborn"
        )


def _check_sklearn_metrics():
    if not HAS_SKLEARN_METRICS:
        raise ImportError(
            "scikit-learn required for ROC/PR curves. Install with: pip install scikit-learn"
        )


def _check_tensorflow_utils():
    if not HAS_TENSORFLOW_UTILS:
        raise ImportError(
            "TensorFlow required for multi-class plotting. Install with: pip install tensorflow"
        )


def _finalize_plot(fig: "Figure", show_arg: Optional[bool]) -> Optional["Figure"]:
    """Centralized logic for showing or returning a plot.

    When display is active, calls ``plt.show()`` and returns ``None`` so
    Jupyter does not render the figure a second time as the cell's return value.
    When display is suppressed, returns the figure for programmatic use.
    """
    should_show = show_arg if show_arg is not None else settings.should_display()
    if should_show:
        plt.show()
        return None  # Prevent Jupyter double-render
    return fig


def _find_metric_columns(df: pd.DataFrame, metric: str) -> Tuple[Optional[str], Optional[str]]:
    """
    Smart column detection. Tries to find train/val columns for a metric
    even if prefixes don't match exactly.
    """
    base = metric.replace("train_", "").replace("val_", "")

    # Candidates for training
    train_candidates = [f"train_{base}", base, f"{base}_train"]
    train_col = next((c for c in train_candidates if c in df.columns), None)

    # Candidates for validation
    val_candidates = [f"val_{base}", f"{base}_val", f"test_{base}"]
    val_col = next((c for c in val_candidates if c and c in df.columns), None)

    return train_col, val_col


def _apply_style(
    ax: "Axes",
    grid: bool = True,
    fontsize: int = 11,
    despine: bool = True,
) -> None:
    """Apply standard Ictonyx axis styling.

    Centralises grid, spine, and font-size settings so all plot functions
    produce consistent visuals and changes propagate from a single place.

    Args:
        ax: The matplotlib Axes to style.
        grid: Draw a subtle dotted grid. Default ``True``.
        fontsize: Tick label font size. Default 11.
        despine: Remove the top and right spines. Default ``True``.
    """
    if grid:
        ax.grid(True, linestyle=":", alpha=0.5, color=settings.THEME["grid"])
        ax.set_axisbelow(True)
    if despine:
        ax.spines["top"].set_visible(False)
        ax.spines["right"].set_visible(False)
    ax.tick_params(labelsize=fontsize)


# --- Standard Plots ---


[docs] def plot_confusion_matrix( cm_df: pd.DataFrame, title: str = "", show: Optional[bool] = None ) -> Optional["Figure"]: """Plot a confusion matrix as an annotated heatmap. Args: cm_df: A square ``pd.DataFrame`` where rows are true labels and columns are predicted labels. Values are integer counts. Typically produced by :func:`~ictonyx.analysis.get_confusion_matrix_df`. title: Plot title. Default: ``"Confusion Matrix"``. show: If ``True``, call ``plt.show()``. If ``False``, return the figure without displaying. If ``None`` (default), defer to the global :func:`~ictonyx.settings.set_display_plots` setting. Returns: The ``matplotlib.figure.Figure``, or ``None`` if display is enabled. """ _check_plotting() fig, ax = plt.subplots(figsize=settings.get_figsize((10, 8)), dpi=150) sns.heatmap(cm_df, annot=True, fmt="d", cmap=settings.THEME["sequential"], ax=ax) ax.set_title(title if title else "Confusion Matrix") ax.set_xlabel("Predicted Label") ax.set_ylabel("True Label") return _finalize_plot(fig, show)
[docs] def plot_training_history( history: Any, title: Optional[str] = None, metrics: Optional[List[str]] = None, show: Optional[bool] = None, ) -> Optional["Figure"]: """Plot training and validation metrics over epochs. Creates one subplot per metric, each showing train and validation curves. Accepts DataFrames, dicts, lists of dicts, or Keras ``History`` objects. Args: history: Training history in any of the supported formats. Column/key naming convention: ``'train_accuracy'`` and ``'val_accuracy'``, or ``'accuracy'`` and ``'val_accuracy'``. title: Plot title. If ``None``, auto-generated from epoch count. metrics: List of metric base names to plot (e.g. ``['accuracy', 'loss']``). If ``None``, auto-detected from available columns. show: Display behavior. See :func:`plot_confusion_matrix`. Returns: The ``matplotlib.figure.Figure``, or ``None`` if display is enabled. """ _check_plotting() # Convert various history formats to DataFrame if isinstance(history, pd.DataFrame): history_df = history elif isinstance(history, dict): history_df = pd.DataFrame(history) elif isinstance(history, list) and all(isinstance(i, dict) for i in history): history_df = pd.DataFrame(history) else: try: history_df = pd.DataFrame(history.history) except AttributeError: settings.logger.error("The provided history object format is not supported.") return None # Determine metrics if metrics is None: available = history_df.columns.tolist() default_metrics = [] if "accuracy" in available or "train_accuracy" in available: default_metrics.append("accuracy") if "loss" in available or "train_loss" in available: default_metrics.append("loss") if default_metrics: metrics = default_metrics else: metrics = [col for col in available if not col.startswith("val_") and col != "epoch"] if not metrics: settings.logger.warning("No metrics found to plot.") return None if title is None: n_epochs = len(history_df) title = f"Training Progress: {n_epochs} Epochs" num_metrics = len(metrics) fig, axes = plt.subplots(1, num_metrics, figsize=(7 * num_metrics, 5)) if num_metrics == 1: axes = [axes] colors = settings.THEME for i, metric in enumerate(metrics): ax = axes[i] # Use smart column finding train_col, val_col = _find_metric_columns(history_df, metric) if not train_col: continue metric_display = metric.replace("_", " ").title() if metric.lower() in ["mse", "mae", "rmse"]: metric_display = metric.upper() elif metric.lower() == "auc": metric_display = "AUC" epochs = range(1, len(history_df) + 1) ax.plot( epochs, history_df[train_col], label=f"Training", color=colors["train"], linewidth=2 ) if val_col: ax.plot( epochs, history_df[val_col], label=f"Validation", color=colors["val"], linewidth=2 ) final_train = history_df[train_col].iloc[-1] final_val = history_df[val_col].iloc[-1] ax.plot([], [], " ", label=f"Final: {final_train:.4f} / {final_val:.4f}") else: final_train = history_df[train_col].iloc[-1] ax.plot([], [], " ", label=f"Final: {final_train:.4f}") ax.set_title(metric_display, fontsize=12, fontweight="bold") ax.set_xlabel("Epoch", fontsize=11) ax.set_ylabel(metric_display, fontsize=11) ax.xaxis.set_major_locator(plt.MaxNLocator(integer=True)) # Set y-axis limits for bounded metrics if any(x in metric.lower() for x in ["accuracy", "auc", "precision", "recall", "f1"]): ax.set_ylim([0, 1.05]) ax.legend(loc="best", framealpha=0.9) ax.grid(True, alpha=0.3, color=colors.get("grid", "#e6e6e6")) fig.suptitle(title, fontsize=14, fontweight="bold") plt.tight_layout(rect=(0, 0, 1, 0.96)) return _finalize_plot(fig, show)
[docs] def plot_roc_curve( model_wrapper: "BaseModelWrapper", X_test: np.ndarray, y_test: np.ndarray, title: str = "", show: Optional[bool] = None, ) -> Optional["Figure"]: """Plot per-class ROC curves with AUC values. Binarizes labels using one-vs-rest and plots one ROC curve per class. Requires the model wrapper to have a ``predict_proba`` method. Args: model_wrapper: A trained :class:`~ictonyx.core.BaseModelWrapper` with ``predict_proba`` support. X_test: Test feature array. y_test: True class labels (integer-encoded). title: Plot title. Default: ``'Receiver Operating Characteristic'``. show: Display behavior. See :func:`plot_confusion_matrix`. Returns: The ``matplotlib.figure.Figure``, or ``None`` if the model does not support probability predictions. """ _check_plotting() _check_sklearn_metrics() if not hasattr(model_wrapper, "predict_proba"): settings.logger.warning( "Model does not have a predict_proba method. Cannot plot ROC curve." ) return None y_score = model_wrapper.predict_proba(X_test) if y_score.ndim == 1: y_score = np.vstack([1 - y_score, y_score]).T num_classes = y_score.shape[1] if HAS_TENSORFLOW_UTILS: y_test_binarized = to_categorical(y_test, num_classes=num_classes) else: y_test_binarized = np.eye(num_classes)[y_test.astype(int)] fig, ax = plt.subplots(figsize=settings.get_figsize((10, 8)), dpi=150) for i in range(num_classes): fpr, tpr, _ = roc_curve(y_test_binarized[:, i], y_score[:, i]) roc_auc = auc(fpr, tpr) ax.plot(fpr, tpr, label=f"ROC curve of class {i} (area = {roc_auc:.2f})") ax.plot([0, 1], [0, 1], "--", color=settings.THEME["baseline"], label="Chance") ax.set_xlim(0.0, 1.0) ax.set_ylim(0.0, 1.05) ax.set_xlabel("False Positive Rate") ax.set_ylabel("True Positive Rate") ax.set_title(title if title else "Receiver Operating Characteristic") ax.legend(loc="lower right") _apply_style(ax) return _finalize_plot(fig, show)
[docs] def plot_precision_recall_curve( model_wrapper: "BaseModelWrapper", X_test: np.ndarray, y_test: np.ndarray, title: str = "", show: Optional[bool] = None, ) -> Optional["Figure"]: """Plot per-class precision-recall curves with AUC values. Binarizes labels using one-vs-rest and plots one PR curve per class. Requires the model wrapper to have a ``predict_proba`` method. Args: model_wrapper: A trained :class:`~ictonyx.core.BaseModelWrapper` with ``predict_proba`` support. X_test: Test feature array. y_test: True class labels (integer-encoded). title: Plot title. Default: ``'Precision-Recall Curve'``. show: Display behavior. See :func:`plot_confusion_matrix`. Returns: The ``matplotlib.figure.Figure``, or ``None`` if the model does not support probability predictions. """ _check_plotting() _check_sklearn_metrics() if not hasattr(model_wrapper, "predict_proba"): settings.logger.warning( "Model does not have a predict_proba method. Cannot plot Precision-Recall curve." ) return None y_score = model_wrapper.predict_proba(X_test) if y_score.ndim == 1: y_score = np.vstack([1 - y_score, y_score]).T num_classes = y_score.shape[1] if HAS_TENSORFLOW_UTILS: y_test_binarized = to_categorical(y_test, num_classes=num_classes) else: y_test_binarized = np.eye(num_classes)[y_test.astype(int)] fig, ax = plt.subplots(figsize=settings.get_figsize((10, 8)), dpi=150) for i in range(num_classes): precision, recall, _ = precision_recall_curve(y_test_binarized[:, i], y_score[:, i]) pr_auc = auc(recall, precision) ax.plot( recall, precision, label=f"Precision-recall curve of class {i} (area = {pr_auc:.2f})" ) ax.set_xlim(0.0, 1.0) ax.set_ylim(0.0, 1.05) ax.set_xlabel("Recall") ax.set_ylabel("Precision") ax.set_title(title if title else "Precision-Recall Curve") ax.legend(loc="lower left") _apply_style(ax) return _finalize_plot(fig, show)
# --- Variability & Comparison Visualizations ---
[docs] def plot_variability_summary( all_runs_metrics_list: Optional[List[pd.DataFrame]] = None, final_metrics_series: Optional[Union[pd.Series, List]] = None, final_test_series: Optional[Union[pd.Series, List]] = None, metric: str = "accuracy", kind: Optional[str] = None, show_histogram: bool = True, show_boxplot: bool = False, show_mean_lines: bool = True, show_train: bool = True, show_val: bool = True, histogram_orientation: str = "vertical", alpha: Optional[float] = None, figsize: Optional[Tuple[float, float]] = None, dpi: int = 300, show: Optional[bool] = None, results: Optional[Any] = None, ) -> Optional["Figure"]: """Plot a multi-panel variability summary for a completed study. Generates a figure with: * **Overlaid training curves** — all runs plotted together, showing epoch-by-epoch convergence and spread. * **Final metric distribution** — histogram and/or boxplot of final-epoch values, showing the spread of outcomes across runs. Args: all_runs_metrics_list: List of per-run DataFrames (one per successful run), as stored in :attr:`VariabilityStudyResults.all_runs_metrics`. final_metrics_series: ``pd.Series`` of final-epoch validation metric values across runs. final_test_series: Optional ``pd.Series`` of test-set metric values. If provided, a second distribution is added. metric: Base metric name for labeling (default ``'accuracy'``). show_histogram: Show histogram panel (default ``True``). show_boxplot: Show boxplot panel (default ``False``). show_mean_lines: Overlay mean training curves on the trajectory panel (default ``True``). Set to ``False`` when runs diverge dramatically and the mean is misleading. show_train: Plot training metric curves (default ``True``). show_val: Plot validation metric curves (default ``True``). histogram_orientation: ``'vertical'`` (default) or ``'horizontal'``. Horizontal orientation aligns the distribution with the y-axis of the trajectory panel. alpha: Opacity of individual run curves. ``None`` (default) auto-calculates based on the number of runs. figsize: Figure size as ``(width, height)`` in inches. ``None`` (default) auto-calculates based on the number of panels. dpi: Figure resolution in dots per inch (default ``150``). show: Display behavior. See :func:`plot_confusion_matrix`. Returns: The ``matplotlib.figure.Figure``, or ``None`` if display is enabled. """ _check_plotting() # ── New dispatch path: results= + kind= ────────────────────────────── if results is not None and kind is not None: _dispatch = { "trajectories": plot_run_trajectories, "distribution": plot_run_distribution, "strip": plot_run_strip, } fn = _dispatch.get(kind) if fn is None: raise ValueError( f"Unknown kind='{kind}'. " f"Valid options: {sorted(_dispatch.keys())}." ) return fn(results, metric=None, show=show) # ── Legacy positional-argument form deprecation ─────────────────────── if all_runs_metrics_list is not None or final_metrics_series is not None: warnings.warn( "Passing all_runs_metrics_list, final_metrics_series, or " "final_test_series directly to plot_variability_summary() is " "deprecated and will be removed in v0.5.0. " "Pass a VariabilityStudyResults object via results= and use " "kind= to select the plot type. Example: " "ix.plot_variability_summary(results=my_results, kind='trajectories').", DeprecationWarning, stacklevel=2, ) # Allow passing a VariabilityStudyResults object directly via results= if results is not None: all_runs_metrics_list = results.all_runs_metrics # Auto-resolve metric base from preferred_metric when results provided resolved_base = results.preferred_metric(metric).replace("test_", "").replace("val_", "") try: final_metrics_series = pd.Series( results.get_metric_values(results.preferred_metric(metric)) ) except (KeyError, ValueError): try: final_metrics_series = pd.Series(results.get_metric_values(f"val_{metric}")) except (KeyError, ValueError): final_metrics_series = pd.Series(results.get_metric_values(metric)) # Also pull test series if available if results.has_test_data and final_test_series is None: test_key = f"test_{resolved_base}" try: final_test_series = pd.Series(results.get_test_metric_values(test_key)) except KeyError: pass if not all_runs_metrics_list: settings.logger.warning("No run metrics provided to plot.") return None if final_metrics_series is None: final_metrics_series = pd.Series([], dtype=float) if histogram_orientation not in ("vertical", "horizontal"): raise ValueError( f"histogram_orientation must be 'vertical' or 'horizontal', " f"got '{histogram_orientation}'" ) if alpha is not None and not (0 < alpha <= 1): raise ValueError(f"alpha must be between 0 and 1, got {alpha}") if not show_train and not show_val: settings.logger.warning( "Both show_train and show_val are False. " "Training curves panel will be empty." ) # Resolve Columns using smart detection sample_run = all_runs_metrics_list[0] train_col, val_col = _find_metric_columns(sample_run, metric) if not train_col and not val_col: settings.logger.warning( f"Could not find columns for metric '{metric}'. Available: {list(sample_run.columns)}" ) return None # Setup layout num_plots = 1 + int(show_histogram) + int(show_boxplot) if figsize is None: figsize = (7 * num_plots, 6) fig, axes = plt.subplots(1, num_plots, figsize=figsize, dpi=dpi) if num_plots == 1: axes = [axes] # PANEL 1: Trajectories ax = axes[0] colors = settings.THEME # Clean metric name for display metric_display = metric.replace("val_", "").replace("train_", "").replace("_", " ").title() # Calculate alpha if not provided if alpha is None: n_runs = len(all_runs_metrics_list) if n_runs <= 3: alpha = 0.6 elif n_runs <= 10: alpha = 0.4 elif n_runs <= 30: alpha = 0.3 else: alpha = max(0.1, 1.5 / n_runs) for df in all_runs_metrics_list: epochs = range(1, len(df) + 1) if show_train and train_col and train_col in df.columns: ax.plot(epochs, df[train_col], color=colors["train"], alpha=alpha) if show_val and val_col and val_col in df.columns: ax.plot(epochs, df[val_col], color=colors["val"], alpha=alpha) # Add Mean Lines if show_mean_lines and len(all_runs_metrics_list) > 1: if show_train and train_col: try: train_stack = np.array([run[train_col].values for run in all_runs_metrics_list]) ax.plot( range(1, train_stack.shape[1] + 1), np.mean(train_stack, axis=0), color=colors["train"], linewidth=3, label="Mean Train", ) except ValueError: pass if show_val and val_col: try: val_stack = np.array([run[val_col].values for run in all_runs_metrics_list]) ax.plot( range(1, val_stack.shape[1] + 1), np.mean(val_stack, axis=0), color=colors["val"], linewidth=3, label="Mean Val", ) except ValueError: pass ax.set_title(f"{metric_display} over {len(all_runs_metrics_list)} Runs") ax.set_xlabel("Epoch") ax.set_ylabel(metric_display) ax.legend() ax.grid(True, alpha=0.3, color=colors.get("grid", "#e6e6e6")) plot_idx = 1 # PANEL 2: Histogram if show_histogram and len(final_metrics_series) > 0: ax = axes[plot_idx] if histogram_orientation == "horizontal": sns.histplot( y=final_metrics_series, kde=True, ax=ax, color=colors["val"], label="Validation" ) if final_test_series is not None: sns.histplot( y=final_test_series, kde=True, ax=ax, color=colors["test"], label="Test" ) ax.set_xlabel("Frequency") ax.set_ylabel(f"Final {metric_display}") else: sns.histplot( final_metrics_series, kde=True, ax=ax, color=colors["val"], label="Validation" ) if final_test_series is not None: sns.histplot(final_test_series, kde=True, ax=ax, color=colors["test"], label="Test") ax.set_xlabel(f"Final {metric_display}") ax.set_ylabel("Frequency") ax.set_title(f"Final {metric_display} Distribution") ax.legend() plot_idx += 1 # PANEL 3: Boxplot if show_boxplot and len(final_metrics_series) > 0: ax = axes[plot_idx] box_dict = {"Val": pd.Series(final_metrics_series)} if final_test_series is not None: box_dict["Test"] = pd.Series(final_test_series) box_df = pd.DataFrame(box_dict).melt(var_name="Split", value_name=metric_display) palette_map = {"Val": colors["val"]} if final_test_series is not None: palette_map["Test"] = colors["test"] sns.boxplot( data=box_df, x="Split", y=metric_display, hue="Split", palette=palette_map, ax=ax, legend=False, ) ax.set_title("Performance Spread") plot_idx += 1 plt.tight_layout() return _finalize_plot(fig, show)
[docs] def plot_run_trajectories( results: "VariabilityStudyResults", metric: Optional[str] = "val_accuracy", ax: Optional["Axes"] = None, show: Optional[bool] = None, ) -> Optional["Figure"]: """Plot per-run epoch trajectories overlaid with the mean curve. Each run is drawn as a thin semi-transparent line; the mean trajectory is drawn as a solid thicker line. High spread at convergence signals seed-dependent behaviour. Args: results: A completed :class:`~ictonyx.runners.VariabilityStudyResults`. metric: Metric name to plot. Default ``'val_accuracy'``. ax: Optional existing Axes. Creates a new figure if ``None``. show: Display behavior. See :func:`plot_confusion_matrix`. Returns: The figure, or ``None`` if the metric is not found or display is active. """ _check_plotting() if metric is None: metric = results.preferred_metric(context="epoch") if ax is None: fig, ax = plt.subplots(figsize=settings.get_figsize((10, 6)), dpi=150) else: fig = ax.figure # type: ignore[assignment] run_dfs = results.all_runs_metrics if not run_dfs: settings.logger.warning("plot_run_trajectories: no run data available.") return None train_col, val_col = _find_metric_columns(run_dfs[0], metric) col = val_col or train_col if col is None: settings.logger.warning(f"plot_run_trajectories: metric '{metric}' not found in run data.") return None color = settings.THEME["val"] if val_col else settings.THEME["train"] all_values = [] for i, run_df in enumerate(run_dfs): if col not in run_df.columns: continue epochs = ( run_df["epoch"].values if "epoch" in run_df.columns else np.arange(1, len(run_df) + 1) ) values = run_df[col].values ax.plot( epochs, values, alpha=0.25, color=color, linewidth=1, label="Individual runs" if i == 0 else "_nolegend_", ) all_values.append(values) if all_values: min_len = min(len(v) for v in all_values) mean_vals = np.mean([v[:min_len] for v in all_values], axis=0) ax.plot( np.arange(1, min_len + 1), mean_vals, color=color, linewidth=2.5, label="Mean", ) ax.set_xlabel("Epoch") ax.set_ylabel(metric) ax.set_title(f"Run Trajectories — {metric} (n={len(all_values)})") ax.legend() _apply_style(ax) return _finalize_plot(fig, show)
[docs] def plot_run_distribution( results: "VariabilityStudyResults", metric: Optional[str] = None, ax: Optional["Axes"] = None, show: Optional[bool] = None, ) -> Optional["Figure"]: """Violin plot of final metric values with individual run points overlaid. Combines a KDE violin with individual run dots. Useful for understanding the shape of the performance distribution. Args: results: A completed :class:`~ictonyx.runners.VariabilityStudyResults`. metric: Metric key. Resolved via ``preferred_metric()`` if ``None``. ax: Optional existing Axes. show: Display behavior. See :func:`plot_confusion_matrix`. Returns: The figure, or ``None`` on error or when display is active. """ _check_plotting() if ax is None: fig, ax = plt.subplots(figsize=settings.get_figsize((5, 6)), dpi=150) else: fig = ax.figure # type: ignore[assignment] resolved = metric or results.preferred_metric() try: values = pd.Series(results.get_metric_values(resolved), name="value") except KeyError as e: settings.logger.warning(f"plot_run_distribution: {e}") return None df_plot = values.to_frame() df_plot["metric"] = resolved sns.violinplot( data=df_plot, y="value", ax=ax, color=settings.THEME["val"], inner=None, cut=0, ) rng = np.random.default_rng(42) x_jitter = rng.uniform(-0.08, 0.08, size=len(values)) ax.scatter( x_jitter, values, color=settings.THEME["point"], alpha=0.7, zorder=5, s=25, ) ax.set_ylabel(resolved) ax.set_title(f"Distribution — {resolved} (n={len(values)})") ax.set_xticks([]) _apply_style(ax, grid=False) return _finalize_plot(fig, show)
[docs] def plot_run_strip( results: "VariabilityStudyResults", metric: Optional[str] = None, ax: Optional["Axes"] = None, show: Optional[bool] = None, ) -> Optional["Figure"]: """Strip plot of final metric values with mean and SD markers. Best for small n (< 20) where a violin would be oversmoothed. Each run is one dot; the mean and ±1 SD are marked with lines. Args: results: A completed :class:`~ictonyx.runners.VariabilityStudyResults`. metric: Metric key. Resolved via ``preferred_metric()`` if ``None``. ax: Optional existing Axes. show: Display behavior. See :func:`plot_confusion_matrix`. Returns: The figure, or ``None`` on error or when display is active. """ _check_plotting() if ax is None: fig, ax = plt.subplots(figsize=settings.get_figsize((4, 6)), dpi=150) else: fig = ax.figure # type: ignore[assignment] resolved = metric or results.preferred_metric() try: values = np.array(results.get_metric_values(resolved)) except KeyError as e: settings.logger.warning(f"plot_run_strip: {e}") return None mean = np.mean(values) sd = np.std(values, ddof=1) rng = np.random.default_rng(42) x_jitter = rng.uniform(-0.18, 0.18, size=len(values)) ax.scatter(x_jitter, values, color=settings.THEME["val"], alpha=0.8, s=35, zorder=5) ax.hlines( mean, -0.35, 0.35, colors=settings.THEME["significant"], linewidth=2.5, label=f"Mean: {mean:.4f}", ) ax.hlines( [mean - sd, mean + sd], -0.28, 0.28, colors=settings.THEME["baseline"], linewidth=1.5, linestyle="--", label=f"±1 SD: {sd:.4f}", ) ax.set_ylabel(resolved) ax.set_title(f"Run Results — {resolved} (n={len(values)})") ax.set_xlim(-0.5, 0.5) ax.set_xticks([]) ax.legend(loc="best", fontsize=9) _apply_style(ax) return _finalize_plot(fig, show)
[docs] def plot_rank_correlation_over_epoch( results: "VariabilityStudyResults", metric: Optional[str] = None, threshold: float = 0.8, window: int = 5, ax: Optional["Axes"] = None, show: Optional[bool] = None, ) -> Optional["Figure"]: """Spearman rank correlation between epoch-k and final-epoch rankings. For each epoch k, computes the Spearman correlation between the per-run metric ranking at epoch k and the per-run ranking at the final epoch. Plotted over epochs with a rolling mean smoother. The first epoch where the smoothed correlation reaches ``threshold`` is annotated. This answers: at what point in training can you predict which seed will produce the best final model? Args: results: A completed :class:`~ictonyx.runners.VariabilityStudyResults` with per-epoch data in ``all_runs_metrics``. metric: Metric column to rank runs by. If ``None``, resolved via ``preferred_metric()``. Should be a validation metric. threshold: Smoothed correlation value at which to draw the annotation line. Default 0.8. window: Rolling mean window size for smoothing. Default 5. ax: Optional existing Axes. show: Display behavior. See :func:`plot_confusion_matrix`. Returns: The figure, or ``None`` on error or when display is active. Raises: ValueError: If ``n_runs < 15`` — rank correlation is unreliable with fewer runs. """ _check_plotting() n_runs = results.n_runs if n_runs < 15: raise ValueError( f"plot_rank_correlation_over_epoch() requires n_runs >= 15 for " f"reliable rank correlation. Got n_runs={n_runs}. Run at least " f"15 seeds to use this plot." ) resolved = metric or results.preferred_metric(context="epoch") run_dfs = results.all_runs_metrics if not run_dfs: settings.logger.warning("plot_rank_correlation_over_epoch: no run data available.") return None train_col, val_col = _find_metric_columns(run_dfs[0], resolved) col = val_col or train_col if col is None: settings.logger.warning( f"plot_rank_correlation_over_epoch: metric '{resolved}' " "not found in run data." ) return None # Build aligned matrix: shape (n_runs, n_epochs) # Use minimum epoch count across all runs to align series = [df[col].values for df in run_dfs if col in df.columns] if len(series) < 15: raise ValueError( f"plot_rank_correlation_over_epoch() requires at least 15 runs " f"with valid data for '{col}'. Found {len(series)}." ) min_epochs = min(len(s) for s in series) if min_epochs < 2: settings.logger.warning("plot_rank_correlation_over_epoch: fewer than 2 epochs available.") return None matrix = np.array([s[:min_epochs] for s in series]) # (n_runs, n_epochs) # Final-epoch rankings (higher metric = better rank = lower rank number) from scipy.stats import rankdata, spearmanr final_ranks = rankdata(-matrix[:, -1]) correlations = [] for epoch_idx in range(min_epochs): epoch_ranks = rankdata(-matrix[:, epoch_idx]) rho, _ = spearmanr(epoch_ranks, final_ranks) correlations.append(float(rho)) epochs = np.arange(1, min_epochs + 1) corr_series = pd.Series(correlations, index=epochs) smoothed = corr_series.rolling(window, center=True, min_periods=1).mean() # Find first epoch where smoothed correlation >= threshold threshold_epoch = None for ep, val in zip(epochs, smoothed.values): if val >= threshold: threshold_epoch = ep break if ax is None: fig, ax = plt.subplots(figsize=settings.get_figsize((10, 6)), dpi=150) else: fig = ax.figure # type: ignore[assignment] # Raw correlation as faint line ax.plot( epochs, correlations, color=settings.THEME["val"], alpha=0.35, linewidth=1, label="Raw Spearman ρ", ) # Smoothed correlation as main line ax.plot( epochs, smoothed.values, color=settings.THEME["val"], linewidth=2.5, label=f"Smoothed (window={window})", ) # Threshold reference line ax.axhline( threshold, color=settings.THEME["neutral"], linestyle="--", linewidth=1.2, label=f"Threshold ρ = {threshold}", ) # Annotation: first epoch at threshold if threshold_epoch is not None: ax.axvline( threshold_epoch, color=settings.THEME["significant"], linestyle=":", linewidth=1.5 ) ax.annotate( f"ρ ≥ {threshold}\nat epoch {threshold_epoch}", xy=(threshold_epoch, threshold), xytext=(threshold_epoch + max(1, min_epochs * 0.05), threshold - 0.08), fontsize=9, color=settings.THEME["significant"], arrowprops=dict( arrowstyle="->", color=settings.THEME["significant"], lw=1.2, ), ) ax.set_xlabel("Epoch") ax.set_ylabel("Spearman ρ (epoch-k vs final ranking)") ax.set_title( f"Rank Correlation Over Training — {resolved}\n" f"(n={len(series)} runs; higher = final ranking predictable earlier)" ) ax.set_ylim(-0.1, 1.05) ax.legend(loc="lower right") _apply_style(ax) return _finalize_plot(fig, show)
[docs] def plot_comparison_boxplots( comparison_results: Dict[str, Any], metric: str = "Accuracy", ax: Optional["Axes"] = None, show: Optional[bool] = None, ) -> Optional["Figure"]: """Side-by-side boxplots comparing metric distributions across models. Each model's runs are shown as a boxplot with individual data points overlaid. If pairwise statistical comparisons are present in the input, significance annotations are added to the title. Args: comparison_results: Either a dict returned by :func:`~ictonyx.api.compare_models` (must contain a ``'raw_data'`` key), or a plain dict mapping model names to lists/arrays of metric values. metric: Label for the y-axis (default ``'Accuracy'``). show: Display behavior. See :func:`plot_confusion_matrix`. Returns: The ``matplotlib.figure.Figure``, or ``None`` if the raw data cannot be extracted. """ _check_plotting() if hasattr(comparison_results, "raw_data"): data_dict = comparison_results.raw_data elif isinstance(comparison_results, dict): data_dict = comparison_results else: settings.logger.error("Could not find raw data for boxplots.") return None records = [] for model_name, scores in data_dict.items(): for score in scores: records.append({"Model": model_name, metric: score}) df = pd.DataFrame(records) if ax is None: fig, ax = plt.subplots(figsize=settings.get_figsize((10, 6)), dpi=150) else: fig = ax.figure # type: ignore[assignment] # Use themed palette if possible, else default sns.boxplot(data=df, x="Model", y=metric, hue="Model", palette="Blues", legend=False, ax=ax) sns.stripplot(data=df, x="Model", y=metric, color=settings.THEME["point"], alpha=0.3, ax=ax) # ── Significance bracket annotations ───────────────────────────────── if hasattr(comparison_results, "significant_comparisons"): sig_pairs = comparison_results.significant_comparisons or [] model_names_sorted = sorted(df["Model"].unique()) y_max = df[metric].max() y_range = df[metric].max() - df[metric].min() step = y_range * 0.08 for k, pair in enumerate(sig_pairs[:5]): # cap at 5 to avoid overlap parts = pair.split("_vs_", maxsplit=1) if len(parts) != 2: continue m1, m2 = parts if m1 not in model_names_sorted or m2 not in model_names_sorted: continue x1 = model_names_sorted.index(m1) x2 = model_names_sorted.index(m2) y = y_max + step * (k + 1) ax.plot( [x1, x1, x2, x2], [y - step * 0.2, y, y, y - step * 0.2], lw=1.2, color=settings.THEME["significant"], ) ax.text( (x1 + x2) / 2, y + step * 0.05, "*", ha="center", va="bottom", color=settings.THEME["significant"], fontsize=14, ) # Statistical Annotations Title if hasattr(comparison_results, "pairwise_comparisons"): pairs = comparison_results.pairwise_comparisons sig_pairs = [k for k, v in pairs.items() if v.is_significant()] if sig_pairs: subtitle = "Significant differences: " + ", ".join(sig_pairs) if len(subtitle) > 80: subtitle = subtitle[:77] + "..." ax.set_title(f"Model Comparison: {metric}\n{subtitle}", fontsize=10) else: ax.set_title( f"Model Comparison: {metric}\nNo significant differences found.", fontsize=10 ) return _finalize_plot(fig, show)
[docs] def plot_comparison_forest( comparison_results: Any, baseline_model: str, metric: str = "Accuracy", ax: Optional["Axes"] = None, show: Optional[bool] = None, ) -> Optional["Figure"]: """Forest plot of effect sizes relative to a baseline model. For each non-baseline model, shows the mean difference from the baseline with a 95% confidence interval, plotted as a horizontal error bar. A vertical dashed line at zero marks no difference. Args: comparison_results: Dict with a ``'raw_data'`` key mapping model names to arrays of metric values. baseline_model: Name of the model to use as the reference. Must be a key in ``raw_data``. metric: Label for the x-axis (default ``'Accuracy'``). show: Display behavior. See :func:`plot_confusion_matrix`. Returns: The ``matplotlib.figure.Figure``, or ``None`` if the baseline model is not found in the data. """ _check_plotting() if hasattr(comparison_results, "raw_data"): data_dict = comparison_results.raw_data elif isinstance(comparison_results, dict): data_dict = comparison_results.get("raw_data", comparison_results) else: settings.logger.error("Could not find raw data.") return None if baseline_model not in data_dict: settings.logger.error(f"Baseline model '{baseline_model}' not found in results.") return None baseline_scores = np.array(data_dict[baseline_model]) baseline_mean = np.mean(baseline_scores) models = [] diff_means = [] cis = [] colors = [] for name, scores in data_dict.items(): if name == baseline_model: continue scores = np.array(scores) diff = scores.mean() - baseline_mean # Extract pre-computed pairwise CI if available (paired, correct). pairwise_comps: Dict = {} if hasattr(comparison_results, "pairwise_comparisons"): pairwise_comps = comparison_results.pairwise_comparisons or {} elif isinstance(comparison_results, dict): pairwise_comps = comparison_results.get("pairwise_comparisons", {}) ci_half = None for key in (f"{name}_vs_{baseline_model}", f"{baseline_model}_vs_{name}"): result = pairwise_comps.get(key) ci = getattr(result, "confidence_interval", None) if ci is not None and isinstance(ci, (list, tuple)) and len(ci) == 2: lo, hi = ci ci_half = (hi - lo) / 2.0 break if ci_half is None: # Fallback: Welch unpaired. ~2.36x too wide for paired data (r≈0.85). se_diff = np.sqrt( np.var(scores, ddof=1) / len(scores) + np.var(baseline_scores, ddof=1) / len(baseline_scores) ) df_welch = len(scores) + len(baseline_scores) - 2 t_crit = _scipy_stats.t.ppf(0.975, df=df_welch) ci_half = t_crit * se_diff ci = ci_half models.append(name) diff_means.append(diff) cis.append(ci) if diff - ci > 0: colors.append(settings.THEME["test"]) # Better elif diff + ci < 0: colors.append(settings.THEME["significant"]) # Worse else: colors.append("gray") # Neutral if ax is None: fig, ax = plt.subplots(figsize=(8, len(models) * 0.8 + 2)) else: fig = ax.figure # type: ignore[assignment] y_pos = np.arange(len(models)) for i, (dm, yp, ci, col) in enumerate(zip(diff_means, y_pos, cis, colors)): ax.errorbar(dm, yp, xerr=ci, fmt="o", color="black", ecolor=col, capsize=5) ax.axvline(0, color=settings.THEME["baseline"], linestyle="--") ax.set_yticks(y_pos) ax.set_yticklabels(models) ax.set_xlabel(f"Difference in {metric} (vs {baseline_model})") ax.set_title(f"Model Performance vs Baseline ({baseline_model})") ax.grid(axis="x", linestyle=":", alpha=0.5) return _finalize_plot(fig, show)
[docs] def plot_pairwise_comparison_matrix( comparison_results: Dict[str, Any], figsize: Tuple[int, int] = (12, 10), show_effect_sizes: bool = True, annotate_significance: bool = True, show: Optional[bool] = None, ) -> Optional["Figure"]: """Heatmap matrix of pairwise p-values, significance, and effect sizes. Creates 2–3 side-by-side heatmaps showing p-values, binary significance indicators, and (optionally) effect sizes for every pair of models. Args: comparison_results: Dict returned by :func:`~ictonyx.api.compare_models`, containing a ``'pairwise_comparisons'`` key. figsize: Figure dimensions. Default ``(12, 10)``. show_effect_sizes: If ``True``, include an effect-size heatmap. Default ``True``. annotate_significance: If ``True``, mark significant cells with asterisks. Default ``True``. show: Display behavior. See :func:`plot_confusion_matrix`. Returns: The ``matplotlib.figure.Figure``, or ``None`` if no pairwise comparisons are present. """ _check_plotting() if not comparison_results: warnings.warn( "plot_pairwise_comparison_matrix: comparison_results is empty.", UserWarning, stacklevel=2, ) return None if hasattr(comparison_results, "pairwise_comparisons"): pairwise = comparison_results.pairwise_comparisons elif isinstance(comparison_results, dict): pairwise = comparison_results.get("pairwise_comparisons") else: pairwise = None if not pairwise: settings.logger.warning("No pairwise comparisons to plot.") return None _name_set: set = set() for comp_name in pairwise.keys(): names = comp_name.split("_vs_", maxsplit=1) _name_set.update(names) model_names = sorted(list(_name_set)) n_models = len(model_names) p_value_matrix = np.ones((n_models, n_models)) significance_matrix = np.zeros((n_models, n_models)) effect_size_matrix = np.zeros((n_models, n_models)) for comp_name, result in pairwise.items(): parts = comp_name.split("_vs_", maxsplit=1) if len(parts) != 2: continue name1, name2 = parts i, j = model_names.index(name1), model_names.index(name2) p_val = result.corrected_p_value if result.corrected_p_value is not None else result.p_value p_value_matrix[i, j] = p_val p_value_matrix[j, i] = p_val if result.is_significant(): significance_matrix[i, j] = 1 significance_matrix[j, i] = 1 if result.effect_size is not None: effect_size_matrix[i, j] = result.effect_size effect_size_matrix[j, i] = result.effect_size n_plots = 2 if not show_effect_sizes or not np.any(effect_size_matrix) else 3 fig, axes = plt.subplots(1, n_plots, figsize=figsize) if n_plots == 1: axes = [axes] def format_annotation(data, i, j): val = data[i, j] if i == j: return "" text = f"{val:.3f}" if annotate_significance and significance_matrix[i, j] == 1: if val < 0.001: text += "\n***" elif val < 0.01: text += "\n**" elif val < 0.05: text += "\n*" return text annot_matrix = np.array( [ [format_annotation(p_value_matrix, i, j) for j in range(n_models)] for i in range(n_models) ] ) mask = np.eye(n_models, dtype=bool) sns.heatmap( p_value_matrix, xticklabels=model_names, yticklabels=model_names, annot=annot_matrix, fmt="", cmap="RdYlGn_r", vmin=0, vmax=0.1, ax=axes[0], mask=mask, cbar_kws={"label": "P-value"}, linewidths=0.5, linecolor="gray", ) axes[0].set_title(f"Corrected P-values", fontsize=12, fontweight="bold") sig_annot = np.array( [ ["✓" if significance_matrix[i, j] and i != j else "" for j in range(n_models)] for i in range(n_models) ] ) sns.heatmap( significance_matrix, xticklabels=model_names, yticklabels=model_names, annot=sig_annot, fmt="", cmap="RdYlGn", vmin=0, vmax=1, ax=axes[1], mask=mask, cbar_kws={"label": "Significant", "ticks": [0, 1]}, linewidths=0.5, linecolor="gray", ) axes[1].set_title("Significant Differences (α = 0.05)", fontsize=12, fontweight="bold") if show_effect_sizes and np.any(effect_size_matrix): effect_annot = np.array( [ [f"{effect_size_matrix[i, j]:.3f}" if i != j else "" for j in range(n_models)] for i in range(n_models) ] ) sns.heatmap( np.abs(effect_size_matrix), xticklabels=model_names, yticklabels=model_names, annot=effect_annot, fmt="", cmap="YlOrRd", ax=axes[2], mask=mask, cbar_kws={"label": "Effect Size (absolute)"}, linewidths=0.5, linecolor="gray", ) axes[2].set_title("Effect Sizes", fontsize=12, fontweight="bold") plt.suptitle(f"Pairwise Comparison Matrix ({n_models} models)", fontsize=14, fontweight="bold") plt.tight_layout(rect=(0, 0, 1, 0.96)) return _finalize_plot(fig, show)
[docs] def plot_grid_study_heatmap( grid_results: Any, x_param: str, y_param: str, metric: Optional[str] = None, stat: str = "mean", show: Optional[bool] = None, ) -> Optional["Figure"]: """Heatmap of a grid study metric across two swept parameters. Args: grid_results: A :class:`~ictonyx.runners.GridStudyResults` object. x_param: Parameter name for the x-axis (columns of heatmap). y_param: Parameter name for the y-axis (rows of heatmap). metric: Metric to display. Defaults to ``grid_results.metric``. stat: Statistic to display — ``'mean'`` (default) or ``'sd'``. show: Display behavior. See :func:`plot_confusion_matrix`. Returns: The ``matplotlib.figure.Figure``. """ _check_plotting() df = grid_results.to_dataframe(metric=metric) if x_param not in df.columns: raise ValueError(f"x_param '{x_param}' not found in grid results.") if y_param not in df.columns: raise ValueError(f"y_param '{y_param}' not found in grid results.") if stat not in ("mean", "sd"): raise ValueError(f"stat must be 'mean' or 'sd', got '{stat}'.") pivot = df.pivot(index=y_param, columns=x_param, values=stat) fig, ax = plt.subplots( figsize=(max(6, len(pivot.columns) * 1.2), max(4, len(pivot.index) * 1.0)) ) cmap = "YlGnBu" if stat == "mean" else "Reds" im = ax.imshow(pivot.values, cmap=cmap, aspect="auto") plt.colorbar(im, ax=ax, label=stat) ax.set_xticks(range(len(pivot.columns))) ax.set_xticklabels([str(v) for v in pivot.columns]) ax.set_yticks(range(len(pivot.index))) ax.set_yticklabels([str(v) for v in pivot.index]) ax.set_xlabel(x_param) ax.set_ylabel(y_param) ax.set_title(f"Grid Study — {stat} {metric or grid_results.metric} " f"({x_param} × {y_param})") for i in range(len(pivot.index)): for j in range(len(pivot.columns)): val = pivot.values[i, j] if not np.isnan(val): ax.text( j, i, f"{val:.4f}", ha="center", va="center", fontsize=8, color="white" if val > pivot.values.max() * 0.7 else "black", ) fig.tight_layout() return _finalize_plot(fig, show)
[docs] def plot_training_stability( stability_results: Optional[Dict[str, Any]] = None, *, results: Optional["VariabilityStudyResults"] = None, metric: str = "loss", window_size: int = 10, figsize: Tuple[int, int] = (12, 8), show: Optional[bool] = None, ) -> Optional["Figure"]: """Plot training stability diagnostics from a stability analysis. Creates a multi-panel figure showing convergence behavior, loss distributions, and coefficient of variation across runs. Two ways to call this function: 1. **Direct**: pass a :class:`VariabilityStudyResults` via ``results=`` and the stability analysis will be computed for you:: ix.plot_training_stability(results=my_results) 2. **Precomputed**: pass a stability dict (useful when you want multiple views of the same analysis):: stability = ix.assess_training_stability( [df['loss'] for df in my_results.all_runs_metrics] ) ix.plot_training_stability(stability) Args: stability_results: Dict produced by :func:`~ictonyx.analysis.assess_training_stability`, containing keys such as ``'n_runs'``, ``'common_length'``, ``'final_loss_mean'``, ``'final_loss_std'``, ``'final_loss_cv'``, ``'stability_assessment'``, ``'converged_runs'``, and ``'final_losses_list'``. Mutually exclusive with ``results``. results: A :class:`VariabilityStudyResults` object. When provided, the stability analysis is computed internally using the specified ``metric`` column from each run's history. Mutually exclusive with ``stability_results``. metric: Name of the column in each run's history DataFrame to analyze. Default ``"loss"``. Only used when ``results`` is provided. window_size: Number of recent epochs for convergence statistics. Default ``10``. Only used when ``results`` is provided. figsize: Figure dimensions in inches. Default ``(12, 8)``. show: Display behavior. See :func:`plot_confusion_matrix`. Returns: The ``matplotlib.figure.Figure``, or ``None`` if display is enabled. Raises: ValueError: If both ``stability_results`` and ``results`` are provided, or if neither is. KeyError: If ``metric`` is not found in a run's history. """ _check_plotting() if results is not None: if stability_results is not None: raise ValueError("Pass either `results` or `stability_results`, not both.") from .analysis import assess_training_stability loss_histories = [pd.Series(run_df[metric]) for run_df in results.all_runs_metrics] stability_results = assess_training_stability(loss_histories, window_size=window_size) elif stability_results is None: raise ValueError( "plot_training_stability requires either `results` or `stability_results`." ) if "error" in stability_results: settings.logger.error(f"Cannot plot training stability: {stability_results['error']}") return None fig, axes = plt.subplots(2, 2, figsize=figsize) # 1. Final loss distribution ax1 = axes[0, 0] if "final_losses_list" in stability_results: final_losses = stability_results["final_losses_list"] mean_loss = stability_results["final_loss_mean"] std_loss = stability_results["final_loss_std"] n_runs = stability_results["n_runs"] ax1.hist( final_losses, bins=min(10, n_runs // 2 + 1), alpha=0.7, color="lightblue", edgecolor="black", ) ax1.axvline( mean_loss, color=settings.THEME["significant"], linestyle="--", label=f"Mean: {mean_loss:.4f}", ) ax1.axvline(mean_loss - std_loss, color=settings.THEME["val"], linestyle=":", alpha=0.7) ax1.axvline(mean_loss + std_loss, color=settings.THEME["val"], linestyle=":", alpha=0.7) ax1.legend() ax1.set_xlabel("Final Loss") ax1.set_ylabel("Frequency") ax1.set_title("Final Loss Distribution") # 2. Stability metrics ax2 = axes[0, 1] ax2.axis("off") stability_text = [ "Training Stability Metrics", "=" * 25, f"Runs analyzed: {stability_results['n_runs']}", f"Epochs per run: {stability_results['common_length']}", "", f"Final Loss:", f" Mean: {stability_results['final_loss_mean']:.4f}", f" Std: {stability_results['final_loss_std']:.4f}", f" CV: {stability_results['final_loss_cv']:.4f}", "", f"Stability: {stability_results['stability_assessment'].upper()}", f"Convergence rate: {stability_results['convergence_rate']:.1%}", ] ax2.text( 0.05, 0.95, "\n".join(stability_text), transform=ax2.transAxes, fontfamily="monospace", fontsize=10, va="top", ) # 3. Convergence status ax3 = axes[1, 0] converged = stability_results["converged_runs"] not_converged = stability_results["n_runs"] - converged ax3.pie( [converged, not_converged], labels=["Converged", "Not Converged"], colors=[settings.THEME["better"], settings.THEME["worse"]], autopct="%1.1f%%", startangle=90, ) ax3.set_title("Convergence Analysis") # 4. Stability level ax4 = axes[1, 1] stability_levels = ["High", "Moderate", "Low"] current_stability = stability_results["stability_assessment"].title() colors = [ settings.THEME["better"] if level == current_stability else settings.THEME["neutral"] for level in stability_levels ] bars = ax4.bar(stability_levels, [1, 1, 1], color=colors, alpha=0.7) if current_stability in stability_levels: bars[stability_levels.index(current_stability)].set_height(1.2) bars[stability_levels.index(current_stability)].set_alpha(1.0) ax4.set_ylabel("Stability Level") ax4.set_title(f"Overall Assessment: {current_stability}") ax4.set_ylim(0, 1.5) plt.tight_layout() return _finalize_plot(fig, show)
[docs] def plot_autocorr_vs_lag( data: Union[pd.Series, List[float]], max_lag: int = 20, title: str = "Autocorrelation of Loss", ax: Optional["Axes"] = None, show: Optional[bool] = None, ) -> Optional["Figure"]: """Plot autocorrelation of a metric series as a function of lag. Useful for diagnosing sequential dependence between runs — if autocorrelation is high at lag 1, consecutive runs may not be independent (e.g. due to incomplete GPU memory cleanup). Args: data: A ``pd.Series`` or list of metric values ordered by run. max_lag: Maximum lag to compute. Default 20. show: Display behavior. See :func:`plot_confusion_matrix`. Returns: The ``matplotlib.figure.Figure``, or ``None`` if display is enabled. """ _check_plotting() if not isinstance(data, pd.Series): data = pd.Series(data) if len(data) <= max_lag: warnings.warn( f"plot_autocorr_vs_lag(): data length ({len(data)}) is not greater " f"than max_lag ({max_lag}). Cannot compute autocorrelation. " "Provide a longer series or reduce max_lag.", UserWarning, stacklevel=2, ) return None autocorr_values = [data.autocorr(lag) for lag in range(1, max_lag + 1)] lags = range(1, max_lag + 1) if ax is None: fig, ax = plt.subplots(figsize=settings.get_figsize((10, 6)), dpi=150) else: fig = ax.figure # type: ignore[assignment] ax.stem(lags, autocorr_values) ax.set_title(title) ax.set_xlabel("Lag") ax.set_ylabel("Autocorrelation") ax.axhline(y=0, color=settings.THEME["significant"], linestyle="--") _apply_style(ax) return _finalize_plot(fig, show)
[docs] def plot_averaged_autocorr( lags: List[float], mean_autocorr: List[float], std_autocorr: List[float], title: str = "Averaged Autocorrelation of Loss", ax: Optional["Axes"] = None, show: Optional[bool] = None, ) -> Optional["Figure"]: """Plot averaged autocorrelation with error bands across multiple studies. Shows the mean autocorrelation at each lag with ±1 standard deviation shaded, summarizing sequential dependence across repeated studies. Args: lags: List of integer lag values. mean_autocorr: Mean autocorrelation at each lag. std_autocorr: Standard deviation of autocorrelation at each lag. show: Display behavior. See :func:`plot_confusion_matrix`. Returns: The ``matplotlib.figure.Figure``, or ``None`` if display is enabled. """ _check_plotting() if ax is None: fig, ax = plt.subplots(figsize=settings.get_figsize((10, 6)), dpi=150) else: fig = ax.figure # type: ignore[assignment] ax.plot(lags, mean_autocorr, color=settings.THEME["val"], label="Mean Autocorrelation") ax.fill_between( lags, np.array(mean_autocorr) - np.array(std_autocorr), np.array(mean_autocorr) + np.array(std_autocorr), color=settings.THEME["val"], alpha=0.2, label="Standard Deviation", ) ax.set_title(title) ax.set_xlabel("Lag") ax.set_ylabel("Autocorrelation") ax.axhline(y=0, color=settings.THEME["significant"], linestyle="--") ax.legend() _apply_style(ax) return _finalize_plot(fig, show)
[docs] def plot_pacf_vs_lag( data: Union[pd.Series, List[float]], max_lag: int = 20, title: str = "Partial Autocorrelation of Loss", alpha: float = 0.05, ax: Optional["Axes"] = None, show: Optional[bool] = None, ) -> Optional["Figure"]: """Plot partial autocorrelation function (PACF) with confidence bands. Requires ``statsmodels``. PACF isolates the direct correlation at each lag, removing the influence of intermediate lags. Args: data: A ``pd.Series`` or list of metric values ordered by run. max_lag: Maximum lag to compute. Default 20. title: Plot title. Default ``'Partial Autocorrelation of Loss'``. alpha: Confidence level for the bands. Default 0.05 (95% CI). show: Display behavior. See :func:`plot_confusion_matrix`. Returns: The ``matplotlib.figure.Figure``, or ``None`` if ``statsmodels`` is not installed or the series is too short. """ _check_plotting() try: from statsmodels.tsa.stattools import pacf except ImportError: settings.logger.warning("statsmodels required for PACF.") return None if not isinstance(data, pd.Series): data = pd.Series(data) if len(data) <= max_lag + 1: return None pacf_values, conf_int = pacf(data, nlags=max_lag, alpha=alpha) pacf_values = pacf_values[1:] conf_int = conf_int[1:] lags = range(1, max_lag + 1) if ax is None: fig, ax = plt.subplots(figsize=settings.get_figsize((10, 6)), dpi=150) else: fig = ax.figure # type: ignore[assignment] ax.stem(lags, pacf_values, label="PACF") # conf_int from statsmodels is already in absolute PACF value units. # Do NOT subtract pacf_values — that recenters the band at zero. ax.fill_between( lags, conf_int[:, 0], conf_int[:, 1], alpha=0.2, color=settings.THEME["neutral"], label=f"{int((1 - alpha) * 100)}% CI", ) ax.set_title(title) ax.legend() _apply_style(ax) return _finalize_plot(fig, show)
[docs] def plot_averaged_pacf( lags: List[float], mean_pacf: List[float], std_pacf: List[float], n_series: int = 0, title: str = "Averaged Partial Autocorrelation of Loss", conf_level: float = 0.95, show: Optional[bool] = None, ) -> Optional["Figure"]: """Plot averaged partial autocorrelation with error bands across runs. Shows the mean PACF at each lag with ±1 standard deviation shaded. Args: lags: List of integer lag values. mean_pacf: Mean PACF at each lag. std_pacf: Standard deviation of PACF at each lag. title: Plot title. Default ``'Averaged Partial Autocorrelation of Loss'``. conf_level: Confidence level for reference lines. Default 0.95. show: Display behavior. See :func:`plot_confusion_matrix`. Returns: The ``matplotlib.figure.Figure``, or ``None`` if display is enabled. deprecated:: ``plot_averaged_pacf`` is deprecated and will be removed in v0.5.0. Use ``plot_run_independence_diagnostics()`` (v0.4.4) instead. """ warnings.warn( "plot_averaged_pacf() is deprecated and will be removed in v0.5.0. " "Use plot_run_independence_diagnostics() (available in v0.4.4) instead.", DeprecationWarning, stacklevel=2, ) _check_plotting() fig, ax = plt.subplots(figsize=settings.get_figsize((10, 6)), dpi=150) ax.plot(lags, mean_pacf, color=settings.THEME["val"], label="Mean PACF", linewidth=2) ax.fill_between( lags, np.array(mean_pacf) - np.array(std_pacf), np.array(mean_pacf) + np.array(std_pacf), color=settings.THEME["val"], alpha=0.2, label="±1 Standard Deviation", ) if n_series > 0: z = _scipy_stats.norm.ppf((1 + conf_level) / 2) conf_bound = z / np.sqrt(n_series) ax.axhline( y=conf_bound, color=settings.THEME["neutral"], linestyle=":", alpha=0.7, label=f"{int(conf_level * 100)}% CI", ) ax.axhline(y=-conf_bound, color=settings.THEME["neutral"], linestyle=":", alpha=0.7) ax.set_title(title) ax.legend() _apply_style(ax) return _finalize_plot(fig, show)
[docs] def plot_paired_deltas( results_a: "VariabilityStudyResults", results_b: "VariabilityStudyResults", metric: Optional[str] = None, *, name_a: str = "Model A", name_b: str = "Model B", show_ci: bool = True, ci_confidence: float = 0.95, ax: Optional["Axes"] = None, figsize: Tuple[float, float] = (8, 6), show: Optional[bool] = None, ) -> Optional["Figure"]: """Per-run paired differences between two model studies. Shows each run's ``metric_a[i] - metric_b[i]`` as a point on a strip, with a horizontal zero line separating "Model A wins" (above) from "Model B wins" (below). When points appear on both sides of zero, the winner *reverses* depending on which seeds you picked — a phenomenon invisible from mean comparisons alone. Requires equal run counts in both studies (paired comparison is undefined otherwise). Args: results_a: First model's variability study. results_b: Second model's variability study. metric: Metric key to compare. If ``None``, resolved from ``results_a.preferred_metric(context="scalar")``. name_a: Label for model A in the legend and axes. Default ``"Model A"``. name_b: Label for model B. Default ``"Model B"``. show_ci: Whether to overlay a bootstrap CI band on the mean delta. Default ``True``. ci_confidence: Confidence level for the mean-delta CI. Default ``0.95``. ax: Optional existing Axes. Creates a new figure if ``None``. figsize: Figure dimensions when ``ax`` is ``None``. Default ``(8, 6)``. show: Display behavior. See :func:`plot_confusion_matrix`. Returns: The ``matplotlib.figure.Figure``, or ``None`` if display is enabled. Raises: ValueError: If the two studies have different run counts. """ _check_plotting() if metric is None: metric = results_a.preferred_metric(context="scalar") # Dispatch to the correct accessor based on metric scope def _fetch(res, m: str) -> np.ndarray: if m.startswith("test_"): # test_* metrics live in final_test_metrics; strip the prefix return np.asarray(res.get_test_metric_values(m[len("test_") :]), dtype=float) return np.asarray(res.get_metric_values(m), dtype=float) values_a = _fetch(results_a, metric) values_b = _fetch(results_b, metric) if len(values_a) != len(values_b): raise ValueError( f"Paired delta requires equal run counts; got " f"{len(values_a)} (A) and {len(values_b)} (B). " f"Use an unpaired comparison for unequal runs." ) deltas = values_a - values_b n = len(deltas) mean_delta = float(np.mean(deltas)) ci_lo, ci_hi = None, None if show_ci and n >= 3: from .bootstrap import bootstrap_ci try: ci = bootstrap_ci(deltas, np.mean, confidence=ci_confidence, method="bca") ci_lo, ci_hi = ci.ci_lower, ci.ci_upper except Exception: ci_lo, ci_hi = None, None if ax is None: fig, ax = plt.subplots(figsize=figsize) else: fig = ax.figure # type: ignore[assignment] run_indices = np.arange(1, n + 1) colors = [ settings.THEME.get("significant", "C0") if d >= 0 else settings.THEME.get("val", "C1") for d in deltas ] ax.scatter( run_indices, deltas, c=colors, s=60, zorder=3, edgecolor="black", linewidth=0.5, ) ax.axhline(0, color="black", linestyle="-", linewidth=1, alpha=0.6, zorder=1) ax.axhline( mean_delta, color=settings.THEME.get("significant", "C0"), linestyle="--", linewidth=1.2, label=f"Mean Δ = {mean_delta:.4f}", zorder=2, ) if ci_lo is not None and ci_hi is not None: ax.axhspan( ci_lo, ci_hi, color=settings.THEME.get("significant", "C0"), alpha=0.15, label=f"{int(ci_confidence * 100)}% CI: [{ci_lo:.4f}, {ci_hi:.4f}]", zorder=0, ) ax.set_xlabel("Run") ax.set_ylabel(f"{name_a}{name_b} ({metric})") ax.set_title(f"Paired Run Differences: {name_a} vs {name_b}") ax.set_xticks(run_indices) ax.legend(loc="best", fontsize=9) ax.grid(True, alpha=0.3, zorder=0) n_above = int(np.sum(deltas > 0)) n_below = int(np.sum(deltas < 0)) if n_above > 0 and n_below > 0: ax.text( 0.02, 0.02, f"Winner reverses: {n_above} favor {name_a}, {n_below} favor {name_b}", transform=ax.transAxes, fontsize=9, style="italic", verticalalignment="bottom", bbox=dict(boxstyle="round,pad=0.3", facecolor="white", alpha=0.8), ) return _finalize_plot(fig, show)
[docs] def plot_run_independence_diagnostics( results: "VariabilityStudyResults", metric: Optional[str] = None, max_lag: int = 5, *, alpha: float = 0.05, ax: Optional["Axes"] = None, figsize: Tuple[float, float] = (8, 5), show: Optional[bool] = None, ) -> Optional["Figure"]: """Autocorrelation diagnostic for run-level independence. Runs from :func:`variability_study` should be IID by ``SeedSequence.spawn()`` construction. This plot helps detect violations — e.g. GPU state, wall-clock ordering, or caching leaking between runs — by showing autocorrelation at lags ``1..max_lag`` of the run-indexed final-metric series. Uses :func:`~ictonyx.analysis.check_independence` internally, which applies a Bonferroni correction across the tested lags to control the familywise false-positive rate at ``alpha``. **At small N the test has low power.** With ``n=10`` runs and ``max_lag=5``, the plot bands are wide; absence of a flagged lag is not strong evidence of independence. Consider ``runs >= 20`` before reading too much into a clean diagnostic. Args: results: A completed :class:`VariabilityStudyResults`. metric: Metric key to diagnose. If ``None``, resolved via ``results.preferred_metric(context="scalar")``. max_lag: Maximum lag to test. Default ``5``. Effective lag may be reduced if the run count is small. alpha: Familywise significance level. Default ``0.05``. ax: Optional existing Axes. Creates a new figure if ``None``. figsize: Figure dimensions when ``ax`` is ``None``. Default ``(8, 5)``. show: Display behavior. See :func:`plot_confusion_matrix`. Returns: The ``matplotlib.figure.Figure``, or ``None`` if display is enabled. """ _check_plotting() from .analysis import check_independence if metric is None: metric = results.preferred_metric(context="scalar") # Fetch run-indexed series (reusing the dispatch logic from paired deltas) if metric.startswith("test_"): values = np.asarray(results.get_test_metric_values(metric[len("test_") :]), dtype=float) else: values = np.asarray(results.get_metric_values(metric), dtype=float) series = pd.Series(values) is_independent, details = check_independence(series, max_lag=max_lag, alpha=alpha) if ax is None: fig, ax = plt.subplots(figsize=figsize) else: fig = ax.figure # type: ignore[assignment] # Handle untestable case (series too short) if is_independent is None: ax.text( 0.5, 0.5, details.get("error", "Series too short to assess independence."), ha="center", va="center", transform=ax.transAxes, fontsize=10, wrap=True, ) ax.set_xticks([]) ax.set_yticks([]) ax.set_title("Run Independence Diagnostic (untestable)") return _finalize_plot(fig, show) # Extract pre-computed per-lag ACFs and thresholds from check_independence autocorrs = details["autocorrelations"] # dict "lag_1" -> float thresholds = details["thresholds"] # dict "lag_1" -> float (±threshold) significant_lags = set(details["significant_lags"]) lags = sorted(int(k.split("_")[1]) for k in autocorrs.keys()) acfs = [autocorrs[f"lag_{k}"] for k in lags] thrs = [thresholds[f"lag_{k}"] for k in lags] # Stem plot: one vertical line per lag from 0 to ACF value color_sig = settings.THEME.get("significant", "C3") color_ok = settings.THEME.get("val", "C0") bar_colors = [color_sig if lag in significant_lags else color_ok for lag in lags] ax.vlines(lags, 0, acfs, colors=bar_colors, linewidth=2.5) ax.scatter(lags, acfs, c=bar_colors, s=60, zorder=3, edgecolor="black", linewidth=0.5) # Zero line ax.axhline(0, color="black", linewidth=0.8, alpha=0.5, zorder=1) # Per-lag Bonferroni-adjusted critical value bands (stepped, not flat) ax.plot( lags, thrs, color="gray", linestyle="--", linewidth=1, label=f"Bonferroni α={alpha} threshold", alpha=0.7, ) ax.plot(lags, [-t for t in thrs], color="gray", linestyle="--", linewidth=1, alpha=0.7) # Labels ax.set_xlabel("Lag") ax.set_ylabel(f"Autocorrelation ({metric})") ax.set_xticks(lags) # Title reflects the result n = details.get("n", len(values)) if is_independent: title = f"Run Independence: OK (n={n}, no lags exceed threshold)" else: flagged = sorted(significant_lags) title = ( f"Run Independence: FLAGGED (n={n}, " f"lag{'s' if len(flagged) > 1 else ''} " f"{', '.join(str(l) for l in flagged)} exceed threshold)" ) ax.set_title(title, fontsize=10) ax.legend(loc="best", fontsize=9) ax.grid(True, alpha=0.3, zorder=0) return _finalize_plot(fig, show)
[docs] def plot_epoch_run_heatmap( results: "VariabilityStudyResults", metric: Optional[str] = None, *, cmap: str = "viridis", ax: Optional["Axes"] = None, figsize: Optional[Tuple[float, float]] = None, show: Optional[bool] = None, ) -> Optional["Figure"]: """Epoch × run heatmap of a per-epoch metric. Rows are runs in run order (top = run 1). Columns are epochs. Cells are the metric value, colored by the supplied ``cmap``. Reveals at a glance which runs train fast, which plateau early, which diverge, and how run-to-run consistency evolves over training. Runs with fewer recorded epochs (early stopping, crashes, varied grid configs) have trailing NaN cells, which render as background. This is intentional — padding with zero or extrapolating would mislead. Uses the epoch-context metric (:func:`preferred_metric` with ``context="epoch"``) because per-epoch plotters show training dynamics, and test-set metrics are single scalars not captured across epochs. Args: results: A completed :class:`VariabilityStudyResults`. metric: Metric to plot. If ``None``, resolved via ``results.preferred_metric(context="epoch")``. cmap: Matplotlib colormap name. Default ``"viridis"``. ax: Optional existing Axes. Creates a new figure if ``None``. figsize: Figure dimensions. If ``None``, auto-sized from the matrix shape. show: Display behavior. See :func:`plot_confusion_matrix`. Returns: The ``matplotlib.figure.Figure``, or ``None`` if the metric is not found in any run or display is enabled. """ _check_plotting() if metric is None: metric = results.preferred_metric(context="epoch") run_dfs = results.all_runs_metrics if not run_dfs: settings.logger.warning("plot_epoch_run_heatmap: no run data available.") return None # Resolve column using the standard helper; prefer val over train train_col, val_col = _find_metric_columns(run_dfs[0], metric) col = val_col or train_col if col is None: settings.logger.warning(f"plot_epoch_run_heatmap: metric '{metric}' not found in run data.") return None # Collect per-run value arrays, skipping runs missing the column per_run_values: List[np.ndarray] = [] for run_df in run_dfs: if col not in run_df.columns: continue per_run_values.append(np.asarray(run_df[col].values, dtype=float)) if not per_run_values: settings.logger.warning(f"plot_epoch_run_heatmap: metric '{metric}' found in no run.") return None # Build a (n_runs, max_epochs) matrix padded with NaN for ragged runs n_runs = len(per_run_values) max_epochs = max(len(v) for v in per_run_values) matrix = np.full((n_runs, max_epochs), np.nan, dtype=float) for i, vals in enumerate(per_run_values): matrix[i, : len(vals)] = vals # Auto-figsize if not provided if figsize is None: figsize = ( max(6.0, min(16.0, max_epochs * 0.25)), max(4.0, min(12.0, n_runs * 0.3)), ) if ax is None: fig, ax = plt.subplots(figsize=figsize) else: fig = ax.figure # type: ignore[assignment] # Render via seaborn; mask NaNs so trailing cells in short runs render # as background rather than color-scale extreme sns.heatmap( matrix, ax=ax, cmap=cmap, cbar_kws={"label": col}, mask=np.isnan(matrix), linewidths=0, ) # Sparsify epoch tick labels when epoch count is large epoch_step = max(1, max_epochs // 20) epoch_ticks = np.arange(0, max_epochs, epoch_step) ax.set_xticks(epoch_ticks + 0.5) ax.set_xticklabels([str(e + 1) for e in epoch_ticks], rotation=0) # Y-tick labels: Run 1..N, but sparsify when run count is large run_step = max(1, n_runs // 20) run_ticks = np.arange(0, n_runs, run_step) ax.set_yticks(run_ticks + 0.5) ax.set_yticklabels([f"Run {r + 1}" for r in run_ticks], rotation=0) ax.set_xlabel("Epoch") ax.set_ylabel("Run") ax.set_title(f"Per-Epoch {col} Across {n_runs} Runs") return _finalize_plot(fig, show)
[docs] def plot_sequential_ci( results: "VariabilityStudyResults", metric: Optional[str] = None, *, min_n: int = 3, method: str = "bca", confidence: float = 0.95, ax: Optional["Axes"] = None, figsize: Tuple[float, float] = (8, 5), show: Optional[bool] = None, ) -> Optional["Figure"]: """Bootstrap CI width as a function of number of runs. For each ``n`` in ``[min_n, N]``, computes a bootstrap CI for the mean of the first ``n`` runs and plots the upper/lower bounds along with the running mean. The narrowing band illustrates how quickly additional runs reduce uncertainty and where returns diminish. Useful for the "how many runs do I need?" methodology argument. Look for the point where the band stops meaningfully narrowing — additional runs past that point buy little precision. Args: results: A completed :class:`VariabilityStudyResults`. metric: Metric key. If ``None``, resolved via ``results.preferred_metric(context="scalar")``. min_n: Smallest N at which to compute a CI. Floored at 3 because bootstrap CIs are meaningless with fewer observations. Default ``3``. method: Bootstrap CI method. ``"bca"`` (default) or ``"percentile"``. confidence: Confidence level in (0, 1). Default ``0.95``. ax: Optional existing Axes. Creates a new figure if ``None``. figsize: Figure dimensions when ``ax`` is ``None``. Default ``(8, 5)``. show: Display behavior. See :func:`plot_confusion_matrix`. Returns: The ``matplotlib.figure.Figure``, or ``None`` if display is enabled or fewer than ``min_n`` runs are available. Raises: ValueError: If ``min_n < 3`` or if the study has fewer than ``min_n`` runs. """ _check_plotting() from .bootstrap import bootstrap_ci if min_n < 3: raise ValueError(f"min_n must be >= 3 for meaningful bootstrap CIs; got {min_n}") if metric is None: metric = results.preferred_metric(context="scalar") # Fetch the scalar run-indexed series (same dispatch as plot_paired_deltas) if metric.startswith("test_"): values = np.asarray(results.get_test_metric_values(metric[len("test_") :]), dtype=float) else: values = np.asarray(results.get_metric_values(metric), dtype=float) n_total = len(values) if n_total < min_n: raise ValueError(f"Sequential CI requires at least min_n={min_n} runs; " f"got {n_total}.") # Compute running CI at each N ns: List[int] = [] means: List[float] = [] lowers: List[float] = [] uppers: List[float] = [] for n in range(min_n, n_total + 1): subset = values[:n] try: ci = bootstrap_ci(subset, np.mean, confidence=confidence, method=method) ns.append(n) means.append(float(np.mean(subset))) lowers.append(ci.ci_lower) uppers.append(ci.ci_upper) except Exception: # Catastrophic bootstrap failure at this N — record a gap ns.append(n) means.append(float(np.mean(subset))) lowers.append(np.nan) uppers.append(np.nan) if ax is None: fig, ax = plt.subplots(figsize=figsize) else: fig = ax.figure # type: ignore[assignment] color_mean = settings.THEME.get("val", "C0") color_band = settings.THEME.get("significant", "C1") # Shaded CI band ax.fill_between( ns, lowers, uppers, color=color_band, alpha=0.2, label=f"{int(confidence * 100)}% CI ({method})", ) # CI bound lines (thin, for emphasis at edges) ax.plot(ns, lowers, color=color_band, linewidth=0.8, alpha=0.6) ax.plot(ns, uppers, color=color_band, linewidth=0.8, alpha=0.6) # Running mean line ax.plot(ns, means, color=color_mean, linewidth=2, label="Running mean") # Annotate final CI width final_width = uppers[-1] - lowers[-1] if not np.isnan(uppers[-1]) else np.nan if not np.isnan(final_width): ax.text( 0.98, 0.02, f"CI width at N={n_total}: {final_width:.4f}", transform=ax.transAxes, fontsize=9, horizontalalignment="right", verticalalignment="bottom", bbox=dict(boxstyle="round,pad=0.3", facecolor="white", alpha=0.8), ) ax.set_xlabel("Number of runs (N)") ax.set_ylabel(metric) ax.set_title(f"Sequential Bootstrap CI: {metric}") ax.set_xticks(ns if len(ns) <= 20 else ns[:: max(1, len(ns) // 20)]) ax.legend(loc="best", fontsize=9) _apply_style(ax) return _finalize_plot(fig, show)