# ============================================================================== # 1. IMPORT NECESSARY LIBRARIES # ============================================================================== import joblib import numpy as np from datetime import date from typing import Annotated, Literal from fastapi import FastAPI, HTTPException, Form from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field # Import custom processing functions from the local 'function.py' module from function import prepare_data_pre, prepare_data_post, predict, explain # ============================================================================== # 2. FASTAPI APPLICATION INITIALIZATION # ============================================================================== app = FastAPI( title="Product Return Prediction API", version="1.0.1", description="API for predicting e-commerce product returns before and after delivery." ) # Configure CORS middleware to allow cross-origin requests app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"] ) # ============================================================================== # 3. PYDANTIC DATA MODELS (INPUT VALIDATION) # ============================================================================== class ConditionalInputPre(BaseModel): """Schema for pre-delivery return prediction input.""" customer_age: int = Field(default=20, ge=18, le=69, description="Customer age between 18 and 69") product_category: Literal['Groceries', 'Home & Living', 'Sports', 'Electronics', 'Beauty', 'Fashion'] = "Electronics" payment_method: Literal['Wallet', 'UPI', 'Credit Card', 'Debit Card', 'Cash on Delivery'] = "Wallet" order_value_usd: float = Field(default=20.00, ge=10.01, le=718.73, description="Order value in USD") order_date: date class ConditionalInputPost(BaseModel): """Schema for post-delivery return prediction input.""" customer_age: int = Field(default=20, ge=18, le=69, description="Customer age between 18 and 69") product_category: Literal['Groceries', 'Home & Living', 'Sports', 'Electronics', 'Beauty', 'Fashion'] = "Electronics" payment_method: Literal['Wallet', 'UPI', 'Credit Card', 'Debit Card', 'Cash on Delivery'] = "Wallet" order_value_usd: float = Field(default=20.00, ge=10.01, le=718.73, description="Order value in USD") delivery_time_days: int = Field(default=3, ge=1, le=14, description="Days taken for delivery") customer_rating: float = Field(default=4.0, ge=1.0, le=5.0, description="Customer rating from 1.0 to 5.0") order_date: date # ============================================================================== # 4. GLOBAL MODEL AND EXPLANATION DATA VARIABLES # ============================================================================== # Pre-delivery model state best_model_pre = None lime_training_data_pre = None # Post-delivery model state best_model_post = None lime_training_data_post = None # ============================================================================== # 5. MODEL LOADING FUNCTION # ============================================================================== def load_models() -> bool: """ Loads pre-trained machine learning models and LIME training data into global memory. Ensures models are only loaded if they haven't been initialized yet. """ global best_model_pre, lime_training_data_pre, best_model_post, lime_training_data_post all_requirements = [best_model_pre, lime_training_data_pre, best_model_post, lime_training_data_post] try: # Check if any model or data is missing from memory if any(req is None for req in all_requirements): print("šŸ”„ [SYSTEM] Loading models and training data into memory...") # --------------------------------------------------------- # PRE-DELIVERY MODELS & DATA # --------------------------------------------------------- if best_model_pre is None: print("šŸ“¦ [LOAD] Loading Pre-Delivery Model...") best_model_pre = joblib.load("product-return-model/pre/best_model.joblib") if lime_training_data_pre is None: print("šŸ“Š [LOAD] Loading Pre-Delivery LIME Data...") lime_training_data_pre = np.load("product-return-model/pre/lime_training_data.npy") # --------------------------------------------------------- # POST-DELIVERY MODELS & DATA # --------------------------------------------------------- if best_model_post is None: print("šŸ“¦ [LOAD] Loading Post-Delivery Model...") best_model_post = joblib.load("product-return-model/post/best_model.joblib") if lime_training_data_post is None: print("šŸ“Š [LOAD] Loading Post-Delivery LIME Data...") lime_training_data_post = np.load("product-return-model/post/lime_training_data.npy") print("āœ… [SYSTEM] All models and data loaded successfully.") else: print("ā­ļø [SYSTEM] Models are already loaded in memory. Skipping load operation.") return True # --------------------------------------------------------- # EXCEPTION HANDLING & ERROR ROUTING # --------------------------------------------------------- except Exception as e: error_type = type(e).__name__ error_msg = str(e).lower() error_raw = str(e) if error_type == "FileNotFoundError" or "no such file" in error_msg: raise HTTPException(status_code=500, detail=f"🚨 [FILE ERROR] {error_type}: Model or data file is missing. Ensure the paths are correct. Details: {error_raw}") elif error_type == "ValueError" or "unpickling" in error_msg: raise HTTPException(status_code=500, detail=f"🚨 [LOAD ERROR] {error_type}: Failed to load file. Corrupted joblib/npy. Details: {error_raw}") else: raise HTTPException(status_code=500, detail=f"🚨 [SYSTEM ERROR] {error_type}: Unexpected error loading models. Details: {error_raw}") # ============================================================================== # 6. ROOT ENDPOINT (API METADATA & DOCUMENTATION) # ============================================================================== @app.get("/") def home() -> dict: """ Root endpoint: Provides server health status, API metadata, and detailed usage documentation. Serves as a friendly landing page for developers integrating this Product Return Prediction API. """ print("🌐 [API] Root endpoint accessed. Serving metadata and documentation.") return { "status": "āœ… Online", "service": "Product Return Prediction & LIME Explanation API (Pre & Post Delivery)", "version": "1.0.1", "live_urls": { "base_url": "https://silvio0-product-return-api.hf.space", "documentation": "https://silvio0-product-return-api.hf.space/docs", "pre_delivery_prediction": "https://silvio0-product-return-api.hf.space/predict/pre", "post_delivery_prediction": "https://silvio0-product-return-api.hf.space/predict/post", "pre_delivery_explanation": "https://silvio0-product-return-api.hf.space/explain/pre", "post_delivery_explanation": "https://silvio0-product-return-api.hf.space/explain/post" }, "usage_guide": { "endpoints": { "/predict/pre": "POST method - Predicts return probability BEFORE delivery based on order details.", "/predict/post": "POST method - Predicts return probability AFTER delivery including shipping time and rating.", "/explain/pre": "POST method - Generates an interactive LIME HTML explanation for pre-delivery factors.", "/explain/post": "POST method - Generates an interactive LIME HTML explanation for post-delivery factors." }, "payload_structure_pre_delivery": { "customer_age": "integer (Required) - Range: 18 to 69.", "product_category": "string (Required) - 'Groceries', 'Home & Living', 'Sports', 'Electronics', 'Beauty', or 'Fashion'.", "payment_method": "string (Required) - 'Wallet', 'UPI', 'Credit Card', 'Debit Card', or 'Cash on Delivery'.", "order_value_usd": "float (Required) - Range: 10.01 to 718.73.", "order_date": "string (Required) - Format: 'YYYY-MM-DD'." }, "payload_structure_post_delivery": { "customer_age": "integer (Required) - Range: 18 to 69.", "product_category": "string (Required) - 'Groceries', 'Home & Living', 'Sports', 'Electronics', 'Beauty', or 'Fashion'.", "payment_method": "string (Required) - 'Wallet', 'UPI', 'Credit Card', 'Debit Card', or 'Cash on Delivery'.", "order_value_usd": "float (Required) - Range: 10.01 to 718.73.", "delivery_time_days": "integer (Required) - Range: 1 to 14.", "customer_rating": "float (Required) - Range: 1.0 to 5.0.", "order_date": "string (Required) - Format: 'YYYY-MM-DD'." }, "payload_example_pre": { "customer_age": 25, "product_category": "Electronics", "payment_method": "Credit Card", "order_value_usd": 150.50, "order_date": "2026-04-10" }, "payload_example_post": { "customer_age": 28, "product_category": "Fashion", "payment_method": "Wallet", "order_value_usd": 45.99, "delivery_time_days": 3, "customer_rating": 4.5, "order_date": "2026-04-12" } }, "author": "Silvio Christian Joe" } # ============================================================================== # 7. PRE-DELIVERY PREDICTION ENDPOINT # ============================================================================== @app.post("/predict/pre") def predict_pre(input: Annotated[ConditionalInputPre, Form()]) -> dict: """ API Endpoint to predict the likelihood of a product return BEFORE delivery. Accepts customer and order details via form data. """ global best_model_pre print(f"\nšŸ“„ [API REQUEST] Received request at '/predict/pre' for Order Value: ${input.order_value_usd}") # Ensure models are loaded into memory before processing if not load_models(): print("āŒ [API ERROR] Failed to load models into memory.") raise HTTPException(status_code=500, detail="🚨 [SYSTEM ERROR] Critical failure: Machine Learning models could not be loaded.") try: print("āš™ļø [PROCESSING] Preparing pre-delivery data features...") # Transform raw input into a model-ready DataFrame using custom function df_testing = prepare_data_pre( customer_age = input.customer_age, product_category = input.product_category, payment_method = input.payment_method, order_value_usd = input.order_value_usd, order_date = input.order_date ) print("🧠 [PREDICTING] Running inference through the Pre-Delivery Model...") # Generate predictions using the loaded model prediction, returned_proba, prediction_conf = predict(best_model_pre, df_testing) print(f"āœ… [SUCCESS] Prediction generated: {prediction} (Confidence: {prediction_conf})") # Return the structured JSON response return { "prediction": prediction, "returned_proba": returned_proba, "prediction_conf": prediction_conf } # --------------------------------------------------------- # EXCEPTION HANDLING & ERROR ROUTING (API LEVEL) # --------------------------------------------------------- except Exception as e: error_type = type(e).__name__ error_msg = str(e).lower() error_raw = str(e) print("\n" + "="*70) print("šŸ’„ [CRITICAL FAILURE] API Request aborted during prediction!") print("-" * 70) # 1. Handling Missing Columns/Features during data preparation if error_type == "KeyError" or "key" in error_msg: print(f"🚨 [DATA ERROR] {error_type}: Missing required feature/column. Details: {error_raw}") print("="*70 + "\n") raise HTTPException(status_code=400, detail=f"[DATA ERROR] {error_type}: A required data field is missing from the processing pipeline. Details: {error_raw}") # 2. Handling Data Type Mismatches (e.g., trying to process string as float) elif error_type == "TypeError" or "type" in error_msg: print(f"🚨 [DATA ERROR] {error_type}: Incompatible data type encountered. Details: {error_raw}") print("="*70 + "\n") raise HTTPException(status_code=400, detail=f"[DATA ERROR] {error_type}: Incorrect data type passed to the processing function. Details: {error_raw}") # 3. Handling Invalid Values (e.g., feature shape mismatch with ML model) elif error_type == "ValueError" or "value" in error_msg or "shape" in error_msg: print(f"🚨 [MODEL ERROR] {error_type}: Dimension mismatch or invalid values for model inference. Details: {error_raw}") print("="*70 + "\n") raise HTTPException(status_code=422, detail=f"[MODEL ERROR] {error_type}: The input data shape or values do not match the model's expectations. Details: {error_raw}") # 4. Handling Corrupted Model Objects (e.g., missing .predict() method) elif error_type == "AttributeError" or "attribute" in error_msg: print(f"🚨 [SYSTEM ERROR] {error_type}: Model object is corrupted or missing methods. Details: {error_raw}") print("="*70 + "\n") raise HTTPException(status_code=500, detail=f"[SYSTEM ERROR] {error_type}: Internal model architecture error. Details: {error_raw}") # 5. Handling Unfitted Models (Scikit-Learn specific error) elif error_type == "NotFittedError" or "fitted" in error_msg: print(f"🚨 [MODEL ERROR] {error_type}: Attempting to predict using an untrained model. Details: {error_raw}") print("="*70 + "\n") raise HTTPException(status_code=500, detail=f"[MODEL ERROR] {error_type}: The loaded machine learning model is not trained. Details: {error_raw}") # 6. Fallback for any other unknown errors else: print(f"🚨 [UNKNOWN ERROR] {error_type}: Unexpected failure during execution. Details: {error_raw}") print("="*70 + "\n") raise HTTPException(status_code=500, detail=f"[UNKNOWN ERROR] {error_type}: An unexpected server error occurred during prediction. Details: {error_raw}") # ============================================================================== # 8. POST-DELIVERY PREDICTION ENDPOINT # ============================================================================== @app.post("/predict/post") def predict_post(input: Annotated[ConditionalInputPost, Form()]) -> dict: """ API Endpoint to predict the likelihood of a product return AFTER delivery. Accepts customer, order, and post-delivery details via form data. """ global best_model_post print(f"\nšŸ“„ [API REQUEST] Received request at '/predict/post' for Order Value: ${input.order_value_usd}") # Ensure models are loaded into memory before processing if not load_models(): print("āŒ [API ERROR] Failed to load models into memory.") raise HTTPException(status_code=500, detail="🚨 [SYSTEM ERROR] Critical failure: Machine Learning models could not be loaded.") try: print("āš™ļø [PROCESSING] Preparing post-delivery data features...") # Transform raw input into a model-ready DataFrame using custom function df_testing = prepare_data_post( customer_age = input.customer_age, product_category = input.product_category, payment_method = input.payment_method, order_value_usd = input.order_value_usd, delivery_time_days = input.delivery_time_days, customer_rating = input.customer_rating, order_date = input.order_date ) print("🧠 [PREDICTING] Running inference through the Post-Delivery Model...") # Generate predictions using the loaded model prediction, returned_proba, prediction_conf = predict(best_model_post, df_testing) print(f"āœ… [SUCCESS] Prediction generated: {prediction} (Confidence: {prediction_conf})") # Return the structured JSON response return { "prediction": prediction, "returned_proba": returned_proba, "prediction_conf": prediction_conf } # --------------------------------------------------------- # EXCEPTION HANDLING & ERROR ROUTING (API LEVEL) # --------------------------------------------------------- except Exception as e: error_type = type(e).__name__ error_msg = str(e).lower() error_raw = str(e) print("\n" + "="*70) print("šŸ’„ [CRITICAL FAILURE] API Request aborted during prediction!") print("-" * 70) # 1. Handling Missing Columns/Features during data preparation if error_type == "KeyError" or "key" in error_msg: print(f"🚨 [DATA ERROR] {error_type}: Missing required feature/column. Details: {error_raw}") print("="*70 + "\n") raise HTTPException(status_code=400, detail=f"[DATA ERROR] {error_type}: A required data field is missing from the processing pipeline. Details: {error_raw}") # 2. Handling Data Type Mismatches (e.g., trying to process string as float) elif error_type == "TypeError" or "type" in error_msg: print(f"🚨 [DATA ERROR] {error_type}: Incompatible data type encountered. Details: {error_raw}") print("="*70 + "\n") raise HTTPException(status_code=400, detail=f"[DATA ERROR] {error_type}: Incorrect data type passed to the processing function. Details: {error_raw}") # 3. Handling Invalid Values (e.g., feature shape mismatch with ML model) elif error_type == "ValueError" or "value" in error_msg or "shape" in error_msg: print(f"🚨 [MODEL ERROR] {error_type}: Dimension mismatch or invalid values for model inference. Details: {error_raw}") print("="*70 + "\n") raise HTTPException(status_code=422, detail=f"[MODEL ERROR] {error_type}: The input data shape or values do not match the model's expectations. Details: {error_raw}") # 4. Handling Corrupted Model Objects (e.g., missing .predict() method) elif error_type == "AttributeError" or "attribute" in error_msg: print(f"🚨 [SYSTEM ERROR] {error_type}: Model object is corrupted or missing methods. Details: {error_raw}") print("="*70 + "\n") raise HTTPException(status_code=500, detail=f"[SYSTEM ERROR] {error_type}: Internal model architecture error. Details: {error_raw}") # 5. Handling Unfitted Models (Scikit-Learn specific error) elif error_type == "NotFittedError" or "fitted" in error_msg: print(f"🚨 [MODEL ERROR] {error_type}: Attempting to predict using an untrained model. Details: {error_raw}") print("="*70 + "\n") raise HTTPException(status_code=500, detail=f"[MODEL ERROR] {error_type}: The loaded machine learning model is not trained. Details: {error_raw}") # 6. Fallback for any other unknown errors else: print(f"🚨 [UNKNOWN ERROR] {error_type}: Unexpected failure during execution. Details: {error_raw}") print("="*70 + "\n") raise HTTPException(status_code=500, detail=f"[UNKNOWN ERROR] {error_type}: An unexpected server error occurred during prediction. Details: {error_raw}") # ============================================================================== # 9. PRE-DELIVERY EXPLANATION ENDPOINT (LIME) # ============================================================================== @app.post("/explain/pre") def explain_pre(input: Annotated[ConditionalInputPre, Form()]) -> dict: """ API Endpoint to generate a LIME explanation for a PRE-delivery prediction. Returns an HTML string detailing feature contributions to the model's decision. """ global best_model_pre, lime_training_data_pre print(f"\nšŸ“„ [API REQUEST] Received request at '/explain/pre' for Order Value: ${input.order_value_usd}") # Ensure models and LIME background data are loaded into memory if not load_models(): print("āŒ [API ERROR] Failed to load models or LIME training data into memory.") raise HTTPException(status_code=500, detail="🚨 [SYSTEM ERROR] Critical failure: Machine Learning assets could not be loaded.") try: print("āš™ļø [PROCESSING] Preparing pre-delivery data features for explanation...") # Transform raw input into a model-ready DataFrame df_testing = prepare_data_pre( customer_age = input.customer_age, product_category = input.product_category, payment_method = input.payment_method, order_value_usd = input.order_value_usd, order_date = input.order_date ) print("šŸ” [EXPLAINING] Generating LIME explanation. This may take a moment...") # Generate the explanation HTML using the local explain function explanation = explain(best_model_pre, lime_training_data_pre, df_testing) print("āœ… [SUCCESS] LIME Explanation HTML generated successfully.") # Return the generated HTML string wrapped in a JSON response return { "explanation_html": explanation } # --------------------------------------------------------- # EXCEPTION HANDLING & ERROR ROUTING (API LEVEL) # --------------------------------------------------------- except Exception as e: error_type = type(e).__name__ error_msg = str(e).lower() error_raw = str(e) print("\n" + "="*70) print("šŸ’„ [CRITICAL FAILURE] API Request aborted during LIME explanation!") print("-" * 70) # 1. Handling Missing Columns/Features if error_type == "KeyError" or "key" in error_msg: print(f"🚨 [DATA ERROR] {error_type}: Missing required feature/column. Details: {error_raw}") print("="*70 + "\n") raise HTTPException(status_code=400, detail=f"[DATA ERROR] {error_type}: A required data field is missing from the processing pipeline. Details: {error_raw}") # 2. Handling Data Type Mismatches elif error_type == "TypeError" or "type" in error_msg: print(f"🚨 [DATA ERROR] {error_type}: Incompatible data type encountered. Details: {error_raw}") print("="*70 + "\n") raise HTTPException(status_code=400, detail=f"[DATA ERROR] {error_type}: Incorrect data type passed to the processing function. Details: {error_raw}") # 3. Handling LIME Data Mismatch (Crucial for Explainable AI) elif error_type == "ValueError" or "value" in error_msg or "shape" in error_msg: print(f"🚨 [LIME ERROR] {error_type}: Background training data mismatch. Details: {error_raw}") print("="*70 + "\n") raise HTTPException(status_code=422, detail=f"[LIME ERROR] {error_type}: The input data shape does not match the LIME background dataset. Details: {error_raw}") # 4. Handling Corrupted Model/Explainer Objects elif error_type == "AttributeError" or "attribute" in error_msg: print(f"🚨 [SYSTEM ERROR] {error_type}: Model or Explainer object is corrupted. Details: {error_raw}") print("="*70 + "\n") raise HTTPException(status_code=500, detail=f"[SYSTEM ERROR] {error_type}: Internal model architecture error during explanation generation. Details: {error_raw}") # 5. Handling Unfitted Models elif error_type == "NotFittedError" or "fitted" in error_msg: print(f"🚨 [MODEL ERROR] {error_type}: Attempting to explain an untrained model. Details: {error_raw}") print("="*70 + "\n") raise HTTPException(status_code=500, detail=f"[MODEL ERROR] {error_type}: The loaded machine learning model is not trained. Details: {error_raw}") # 6. Fallback for any other unknown errors else: print(f"🚨 [UNKNOWN ERROR] {error_type}: Unexpected failure during execution. Details: {error_raw}") print("="*70 + "\n") raise HTTPException(status_code=500, detail=f"[UNKNOWN ERROR] {error_type}: An unexpected server error occurred during LIME generation. Details: {error_raw}") # ============================================================================== # 10. POST-DELIVERY EXPLANATION ENDPOINT (LIME) # ============================================================================== @app.post("/explain/post") def explain_post(input: Annotated[ConditionalInputPost, Form()]) -> dict: """ API Endpoint to generate a LIME explanation for a POST-delivery prediction. Returns an HTML string detailing feature contributions to the model's decision, including post-delivery metrics like delivery time and customer rating. """ global best_model_post, lime_training_data_post print(f"\nšŸ“„ [API REQUEST] Received request at '/explain/post' for Order Value: ${input.order_value_usd}") # Ensure models and LIME background data are loaded into memory if not load_models(): print("āŒ [API ERROR] Failed to load Post-Delivery models or LIME training data into memory.") raise HTTPException(status_code=500, detail="🚨 [SYSTEM ERROR] Critical failure: Machine Learning assets could not be loaded.") try: print("āš™ļø [PROCESSING] Preparing post-delivery data features for explanation...") # Transform raw input into a model-ready DataFrame df_testing = prepare_data_post( customer_age = input.customer_age, product_category = input.product_category, payment_method = input.payment_method, order_value_usd = input.order_value_usd, delivery_time_days = input.delivery_time_days, customer_rating = input.customer_rating, order_date = input.order_date ) print("šŸ” [EXPLAINING] Generating Post-Delivery LIME explanation. This may take a moment...") # Generate the explanation HTML using the local explain function explanation = explain(best_model_post, lime_training_data_post, df_testing) print("āœ… [SUCCESS] Post-Delivery LIME Explanation HTML generated successfully.") # Return the generated HTML string wrapped in a JSON response return { "explanation_html": explanation } # --------------------------------------------------------- # EXCEPTION HANDLING & ERROR ROUTING (API LEVEL) # --------------------------------------------------------- except Exception as e: error_type = type(e).__name__ error_msg = str(e).lower() error_raw = str(e) print("\n" + "="*70) print("šŸ’„ [CRITICAL FAILURE] API Request aborted during LIME explanation!") print("-" * 70) # 1. Handling Missing Columns/Features if error_type == "KeyError" or "key" in error_msg: print(f"🚨 [DATA ERROR] {error_type}: Missing required feature/column. Details: {error_raw}") print("="*70 + "\n") raise HTTPException(status_code=400, detail=f"[DATA ERROR] {error_type}: A required data field is missing from the processing pipeline. Details: {error_raw}") # 2. Handling Data Type Mismatches elif error_type == "TypeError" or "type" in error_msg: print(f"🚨 [DATA ERROR] {error_type}: Incompatible data type encountered. Details: {error_raw}") print("="*70 + "\n") raise HTTPException(status_code=400, detail=f"[DATA ERROR] {error_type}: Incorrect data type passed to the processing function. Details: {error_raw}") # 3. Handling LIME Data Mismatch (Crucial for Explainable AI) elif error_type == "ValueError" or "value" in error_msg or "shape" in error_msg: print(f"🚨 [LIME ERROR] {error_type}: Background training data mismatch. Details: {error_raw}") print("="*70 + "\n") raise HTTPException(status_code=422, detail=f"[LIME ERROR] {error_type}: The input data shape does not match the LIME background dataset. Details: {error_raw}") # 4. Handling Corrupted Model/Explainer Objects elif error_type == "AttributeError" or "attribute" in error_msg: print(f"🚨 [SYSTEM ERROR] {error_type}: Model or Explainer object is corrupted. Details: {error_raw}") print("="*70 + "\n") raise HTTPException(status_code=500, detail=f"[SYSTEM ERROR] {error_type}: Internal model architecture error during explanation generation. Details: {error_raw}") # 5. Handling Unfitted Models elif error_type == "NotFittedError" or "fitted" in error_msg: print(f"🚨 [MODEL ERROR] {error_type}: Attempting to explain an untrained model. Details: {error_raw}") print("="*70 + "\n") raise HTTPException(status_code=500, detail=f"[MODEL ERROR] {error_type}: The loaded machine learning model is not trained. Details: {error_raw}") # 6. Fallback for any other unknown errors else: print(f"🚨 [UNKNOWN ERROR] {error_type}: Unexpected failure during execution. Details: {error_raw}") print("="*70 + "\n") raise HTTPException(status_code=500, detail=f"[UNKNOWN ERROR] {error_type}: An unexpected server error occurred during LIME generation. Details: {error_raw}")