Pipeline

---

pipeline

Main entry point module for the medpipe machine learning package.

Provides unified, high-level interfaces (MedpipeClassifier for binary classification outcomes, MedpipeRegressor for regression outcomes) orchestrating data preparation, model fitting, inference, and TRIPOD+AI compliant evaluation.

MedpipeClassifier

User entry point and high-level pipeline runner for MedpipeClassifier.

Coordinates the complete machine learning lifecycle, delegating data ingress and preprocessing setup to MedpipeOrchestrator, model cross-validation and fitting to MedpipeClassifierRunner, and prediction and TRIPOD+AI evaluation to MedpipeClassifierEvaluator.

Parameters:

Name	Type	Description	Default
config	str, Path, or MedpipeClassifierConfig	Path to the TOML configuration file or an instantiated MedpipeClassifierConfig object.	required
base_artifact_dir	str or Path	Root directory where versioned execution run artifacts, logs, and models are stored.	"artifacts"
verbose_override	Union[bool, int, str, None]	Console verbosity setting configuration override.	None

Attributes:

Name	Type	Description
mp_config	MedpipeClassifierConfig	Validated configuration driving the pipeline.
displayer	MedpipeClassifierDisplayer	Visualisation engine rendering and persisting evaluation figures.

Source code in medpipe/pipeline/pipeline.py

class MedpipeClassifier:
    """
    User entry point and high-level pipeline runner for MedpipeClassifier.

    Coordinates the complete machine learning lifecycle, delegating data ingress
    and preprocessing setup to `MedpipeOrchestrator`, model cross-validation
    and fitting to `MedpipeClassifierRunner`, and prediction and TRIPOD+AI evaluation to
    `MedpipeClassifierEvaluator`.

    Parameters
    ----------
    config : str, Path, or MedpipeClassifierConfig
        Path to the TOML configuration file or an instantiated
        MedpipeClassifierConfig object.
    base_artifact_dir : str or Path, default="artifacts"
        Root directory where versioned execution run artifacts, logs, and
        models are stored.
    verbose_override : Union[bool, int, str, None], default=None
        Console verbosity setting configuration override.

    Attributes
    ----------
    mp_config : MedpipeClassifierConfig
        Validated configuration driving the pipeline.
    displayer : MedpipeClassifierDisplayer
        Visualisation engine rendering and persisting evaluation figures.

    """

    def __init__(
        self,
        config: str | Path | MedpipeClassifierConfig,
        base_artifact_dir: str | Path = "artifacts",
        verbose_override: bool | int | str | None = None,
    ) -> None:
        self._logger = get_console_logger("medpipe")

        self._logger.info("Initialising MedpipeClassifier end-to-end pipeline.")

        self._orchestrator = MedpipeOrchestrator(
            config, base_artifact_dir, verbose_override
        )
        self.mp_config = self._orchestrator.config
        self._runner = MedpipeClassifierRunner(orchestrator=self._orchestrator)
        self._evaluator = MedpipeClassifierEvaluator(
            orchestrator=self._orchestrator,
            runner=self._runner,
        )
        self.displayer = MedpipeClassifierDisplayer(orchestrator=self._orchestrator)

        self._logger.info("MedpipeClassifier initialisation complete.")

    @property
    def models(self) -> dict[str, Pipeline | CalibratedClassifierCV]:
        """Return the fitted models.

        Returns
        -------
        fitted_models : Dict[str, Union[Pipeline, CalibratedClassifierCV]]
            Fitted models from the MedpipeClassifierRunner object.

        """
        return self._runner.fitted_models

    @property
    def is_fitted(self) -> bool:
        """Checks if the MedpipeClassifier is fitted by looking at
        MedpipeClassifierRunner.

        Returns
        -------
        fitted : bool
            True if the models have been fitted, False otherwise.

        """
        return self._runner.fitted_models != {}

    @property
    def run_dir(self) -> Path:
        """Returns the run_dir from the MedpipeOrchestrator object.

        Returns
        -------
        run_dir : Path
            Path to the run_dir.

        """
        return self._orchestrator.run_dir

    @property
    def data_split(self) -> DataSplits:
        """Access any of the data splits.

        Return
        ------
        DataSplits
            Any of the DataSplits attributes.

        """
        return self._orchestrator.splits

    @property
    def fairness_split(self) -> FairnessSplits | None:
        """Access fairness stratification columns aligned with each data
        split.

        Lets fairness analyses be run directly against the relevant
        columns (e.g. `pipeline.fairness_split.test["HOSPITAL"]") without
        re-extracting them from the raw dataset, and independently of
        whether those columns are also model predictors.

        Return
        ------
        FairnessSplits or None
            Any of the FairnessSplits attributes, or None if no
            `workflow.evaluation.fairness` configuration is set.

        """
        return self._orchestrator.fairness_splits

    @property
    def prepare_data(self) -> Callable[..., tuple]:
        """Access the orchestrator's data preparation function.

        Ingests the configured dataset and creates the train/recal/test
        splits, e.g. `X_train, y_train, X_recal, y_recal, X_test, y_test,
        groups_train = pipeline.prepare_data()`. Keyword arguments are passed
        to the underlying Pandas reader function.

        Returns
        -------
        Callable
            The bound `MedpipeOrchestrator.prepare_data` method.

        """
        return self._orchestrator.prepare_data

    @property
    def metrics(self) -> list[str]:
        """Returns metrics used during evaluation.

        Returns
        -------
        metrics : list[str]
            List of metrics used during evaluation.

        """
        return self._evaluator.metrics

    def fit(
        self,
        X_train: pd.DataFrame,
        y_train: pd.DataFrame,
        X_recal: pd.DataFrame | None = None,
        y_recal: pd.DataFrame | None = None,
        groups_train: np.ndarray | None = None,
    ) -> dict[str, Any]:
        """
        Fit machine learning models across configured target outcomes via
        MedpipeClassifierRunner.

        Parameters
        ----------
        X_train : pandas.DataFrame
            Training feature dataset.
        y_train : pandas.DataFrame
            Training outcome target labels (can contain multiple outcome columns).
        X_recal : pandas.DataFrame, optional
            Recalibration feature dataset for post-hoc model calibration, default=None.
        y_recal : pandas.DataFrame, optional
            Recalibration outcome target labels, default=None.
        groups_train : numpy.ndarray, optional
            Group identifier array for group-based cross-validation splits,
            default=None.

        Returns
        -------
        fitted_models : dict of str to object
            Dictionary mapping outcome target names to their finalized fitted models.

        """
        self._logger.info("Starting model fitting across configured target outcomes.")
        fitted_models = self._runner.run(
            X_train=X_train,
            y_train_df=y_train,
            X_recal=X_recal,
            y_recal_df=y_recal,
            groups_train=groups_train,
        )
        self._logger.info("Completed model fitting for all target outcomes.")
        return fitted_models

    def predict(
        self,
        X: pd.DataFrame | npt.NDArray,
        model: Any | None = None,
        outcome: str | None = None,
    ) -> npt.NDArray:
        """
        Predict class labels for input samples.

        Parameters
        ----------
        X : pandas.DataFrame or numpy.ndarray
            Features dataset of shape (n_samples, n_features).
        model : object, optional
            Fitted model instance. If None, resolved via `outcome` or
            `runner.fitted_models`.
        outcome : str, optional
            Outcome name corresponding to an entry in `runner.fitted_models`.

        Returns
        -------
        y_pred : numpy.ndarray
            Predicted class labels of shape (n_samples,).

        """
        return self._evaluator.predict(X=X, model=model, outcome=outcome)

    def predict_proba(
        self,
        X: pd.DataFrame | npt.NDArray,
        model: Any | None = None,
        outcome: str | None = None,
    ) -> npt.NDArray:
        """
        Predict class probabilities for input samples.

        Parameters
        ----------
        X : pandas.DataFrame or numpy.ndarray
            Features dataset of shape (n_samples, n_features).
        model : object, optional
            Fitted model instance. If None, resolved via `outcome` or
            `runner.fitted_models`.
        outcome : str, optional
            Outcome name corresponding to an entry in `runner.fitted_models`.

        Returns
        -------
        y_proba : numpy.ndarray
            Predicted class probabilities of shape (n_samples, n_classes) or
            (n_samples,).

        """
        return self._evaluator.predict_proba(X=X, model=model, outcome=outcome)

    def decision_function(
        self,
        X: pd.DataFrame | npt.NDArray,
        model: Any | None = None,
        outcome: str | None = None,
    ) -> npt.NDArray:
        """
        Compute decision function confidence scores for input samples.

        Parameters
        ----------
        X : pandas.DataFrame or numpy.ndarray
            Features dataset of shape (n_samples, n_features).
        model : object, optional
            Fitted model instance. If None, resolved via `outcome` or
            `runner.fitted_models`.
        outcome : str, optional
            Outcome name corresponding to an entry in `runner.fitted_models`.

        Returns
        -------
        scores : numpy.ndarray
            Confidence scores or decision function values of shape (n_samples,).

        """
        return self._evaluator.decision_function(X=X, model=model, outcome=outcome)

    def evaluate(
        self,
        X: pd.DataFrame,
        y: pd.Series | npt.NDArray,
        outcome: str | None = None,
        model: Any | None = None,
        metrics: list[str] | None = None,
        subgroup_specs: dict[str, str | Callable[[pd.DataFrame], pd.Series]]
        | None = None,
        fairness_data: pd.DataFrame | None = None,
        save_artifacts: bool = True,
    ) -> dict[str, Any]:
        """
        Evaluate model performance with confidence intervals on full datasets
        and subgroups.

        Parameters
        ----------
        X : pandas.DataFrame
            Feature dataset of shape (n_samples, n_features).
        y : pandas.Series or numpy.ndarray
            Ground truth target values of shape (n_samples,).
        outcome : str, optional
            Outcome key name. Used for logging, model resolution, and artifact
            saving.
        model : object, optional
            Explicit fitted model instance. If None, resolved from
            `runner.fitted_models`.
        metrics : list of str, optional
            List of metrics to evaluate. If None, defaults to
            `self._evaluator.metrics`.
        subgroup_specs : dict of str to (str or callable), optional
            Specifications for extracting demographic or clinical subgroups.
        fairness_data : pandas.DataFrame, optional
            Fairness stratification columns aligned with `X` (see
            `fairness_split`), used to resolve `subgroup_specs` for strata
            that are not themselves model predictors (e.g. "HOSPITAL").
            Defaults to `X` when not provided.
        save_artifacts : bool, default=True
            Whether to write evaluation summary results to disk via
            `ArtifactManager`.

        Returns
        -------
        results : dict of str to Any
            Nested dictionary containing outcome identifier, overall slice
            metrics, and subgroup performance.

        """
        # Resolve single target Series/ndarray if y is supplied as a DataFrame
        if isinstance(y, pd.DataFrame):
            if outcome and outcome in y.columns:
                y_eval = y[outcome]
            elif y.shape[1] == 1:
                y_eval = y.iloc[:, 0]
            else:
                y_eval = y
        else:
            y_eval = y

        return self._evaluator.evaluate(
            X=X,
            y=y_eval,
            outcome=outcome,
            model=model,
            metrics=metrics,
            subgroup_specs=subgroup_specs,
            fairness_data=fairness_data,
            save_artifacts=save_artifacts,
        )

    def run(
        self,
        groups_train: np.ndarray | None = None,
    ) -> dict[str, Any]:
        """
        Execute full end-to-end pipeline (data preparation, model fitting, and
        test evaluation).

        Automates data ingestion, split creation, model cross-validation and fitting,
        and test evaluation with TRIPOD+AI reporting across all target outcomes.
        The model fitting duration and total run duration are recorded at
        the debug log level.

        Parameters
        ----------
        groups_train : numpy.ndarray, optional
            Group labels for training samples if group-based cross-validation
            is configured.

        Returns
        -------
        pipeline_results : dict of str to Any
            Dictionary containing:
            - `"fitted_models"`: Dictionary mapping outcome target names
            to fitted model instances.
            - `"evaluations"`: Dictionary mapping outcome target names
            to evaluation results dictionaries.
            - `"plots"`: (Audit / Eval mode only) Nested dictionary mapping
            outcome target names to figure objects.

        """
        self._logger.info("Executing full MedpipeClassifier pipeline end-to-end.")
        run_start_time = time.perf_counter()

        run_mode = self.mp_config.meta.run_mode

        self._logger.debug(
            f"Executing full MedpipeClassifier pipeline in {run_mode} mode."
        )

        n_steps = 3
        if run_mode == "audit" or run_mode == "eval":
            n_steps = 4

        # 1. Prepare data splits via orchestrator
        data_kwargs = self.mp_config.data.kwargs  # Get extra data arguments
        self._logger.info(f"Step 1/{n_steps}: Ingesting and splitting dataset.")
        X_train, y_train, X_recal, y_recal, X_test, y_test, groups_train = (
            self._orchestrator.prepare_data(**data_kwargs)
        )

        # 2. Fit models via runner
        self._logger.info(f"Step 2/{n_steps}: Fitting outcome models.")
        fit_start_time = time.perf_counter()
        fitted_models = self.fit(
            X_train=X_train,
            y_train=y_train,
            X_recal=X_recal,
            y_recal=y_recal,
            groups_train=groups_train,
        )
        fit_duration = time.perf_counter() - fit_start_time
        self._logger.debug(f"Model fitting completed in {fit_duration:.2f} seconds.")

        # 3. Evaluate models on test set via evaluator
        self._logger.info(
            f"Step 3/{n_steps}: Evaluating models on holdout test set.",
        )
        evaluations: dict[str, Any] = {}
        plots: dict[str, dict[str, tuple[Figure | SubFigure, Axes]]] = {}
        outcomes = self._orchestrator.config.data.outcomes
        subgroup_specs = self._orchestrator.get_subgroup_specs()
        fairness_splits = self._orchestrator.fairness_splits
        fairness_data = fairness_splits.test if fairness_splits is not None else None

        for outcome in outcomes:
            y_test_outcome = y_test[outcome]
            evaluations[outcome] = self.evaluate(
                X=X_test,
                y=y_test_outcome.to_numpy(),
                outcome=outcome,
                subgroup_specs=subgroup_specs,
                fairness_data=fairness_data,
                save_artifacts=True,
            )

        if run_mode == "audit" or run_mode == "eval":
            self._logger.info(f"Step 4/{n_steps}: Plotting graphs.")
            for outcome in outcomes:
                y_true_outcome = y_test[outcome].to_numpy()
                probas_outcome = self.predict_proba(X=X_test, outcome=outcome)

                plots[outcome] = self.plot_all(
                    y_true=y_true_outcome,
                    probas=probas_outcome,
                    outcome=outcome,
                )
            # Generate cross-outcome strata heatmaps per metric
            if fairness_splits is None:
                self._logger.info(
                    "No fairness configuration set; skipping subgroup strata heatmaps."
                )
            else:
                self._logger.info(
                    "Generating subgroup strata heatmaps across outcomes."
                )
                strata_heatmaps = self.displayer.plot_all_heatmaps(
                    evaluations=evaluations,
                )
                plots["strata_heatmaps"] = strata_heatmaps

        run_duration = time.perf_counter() - run_start_time
        self._logger.debug(
            f"Full pipeline run completed in {run_duration:.2f} seconds."
        )
        self._logger.info(
            "Full MedpipeClassifier pipeline execution finished successfully."
        )

        results: dict[str, Any] = {
            "fitted_models": fitted_models,
            "evaluations": evaluations,
        }
        if plots:
            results["plots"] = plots

        return results

    def plot_all(
        self,
        y_true: np.ndarray,
        probas: np.ndarray,
        outcome: str = "default",
        n_bootstraps: int | None = None,
        save: bool | None = None,
        show: bool | None = None,
        **style_kwargs: Any,
    ) -> dict[str, tuple[Figure | SubFigure, Axes]]:
        """Execute all core evaluation visualization routines for a given outcome.

        Generates and optionally persists the ROC curve, Precision-Recall curve,
        Data Distribution histogram, Reliability Diagram, and Decision
        Curve Analysis (DCA).

        Parameters
        ----------
        y_true : numpy.ndarray
            Ground truth binary target labels of shape (n_samples,).
        probas : numpy.ndarray
            Predicted probabilities of shape (n_samples, 2) or (n_samples,).
        outcome : str, default="default"
            Outcome identifier used for figure titles and directory structuring.
        n_bootstraps : int, optional
            Number of bootstrap iterations for confidence interval estimation on ROC,
            PR, and reliability curves. If None, resolved from the display
            configuration.
        save : bool, optional
            Automatically save all generated plot artifacts to the run directory.
            If None, resolved from the display configuration.
        show : bool, optional
            Whether to display figures interactively before closing. If None,
            resolved from the display configuration.
        **style_kwargs : Any
            Additional style parameters forwarded to underlying drawing primitives.

        Returns
        -------
        plots : dict of str to tuple of (matplotlib.figure.Figure, matplotlib.axes.Axes)
            Dictionary mapping plot keys
            ('roc', 'pr', 'distribution', 'reliability', 'dca')
            to their rendered (Figure, Axes) Matplotlib objects.

        """
        return self.displayer.plot_all(
            y_true=y_true,
            probas=probas,
            outcome=outcome,
            n_bootstraps=n_bootstraps,
            save=save,
            show=show,
            **style_kwargs,
        )

    @classmethod
    def load(cls, run_dir: str | Path) -> MedpipeClassifier:
        """Reconstruct a MedpipeClassifier instance from a run artifact directory.

        Parses the saved JSON configuration and restores serialized outcome
        model artifacts into the runner and evaluator engines.

        Parameters
        ----------
        run_dir : str or Path
            Directory path of a previously executed MedpipeClassifier run artifact.

        Returns
        -------
        pipe : MedpipeClassifier
            Re-instantiated MedpipeClassifier object ready for inference,
            evaluation, or visualization.

        Raises
        ------
        FileNotFoundError
            If the resolved configuration JSON file is missing from the run directory.

        """
        run_path = Path(run_dir)
        config_path = run_path / "env/resolved_config.json"
        models_dir = run_path / "models"

        if not config_path.exists():
            raise FileNotFoundError(
                "Cannot load MedpipeClassifier instance: Configuration JSON missing "
                f"in '{run_path}'"
            )

        # 1. Load JSON dict and instantiate MedpipeClassifierConfig
        import json

        import joblib

        with open(config_path, encoding="utf-8") as f:
            config_dict = json.load(f)

        mp_config = MedpipeClassifierConfig.model_validate(config_dict)

        # 2. Instantiate MedpipeClassifier with reconstructed MedpipeClassifierConfig
        new_run_path = run_path / "eval"
        pipe = cls(config=mp_config, base_artifact_dir=new_run_path)
        pipe._orchestrator.run_dir = new_run_path
        pipe.displayer.run_dir = new_run_path

        # 3. Restore serialized model binaries into runner and evaluator engines
        if models_dir.exists():
            project_name = pipe.mp_config.meta.project_name
            pipe._runner.fitted_models = joblib.load(
                models_dir / f"{project_name}_fitted.joblib"
            )
            pipe._evaluator.fitted_models = pipe._runner.fitted_models
        pipe._logger.info(f"Successfully loaded MedpipeClassifier from {run_dir}")

        return pipe

models property

models: dict[str, Pipeline | CalibratedClassifierCV]

Return the fitted models.

Returns:

Name	Type	Description
fitted_models	Dict[str, Union[Pipeline, CalibratedClassifierCV]]	Fitted models from the MedpipeClassifierRunner object.

is_fitted property

is_fitted: bool

Checks if the MedpipeClassifier is fitted by looking at MedpipeClassifierRunner.

Returns:

Name	Type	Description
fitted	bool	True if the models have been fitted, False otherwise.

run_dir property

run_dir: Path

Returns the run_dir from the MedpipeOrchestrator object.

Returns:

Name	Type	Description
run_dir	Path	Path to the run_dir.

data_split property

data_split: DataSplits

Access any of the data splits.

Return

DataSplits Any of the DataSplits attributes.

fairness_split property

fairness_split: FairnessSplits | None

Access fairness stratification columns aligned with each data split.

