Validation Cross Validation

cross_validation

FUNCTION	DESCRIPTION
cross_validation	Evaluate estimator performance using cross-validation on a single dataset.

Classes

Functions

cross_validation(estimator: Any, X: torch.Tensor, fit_score_fn: Callable, weights: torch.Tensor | None = None, target: torch.Tensor | None = None, n_cv: int | Any = 5, shuffle: bool = False, random_state: int | None = None, device: str | torch.device | None = None, n_jobs: int = 1, backend: str = 'threading', verbose: int = 0, pre_dispatch: str = '2*n_jobs', return_required: list[str] = ['score_test', 'score_train'], return_estimator: bool = False, return_times: bool = True, error_score: float | str = 'raise', dtype: torch.dtype = torch.float32) -> dict[str, Any]

Evaluate estimator performance using cross-validation on a single dataset.

Performs K-fold cross-validation by splitting data into train/test folds, fitting the estimator on each train fold, and scoring on the test fold.

This function wraps dispatch_cross_validation() with simplified interface for single-dataset validation.

PARAMETER	DESCRIPTION
estimator	Unfitted estimator object. Must have fit() method. TYPE: Any
X	Input data of shape (n_samples, ...) TYPE: Tensor
fit_score_fn	User-defined fit and score function with signature: fit_score_fn(estimator, train_data, test_data) -> dict[str, Tensor] train_data: dict with key 'X' -> DataBatch (train split) test_data: dict with key 'X' -> DataBatch (test split) Must return dict with keys: 'score_test': Test fold score (scalar Tensor, required) 'score_train': Train fold score (scalar Tensor, optional) TYPE: Callable
weights	Sample weights of shape (n_samples,). TYPE: torch.Tensor | None, optional, by default None DEFAULT: None
target	Target values of shape (n_samples,) for supervised methods. TYPE: torch.Tensor | None, optional, by default None DEFAULT: None
n_cv	Number of cross-validation folds (int), or a CV splitter object (e.g., KFold, StratifiedKFold from sklearn). TYPE: int | Any, by default 5 DEFAULT: 5
shuffle	Whether to shuffle data before splitting into folds. Only used when n_cv is an int. TYPE: bool, by default False DEFAULT: False
random_state	Random seed for reproducible shuffling. Only used when n_cv is an int. TYPE: int | None, optional, by default None DEFAULT: None
device	Device to run cross-validation on ('cpu', 'cuda', 'cuda:0', etc.). If None, uses the device of input data X. When using CUDA device, parallel processing is disabled (n_jobs=1) and folds are executed sequentially on GPU. TYPE: str | torch.device | None, optional, by default None DEFAULT: None
n_jobs	Number of parallel jobs. -1 uses all processors. Automatically set to 1 when using CUDA device. TYPE: int, by default 1 DEFAULT: 1
backend	Joblib backend for parallelization. TYPE: str, by default "threading" DEFAULT: 'threading'
verbose	Verbosity level for joblib. TYPE: int, by default 0 DEFAULT: 0
pre_dispatch	Number of jobs to pre-dispatch for parallel execution. TYPE: str, by default "2*n_jobs" DEFAULT: '2*n_jobs'
return_required	Required key to return from cross validation. TYPE: list[str] = ["score_test", "score_train"], DEFAULT: ['score_test', 'score_train']
return_estimator	Whether to return fitted estimators from each fold. TYPE: bool, by default False DEFAULT: False
return_times	Whether to include fold execution times in output. TYPE: bool, by default True DEFAULT: True
error_score	Behavior on fit or score errors: 'raise': Re-raise exception 'warn': Assign nan and issue warning numeric value: Assign that value TYPE: float | str, by default "raise" DEFAULT: 'raise'
dtype	Data type for error score tensors. TYPE: dtype DEFAULT: torch.float32

RETURNS	DESCRIPTION
dict[str, Any]	Dictionary with keys: 'score_test': Test scores per fold, shape (n_cv,) 'score_train': Train scores per fold, shape (n_cv,) (always included) 'time': Fold execution times in seconds, length n_cv (if return_times=True) 'estimator': List of fitted estimators from each fold (if return_estimator=True)

Examples:

Basic 5-fold cross-validation:

