Data

churn_prediction.data.ingestion

Kaggle Dataset Acquisition Module.

Provides utilities for downloading and managing customer churn datasets from Kaggle. It handles API authentication, idempotent downloads, and cleanup of unnecessary artifacts to ensure a clean raw data directory.

get_dataset(output_dir, expected_files=None, competition_name='playground-series-s6e3', *, force_download=False)

Download and prepare the Kaggle competition dataset.

The function operates idempotently: when all expected dataset artifacts already exist locally and pass integrity checks, no download is performed. Otherwise, the dataset is downloaded, converted into the required format, and cleaned of unnecessary files.

Parameters:

Name	Type	Description	Default
output_dir	Path or str	Destination directory for the dataset artifacts.	required
expected_files	frozenset of str or None	Set of filenames required for a valid local dataset. If None, the module's default expected file set is used.	None
competition_name	str	Kaggle competition identifier used to retrieve the dataset.	"playground-series-s6e3"
force_download	bool	Whether to force dataset retrieval and processing even when the existing local files pass validation.	False

Raises:

Type	Description
ValueError	If required Kaggle credentials are unavailable.
RuntimeError	If downloading, extracting, converting, or preparing the dataset fails.
OSError	If required directories or dataset files cannot be created or modified.

Source code in src/churn_prediction/data/ingestion.py

def get_dataset(
    output_dir: Path | str,
    expected_files: frozenset[str] | None = None,
    competition_name: str = "playground-series-s6e3",
    *,
    force_download: bool = False,
) -> None:
    """Download and prepare the Kaggle competition dataset.

    The function operates idempotently: when all expected dataset artifacts
    already exist locally and pass integrity checks, no download is performed.
    Otherwise, the dataset is downloaded, converted into the required format,
    and cleaned of unnecessary files.

    Parameters
    ----------
    output_dir : pathlib.Path or str
        Destination directory for the dataset artifacts.
    expected_files : frozenset of str or None, optional
        Set of filenames required for a valid local dataset. If ``None``,
        the module's default expected file set is used.
    competition_name : str, default="playground-series-s6e3"
        Kaggle competition identifier used to retrieve the dataset.
    force_download : bool, default=False
        Whether to force dataset retrieval and processing even when the
        existing local files pass validation.

    Raises
    ------
    ValueError
        If required Kaggle credentials are unavailable.
    RuntimeError
        If downloading, extracting, converting, or preparing the dataset
        fails.
    OSError
        If required directories or dataset files cannot be created or modified.
    """
    logger = get_logger()
    load_dotenv(override=False)

    output_dir = Path(output_dir).expanduser().resolve()
    targets = expected_files if expected_files is not None else DEFAULT_EXPECTED_FILES

    # 0. Check if dataset already exists locally and is non-corrupt
    if not force_download:
        if _has_valid_files(output_dir, targets):
            logger.info("Valid dataset already exists at %s. Skipping download.", output_dir)
            return
        logger.debug(
            "Local dataset at %s is missing or invalid. Proceeding with download.", output_dir
        )
    else:
        logger.debug("force_download=True: bypassing local validation for %s", output_dir)

    # 1. Validate and configure Kaggle API credentials
    username = os.getenv("KAGGLE_USERNAME")
    key = os.getenv("KAGGLE_KEY")

    if not username or not key:
        logger.error("Kaggle credentials missing. KAGGLE_USERNAME or KAGGLE_KEY not set.")
        raise ValueError(
            "Missing Kaggle API credentials. Set KAGGLE_USERNAME and KAGGLE_KEY in your environment or .env file."
        )

    # Set Kaggle API credentials if not already set
    os.environ["KAGGLE_API_TOKEN"] = key
    logger.info("Configured Kaggle API environment credentials for user: %s", username)

    # 2. Download competition files via KaggleHub
    try:
        logger.info("Fetching Kaggle competition dataset '%s' via KaggleHub...", competition_name)
        cache_path_str = kagglehub.competition_download(
            competition_name,
            force_download=force_download,
        )
        cache_path = Path(cache_path_str).expanduser().resolve()
    except Exception as exc:
        logger.error("KaggleHub competition download failed for '%s'.", competition_name)
        raise RuntimeError(f"Failed to download Kaggle competition '{competition_name}'.") from exc

    # 3. Efficient Processing directly from Cache into output_dir
    try:
        logger.debug("Processing items from cache %s to output %s", cache_path, output_dir)

        for item in cache_path.iterdir():
            if item.suffix.lower() == ".csv":
                if item.name == "sample_submission.csv":
                    logger.debug("Skipping conversion for %s (ignored file)", item.name)
                    continue

                # ONLY convert train.csv; preserve test.csv
                if item.name == "test.csv":
                    dest_item = output_dir / "test.csv"
                    shutil.copy2(item, dest_item)
                    logger.info("Preserved CSV file %s -> %s", item.name, output_dir)
                else:
                    parquet_dest = output_dir / f"{item.stem}.parquet"
                    _convert_csv_to_parquet_atomic(item, parquet_dest)

            elif item.is_file():
                dest_item = output_dir / item.name
                shutil.copy2(item, dest_item)
                logger.debug("Copied file %s -> %s", item.name, output_dir)
            elif item.is_dir():
                dest_item = output_dir / item.name
                if dest_item.exists():
                    logger.debug("Removing existing directory %s before copy", dest_item)
                    shutil.rmtree(dest_item)
                shutil.copytree(item, dest_item)

    except Exception as exc:
        logger.exception("KaggleHub competition download failed for '%s'.", competition_name)
        raise RuntimeError(
            f"Failed to download Kaggle competition '{competition_name}'. Check credentials or network connectivity."
        ) from exc

    # 4. Maintain Git structure
    gitkeep = output_dir / ".gitkeep"
    if not gitkeep.exists():
        try:
            gitkeep.touch(exist_ok=True)
            logger.debug("Created .gitkeep in %s", output_dir)
        except OSError as err:
            logger.warning("Could not create .gitkeep file at %s: %s", output_dir, err)

    # 5. Clean up unwanted artifacts in local output_dir (leaving cache pristine)
    _remove_unwanted_paths(output_dir, UNWANTED_DATA)
    logger.info("Dataset successfully prepared and cleaned at %s", output_dir)

