API: modeling.cross_validation

Cross-validation module providing five strategies: K-Fold, Stratified K-Fold, Shuffle Split, Time Series Split, and Nested CV.

The main entry point is perform_cross_validation(), which dispatches to the appropriate sklearn splitter (or the custom nested CV loop). For usage details and examples, see the Cross-Validation Guide.

Key functions:

• perform_cross_validation() — Run CV with any of the five methods.
• _sort_by_time() — Auto-sort data chronologically for Time Series Split.
• _build_splitter() — Build a sklearn splitter from a cv_type string.
• _perform_nested_cv() — Dual-loop nested CV (outer evaluation + inner HP stability).
• _aggregate_metrics() — Compute mean/std/min/max across folds.

skyulf.modeling.cross_validation

Cross-validation logic for V2 modeling.

perform_cross_validation(calculator, applier, X, y, config, n_folds=5, cv_type='k_fold', shuffle=True, random_state=DEFAULT_RANDOM_STATE, time_column=None, progress_callback=None, log_callback=None, preprocessing=None, *, cv_nested_type='auto', group_column=None, gap=0, test_size=None, max_train_size=None, inner_folds=None, sample_weight=None)

Performs K-Fold cross-validation.

Parameters:

Name	Type	Description	Default
sample_weight	Any	Optional positional training weights; held-out metrics stay unweighted.	None
calculator	BaseModelCalculator	The model calculator (fit logic).	required
applier	BaseModelApplier	The model applier (predict logic).	required
X	DataFrame | SkyulfDataFrame	Features.	required
y	Series | Any	Target.	required
config	dict[str, Any]	Model configuration.	required
n_folds	int	Number of folds.	5
cv_type	str	Type of CV.	'k_fold'
shuffle	bool	Whether to shuffle data before splitting (for KFold/Stratified).	True
random_state	int	Random seed for shuffling.	DEFAULT_RANDOM_STATE
time_column	str | None	Optional column name for sorting when using time_series_split.	None
cv_nested_type	str	Explicit nested split policy; auto retains task defaults.	'auto'
group_column	str | None	Split-only entity identifier excluded from model features.	None
gap	int	Number of omitted rows between temporal training and validation.	0
test_size	int | None	Optional row count per temporal validation fold.	None
max_train_size	int | None	Optional rolling training window row limit.	None
inner_folds	int | None	Optional fold count for explicit nested policies.	None
progress_callback	Callable[[int, int], None] | None	Optional callback(current_fold, total_folds).	None
log_callback	Callable[[str], None] | None	Optional callback for logging messages.	None
preprocessing	FoldPreprocessor | None	Optional per-fold preprocessor (F-15). When given, it is re-fit on each fold's training rows and applied to the held-out rows, so preprocessing statistics never see held-out data. None means the caller already transformed the data before splitting.	None

Returns:

Type	Description
dict[str, Any]	Dict containing aggregated metrics and per-fold details.

Source code in skyulf-core/skyulf/modeling/cross_validation.py