>>> from spectre.decomposition import PrincipalComponentAnalysis as PCA
>>> from spectre.validation import cross_validate
>>> import torch
>>>
>>> # Define fit/score callback
>>> def fit_score(est, train_data, test_data):
...     train_batch = train_data["X"]
...     test_batch = test_data["X"]
...
...     # Fit on train fold
...     est.fit(train_batch.data, weights=train_batch.weights)
...
...     # Evaluate on both folds
...     score_train = est.score(train_batch.data)
...     score_test = est.score(test_batch.data)
...
...     return {"score_train": score_train, "score_test": score_test}
>>>
>>> # Run cross-validation
>>> pca = PCA(n_components=5)
>>> X = torch.randn(100, 20)
>>> results = cross_validate(pca, X, fit_score_fn=fit_score, n_cv=5)

With shuffled K-fold and weighted samples:

>>> results = cross_validate(
...     pca,
...     X,
...     fit_score_fn=fit_score,
...     weights=torch.ones(100),
...     n_cv=10,
...     shuffle=True,
...     random_state=42,
... )

With stratified CV splitter (for classification):

>>> from sklearn.model_selection import StratifiedKFold
>>>
>>> y = torch.randint(0, 3, (100,))  # 3 classes
>>> cv_splitter = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
>>> results = cross_validate(
...     pca,
...     X,
...     fit_score_fn=fit_score,
...     target=y,
...     n_cv=cv_splitter,
... )

Parallel evaluation on CPU with fitted estimator retention:

>>> results = cross_validate(
...     pca,
...     X,
...     fit_score_fn=fit_score,
...     n_cv=5,
...     n_jobs=-1,  # All cores
...     return_estimator=True,
...     return_times=True,
... )
>>> # Access fold-specific estimators
>>> fold_1_model = results["estimator"][0]

Source code in spectre/validation/cross_validation.py