Lets fairness analyses be run directly against the relevant columns (e.g. `pipeline.fairness_split.test["HOSPITAL"]") without re-extracting them from the raw dataset, and independently of whether those columns are also model predictors.

Return

FairnessSplits or None Any of the FairnessSplits attributes, or None if no workflow.evaluation.fairness configuration is set.

prepare_data property

prepare_data: Callable[..., tuple]

Access the orchestrator's data preparation function.

Ingests the configured dataset and creates the train/recal/test splits, e.g. X_train, y_train, X_recal, y_recal, X_test, y_test, groups_train = pipeline.prepare_data(). Keyword arguments are passed to the underlying Pandas reader function.

Returns:

Type	Description
Callable	The bound MedpipeOrchestrator.prepare_data method.

metrics property

metrics: list[str]

Returns metrics used during evaluation.

Returns:

Name	Type	Description
metrics	list[str]	List of metrics used during evaluation.

fit

fit(
    X_train: DataFrame,
    y_train: DataFrame,
    X_recal: DataFrame | None = None,
    y_recal: DataFrame | None = None,
    groups_train: ndarray | None = None,
) -> dict[str, Any]

Fit machine learning models across configured target outcomes via MedpipeClassifierRunner.

Parameters:

Name	Type	Description	Default
X_train	DataFrame	Training feature dataset.	required
y_train	DataFrame	Training outcome target labels (can contain multiple outcome columns).	required
X_recal	DataFrame	Recalibration feature dataset for post-hoc model calibration, default=None.	None
y_recal	DataFrame	Recalibration outcome target labels, default=None.	None
groups_train	ndarray	Group identifier array for group-based cross-validation splits, default=None.	None

Returns:

Name	Type	Description
fitted_models	dict of str to object	Dictionary mapping outcome target names to their finalized fitted models.

Source code in medpipe/pipeline/pipeline.py

def fit(
    self,
    X_train: pd.DataFrame,
    y_train: pd.DataFrame,
    X_recal: pd.DataFrame | None = None,
    y_recal: pd.DataFrame | None = None,
    groups_train: np.ndarray | None = None,
) -> dict[str, Any]:
    """
    Fit machine learning models across configured target outcomes via
    MedpipeClassifierRunner.

    Parameters
    ----------
    X_train : pandas.DataFrame
        Training feature dataset.
    y_train : pandas.DataFrame
        Training outcome target labels (can contain multiple outcome columns).
    X_recal : pandas.DataFrame, optional
        Recalibration feature dataset for post-hoc model calibration, default=None.
    y_recal : pandas.DataFrame, optional
        Recalibration outcome target labels, default=None.
    groups_train : numpy.ndarray, optional
        Group identifier array for group-based cross-validation splits,
        default=None.

    Returns
    -------
    fitted_models : dict of str to object
        Dictionary mapping outcome target names to their finalized fitted models.

    """
    self._logger.info("Starting model fitting across configured target outcomes.")
    fitted_models = self._runner.run(
        X_train=X_train,
        y_train_df=y_train,
        X_recal=X_recal,
        y_recal_df=y_recal,
        groups_train=groups_train,
    )
    self._logger.info("Completed model fitting for all target outcomes.")
    return fitted_models

predict

predict(
    X: DataFrame | NDArray,
    model: Any | None = None,
    outcome: str | None = None,
) -> NDArray

Predict class labels for input samples.

Parameters:

Name	Type	Description	Default
X	DataFrame or ndarray	Features dataset of shape (n_samples, n_features).	required
model	object	Fitted model instance. If None, resolved via outcome or runner.fitted_models.	None
outcome	str	Outcome name corresponding to an entry in runner.fitted_models.	None

Returns:

Name	Type	Description
y_pred	ndarray	Predicted class labels of shape (n_samples,).

Source code in medpipe/pipeline/pipeline.py

def predict(
    self,
    X: pd.DataFrame | npt.NDArray,
    model: Any | None = None,
    outcome: str | None = None,
) -> npt.NDArray:
    """
    Predict class labels for input samples.

    Parameters
    ----------
    X : pandas.DataFrame or numpy.ndarray
        Features dataset of shape (n_samples, n_features).
    model : object, optional
        Fitted model instance. If None, resolved via `outcome` or
        `runner.fitted_models`.
    outcome : str, optional
        Outcome name corresponding to an entry in `runner.fitted_models`.

    Returns
    -------
    y_pred : numpy.ndarray
        Predicted class labels of shape (n_samples,).

    """
    return self._evaluator.predict(X=X, model=model, outcome=outcome)

predict_proba

predict_proba(
    X: DataFrame | NDArray,
    model: Any | None = None,
    outcome: str | None = None,
) -> NDArray

Predict class probabilities for input samples.

Parameters:

Name	Type	Description	Default
X	DataFrame or ndarray	Features dataset of shape (n_samples, n_features).	required
model	object	Fitted model instance. If None, resolved via outcome or runner.fitted_models.	None
outcome	str	Outcome name corresponding to an entry in runner.fitted_models.	None

Returns:

Name	Type	Description
y_proba	ndarray	Predicted class probabilities of shape (n_samples, n_classes) or (n_samples,).

Source code in medpipe/pipeline/pipeline.py

def predict_proba(
    self,
    X: pd.DataFrame | npt.NDArray,
    model: Any | None = None,
    outcome: str | None = None,
) -> npt.NDArray:
    """
    Predict class probabilities for input samples.

    Parameters
    ----------
    X : pandas.DataFrame or numpy.ndarray
        Features dataset of shape (n_samples, n_features).
    model : object, optional
        Fitted model instance. If None, resolved via `outcome` or
        `runner.fitted_models`.
    outcome : str, optional
        Outcome name corresponding to an entry in `runner.fitted_models`.

    Returns
    -------
    y_proba : numpy.ndarray
        Predicted class probabilities of shape (n_samples, n_classes) or
        (n_samples,).

    """
    return self._evaluator.predict_proba(X=X, model=model, outcome=outcome)

decision_function

decision_function(
    X: DataFrame | NDArray,
    model: Any | None = None,
    outcome: str | None = None,
) -> NDArray

Compute decision function confidence scores for input samples.

Parameters:

Name	Type	Description	Default
X	DataFrame or ndarray	Features dataset of shape (n_samples, n_features).	required
model	object	Fitted model instance. If None, resolved via outcome or runner.fitted_models.	None
outcome	str	Outcome name corresponding to an entry in runner.fitted_models.	None

Returns:

Name	Type	Description
scores	ndarray	Confidence scores or decision function values of shape (n_samples,).

Source code in medpipe/pipeline/pipeline.py

def decision_function(
    self,
    X: pd.DataFrame | npt.NDArray,
    model: Any | None = None,
    outcome: str | None = None,
) -> npt.NDArray:
    """
    Compute decision function confidence scores for input samples.

    Parameters
    ----------
    X : pandas.DataFrame or numpy.ndarray
        Features dataset of shape (n_samples, n_features).
    model : object, optional
        Fitted model instance. If None, resolved via `outcome` or
        `runner.fitted_models`.
    outcome : str, optional
        Outcome name corresponding to an entry in `runner.fitted_models`.

    Returns
    -------
    scores : numpy.ndarray
        Confidence scores or decision function values of shape (n_samples,).

    """
    return self._evaluator.decision_function(X=X, model=model, outcome=outcome)

evaluate

evaluate(
    X: DataFrame,
    y: Series | NDArray,
    outcome: str | None = None,
    model: Any | None = None,
    metrics: list[str] | None = None,
    subgroup_specs: dict[
        str, str | Callable[[DataFrame], Series]
    ]
    | None = None,
    fairness_data: DataFrame | None = None,
    save_artifacts: bool = True,
) -> dict[str, Any]

Evaluate model performance with confidence intervals on full datasets and subgroups.

Parameters:

Name	Type	Description	Default
X	DataFrame	Feature dataset of shape (n_samples, n_features).	required
y	Series or ndarray	Ground truth target values of shape (n_samples,).	required
outcome	str	Outcome key name. Used for logging, model resolution, and artifact saving.	None
model	object	Explicit fitted model instance. If None, resolved from runner.fitted_models.	None
metrics	list of str	List of metrics to evaluate. If None, defaults to self._evaluator.metrics.	None
subgroup_specs	dict of str to (str or callable)	Specifications for extracting demographic or clinical subgroups.	None
fairness_data	DataFrame	Fairness stratification columns aligned with X (see fairness_split), used to resolve subgroup_specs for strata that are not themselves model predictors (e.g. "HOSPITAL"). Defaults to X when not provided.	None
save_artifacts	bool	Whether to write evaluation summary results to disk via ArtifactManager.	True

Returns:

Name	Type	Description
results	dict of str to Any	Nested dictionary containing outcome identifier, overall slice metrics, and subgroup performance.

Source code in medpipe/pipeline/pipeline.py

def evaluate(
    self,
    X: pd.DataFrame,
    y: pd.Series | npt.NDArray,
    outcome: str | None = None,
    model: Any | None = None,
    metrics: list[str] | None = None,
    subgroup_specs: dict[str, str | Callable[[pd.DataFrame], pd.Series]]
    | None = None,
    fairness_data: pd.DataFrame | None = None,
    save_artifacts: bool = True,
) -> dict[str, Any]:
    """
    Evaluate model performance with confidence intervals on full datasets
    and subgroups.

    Parameters
    ----------
    X : pandas.DataFrame
        Feature dataset of shape (n_samples, n_features).
    y : pandas.Series or numpy.ndarray
        Ground truth target values of shape (n_samples,).
    outcome : str, optional
        Outcome key name. Used for logging, model resolution, and artifact
        saving.
    model : object, optional
        Explicit fitted model instance. If None, resolved from
        `runner.fitted_models`.
    metrics : list of str, optional
        List of metrics to evaluate. If None, defaults to
        `self._evaluator.metrics`.
    subgroup_specs : dict of str to (str or callable), optional
        Specifications for extracting demographic or clinical subgroups.
    fairness_data : pandas.DataFrame, optional
        Fairness stratification columns aligned with `X` (see
        `fairness_split`), used to resolve `subgroup_specs` for strata
        that are not themselves model predictors (e.g. "HOSPITAL").
        Defaults to `X` when not provided.
    save_artifacts : bool, default=True
        Whether to write evaluation summary results to disk via
        `ArtifactManager`.

    Returns
    -------
    results : dict of str to Any
        Nested dictionary containing outcome identifier, overall slice
        metrics, and subgroup performance.

    """
    # Resolve single target Series/ndarray if y is supplied as a DataFrame
    if isinstance(y, pd.DataFrame):
        if outcome and outcome in y.columns:
            y_eval = y[outcome]
        elif y.shape[1] == 1:
            y_eval = y.iloc[:, 0]
        else:
            y_eval = y
    else:
        y_eval = y

    return self._evaluator.evaluate(
        X=X,
        y=y_eval,
        outcome=outcome,
        model=model,
        metrics=metrics,
        subgroup_specs=subgroup_specs,
        fairness_data=fairness_data,
        save_artifacts=save_artifacts,
    )

run

run(groups_train: ndarray | None = None) -> dict[str, Any]

Execute full end-to-end pipeline (data preparation, model fitting, and test evaluation).

Automates data ingestion, split creation, model cross-validation and fitting, and test evaluation with TRIPOD+AI reporting across all target outcomes. The model fitting duration and total run duration are recorded at the debug log level.

Parameters:

Name	Type	Description	Default
groups_train	ndarray	Group labels for training samples if group-based cross-validation is configured.	None

Returns:

Name	Type	Description
pipeline_results	dict of str to Any	Dictionary containing: - "fitted_models": Dictionary mapping outcome target names to fitted model instances. - "evaluations": Dictionary mapping outcome target names to evaluation results dictionaries. - "plots": (Audit / Eval mode only) Nested dictionary mapping outcome target names to figure objects.

Source code in medpipe/pipeline/pipeline.py

def run(
    self,
    groups_train: np.ndarray | None = None,
) -> dict[str, Any]:
    """
    Execute full end-to-end pipeline (data preparation, model fitting, and
    test evaluation).

    Automates data ingestion, split creation, model cross-validation and fitting,
    and test evaluation with TRIPOD+AI reporting across all target outcomes.
    The model fitting duration and total run duration are recorded at
    the debug log level.

    Parameters
    ----------
    groups_train : numpy.ndarray, optional
        Group labels for training samples if group-based cross-validation
        is configured.

    Returns
    -------
    pipeline_results : dict of str to Any
        Dictionary containing:
        - `"fitted_models"`: Dictionary mapping outcome target names
        to fitted model instances.
        - `"evaluations"`: Dictionary mapping outcome target names
        to evaluation results dictionaries.
        - `"plots"`: (Audit / Eval mode only) Nested dictionary mapping
        outcome target names to figure objects.

    """
    self._logger.info("Executing full MedpipeClassifier pipeline end-to-end.")
    run_start_time = time.perf_counter()

    run_mode = self.mp_config.meta.run_mode

    self._logger.debug(
        f"Executing full MedpipeClassifier pipeline in {run_mode} mode."
    )

    n_steps = 3
    if run_mode == "audit" or run_mode == "eval":
        n_steps = 4

    # 1. Prepare data splits via orchestrator
    data_kwargs = self.mp_config.data.kwargs  # Get extra data arguments
    self._logger.info(f"Step 1/{n_steps}: Ingesting and splitting dataset.")
    X_train, y_train, X_recal, y_recal, X_test, y_test, groups_train = (
        self._orchestrator.prepare_data(**data_kwargs)
    )

    # 2. Fit models via runner
    self._logger.info(f"Step 2/{n_steps}: Fitting outcome models.")
    fit_start_time = time.perf_counter()
    fitted_models = self.fit(
        X_train=X_train,
        y_train=y_train,
        X_recal=X_recal,
        y_recal=y_recal,
        groups_train=groups_train,
    )
    fit_duration = time.perf_counter() - fit_start_time
    self._logger.debug(f"Model fitting completed in {fit_duration:.2f} seconds.")

    # 3. Evaluate models on test set via evaluator
    self._logger.info(
        f"Step 3/{n_steps}: Evaluating models on holdout test set.",
    )
    evaluations: dict[str, Any] = {}
    plots: dict[str, dict[str, tuple[Figure | SubFigure, Axes]]] = {}
    outcomes = self._orchestrator.config.data.outcomes
    subgroup_specs = self._orchestrator.get_subgroup_specs()
    fairness_splits = self._orchestrator.fairness_splits
    fairness_data = fairness_splits.test if fairness_splits is not None else None

    for outcome in outcomes:
        y_test_outcome = y_test[outcome]
        evaluations[outcome] = self.evaluate(
            X=X_test,
            y=y_test_outcome.to_numpy(),
            outcome=outcome,
            subgroup_specs=subgroup_specs,
            fairness_data=fairness_data,
            save_artifacts=True,
        )

    if run_mode == "audit" or run_mode == "eval":
        self._logger.info(f"Step 4/{n_steps}: Plotting graphs.")
        for outcome in outcomes:
            y_true_outcome = y_test[outcome].to_numpy()
            probas_outcome = self.predict_proba(X=X_test, outcome=outcome)

            plots[outcome] = self.plot_all(
                y_true=y_true_outcome,
                probas=probas_outcome,
                outcome=outcome,
            )
        # Generate cross-outcome strata heatmaps per metric
        if fairness_splits is None:
            self._logger.info(
                "No fairness configuration set; skipping subgroup strata heatmaps."
            )
        else:
            self._logger.info(
                "Generating subgroup strata heatmaps across outcomes."
            )
            strata_heatmaps = self.displayer.plot_all_heatmaps(
                evaluations=evaluations,
            )
            plots["strata_heatmaps"] = strata_heatmaps

    run_duration = time.perf_counter() - run_start_time
    self._logger.debug(
        f"Full pipeline run completed in {run_duration:.2f} seconds."
    )
    self._logger.info(
        "Full MedpipeClassifier pipeline execution finished successfully."
    )

    results: dict[str, Any] = {
        "fitted_models": fitted_models,
        "evaluations": evaluations,
    }
    if plots:
        results["plots"] = plots

    return results

plot_all

plot_all(
    y_true: ndarray,
    probas: ndarray,
    outcome: str = "default",
    n_bootstraps: int | None = None,
    save: bool | None = None,
    show: bool | None = None,
    **style_kwargs: Any,
) -> dict[str, tuple[Figure | SubFigure, Axes]]

Execute all core evaluation visualization routines for a given outcome.

Generates and optionally persists the ROC curve, Precision-Recall curve, Data Distribution histogram, Reliability Diagram, and Decision Curve Analysis (DCA).

Parameters:

Name	Type	Description	Default
y_true	ndarray	Ground truth binary target labels of shape (n_samples,).	required
probas	ndarray	Predicted probabilities of shape (n_samples, 2) or (n_samples,).	required
outcome	str	Outcome identifier used for figure titles and directory structuring.	"default"
n_bootstraps	int	Number of bootstrap iterations for confidence interval estimation on ROC, PR, and reliability curves. If None, resolved from the display configuration.	None
save	bool	Automatically save all generated plot artifacts to the run directory. If None, resolved from the display configuration.	None
show	bool	Whether to display figures interactively before closing. If None, resolved from the display configuration.	None
**style_kwargs	Any	Additional style parameters forwarded to underlying drawing primitives.	{}

Returns:

Name	Type	Description
plots	dict of str to tuple of (matplotlib.figure.Figure, matplotlib.axes.Axes)	Dictionary mapping plot keys ('roc', 'pr', 'distribution', 'reliability', 'dca') to their rendered (Figure, Axes) Matplotlib objects.

Source code in medpipe/pipeline/pipeline.py

def plot_all(
    self,
    y_true: np.ndarray,
    probas: np.ndarray,
    outcome: str = "default",
    n_bootstraps: int | None = None,
    save: bool | None = None,
    show: bool | None = None,
    **style_kwargs: Any,
) -> dict[str, tuple[Figure | SubFigure, Axes]]:
    """Execute all core evaluation visualization routines for a given outcome.

    Generates and optionally persists the ROC curve, Precision-Recall curve,
    Data Distribution histogram, Reliability Diagram, and Decision
    Curve Analysis (DCA).

    Parameters
    ----------
    y_true : numpy.ndarray
        Ground truth binary target labels of shape (n_samples,).
    probas : numpy.ndarray
        Predicted probabilities of shape (n_samples, 2) or (n_samples,).
    outcome : str, default="default"
        Outcome identifier used for figure titles and directory structuring.
    n_bootstraps : int, optional
        Number of bootstrap iterations for confidence interval estimation on ROC,
        PR, and reliability curves. If None, resolved from the display
        configuration.
    save : bool, optional
        Automatically save all generated plot artifacts to the run directory.
        If None, resolved from the display configuration.
    show : bool, optional
        Whether to display figures interactively before closing. If None,
        resolved from the display configuration.
    **style_kwargs : Any
        Additional style parameters forwarded to underlying drawing primitives.

    Returns
    -------
    plots : dict of str to tuple of (matplotlib.figure.Figure, matplotlib.axes.Axes)
        Dictionary mapping plot keys
        ('roc', 'pr', 'distribution', 'reliability', 'dca')
        to their rendered (Figure, Axes) Matplotlib objects.

    """
    return self.displayer.plot_all(
        y_true=y_true,
        probas=probas,
        outcome=outcome,
        n_bootstraps=n_bootstraps,
        save=save,
        show=show,
        **style_kwargs,
    )

load classmethod

load(run_dir: str | Path) -> MedpipeClassifier

Reconstruct a MedpipeClassifier instance from a run artifact directory.

Parses the saved JSON configuration and restores serialized outcome model artifacts into the runner and evaluator engines.

Parameters:

Name	Type	Description	Default
run_dir	str or Path	Directory path of a previously executed MedpipeClassifier run artifact.	required

Returns:

Name	Type	Description
pipe	MedpipeClassifier	Re-instantiated MedpipeClassifier object ready for inference, evaluation, or visualization.

Raises:

Type	Description
FileNotFoundError	If the resolved configuration JSON file is missing from the run directory.

Source code in medpipe/pipeline/pipeline.py

@classmethod
def load(cls, run_dir: str | Path) -> MedpipeClassifier:
    """Reconstruct a MedpipeClassifier instance from a run artifact directory.

    Parses the saved JSON configuration and restores serialized outcome
    model artifacts into the runner and evaluator engines.

    Parameters
    ----------
    run_dir : str or Path
        Directory path of a previously executed MedpipeClassifier run artifact.

    Returns
    -------
    pipe : MedpipeClassifier
        Re-instantiated MedpipeClassifier object ready for inference,
        evaluation, or visualization.

    Raises
    ------
    FileNotFoundError
        If the resolved configuration JSON file is missing from the run directory.

    """
    run_path = Path(run_dir)
    config_path = run_path / "env/resolved_config.json"
    models_dir = run_path / "models"

    if not config_path.exists():
        raise FileNotFoundError(
            "Cannot load MedpipeClassifier instance: Configuration JSON missing "
            f"in '{run_path}'"
        )

    # 1. Load JSON dict and instantiate MedpipeClassifierConfig
    import json

    import joblib

    with open(config_path, encoding="utf-8") as f:
        config_dict = json.load(f)

    mp_config = MedpipeClassifierConfig.model_validate(config_dict)

    # 2. Instantiate MedpipeClassifier with reconstructed MedpipeClassifierConfig
    new_run_path = run_path / "eval"
    pipe = cls(config=mp_config, base_artifact_dir=new_run_path)
    pipe._orchestrator.run_dir = new_run_path
    pipe.displayer.run_dir = new_run_path

    # 3. Restore serialized model binaries into runner and evaluator engines
    if models_dir.exists():
        project_name = pipe.mp_config.meta.project_name
        pipe._runner.fitted_models = joblib.load(
            models_dir / f"{project_name}_fitted.joblib"
        )
        pipe._evaluator.fitted_models = pipe._runner.fitted_models
    pipe._logger.info(f"Successfully loaded MedpipeClassifier from {run_dir}")

    return pipe

MedpipeRegressor

User entry point and high-level pipeline runner for MedpipeRegressor.

Coordinates the complete machine learning lifecycle for regression outcomes, delegating data ingress and preprocessing setup to MedpipeOrchestrator, model cross-validation and fitting to MedpipeRegressorRunner, and prediction and TRIPOD+AI evaluation to MedpipeRegressorEvaluator.

Unlike MedpipeClassifier, there is no post-hoc recalibration step (not a meaningful concept for regression outcomes). Distributional diagnostic figures (coverage, sharpness, Winkler score, marginal calibration, PIT histogram, and target-value data distribution) are available via plot_all, which run() calls in the "audit" and "eval" run modes. The figures are only available for models producing a predictive CDF through predict_dist (currently OrdBoost only); run() skips them with a warning for any other model.

Parameters:

Name	Type	Description	Default
config	str, Path, or MedpipeRegressorConfig	Path to a TOML configuration file (parsed by MedpipeOrchestrator via read_regressor_toml_configuration) or an instantiated MedpipeRegressorConfig object.	required
base_artifact_dir	str or Path	Root directory where versioned execution run artifacts, logs, and models are stored.	"artifacts"
verbose_override	Union[bool, int, str, None]	Console verbosity setting configuration override.	None

Attributes:

Name	Type	Description
mp_config	MedpipeRegressorConfig	Validated configuration driving the pipeline.
displayer	MedpipeRegressorDisplayer	Visualisation engine rendering and persisting distributional diagnostic figures.

Source code in medpipe/pipeline/pipeline.py

class MedpipeRegressor:
    """
    User entry point and high-level pipeline runner for MedpipeRegressor.

    Coordinates the complete machine learning lifecycle for regression
    outcomes, delegating data ingress and preprocessing setup to
    `MedpipeOrchestrator`, model cross-validation and fitting to
    `MedpipeRegressorRunner`, and prediction and TRIPOD+AI evaluation to
    `MedpipeRegressorEvaluator`.

    Unlike `MedpipeClassifier`, there is no post-hoc recalibration step (not
    a meaningful concept for regression outcomes). Distributional diagnostic
    figures (coverage, sharpness, Winkler score, marginal calibration, PIT
    histogram, and target-value data distribution) are available via
    `plot_all`, which `run()` calls in the "audit" and "eval" run modes. The
    figures are only available for models producing a predictive CDF through
    `predict_dist` (currently OrdBoost only); `run()` skips them with a
    warning for any other model.

    Parameters
    ----------
    config : str, Path, or MedpipeRegressorConfig
        Path to a TOML configuration file (parsed by `MedpipeOrchestrator`
        via `read_regressor_toml_configuration`) or an instantiated
        `MedpipeRegressorConfig` object.
    base_artifact_dir : str or Path, default="artifacts"
        Root directory where versioned execution run artifacts, logs, and
        models are stored.
    verbose_override : Union[bool, int, str, None], default=None
        Console verbosity setting configuration override.

    Attributes
    ----------
    mp_config : MedpipeRegressorConfig
        Validated configuration driving the pipeline.
    displayer : MedpipeRegressorDisplayer
        Visualisation engine rendering and persisting distributional
        diagnostic figures.

    """

    def __init__(
        self,
        config: str | Path | MedpipeRegressorConfig,
        base_artifact_dir: str | Path = "artifacts",
        verbose_override: bool | int | str | None = None,
    ) -> None:
        self._logger = get_console_logger("medpipe")

        self._logger.info("Initialising MedpipeRegressor end-to-end pipeline.")

        self._orchestrator = MedpipeOrchestrator(
            config, base_artifact_dir, verbose_override, is_classifier=False
        )
        self.mp_config = self._orchestrator.config
        self._runner = MedpipeRegressorRunner(orchestrator=self._orchestrator)
        self._evaluator = MedpipeRegressorEvaluator(
            orchestrator=self._orchestrator,
            runner=self._runner,
        )
        self.displayer = MedpipeRegressorDisplayer(orchestrator=self._orchestrator)

        self._logger.info("MedpipeRegressor initialisation complete.")

    @property
    def models(self) -> dict[str, DistributionalPipeline]:
        """Return the fitted models.

        Returns
        -------
        fitted_models : Dict[str, DistributionalPipeline]
            Fitted models from the MedpipeRegressorRunner object.

        """
        return self._runner.fitted_models

    @property
    def is_fitted(self) -> bool:
        """Checks if the MedpipeRegressor is fitted by looking at
        MedpipeRegressorRunner.

        Returns
        -------
        fitted : bool
            True if the models have been fitted, False otherwise.

        """
        return self._runner.fitted_models != {}

    @property
    def run_dir(self) -> Path:
        """Returns the run_dir from the MedpipeOrchestrator object.

        Returns
        -------
        run_dir : Path
            Path to the run_dir.

        """
        return self._orchestrator.run_dir

    @property
    def data_split(self) -> DataSplits:
        """Access any of the data splits.

        Return
        ------
        DataSplits
            Any of the DataSplits attributes.

        """
        return self._orchestrator.splits

    @property
    def fairness_split(self) -> FairnessSplits | None:
        """Access fairness stratification columns aligned with each data
        split.

        Lets fairness analyses be run directly against the relevant
        columns (e.g. `pipeline.fairness_split.test["HOSPITAL"]") without
        re-extracting them from the raw dataset, and independently of
        whether those columns are also model predictors.

        Return
        ------
        FairnessSplits or None
            Any of the FairnessSplits attributes, or None if no
            `workflow.evaluation.fairness` configuration is set.

        """
        return self._orchestrator.fairness_splits

    @property
    def prepare_data(self) -> Callable[..., tuple]:
        """Access the orchestrator's data preparation function.

        Ingests the configured dataset and creates the train/recal/test
        splits, e.g. `X_train, y_train, X_recal, y_recal, X_test, y_test,
        groups_train = pipeline.prepare_data()`. Keyword arguments are passed
        to the underlying Pandas reader function.

        Returns
        -------
        Callable
            The bound `MedpipeOrchestrator.prepare_data` method.

        """
        return self._orchestrator.prepare_data

    @property
    def metrics(self) -> list[str]:
        """Returns metrics used during evaluation.

        Returns
        -------
        metrics : list[str]
            List of metrics used during evaluation.

        """
        return self._evaluator.metrics

    def fit(
        self,
        X_train: pd.DataFrame,
        y_train: pd.DataFrame,
        X_recal: pd.DataFrame | None = None,
        y_recal: pd.DataFrame | None = None,
        groups_train: np.ndarray | None = None,
    ) -> dict[str, Any]:
        """
        Fit machine learning models across configured target outcomes via
        MedpipeRegressorRunner.

        Parameters
        ----------
        X_train : pandas.DataFrame
            Training feature dataset.
        y_train : pandas.DataFrame
            Training outcome target labels (can contain multiple outcome columns).
        X_recal : pandas.DataFrame, optional
            Unused (no recalibration step for regression); accepted for
            interface symmetry with `MedpipeClassifier.fit`.
        y_recal : pandas.DataFrame, optional
            Unused (no recalibration step for regression); accepted for
            interface symmetry with `MedpipeClassifier.fit`.
        groups_train : numpy.ndarray, optional
            Group identifier array for group-based cross-validation splits,
            default=None.

        Returns
        -------
        fitted_models : dict of str to object
            Dictionary mapping outcome target names to their finalized fitted models.

        """
        self._logger.info("Starting model fitting across configured target outcomes.")
        fitted_models = self._runner.run(
            X_train=X_train,
            y_train_df=y_train,
            X_recal=X_recal,
            y_recal_df=y_recal,
            groups_train=groups_train,
        )
        self._logger.info("Completed model fitting for all target outcomes.")
        return fitted_models

    def predict(
        self,
        X: pd.DataFrame | npt.NDArray,
        model: Any | None = None,
        outcome: str | None = None,
    ) -> npt.NDArray:
        """
        Predict continuous values for input samples.

        Parameters
        ----------
        X : pandas.DataFrame or numpy.ndarray
            Features dataset of shape (n_samples, n_features).
        model : object, optional
            Fitted model instance. If None, resolved via `outcome` or
            `runner.fitted_models`.
        outcome : str, optional
            Outcome name corresponding to an entry in `runner.fitted_models`.

        Returns
        -------
        y_pred : numpy.ndarray
            Predicted continuous values of shape (n_samples,).

        """
        return self._evaluator.predict(X=X, model=model, outcome=outcome)

    def predict_dist(
        self,
        X: pd.DataFrame | npt.NDArray,
        model: Any | None = None,
        outcome: str | None = None,
    ) -> Any:
        """
        Predict the full predictive distribution for input samples.

        Parameters
        ----------
        X : pandas.DataFrame or numpy.ndarray
            Features dataset of shape (n_samples, n_features).
        model : object, optional
            Fitted model instance. If None, resolved via `outcome` or
            `runner.fitted_models`.
        outcome : str, optional
            Outcome name corresponding to an entry in `runner.fitted_models`.

        Returns
        -------
        Any
            The distribution object returned by the resolved model's
            `predict_dist` method.

        """
        return self._evaluator.predict_dist(X=X, model=model, outcome=outcome)

    def evaluate(
        self,
        X: pd.DataFrame,
        y: pd.Series | npt.NDArray,
        outcome: str | None = None,
        model: Any | None = None,
        metrics: list[str] | None = None,
        subgroup_specs: dict[str, str | Callable[[pd.DataFrame], pd.Series]]
        | None = None,
        fairness_data: pd.DataFrame | None = None,
        save_artifacts: bool = True,
    ) -> dict[str, Any]:
        """
        Evaluate model performance with confidence intervals on full datasets
        and subgroups.

        Parameters
        ----------
        X : pandas.DataFrame
            Feature dataset of shape (n_samples, n_features).
        y : pandas.Series or numpy.ndarray
            Ground truth target values of shape (n_samples,).
        outcome : str, optional
            Outcome key name. Used for logging, model resolution, and artifact
            saving.
        model : object, optional
            Explicit fitted model instance. If None, resolved from
            `runner.fitted_models`.
        metrics : list of str, optional
            List of metrics to evaluate. If None, defaults to
            `self._evaluator.metrics`.
        subgroup_specs : dict of str to (str or callable), optional
            Specifications for extracting demographic or clinical subgroups.
        fairness_data : pandas.DataFrame, optional
            Fairness stratification columns aligned with `X` (see
            `fairness_split`), used to resolve `subgroup_specs` for strata
            that are not themselves model predictors (e.g. "HOSPITAL").
            Defaults to `X` when not provided.
        save_artifacts : bool, default=True
            Whether to write evaluation summary results to disk via
            `ArtifactManager`.

        Returns
        -------
        results : dict of str to Any
            Nested dictionary containing outcome identifier, overall slice
            metrics, and subgroup performance.

        """
        # Resolve single target Series/ndarray if y is supplied as a DataFrame
        if isinstance(y, pd.DataFrame):
            if outcome and outcome in y.columns:
                y_eval = y[outcome]
            elif y.shape[1] == 1:
                y_eval = y.iloc[:, 0]
            else:
                y_eval = y
        else:
            y_eval = y

        return self._evaluator.evaluate(
            X=X,
            y=y_eval,
            outcome=outcome,
            model=model,
            metrics=metrics,
            subgroup_specs=subgroup_specs,
            fairness_data=fairness_data,
            save_artifacts=save_artifacts,
        )

    def run(
        self,
        groups_train: np.ndarray | None = None,
    ) -> dict[str, Any]:
        """
        Execute full end-to-end pipeline (data preparation, model fitting, and
        test evaluation).

        Automates data ingestion, split creation, model cross-validation and fitting,
        and test evaluation with TRIPOD+AI reporting across all target outcomes.
        The model fitting duration and total run duration are recorded at
        the debug log level.

        Note: only models that support `predict_dist`, are able to generate plots.

        Parameters
        ----------
        groups_train : numpy.ndarray, optional
            Group labels for training samples if group-based cross-validation
            is configured.

        Returns
        -------
        pipeline_results : dict of str to Any
            Dictionary containing:
            - `"fitted_models"`: Dictionary mapping outcome target names
            to fitted model instances.
            - `"evaluations"`: Dictionary mapping outcome target names
            to evaluation results dictionaries.
            - `"plots"`: (Audit / Eval mode only) Nested dictionary mapping
            outcome target names to figure objects.

        """
        self._logger.info("Executing full MedpipeRegressor pipeline end-to-end.")
        run_start_time = time.perf_counter()

        run_mode = self.mp_config.meta.run_mode
        self._logger.debug(
            f"Executing full MedpipeRegressor pipeline in {run_mode} mode."
        )

        n_steps = 3
        if run_mode == "audit" or run_mode == "eval":
            n_steps = 4

        # 1. Prepare data splits via orchestrator
        data_kwargs = self.mp_config.data.kwargs  # Get extra data arguments
        self._logger.info(f"Step 1/{n_steps}: Ingesting and splitting dataset.")
        X_train, y_train, X_recal, y_recal, X_test, y_test, groups_train = (
            self._orchestrator.prepare_data(**data_kwargs)
        )

        # 2. Fit models via runner
        self._logger.info(f"Step 2/{n_steps}: Fitting outcome models.")
        fit_start_time = time.perf_counter()
        fitted_models = self.fit(
            X_train=X_train,
            y_train=y_train,
            X_recal=X_recal,
            y_recal=y_recal,
            groups_train=groups_train,
        )
        fit_duration = time.perf_counter() - fit_start_time
        self._logger.debug(f"Model fitting completed in {fit_duration:.2f} seconds.")

        # 3. Evaluate models on test set via evaluator
        self._logger.info(
            f"Step 3/{n_steps}: Evaluating models on holdout test set.",
        )
        evaluations: dict[str, Any] = {}
        outcomes = self._orchestrator.config.data.outcomes
        plots: dict[str, dict[str, tuple[Figure | SubFigure, Axes]]] = {}
        subgroup_specs = self._orchestrator.get_subgroup_specs()
        fairness_splits = self._orchestrator.fairness_splits
        fairness_data = fairness_splits.test if fairness_splits is not None else None

        for outcome in outcomes:
            y_test_outcome = y_test[outcome]
            evaluations[outcome] = self.evaluate(
                X=X_test,
                y=y_test_outcome.to_numpy(),
                outcome=outcome,
                subgroup_specs=subgroup_specs,
                fairness_data=fairness_data,
                save_artifacts=True,
            )

        if run_mode == "audit" or run_mode == "eval":
            self._logger.info(f"Step 4/{n_steps}: Plotting graphs.")
            for outcome in outcomes:
                if not self._supports_distribution_plots(outcome):
                    self._logger.warning(
                        f"[{outcome}] Skipping the figures: the model "
                        f"{self._model_name(outcome)} does not provide a predictive "
                        "distribution that the figures support (OrdBoost "
                        "models only)."
                    )
                    continue
                y_true_outcome = y_test[outcome].to_numpy()
                dist_outcome = self.predict_dist(X=X_test, outcome=outcome)

                plots[outcome] = self.plot_all(
                    y_true=y_true_outcome,
                    dist=dist_outcome,
                    outcome=outcome,
                )
            # Generate cross-outcome strata heatmaps per metric
            if fairness_splits is None:
                self._logger.info(
                    "No fairness configuration set; skipping subgroup strata heatmaps."
                )
            else:
                self._logger.info(
                    "Generating subgroup strata heatmaps across outcomes."
                )
                strata_heatmaps = self.displayer.plot_all_heatmaps(
                    evaluations=evaluations,
                )
                plots["strata_heatmaps"] = strata_heatmaps

        run_duration = time.perf_counter() - run_start_time
        self._logger.debug(
            f"Full pipeline run completed in {run_duration:.2f} seconds."
        )
        self._logger.info(
            "Full MedpipeRegressor pipeline execution finished successfully."
        )

        results: dict[str, Any] = {
            "fitted_models": fitted_models,
            "evaluations": evaluations,
        }
        if plots:
            results["plots"] = plots

        return results

    def _model_name(self, outcome: str) -> str:
        """Return the class name of the final estimator fitted for an outcome."""
        model = self.models.get(outcome)
        if model is None:
            return "unknown"
        return type(model["regressor"]).__name__

    def _supports_distribution_plots(self, outcome: str) -> bool:
        """Check that the model of an outcome supports the distributional figures.

        The figures need a model that provides `predict_dist` and exposes the
        bin mapper used by the PIT histogram and the marginal calibration
        curve, which is the case of the OrdBoost models.

        Parameters
        ----------
        outcome : str
            Name of the outcome.

        Returns
        -------
        bool
            True if the figures can be drawn for the model of the outcome.

        """
        model = self.models.get(outcome)
        if model is None:
            return False
        regressor = model["regressor"]
        return (
            hasattr(regressor, "predict_dist")
            and getattr(regressor, "mapper_", None) is not None
        )

    def plot_all(
        self,
        y_true: npt.NDArray,
        dist: ContinuousPredictiveDistribution,
        outcome: str = "default",
        coverage_levels: npt.NDArray | None = None,
        n_bootstraps: int | None = None,
        save: bool | None = None,
        show: bool | None = None,
        **style_kwargs: Any,
    ) -> dict[str, tuple[Figure | SubFigure, Axes]]:
        """
        Execute all core distributional diagnostic visualization routines
        for a given outcome.

        Generates and optionally persists the coverage reliability curve,
        sharpness curve, Winkler score curve, marginal calibration curve,
        PIT histogram, and target-value data distribution histogram. Only
        supported for models that produce a predictive CDF via
        `predict_dist` (currently OrdBoost only). The
        bin mapper required for the PIT histogram is extracted directly
        from `self.models[outcome]["regressor"].mapper_`, so `outcome` must
        match an entry in `runner.fitted_models` for the PIT histogram to
        render; it stays skipped-with-a-clear-error otherwise, same as
        when no mapper is available.

        Called by `run()` in the "audit" and "eval" run modes for the outcomes
        whose model is distribution-capable (e.g. `OrdBoostRegressor`).

        Parameters
        ----------
        y_true : numpy.ndarray
            Ground truth continuous target values of shape (n_samples,).
        dist : ContinuousPredictiveDistribution
            Predictive distribution for the same samples, e.g. from
            `self.predict_dist(X, outcome=outcome)`.
        outcome : str, default="default"
            Outcome identifier used for figure titles, output folder
            structuring, and to resolve the fitted model (and its bin
            mapper) from `runner.fitted_models`.
        coverage_levels : numpy.ndarray, optional
            Nominal central-interval coverage levels in percent, used for the
            coverage, sharpness, and Winkler score curves. Defaults to
            `np.arange(5, 100, 10)`.
        n_bootstraps : int, optional
            Number of bootstrap iterations for coverage, sharpness, and
            Winkler score curves. If None, resolved from the display
            configuration.
        save : bool, optional
            Automatically save all generated plot artifacts to the run directory.
            If None, resolved from the display configuration.
        show : bool, optional
            Whether to display figures interactively before closing. If None,
            resolved from the display configuration.
        **style_kwargs : Any
            Additional style parameters forwarded to underlying drawing primitives.

        Returns
        -------
        plots : dict of str to tuple of (matplotlib.figure.Figure, matplotlib.axes.Axes)
            Dictionary mapping plot keys
            ('coverage', 'sharpness', 'winkler', 'marginal_calibration',
            'pit_histogram', 'data_distribution') to their rendered (Figure,
            Axes) Matplotlib objects.

        """
        fitted_model = self.models.get(outcome)
        mapper = (
            getattr(fitted_model["regressor"], "mapper_", None)
            if fitted_model is not None
            else None
        )

        return self.displayer.plot_all(
            y_true=y_true,
            dist=dist,
            mapper=mapper,
            outcome=outcome,
            coverage_levels=coverage_levels,
            n_bootstraps=n_bootstraps,
            save=save,
            show=show,
            **style_kwargs,
        )

    @classmethod
    def load(cls, run_dir: str | Path) -> MedpipeRegressor:
        """Reconstruct a MedpipeRegressor instance from a run artifact directory.

        Parses the saved JSON configuration and restores serialized outcome
        model artifacts into the runner and evaluator engines.

        Parameters
        ----------
        run_dir : str or Path
            Directory path of a previously executed MedpipeRegressor run artifact.

        Returns
        -------
        pipe : MedpipeRegressor
            Re-instantiated MedpipeRegressor object ready for inference,
            evaluation, or visualization.

        Raises
        ------
        FileNotFoundError
            If the resolved configuration JSON file is missing from the run directory.

        """
        run_path = Path(run_dir)
        config_path = run_path / "env/resolved_config.json"
        models_dir = run_path / "models"

        if not config_path.exists():
            raise FileNotFoundError(
                "Cannot load MedpipeRegressor instance: Configuration JSON missing "
                f"in '{run_path}'"
            )

        # 1. Load JSON dict and instantiate MedpipeRegressorConfig
        import json

        import joblib

        with open(config_path, encoding="utf-8") as f:
            config_dict = json.load(f)

        mp_config = MedpipeRegressorConfig.model_validate(config_dict)

        # 2. Instantiate MedpipeRegressor with reconstructed MedpipeRegressorConfig
        new_run_path = run_path / "eval"
        pipe = cls(config=mp_config, base_artifact_dir=new_run_path)
        pipe._orchestrator.run_dir = new_run_path
        pipe.displayer.run_dir = new_run_path

        # 3. Restore serialized model binaries into runner and evaluator engines
        if models_dir.exists():
            project_name = pipe.mp_config.meta.project_name
            pipe._runner.fitted_models = joblib.load(
                models_dir / f"{project_name}_fitted.joblib"
            )
            pipe._evaluator.fitted_models = pipe._runner.fitted_models
        pipe._logger.info(f"Successfully loaded MedpipeRegressor from {run_dir}")

        return pipe

models property

models: dict[str, DistributionalPipeline]

Return the fitted models.

Returns:

Name	Type	Description
fitted_models	Dict[str, DistributionalPipeline]	Fitted models from the MedpipeRegressorRunner object.

is_fitted property

is_fitted: bool

Checks if the MedpipeRegressor is fitted by looking at MedpipeRegressorRunner.

Returns:

Name	Type	Description
fitted	bool	True if the models have been fitted, False otherwise.

run_dir property

run_dir: Path

Returns the run_dir from the MedpipeOrchestrator object.

Returns:

Name	Type	Description
run_dir	Path	Path to the run_dir.

data_split property

data_split: DataSplits

Access any of the data splits.

Return

DataSplits Any of the DataSplits attributes.

fairness_split property

fairness_split: FairnessSplits | None

Access fairness stratification columns aligned with each data split.

Lets fairness analyses be run directly against the relevant columns (e.g. `pipeline.fairness_split.test["HOSPITAL"]") without re-extracting them from the raw dataset, and independently of whether those columns are also model predictors.

Return

FairnessSplits or None Any of the FairnessSplits attributes, or None if no workflow.evaluation.fairness configuration is set.

prepare_data property

prepare_data: Callable[..., tuple]

Access the orchestrator's data preparation function.

Ingests the configured dataset and creates the train/recal/test splits, e.g. X_train, y_train, X_recal, y_recal, X_test, y_test, groups_train = pipeline.prepare_data(). Keyword arguments are passed to the underlying Pandas reader function.

Returns:

Type	Description
Callable	The bound MedpipeOrchestrator.prepare_data method.

metrics property

metrics: list[str]

Returns metrics used during evaluation.

Returns:

Name	Type	Description
metrics	list[str]	List of metrics used during evaluation.

fit

fit(
    X_train: DataFrame,
    y_train: DataFrame,
    X_recal: DataFrame | None = None,
    y_recal: DataFrame | None = None,
    groups_train: ndarray | None = None,
) -> dict[str, Any]

Fit machine learning models across configured target outcomes via MedpipeRegressorRunner.

Parameters:

Name	Type	Description	Default
X_train	DataFrame	Training feature dataset.	required
y_train	DataFrame	Training outcome target labels (can contain multiple outcome columns).	required
X_recal	DataFrame	Unused (no recalibration step for regression); accepted for interface symmetry with MedpipeClassifier.fit.	None
y_recal	DataFrame	Unused (no recalibration step for regression); accepted for interface symmetry with MedpipeClassifier.fit.	None
groups_train	ndarray	Group identifier array for group-based cross-validation splits, default=None.	None

Returns:

Name	Type	Description
fitted_models	dict of str to object	Dictionary mapping outcome target names to their finalized fitted models.

Source code in medpipe/pipeline/pipeline.py

def fit(
    self,
    X_train: pd.DataFrame,
    y_train: pd.DataFrame,
    X_recal: pd.DataFrame | None = None,
    y_recal: pd.DataFrame | None = None,
    groups_train: np.ndarray | None = None,
) -> dict[str, Any]:
    """
    Fit machine learning models across configured target outcomes via
    MedpipeRegressorRunner.

    Parameters
    ----------
    X_train : pandas.DataFrame
        Training feature dataset.
    y_train : pandas.DataFrame
        Training outcome target labels (can contain multiple outcome columns).
    X_recal : pandas.DataFrame, optional
        Unused (no recalibration step for regression); accepted for
        interface symmetry with `MedpipeClassifier.fit`.
    y_recal : pandas.DataFrame, optional
        Unused (no recalibration step for regression); accepted for
        interface symmetry with `MedpipeClassifier.fit`.
    groups_train : numpy.ndarray, optional
        Group identifier array for group-based cross-validation splits,
        default=None.

    Returns
    -------
    fitted_models : dict of str to object
        Dictionary mapping outcome target names to their finalized fitted models.

    """
    self._logger.info("Starting model fitting across configured target outcomes.")
    fitted_models = self._runner.run(
        X_train=X_train,
        y_train_df=y_train,
        X_recal=X_recal,
        y_recal_df=y_recal,
        groups_train=groups_train,
    )
    self._logger.info("Completed model fitting for all target outcomes.")
    return fitted_models

predict

predict(
    X: DataFrame | NDArray,
    model: Any | None = None,
    outcome: str | None = None,
) -> NDArray

Predict continuous values for input samples.

Parameters:

Name	Type	Description	Default
X	DataFrame or ndarray	Features dataset of shape (n_samples, n_features).	required
model	object	Fitted model instance. If None, resolved via outcome or runner.fitted_models.	None
outcome	str	Outcome name corresponding to an entry in runner.fitted_models.	None

Returns:

Name	Type	Description
y_pred	ndarray	Predicted continuous values of shape (n_samples,).

Source code in medpipe/pipeline/pipeline.py

def predict(
    self,
    X: pd.DataFrame | npt.NDArray,
    model: Any | None = None,
    outcome: str | None = None,
) -> npt.NDArray:
    """
    Predict continuous values for input samples.

    Parameters
    ----------
    X : pandas.DataFrame or numpy.ndarray
        Features dataset of shape (n_samples, n_features).
    model : object, optional
        Fitted model instance. If None, resolved via `outcome` or
        `runner.fitted_models`.
    outcome : str, optional
        Outcome name corresponding to an entry in `runner.fitted_models`.

    Returns
    -------
    y_pred : numpy.ndarray
        Predicted continuous values of shape (n_samples,).

    """
    return self._evaluator.predict(X=X, model=model, outcome=outcome)

predict_dist

predict_dist(
    X: DataFrame | NDArray,
    model: Any | None = None,
    outcome: str | None = None,
) -> Any

Predict the full predictive distribution for input samples.

Parameters:

Name	Type	Description	Default
X	DataFrame or ndarray	Features dataset of shape (n_samples, n_features).	required
model	object	Fitted model instance. If None, resolved via outcome or runner.fitted_models.	None
outcome	str	Outcome name corresponding to an entry in runner.fitted_models.	None

Returns:

Type	Description
Any	The distribution object returned by the resolved model's predict_dist method.

Source code in medpipe/pipeline/pipeline.py

def predict_dist(
    self,
    X: pd.DataFrame | npt.NDArray,
    model: Any | None = None,
    outcome: str | None = None,
) -> Any:
    """
    Predict the full predictive distribution for input samples.

    Parameters
    ----------
    X : pandas.DataFrame or numpy.ndarray
        Features dataset of shape (n_samples, n_features).
    model : object, optional
        Fitted model instance. If None, resolved via `outcome` or
        `runner.fitted_models`.
    outcome : str, optional
        Outcome name corresponding to an entry in `runner.fitted_models`.

    Returns
    -------
    Any
        The distribution object returned by the resolved model's
        `predict_dist` method.

    """
    return self._evaluator.predict_dist(X=X, model=model, outcome=outcome)

evaluate

evaluate(
    X: DataFrame,
    y: Series | NDArray,
    outcome: str | None = None,
    model: Any | None = None,
    metrics: list[str] | None = None,
    subgroup_specs: dict[
        str, str | Callable[[DataFrame], Series]
    ]
    | None = None,
    fairness_data: DataFrame | None = None,
    save_artifacts: bool = True,
) -> dict[str, Any]

Evaluate model performance with confidence intervals on full datasets and subgroups.

Parameters:

Name	Type	Description	Default
X	DataFrame	Feature dataset of shape (n_samples, n_features).	required
y	Series or ndarray	Ground truth target values of shape (n_samples,).	required
outcome	str	Outcome key name. Used for logging, model resolution, and artifact saving.	None
model	object	Explicit fitted model instance. If None, resolved from runner.fitted_models.	None
metrics	list of str	List of metrics to evaluate. If None, defaults to self._evaluator.metrics.	None
subgroup_specs	dict of str to (str or callable)	Specifications for extracting demographic or clinical subgroups.	None
fairness_data	DataFrame	Fairness stratification columns aligned with X (see fairness_split), used to resolve subgroup_specs for strata that are not themselves model predictors (e.g. "HOSPITAL"). Defaults to X when not provided.	None
save_artifacts	bool	Whether to write evaluation summary results to disk via ArtifactManager.	True

Returns:

Name	Type	Description
results	dict of str to Any	Nested dictionary containing outcome identifier, overall slice metrics, and subgroup performance.

Source code in medpipe/pipeline/pipeline.py

def evaluate(
    self,
    X: pd.DataFrame,
    y: pd.Series | npt.NDArray,
    outcome: str | None = None,
    model: Any | None = None,
    metrics: list[str] | None = None,
    subgroup_specs: dict[str, str | Callable[[pd.DataFrame], pd.Series]]
    | None = None,
    fairness_data: pd.DataFrame | None = None,
    save_artifacts: bool = True,
) -> dict[str, Any]:
    """
    Evaluate model performance with confidence intervals on full datasets
    and subgroups.

    Parameters
    ----------
    X : pandas.DataFrame
        Feature dataset of shape (n_samples, n_features).
    y : pandas.Series or numpy.ndarray
        Ground truth target values of shape (n_samples,).
    outcome : str, optional
        Outcome key name. Used for logging, model resolution, and artifact
        saving.
    model : object, optional
        Explicit fitted model instance. If None, resolved from
        `runner.fitted_models`.
    metrics : list of str, optional
        List of metrics to evaluate. If None, defaults to
        `self._evaluator.metrics`.
    subgroup_specs : dict of str to (str or callable), optional
        Specifications for extracting demographic or clinical subgroups.
    fairness_data : pandas.DataFrame, optional
        Fairness stratification columns aligned with `X` (see
        `fairness_split`), used to resolve `subgroup_specs` for strata
        that are not themselves model predictors (e.g. "HOSPITAL").
        Defaults to `X` when not provided.
    save_artifacts : bool, default=True
        Whether to write evaluation summary results to disk via
        `ArtifactManager`.

    Returns
    -------
    results : dict of str to Any
        Nested dictionary containing outcome identifier, overall slice
        metrics, and subgroup performance.

    """
    # Resolve single target Series/ndarray if y is supplied as a DataFrame
    if isinstance(y, pd.DataFrame):
        if outcome and outcome in y.columns:
            y_eval = y[outcome]
        elif y.shape[1] == 1:
            y_eval = y.iloc[:, 0]
        else:
            y_eval = y
    else:
        y_eval = y

    return self._evaluator.evaluate(
        X=X,
        y=y_eval,
        outcome=outcome,
        model=model,
        metrics=metrics,
        subgroup_specs=subgroup_specs,
        fairness_data=fairness_data,
        save_artifacts=save_artifacts,
    )

run

run(groups_train: ndarray | None = None) -> dict[str, Any]

Execute full end-to-end pipeline (data preparation, model fitting, and test evaluation).

Automates data ingestion, split creation, model cross-validation and fitting, and test evaluation with TRIPOD+AI reporting across all target outcomes. The model fitting duration and total run duration are recorded at the debug log level.

Note: only models that support predict_dist, are able to generate plots.

Parameters:

Name	Type	Description	Default
groups_train	ndarray	Group labels for training samples if group-based cross-validation is configured.	None

Returns:

Name	Type	Description
pipeline_results	dict of str to Any	Dictionary containing: - "fitted_models": Dictionary mapping outcome target names to fitted model instances. - "evaluations": Dictionary mapping outcome target names to evaluation results dictionaries. - "plots": (Audit / Eval mode only) Nested dictionary mapping outcome target names to figure objects.

Source code in medpipe/pipeline/pipeline.py

def run(
    self,
    groups_train: np.ndarray | None = None,
) -> dict[str, Any]:
    """
    Execute full end-to-end pipeline (data preparation, model fitting, and
    test evaluation).

    Automates data ingestion, split creation, model cross-validation and fitting,
    and test evaluation with TRIPOD+AI reporting across all target outcomes.
    The model fitting duration and total run duration are recorded at
    the debug log level.

    Note: only models that support `predict_dist`, are able to generate plots.

    Parameters
    ----------
    groups_train : numpy.ndarray, optional
        Group labels for training samples if group-based cross-validation
        is configured.

    Returns
    -------
    pipeline_results : dict of str to Any
        Dictionary containing:
        - `"fitted_models"`: Dictionary mapping outcome target names
        to fitted model instances.
        - `"evaluations"`: Dictionary mapping outcome target names
        to evaluation results dictionaries.
        - `"plots"`: (Audit / Eval mode only) Nested dictionary mapping
        outcome target names to figure objects.

    """
    self._logger.info("Executing full MedpipeRegressor pipeline end-to-end.")
    run_start_time = time.perf_counter()

    run_mode = self.mp_config.meta.run_mode
    self._logger.debug(
        f"Executing full MedpipeRegressor pipeline in {run_mode} mode."
    )

    n_steps = 3
    if run_mode == "audit" or run_mode == "eval":
        n_steps = 4

    # 1. Prepare data splits via orchestrator
    data_kwargs = self.mp_config.data.kwargs  # Get extra data arguments
    self._logger.info(f"Step 1/{n_steps}: Ingesting and splitting dataset.")
    X_train, y_train, X_recal, y_recal, X_test, y_test, groups_train = (
        self._orchestrator.prepare_data(**data_kwargs)
    )

    # 2. Fit models via runner
    self._logger.info(f"Step 2/{n_steps}: Fitting outcome models.")
    fit_start_time = time.perf_counter()
    fitted_models = self.fit(
        X_train=X_train,
        y_train=y_train,
        X_recal=X_recal,
        y_recal=y_recal,
        groups_train=groups_train,
    )
    fit_duration = time.perf_counter() - fit_start_time
    self._logger.debug(f"Model fitting completed in {fit_duration:.2f} seconds.")

    # 3. Evaluate models on test set via evaluator
    self._logger.info(
        f"Step 3/{n_steps}: Evaluating models on holdout test set.",
    )
    evaluations: dict[str, Any] = {}
    outcomes = self._orchestrator.config.data.outcomes
    plots: dict[str, dict[str, tuple[Figure | SubFigure, Axes]]] = {}
    subgroup_specs = self._orchestrator.get_subgroup_specs()
    fairness_splits = self._orchestrator.fairness_splits
    fairness_data = fairness_splits.test if fairness_splits is not None else None

    for outcome in outcomes:
        y_test_outcome = y_test[outcome]
        evaluations[outcome] = self.evaluate(
            X=X_test,
            y=y_test_outcome.to_numpy(),
            outcome=outcome,
            subgroup_specs=subgroup_specs,
            fairness_data=fairness_data,
            save_artifacts=True,
        )

    if run_mode == "audit" or run_mode == "eval":
        self._logger.info(f"Step 4/{n_steps}: Plotting graphs.")
        for outcome in outcomes:
            if not self._supports_distribution_plots(outcome):
                self._logger.warning(
                    f"[{outcome}] Skipping the figures: the model "
                    f"{self._model_name(outcome)} does not provide a predictive "
                    "distribution that the figures support (OrdBoost "
                    "models only)."
                )
                continue
            y_true_outcome = y_test[outcome].to_numpy()
            dist_outcome = self.predict_dist(X=X_test, outcome=outcome)

            plots[outcome] = self.plot_all(
                y_true=y_true_outcome,
                dist=dist_outcome,
                outcome=outcome,
            )
        # Generate cross-outcome strata heatmaps per metric
        if fairness_splits is None:
            self._logger.info(
                "No fairness configuration set; skipping subgroup strata heatmaps."
            )
        else:
            self._logger.info(
                "Generating subgroup strata heatmaps across outcomes."
            )
            strata_heatmaps = self.displayer.plot_all_heatmaps(
                evaluations=evaluations,
            )
            plots["strata_heatmaps"] = strata_heatmaps

    run_duration = time.perf_counter() - run_start_time
    self._logger.debug(
        f"Full pipeline run completed in {run_duration:.2f} seconds."
    )
    self._logger.info(
        "Full MedpipeRegressor pipeline execution finished successfully."
    )

    results: dict[str, Any] = {
        "fitted_models": fitted_models,
        "evaluations": evaluations,
    }
    if plots:
        results["plots"] = plots

    return results

plot_all

plot_all(
    y_true: NDArray,
    dist: ContinuousPredictiveDistribution,
    outcome: str = "default",
    coverage_levels: NDArray | None = None,
    n_bootstraps: int | None = None,
    save: bool | None = None,
    show: bool | None = None,
    **style_kwargs: Any,
) -> dict[str, tuple[Figure | SubFigure, Axes]]

Execute all core distributional diagnostic visualization routines for a given outcome.

Generates and optionally persists the coverage reliability curve, sharpness curve, Winkler score curve, marginal calibration curve, PIT histogram, and target-value data distribution histogram. Only supported for models that produce a predictive CDF via predict_dist (currently OrdBoost only). The bin mapper required for the PIT histogram is extracted directly from self.models[outcome]["regressor"].mapper_, so outcome must match an entry in runner.fitted_models for the PIT histogram to render; it stays skipped-with-a-clear-error otherwise, same as when no mapper is available.

Called by run() in the "audit" and "eval" run modes for the outcomes whose model is distribution-capable (e.g. OrdBoostRegressor).

Parameters:

Name	Type	Description	Default
y_true	ndarray	Ground truth continuous target values of shape (n_samples,).	required
dist	ContinuousPredictiveDistribution	Predictive distribution for the same samples, e.g. from self.predict_dist(X, outcome=outcome).	required
outcome	str	Outcome identifier used for figure titles, output folder structuring, and to resolve the fitted model (and its bin mapper) from runner.fitted_models.	"default"
coverage_levels	ndarray	Nominal central-interval coverage levels in percent, used for the coverage, sharpness, and Winkler score curves. Defaults to np.arange(5, 100, 10).	None
n_bootstraps	int	Number of bootstrap iterations for coverage, sharpness, and Winkler score curves. If None, resolved from the display configuration.	None
save	bool	Automatically save all generated plot artifacts to the run directory. If None, resolved from the display configuration.	None
show	bool	Whether to display figures interactively before closing. If None, resolved from the display configuration.	None
**style_kwargs	Any	Additional style parameters forwarded to underlying drawing primitives.	{}

Returns:

Name	Type	Description
plots	dict of str to tuple of (matplotlib.figure.Figure, matplotlib.axes.Axes)	Dictionary mapping plot keys ('coverage', 'sharpness', 'winkler', 'marginal_calibration', 'pit_histogram', 'data_distribution') to their rendered (Figure, Axes) Matplotlib objects.

Source code in medpipe/pipeline/pipeline.py

def plot_all(
    self,
    y_true: npt.NDArray,
    dist: ContinuousPredictiveDistribution,
    outcome: str = "default",
    coverage_levels: npt.NDArray | None = None,
    n_bootstraps: int | None = None,
    save: bool | None = None,
    show: bool | None = None,
    **style_kwargs: Any,
) -> dict[str, tuple[Figure | SubFigure, Axes]]:
    """
    Execute all core distributional diagnostic visualization routines
    for a given outcome.

    Generates and optionally persists the coverage reliability curve,
    sharpness curve, Winkler score curve, marginal calibration curve,
    PIT histogram, and target-value data distribution histogram. Only
    supported for models that produce a predictive CDF via
    `predict_dist` (currently OrdBoost only). The
    bin mapper required for the PIT histogram is extracted directly
    from `self.models[outcome]["regressor"].mapper_`, so `outcome` must
    match an entry in `runner.fitted_models` for the PIT histogram to
    render; it stays skipped-with-a-clear-error otherwise, same as
    when no mapper is available.

    Called by `run()` in the "audit" and "eval" run modes for the outcomes
    whose model is distribution-capable (e.g. `OrdBoostRegressor`).

    Parameters
    ----------
    y_true : numpy.ndarray
        Ground truth continuous target values of shape (n_samples,).
    dist : ContinuousPredictiveDistribution
        Predictive distribution for the same samples, e.g. from
        `self.predict_dist(X, outcome=outcome)`.
    outcome : str, default="default"
        Outcome identifier used for figure titles, output folder
        structuring, and to resolve the fitted model (and its bin
        mapper) from `runner.fitted_models`.
    coverage_levels : numpy.ndarray, optional
        Nominal central-interval coverage levels in percent, used for the
        coverage, sharpness, and Winkler score curves. Defaults to
        `np.arange(5, 100, 10)`.
    n_bootstraps : int, optional
        Number of bootstrap iterations for coverage, sharpness, and
        Winkler score curves. If None, resolved from the display
        configuration.
    save : bool, optional
        Automatically save all generated plot artifacts to the run directory.
        If None, resolved from the display configuration.
    show : bool, optional
        Whether to display figures interactively before closing. If None,
        resolved from the display configuration.
    **style_kwargs : Any
        Additional style parameters forwarded to underlying drawing primitives.

    Returns
    -------
    plots : dict of str to tuple of (matplotlib.figure.Figure, matplotlib.axes.Axes)
        Dictionary mapping plot keys
        ('coverage', 'sharpness', 'winkler', 'marginal_calibration',
        'pit_histogram', 'data_distribution') to their rendered (Figure,
        Axes) Matplotlib objects.

    """
    fitted_model = self.models.get(outcome)
    mapper = (
        getattr(fitted_model["regressor"], "mapper_", None)
        if fitted_model is not None
        else None
    )

    return self.displayer.plot_all(
        y_true=y_true,
        dist=dist,
        mapper=mapper,
        outcome=outcome,
        coverage_levels=coverage_levels,
        n_bootstraps=n_bootstraps,
        save=save,
        show=show,
        **style_kwargs,
    )

load classmethod

load(run_dir: str | Path) -> MedpipeRegressor

Reconstruct a MedpipeRegressor instance from a run artifact directory.

Parses the saved JSON configuration and restores serialized outcome model artifacts into the runner and evaluator engines.

Parameters:

Name	Type	Description	Default
run_dir	str or Path	Directory path of a previously executed MedpipeRegressor run artifact.	required

Returns:

Name	Type	Description
pipe	MedpipeRegressor	Re-instantiated MedpipeRegressor object ready for inference, evaluation, or visualization.

Raises:

Type	Description
FileNotFoundError	If the resolved configuration JSON file is missing from the run directory.

Source code in medpipe/pipeline/pipeline.py

@classmethod
def load(cls, run_dir: str | Path) -> MedpipeRegressor:
    """Reconstruct a MedpipeRegressor instance from a run artifact directory.

    Parses the saved JSON configuration and restores serialized outcome
    model artifacts into the runner and evaluator engines.

    Parameters
    ----------
    run_dir : str or Path
        Directory path of a previously executed MedpipeRegressor run artifact.

    Returns
    -------
    pipe : MedpipeRegressor
        Re-instantiated MedpipeRegressor object ready for inference,
        evaluation, or visualization.

    Raises
    ------
    FileNotFoundError
        If the resolved configuration JSON file is missing from the run directory.

    """
    run_path = Path(run_dir)
    config_path = run_path / "env/resolved_config.json"
    models_dir = run_path / "models"

    if not config_path.exists():
        raise FileNotFoundError(
            "Cannot load MedpipeRegressor instance: Configuration JSON missing "
            f"in '{run_path}'"
        )

    # 1. Load JSON dict and instantiate MedpipeRegressorConfig
    import json

    import joblib

    with open(config_path, encoding="utf-8") as f:
        config_dict = json.load(f)

    mp_config = MedpipeRegressorConfig.model_validate(config_dict)

    # 2. Instantiate MedpipeRegressor with reconstructed MedpipeRegressorConfig
    new_run_path = run_path / "eval"
    pipe = cls(config=mp_config, base_artifact_dir=new_run_path)
    pipe._orchestrator.run_dir = new_run_path
    pipe.displayer.run_dir = new_run_path

    # 3. Restore serialized model binaries into runner and evaluator engines
    if models_dir.exists():
        project_name = pipe.mp_config.meta.project_name
        pipe._runner.fitted_models = joblib.load(
            models_dir / f"{project_name}_fitted.joblib"
        )
        pipe._evaluator.fitted_models = pipe._runner.fitted_models
    pipe._logger.info(f"Successfully loaded MedpipeRegressor from {run_dir}")

    return pipe

orchestrator

DataSplits dataclass

Container holding prepared feature and target splits.

Attributes:

Name	Type	Description
X_train	DataFrame	Samples used for training.
y_train	DataFrame	Training labels.
X_recal	DataFrame | None	Samples used for recalibration, if available.
y_recal	DataFrame | None	Recalibration labels, if available.
X_test	DataFrame	Samples used for testing.
y_test	DataFrame	Testing labels.
groups_train	NDArray	Groups used for splitting the training set, if available.

Source code in medpipe/pipeline/orchestrator.py

@dataclass(frozen=True)
class DataSplits:
    """Container holding prepared feature and target splits.

    Attributes
    ----------
    X_train : pd.DataFrame
        Samples used for training.
    y_train : pd.DataFrame
        Training labels.
    X_recal : pd.DataFrame | None
        Samples used for recalibration, if available.
    y_recal : pd.DataFrame | None
        Recalibration labels, if available.
    X_test : pd.DataFrame
        Samples used for testing.
    y_test : pd.DataFrame
        Testing labels.
    groups_train : NDArray
        Groups used for splitting the training set, if available.

    """

    X_train: pd.DataFrame
    y_train: pd.DataFrame
    X_test: pd.DataFrame
    y_test: pd.DataFrame
    X_recal: pd.DataFrame | None = None
    y_recal: pd.DataFrame | None = None
    groups_train: NDArray | None = None

FairnessSplits dataclass

Container holding fairness stratification columns aligned with each prepared data split.

Populated whenever workflow.evaluation.fairness.strata is configured. Holds every fairness stratum column exactly as ingested, independent of whether a given column is also a model predictor, so fairness columns (e.g. a "HOSPITAL" column used only for a spatial comparison) are always available here without ever being merged into DataSplits' feature frames.

Attributes:

Name	Type	Description
train	DataFrame	Fairness stratum columns for the rows in DataSplits.X_train.
test	DataFrame	Fairness stratum columns for the rows in DataSplits.X_test.
recal	DataFrame | None	Fairness stratum columns for the rows in DataSplits.X_recal, if a recalibration split is configured.

Source code in medpipe/pipeline/orchestrator.py

@dataclass(frozen=True)
class FairnessSplits:
    """Container holding fairness stratification columns aligned with each
    prepared data split.

    Populated whenever `workflow.evaluation.fairness.strata` is configured.
    Holds every fairness stratum column exactly as ingested, independent of
    whether a given column is also a model predictor, so fairness columns
    (e.g. a "HOSPITAL" column used only for a spatial comparison) are always
    available here without ever being merged into `DataSplits`' feature
    frames.

    Attributes
    ----------
    train : pd.DataFrame
        Fairness stratum columns for the rows in `DataSplits.X_train`.
    test : pd.DataFrame
        Fairness stratum columns for the rows in `DataSplits.X_test`.
    recal : pd.DataFrame | None
        Fairness stratum columns for the rows in `DataSplits.X_recal`, if
        a recalibration split is configured.

    """

    train: pd.DataFrame
    test: pd.DataFrame
    recal: pd.DataFrame | None = None

MedpipeOrchestrator

Handles data ingress, transformation preparation, and reproducibility management.

This orchestrator acts as the primary entry point for setting up the machine learning pipeline. It establishes the reproducibility environment, configures logging, ingests raw data, builds the scikit-learn preprocessing pipeline, and resolves configuration overrides and data splits.

Parameters:

Name	Type	Description	Default
config	Union[str, Path, MedpipeClassifierConfig, MedpipeRegressorConfig]	Path to a TOML configuration file, or an already-instantiated MedpipeClassifierConfig or MedpipeRegressorConfig object. A bare path string/Path is parsed via read_classifier_toml_configuration or read_regressor_toml_configuration depending on is_classifier.	required
base_artifact_dir	Union[str, Path]	Root directory where the versioned run artifacts and logs will be saved.	"artifacts"
verbose_override	Union[bool, int, str, None]	Console verbosity setting configuration override.	None
is_classifier	bool	Whether config (when passed as a path) should be parsed as a MedpipeClassifierConfig (True) or a MedpipeRegressorConfig (False). Ignored when config is already a config object.	True

Attributes:

Name	Type	Description
config	MedpipeClassifierConfig	The resolved configuration object driving the pipeline.
artifact_manager	ArtifactManager	Manager handling the creation and population of reproducibility artifacts.
run_dir	Path	The specific, versioned directory path for the current execution run.
logger	Logger	The configured logger instance for the orchestrator, routing to both console and the artifact directory.

Methods:

Name	Description
ingest_data	Loads and validates the raw dataset specified in the configuration.
prepare_data	Extracts labels and applies sequential splits for test and recalibration sets.
extract_stratum_subgroup	Extract a stratified subgroup from features (and optional target labels).
build_preprocessor	Constructs an sklearn Pipeline for data transformation.
_save_reproducibility_artifacts	Persists the resolved configuration, original TOML file, and environment state.
_check_operation	Validates and retrieves a preprocessing operation.

Source code in medpipe/pipeline/orchestrator.py

class MedpipeOrchestrator:
    """
    Handles data ingress, transformation preparation, and reproducibility management.

    This orchestrator acts as the primary entry point for setting up the machine
    learning pipeline. It establishes the reproducibility environment, configures
    logging, ingests raw data, builds the scikit-learn preprocessing pipeline,
    and resolves configuration overrides and data splits.

    Parameters
    ----------
    config : Union[str, Path, MedpipeClassifierConfig, MedpipeRegressorConfig]
        Path to a TOML configuration file, or an already-instantiated
        `MedpipeClassifierConfig` or `MedpipeRegressorConfig` object. A bare
        path string/Path is parsed via `read_classifier_toml_configuration`
        or `read_regressor_toml_configuration` depending on `is_classifier`.
    base_artifact_dir : Union[str, Path], default="artifacts"
        Root directory where the versioned run artifacts and logs will be saved.
    verbose_override : Union[bool, int, str, None], default=None
        Console verbosity setting configuration override.
    is_classifier : bool, default=True
        Whether `config` (when passed as a path) should be parsed as a
        `MedpipeClassifierConfig` (True) or a `MedpipeRegressorConfig`
        (False). Ignored when `config` is already a config object.

    Attributes
    ----------
    config : MedpipeClassifierConfig
        The resolved configuration object driving the pipeline.
    artifact_manager : ArtifactManager
        Manager handling the creation and population of reproducibility artifacts.
    run_dir : Path
        The specific, versioned directory path for the current execution run.
    logger : logging.Logger
        The configured logger instance for the orchestrator, routing to both
        console and the artifact directory.

    Methods
    -------
    ingest_data(**kwargs)
        Loads and validates the raw dataset specified in the configuration.
    prepare_data(**kwargs)
        Extracts labels and applies sequential splits for test and recalibration sets.
    extract_stratum_subgroup(X, column, group, y=None)
        Extract a stratified subgroup from features (and optional target labels).
    build_preprocessor()
        Constructs an sklearn Pipeline for data transformation.
    _save_reproducibility_artifacts()
        Persists the resolved configuration, original TOML file, and
        environment state.
    _check_operation(op)
        Validates and retrieves a preprocessing operation.

    """

    def __init__(
        self,
        config: str | Path | MedpipeClassifierConfig | MedpipeRegressorConfig,
        base_artifact_dir: str | Path = "artifacts",
        verbose_override: bool | int | str | None = None,
        is_classifier: bool = True,
    ) -> None:
        self._config_path: Path | None = None
        if isinstance(config, (str, Path)):
            self.config = (
                read_classifier_toml_configuration(config)
                if is_classifier
                else read_regressor_toml_configuration(config)
            )
            self._config_path = Path(config)
        elif isinstance(config, (MedpipeClassifierConfig, MedpipeRegressorConfig)):
            self.config = config
        else:
            raise ValueError(
                "A configuration file, a MedpipeClassifierConfig, or a "
                "MedpipeRegressorConfig must be specified."
            )

        if verbose_override is not None:
            set_verbosity(verbose_override)
        else:
            set_verbosity(self.config.meta.verbose)
        self.artifact_manager = ArtifactManager(base_artifact_dir)
        self.run_dir = self.artifact_manager.create_run_directory()

        self.logger = get_console_logger("medpipe.orchestrator")
        add_file_handler(self.logger, log_dir=self.run_dir)

        self.logger.info(
            f"Initialised MedpipeOrchestrator. Run directory: {self.run_dir}"
        )

        self._splits: DataSplits | None = None  # Holds the data splits
        self._fairness_splits: FairnessSplits | None = None
        self._fairness_raw: pd.DataFrame | None = None
        self._save_reproducibility_artifacts()

    @property
    def splits(self) -> DataSplits:
        """Access cached data splits.

        Returns
        ------
        DataSplits
            Access to the DataSplits attributes.

        Raises
        ------
        RuntimeError
            If the splits have not been populated and 'prepare_data()'
            or 'run()' need to be called.

        """
        if self._splits is None:
            raise RuntimeError(
                "Data has not been prepared yet. Call 'prepare_data()' "
                "or 'run()' before accessing dataset splits."
            )
        return self._splits

    @property
    def fairness_splits(self) -> FairnessSplits | None:
        """Access cached fairness stratification splits.

        Returns
        -------
        FairnessSplits or None
            Access to the FairnessSplits attributes, or None if no
            `workflow.evaluation.fairness` configuration is set.

        Raises
        ------
        RuntimeError
            If data has not been prepared yet and 'prepare_data()' or
            'run()' need to be called.

        """
        if self._splits is None:
            raise RuntimeError(
                "Data has not been prepared yet. Call 'prepare_data()' "
                "or 'run()' before accessing fairness splits."
            )
        return self._fairness_splits

    def _save_reproducibility_artifacts(self) -> None:
        """
        Saves the resolved configuration, original TOML file, and
        environment state to the artifact directory.

        This method extracts the configuration state (handling different Pydantic
        versions) and writes it to disk alongside the runtime environment metadata.
        If the orchestrator was initialized from a TOML file, a copy of that
        original file is also saved for easy inspection and reuse.

        """
        config_dict = (
            self.config.model_dump()
            if hasattr(self.config, "model_dump")
            else dict(self.config)
        )

        dataset_path = self.config.data.path if hasattr(self.config, "data") else None
        dest_dir = self.run_dir / "env"

        self.artifact_manager.save_env_state(
            destination_dir=dest_dir,
            config=config_dict,
            dataset_path=dataset_path,
        )
        self.artifact_manager.save_resolved_config(config_dict, dest_dir)

        if self._config_path is not None:
            self.artifact_manager.save_toml_config(self._config_path, dest_dir)
        self.logger.info("Reproducibility artifacts saved successfully.")

    def prepare_data(
        self, **kwargs
    ) -> tuple[
        pd.DataFrame,
        pd.DataFrame,
        pd.DataFrame | None,
        pd.DataFrame | None,
        pd.DataFrame,
        pd.DataFrame,
        NDArray | None,
    ]:
        """
        Ingests data, extracts labels, and performs configured train/recal/test splits.

        Also populates `fairness_splits` with the configured fairness
        stratification columns (see `FairnessSplits`), aligned to the same
        row indices as the returned `X_train`/`X_test`/`X_recal`.

        Parameters
        ----------
        **kwargs : Any
            Additional keyword arguments passed to the underlying Pandas reader function
            (e.g., `sheet_name` for Excel, `sep` for CSV).

        Returns
        -------
        X_train : pd.DataFrame
        y_train : pd.DataFrame
        X_recal : Optional[pd.DataFrame]
        y_recal : Optional[pd.DataFrame]
        X_test : pd.DataFrame
        y_test : pd.DataFrame
        groups_train : Optional[npt.NDArray]

        """
        data = self.ingest_data(**kwargs)
        outcomes = self.config.data.outcomes
        outcome_columns = pd.Index(outcomes)

        X, y_arr = extract_labels(data, outcomes)

        val_config = getattr(self.config.workflow, "validation", None)
        if not val_config:
            raise ValueError("Validation configuration is missing from workflow.")

        # 1. Apply Test Split
        test_cfg = val_config.test_split
        X_temp, y_temp_arr, X_test, y_test_arr = split_data(
            features=X,
            labels=y_arr,
            strategy=test_cfg.strategy,
            group_column=getattr(test_cfg, "group_column", None),
            values=getattr(test_cfg, "values", None),
            test_size=getattr(test_cfg, "test_size", None),
            random_state=self.config.workflow.random_state,
        )

        # Re-wrap multi-label arrays into DataFrames aligned with their X indices
        y_temp_df = pd.DataFrame(
            y_temp_arr, columns=outcome_columns, index=X_temp.index
        )
        y_test_df = pd.DataFrame(
            y_test_arr, columns=outcome_columns, index=X_test.index
        )

        # Apply Recalibration Split (Optional)
        recal_cfg = getattr(val_config, "recalibration_split", None)

        if recal_cfg:
            X_train, y_train_arr, X_recal, y_recal_arr = split_data(
                features=X_temp,
                labels=y_temp_df.to_numpy(),
                strategy=recal_cfg.strategy,
                group_column=getattr(recal_cfg, "group_column", None),
                values=getattr(recal_cfg, "values", None),
                recalibration_size=getattr(recal_cfg, "recalibration_size", None),
                random_state=self.config.workflow.random_state,
            )
            y_train_df = pd.DataFrame(
                y_train_arr, columns=outcome_columns, index=X_train.index
            )
            y_recal_df = pd.DataFrame(
                y_recal_arr, columns=outcome_columns, index=X_recal.index
            )
            self.logger.info(
                f"Data successfully split into Train ({len(X_train):,}), "
                f"Test ({len(X_test):,}), and "
                f"Recalibration ({len(X_recal):,}) sets."
            )
            self.logger.debug(f"Train set initial shape: {X_train.shape}")
            self.logger.debug(f"Test set initial shape: {X_test.shape}")
            self.logger.debug(f"Recalibration set initial shape: {X_recal.shape}")

        else:
            X_train = X_temp
            y_train_df = y_temp_df
            X_recal = None
            y_recal_df = None

            self.logger.info(
                f"Data successfully split into Train ({len(X_train):,}), and "
                f"Test ({len(X_test):,}) sets."
            )
            self.logger.debug(f"Train set initial shape: {X_train.shape}")
            self.logger.debug(f"Test set initial shape: {X_test.shape}")

        # Extract groups if required
        groups_train = None
        cv_cfg = getattr(val_config, "cross_validation", None)
        if cv_cfg and cv_cfg.group_column:
            if cv_cfg.group_column in X_train.columns:
                X_train, groups_train = extract_labels(X_train, [cv_cfg.group_column])
                groups_train = asarray(groups_train).ravel()
            else:
                raise KeyError(
                    f"Cross-validation group column '{cv_cfg.group_column}' "
                    f"was not found in dataset columns."
                )

            self.logger.info("Data successfully prepared with CV groups.")
            self.logger.debug(
                f"Train set shape after CV groups removed: {X_train.shape}",
            )

        val_columns = self._get_validation_columns()
        if len(val_columns) != 0:
            # Drop the columns from the validation configuration
            X_train = X_train.drop(columns=val_columns, errors="ignore")
            X_test = X_test.drop(columns=val_columns)
            if X_recal is not None:
                X_recal = X_recal.drop(columns=val_columns)

            self.logger.info(f"Removed {val_columns} from data.")

        # Final shapes printed to debug
        self.logger.debug(f"Train set final shape: {X_train.shape}")
        self.logger.debug(f"Test set final shape: {X_test.shape}")
        if X_recal is not None:
            self.logger.debug(f"Recalibration set final shape: {X_recal.shape}")

        self._splits = DataSplits(  # Populate DataSplit object
            X_train=X_train,
            y_train=y_train_df,
            X_recal=X_recal,
            y_recal=y_recal_df,
            X_test=X_test,
            y_test=y_test_df,
            groups_train=groups_train,
        )

        # Fairness stratification columns (e.g. a "HOSPITAL" column used
        # only for a spatial subgroup comparison) are ingested independently
        # of `data.predictors` and are never merged into the feature
        # frames above; align that raw slice with each split's row index so
        # it stays available for fairness evaluation.
        self._fairness_splits = (
            FairnessSplits(
                train=self._fairness_raw.loc[X_train.index].copy(),
                test=self._fairness_raw.loc[X_test.index].copy(),
                recal=(
                    self._fairness_raw.loc[X_recal.index].copy()
                    if X_recal is not None
                    else None
                ),
            )
            if self._fairness_raw is not None
            else None
        )

        return X_train, y_train_df, X_recal, y_recal_df, X_test, y_test_df, groups_train

    def ingest_data(self, **kwargs) -> pd.DataFrame:
        """
        Loads the raw dataset specified in the configuration and filters it
        to retain only predictors, outcomes, and configured group columns.

        Fairness stratification columns (`workflow.evaluation.fairness.strata`)
        are validated against the raw dataset here too, but are kept out of
        the returned frame; they are cached separately (`_fairness_raw`) for
        `prepare_data()` to align into `FairnessSplits`.

        Parameters
        ----------
        **kwargs : Any
            Additional keyword arguments passed to the underlying Pandas reader function
            (e.g., `sheet_name` for Excel, `sep` for CSV).

        Returns
        -------
        pd.DataFrame
            The loaded and filtered dataset.

        Raises
        ------
        TypeError
            If the loaded data is not a pandas DataFrame.
        KeyError
            If any configured required or fairness stratum column is
            missing from the raw dataset.

        """
        dataset_path = self.config.data.path
        self.logger.info(f"Ingesting data from {dataset_path}")

        data = load_data(dataset_path, **kwargs)
        if not isinstance(data, pd.DataFrame):
            raise TypeError(
                f"Input data should be a pd.DataFrame, but got {type(data)}"
            )

        self.logger.debug(f"Loaded data shape: {data.shape}")

        # Collect required columns (predictors & outcomes)
        required_cols: list[str] = []
        if hasattr(self.config, "data"):
            required_cols.extend(getattr(self.config.data, "predictors", []))
            required_cols.extend(getattr(self.config.data, "outcomes", []))

        # Collect group columns from validation splits (test, recalibration,
        # cross-validation)
        required_cols += self._get_validation_columns()

        # Deduplicate while preserving order
        unique_cols = list(dict.fromkeys(required_cols))

        # Fairness stratification columns (e.g. a "HOSPITAL" column used
        # only for a spatial subgroup comparison) are validated up front
        # here too, so a missing column fails fast instead of only once
        # evaluation runs later — but they are never merged into the
        # modelling frame below; they are kept as a separate raw slice
        # (see `_fairness_raw`) so they never reach the model.
        fairness_cols = self._get_fairness_columns()

        # Validate column existence
        missing_cols = [
            col
            for col in list(dict.fromkeys(unique_cols + fairness_cols))
            if col not in data.columns
        ]
        if missing_cols:
            raise KeyError(
                "The following required columns were missing from the "
                f"dataset: {missing_cols}"
            )

        self._fairness_raw = data[fairness_cols].copy() if fairness_cols else None

        # Filter out unneeded columns
        filtered_data = data[unique_cols].copy()
        self.logger.info(
            f"Filtered dataset from {len(data.columns)} down to "
            f"{len(unique_cols)} required columns."
        )
        self.logger.debug(f"Filtered data shape: {filtered_data.shape}")

        return pd.DataFrame(filtered_data)

    def _get_validation_columns(self) -> list[str]:
        """
        Collect group columns from validation splits
        (test, recalibration, cross-validation)

        Returns
        -------
        unique_cols : list[str]
            Unique columns from the validation configuration.

        """
        val_columns = []
        # Collect group columns from validation splits (test, recalibration,
        # cross-validation)
        val_cfg = getattr(getattr(self.config, "workflow", None), "validation", None)
        if val_cfg:
            for split_name in ["test_split", "recalibration_split", "cross_validation"]:
                split = getattr(val_cfg, split_name, None)
                if split and getattr(split, "group_column", None):
                    val_columns.append(split.group_column)

        unique_cols = list(dict.fromkeys(val_columns))

        return unique_cols

    def _get_fairness_columns(self) -> list[str]:
        """
        Collect stratification columns from the fairness configuration.

        These columns may or may not also be model predictors (e.g. a
        "HOSPITAL" column used only for a spatial subgroup comparison), so
        they are collected separately to be validated and retained
        independently of `data.predictors`.

        Returns
        -------
        unique_cols : list of str
            Unique fairness stratum column names, or an empty list if no
            fairness configuration is set.

        """
        eval_cfg = getattr(getattr(self.config, "workflow", None), "evaluation", None)
        fairness_cfg = getattr(eval_cfg, "fairness", None)
        strata = getattr(fairness_cfg, "strata", None) or []

        return list(dict.fromkeys(strata))

    def get_subgroup_specs(self) -> dict[str, Any]:
        """
        Parses `workflow.evaluation.fairness` settings into a dictionary
        compatible with `MedpipeClassifierEvaluator.extract_subgroups`.

        Returns
        -------
        subgroup_specs : dict of str to Any
            Maps stratum column names either to column strings
            (for discrete categorical groupings) or to lists of range bounds
            defined under `groups`.

        """
        eval_cfg = getattr(getattr(self.config, "workflow", None), "evaluation", None)
        fairness_cfg = getattr(eval_cfg, "fairness", None)

        if not fairness_cfg or not getattr(fairness_cfg, "strata", None):
            return {}

        strata = fairness_cfg.strata
        custom_groups = getattr(fairness_cfg, "groups", {}) or {}

        subgroup_specs = {}
        for column in strata:
            if column in custom_groups:
                # Custom ranges e.g. {"AGE": [[18, 50], [51, 120]]}
                subgroup_specs[column] = custom_groups[column]
            else:
                # Column name string for discrete categorical groupby (e.g. "SEX")
                subgroup_specs[column] = column

        return subgroup_specs

    def extract_stratum_subgroup(
        self,
        X: pd.DataFrame,
        column: str,
        group: Any,
        y: pd.DataFrame | None = None,
    ) -> tuple[pd.DataFrame, pd.DataFrame | None]:
        """
        Extract a stratified subgroup from features (and optional target labels).

        Resolves continuous range bounds or discrete categorical identifiers.

        Parameters
        ----------
        X : pd.DataFrame
            Feature matrix containing the stratification column.
        column : str
            Name of the column defining the stratum (e.g., 'AGE', 'SEX').
        group : Any
            The specific subgroup identifier or range (e.g., 'M', (18, 65), '[18, 65)').
        y : pd.DataFrame, optional
            Corresponding ground truth label DataFrame aligned with X. Default is None.

        Returns
        -------
        X_subgroup : pd.DataFrame
            Filtered feature DataFrame for the specified stratum group.
        y_subgroup : pd.DataFrame or None
            Filtered label DataFrame matching X_subgroup index, or None if y
            was not provided.

        """
        self.logger.info(
            f"Extracting subgroup for stratum '{column}' matching group: {group}"
        )

        mask = resolve_subgroup_mask(X, column=column, group=group)
        n_matched = mask.sum()

        if n_matched == 0:
            self.logger.warning(
                f"Subgroup extraction returned 0 samples for column '{column}' "
                f"and group '{group}'."
            )

        X_subgroup = X.loc[mask].copy()
        y_subgroup = y.loc[mask].copy() if y is not None else None

        self.logger.debug(
            f"{len(mask)} samples in stratum '{column}' matching group: {group}",
        )
        return X_subgroup, y_subgroup

    def build_preprocessor(self) -> Pipeline | None:
        """
        Constructs an sklearn Pipeline for data transformation based on the
        configuration.

        Parses the workflow configuration to build a sequence of `ColumnTransformer`
        objects, mapping specific operations to requested columns.

        Returns
        -------
        Optional[Pipeline]
            A configured scikit-learn `Pipeline` containing the cascaded
            transformations, or `None` if preprocessing is disabled or omitted.

        """
        preprocessing_dict = self.config.workflow.preprocessing

        if preprocessing_dict and preprocessing_dict.preprocess:
            self.logger.info("Constructing preprocessing pipeline.")
            steps = []

            if preprocessing_dict.operations:
                ct_columns_dict = {pred: pred for pred in self.config.data.predictors}

                for i, operation in enumerate(preprocessing_dict.operations):
                    op_type = self._check_operation(operation.name)
                    op_extras = (
                        {} if not operation.model_extra else operation.model_extra
                    )

                    unknown = [
                        column
                        for column in operation.columns
                        if column not in ct_columns_dict
                    ]
                    if unknown:
                        raise ValueError(
                            f"Preprocessing operation '{operation.name}' refers to "
                            f"columns that are not in data.predictors: {unknown}"
                        )
                    ct_columns = [
                        ct_columns_dict[column] for column in operation.columns
                    ]

                    self.logger.debug(
                        f"Creating operation {i + 1}: {operation} on columns "
                        f"{ct_columns}"
                    )
                    ct = ColumnTransformer(
                        [(f"op_{i + 1}", op_type(**op_extras), ct_columns)],
                        remainder="passthrough",
                    )
                    ct.set_output(transform="pandas")

                    if i == len(preprocessing_dict.operations) - 1:
                        ct.set_output(transform="default")

                    steps.append((f"transformer_{i + 1}", ct))

                    ct_columns_dict = {
                        pred: (
                            f"remainder__{column}"
                            if column not in ct_columns
                            else f"op_{i + 1}__{column}"
                        )
                        for (pred, column) in ct_columns_dict.items()
                    }

            return Pipeline(steps=steps)

        self.logger.info(
            "No preprocessing steps specified; skipping pipeline creation."
        )
        return None

    def _check_operation(self, op: str) -> type:
        """
        Validates whether a requested preprocessing operation exists.

        Delegates the lookup to the `PreprocessorRegistry`.

        Parameters
        ----------
        op : str
            The string name of the operation to validate and retrieve.

        Returns
        -------
        type
            The uninstantiated class for the preprocessing operation.

        Raises
        ------
        ValueError
            If the operation cannot be found in the registry or scikit-learn.

        """
        return PreprocessorRegistry.get(op)

splits property

splits: DataSplits

Access cached data splits.

Returns:

Type	Description
DataSplits	Access to the DataSplits attributes.

Raises:

Type	Description
RuntimeError	If the splits have not been populated and 'prepare_data()' or 'run()' need to be called.

fairness_splits property

fairness_splits: FairnessSplits | None

Access cached fairness stratification splits.

Returns:

Type	Description
FairnessSplits or None	Access to the FairnessSplits attributes, or None if no workflow.evaluation.fairness configuration is set.

Raises:

Type	Description
RuntimeError	If data has not been prepared yet and 'prepare_data()' or 'run()' need to be called.

prepare_data

prepare_data(
    **kwargs,
) -> tuple[
    DataFrame,
    DataFrame,
    DataFrame | None,
    DataFrame | None,
    DataFrame,
    DataFrame,
    NDArray | None,
]

Ingests data, extracts labels, and performs configured train/recal/test splits.

Also populates fairness_splits with the configured fairness stratification columns (see FairnessSplits), aligned to the same row indices as the returned X_train/X_test/X_recal.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Additional keyword arguments passed to the underlying Pandas reader function (e.g., sheet_name for Excel, sep for CSV).	{}

Returns:

Name	Type	Description
X_train	DataFrame
y_train	DataFrame
X_recal	Optional[DataFrame]
y_recal	Optional[DataFrame]
X_test	DataFrame
y_test	DataFrame
groups_train	Optional[NDArray]

Source code in medpipe/pipeline/orchestrator.py

def prepare_data(
    self, **kwargs
) -> tuple[
    pd.DataFrame,
    pd.DataFrame,
    pd.DataFrame | None,
    pd.DataFrame | None,
    pd.DataFrame,
    pd.DataFrame,
    NDArray | None,
]:
    """
    Ingests data, extracts labels, and performs configured train/recal/test splits.

    Also populates `fairness_splits` with the configured fairness
    stratification columns (see `FairnessSplits`), aligned to the same
    row indices as the returned `X_train`/`X_test`/`X_recal`.

    Parameters
    ----------
    **kwargs : Any
        Additional keyword arguments passed to the underlying Pandas reader function
        (e.g., `sheet_name` for Excel, `sep` for CSV).

    Returns
    -------
    X_train : pd.DataFrame
    y_train : pd.DataFrame
    X_recal : Optional[pd.DataFrame]
    y_recal : Optional[pd.DataFrame]
    X_test : pd.DataFrame
    y_test : pd.DataFrame
    groups_train : Optional[npt.NDArray]

    """
    data = self.ingest_data(**kwargs)
    outcomes = self.config.data.outcomes
    outcome_columns = pd.Index(outcomes)

    X, y_arr = extract_labels(data, outcomes)

    val_config = getattr(self.config.workflow, "validation", None)
    if not val_config:
        raise ValueError("Validation configuration is missing from workflow.")

    # 1. Apply Test Split
    test_cfg = val_config.test_split
    X_temp, y_temp_arr, X_test, y_test_arr = split_data(
        features=X,
        labels=y_arr,
        strategy=test_cfg.strategy,
        group_column=getattr(test_cfg, "group_column", None),
        values=getattr(test_cfg, "values", None),
        test_size=getattr(test_cfg, "test_size", None),
        random_state=self.config.workflow.random_state,
    )

    # Re-wrap multi-label arrays into DataFrames aligned with their X indices
    y_temp_df = pd.DataFrame(
        y_temp_arr, columns=outcome_columns, index=X_temp.index
    )
    y_test_df = pd.DataFrame(
        y_test_arr, columns=outcome_columns, index=X_test.index
    )

    # Apply Recalibration Split (Optional)
    recal_cfg = getattr(val_config, "recalibration_split", None)

    if recal_cfg:
        X_train, y_train_arr, X_recal, y_recal_arr = split_data(
            features=X_temp,
            labels=y_temp_df.to_numpy(),
            strategy=recal_cfg.strategy,
            group_column=getattr(recal_cfg, "group_column", None),
            values=getattr(recal_cfg, "values", None),
            recalibration_size=getattr(recal_cfg, "recalibration_size", None),
            random_state=self.config.workflow.random_state,
        )
        y_train_df = pd.DataFrame(
            y_train_arr, columns=outcome_columns, index=X_train.index
        )
        y_recal_df = pd.DataFrame(
            y_recal_arr, columns=outcome_columns, index=X_recal.index
        )
        self.logger.info(
            f"Data successfully split into Train ({len(X_train):,}), "
            f"Test ({len(X_test):,}), and "
            f"Recalibration ({len(X_recal):,}) sets."
        )
        self.logger.debug(f"Train set initial shape: {X_train.shape}")
        self.logger.debug(f"Test set initial shape: {X_test.shape}")
        self.logger.debug(f"Recalibration set initial shape: {X_recal.shape}")

    else:
        X_train = X_temp
        y_train_df = y_temp_df
        X_recal = None
        y_recal_df = None

        self.logger.info(
            f"Data successfully split into Train ({len(X_train):,}), and "
            f"Test ({len(X_test):,}) sets."
        )
        self.logger.debug(f"Train set initial shape: {X_train.shape}")
        self.logger.debug(f"Test set initial shape: {X_test.shape}")

    # Extract groups if required
    groups_train = None
    cv_cfg = getattr(val_config, "cross_validation", None)
    if cv_cfg and cv_cfg.group_column:
        if cv_cfg.group_column in X_train.columns:
            X_train, groups_train = extract_labels(X_train, [cv_cfg.group_column])
            groups_train = asarray(groups_train).ravel()
        else:
            raise KeyError(
                f"Cross-validation group column '{cv_cfg.group_column}' "
                f"was not found in dataset columns."
            )

        self.logger.info("Data successfully prepared with CV groups.")
        self.logger.debug(
            f"Train set shape after CV groups removed: {X_train.shape}",
        )

    val_columns = self._get_validation_columns()
    if len(val_columns) != 0:
        # Drop the columns from the validation configuration
        X_train = X_train.drop(columns=val_columns, errors="ignore")
        X_test = X_test.drop(columns=val_columns)
        if X_recal is not None:
            X_recal = X_recal.drop(columns=val_columns)

        self.logger.info(f"Removed {val_columns} from data.")

    # Final shapes printed to debug
    self.logger.debug(f"Train set final shape: {X_train.shape}")
    self.logger.debug(f"Test set final shape: {X_test.shape}")
    if X_recal is not None:
        self.logger.debug(f"Recalibration set final shape: {X_recal.shape}")

    self._splits = DataSplits(  # Populate DataSplit object
        X_train=X_train,
        y_train=y_train_df,
        X_recal=X_recal,
        y_recal=y_recal_df,
        X_test=X_test,
        y_test=y_test_df,
        groups_train=groups_train,
    )

    # Fairness stratification columns (e.g. a "HOSPITAL" column used
    # only for a spatial subgroup comparison) are ingested independently
    # of `data.predictors` and are never merged into the feature
    # frames above; align that raw slice with each split's row index so
    # it stays available for fairness evaluation.
    self._fairness_splits = (
        FairnessSplits(
            train=self._fairness_raw.loc[X_train.index].copy(),
            test=self._fairness_raw.loc[X_test.index].copy(),
            recal=(
                self._fairness_raw.loc[X_recal.index].copy()
                if X_recal is not None
                else None
            ),
        )
        if self._fairness_raw is not None
        else None
    )

    return X_train, y_train_df, X_recal, y_recal_df, X_test, y_test_df, groups_train

ingest_data

ingest_data(**kwargs) -> DataFrame

Loads the raw dataset specified in the configuration and filters it to retain only predictors, outcomes, and configured group columns.

Fairness stratification columns (workflow.evaluation.fairness.strata) are validated against the raw dataset here too, but are kept out of the returned frame; they are cached separately (_fairness_raw) for prepare_data() to align into FairnessSplits.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Additional keyword arguments passed to the underlying Pandas reader function (e.g., sheet_name for Excel, sep for CSV).	{}

Returns:

Type	Description
DataFrame	The loaded and filtered dataset.

Raises:

Type	Description
TypeError	If the loaded data is not a pandas DataFrame.
KeyError	If any configured required or fairness stratum column is missing from the raw dataset.

Source code in medpipe/pipeline/orchestrator.py

def ingest_data(self, **kwargs) -> pd.DataFrame:
    """
    Loads the raw dataset specified in the configuration and filters it
    to retain only predictors, outcomes, and configured group columns.

    Fairness stratification columns (`workflow.evaluation.fairness.strata`)
    are validated against the raw dataset here too, but are kept out of
    the returned frame; they are cached separately (`_fairness_raw`) for
    `prepare_data()` to align into `FairnessSplits`.

    Parameters
    ----------
    **kwargs : Any
        Additional keyword arguments passed to the underlying Pandas reader function
        (e.g., `sheet_name` for Excel, `sep` for CSV).

    Returns
    -------
    pd.DataFrame
        The loaded and filtered dataset.

    Raises
    ------
    TypeError
        If the loaded data is not a pandas DataFrame.
    KeyError
        If any configured required or fairness stratum column is
        missing from the raw dataset.

    """
    dataset_path = self.config.data.path
    self.logger.info(f"Ingesting data from {dataset_path}")

    data = load_data(dataset_path, **kwargs)
    if not isinstance(data, pd.DataFrame):
        raise TypeError(
            f"Input data should be a pd.DataFrame, but got {type(data)}"
        )

    self.logger.debug(f"Loaded data shape: {data.shape}")

    # Collect required columns (predictors & outcomes)
    required_cols: list[str] = []
    if hasattr(self.config, "data"):
        required_cols.extend(getattr(self.config.data, "predictors", []))
        required_cols.extend(getattr(self.config.data, "outcomes", []))

    # Collect group columns from validation splits (test, recalibration,
    # cross-validation)
    required_cols += self._get_validation_columns()

    # Deduplicate while preserving order
    unique_cols = list(dict.fromkeys(required_cols))

    # Fairness stratification columns (e.g. a "HOSPITAL" column used
    # only for a spatial subgroup comparison) are validated up front
    # here too, so a missing column fails fast instead of only once
    # evaluation runs later — but they are never merged into the
    # modelling frame below; they are kept as a separate raw slice
    # (see `_fairness_raw`) so they never reach the model.
    fairness_cols = self._get_fairness_columns()

    # Validate column existence
    missing_cols = [
        col
        for col in list(dict.fromkeys(unique_cols + fairness_cols))
        if col not in data.columns
    ]
    if missing_cols:
        raise KeyError(
            "The following required columns were missing from the "
            f"dataset: {missing_cols}"
        )

    self._fairness_raw = data[fairness_cols].copy() if fairness_cols else None

    # Filter out unneeded columns
    filtered_data = data[unique_cols].copy()
    self.logger.info(
        f"Filtered dataset from {len(data.columns)} down to "
        f"{len(unique_cols)} required columns."
    )
    self.logger.debug(f"Filtered data shape: {filtered_data.shape}")

    return pd.DataFrame(filtered_data)

get_subgroup_specs

get_subgroup_specs() -> dict[str, Any]

Parses workflow.evaluation.fairness settings into a dictionary compatible with MedpipeClassifierEvaluator.extract_subgroups.

Returns:

Name	Type	Description
subgroup_specs	dict of str to Any	Maps stratum column names either to column strings (for discrete categorical groupings) or to lists of range bounds defined under groups.

Source code in medpipe/pipeline/orchestrator.py

def get_subgroup_specs(self) -> dict[str, Any]:
    """
    Parses `workflow.evaluation.fairness` settings into a dictionary
    compatible with `MedpipeClassifierEvaluator.extract_subgroups`.

    Returns
    -------
    subgroup_specs : dict of str to Any
        Maps stratum column names either to column strings
        (for discrete categorical groupings) or to lists of range bounds
        defined under `groups`.

    """
    eval_cfg = getattr(getattr(self.config, "workflow", None), "evaluation", None)
    fairness_cfg = getattr(eval_cfg, "fairness", None)

    if not fairness_cfg or not getattr(fairness_cfg, "strata", None):
        return {}

    strata = fairness_cfg.strata
    custom_groups = getattr(fairness_cfg, "groups", {}) or {}

    subgroup_specs = {}
    for column in strata:
        if column in custom_groups:
            # Custom ranges e.g. {"AGE": [[18, 50], [51, 120]]}
            subgroup_specs[column] = custom_groups[column]
        else:
            # Column name string for discrete categorical groupby (e.g. "SEX")
            subgroup_specs[column] = column

    return subgroup_specs

extract_stratum_subgroup

extract_stratum_subgroup(
    X: DataFrame,
    column: str,
    group: Any,
    y: DataFrame | None = None,
) -> tuple[DataFrame, DataFrame | None]

Extract a stratified subgroup from features (and optional target labels).

Resolves continuous range bounds or discrete categorical identifiers.

Parameters:

Name	Type	Description	Default
X	DataFrame	Feature matrix containing the stratification column.	required
column	str	Name of the column defining the stratum (e.g., 'AGE', 'SEX').	required
group	Any	The specific subgroup identifier or range (e.g., 'M', (18, 65), '[18, 65)').	required
y	DataFrame	Corresponding ground truth label DataFrame aligned with X. Default is None.	None

Returns:

Name	Type	Description
X_subgroup	DataFrame	Filtered feature DataFrame for the specified stratum group.
y_subgroup	DataFrame or None	Filtered label DataFrame matching X_subgroup index, or None if y was not provided.

Source code in medpipe/pipeline/orchestrator.py

def extract_stratum_subgroup(
    self,
    X: pd.DataFrame,
    column: str,
    group: Any,
    y: pd.DataFrame | None = None,
) -> tuple[pd.DataFrame, pd.DataFrame | None]:
    """
    Extract a stratified subgroup from features (and optional target labels).

    Resolves continuous range bounds or discrete categorical identifiers.

    Parameters
    ----------
    X : pd.DataFrame
        Feature matrix containing the stratification column.
    column : str
        Name of the column defining the stratum (e.g., 'AGE', 'SEX').
    group : Any
        The specific subgroup identifier or range (e.g., 'M', (18, 65), '[18, 65)').
    y : pd.DataFrame, optional
        Corresponding ground truth label DataFrame aligned with X. Default is None.

    Returns
    -------
    X_subgroup : pd.DataFrame
        Filtered feature DataFrame for the specified stratum group.
    y_subgroup : pd.DataFrame or None
        Filtered label DataFrame matching X_subgroup index, or None if y
        was not provided.

    """
    self.logger.info(
        f"Extracting subgroup for stratum '{column}' matching group: {group}"
    )

    mask = resolve_subgroup_mask(X, column=column, group=group)
    n_matched = mask.sum()

    if n_matched == 0:
        self.logger.warning(
            f"Subgroup extraction returned 0 samples for column '{column}' "
            f"and group '{group}'."
        )

    X_subgroup = X.loc[mask].copy()
    y_subgroup = y.loc[mask].copy() if y is not None else None

    self.logger.debug(
        f"{len(mask)} samples in stratum '{column}' matching group: {group}",
    )
    return X_subgroup, y_subgroup

build_preprocessor

build_preprocessor() -> Pipeline | None

Constructs an sklearn Pipeline for data transformation based on the configuration.

Parses the workflow configuration to build a sequence of ColumnTransformer objects, mapping specific operations to requested columns.

Returns:

Type	Description
Optional[Pipeline]	A configured scikit-learn Pipeline containing the cascaded transformations, or None if preprocessing is disabled or omitted.

Source code in medpipe/pipeline/orchestrator.py

def build_preprocessor(self) -> Pipeline | None:
    """
    Constructs an sklearn Pipeline for data transformation based on the
    configuration.

    Parses the workflow configuration to build a sequence of `ColumnTransformer`
    objects, mapping specific operations to requested columns.

    Returns
    -------
    Optional[Pipeline]
        A configured scikit-learn `Pipeline` containing the cascaded
        transformations, or `None` if preprocessing is disabled or omitted.

    """
    preprocessing_dict = self.config.workflow.preprocessing

    if preprocessing_dict and preprocessing_dict.preprocess:
        self.logger.info("Constructing preprocessing pipeline.")
        steps = []

        if preprocessing_dict.operations:
            ct_columns_dict = {pred: pred for pred in self.config.data.predictors}

            for i, operation in enumerate(preprocessing_dict.operations):
                op_type = self._check_operation(operation.name)
                op_extras = (
                    {} if not operation.model_extra else operation.model_extra
                )

                unknown = [
                    column
                    for column in operation.columns
                    if column not in ct_columns_dict
                ]
                if unknown:
                    raise ValueError(
                        f"Preprocessing operation '{operation.name}' refers to "
                        f"columns that are not in data.predictors: {unknown}"
                    )
                ct_columns = [
                    ct_columns_dict[column] for column in operation.columns
                ]

                self.logger.debug(
                    f"Creating operation {i + 1}: {operation} on columns "
                    f"{ct_columns}"
                )
                ct = ColumnTransformer(
                    [(f"op_{i + 1}", op_type(**op_extras), ct_columns)],
                    remainder="passthrough",
                )
                ct.set_output(transform="pandas")

                if i == len(preprocessing_dict.operations) - 1:
                    ct.set_output(transform="default")

                steps.append((f"transformer_{i + 1}", ct))

                ct_columns_dict = {
                    pred: (
                        f"remainder__{column}"
                        if column not in ct_columns
                        else f"op_{i + 1}__{column}"
                    )
                    for (pred, column) in ct_columns_dict.items()
                }

        return Pipeline(steps=steps)

    self.logger.info(
        "No preprocessing steps specified; skipping pipeline creation."
    )
    return None

runner

BaseRunner

Shared execution engine for training, hyperparameter tuning, and persisting outcome models, independent of outcome type.

Handles instantiating estimators via the ModelRegistry, running GridSearchCV or standard cross-validation, and persisting the finalized models to the run directory. Outcome-type-specific behavior (CV splitter selection, the fitted pipeline's final step name and class, the default metrics fallback, and any post-hoc adjustment of the fitted pipeline) is provided by subclasses via the _final_step_name, _pipeline_cls, and _default_metrics class attributes and the _create_cv_splitter and _postfit hook methods.

Parameters:

Name	Type	Description	Default
orchestrator	MedpipeOrchestrator	The configured orchestrator instance, providing resolved configurations, preprocessing pipelines, and the execution run directory.	required

Attributes:

Name	Type	Description
orchestrator	MedpipeOrchestrator	Orchestrator instance driving the environment state.
logger	Logger	Logger instance for the runner.
fitted_models	Dict[str, Any]	Dictionary storing the finalized models, keyed by outcome name.

Methods:

Name	Description
fit_outcome
groups_train=None)	Trains and optionally post-processes a model for a single outcome.
run	Executes the pipelines for all configured target outcomes.

Source code in medpipe/pipeline/runner.py

class BaseRunner:
    """
    Shared execution engine for training, hyperparameter tuning, and
    persisting outcome models, independent of outcome type.

    Handles instantiating estimators via the ModelRegistry, running
    GridSearchCV or standard cross-validation, and persisting the finalized
    models to the run directory. Outcome-type-specific behavior (CV splitter
    selection, the fitted pipeline's final step name and class, the default
    metrics fallback, and any post-hoc adjustment of the fitted pipeline) is
    provided by subclasses via the `_final_step_name`, `_pipeline_cls`, and
    `_default_metrics` class attributes and the `_create_cv_splitter` and
    `_postfit` hook methods.

    Parameters
    ----------
    orchestrator : MedpipeOrchestrator
        The configured orchestrator instance, providing resolved configurations,
        preprocessing pipelines, and the execution run directory.

    Attributes
    ----------
    orchestrator : MedpipeOrchestrator
        Orchestrator instance driving the environment state.
    logger : logging.Logger
        Logger instance for the runner.
    fitted_models : Dict[str, Any]
        Dictionary storing the finalized models, keyed by outcome name.

    Methods
    -------
    fit_outcome(outcome, X_train, y_train, X_recal=None, y_recal=None,
    groups_train=None)
        Trains and optionally post-processes a model for a single outcome.
    run(X_train, y_train_df, X_recal=None, y_recal_df=None, groups_train=None)
        Executes the pipelines for all configured target outcomes.

    """

    _final_step_name: ClassVar[str]
    _pipeline_cls: ClassVar[type[Pipeline]] = Pipeline
    _default_metrics: ClassVar[list[str]]

    def __init__(self, orchestrator: MedpipeOrchestrator) -> None:
        self.orchestrator = orchestrator
        self.logger = get_console_logger("medpipe.runner")
        self.fitted_models: dict[str, Any] = {}

    def _instantiate_estimator(
        self, algo_name: str, params: dict[str, Any]
    ) -> BaseEstimator:
        """
        Instantiates an estimator using the ModelRegistry.

        Parameters
        ----------
        algo_name : str
            The lookup name for the algorithm in the ModelRegistry.
        params : Dict[str, Any]
            The hyperparameters to initialize the estimator with.

        Returns
        -------
        BaseEstimator
            The un-fitted scikit-learn estimator.

        """
        estimator_class = ModelRegistry.get(algo_name)

        # Filter out hyperparameters that are lists/tuples if we are doing GridSearch
        # Initial instantiation should only use scalar/default values if available
        init_params = {
            k: (v[0] if isinstance(v, (list, tuple)) else v) for k, v in params.items()
        }

        estimator = estimator_class(**init_params)
        self.logger.info(
            f"Estimator {estimator_class.__name__} successfully instantiated"
        )
        self.logger.debug(f"Estimator instantiated with parameters {init_params}")

        return estimator

    def _create_cv_splitter(
        self,
        strategy: str,
        n_splits: int | None,
        random_state: int | None,
        shuffle: bool = True,
    ) -> BaseCrossValidator:
        """
        Instantiates the appropriate cross-validation splitter.

        Subclasses must implement this to return a splitter suited to their
        outcome type (e.g. stratified splitters for classification, plain
        K-fold splitters for regression).

        Parameters
        ----------
        strategy : {'random', 'group', 'loso'}
            The CV strategy to employ. The 'loso' strategy holds out one
            group per fold (leave-one-group-out).
        n_splits : int | None
            The number of cross-validation folds. Ignored by the 'loso'
            strategy, which has one fold per group.
        random_state : int | None
            The random seed for reproducibility. Only used when `shuffle` is
            True, since scikit-learn rejects a seed for unshuffled splitters.
        shuffle : bool, default=True
            Whether to shuffle the samples before splitting into folds.
            Ignored by the 'loso' strategy.

        Returns
        -------
        BaseCrossValidator
            The instantiated cross-validation splitter object.

        Raises
        ------
        ValueError
            If an unsupported strategy is provided.

        """
        raise NotImplementedError

    def _save_model(self, model: Any, outcome: str) -> None:
        """
        Saves the fitted model to the orchestrator's run directory.

        Parameters
        ----------
        model : Any
            The finalized, fully fitted model.
        outcome : str
            The name of the outcome, used to name the saved file.

        """
        save_dir = self.orchestrator.run_dir / "models"
        save_dir.mkdir(exist_ok=True, parents=True)

        file_path = save_dir / f"{outcome}_model.joblib"
        joblib.dump(model, file_path, compress=3)
        self.logger.info(f"[{outcome}] Model saved to {file_path}")

    def _save_final_models(self) -> None:
        """
        Saves the complete dictionary of fitted outcome models into a single
        file in the models directory for downstream evaluation.

        """
        save_dir = self.orchestrator.run_dir / "models"
        save_dir.mkdir(exist_ok=True, parents=True)

        project_name = self.orchestrator.config.meta.project_name
        models_path = save_dir / f"{project_name}_fitted.joblib"
        joblib.dump(self.fitted_models, models_path, compress=3)
        self.logger.info(f"Fitted final models saved to {models_path}")

    def _save_cv_results(
        self,
        outcome: str,
        cv_results_df: pd.DataFrame,
        best_index: int | None = None,
    ) -> None:
        """
        Save cross-validation results to disk in the artifacts directory.

        Exports a detailed fold-level CSV file for spreadsheet auditing alongside
        a JSON file containing summary statistics (mean and standard deviation)
        for each test evaluation metric. For the results of a grid search, the
        summary is the one of the selected hyperparameters.

        Parameters
        ----------
        outcome : str
            The name of the outcome being evaluated.
        cv_results_df : pd.DataFrame
            DataFrame containing fold-by-fold cross-validation metrics and timing
            information derived from scikit-learn's `cross_validate` or `GridSearchCV`.
        best_index : int, optional
            Row of `cv_results_df` of the selected hyperparameters when the results
            come from `GridSearchCV`. Leave as None for `cross_validate` results.

        Returns
        -------
        None

        """
        artifacts_dir = self.orchestrator.run_dir / "CV"
        artifacts_dir.mkdir(exist_ok=True, parents=True)

        # Save detailed fold-level CSV
        csv_path = artifacts_dir / f"{outcome}_cv_results.csv"
        cv_results_df.to_csv(csv_path, index=False)
        self.logger.info(f"[{outcome}] Detailed CV results saved to {csv_path}")

        # Extract and save summary report (mean ± std dev for test scores)
        summary = {}
        if best_index is not None:
            # GridSearchCV results hold one row per candidate, with the mean and
            # standard deviation over the folds already computed
            best = cv_results_df.iloc[int(best_index)]
            for col in cv_results_df.columns:
                if col.startswith("mean_test_"):
                    metric_name = col.removeprefix("mean_test_")
                    summary[metric_name] = {
                        "mean": float(best[col]),
                        "std": float(best[f"std_test_{metric_name}"]),
                    }
        else:
            for col in cv_results_df.columns:
                if col.startswith("test_"):
                    metric_name = col.replace("test_", "")
                    summary[metric_name] = {
                        "mean": float(cv_results_df[col].mean()),
                        "std": float(cv_results_df[col].std()),
                    }

        json_path = self.orchestrator.artifact_manager.save_json(
            summary, artifacts_dir, f"{outcome}_cv_summary.json"
        )
        self.logger.info(f"[{outcome}] CV summary saved to {json_path}")

    def _train_model_cv(
        self,
        outcome: str,
        pipeline: Pipeline,
        hyperparams: dict,
        X_train: pd.DataFrame,
        y_train: np.ndarray,
        cv_splitter: BaseCrossValidator,
        groups_train: np.ndarray | None,
    ) -> Pipeline:
        """
        Handles hyperparameter tuning via GridSearchCV or standard cross-validation.

        Parameters
        ----------
        outcome : str
            The name of the outcome being modeled.
        pipeline : sklearn.pipeline.Pipeline
            The base pipeline containing the preprocessor and estimator.
        hyperparams : dict
            A dictionary of hyperparameters to apply or search over.
        X_train : pd.DataFrame
            The training feature set.
        y_train : np.ndarray
            The 1D training target labels.
        cv_splitter : BaseCrossValidator
            The cross-validation splitting strategy object.
        groups_train : Optional[np.ndarray]
            Group labels for the training set, used for group-based
            cross-validation splits.

        Returns
        -------
        sklearn.pipeline.Pipeline
            The best fitted pipeline after cross-validation or grid search.

        Raises
        ------
        ValueError
            If `groups_train` is None when the strategy is 'group' or 'loso'.

        """
        # Retrieve the list of metrics from the configuration safely
        try:
            configured_metrics = (
                self.orchestrator.config.workflow.evaluation.metrics.metrics
            )
        except AttributeError:
            configured_metrics = self._default_metrics

        n_jobs = self.orchestrator.config.workflow.n_jobs
        cv_cfg = self.orchestrator.config.workflow.validation.cross_validation
        assert cv_cfg

        if cv_cfg.strategy in ("group", "loso") and groups_train is None:
            raise ValueError(
                f"The 'groups' parameter should not be None for the "
                f"'{cv_cfg.strategy}' strategy"
            )

        scorers_dict = build_scorers(configured_metrics)

        self.logger.debug(
            f"[{outcome}] Cross-validation input dimensions - "
            f"X_train: {X_train.shape}, y_train: {y_train.shape}, "
            f"Positives: {int(np.sum(y_train))}"
        )
        if groups_train is not None:
            unique_groups = len(np.unique(groups_train))
            self.logger.debug(
                f"[{outcome}] Group CV enabled with {unique_groups} unique "
                "group identifiers."
            )

        self.logger.debug(
            f"[{outcome}] Active metrics scorers: {list(scorers_dict.keys())}"
        )

        if cv_cfg.grid_search:
            self.logger.info(f"[{outcome}] Running GridSearchCV tuning.")

            pipeline_params = {
                f"{self._final_step_name}__{k}": (
                    v if isinstance(v, (list, tuple)) else [v]
                )
                for k, v in hyperparams.items()
            }

            self.logger.debug(
                f"[{outcome}] GridSearchCV parameter grid: {pipeline_params}"
            )
            self.logger.debug(
                f"[{outcome}] GridSearchCV parallel jobs number: {n_jobs}"
            )
            search = GridSearchCV(
                estimator=pipeline,
                param_grid=pipeline_params,
                cv=cv_splitter,
                scoring=scorers_dict,
                refit=configured_metrics[0],  # type: ignore
                n_jobs=n_jobs,
            )
            search.fit(X_train, y_train, groups=groups_train)

            # Convert search results to DataFrame & save artifacts
            cv_results_df = pd.DataFrame(search.cv_results_)
            self._save_cv_results(outcome, cv_results_df, best_index=search.best_index_)

            self.logger.debug(
                f"[{outcome}] GridSearchCV best index: {search.best_index_} | "
                f"Best score ({configured_metrics[0]}): {search.best_score_:.4f} | "
                f"Best parameters: {search.best_params_}"
            )
            return search.best_estimator_

        else:
            self.logger.info(f"[{outcome}] Running standard cross-validation.")

            self.logger.debug(
                f"[{outcome}] Cross-validation parallel jobs number: {n_jobs}"
            )
            results = cross_validate(
                estimator=pipeline,
                X=X_train,
                y=y_train,
                groups=groups_train,
                cv=cv_splitter,
                scoring=scorers_dict,
                n_jobs=n_jobs,
            )

            # Convert cross_validate dict to DataFrame & save artifacts
            cv_results_df = pd.DataFrame(results)
            self._save_cv_results(outcome, cv_results_df)

            # Debug log per-fold metric breakdown
            for col in cv_results_df.columns:
                if col.startswith("test_"):
                    scores = cv_results_df[col].to_numpy()
                    self.logger.debug(
                        f"[{outcome}] Fold scores for '{col}': "
                        f"{np.round(scores, 4).tolist()}"
                    )

            self.logger.info(
                f"[{outcome}] Fitting final base pipeline on full training data."
            )
            return pipeline.fit(X_train, y_train)

    def _postfit(
        self,
        outcome: str,
        best_pipeline: Pipeline,
        model_config: dict,
        X_recal: pd.DataFrame | None,
        y_recal: np.ndarray | None,
    ) -> Any:
        """
        Hook for outcome-type-specific post-processing of the fitted pipeline
        (e.g. classifier recalibration). Default is a no-op passthrough.

        Parameters
        ----------
        outcome : str
            The name of the outcome being modeled.
        best_pipeline : sklearn.pipeline.Pipeline
            The fitted base pipeline.
        model_config : dict
            The configuration dictionary for the specific outcome.
        X_recal : pd.DataFrame or None
            The recalibration feature set, if available.
        y_recal : np.ndarray or None
            The 1D recalibration target labels, if available.

        Returns
        -------
        Any
            The final model to persist and evaluate.

        """
        return best_pipeline

    def fit_outcome(
        self,
        outcome: str,
        X_train: pd.DataFrame,
        y_train: np.ndarray,
        X_recal: pd.DataFrame | None = None,
        y_recal: np.ndarray | None = None,
        groups_train: np.ndarray | None = None,
    ) -> Any:
        """
        Trains and optionally post-processes a model for a single outcome.

        Parameters
        ----------
        outcome : str
            The name of the outcome being modeled.
        X_train : pd.DataFrame
            The training feature set.
        y_train : np.ndarray
            The 1D training target labels.
        X_recal : Optional[pd.DataFrame], default=None
            The recalibration feature set.
        y_recal : Optional[np.ndarray], default=None
            The 1D recalibration target labels.
        groups_train : Optional[np.ndarray], default=None
            Group labels for the training set, required if strategy is 'group'.

        Returns
        -------
        Any
            The fully trained model ready for evaluation.

        Raises
        ------
        ValueError
            If no algorithm is specified for an outcome

        """
        self.logger.info(f"--- Starting execution for outcome: {outcome} ---")

        model_config = self.orchestrator.config.resolved_models[outcome]
        algo_name = model_config.algorithm

        if not algo_name:
            raise ValueError(f"No algorithm specified for outcome: {outcome}")

        hyperparams = model_config.hyperparameters

        # 1. Build Base Pipeline
        preprocessor = self.orchestrator.build_preprocessor()
        estimator = self._instantiate_estimator(algo_name, hyperparams)

        steps = []
        if preprocessor is not None:
            steps.append(("preprocessor", preprocessor))
        steps.append((self._final_step_name, estimator))
        pipeline = self._pipeline_cls(steps)

        # 2. Configure Cross-Validation
        cv_config = self.orchestrator.config.workflow.validation.cross_validation

        # 3. Training Loop
        if self.orchestrator.config.meta.run_mode in ["audit", "cv"]:
            assert cv_config
            assert cv_config.strategy == "loso" or cv_config.n_splits

            cv_splitter = self._create_cv_splitter(
                strategy=cv_config.strategy,
                n_splits=cv_config.n_splits,
                random_state=self.orchestrator.config.workflow.random_state,
                shuffle=True if cv_config.shuffle is None else cv_config.shuffle,
            )
            self.logger.debug(
                f"[{outcome}] CV {type(cv_splitter).__name__} "
                f"with arguments {cv_splitter.__dict__}"
            )
            best_pipeline = self._train_model_cv(
                outcome=outcome,
                pipeline=pipeline,
                hyperparams=hyperparams,
                X_train=X_train,
                y_train=y_train,
                groups_train=groups_train,
                cv_splitter=cv_splitter,
            )
        else:
            self.logger.debug(f"[{outcome}] Fitting without any CV")
            best_pipeline = pipeline.fit(X_train, y_train)

        # 4. Optional Post-Hoc Adjustment
        final_model = self._postfit(
            outcome=outcome,
            best_pipeline=best_pipeline,
            model_config=model_config.model_dump(),
            X_recal=X_recal,
            y_recal=y_recal,
        )

        # 5. Save and Return
        self._save_model(final_model, outcome)
        self.logger.info(f"--- Finished execution for outcome: {outcome} ---")
        return final_model

    def run(
        self,
        X_train: pd.DataFrame,
        y_train_df: pd.DataFrame,
        X_recal: pd.DataFrame | None = None,
        y_recal_df: pd.DataFrame | None = None,
        groups_train: np.ndarray | None = None,
    ) -> dict[str, Any]:
        """
        Executes the pipelines for all configured target outcomes.

        Parameters
        ----------
        X_train : pd.DataFrame
            The training feature set.
        y_train_df : pd.DataFrame
            The training targets DataFrame (can contain multiple outcome columns).
        X_recal : Optional[pd.DataFrame], default=None
            The recalibration feature set.
        y_recal_df : Optional[pd.DataFrame], default=None
            The recalibration targets DataFrame.
        groups_train : Optional[np.ndarray], default=None
            Group labels for the training set.

        Returns
        -------
        Dict[str, Any]
            A dictionary mapping outcome names to their finalized, fitted models.

        """
        outcomes = self.orchestrator.config.data.outcomes
        self.logger.debug(f"Starting fitting of {len(outcomes)} models.")

        for outcome in outcomes:
            # We assume y_train_df columns correspond to the requested outcomes
            fit_start_time = time.perf_counter()

            y_train = y_train_df[outcome].to_numpy().ravel()

            y_recal = None
            if y_recal_df is not None and outcome in y_recal_df.columns:
                y_recal = y_recal_df[outcome].to_numpy().ravel()

            final_model = self.fit_outcome(
                outcome=outcome,
                X_train=X_train,
                y_train=y_train,
                X_recal=X_recal,
                y_recal=y_recal,
                groups_train=groups_train,
            )

            self.fitted_models[outcome] = final_model

            fit_duration = time.perf_counter() - fit_start_time
            self.logger.debug(
                f"[{outcome}] Fitting completed in {fit_duration:.2f} seconds."
            )

        # Save final models dictionary
        self._save_final_models()

        return self.fitted_models

fit_outcome

fit_outcome(
    outcome: str,
    X_train: DataFrame,
    y_train: ndarray,
    X_recal: DataFrame | None = None,
    y_recal: ndarray | None = None,
    groups_train: ndarray | None = None,
) -> Any

Trains and optionally post-processes a model for a single outcome.

Parameters:

Name	Type	Description	Default
outcome	str	The name of the outcome being modeled.	required
X_train	DataFrame	The training feature set.	required
y_train	ndarray	The 1D training target labels.	required
X_recal	Optional[DataFrame]	The recalibration feature set.	None
y_recal	Optional[ndarray]	The 1D recalibration target labels.	None
groups_train	Optional[ndarray]	Group labels for the training set, required if strategy is 'group'.	None

Returns:

Type	Description
Any	The fully trained model ready for evaluation.

Raises:

Type	Description
ValueError	If no algorithm is specified for an outcome

Source code in medpipe/pipeline/runner.py

def fit_outcome(
    self,
    outcome: str,
    X_train: pd.DataFrame,
    y_train: np.ndarray,
    X_recal: pd.DataFrame | None = None,
    y_recal: np.ndarray | None = None,
    groups_train: np.ndarray | None = None,
) -> Any:
    """
    Trains and optionally post-processes a model for a single outcome.

    Parameters
    ----------
    outcome : str
        The name of the outcome being modeled.
    X_train : pd.DataFrame
        The training feature set.
    y_train : np.ndarray
        The 1D training target labels.
    X_recal : Optional[pd.DataFrame], default=None
        The recalibration feature set.
    y_recal : Optional[np.ndarray], default=None
        The 1D recalibration target labels.
    groups_train : Optional[np.ndarray], default=None
        Group labels for the training set, required if strategy is 'group'.

    Returns
    -------
    Any
        The fully trained model ready for evaluation.

    Raises
    ------
    ValueError
        If no algorithm is specified for an outcome

    """
    self.logger.info(f"--- Starting execution for outcome: {outcome} ---")

    model_config = self.orchestrator.config.resolved_models[outcome]
    algo_name = model_config.algorithm

    if not algo_name:
        raise ValueError(f"No algorithm specified for outcome: {outcome}")

    hyperparams = model_config.hyperparameters

    # 1. Build Base Pipeline
    preprocessor = self.orchestrator.build_preprocessor()
    estimator = self._instantiate_estimator(algo_name, hyperparams)

    steps = []
    if preprocessor is not None:
        steps.append(("preprocessor", preprocessor))
    steps.append((self._final_step_name, estimator))
    pipeline = self._pipeline_cls(steps)

    # 2. Configure Cross-Validation
    cv_config = self.orchestrator.config.workflow.validation.cross_validation

    # 3. Training Loop
    if self.orchestrator.config.meta.run_mode in ["audit", "cv"]:
        assert cv_config
        assert cv_config.strategy == "loso" or cv_config.n_splits

        cv_splitter = self._create_cv_splitter(
            strategy=cv_config.strategy,
            n_splits=cv_config.n_splits,
            random_state=self.orchestrator.config.workflow.random_state,
            shuffle=True if cv_config.shuffle is None else cv_config.shuffle,
        )
        self.logger.debug(
            f"[{outcome}] CV {type(cv_splitter).__name__} "
            f"with arguments {cv_splitter.__dict__}"
        )
        best_pipeline = self._train_model_cv(
            outcome=outcome,
            pipeline=pipeline,
            hyperparams=hyperparams,
            X_train=X_train,
            y_train=y_train,
            groups_train=groups_train,
            cv_splitter=cv_splitter,
        )
    else:
        self.logger.debug(f"[{outcome}] Fitting without any CV")
        best_pipeline = pipeline.fit(X_train, y_train)

    # 4. Optional Post-Hoc Adjustment
    final_model = self._postfit(
        outcome=outcome,
        best_pipeline=best_pipeline,
        model_config=model_config.model_dump(),
        X_recal=X_recal,
        y_recal=y_recal,
    )

    # 5. Save and Return
    self._save_model(final_model, outcome)
    self.logger.info(f"--- Finished execution for outcome: {outcome} ---")
    return final_model

run

run(
    X_train: DataFrame,
    y_train_df: DataFrame,
    X_recal: DataFrame | None = None,
    y_recal_df: DataFrame | None = None,
    groups_train: ndarray | None = None,
) -> dict[str, Any]

Executes the pipelines for all configured target outcomes.

Parameters:

Name	Type	Description	Default
X_train	DataFrame	The training feature set.	required
y_train_df	DataFrame	The training targets DataFrame (can contain multiple outcome columns).	required
X_recal	Optional[DataFrame]	The recalibration feature set.	None
y_recal_df	Optional[DataFrame]	The recalibration targets DataFrame.	None
groups_train	Optional[ndarray]	Group labels for the training set.	None

Returns:

Type	Description
Dict[str, Any]	A dictionary mapping outcome names to their finalized, fitted models.

Source code in medpipe/pipeline/runner.py

def run(
    self,
    X_train: pd.DataFrame,
    y_train_df: pd.DataFrame,
    X_recal: pd.DataFrame | None = None,
    y_recal_df: pd.DataFrame | None = None,
    groups_train: np.ndarray | None = None,
) -> dict[str, Any]:
    """
    Executes the pipelines for all configured target outcomes.

    Parameters
    ----------
    X_train : pd.DataFrame
        The training feature set.
    y_train_df : pd.DataFrame
        The training targets DataFrame (can contain multiple outcome columns).
    X_recal : Optional[pd.DataFrame], default=None
        The recalibration feature set.
    y_recal_df : Optional[pd.DataFrame], default=None
        The recalibration targets DataFrame.
    groups_train : Optional[np.ndarray], default=None
        Group labels for the training set.

    Returns
    -------
    Dict[str, Any]
        A dictionary mapping outcome names to their finalized, fitted models.

    """
    outcomes = self.orchestrator.config.data.outcomes
    self.logger.debug(f"Starting fitting of {len(outcomes)} models.")

    for outcome in outcomes:
        # We assume y_train_df columns correspond to the requested outcomes
        fit_start_time = time.perf_counter()

        y_train = y_train_df[outcome].to_numpy().ravel()

        y_recal = None
        if y_recal_df is not None and outcome in y_recal_df.columns:
            y_recal = y_recal_df[outcome].to_numpy().ravel()

        final_model = self.fit_outcome(
            outcome=outcome,
            X_train=X_train,
            y_train=y_train,
            X_recal=X_recal,
            y_recal=y_recal,
            groups_train=groups_train,
        )

        self.fitted_models[outcome] = final_model

        fit_duration = time.perf_counter() - fit_start_time
        self.logger.debug(
            f"[{outcome}] Fitting completed in {fit_duration:.2f} seconds."
        )

    # Save final models dictionary
    self._save_final_models()

    return self.fitted_models

MedpipeClassifierRunner

Bases: BaseRunner

Executes the training, hyperparameter tuning, and optional recalibration loops for binary classification outcomes.

The runner acts purely as an execution engine. It requests instantiated preprocessing pipelines from the orchestrator, initializes models via the ModelRegistry, handles GridSearchCV or standard cross-validation, applies post-hoc calibration, and persists the finalized models to the run directory.

Parameters:

Name	Type	Description	Default
orchestrator	MedpipeOrchestrator	The configured orchestrator instance, providing resolved configurations, preprocessing pipelines, and the execution run directory.	required

Attributes:

Name	Type	Description
orchestrator	MedpipeOrchestrator	Orchestrator instance driving the environment state.
logger	Logger	Logger instance for the runner.
fitted_models	Dict[str, Union[Pipeline, CalibratedClassifierCV]]	Dictionary storing the finalized models, keyed by outcome name.

Methods:

Name	Description
fit_outcome
groups_train=None)	Trains, evaluates, and optionally calibrates a model for a single outcome.
run	Executes the pipelines for all configured target outcomes.

Source code in medpipe/pipeline/runner.py

class MedpipeClassifierRunner(BaseRunner):
    """
    Executes the training, hyperparameter tuning, and optional recalibration loops
    for binary classification outcomes.

    The runner acts purely as an execution engine. It requests instantiated
    preprocessing pipelines from the orchestrator, initializes models via the
    ModelRegistry, handles GridSearchCV or standard cross-validation, applies
    post-hoc calibration, and persists the finalized models to the run directory.

    Parameters
    ----------
    orchestrator : MedpipeOrchestrator
        The configured orchestrator instance, providing resolved configurations,
        preprocessing pipelines, and the execution run directory.

    Attributes
    ----------
    orchestrator : MedpipeOrchestrator
        Orchestrator instance driving the environment state.
    logger : logging.Logger
        Logger instance for the runner.
    fitted_models : Dict[str, Union[Pipeline, CalibratedClassifierCV]]
        Dictionary storing the finalized models, keyed by outcome name.

    Methods
    -------
    fit_outcome(outcome, X_train, y_train, X_recal=None, y_recal=None,
    groups_train=None)
        Trains, evaluates, and optionally calibrates a model for a single outcome.
    run(X_train, y_train_df, X_recal=None, y_recal_df=None, groups_train=None)
        Executes the pipelines for all configured target outcomes.

    """

    _final_step_name: ClassVar[str] = "classifier"
    _default_metrics: ClassVar[list[str]] = ["roc_auc"]

    def _create_cv_splitter(
        self,
        strategy: str,
        n_splits: int | None,
        random_state: int | None,
        shuffle: bool = True,
    ) -> StratifiedKFold | StratifiedGroupKFold | LeaveOneGroupOut:
        """
        Instantiates the appropriate cross-validation splitter.

        Parameters
        ----------
        strategy : {'random', 'group', 'loso'}
            The CV strategy to employ. The 'loso' strategy holds out one
            group per fold (leave-one-group-out).
        n_splits : int | None
            The number of cross-validation folds. Ignored by the 'loso'
            strategy, which has one fold per group.
        random_state : int | None
            The random seed for reproducibility. Only used when `shuffle` is
            True, since scikit-learn rejects a seed for unshuffled splitters.
        shuffle : bool, default=True
            Whether to shuffle the samples before splitting into folds.
            Ignored by the 'loso' strategy.

        Returns
        -------
        Union[StratifiedKFold, StratifiedGroupKFold, LeaveOneGroupOut]
            The instantiated cross-validation splitter object.

        Raises
        ------
        ValueError
            If an unsupported strategy is provided.

        """
        seed = random_state if shuffle else None
        if strategy == "random":
            return StratifiedKFold(
                n_splits=n_splits, shuffle=shuffle, random_state=seed
            )
        elif strategy == "group":
            return StratifiedGroupKFold(
                n_splits=n_splits, shuffle=shuffle, random_state=seed
            )
        elif strategy == "loso":
            return LeaveOneGroupOut()
        else:
            raise ValueError(
                f"Strategy must be 'random', 'group' or 'loso', got {strategy}"
            )

    def _calibrate_model(
        self,
        outcome: str,
        best_pipeline: Pipeline,
        model_config: dict,
        X_recal: pd.DataFrame | None,
        y_recal: np.ndarray | None,
    ) -> Pipeline | CalibratedClassifierCV:
        """
        Wraps the model in a FrozenEstimator and fits a calibrator if holdout
        data exists.

        Parameters
        ----------
        outcome : str
            The name of the outcome being modeled.
        best_pipeline : sklearn.pipeline.Pipeline
            The fitted base pipeline to be calibrated.
        model_config : dict
            The configuration dictionary for the specific outcome, containing
            recalibration settings.
        X_recal : pd.DataFrame or None
            The recalibration feature set. Can be None if recalibration
            is not performed.
        y_recal : np.ndarray or None
            The 1D recalibration target labels. Can be None if recalibration
            is not performed.

        Returns
        -------
        Union[sklearn.pipeline.Pipeline, sklearn.calibration.CalibratedClassifierCV]
            The calibrated model if recalibration was performed, otherwise
            the original fitted pipeline.

        """
        recal_config = model_config.get("recalibration")
        recal_flag = False

        if recal_config is not None:
            recal_flag = recal_config["recalibrate"]

        if X_recal is not None and not X_recal.empty and recal_flag:
            assert recal_config is not None  # Avoid diagnostic error
            assert y_recal is not None  # Avoid diagnostic error
            self.logger.info(f"[{outcome}] Fitting recalibrator on holdout dataset.")
            self.logger.debug(
                f"[{outcome}] Recalibration dataset shape: {X_recal.shape} | "
                f"Positives: {int(np.sum(y_recal))} ({np.mean(y_recal):.2%})"
            )
            calibrator_method = recal_config.get("method")

            self.logger.debug(f"[{outcome}] Recalibration methhod {calibrator_method}")
            calibrator = CalibratedClassifierCV(
                estimator=FrozenEstimator(best_pipeline),
                cv=2,
                method=calibrator_method,
            )
            return calibrator.fit(X_recal, y_recal)

        else:
            self.logger.info(
                f"[{outcome}] Skipping recalibration (missing dataset or "
                "configuration)."
            )
            self.logger.debug(
                f"[{outcome}] Skipping recalibration "
                f"(X_recal present: {X_recal is not None}, "
                f"Config present: {recal_config is not None}, "
                f"Recalibration flag: {recal_flag})."
            )
            return best_pipeline

    def _postfit(
        self,
        outcome: str,
        best_pipeline: Pipeline,
        model_config: dict,
        X_recal: pd.DataFrame | None,
        y_recal: np.ndarray | None,
    ) -> Pipeline | CalibratedClassifierCV:
        """
        Applies optional post-hoc probability recalibration.

        Parameters
        ----------
        outcome : str
            The name of the outcome being modeled.
        best_pipeline : sklearn.pipeline.Pipeline
            The fitted base pipeline.
        model_config : dict
            The configuration dictionary for the specific outcome, containing
            recalibration settings.
        X_recal : pd.DataFrame or None
            The recalibration feature set, if available.
        y_recal : np.ndarray or None
            The 1D recalibration target labels, if available.

        Returns
        -------
        Union[sklearn.pipeline.Pipeline, sklearn.calibration.CalibratedClassifierCV]
            The calibrated model if recalibration was performed, otherwise
            the original fitted pipeline.

        """
        return self._calibrate_model(
            outcome=outcome,
            best_pipeline=best_pipeline,
            model_config=model_config,
            X_recal=X_recal,
            y_recal=y_recal,
        )

MedpipeRegressorRunner

Bases: BaseRunner

Executes the training and hyperparameter tuning loops for regression outcomes.

Fitted pipelines are built as DistributionalPipeline instances so that distributional estimators (e.g. NGBoost, OrdBoost) expose a unified predict_dist method. Post-hoc recalibration is not supported on the regression track, so fit_outcome returns the best fitted pipeline unchanged.

Parameters:

Name	Type	Description	Default
orchestrator	MedpipeOrchestrator	The configured orchestrator instance, providing resolved configurations, preprocessing pipelines, and the execution run directory.	required

Attributes:

Name	Type	Description
orchestrator	MedpipeOrchestrator	Orchestrator instance driving the environment state.
logger	Logger	Logger instance for the runner.
fitted_models	Dict[str, DistributionalPipeline]	Dictionary storing the finalized models, keyed by outcome name.

Methods:

Name	Description
fit_outcome
groups_train=None)	Trains a model for a single outcome.
run	Executes the pipelines for all configured target outcomes.

Source code in medpipe/pipeline/runner.py

class MedpipeRegressorRunner(BaseRunner):
    """
    Executes the training and hyperparameter tuning loops for regression
    outcomes.

    Fitted pipelines are built as `DistributionalPipeline` instances so that
    distributional estimators (e.g. NGBoost, OrdBoost) expose a unified
    `predict_dist` method. Post-hoc recalibration is not supported on the
    regression track, so `fit_outcome` returns the best fitted pipeline
    unchanged.

    Parameters
    ----------
    orchestrator : MedpipeOrchestrator
        The configured orchestrator instance, providing resolved configurations,
        preprocessing pipelines, and the execution run directory.

    Attributes
    ----------
    orchestrator : MedpipeOrchestrator
        Orchestrator instance driving the environment state.
    logger : logging.Logger
        Logger instance for the runner.
    fitted_models : Dict[str, DistributionalPipeline]
        Dictionary storing the finalized models, keyed by outcome name.

    Methods
    -------
    fit_outcome(outcome, X_train, y_train, X_recal=None, y_recal=None,
    groups_train=None)
        Trains a model for a single outcome.
    run(X_train, y_train_df, X_recal=None, y_recal_df=None, groups_train=None)
        Executes the pipelines for all configured target outcomes.

    """

    _final_step_name: ClassVar[str] = "regressor"
    _pipeline_cls: ClassVar[type[Pipeline]] = DistributionalPipeline
    _default_metrics: ClassVar[list[str]] = ["rmse"]

    def _create_cv_splitter(
        self,
        strategy: str,
        n_splits: int | None,
        random_state: int | None,
        shuffle: bool = True,
    ) -> KFold | GroupKFold | LeaveOneGroupOut:
        """
        Instantiates the appropriate cross-validation splitter.

        Parameters
        ----------
        strategy : {'random', 'group', 'loso'}
            The CV strategy to employ. The 'loso' strategy holds out one
            group per fold (leave-one-group-out).
        n_splits : int | None
            The number of cross-validation folds. Ignored by the 'loso'
            strategy, which has one fold per group.
        random_state : int | None
            The random seed for reproducibility. Only used when `shuffle` is
            True, since scikit-learn rejects a seed for unshuffled splitters.
        shuffle : bool, default=True
            Whether to shuffle the samples before splitting into folds.
            Ignored by the 'loso' strategy.

        Returns
        -------
        Union[KFold, GroupKFold, LeaveOneGroupOut]
            The instantiated cross-validation splitter object.

        Raises
        ------
        ValueError
            If an unsupported strategy is provided.

        """
        seed = random_state if shuffle else None
        if strategy == "random":
            return KFold(n_splits=n_splits, shuffle=shuffle, random_state=seed)
        elif strategy == "group":
            return GroupKFold(n_splits=n_splits, shuffle=shuffle, random_state=seed)
        elif strategy == "loso":
            return LeaveOneGroupOut()
        else:
            raise ValueError(
                f"Strategy must be 'random', 'group' or 'loso', got {strategy}"
            )

evaluator

Evaluator module for MedpipeClassifier machine learning models and pipelines.

Provides TRIPOD+AI compliant model evaluation, subgroup performance analysis, bootstrap confidence interval estimation, logging, and artifact management.

BaseEvaluator

Shared evaluation engine for computing point estimates and bootstrap confidence intervals across full datasets and subgroups, independent of outcome type.

Provides model resolution, subgroup extraction, TRIPOD+AI compliant slice evaluation, and artifact persistence. Outcome-type-specific prediction retrieval (e.g. probability/decision scores for classification, point or distributional predictions for regression) is provided by subclasses via the _get_predictions hook.

Parameters:

Name	Type	Description	Default
orchestrator	MedpipeOrchestrator	The pipeline orchestrator instance containing workflow configuration, run directories, and the ArtifactManager.	required
runner	BaseRunner	The pipeline runner instance containing the dictionary of fitted models (fitted_models).	required

Attributes:

Name	Type	Description
orchestrator	MedpipeOrchestrator	Pipeline orchestrator instance.
runner	BaseRunner	Pipeline runner instance.
fitted_models	dict of str to object	Dictionary mapping outcome names to fitted estimators or pipelines.
metrics	list of str	List of metric names used during evaluation.
n_bootstraps	int	Number of bootstrap iterations.
ci_level	float	Target confidence interval level.
random_state	int, np.random.Generator, or None	Random state instance for resampling.
n_jobs	int or None	Number of parallel jobs used to compute each slice's bootstrap resamples (see bootstrap_confidence_intervals).
logger	Logger	Logger instance configured under "medpipe.evaluator".

Methods:

Name	Description
predict	Predict outcomes for samples in X.
extract_subgroups	Extract index subsets for specified data subgroups.
evaluate	Evaluate model performance with confidence intervals across full data and subgroups.

Source code in medpipe/pipeline/evaluator.py

class BaseEvaluator:
    """
    Shared evaluation engine for computing point estimates and bootstrap
    confidence intervals across full datasets and subgroups, independent of
    outcome type.

    Provides model resolution, subgroup extraction, TRIPOD+AI compliant
    slice evaluation, and artifact persistence. Outcome-type-specific
    prediction retrieval (e.g. probability/decision scores for
    classification, point or distributional predictions for regression) is
    provided by subclasses via the `_get_predictions` hook.

    Parameters
    ----------
    orchestrator : MedpipeOrchestrator
        The pipeline orchestrator instance containing workflow configuration,
        run directories, and the `ArtifactManager`.
    runner : BaseRunner
        The pipeline runner instance containing the dictionary of fitted models
        (`fitted_models`).

    Attributes
    ----------
    orchestrator : MedpipeOrchestrator
        Pipeline orchestrator instance.
    runner : BaseRunner
        Pipeline runner instance.
    fitted_models : dict of str to object
        Dictionary mapping outcome names to fitted estimators or pipelines.
    metrics : list of str
        List of metric names used during evaluation.
    n_bootstraps : int
        Number of bootstrap iterations.
    ci_level : float
        Target confidence interval level.
    random_state : int, np.random.Generator, or None
        Random state instance for resampling.
    n_jobs : int or None
        Number of parallel jobs used to compute each slice's bootstrap
        resamples (see `bootstrap_confidence_intervals`).
    logger : logging.Logger
        Logger instance configured under `"medpipe.evaluator"`.

    Methods
    -------
    predict(X, model=None, outcome=None)
        Predict outcomes for samples in X.
    extract_subgroups(X, subgroup_specs)
        Extract index subsets for specified data subgroups.
    evaluate(X, y, outcome=None, metrics=None, subgroup_specs=None, save_artifacts=True)
        Evaluate model performance with confidence intervals across full data
        and subgroups.

    """

    def __init__(
        self,
        orchestrator: MedpipeOrchestrator,
        runner: BaseRunner,
    ) -> None:
        self.orchestrator = orchestrator
        self.runner = runner
        eval_config = self.orchestrator.config.workflow.evaluation

        self.fitted_models: dict[str, Any] = getattr(runner, "fitted_models", {})
        self.n_bootstraps = eval_config.metrics.n_bootstraps
        self.ci_level = eval_config.metrics.ci_level
        self.random_state = self.orchestrator.config.workflow.random_state
        self.n_jobs = self.orchestrator.config.workflow.n_jobs

        # `ici` is registered globally with a fixed `cv_splines` default;
        # rebind it to this run's configured value (regression configs have
        # no such metric/field, hence the fallback).
        cv_splines = getattr(eval_config.metrics, "cv_splines", 3)
        MetricRegistry.register_spec(
            replace(
                MetricRegistry.get("ici"),
                func=partial(ici_score, cv_splines=cv_splines),
            )
        )
        self.logger = get_console_logger("medpipe.evaluator")

        self.metrics = eval_config.metrics.metrics

    def _get_model(self, model: Any | None = None, outcome: str | None = None) -> Any:
        """
        Resolve estimator model from argument, outcome key, or fitted dictionary.

        Parameters
        ----------
        model : object, optional
            Explicit fitted model instance.
        outcome : str, optional
            Outcome name key corresponding to an entry in `self.fitted_models`.

        Returns
        -------
        resolved_model : object
            Fitted estimator or pipeline instance.

        Raises
        ------
        KeyError
            If specified outcome is not found in `self.fitted_models`.
        ValueError
            If model is unspecified and `self.fitted_models` contains multiple models.

        """
        if model is not None:
            return model
        if outcome is not None:
            if outcome in self.fitted_models:
                return self.fitted_models[outcome]
            raise KeyError(
                f"Outcome '{outcome}' not found in runner.fitted_models dictionary."
            )
        if len(self.fitted_models) == 1:
            return next(iter(self.fitted_models.values()))
        raise ValueError(
            "Multiple models found in runner.fitted_models. "
            "Please explicitly specify 'outcome' or pass 'model'."
        )

    def predict(
        self,
        X: pd.DataFrame | npt.NDArray,
        model: Any | None = None,
        outcome: str | None = None,
    ) -> npt.NDArray:
        """
        Predict outcomes for samples in X.

        Parameters
        ----------
        X : pandas.DataFrame or numpy.ndarray
            Features dataset of shape (n_samples, n_features).
        model : object, optional
            Fitted model instance. If None, resolved via `outcome` or `fitted_models`.
        outcome : str, optional
            Outcome key to look up in `self.fitted_models`.

        Returns
        -------
        y_pred : numpy.ndarray
            Predicted values of shape (n_samples,).

        Raises
        ------
        AttributeError
            If the resolved model does not implement a `predict` method.

        """
        target_model = self._get_model(model, outcome)
        if not hasattr(target_model, "predict"):
            raise AttributeError("The underlying model does not implement 'predict'.")
        return np.asarray(target_model.predict(X))

    def _get_predictions(
        self,
        X: pd.DataFrame | npt.NDArray,
        target_model: Any,
        metrics: list[str],
    ) -> npt.NDArray | PredictionBundle:
        """
        Retrieve the model predictions used to compute the requested metrics.

        Subclasses must implement this to resolve the outcome-type-specific
        prediction method.

        Parameters
        ----------
        X : pandas.DataFrame or numpy.ndarray
            Features dataset of shape (n_samples, n_features).
        target_model : object
            The resolved, fitted model to predict with.
        metrics : list of str
            The metric names being evaluated, made available so subclasses
            can decide which prediction method(s) are needed.

        Returns
        -------
        y_pred : numpy.ndarray
            Predictions used as input to metric computation.

        """
        raise NotImplementedError

    def extract_subgroups(
        self,
        X: pd.DataFrame,
        subgroup_specs: dict[str, str | Callable[[pd.DataFrame], pd.Series]],
    ) -> dict[str, dict[str, pd.Index]]:
        """
        Extract sample index subsets for specified data subgroups.

        Parameters
        ----------
        X : pandas.DataFrame
            Input feature dataset.
        subgroup_specs : dict of str to (str or callable)
            Mapping where keys are subgroup category names (e.g., `'age_group'`)
            and values are:
            - Column name (`str`): Groups samples by unique column values.
            - Predicate (`callable`): Function taking `X` and returning a
              boolean `pd.Series` mask.

        Returns
        -------
        subgroups : dict of str to dict of str to pandas.Index
            Nested dictionary structured as
            `{category_name: {subgroup_val: row_indices}}`.

        Raises
        ------
        KeyError
            If a specified column string is missing from `X`.
        TypeError
            If a specification is neither a string nor a callable.

        """
        self.logger.debug(
            f"Extracting subgroups for {len(subgroup_specs)} categories.",
        )
        subgroups: dict[str, dict[str, pd.Index]] = {}

        for cat_name, spec in subgroup_specs.items():
            cat_subgroups: dict[str, pd.Index] = {}

            if isinstance(spec, str) and spec in X.columns:
                # Column string: discrete categorical groupby
                for val, group_df in X.groupby(spec, observed=False):
                    cat_subgroups[str(val)] = group_df.index

            elif isinstance(spec, (list, tuple)):
                # List of custom ranges e.g. [[18, 50], [51, 120]]
                for grp in spec:
                    mask = resolve_subgroup_mask(df=X, column=cat_name, group=grp)
                    grp_key = (
                        f"[{grp[0]}, {grp[1]}]"
                        if isinstance(grp, (list, tuple)) and len(grp) == 2
                        else str(grp)
                    )
                    cat_subgroups[grp_key] = X.index[mask]

            elif callable(spec):
                mask = spec(X)
                if not isinstance(mask, pd.Series):
                    mask = pd.Series(mask, index=X.index)
                cat_subgroups["true"] = X.index[mask]
                cat_subgroups["false"] = X.index[~mask]

            else:
                mask = resolve_subgroup_mask(df=X, column=cat_name, group=spec)
                cat_subgroups[str(spec)] = X.index[mask]

            subgroups[cat_name] = cat_subgroups
            self.logger.debug(
                f"Extracted stratum '{cat_name}' matching groups: {spec}",
            )

        return subgroups

    def _evaluate_slice(
        self,
        y_true: npt.NDArray,
        y_pred: npt.NDArray | PredictionBundle,
        metrics: list[str],
    ) -> dict[str, dict[str, float]]:
        """
        Evaluate a single slice of data with TRIPOD+AI compliant confidence intervals.

        Parameters
        ----------
        y_true : numpy.ndarray
            Ground truth target array for the data slice.
        y_pred : numpy.ndarray
            Predicted probabilities or scores for the data slice.
        metrics : list of str
            List of metric names to evaluate.

        Returns
        -------
        scores : dict of str to dict of str to float
            Nested dictionary mapping metric names to dictionaries containing:
            - `"point_estimate"`: Metric score computed on original dataset slice.
            - `"ci_lower"`: Lower bound of the bootstrap confidence interval.
            - `"ci_upper"`: Upper bound of the bootstrap confidence interval.

        """
        if self.n_bootstraps == 0:
            # No bootstrap requested: report point estimates only
            return self._point_estimates(metrics, y_true, y_pred)

        try:
            return bootstrap_confidence_intervals(
                metrics=metrics,
                y_true=y_true,
                y_pred=y_pred,
                n_bootstraps=self.n_bootstraps,
                ci_level=self.ci_level,
                random_state=self.random_state,
                n_jobs=self.n_jobs,
            )
        except Exception as err:
            self.logger.warning(
                "Bootstrap CI calculation failed for slice (%s). "
                "Falling back to point estimates.",
                err,
            )
            return self._point_estimates(metrics, y_true, y_pred)

    @staticmethod
    def _point_estimates(
        metrics: list[str],
        y_true: npt.NDArray,
        y_pred: npt.NDArray | PredictionBundle,
    ) -> dict[str, dict[str, float]]:
        """
        Compute the metrics on a slice of data without confidence intervals.

        Parameters
        ----------
        y_true : numpy.ndarray
            Ground truth target array for the data slice.
        y_pred : numpy.ndarray
            Predicted probabilities or scores for the data slice.
        metrics : list of str
            List of metric names to evaluate.

        Returns
        -------
        scores : dict of str to dict of str to float
            Nested dictionary mapping metric names to dictionaries with the
            `"point_estimate"` and NaN `"ci_lower"` and `"ci_upper"` bounds.
            The point estimate is NaN if the metrics cannot be computed.

        """
        try:
            point_scores = compute_metrics(metrics, y_true, y_pred)
            return {
                metric: {
                    "point_estimate": float(score),
                    "ci_lower": np.nan,
                    "ci_upper": np.nan,
                }
                for metric, score in zip(metrics, point_scores, strict=True)
            }
        except Exception:
            return {
                metric: {
                    "point_estimate": np.nan,
                    "ci_lower": np.nan,
                    "ci_upper": np.nan,
                }
                for metric in metrics
            }

    def _evaluate_stratum(
        self,
        cat_name: str,
        group_val: str,
        indices: pd.Index,
        X: pd.DataFrame,
        y_arr: npt.NDArray,
        y_pred: npt.NDArray | PredictionBundle,
        eval_metrics: list[str],
        outcome: str | None,
    ) -> tuple[str, str, dict[str, dict[str, float]] | None]:
        """
        Evaluate a single subgroup slice, isolated from `evaluate`'s stratum
        loop for readability (the loop itself runs sequentially; parallelism
        is instead applied to each slice's own bootstrap loop, see
        `bootstrap_confidence_intervals`'s `n_jobs`).

        Parameters
        ----------
        cat_name : str
            Name of the subgroup category (e.g., 'age_group').
        group_val : str
            The specific subgroup value being evaluated (e.g., '[18, 50]').
        indices : pandas.Index
            Row indices (into `X`) belonging to this subgroup.
        X : pandas.DataFrame
            Full feature dataset used for prediction.
        y_arr : numpy.ndarray
            Full ground truth target array aligned with `X`.
        y_pred : numpy.ndarray or PredictionBundle
            Full model predictions aligned with `X`.
        eval_metrics : list of str
            List of metric names to evaluate.
        outcome : str, optional
            Outcome key name, used for logging only.

        Returns
        -------
        cat_name : str
            Echoed subgroup category name, for reassembly by the caller.
        group_val : str
            Echoed subgroup value, for reassembly by the caller.
        scores : dict of str to dict of str to float, or None
            Metric results as returned by `_evaluate_slice`, or `None` if
            the subgroup was empty and evaluation was skipped.

        """
        if len(indices) == 0:
            self.logger.warning(
                "Subgroup '%s=%s' is empty. Skipping.", cat_name, group_val
            )
            return cat_name, group_val, None

        pos_idx = X.index.get_indexer(indices)
        y_sub = y_arr[pos_idx]

        y_pred_sub: npt.NDArray | PredictionBundle
        if isinstance(y_pred, PredictionBundle):
            point_sub = y_pred.point[pos_idx] if y_pred.point is not None else None
            dist_sub = y_pred.dist[pos_idx] if y_pred.dist is not None else None
            y_pred_sub = PredictionBundle(point=point_sub, dist=dist_sub)
        else:
            y_pred_sub = y_pred[pos_idx]

        self.logger.debug(
            f"[{outcome}] Evaluating stratum '{cat_name}' matching group: {group_val}.",
        )
        self.logger.debug(f"[{outcome}] Number of samples in stratum: {len(y_sub)}.")

        return (
            cat_name,
            group_val,
            self._evaluate_slice(y_sub, y_pred_sub, eval_metrics),
        )

    def evaluate(
        self,
        X: pd.DataFrame,
        y: pd.Series | npt.NDArray,
        outcome: str | None = None,
        model: Any | None = None,
        metrics: list[str] | None = None,
        subgroup_specs: dict[str, str | Callable[[pd.DataFrame], pd.Series]]
        | None = None,
        fairness_data: pd.DataFrame | None = None,
        save_artifacts: bool = True,
    ) -> dict[str, Any]:
        """
        Evaluate model performance across the full dataset and optional subgroups.

        Computes point estimates and bootstrap confidence intervals for all requested
        metrics in accordance with TRIPOD+AI guidelines. Results are optionally saved
        to disk via the orchestrator's `ArtifactManager`.

        Parameters
        ----------
        X : pandas.DataFrame
            Feature dataset of shape (n_samples, n_features), used for
            prediction. Contains model predictors only.
        y : pandas.Series or numpy.ndarray
            Ground truth target values of shape (n_samples,).
        outcome : str, optional
            Outcome key name. Used for logging, model resolution, and artifact saving.
        model : object, optional
            Explicit fitted model instance. If None, resolved from `self.fitted_models`.
        metrics : list of str, optional
            List of metrics to evaluate. If None, defaults to `self.metrics`.
        subgroup_specs : dict of str to (str or callable), optional
            Specifications for extracting demographic/clinical subgroups.
        fairness_data : pandas.DataFrame, optional
            Fairness stratification columns aligned with `X` (see
            `MedpipeOrchestrator.fairness_splits`), used to resolve
            `subgroup_specs` instead of `X` when a stratum is not itself a
            model predictor (e.g. a "HOSPITAL" column). Defaults to `X`
            when not provided, preserving strata that are also predictors.
        save_artifacts : bool, default=True
            Whether to write evaluation summary results to disk via `ArtifactManager`.

        Returns
        -------
        results : dict of str to Any
            Nested dictionary containing:
            - `"outcome"`: Name of outcome evaluated.
            - `"overall"`: Dict mapping metric names to dicts of point
              estimates and CIs.
            - `"subgroups"`: Dict mapping subgroup strata to slice metric results.

        """
        outcome_name = outcome or "default_outcome"
        self.logger.info(f"--- Starting model evaluation for outcome: {outcome} ---")

        eval_metrics = metrics if metrics is not None else self.metrics
        target_model = self._get_model(model, outcome)

        y_arr = np.asarray(y)

        # Retrieve model predictions
        y_pred = self._get_predictions(X, target_model, eval_metrics)

        # 1. Compute overall evaluation with confidence intervals
        if self.n_bootstraps == 0:
            self.logger.info(
                "[%s] Computing overall metrics without confidence intervals.",
                outcome_name,
            )
        else:
            self.logger.info(
                "[%s] Computing overall metrics with %d bootstrap iterations "
                "(CI=%.2f).",
                outcome_name,
                self.n_bootstraps,
                self.ci_level,
            )
        results: dict[str, Any] = {
            "outcome": outcome_name,
            "overall": self._evaluate_slice(y_arr, y_pred, eval_metrics),
        }

        # 2. Compute subgroup metrics using identical slice evaluation logic
        if subgroup_specs:
            self.logger.info(f"[{outcome_name}] Evaluating performance across strata.")
            subgroup_source = fairness_data if fairness_data is not None else X
            subgroups = self.extract_subgroups(subgroup_source, subgroup_specs)

            stratum_tasks = [
                (cat_name, group_val, indices)
                for cat_name, cat_groups in subgroups.items()
                for group_val, indices in cat_groups.items()
            ]

            # Evaluated sequentially: the dominant cost is each slice's own
            # bootstrap loop (see `bootstrap_confidence_intervals`'s
            # `n_jobs`), which already spends the configured parallelism
            # budget. Parallelizing this outer loop too would nest
            # `n_jobs` worker processes inside `n_jobs` worker processes,
            # oversubscribing cores for comparatively little gain, since
            # there are far fewer strata than bootstrap resamples.
            stratum_outputs = [
                self._evaluate_stratum(
                    cat_name,
                    group_val,
                    indices,
                    X,
                    y_arr,
                    y_pred,
                    eval_metrics,
                    outcome,
                )
                for cat_name, group_val, indices in stratum_tasks
            ]

            subgroup_results: dict[str, dict[str, dict[str, dict[str, float]]]] = {
                cat_name: {} for cat_name in subgroups
            }
            for cat_name, group_val, scores in stratum_outputs:
                if scores is not None:
                    subgroup_results[cat_name][group_val] = scores

            results["strata"] = subgroup_results

        # 3. Save artifacts using ArtifactManager
        if save_artifacts:
            self._save_evaluation_artifacts(results, outcome=outcome_name)

        self.logger.info(f"--- Finished model evaluation for outcome: {outcome} ---")

        return results

    def _save_evaluation_artifacts(
        self,
        results: dict[str, Any],
        outcome: str,
    ) -> Path:
        """
        Save evaluation outcomes to disk using the ArtifactManager or run directory.

        Parameters
        ----------
        results : dict of str to Any
            Evaluation results dictionary containing overall and subgroup metrics.
        outcome : str
            Name of the outcome evaluated.

        Returns
        -------
        saved_path : pathlib.Path
            Path to the saved artifact file on disk.

        """
        filename = f"{outcome}_evaluation_results.json"
        saved_path: Path | None = None

        artifacts_dir = self.orchestrator.run_dir / "results"
        artifact_mgr = self.orchestrator.artifact_manager

        saved_path = artifact_mgr.save_json(results, artifacts_dir, filename)

        self.logger.info(
            f"[{outcome}] Successfully saved evaluation artifacts to {saved_path}.",
        )
        return saved_path

predict

predict(
    X: DataFrame | NDArray,
    model: Any | None = None,
    outcome: str | None = None,
) -> NDArray

Predict outcomes for samples in X.

Parameters:

Name	Type	Description	Default
X	DataFrame or ndarray	Features dataset of shape (n_samples, n_features).	required
model	object	Fitted model instance. If None, resolved via outcome or fitted_models.	None
outcome	str	Outcome key to look up in self.fitted_models.	None

Returns:

Name	Type	Description
y_pred	ndarray	Predicted values of shape (n_samples,).

Raises:

Type	Description
AttributeError	If the resolved model does not implement a predict method.

Source code in medpipe/pipeline/evaluator.py

def predict(
    self,
    X: pd.DataFrame | npt.NDArray,
    model: Any | None = None,
    outcome: str | None = None,
) -> npt.NDArray:
    """
    Predict outcomes for samples in X.

    Parameters
    ----------
    X : pandas.DataFrame or numpy.ndarray
        Features dataset of shape (n_samples, n_features).
    model : object, optional
        Fitted model instance. If None, resolved via `outcome` or `fitted_models`.
    outcome : str, optional
        Outcome key to look up in `self.fitted_models`.

    Returns
    -------
    y_pred : numpy.ndarray
        Predicted values of shape (n_samples,).

    Raises
    ------
    AttributeError
        If the resolved model does not implement a `predict` method.

    """
    target_model = self._get_model(model, outcome)
    if not hasattr(target_model, "predict"):
        raise AttributeError("The underlying model does not implement 'predict'.")
    return np.asarray(target_model.predict(X))

extract_subgroups

extract_subgroups(
    X: DataFrame,
    subgroup_specs: dict[
        str, str | Callable[[DataFrame], Series]
    ],
) -> dict[str, dict[str, Index]]

Extract sample index subsets for specified data subgroups.

Parameters:

Name	Type	Description	Default
X	DataFrame	Input feature dataset.	required
subgroup_specs	dict of str to (str or callable)	Mapping where keys are subgroup category names (e.g., 'age_group') and values are: - Column name (str): Groups samples by unique column values. - Predicate (callable): Function taking X and returning a boolean pd.Series mask.	required

Returns:

Name	Type	Description
subgroups	dict of str to dict of str to pandas.Index	Nested dictionary structured as {category_name: {subgroup_val: row_indices}}.

Raises:

Type	Description
KeyError	If a specified column string is missing from X.
TypeError	If a specification is neither a string nor a callable.

Source code in medpipe/pipeline/evaluator.py

def extract_subgroups(
    self,
    X: pd.DataFrame,
    subgroup_specs: dict[str, str | Callable[[pd.DataFrame], pd.Series]],
) -> dict[str, dict[str, pd.Index]]:
    """
    Extract sample index subsets for specified data subgroups.

    Parameters
    ----------
    X : pandas.DataFrame
        Input feature dataset.
    subgroup_specs : dict of str to (str or callable)
        Mapping where keys are subgroup category names (e.g., `'age_group'`)
        and values are:
        - Column name (`str`): Groups samples by unique column values.
        - Predicate (`callable`): Function taking `X` and returning a
          boolean `pd.Series` mask.

    Returns
    -------
    subgroups : dict of str to dict of str to pandas.Index
        Nested dictionary structured as
        `{category_name: {subgroup_val: row_indices}}`.

    Raises
    ------
    KeyError
        If a specified column string is missing from `X`.
    TypeError
        If a specification is neither a string nor a callable.

    """
    self.logger.debug(
        f"Extracting subgroups for {len(subgroup_specs)} categories.",
    )
    subgroups: dict[str, dict[str, pd.Index]] = {}

    for cat_name, spec in subgroup_specs.items():
        cat_subgroups: dict[str, pd.Index] = {}

        if isinstance(spec, str) and spec in X.columns:
            # Column string: discrete categorical groupby
            for val, group_df in X.groupby(spec, observed=False):
                cat_subgroups[str(val)] = group_df.index

        elif isinstance(spec, (list, tuple)):
            # List of custom ranges e.g. [[18, 50], [51, 120]]
            for grp in spec:
                mask = resolve_subgroup_mask(df=X, column=cat_name, group=grp)
                grp_key = (
                    f"[{grp[0]}, {grp[1]}]"
                    if isinstance(grp, (list, tuple)) and len(grp) == 2
                    else str(grp)
                )
                cat_subgroups[grp_key] = X.index[mask]

        elif callable(spec):
            mask = spec(X)
            if not isinstance(mask, pd.Series):
                mask = pd.Series(mask, index=X.index)
            cat_subgroups["true"] = X.index[mask]
            cat_subgroups["false"] = X.index[~mask]

        else:
            mask = resolve_subgroup_mask(df=X, column=cat_name, group=spec)
            cat_subgroups[str(spec)] = X.index[mask]

        subgroups[cat_name] = cat_subgroups
        self.logger.debug(
            f"Extracted stratum '{cat_name}' matching groups: {spec}",
        )

    return subgroups

evaluate

evaluate(
    X: DataFrame,
    y: Series | NDArray,
    outcome: str | None = None,
    model: Any | None = None,
    metrics: list[str] | None = None,
    subgroup_specs: dict[
        str, str | Callable[[DataFrame], Series]
    ]
    | None = None,
    fairness_data: DataFrame | None = None,
    save_artifacts: bool = True,
) -> dict[str, Any]

Evaluate model performance across the full dataset and optional subgroups.

Computes point estimates and bootstrap confidence intervals for all requested metrics in accordance with TRIPOD+AI guidelines. Results are optionally saved to disk via the orchestrator's ArtifactManager.

Parameters:

Name	Type	Description	Default
X	DataFrame	Feature dataset of shape (n_samples, n_features), used for prediction. Contains model predictors only.	required
y	Series or ndarray	Ground truth target values of shape (n_samples,).	required
outcome	str	Outcome key name. Used for logging, model resolution, and artifact saving.	None
model	object	Explicit fitted model instance. If None, resolved from self.fitted_models.	None
metrics	list of str	List of metrics to evaluate. If None, defaults to self.metrics.	None
subgroup_specs	dict of str to (str or callable)	Specifications for extracting demographic/clinical subgroups.	None
fairness_data	DataFrame	Fairness stratification columns aligned with X (see MedpipeOrchestrator.fairness_splits), used to resolve subgroup_specs instead of X when a stratum is not itself a model predictor (e.g. a "HOSPITAL" column). Defaults to X when not provided, preserving strata that are also predictors.	None
save_artifacts	bool	Whether to write evaluation summary results to disk via ArtifactManager.	True

Returns:

Name	Type	Description
results	dict of str to Any	Nested dictionary containing: - "outcome": Name of outcome evaluated. - "overall": Dict mapping metric names to dicts of point estimates and CIs. - "subgroups": Dict mapping subgroup strata to slice metric results.

Source code in medpipe/pipeline/evaluator.py

def evaluate(
    self,
    X: pd.DataFrame,
    y: pd.Series | npt.NDArray,
    outcome: str | None = None,
    model: Any | None = None,
    metrics: list[str] | None = None,
    subgroup_specs: dict[str, str | Callable[[pd.DataFrame], pd.Series]]
    | None = None,
    fairness_data: pd.DataFrame | None = None,
    save_artifacts: bool = True,
) -> dict[str, Any]:
    """
    Evaluate model performance across the full dataset and optional subgroups.

    Computes point estimates and bootstrap confidence intervals for all requested
    metrics in accordance with TRIPOD+AI guidelines. Results are optionally saved
    to disk via the orchestrator's `ArtifactManager`.

    Parameters
    ----------
    X : pandas.DataFrame
        Feature dataset of shape (n_samples, n_features), used for
        prediction. Contains model predictors only.
    y : pandas.Series or numpy.ndarray
        Ground truth target values of shape (n_samples,).
    outcome : str, optional
        Outcome key name. Used for logging, model resolution, and artifact saving.
    model : object, optional
        Explicit fitted model instance. If None, resolved from `self.fitted_models`.
    metrics : list of str, optional
        List of metrics to evaluate. If None, defaults to `self.metrics`.
    subgroup_specs : dict of str to (str or callable), optional
        Specifications for extracting demographic/clinical subgroups.
    fairness_data : pandas.DataFrame, optional
        Fairness stratification columns aligned with `X` (see
        `MedpipeOrchestrator.fairness_splits`), used to resolve
        `subgroup_specs` instead of `X` when a stratum is not itself a
        model predictor (e.g. a "HOSPITAL" column). Defaults to `X`
        when not provided, preserving strata that are also predictors.
    save_artifacts : bool, default=True
        Whether to write evaluation summary results to disk via `ArtifactManager`.

    Returns
    -------
    results : dict of str to Any
        Nested dictionary containing:
        - `"outcome"`: Name of outcome evaluated.
        - `"overall"`: Dict mapping metric names to dicts of point
          estimates and CIs.
        - `"subgroups"`: Dict mapping subgroup strata to slice metric results.

    """
    outcome_name = outcome or "default_outcome"
    self.logger.info(f"--- Starting model evaluation for outcome: {outcome} ---")

    eval_metrics = metrics if metrics is not None else self.metrics
    target_model = self._get_model(model, outcome)

    y_arr = np.asarray(y)

    # Retrieve model predictions
    y_pred = self._get_predictions(X, target_model, eval_metrics)

    # 1. Compute overall evaluation with confidence intervals
    if self.n_bootstraps == 0:
        self.logger.info(
            "[%s] Computing overall metrics without confidence intervals.",
            outcome_name,
        )
    else:
        self.logger.info(
            "[%s] Computing overall metrics with %d bootstrap iterations "
            "(CI=%.2f).",
            outcome_name,
            self.n_bootstraps,
            self.ci_level,
        )
    results: dict[str, Any] = {
        "outcome": outcome_name,
        "overall": self._evaluate_slice(y_arr, y_pred, eval_metrics),
    }

    # 2. Compute subgroup metrics using identical slice evaluation logic
    if subgroup_specs:
        self.logger.info(f"[{outcome_name}] Evaluating performance across strata.")
        subgroup_source = fairness_data if fairness_data is not None else X
        subgroups = self.extract_subgroups(subgroup_source, subgroup_specs)

        stratum_tasks = [
            (cat_name, group_val, indices)
            for cat_name, cat_groups in subgroups.items()
            for group_val, indices in cat_groups.items()
        ]

        # Evaluated sequentially: the dominant cost is each slice's own
        # bootstrap loop (see `bootstrap_confidence_intervals`'s
        # `n_jobs`), which already spends the configured parallelism
        # budget. Parallelizing this outer loop too would nest
        # `n_jobs` worker processes inside `n_jobs` worker processes,
        # oversubscribing cores for comparatively little gain, since
        # there are far fewer strata than bootstrap resamples.
        stratum_outputs = [
            self._evaluate_stratum(
                cat_name,
                group_val,
                indices,
                X,
                y_arr,
                y_pred,
                eval_metrics,
                outcome,
            )
            for cat_name, group_val, indices in stratum_tasks
        ]

        subgroup_results: dict[str, dict[str, dict[str, dict[str, float]]]] = {
            cat_name: {} for cat_name in subgroups
        }
        for cat_name, group_val, scores in stratum_outputs:
            if scores is not None:
                subgroup_results[cat_name][group_val] = scores

        results["strata"] = subgroup_results

    # 3. Save artifacts using ArtifactManager
    if save_artifacts:
        self._save_evaluation_artifacts(results, outcome=outcome_name)

    self.logger.info(f"--- Finished model evaluation for outcome: {outcome} ---")

    return results

MedpipeClassifierEvaluator

Bases: BaseEvaluator

Evaluation engine for MedpipeClassifier machine learning models and pipelines.

Provides standard inference interfaces (predict, predict_proba, decision_function) and structured performance evaluation across full datasets and extracted data subgroups. In compliance with TRIPOD+AI reporting guidelines, evaluation metrics include bootstrap confidence intervals. Results are automatically logged and saved to disk using the orchestrator's ArtifactManager.

Parameters:

Name	Type	Description	Default
orchestrator	MedpipeOrchestrator	The pipeline orchestrator instance containing workflow configuration, run directories, and the ArtifactManager.	required
runner	MedpipeClassifierRunner	The pipeline runner instance containing the dictionary of fitted models (fitted_models).	required

Attributes:

Name	Type	Description
orchestrator	MedpipeOrchestrator	Pipeline orchestrator instance.
runner	MedpipeClassifierRunner	Pipeline runner instance.
fitted_models	dict of str to object	Dictionary mapping outcome names to fitted estimators or pipelines.
metrics	list of str	List of metric names used during evaluation.
n_bootstraps	int	Number of bootstrap iterations.
ci_level	float	Target confidence interval level.
random_state	int, np.random.Generator, or None	Random state instance for resampling.
n_jobs	int or None	Number of parallel jobs used to compute each slice's bootstrap resamples (see bootstrap_confidence_intervals).
logger	Logger	Logger instance configured under "medpipe.evaluator".

Methods:

Name	Description
predict	Predict class labels for samples in X.
predict_proba	Predict class probabilities for samples in X.
decision_function	Compute decision function scores for samples in X.
extract_subgroups	Extract index subsets for specified data subgroups.
evaluate	Evaluate model performance with confidence intervals across full data and subgroups.

Source code in medpipe/pipeline/evaluator.py

class MedpipeClassifierEvaluator(BaseEvaluator):
    """
    Evaluation engine for MedpipeClassifier machine learning models and pipelines.

    Provides standard inference interfaces (`predict`, `predict_proba`,
    `decision_function`) and structured performance evaluation across full
    datasets and extracted data subgroups. In compliance with TRIPOD+AI
    reporting guidelines, evaluation metrics include bootstrap confidence
    intervals. Results are automatically logged and saved to disk using the
    orchestrator's `ArtifactManager`.

    Parameters
    ----------
    orchestrator : MedpipeOrchestrator
        The pipeline orchestrator instance containing workflow configuration,
        run directories, and the `ArtifactManager`.
    runner : MedpipeClassifierRunner
        The pipeline runner instance containing the dictionary of fitted models
        (`fitted_models`).

    Attributes
    ----------
    orchestrator : MedpipeOrchestrator
        Pipeline orchestrator instance.
    runner : MedpipeClassifierRunner
        Pipeline runner instance.
    fitted_models : dict of str to object
        Dictionary mapping outcome names to fitted estimators or pipelines.
    metrics : list of str
        List of metric names used during evaluation.
    n_bootstraps : int
        Number of bootstrap iterations.
    ci_level : float
        Target confidence interval level.
    random_state : int, np.random.Generator, or None
        Random state instance for resampling.
    n_jobs : int or None
        Number of parallel jobs used to compute each slice's bootstrap
        resamples (see `bootstrap_confidence_intervals`).
    logger : logging.Logger
        Logger instance configured under `"medpipe.evaluator"`.

    Methods
    -------
    predict(X, model=None, outcome=None)
        Predict class labels for samples in X.
    predict_proba(X, model=None, outcome=None)
        Predict class probabilities for samples in X.
    decision_function(X, model=None, outcome=None)
        Compute decision function scores for samples in X.
    extract_subgroups(X, subgroup_specs)
        Extract index subsets for specified data subgroups.
    evaluate(X, y, outcome=None, metrics=None, subgroup_specs=None, save_artifacts=True)
        Evaluate model performance with confidence intervals across full data
        and subgroups.

    """

    def predict_proba(
        self,
        X: pd.DataFrame | npt.NDArray,
        model: Any | None = None,
        outcome: str | None = None,
    ) -> npt.NDArray:
        """
        Predict class probabilities for samples in X.

        Parameters
        ----------
        X : pandas.DataFrame or numpy.ndarray
            Features dataset of shape (n_samples, n_features).
        model : object, optional
            Fitted model instance. If None, resolved via `outcome` or `fitted_models`.
        outcome : str, optional
            Outcome key to look up in `self.fitted_models`.

        Returns
        -------
        y_proba : numpy.ndarray
            Predicted class probabilities of shape (n_samples, n_classes) or
            (n_samples,).

        Raises
        ------
        AttributeError
            If the resolved model does not implement a `predict_proba` method.

        """
        target_model = self._get_model(model, outcome)
        if not hasattr(target_model, "predict_proba"):
            raise AttributeError(
                "The underlying model does not implement 'predict_proba'."
            )
        return np.asarray(target_model.predict_proba(X))

    def decision_function(
        self,
        X: pd.DataFrame | npt.NDArray,
        model: Any | None = None,
        outcome: str | None = None,
    ) -> npt.NDArray:
        """
        Compute decision function scores for samples in X.

        Parameters
        ----------
        X : pandas.DataFrame or numpy.ndarray
            Features dataset of shape (n_samples, n_features).
        model : object, optional
            Fitted model instance. If None, resolved via `outcome` or `fitted_models`.
        outcome : str, optional
            Outcome key to look up in `self.fitted_models`.

        Returns
        -------
        scores : numpy.ndarray
            Confidence scores or decision function values of shape (n_samples,).

        Raises
        ------
        AttributeError
            If the resolved model does not implement a `decision_function` method.

        """
        target_model = self._get_model(model, outcome)
        if not hasattr(target_model, "decision_function"):
            raise AttributeError(
                "The underlying model does not implement 'decision_function'."
            )
        return np.asarray(target_model.decision_function(X))

    def _get_predictions(
        self,
        X: pd.DataFrame | npt.NDArray,
        target_model: Any,
        metrics: list[str],
    ) -> npt.NDArray:
        """
        Resolve predictions using probability, decision, or label output, in
        that order of preference.

        Parameters
        ----------
        X : pandas.DataFrame or numpy.ndarray
            Features dataset of shape (n_samples, n_features).
        target_model : object
            The resolved, fitted model to predict with.
        metrics : list of str
            Unused for classification; accepted for interface compatibility
            with `BaseEvaluator`.

        Returns
        -------
        y_pred : numpy.ndarray
            Predicted probabilities, decision scores, or class labels.

        """
        if hasattr(target_model, "predict_proba"):
            return self.predict_proba(X, model=target_model)
        elif hasattr(target_model, "decision_function"):
            return self.decision_function(X, model=target_model)
        return self.predict(X, model=target_model)

predict_proba

predict_proba(
    X: DataFrame | NDArray,
    model: Any | None = None,
    outcome: str | None = None,
) -> NDArray

Predict class probabilities for samples in X.

Parameters:

Name	Type	Description	Default
X	DataFrame or ndarray	Features dataset of shape (n_samples, n_features).	required
model	object	Fitted model instance. If None, resolved via outcome or fitted_models.	None
outcome	str	Outcome key to look up in self.fitted_models.	None

Returns:

Name	Type	Description
y_proba	ndarray	Predicted class probabilities of shape (n_samples, n_classes) or (n_samples,).

Raises:

Type	Description
AttributeError	If the resolved model does not implement a predict_proba method.

Source code in medpipe/pipeline/evaluator.py

def predict_proba(
    self,
    X: pd.DataFrame | npt.NDArray,
    model: Any | None = None,
    outcome: str | None = None,
) -> npt.NDArray:
    """
    Predict class probabilities for samples in X.

    Parameters
    ----------
    X : pandas.DataFrame or numpy.ndarray
        Features dataset of shape (n_samples, n_features).
    model : object, optional
        Fitted model instance. If None, resolved via `outcome` or `fitted_models`.
    outcome : str, optional
        Outcome key to look up in `self.fitted_models`.

    Returns
    -------
    y_proba : numpy.ndarray
        Predicted class probabilities of shape (n_samples, n_classes) or
        (n_samples,).

    Raises
    ------
    AttributeError
        If the resolved model does not implement a `predict_proba` method.

    """
    target_model = self._get_model(model, outcome)
    if not hasattr(target_model, "predict_proba"):
        raise AttributeError(
            "The underlying model does not implement 'predict_proba'."
        )
    return np.asarray(target_model.predict_proba(X))

decision_function

decision_function(
    X: DataFrame | NDArray,
    model: Any | None = None,
    outcome: str | None = None,
) -> NDArray

Compute decision function scores for samples in X.

Parameters:

Name	Type	Description	Default
X	DataFrame or ndarray	Features dataset of shape (n_samples, n_features).	required
model	object	Fitted model instance. If None, resolved via outcome or fitted_models.	None
outcome	str	Outcome key to look up in self.fitted_models.	None

Returns:

Name	Type	Description
scores	ndarray	Confidence scores or decision function values of shape (n_samples,).

Raises:

Type	Description
AttributeError	If the resolved model does not implement a decision_function method.

Source code in medpipe/pipeline/evaluator.py

def decision_function(
    self,
    X: pd.DataFrame | npt.NDArray,
    model: Any | None = None,
    outcome: str | None = None,
) -> npt.NDArray:
    """
    Compute decision function scores for samples in X.

    Parameters
    ----------
    X : pandas.DataFrame or numpy.ndarray
        Features dataset of shape (n_samples, n_features).
    model : object, optional
        Fitted model instance. If None, resolved via `outcome` or `fitted_models`.
    outcome : str, optional
        Outcome key to look up in `self.fitted_models`.

    Returns
    -------
    scores : numpy.ndarray
        Confidence scores or decision function values of shape (n_samples,).

    Raises
    ------
    AttributeError
        If the resolved model does not implement a `decision_function` method.

    """
    target_model = self._get_model(model, outcome)
    if not hasattr(target_model, "decision_function"):
        raise AttributeError(
            "The underlying model does not implement 'decision_function'."
        )
    return np.asarray(target_model.decision_function(X))

MedpipeRegressorEvaluator

Bases: BaseEvaluator

Evaluation engine for MedpipeRegressor machine learning models and pipelines.

Provides the standard predict inference interface and structured performance evaluation across full datasets and extracted data subgroups, using point predictions for metrics such as RMSE and MAE. In compliance with TRIPOD+AI reporting guidelines, evaluation metrics include bootstrap confidence intervals. Results are automatically logged and saved to disk using the orchestrator's ArtifactManager.

Parameters:

Name	Type	Description	Default
orchestrator	MedpipeOrchestrator	The pipeline orchestrator instance containing workflow configuration, run directories, and the ArtifactManager.	required
runner	MedpipeRegressorRunner	The pipeline runner instance containing the dictionary of fitted models (fitted_models).	required

Attributes:

Name	Type	Description
orchestrator	MedpipeOrchestrator	Pipeline orchestrator instance.
runner	MedpipeRegressorRunner	Pipeline runner instance.
fitted_models	dict of str to object	Dictionary mapping outcome names to fitted estimators or pipelines.
metrics	list of str	List of metric names used during evaluation.
n_bootstraps	int	Number of bootstrap iterations.
ci_level	float	Target confidence interval level.
random_state	int, np.random.Generator, or None	Random state instance for resampling.
n_jobs	int or None	Number of parallel jobs used to compute each slice's bootstrap resamples (see bootstrap_confidence_intervals).
logger	Logger	Logger instance configured under "medpipe.evaluator".

Methods:

Name	Description
predict	Predict continuous values for samples in X.
predict_dist	Predict the full predictive distribution for samples in X.
extract_subgroups	Extract index subsets for specified data subgroups.
evaluate	Evaluate model performance with confidence intervals across full data and subgroups.

Source code in medpipe/pipeline/evaluator.py

class MedpipeRegressorEvaluator(BaseEvaluator):
    """
    Evaluation engine for MedpipeRegressor machine learning models and
    pipelines.

    Provides the standard `predict` inference interface and structured
    performance evaluation across full datasets and extracted data
    subgroups, using point predictions for metrics such as RMSE and MAE.
    In compliance with TRIPOD+AI reporting guidelines, evaluation metrics
    include bootstrap confidence intervals. Results are automatically
    logged and saved to disk using the orchestrator's `ArtifactManager`.

    Parameters
    ----------
    orchestrator : MedpipeOrchestrator
        The pipeline orchestrator instance containing workflow configuration,
        run directories, and the `ArtifactManager`.
    runner : MedpipeRegressorRunner
        The pipeline runner instance containing the dictionary of fitted models
        (`fitted_models`).

    Attributes
    ----------
    orchestrator : MedpipeOrchestrator
        Pipeline orchestrator instance.
    runner : MedpipeRegressorRunner
        Pipeline runner instance.
    fitted_models : dict of str to object
        Dictionary mapping outcome names to fitted estimators or pipelines.
    metrics : list of str
        List of metric names used during evaluation.
    n_bootstraps : int
        Number of bootstrap iterations.
    ci_level : float
        Target confidence interval level.
    random_state : int, np.random.Generator, or None
        Random state instance for resampling.
    n_jobs : int or None
        Number of parallel jobs used to compute each slice's bootstrap
        resamples (see `bootstrap_confidence_intervals`).
    logger : logging.Logger
        Logger instance configured under `"medpipe.evaluator"`.

    Methods
    -------
    predict(X, model=None, outcome=None)
        Predict continuous values for samples in X.
    predict_dist(X, model=None, outcome=None)
        Predict the full predictive distribution for samples in X.
    extract_subgroups(X, subgroup_specs)
        Extract index subsets for specified data subgroups.
    evaluate(X, y, outcome=None, metrics=None, subgroup_specs=None, save_artifacts=True)
        Evaluate model performance with confidence intervals across full data
        and subgroups.

    """

    def predict_dist(
        self,
        X: pd.DataFrame | npt.NDArray,
        model: Any | None = None,
        outcome: str | None = None,
    ) -> Any:
        """
        Predict the full predictive distribution for samples in X.

        Parameters
        ----------
        X : pandas.DataFrame or numpy.ndarray
            Features dataset of shape (n_samples, n_features).
        model : object, optional
            Fitted model instance. If None, resolved via `outcome` or `fitted_models`.
        outcome : str, optional
            Outcome key to look up in `self.fitted_models`.

        Returns
        -------
        Any
            The distribution object returned by the resolved model's
            `predict_dist` method (e.g. an `ngboost.distns.Normal` or
            `ordboost.distributions.ContinuousPredictiveDistribution`).

        Raises
        ------
        AttributeError
            If the resolved model does not implement a `predict_dist` method.

        """
        target_model = self._get_model(model, outcome)
        if not hasattr(target_model, "predict_dist"):
            raise AttributeError(
                "The underlying model does not implement 'predict_dist'."
            )
        return target_model.predict_dist(X)

    def _get_predictions(
        self,
        X: pd.DataFrame | npt.NDArray,
        target_model: Any,
        metrics: list[str],
    ) -> PredictionBundle:
        """
        Resolve point predictions, and a full predictive distribution when
        needed, used to compute the requested metrics.

        Parameters
        ----------
        X : pandas.DataFrame or numpy.ndarray
            Features dataset of shape (n_samples, n_features).
        target_model : object
            The resolved, fitted model to predict with.
        metrics : list of str
            The metric names being evaluated. If any requires a
            `predict_dist` response method (e.g. crps), `predict_dist(X)`
            is also called.

        Returns
        -------
        PredictionBundle
            Bundle carrying point predictions and, when needed, the full
            predictive distribution.

        """
        needs_dist = any(
            MetricRegistry.get(m).response_method == "predict_dist" for m in metrics
        )
        point = self.predict(X, model=target_model)
        dist = self.predict_dist(X, model=target_model) if needs_dist else None
        return PredictionBundle(point=point, dist=dist)

predict_dist

predict_dist(
    X: DataFrame | NDArray,
    model: Any | None = None,
    outcome: str | None = None,
) -> Any

Predict the full predictive distribution for samples in X.

Parameters:

Name	Type	Description	Default
X	DataFrame or ndarray	Features dataset of shape (n_samples, n_features).	required
model	object	Fitted model instance. If None, resolved via outcome or fitted_models.	None
outcome	str	Outcome key to look up in self.fitted_models.	None

Returns:

Type	Description
Any	The distribution object returned by the resolved model's predict_dist method (e.g. an ngboost.distns.Normal or ordboost.distributions.ContinuousPredictiveDistribution).

Raises:

Type	Description
AttributeError	If the resolved model does not implement a predict_dist method.

Source code in medpipe/pipeline/evaluator.py

def predict_dist(
    self,
    X: pd.DataFrame | npt.NDArray,
    model: Any | None = None,
    outcome: str | None = None,
) -> Any:
    """
    Predict the full predictive distribution for samples in X.

    Parameters
    ----------
    X : pandas.DataFrame or numpy.ndarray
        Features dataset of shape (n_samples, n_features).
    model : object, optional
        Fitted model instance. If None, resolved via `outcome` or `fitted_models`.
    outcome : str, optional
        Outcome key to look up in `self.fitted_models`.

    Returns
    -------
    Any
        The distribution object returned by the resolved model's
        `predict_dist` method (e.g. an `ngboost.distns.Normal` or
        `ordboost.distributions.ContinuousPredictiveDistribution`).

    Raises
    ------
    AttributeError
        If the resolved model does not implement a `predict_dist` method.

    """
    target_model = self._get_model(model, outcome)
    if not hasattr(target_model, "predict_dist"):
        raise AttributeError(
            "The underlying model does not implement 'predict_dist'."
        )
    return target_model.predict_dist(X)

estimator

Pipeline subclass exposing a unified distributional prediction interface.

DistributionalPipeline

Bases: Pipeline

A scikit-learn Pipeline that exposes a unified predict_dist method, delegating to whichever of predict_dist or pred_dist the final estimator implements.

Different probabilistic boosting libraries name their distributional prediction method differently (e.g. NGBoost's NGBRegressor.pred_dist, OrdBoost's predict_dist). This subclass lets the rest of MedPipe call a single, consistent method regardless of which library produced the final estimator.

Adds no constructor parameters or instance state beyond Pipeline itself, so sklearn.base.clone() (used internally by GridSearchCV and cross_validate) preserves this subclass unchanged.

Source code in medpipe/pipeline/estimator.py

class DistributionalPipeline(Pipeline):
    """
    A scikit-learn Pipeline that exposes a unified `predict_dist` method,
    delegating to whichever of `predict_dist` or `pred_dist` the final
    estimator implements.

    Different probabilistic boosting libraries name their distributional
    prediction method differently (e.g. NGBoost's `NGBRegressor.pred_dist`,
    OrdBoost's `predict_dist`). This subclass lets the rest of
    MedPipe call a single, consistent method regardless of which library
    produced the final estimator.

    Adds no constructor parameters or instance state beyond `Pipeline`
    itself, so `sklearn.base.clone()` (used internally by `GridSearchCV`
    and `cross_validate`) preserves this subclass unchanged.

    """

    def __sklearn_is_fitted__(self) -> bool:
        """
        Indicate whether the pipeline's final estimator has been fitted.

        Falls back to checking NGBoost's `base_models` list when the
        default scikit-learn check reports "not fitted". NGBoost estimators
        (e.g. `NGBRegressor`) never set any scikit-learn-conventional
        fitted attribute (a name ending in an underscore, not starting with
        `__`), which is exactly what `sklearn.utils.validation
        .check_is_fitted`'s default heuristic looks for on the final
        estimator. That makes `Pipeline.__sklearn_is_fitted__()` report
        "not fitted" even immediately after a fully successful `fit()`
        call, which in turn makes every `Pipeline` method that starts with
        `check_is_fitted(self)` (`predict`, `predict_proba`, `score`, ...)
        raise `NotFittedError` unconditionally for a fitted NGBoost model.
        `NGBoost.fit()` always initializes `base_models = []` and appends
        one entry per completed boosting round, so a non-empty list is a
        reliable "has been fitted" signal across the NGBoost model family.

        Returns
        -------
        bool
            True if the pipeline's final estimator has been fitted.

        """
        if super().__sklearn_is_fitted__():
            return True

        final_estimator = self.steps[-1][1] if self.steps else None
        return bool(getattr(final_estimator, "base_models", None))

    def predict_dist(self, X: pd.DataFrame | np.ndarray) -> Any:
        """
        Predict the full predictive distribution for samples in X.

        Parameters
        ----------
        X : pandas.DataFrame or numpy.ndarray
            Features dataset of shape (n_samples, n_features).

        Returns
        -------
        Any
            The distribution object returned by the final estimator's
            `predict_dist` or `pred_dist` method (e.g. an
            `ngboost.distns.Normal` instance).

        Raises
        ------
        AttributeError
            If the final estimator implements neither `predict_dist` nor
            `pred_dist`.

        """
        Xt = X
        if len(self.steps) > 1:
            Xt = self[:-1].transform(Xt)

        final_estimator = self.steps[-1][1]

        if hasattr(final_estimator, "predict_dist"):
            return final_estimator.predict_dist(Xt)
        elif hasattr(final_estimator, "pred_dist"):
            return final_estimator.pred_dist(Xt)

        raise AttributeError(
            f"{type(final_estimator).__name__} implements neither "
            "'predict_dist' nor 'pred_dist'."
        )

predict_dist

predict_dist(X: DataFrame | ndarray) -> Any

Predict the full predictive distribution for samples in X.

Parameters:

Name	Type	Description	Default
X	DataFrame or ndarray	Features dataset of shape (n_samples, n_features).	required

Returns:

Type	Description
Any	The distribution object returned by the final estimator's predict_dist or pred_dist method (e.g. an ngboost.distns.Normal instance).

Raises:

Type	Description
AttributeError	If the final estimator implements neither predict_dist nor pred_dist.

Source code in medpipe/pipeline/estimator.py

def predict_dist(self, X: pd.DataFrame | np.ndarray) -> Any:
    """
    Predict the full predictive distribution for samples in X.

    Parameters
    ----------
    X : pandas.DataFrame or numpy.ndarray
        Features dataset of shape (n_samples, n_features).

    Returns
    -------
    Any
        The distribution object returned by the final estimator's
        `predict_dist` or `pred_dist` method (e.g. an
        `ngboost.distns.Normal` instance).

    Raises
    ------
    AttributeError
        If the final estimator implements neither `predict_dist` nor
        `pred_dist`.

    """
    Xt = X
    if len(self.steps) > 1:
        Xt = self[:-1].transform(Xt)

    final_estimator = self.steps[-1][1]

    if hasattr(final_estimator, "predict_dist"):
        return final_estimator.predict_dist(Xt)
    elif hasattr(final_estimator, "pred_dist"):
        return final_estimator.pred_dist(Xt)

    raise AttributeError(
        f"{type(final_estimator).__name__} implements neither "
        "'predict_dist' nor 'pred_dist'."
    )