Skip to content

Pipelines

churn_prediction.pipelines.training_pipeline

Training Pipeline for Churn Prediction.

This module provides a CLI interface to orchestrate the training pipeline, including ingestion, schema validation, preprocessing of training and validation datasets and training the xgboost model. It is designed to be integrated with DVC (Data Version Control) for pipeline reproducibility.

run_all_stages(settings)

Execute the complete training pipeline.

Runs ingestion, raw-data validation, preprocessing, processed-data validation, and model training sequentially using the supplied application settings.

Source code in src/churn_prediction/pipelines/training_pipeline.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
def run_all_stages(settings: Settings) -> None:
    """Execute the complete training pipeline.

    Runs ingestion, raw-data validation, preprocessing, processed-data
    validation, and model training sequentially using the supplied
    application settings.
    """
    run_ingestion(settings)
    run_raw_validation(settings)
    run_preprocessing(settings)
    run_train_validation(settings)
    run_val_validation(settings)
    run_training(settings)

run_ingestion(settings)

Execute the dataset ingestion stage.

Downloads or retrieves the configured raw dataset and prepares the expected local dataset artifacts.

Source code in src/churn_prediction/pipelines/training_pipeline.py 
21
22
23
24
25
26
27
28
29
30
31
def run_ingestion(settings: Settings) -> None:
    """Execute the dataset ingestion stage.

    Downloads or retrieves the configured raw dataset and prepares the
    expected local dataset artifacts.
    """
    # Resolve Base Directories
    raw_data_dir = settings.RAW_DATA_DIR.expanduser().resolve()

    # Execute Ingestion
    get_dataset(raw_data_dir)

run_preprocessing(settings)

Execute the training preprocessing stage.

Transforms the raw dataset into model-ready training and validation datasets and persists the fitted preprocessing artifact.

Source code in src/churn_prediction/pipelines/training_pipeline.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def run_preprocessing(settings: Settings) -> None:
    """Execute the training preprocessing stage.

    Transforms the raw dataset into model-ready training and validation
    datasets and persists the fitted preprocessing artifact.
    """
    # Resolve Base Directories
    raw_dir = settings.RAW_DATA_DIR.expanduser().resolve()
    processed_dir = settings.PROCESSED_DATA_DIR.expanduser().resolve()
    model_dir = settings.MODEL_DIR.expanduser().resolve()

    # Derive Specific File Paths
    raw_file = raw_dir / "train.parquet"
    train_file = processed_dir / "processed_train.parquet"
    val_file = processed_dir / "processed_val.parquet"
    preprocessor_file = model_dir / "preprocessor.joblib"

    # Execute Preprocessing
    preprocess_train_dataset(
        raw_input_path=raw_file,
        train_output_path=train_file,
        val_output_path=val_file,
        preprocessor_path=preprocessor_file,
    )

run_raw_validation(settings)

Execute raw dataset validation.

Validates the ingested raw dataset against RAW_SCHEMA and writes validation diagnostics to the configured report location.

Source code in src/churn_prediction/pipelines/training_pipeline.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def run_raw_validation(settings: Settings) -> None:
    """Execute raw dataset validation.

    Validates the ingested raw dataset against ``RAW_SCHEMA`` and writes
    validation diagnostics to the configured report location.
    """
    # Resolve Base Directories
    raw_data_dir = settings.RAW_DATA_DIR.expanduser().resolve()
    validation_report_dir = settings.VALIDATION_REPORT_DIR.expanduser().resolve()

    # Derive Specific File Paths
    raw_data_file = raw_data_dir / "train.parquet"
    raw_validation_report = validation_report_dir / "raw.json"

    # Execute Validation
    validate_dataset(
        dataset_path=raw_data_file,
        report_path=raw_validation_report,
        schema=RAW_SCHEMA,
    )

run_train_validation(settings)

Validate the processed training dataset.

Checks the generated training dataset against PROCESSED_SCHEMA and records any validation failures in the configured validation report.

Source code in src/churn_prediction/pipelines/training_pipeline.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def run_train_validation(settings: Settings) -> None:
    """Validate the processed training dataset.

    Checks the generated training dataset against ``PROCESSED_SCHEMA`` and
    records any validation failures in the configured validation report.
    """
    # Resolve Base Directories
    processed_dir = settings.PROCESSED_DATA_DIR.expanduser().resolve()
    validation_report_dir = settings.VALIDATION_REPORT_DIR.expanduser().resolve()

    # Derive Specific File Paths
    train_file = processed_dir / "processed_train.parquet"
    train_validation_report = validation_report_dir / "train.json"

    # Execute Validation
    validate_dataset(
        dataset_path=train_file,
        report_path=train_validation_report,
        schema=PROCESSED_SCHEMA,
    )

run_training(settings)

Execute the model training stage.

Trains the configured XGBoost model using the processed training and validation datasets and exports the resulting model, model card, and training report.

Source code in src/churn_prediction/pipelines/training_pipeline.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def run_training(settings: Settings) -> None:
    """Execute the model training stage.

    Trains the configured XGBoost model using the processed training and
    validation datasets and exports the resulting model, model card, and
    training report.
    """
    # Resolve Base Directories
    processed_dir = settings.PROCESSED_DATA_DIR.expanduser().resolve()
    model_dir = settings.MODEL_DIR.expanduser().resolve()
    report_dir = settings.TRAINING_REPORT_DIR.expanduser().resolve()

    # Derive Specific File Paths
    train_file = processed_dir / "processed_train.parquet"
    val_file = processed_dir / "processed_val.parquet"
    model_file = model_dir / "model.ubj"
    model_card_file = model_dir / "card.json"
    report_file = report_dir / "training.json"

    # Execute Training
    train_xgb_model(
        train_data_path=train_file,
        val_data_path=val_file,
        model_filepath=model_file,
        model_card_path=model_card_file,
        report_path=report_file,
    )

