from __future__ import annotations from pathlib import Path from typing import Dict, Tuple import gradio as gr import joblib import pandas as pd MODEL_PATH = Path("model/model.joblib") DEFAULT_VALUES = { "customer_tenure": 14.0, "total_spent": 850.0, "last_purchase_days": 24, "visit_count": 22, "email_open_rate": 0.58, "discount_usage": 0.24, "support_tickets": 1, "satisfaction_score": 7.8, } class CustomerPredictorModel: def __init__(self, model_path: Path) -> None: if not model_path.exists(): raise FileNotFoundError( "Model file is missing. Run `python train_model.py` first." ) artifacts = joblib.load(model_path) self.pipeline = artifacts["pipeline"] self.metadata = artifacts["metadata"] def _build_explanation(self, inputs: Dict[str, float], label: str) -> str: medians = self.metadata["medians"] importance = self.metadata["feature_importance"] signals = {} for feature, value in inputs.items(): baseline = medians.get(feature, 1.0) if baseline == 0: delta = value else: delta = (value - baseline) / baseline # Align direction so positive value means stronger lead quality. if feature in {"last_purchase_days", "discount_usage", "support_tickets"}: delta = -delta signals[feature] = abs(delta) * importance.get(feature, 0.05) top = sorted(signals.items(), key=lambda x: x[1], reverse=True)[:3] labels = { "customer_tenure": "customer tenure", "total_spent": "lifetime spend", "last_purchase_days": "purchase recency", "visit_count": "visit activity", "email_open_rate": "email engagement", "discount_usage": "discount dependence", "support_tickets": "support load", "satisfaction_score": "satisfaction score", } top_text = ", ".join(labels.get(k, k) for k, _ in top) recommendation = { "Hot Lead": "Prioritize this account for upsell bundles and immediate follow-up.", "Normal Lead": "Keep in nurture flow with segmented offers and reminder campaigns.", "Churn Risk": "Trigger retention workflow with support outreach and win-back incentive.", }[label] return f"Top lead signals: {top_text}. {recommendation}" def predict_one(self, inputs: Dict[str, float]) -> Dict[str, str]: df = pd.DataFrame([inputs]) pred = self.pipeline.predict(df)[0] probas = self.pipeline.predict_proba(df)[0] classes = self.pipeline.classes_ confidence = float(probas[list(classes).index(pred)]) return { "lead_status": pred, "confidence": f"{confidence:.2%}", "explanation": self._build_explanation(inputs, pred), } def predict_csv(self, file_path: str) -> pd.DataFrame: input_df = pd.read_csv(file_path) required = self.metadata["feature_columns"] missing = [col for col in required if col not in input_df.columns] if missing: raise ValueError(f"Missing columns in CSV: {missing}") preds = [] for _, row in input_df.iterrows(): prediction = self.predict_one({col: row[col] for col in required}) preds.append(prediction) return pd.concat([input_df.reset_index(drop=True), pd.DataFrame(preds)], axis=1) model = CustomerPredictorModel(MODEL_PATH) def predict_customer_status( customer_tenure: float, total_spent: float, last_purchase_days: int, visit_count: int, email_open_rate: float, discount_usage: float, support_tickets: int, satisfaction_score: float, ) -> Tuple[str, str, str]: inputs = { "customer_tenure": customer_tenure, "total_spent": total_spent, "last_purchase_days": int(last_purchase_days), "visit_count": int(visit_count), "email_open_rate": email_open_rate, "discount_usage": discount_usage, "support_tickets": int(support_tickets), "satisfaction_score": satisfaction_score, } result = model.predict_one(inputs) return result["lead_status"], result["confidence"], result["explanation"] def score_csv(file_obj): if file_obj is None: raise gr.Error("Upload a CSV file to continue.") file_path = file_obj if isinstance(file_obj, str) else file_obj.name return model.predict_csv(file_path) demo = gr.Blocks(title="Lead.AI Customer Predictor") with demo: gr.Markdown( "# Lead.AI Customer Predictor\n" "Find which customers are likely to buy again. " "Predict Hot Lead, Normal Lead, or Churn Risk." ) with gr.Tab("Live Demo"): with gr.Row(): customer_tenure = gr.Number(label="customer_tenure", value=DEFAULT_VALUES["customer_tenure"]) total_spent = gr.Number(label="total_spent", value=DEFAULT_VALUES["total_spent"]) last_purchase_days = gr.Number(label="last_purchase_days", value=DEFAULT_VALUES["last_purchase_days"]) visit_count = gr.Number(label="visit_count", value=DEFAULT_VALUES["visit_count"]) with gr.Row(): email_open_rate = gr.Slider(label="email_open_rate", minimum=0, maximum=1, value=DEFAULT_VALUES["email_open_rate"], step=0.01) discount_usage = gr.Slider(label="discount_usage", minimum=0, maximum=1, value=DEFAULT_VALUES["discount_usage"], step=0.01) support_tickets = gr.Number(label="support_tickets", value=DEFAULT_VALUES["support_tickets"]) satisfaction_score = gr.Slider(label="satisfaction_score", minimum=1, maximum=10, value=DEFAULT_VALUES["satisfaction_score"], step=0.1) predict_btn = gr.Button("Score Lead") label_output = gr.Label(label="Lead Status") confidence_output = gr.Textbox(label="Model Confidence") explanation_output = gr.Textbox(label="Explanation", lines=3) predict_btn.click( predict_customer_status, inputs=[ customer_tenure, total_spent, last_purchase_days, visit_count, email_open_rate, discount_usage, support_tickets, satisfaction_score, ], outputs=[label_output, confidence_output, explanation_output], ) gr.Examples( examples=[ [24, 3400, 8, 42, 0.76, 0.12, 0, 9.1], [7, 420, 56, 11, 0.34, 0.48, 2, 6.4], [4, 110, 140, 5, 0.12, 0.62, 4, 4.2], ], inputs=[ customer_tenure, total_spent, last_purchase_days, visit_count, email_open_rate, discount_usage, support_tickets, satisfaction_score, ], label="Try Example", ) with gr.Tab("Upload CSV"): csv_input = gr.File(label="Upload CSV with required feature columns", file_types=[".csv"]) csv_btn = gr.Button("Batch Score") csv_output = gr.Dataframe(label="Scored Results") csv_btn.click(score_csv, inputs=[csv_input], outputs=[csv_output]) gr.Markdown( "### Pricing\n" "- Starter: $49/month (single-model access for SMB teams)\n" "- Business: $149/month (lead scoring + CRM export + automation)\n" "- Custom AI Setup: $499-$1,500 one-time" ) if __name__ == "__main__": demo.launch()