churn_prediction.data.schemas

Dataset Validation Schemas.

This module defines Pandera schemas for validating the structure, data types, and value constraints of customer churn datasets. It ensures data quality by enforcing strict rules on nullability, uniqueness, and allowed categories across both raw and processed data stages.

churn_prediction.data.validation

Dataset Validation Module.

Implements the validation logic for churn prediction datasets using Pandera schemas. It provides a workflow to load tabular data, verify its integrity against structural constraints, and persist detailed diagnostic reports in JSON format.

validate_dataset(dataset_path, report_path, schema, *, lazy=True)

Validate a tabular dataset against a Pandera schema.

Loads the dataset according to its file format, validates its structure and values against the supplied Pandera schema, and writes a diagnostic JSON report describing the validation result.

Parameters:

Name	Type	Description	Default
dataset_path	str or Path	Path to the dataset being validated. Supported formats depend on the dataset loading implementation.	required
report_path	str or Path	Destination path for the generated JSON validation report.	required
schema	DataFrameSchema	Pandera schema defining the expected columns, data types, nullability, uniqueness constraints, and value restrictions.	required
lazy	bool	Whether Pandera should collect all validation failures before raising an exception. When True, the resulting report contains all detected validation failures.	True

Raises:

Type	Description
FileNotFoundError	If dataset_path does not exist.
ValueError	If the dataset cannot be loaded or violates the supplied schema.
OSError	If the validation report cannot be written to disk.

Source code in src/churn_prediction/data/validation.py

