Client¶

`churn_prediction.client.api_client` ¶

HTTP wrappers around the churn-prediction FastAPI endpoints.

This module provides functions to interact with the churn-prediction API endpoints for batch prediction, downloading predictions, and explaining predictions. It uses the requests library to send HTTP requests and handle responses.

`check_health()` ¶

Check the availability of the churn prediction API.

Sends a request to the configured health endpoint and returns the service health information.

Returns:

Type	Description
`dict[str, str]`	JSON health status returned by the API.

Raises:

Type	Description
`HTTPError`	If the API returns an unsuccessful HTTP status.
`RequestException`	If the health-check request fails.

Source code in src/churn_prediction/client/api_client.py

def check_health() -> dict[str, str]:
    """Check the availability of the churn prediction API.

    Sends a request to the configured health endpoint and returns the
    service health information.

    Returns
    -------
    dict[str, str]
        JSON health status returned by the API.

    Raises
    ------
    requests.HTTPError
        If the API returns an unsuccessful HTTP status.
    requests.RequestException
        If the health-check request fails.
    """
    settings = get_settings()

    response = requests.get(settings.HEALTH_API_URL, timeout=10)
    response.raise_for_status()
    return response.json()

`download_predictions(batch_id, save_path)` ¶

Download prediction results for a completed batch.

Requests the prediction artifact associated with batch_id from the API and writes the returned file contents to save_path.

Parameters:

Name	Type	Description	Default
`batch_id`	`str`	Unique identifier of the prediction batch.	required
`save_path`	`str`	Local destination path for the downloaded prediction file.	required

Returns:

Type	Description
`str`	Path to the saved prediction file.

Raises:

Type	Description
`HTTPError`	If the API returns an unsuccessful HTTP status.
`RequestException`	If the download request fails.
`OSError`	If the prediction file cannot be written to disk.

Source code in src/churn_prediction/client/api_client.py

def download_predictions(batch_id: str, save_path: str) -> str:
    """Download prediction results for a completed batch.

    Requests the prediction artifact associated with ``batch_id`` from the
    API and writes the returned file contents to ``save_path``.

    Parameters
    ----------
    batch_id : str
        Unique identifier of the prediction batch.
    save_path : str
        Local destination path for the downloaded prediction file.

    Returns
    -------
    str
        Path to the saved prediction file.

    Raises
    ------
    requests.HTTPError
        If the API returns an unsuccessful HTTP status.
    requests.RequestException
        If the download request fails.
    OSError
        If the prediction file cannot be written to disk.
    """
    settings = get_settings()

    response = requests.get(f"{settings.DOWNLOAD_API_URL}/{batch_id}", timeout=30)
    response.raise_for_status()
    with Path(save_path).open("wb") as f:
        f.write(response.content)
    return save_path

`explain_batch(batch_id)` ¶

Request explainability artifacts for a prediction batch.

Sends an explainability request for the specified batch and returns the API response containing information about the generated SHAP artifacts.

Parameters:

Name	Type	Description	Default
`batch_id`	`str`	Unique identifier of the prediction batch to explain.	required

Returns:

Type	Description
`dict[str, str]`	JSON response returned by the explainability endpoint.

Raises:

Type	Description
`HTTPError`	If the API returns an unsuccessful HTTP status.
`RequestException`	If the HTTP request fails.

Source code in src/churn_prediction/client/api_client.py

def explain_batch(batch_id: str) -> dict[str, str]:
    """Request explainability artifacts for a prediction batch.

    Sends an explainability request for the specified batch and returns the
    API response containing information about the generated SHAP artifacts.

    Parameters
    ----------
    batch_id : str
        Unique identifier of the prediction batch to explain.

    Returns
    -------
    dict[str, str]
        JSON response returned by the explainability endpoint.

    Raises
    ------
    requests.HTTPError
        If the API returns an unsuccessful HTTP status.
    requests.RequestException
        If the HTTP request fails.
    """
    settings = get_settings()

    response = requests.post(f"{settings.EXPLAIN_API_URL}/{batch_id}", timeout=120)
    response.raise_for_status()
    return response.json()