run_val_validation(settings)

Validate the processed validation dataset.

Checks the generated validation dataset against PROCESSED_SCHEMA and records any validation failures in the configured validation report.

Source code in src/churn_prediction/pipelines/training_pipeline.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def run_val_validation(settings: Settings) -> None:
    """Validate the processed validation dataset.

    Checks the generated validation dataset against ``PROCESSED_SCHEMA``
    and records any validation failures in the configured validation report.
    """
    # Resolve Base Directories
    processed_dir = settings.PROCESSED_DATA_DIR.expanduser().resolve()
    validation_report_dir = settings.VALIDATION_REPORT_DIR.expanduser().resolve()

    # Derive Specific File Paths
    val_file = processed_dir / "processed_val.parquet"
    val_validation_report = validation_report_dir / "val.json"

    # Execute Validation
    validate_dataset(
        dataset_path=val_file,
        report_path=val_validation_report,
        schema=PROCESSED_SCHEMA,
    )

churn_prediction.pipelines.prediction_pipeline

Prediction Pipeline for Churn Prediction.

This module defines the end-to-end workflow for generating predictions on new data, integrating dataset validation, feature preprocessing, and model inference.

run_prediction_pipeline(batch_id, raw_file_path, loaded_encoder, loaded_booster)

Execute the complete batch prediction workflow.

The pipeline validates the raw input dataset, transforms it using the pre-fitted encoder, validates the resulting model-ready dataset, and generates predictions using the trained XGBoost booster.

Parameters:

Name Type Description Default
batch_id str

Unique identifier for the current prediction batch.

 required
raw_file_path str or Path

Path to the raw customer dataset to be scored.

 required
loaded_encoder OrdinalEncoder

Pre-fitted encoder used to transform categorical features.

 required
loaded_booster Booster

Trained XGBoost model used to generate churn predictions.

 required

Raises:

Type Description
FileNotFoundError

If the input dataset or required output directories are unavailable.
ValueError

If raw or processed data fails validation or preprocessing fails.
RuntimeError

If model inference or artifact persistence fails.
Source code in src/churn_prediction/pipelines/prediction_pipeline.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def run_prediction_pipeline(
    batch_id: str,
    raw_file_path: str | Path,
    loaded_encoder: OrdinalEncoder,
    loaded_booster: Booster,
) -> None:
    """Execute the complete batch prediction workflow.

    The pipeline validates the raw input dataset, transforms it using the
    pre-fitted encoder, validates the resulting model-ready dataset, and
    generates predictions using the trained XGBoost booster.

    Parameters
    ----------
    batch_id : str
        Unique identifier for the current prediction batch.
    raw_file_path : str or pathlib.Path
        Path to the raw customer dataset to be scored.
    loaded_encoder : sklearn.preprocessing.OrdinalEncoder
        Pre-fitted encoder used to transform categorical features.
    loaded_booster : xgboost.Booster
        Trained XGBoost model used to generate churn predictions.

    Raises
    ------
    FileNotFoundError
        If the input dataset or required output directories are unavailable.
    ValueError
        If raw or processed data fails validation or preprocessing fails.
    RuntimeError
        If model inference or artifact persistence fails.
    """
    # Initialize settings
    settings = get_settings()
    batch_id = batch_id.strip()

    # Resolve base directories
    processed_dir = settings.PROCESSED_DATA_DIR.expanduser().resolve()
    prediction_dir = settings.PREDICTION_DATA_DIR.expanduser().resolve()
    model_dir = settings.MODEL_DIR.expanduser().resolve()
    validation_report_dir = settings.VALIDATION_REPORT_DIR.expanduser().resolve()

    # Derive specific file paths
    raw_file_path = Path(raw_file_path).expanduser().resolve()
    processed_file_path = processed_dir / f"processed_{batch_id}.parquet"
    prediction_file_path = prediction_dir / f"prediction_{batch_id}.csv"
    model_card_path = model_dir / "card.json"
    raw_report_path = validation_report_dir / f"raw_{batch_id}.json"
    processed_report_path = validation_report_dir / f"processed_{batch_id}.json"

    # 1. Validate the raw dataset
    validate_dataset(
        dataset_path=raw_file_path,
        report_path=raw_report_path,
        schema=RAW_BASE_SCHEMA,
    )

    # 2. Preprocess the raw dataset for inference
    preprocess_inference_dataset(
        raw_input_path=raw_file_path,
        processed_output_path=processed_file_path,
        fitted_encoder=loaded_encoder,
    )

    # 3. Validate the processed dataset
    validate_dataset(
        dataset_path=processed_file_path,
        report_path=processed_report_path,
        schema=PROCESSED_BASE_SCHEMA,
    )

    # 4. Run inference on the processed dataset
    run_inference(
        booster=loaded_booster,
        raw_filepath=raw_file_path,
        input_filepath=processed_file_path,
        output_filepath=prediction_file_path,
        model_card_path=model_card_path,
    )