def validate_dataset(
    dataset_path: str | Path,
    report_path: str | Path,
    schema: pa.DataFrameSchema,
    *,
    lazy: bool = True,
) -> None:
    """Validate a tabular dataset against a Pandera schema.

    Loads the dataset according to its file format, validates its structure
    and values against the supplied Pandera schema, and writes a diagnostic
    JSON report describing the validation result.

    Parameters
    ----------
    dataset_path : str or pathlib.Path
        Path to the dataset being validated. Supported formats depend on the
        dataset loading implementation.
    report_path : str or pathlib.Path
        Destination path for the generated JSON validation report.
    schema : pandera.DataFrameSchema
        Pandera schema defining the expected columns, data types, nullability,
        uniqueness constraints, and value restrictions.
    lazy : bool, default=True
        Whether Pandera should collect all validation failures before raising
        an exception. When ``True``, the resulting report contains all
        detected validation failures.

    Raises
    ------
    FileNotFoundError
        If ``dataset_path`` does not exist.
    ValueError
        If the dataset cannot be loaded or violates the supplied schema.
    OSError
        If the validation report cannot be written to disk.
    """
    logger = get_logger()
    dataset_path = Path(dataset_path).expanduser().resolve()
    report_path = Path(report_path).expanduser().resolve()

    # 0. Validate input path existence
    if not dataset_path.exists():
        logger.error("Dataset path does not exist: %s", dataset_path)
        raise FileNotFoundError(f"Dataset file not found: {dataset_path}")

    if not dataset_path.is_file():
        logger.error("Dataset path is not a valid file: %s", dataset_path)
        raise ValueError(f"Target path is a directory or special file: {dataset_path}")

    # 1. Load the dataset into a Pandas DataFrame
    try:
        logger.debug("Loading dataset from %s into Pandas DataFrame", dataset_path)
        df = pd.read_parquet(dataset_path)
    except Exception as err:
        logger.exception("Failed to load dataset file %s: %s", dataset_path, err)
        raise ValueError(f"Failed to load dataset from {dataset_path}: {err}") from err

    logger.info(
        "Successfully loaded dataset from %s (rows=%d, cols=%d, memory=%.2f MB)",
        dataset_path,
        len(df),
        len(df.columns),
        df.memory_usage(deep=True).sum() / (1024 * 1024),
    )

    # 2. Prepare Diagnostic Structure
    report: dict[str, Any] = {
        "dataset_path": str(dataset_path),
        "schema_name": schema.name,
        "total_rows": len(df),
        "total_columns": len(df.columns),
        "columns_present": list(df.columns),
        "status": "PASSED",
        "failure_count": 0,
        "failures": [],
    }

    # 3. Perform Validation
    try:
        logger.debug("Executing schema validation (lazy=%s) for %s", lazy, schema.name)
        schema.validate(df, lazy=lazy)
        logger.info(
            "Dataframe validation PASSED for %s against schema context: %s",
            dataset_path.name,
            schema.name,
        )
    except (pa_errors.SchemaError, pa_errors.SchemaErrors) as err:
        report["status"] = "FAILED"
        failures = _extract_failures(err)
        report["failures"] = failures
        report["failure_count"] = len(failures)

        # Extract specific failing columns for high-level logging summary
        failed_columns: set[str] = {
            str(f["column"]) for f in failures if f.get("column") is not None
        }

        logger.error(
            "Dataframe validation FAILED for %s with %d violation(s) across %d column(s) %s",
            dataset_path.name,
            report["failure_count"],
            len(failed_columns),
            sorted(failed_columns) if failed_columns else [],
        )

        # 4. Persist Diagnostic Report
        _write_report(report_path, report)

    # 5. Raise Exception if Validation Failed
    if report["status"] == "FAILED":
        raise ValueError(
            f"Dataset validation failed with {report['failure_count']} violation(s). "
            f"Detailed diagnostic report persisted to: {report_path}"
        )