`predict_batch(csv_path)` ¶

Submit a CSV dataset to the batch prediction API.

Opens the supplied CSV file and sends it as a multipart upload to the configured prediction endpoint. The API response is returned after validating the HTTP status.

Parameters:

Name	Type	Description	Default
`csv_path`	`str or Path`	Path to the CSV file containing customer records to score.	required

Returns:

Type	Description
`dict[str, str]`	JSON response returned by the prediction API.

Raises:

Type	Description
`FileNotFoundError`	If the CSV file does not exist.
`HTTPError`	If the API returns an unsuccessful HTTP status.
`RequestException`	If the HTTP request fails.

Source code in src/churn_prediction/client/api_client.py

def predict_batch(csv_path: str | Path) -> dict[str, str]:
    """Submit a CSV dataset to the batch prediction API.

    Opens the supplied CSV file and sends it as a multipart upload to the
    configured prediction endpoint. The API response is returned after
    validating the HTTP status.

    Parameters
    ----------
    csv_path : str or pathlib.Path
        Path to the CSV file containing customer records to score.

    Returns
    -------
    dict[str, str]
        JSON response returned by the prediction API.

    Raises
    ------
    FileNotFoundError
        If the CSV file does not exist.
    requests.HTTPError
        If the API returns an unsuccessful HTTP status.
    requests.RequestException
        If the HTTP request fails.
    """
    settings = get_settings()
    csv_path = Path(csv_path).expanduser().resolve()

    with csv_path.open("rb") as f:
        files = {"file": (csv_path.name, f, "text/csv")}
        response = requests.post(settings.PREDICT_API_URL, files=files, timeout=120)
    response.raise_for_status()
    return response.json()

`churn_prediction.client.dashboard` ¶

Churn Prediction Interpretation Module.

This module provides utilities for interpreting machine learning model outputs related to customer churn. It includes functionality to load model artifacts, decode ordinal features back to their original categories, map feature importance to human-readable recommendations, and extract SHAP-based risk breakdowns for individual customers.

`customer_shap_breakdown(risk_profiles, customer_id)` ¶

Extract the SHAP contribution breakdown for one customer.

Combines the customer's strongest churn drivers and retention factors, decodes categorical feature values, and returns the contributions sorted by SHAP value.

Parameters:

Name	Type	Description	Default
`risk_profiles`	`DataFrame`	Customer risk profile table produced by the explainability pipeline.	required
`customer_id`	`int`	Customer identifier for which the SHAP breakdown should be retrieved.	required

Returns:

Type	Description
`DataFrame`	SHAP contribution table containing `feature`, `shap_value`, `feature_value`, and `display_value` columns. An empty DataFrame with the expected columns is returned when the customer does not exist or has no recorded drivers.

Source code in src/churn_prediction/client/dashboard.py

def customer_shap_breakdown(
    risk_profiles: pd.DataFrame,
    customer_id: int,
) -> pd.DataFrame:
    """Extract the SHAP contribution breakdown for one customer.

    Combines the customer's strongest churn drivers and retention factors,
    decodes categorical feature values, and returns the contributions sorted
    by SHAP value.

    Parameters
    ----------
    risk_profiles : pandas.DataFrame
        Customer risk profile table produced by the explainability pipeline.
    customer_id : int
        Customer identifier for which the SHAP breakdown should be retrieved.

    Returns
    -------
    pandas.DataFrame
        SHAP contribution table containing ``feature``, ``shap_value``,
        ``feature_value``, and ``display_value`` columns. An empty DataFrame
        with the expected columns is returned when the customer does not
        exist or has no recorded drivers.
    """
    row: pd.DataFrame = risk_profiles[risk_profiles["customer_id"] == customer_id]
    if row.empty:
        return pd.DataFrame(
            {
                "feature": [],
                "shap_value": [],
                "feature_value": [],
                "display_value": [],
            }
        )

    drivers: list[dict[str, Any]] = (
        row.iloc[0]["top_churn_drivers"] + row.iloc[0]["top_retention_factors"]
    )
    if not drivers:
        return pd.DataFrame(
            {
                "feature": [],
                "shap_value": [],
                "feature_value": [],
                "display_value": [],
            }
        )

    breakdown = pd.DataFrame(drivers)
    breakdown["display_value"] = breakdown.apply(
        lambda r: decode_features(r["feature"], r["feature_value"]), axis=1
    )
    return breakdown.sort_values("shap_value").reset_index(drop=True)

