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=42, time_column=None, progress_callback=None, log_callback=None)

Performs K-Fold cross-validation.

Parameters:

Name	Type	Description	Default
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.	42
time_column	str | None	Optional column name for sorting when using time_series_split.	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

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 = 42,
    time_column: str | None = None,
    progress_callback: Callable[[int, int], None] | None = None,
    log_callback: Callable[[str], None] | None = None,
) -> dict[str, Any]:
    """
    Performs K-Fold cross-validation.

    Args:
        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.
        progress_callback: Optional callback(current_fold, total_folds).
        log_callback: Optional callback for logging messages.

    Returns:
        Dict containing aggregated metrics and per-fold details.
    """
    import logging

    logger = logging.getLogger(__name__)
    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 = _sort_by_time(X, y, time_column, log_callback, logger)

    # 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,
        )

    # 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))

    # 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,
            )
        )

    # 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,
        },
    }