def perform_cross_validation(
    calculator: BaseModelCalculator,
    applier: BaseModelApplier,
    X: pd.DataFrame | SkyulfDataFrame,
    y: pd.Series | Any,
    config: dict[str, Any],
    n_folds: int = 5,
    cv_type: str = "k_fold",  # k_fold, stratified_k_fold, time_series_split, shuffle_split, nested_cv
    shuffle: bool = True,
    random_state: int = DEFAULT_RANDOM_STATE,
    time_column: str | None = None,
    progress_callback: Callable[[int, int], None] | None = None,
    log_callback: Callable[[str], None] | None = None,
    preprocessing: FoldPreprocessor | None = None,
    *,
    cv_nested_type: str = "auto",
    group_column: str | None = None,
    gap: int = 0,
    test_size: int | None = None,
    max_train_size: int | None = None,
    inner_folds: int | None = None,
    sample_weight: Any = None,
) -> dict[str, Any]:
    """Performs K-Fold cross-validation.

    Args:
        sample_weight: Optional positional training weights; held-out metrics stay unweighted.
        calculator: The model calculator (fit logic).
        applier: The model applier (predict logic).
        X: Features.
        y: Target.
        config: Model configuration.
        n_folds: Number of folds.
        cv_type: Type of CV.
        shuffle: Whether to shuffle data before splitting (for KFold/Stratified).
        random_state: Random seed for shuffling.
        time_column: Optional column name for sorting when using time_series_split.
        cv_nested_type: Explicit nested split policy; auto retains task defaults.
        group_column: Split-only entity identifier excluded from model features.
        gap: Number of omitted rows between temporal training and validation.
        test_size: Optional row count per temporal validation fold.
        max_train_size: Optional rolling training window row limit.
        inner_folds: Optional fold count for explicit nested policies.
        progress_callback: Optional callback(current_fold, total_folds).
        log_callback: Optional callback for logging messages.
        preprocessing: Optional per-fold preprocessor (F-15). When given, it is
            re-fit on each fold's training rows and applied to the held-out rows,
            so preprocessing statistics never see held-out data. ``None`` means
            the caller already transformed the data before splitting.

    Returns:
        Dict containing aggregated metrics and per-fold details.
    """
    from ._policy_cv import (  # noqa: PLC0415 - avoid tuning import cycle
        perform_policy_cv,
        policy_config,
    )
    from ._tuning.cv_policy import (  # noqa: PLC0415 - avoid tuning import cycle
        uses_explicit_policy,
    )
    from ._tuning.history_policy import has_temporal_history  # noqa: PLC0415 - tuning import cycle

    sample_weight = prepare_weights(sample_weight, len(X), preprocessing)
    y = _indexable_target(y)
    policy = policy_config(
        cv_type,
        n_folds,
        shuffle,
        random_state,
        time_column,
        {
            "cv_nested_type": cv_nested_type,
            "group_column": group_column,
            "gap": gap,
            "test_size": test_size,
            "max_train_size": max_train_size,
            "inner_folds": inner_folds,
        },
    )
    if uses_explicit_policy(policy) or has_temporal_history(preprocessing):
        return perform_policy_cv(
            calculator,
            X,
            y,
            config,
            policy,
            preprocessing,
            log_callback,
            progress_callback,
            sample_weight,
        )
    problem_type = calculator.problem_type

    if log_callback:
        log_callback(f"Starting Cross-Validation (Folds: {n_folds}, Type: {cv_type})")

    # For Time Series Split, sort data chronologically. Only applies to
    # DataFrame-like X (pandas or Polars); a plain array has no columns to
    # sort/drop by, so time_series_split relies on the caller's row order.
    if cv_type == "time_series_split" and hasattr(X, "columns"):
        X, y, positions = _sort_by_time(X, y, time_column, log_callback, logger, True)
        sample_weight = take_weights(sample_weight, positions)

    # Handle nested CV separately
    if cv_type == "nested_cv":
        return _perform_nested_cv(
            calculator=calculator,
            applier=applier,
            X=X,
            y=y,
            config=config,
            n_folds=n_folds,
            shuffle=shuffle,
            random_state=random_state,
            progress_callback=progress_callback,
            log_callback=log_callback,
            preprocessing=preprocessing,
            sample_weight=sample_weight,
        )

    # 1. Setup Splitter (delegates to _build_splitter so unknown cv_type
    # values get the same warning/fallback behavior in both call paths).
    splitter = _build_splitter(
        cv_type=cv_type,
        n_folds=n_folds,
        problem_type=problem_type,
        shuffle=shuffle,
        random_state=random_state,
    )

    fold_results = []

    # Ensure numpy for splitting using the Bridge
    X_arr, y_arr = SklearnBridge.to_sklearn((X, y))

    preflight_weights(sample_weight, splitter, X_arr, y_arr)
    # 2. Iterate Folds
    for fold_idx, (train_idx, val_idx) in enumerate(splitter.split(X_arr, y_arr)):
        fold_results.append(
            _run_cv_fold(
                calculator=calculator,
                X=X,
                y=y,
                train_idx=train_idx,
                val_idx=val_idx,
                config=config,
                problem_type=problem_type,
                fold_idx=fold_idx,
                n_folds=n_folds,
                progress_callback=progress_callback,
                log_callback=log_callback,
                preprocessing=preprocessing,
                sample_weight=sample_weight,
            )
        )

    # 3. Aggregate
    fold_metrics = [cast(dict[str, float], r["metrics"]) for r in fold_results]
    aggregated = _aggregate_metrics(fold_metrics)

    if log_callback:
        log_callback(f"Cross-Validation Completed. Aggregated Metrics: {aggregated}")

    return {
        "aggregated_metrics": aggregated,
        "folds": fold_results,
        "cv_config": {
            "n_folds": n_folds,
            "cv_type": cv_type,
            "shuffle": shuffle,
            "random_state": random_state,
        },
    }