def cross_validation(
    estimator: Any,
    X: torch.Tensor,
    fit_score_fn: Callable,
    weights: torch.Tensor | None = None,
    target: torch.Tensor | None = None,
    n_cv: int | Any = 5,
    shuffle: bool = False,
    random_state: int | None = None,
    device: str | torch.device | None = None,
    n_jobs: int = 1,
    backend: str = "threading",
    verbose: int = 0,
    pre_dispatch: str = "2*n_jobs",
    return_required: list[str] = ["score_test", "score_train"],
    return_estimator: bool = False,
    return_times: bool = True,
    error_score: float | str = "raise",
    dtype: torch.dtype = torch.float32,
) -> dict[str, Any]:
    """
    Evaluate estimator performance using cross-validation on a single dataset.

    Performs K-fold cross-validation by splitting data into train/test folds,
    fitting the estimator on each train fold, and scoring on the test fold.

    This function wraps `dispatch_cross_validation()` with simplified interface
    for single-dataset validation.

    Parameters
    ----------
    estimator : Any
        Unfitted estimator object. Must have `fit()` method.

    X : torch.Tensor
        Input data of shape `(n_samples, ...)`

    fit_score_fn : Callable
        User-defined fit and score function with signature:

        `fit_score_fn(estimator, train_data, test_data) -> dict[str, Tensor]`

        - `train_data`: dict with key `'X'` -> `DataBatch` (train split)
        - `test_data`: dict with key `'X'` -> `DataBatch` (test split)
        - Must return dict with keys:

          - `'score_test'`: Test fold score (scalar `Tensor`, required)
          - `'score_train'`: Train fold score (scalar `Tensor`, optional)

    weights : torch.Tensor | None, optional, by default None
        Sample weights of shape (n_samples,).

    target : torch.Tensor | None, optional, by default None
        Target values of shape (n_samples,) for supervised methods.

    n_cv : int | Any, by default 5
        Number of cross-validation folds (int), or a CV splitter object
        (e.g., KFold, StratifiedKFold from sklearn).

    shuffle : bool, by default False
        Whether to shuffle data before splitting into folds.
        Only used when `n_cv` is an int.

    random_state : int | None, optional, by default None
        Random seed for reproducible shuffling.
        Only used when `n_cv` is an int.

    device : str | torch.device | None, optional, by default None
        Device to run cross-validation on ('cpu', 'cuda', 'cuda:0', etc.).
        If None, uses the device of input data X.
        When using CUDA device, parallel processing is disabled (n_jobs=1)
        and folds are executed sequentially on GPU.

    n_jobs : int, by default 1
        Number of parallel jobs. -1 uses all processors.
        Automatically set to 1 when using CUDA device.

    backend : str, by default "threading"
        Joblib backend for parallelization.

    verbose : int, by default 0
        Verbosity level for joblib.

    pre_dispatch : str, by default "2*n_jobs"
        Number of jobs to pre-dispatch for parallel execution.

    return_required : list[str] = ["score_test", "score_train"],
        Required key to return from cross validation.

    return_estimator : bool, by default False
        Whether to return fitted estimators from each fold.

    return_times : bool, by default True
        Whether to include fold execution times in output.

    error_score : float | str, by default "raise"
        Behavior on fit or score errors:

        - `'raise'`: Re-raise exception
        - `'warn'`: Assign `nan` and issue warning
        - numeric value: Assign that value

    dtype : torch.dtype, default torch.float32
        Data type for error score tensors.

    Returns
    -------
    dict[str, Any]
        Dictionary with keys:

        - `'score_test'`: Test scores per fold, shape `(n_cv,)`
        - `'score_train'`: Train scores per fold, shape `(n_cv,)`
          (always included)
        - `'time'`: Fold execution times in seconds, length `n_cv`
          (if `return_times=True`)
        - `'estimator'`: List of fitted estimators from each fold
          (if `return_estimator=True`)

    Examples
    --------
    Basic 5-fold cross-validation:

    >>> from spectre.decomposition import PrincipalComponentAnalysis as PCA
    >>> from spectre.validation import cross_validate
    >>> import torch
    >>>
    >>> # Define fit/score callback
    >>> def fit_score(est, train_data, test_data):
    ...     train_batch = train_data["X"]
    ...     test_batch = test_data["X"]
    ...
    ...     # Fit on train fold
    ...     est.fit(train_batch.data, weights=train_batch.weights)
    ...
    ...     # Evaluate on both folds
    ...     score_train = est.score(train_batch.data)
    ...     score_test = est.score(test_batch.data)
    ...
    ...     return {"score_train": score_train, "score_test": score_test}
    >>>
    >>> # Run cross-validation
    >>> pca = PCA(n_components=5)
    >>> X = torch.randn(100, 20)
    >>> results = cross_validate(pca, X, fit_score_fn=fit_score, n_cv=5)

    With shuffled K-fold and weighted samples:

    >>> results = cross_validate(
    ...     pca,
    ...     X,
    ...     fit_score_fn=fit_score,
    ...     weights=torch.ones(100),
    ...     n_cv=10,
    ...     shuffle=True,
    ...     random_state=42,
    ... )

    With stratified CV splitter (for classification):

    >>> from sklearn.model_selection import StratifiedKFold
    >>>
    >>> y = torch.randint(0, 3, (100,))  # 3 classes
    >>> cv_splitter = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
    >>> results = cross_validate(
    ...     pca,
    ...     X,
    ...     fit_score_fn=fit_score,
    ...     target=y,
    ...     n_cv=cv_splitter,
    ... )

    Parallel evaluation on CPU with fitted estimator retention:

    >>> results = cross_validate(
    ...     pca,
    ...     X,
    ...     fit_score_fn=fit_score,
    ...     n_cv=5,
    ...     n_jobs=-1,  # All cores
    ...     return_estimator=True,
    ...     return_times=True,
    ... )
    >>> # Access fold-specific estimators
    >>> fold_1_model = results["estimator"][0]
    """
    batch = DataBatch(data=X, weights=weights, target=target)

    # Delegate to dispatch_cross_validation
    return dispatch_cross_validation(
        estimator=estimator,
        datasets={"X": batch},
        fit_score_fn=fit_score_fn,
        n_cv=n_cv,
        shuffle=shuffle,
        random_state=random_state,
        device=device,
        n_jobs=n_jobs,
        backend=backend,
        verbose=verbose,
        pre_dispatch=pre_dispatch,
        return_required=return_required,
        return_estimator=return_estimator,
        return_times=return_times,
        error_score=error_score,
        dtype=dtype,
    )