`decode_features(feature, value)` ¶

Decode an ordinal feature value into its original category.

Uses the fitted training encoder to translate a numeric ordinal value back into its original categorical representation. Features that are not ordinally encoded are returned unchanged.

Parameters:

Name	Type	Description	Default
`feature`	`str`	Name of the feature containing the encoded value.	required
`value`	`float`	Numeric ordinal value produced by the preprocessing pipeline.	required

Returns:

Type	Description
`str`	Original categorical value when the feature is ordinal and the value is valid. Returns the string representation of `value` for non-ordinal features and `"Unknown"` for invalid ordinal values.

Source code in src/churn_prediction/client/dashboard.py

def decode_features(feature: str, value: float) -> str:
    """Decode an ordinal feature value into its original category.

    Uses the fitted training encoder to translate a numeric ordinal value
    back into its original categorical representation. Features that are not
    ordinally encoded are returned unchanged.

    Parameters
    ----------
    feature : str
        Name of the feature containing the encoded value.
    value : float
        Numeric ordinal value produced by the preprocessing pipeline.

    Returns
    -------
    str
        Original categorical value when the feature is ordinal and the value
        is valid. Returns the string representation of ``value`` for
        non-ordinal features and ``"Unknown"`` for invalid ordinal values.
    """
    encoder, ordinal_columns = _load_encoder()
    if feature not in ordinal_columns:
        return str(value)

    categories = encoder.categories_[ordinal_columns.index(feature)]
    if 0 <= value < len(categories):
        return str(categories[int(value)])
    return "Unknown"

`load_artifacts(artifacts_dir)` ¶

Load dashboard artifacts generated by the explainability pipeline.

Reads global feature-importance metadata and customer-level SHAP risk profiles from the supplied artifact directory and converts them into dashboard-ready Python and pandas objects.

Parameters:

Name	Type	Description	Default
`artifacts_dir`	`str or Path`	Directory containing `metadata.json` and `customer_risk_profiles.parquet`.	required

Returns:

Type	Description
`dict`	Dictionary containing: `metadata` : dict Global explainability metadata. `feature_importance` : pandas.DataFrame Ranked global feature importance values. `risk_profiles` : pandas.DataFrame Customer-level churn probabilities and SHAP drivers. `id_col` : str Name of the customer identifier column. `prob_col` : str Name of the calculated churn probability column.

Raises:

Type	Description
`FileNotFoundError`	If the artifact directory or required artifact files do not exist.
`NotADirectoryError`	If `artifacts_dir` is not a directory.

Source code in src/churn_prediction/client/dashboard.py

