function.py

# ==============================================================================
# 1. IMPORT NECESSARY LIBRARIES
# ==============================================================================
import pandas as pd
import scipy
import lime
import lime.lime_tabular
import numpy as np
from datetime import date

# ==============================================================================
# 2. GLOBAL CONFIGURATION
# ==============================================================================
RANDOM_SEED = 42
TARGET_LABELS = ["Not Returned (0)", "Returned (1)"]

# ==============================================================================
# 3. DATA PREPARATION FUNCTIONS (FEATURE ENGINEERING)
# ==============================================================================
def prepare_data_pre(
    customer_age: int, 
    product_category: str, 
    payment_method: str, 
    order_value_usd: float, 
    order_date: date
) -> pd.DataFrame:
    """
    Transforms raw pre-delivery input parameters into a structured Pandas DataFrame.
    Extracts temporal features from the order date.
    """
    print("⏳ [FEATURE ENG] Structuring raw Pre-Delivery inputs...")
    
    # Wrap scalars in lists to correctly construct a single-row DataFrame
    df_testing = pd.DataFrame({
        "customer_age": [customer_age],
        "product_category": [product_category],
        "payment_method": [payment_method],
        "order_value_usd": [order_value_usd],
        "order_date": [order_date]
    })
    
    print("📅 [FEATURE ENG] Extracting temporal features from 'order_date'...")
    df_testing["order_date"] = pd.to_datetime(df_testing["order_date"])
    df_testing["day"] = df_testing["order_date"].dt.day
    df_testing["month"] = df_testing["order_date"].dt.month
    df_testing["week"] = df_testing["order_date"].dt.isocalendar().week.astype(int)
    
    # Boolean flag: True if Monday-Friday, False if Saturday/Sunday
    df_testing["working_day"] = df_testing["order_date"].dt.day_of_week < 5
    
    # Drop the original datetime column to prevent model errors
    df_testing.drop("order_date", axis=1, inplace=True)

    print(f"✅ [FEATURE ENG] Pre-Delivery DataFrame ready. Shape: {df_testing.shape}")
    return df_testing

def prepare_data_post(
    customer_age: int, 
    product_category: str, 
    payment_method: str, 
    order_value_usd: float, 
    delivery_time_days: int, 
    customer_rating: float, 
    order_date: date
) -> pd.DataFrame:
    """
    Transforms raw post-delivery input parameters into a structured Pandas DataFrame.
    Includes additional post-delivery metrics (delivery time and rating).
    """
    print("⏳ [FEATURE ENG] Structuring raw Post-Delivery inputs...")
    
    df_testing = pd.DataFrame({
        "customer_age": [customer_age],
        "product_category": [product_category],
        "payment_method": [payment_method],
        "order_value_usd": [order_value_usd],
        "delivery_time_days": [delivery_time_days],
        "customer_rating": [customer_rating],
        "order_date": [order_date]
    })
    
    print("📅 [FEATURE ENG] Extracting temporal features from 'order_date'...")
    df_testing["order_date"] = pd.to_datetime(df_testing["order_date"])
    df_testing["day"] = df_testing["order_date"].dt.day
    df_testing["month"] = df_testing["order_date"].dt.month
    df_testing["week"] = df_testing["order_date"].dt.isocalendar().week.astype(int)
    df_testing["working_day"] = df_testing["order_date"].dt.day_of_week < 5
    
    df_testing.drop("order_date", axis=1, inplace=True)

    print(f"✅ [FEATURE ENG] Post-Delivery DataFrame ready. Shape: {df_testing.shape}")
    return df_testing

# ==============================================================================
# 4. INFERENCE FUNCTION (PREDICTION)
# ==============================================================================
def predict(best_model, df_testing: pd.DataFrame, target_labels: list = TARGET_LABELS) -> tuple:
    """
    Executes model inference on the prepared DataFrame.
    Extracts the predicted class, return probability, and overall confidence for a single instance.
    """
    print("🧠 [INFERENCE] Executing model prediction...")
    
    # Generate predictions and probabilities using the loaded model
    y_pred = best_model.predict(df_testing)
    y_pred_proba = best_model.predict_proba(df_testing)

    # Map the numeric prediction to the corresponding string label
    result = [target_labels[pred] for pred in y_pred]

    # Extract the specific values for the single transaction (index 0)
    prediction = result[0]
    returned_proba = ((y_pred_proba[:, 1] * 100).round(2).astype(str) + '%')[0]
    prediction_conf = ((y_pred_proba.max(axis=1) * 100).round(2).astype(str) + '%')[0]

    # Log the extracted metrics to the terminal for observability
    print(f"🎯 [INFERENCE] Result: {prediction} | Prob: {returned_proba} | Conf: {prediction_conf}")
    
    return prediction, returned_proba, prediction_conf

# ==============================================================================
# 5. EXPLAINABILITY FUNCTION (LIME)
# ==============================================================================
def explain(best_model, X_train_processed: np.ndarray, df_testing: pd.DataFrame, target_label: list = TARGET_LABELS) -> str:
    """
    Generates a LIME (Local Interpretable Model-agnostic Explanations) HTML output.
    Bypasses background data transformation as X_train_processed is already encoded.
    """
    print("🔍 [XAI] Initializing LIME Explainer...")
    
    # Extract the preprocessing step and the machine learning model from the pipeline
    try:
        preprocessor = best_model.named_steps["Preprocessor"]
        ml_model = best_model.named_steps["Model"]
    except KeyError as e:
        raise KeyError(f"🚨 [PIPELINE ERROR] Expected steps 'Preprocessor' and 'Model' not found in pipeline. Details: {e}")

    print("🧮 [XAI] Loading pre-processed background training data...")
    
    # Ensure the background data is a dense array for LIME compatibility
    if scipy.sparse.issparse(X_train_processed):
        X_train_processed = X_train_processed.toarray()

    # Extract feature names directly from the preprocessor to label the LIME output
    features = preprocessor.get_feature_names_out()

    # Initialize the Tabular Explainer with the pre-processed background data
    explainer = lime.lime_tabular.LimeTabularExplainer(
        training_data=X_train_processed,
        feature_names=features,
        class_names=target_label,
        mode="classification",
        random_state=RANDOM_SEED
    )

    print("📊 [XAI] Processing single transaction instance for explanation...")
    
    # Isolate the single row to explain and apply the preprocessing pipeline
    transaction_raw = df_testing.iloc[[0]]
    processed_data = preprocessor.transform(transaction_raw)

    # Convert the processed single instance to a dense array if necessary
    if scipy.sparse.issparse(processed_data):
        processed_data = processed_data.toarray()

    # Flatten to a 1D array as required by LIME's explain_instance method
    transaction_data = processed_data[0]

    print("🧩 [XAI] Generating instance explanation (Top 10 features)...")
    
    # Generate the explanation using the isolated ML model's predict_proba method
    explanation = explainer.explain_instance(
        data_row=transaction_data,
        predict_fn=ml_model.predict_proba,
        num_features=10
    )

    print("✅ [XAI] LIME HTML generation complete.")
    return explanation.as_html()