def load_artifacts(artifacts_dir: str | Path) -> dict[str, Any]:
    """Load dashboard artifacts generated by the explainability pipeline.

    Reads global feature-importance metadata and customer-level SHAP risk
    profiles from the supplied artifact directory and converts them into
    dashboard-ready Python and pandas objects.

    Parameters
    ----------
    artifacts_dir : str or pathlib.Path
        Directory containing ``metadata.json`` and
        ``customer_risk_profiles.parquet``.

    Returns
    -------
    dict
        Dictionary containing:

        - ``metadata`` : dict
            Global explainability metadata.
        - ``feature_importance`` : pandas.DataFrame
            Ranked global feature importance values.
        - ``risk_profiles`` : pandas.DataFrame
            Customer-level churn probabilities and SHAP drivers.
        - ``id_col`` : str
            Name of the customer identifier column.
        - ``prob_col`` : str
            Name of the calculated churn probability column.

    Raises
    ------
    FileNotFoundError
        If the artifact directory or required artifact files do not exist.
    NotADirectoryError
        If ``artifacts_dir`` is not a directory.
    """
    base = Path(artifacts_dir)
    if not base.exists():
        raise FileNotFoundError(f"Artifacts directory not found: {base}")
    if not base.is_dir():
        raise NotADirectoryError(f"Artifacts directory is not a directory: {base}")

    # Load metadata from the JSON file
    metadata: dict[str, Any] = json.loads((base / "metadata.json").read_text())
    feature_importance: pd.DataFrame = pd.DataFrame(metadata["global_feature_importance"]).head(10)[
        ["feature", "mean_abs_shap"]
    ]

    # Load risk profiles from the Parquet file
    risk_profiles = pd.read_parquet(base / "customer_risk_profiles.parquet")
    risk_profiles.index.name = "customer_id"
    risk_profiles = risk_profiles.reset_index()
    risk_profiles["churn_probability"] = _sigmoid(risk_profiles["predicted_margin"].to_numpy())
    risk_profiles["top_churn_drivers"] = risk_profiles["top_churn_drivers"].apply(json.loads)
    risk_profiles["top_retention_factors"] = risk_profiles["top_retention_factors"].apply(
        json.loads
    )

    return {
        "metadata": metadata,
        "feature_importance": feature_importance,
        "risk_profiles": risk_profiles,
        "id_col": "customer_id",
        "prob_col": "churn_probability",
    }

`recommend_for_feature(feature_name)` ¶

Return a retention recommendation for a churn-risk feature.

Normalizes the supplied feature name and maps it to the corresponding business recommendation. Unknown features receive a generic customer health recommendation.

Parameters:

Name	Type	Description	Default
`feature_name`	`str`	Feature name associated with the identified churn risk factor.	required

Returns:

Type	Description
`str`	Business recommendation corresponding to the feature, or a generic account-health recommendation when no feature-specific recommendation exists.

Source code in src/churn_prediction/client/dashboard.py

def recommend_for_feature(feature_name: str) -> str:
    """Return a retention recommendation for a churn-risk feature.

    Normalizes the supplied feature name and maps it to the corresponding
    business recommendation. Unknown features receive a generic customer
    health recommendation.

    Parameters
    ----------
    feature_name : str
        Feature name associated with the identified churn risk factor.

    Returns
    -------
    str
        Business recommendation corresponding to the feature, or a generic
        account-health recommendation when no feature-specific recommendation
        exists.
    """
    key = feature_name.strip().lower().replace("_", "").replace(" ", "")
    normalized_map = {k.replace("_", ""): k for k in RECOMMENDATIONS}
    matched_key = normalized_map.get(key)
    if matched_key:
        return RECOMMENDATIONS[matched_key]
    return "Review overall account engagement and conduct a proactive customer health check-in."

Client¶

churn_prediction.client.api_client ¶

check_health() ¶

download_predictions(batch_id, save_path) ¶

explain_batch(batch_id) ¶

predict_batch(csv_path) ¶

churn_prediction.client.dashboard ¶

customer_shap_breakdown(risk_profiles, customer_id) ¶

decode_features(feature, value) ¶

load_artifacts(artifacts_dir) ¶

recommend_for_feature(feature_name) ¶

`churn_prediction.client.api_client` ¶

`check_health()` ¶

`download_predictions(batch_id, save_path)` ¶

`explain_batch(batch_id)` ¶

`predict_batch(csv_path)` ¶

`churn_prediction.client.dashboard` ¶

`customer_shap_breakdown(risk_profiles, customer_id)` ¶

`decode_features(feature, value)` ¶

`load_artifacts(artifacts_dir)` ¶

`recommend_for_feature(feature_name)` ¶