Spaces:
Runtime error
Runtime error
| import numpy as np | |
| import pandas as pd | |
| import shap | |
| FEATURE_LABEL_MAP = { | |
| "recency_days": "Days Inactive", | |
| "frequency_purchases": "Total Purchases", | |
| "monetary_total_spend": "Lifetime Spend", | |
| "avg_order_value": "Avg Order Value", | |
| "website_visits_30d": "30-Day Website Visits", | |
| "product_views_30d": "30-Day Product Views", | |
| "cart_additions_30d": "Cart Additions", | |
| "cart_abandonments_30d": "Abandoned Carts", | |
| "email_opens_30d": "Email Opens", | |
| "email_clicks_30d": "Email Clicks", | |
| "support_tickets_30d": "Support Tickets", | |
| "engagement_change_pct": "Engagement Trend %", | |
| "return_rate": "Return Rate %", | |
| "discount_dependency_ratio": "Discount Dependency", | |
| "spend_30d": "30-Day Spend", | |
| "spend_90d": "90-Day Spend", | |
| "purchase_velocity": "Days Between Purchases", | |
| "cart_abandon_ratio": "Cart Abandon Rate", | |
| "email_click_ratio": "Email Click Rate", | |
| "rfm_score": "RFM Score" | |
| } | |
| class ShapExplainabilityEngine: | |
| def __init__(self, trained_model, feature_names): | |
| self.trained_model = trained_model | |
| self.feature_names = feature_names | |
| try: | |
| # TreeExplainer for XGBoost/RandomForest | |
| self.explainer = shap.TreeExplainer(self.trained_model) | |
| except Exception: | |
| # Fallback | |
| self.explainer = None | |
| def explain_instance(self, customer_features: pd.DataFrame, top_n: int = 5) -> list[dict]: | |
| """ | |
| Calculates SHAP feature attributions for a single customer sample. | |
| Returns top_n positive & negative behavioral drivers. | |
| """ | |
| if customer_features.ndim == 1: | |
| customer_features = customer_features.to_frame().T | |
| features_df = customer_features[self.feature_names] | |
| if self.explainer is not None: | |
| try: | |
| shap_values = self.explainer.shap_values(features_df) | |
| # Handle binary classification 2D or 3D output | |
| if isinstance(shap_values, list): | |
| shap_vals = shap_values[1][0] | |
| elif shap_values.ndim == 2: | |
| shap_vals = shap_values[0] | |
| else: | |
| shap_vals = shap_values[0] | |
| except Exception: | |
| shap_vals = self._fallback_feature_importance(features_df) | |
| else: | |
| shap_vals = self._fallback_feature_importance(features_df) | |
| explanations = [] | |
| for feat_name, val, s_val in zip(self.feature_names, features_df.iloc[0], shap_vals): | |
| label = FEATURE_LABEL_MAP.get(feat_name, feat_name) | |
| # Format value for human display | |
| if "pct" in feat_name or "rate" in feat_name or "ratio" in feat_name: | |
| formatted_val = f"{val * 100:.1f}%" if "pct" not in feat_name else f"{val:.1f}%" | |
| elif "spend" in feat_name or "value" in feat_name or "clv" in feat_name: | |
| formatted_val = f"Rs.{val:,.0f}" | |
| else: | |
| formatted_val = f"{val:g}" if isinstance(val, (int, float)) else str(val) | |
| impact_direction = "INCREASES_RISK" if s_val > 0 else "DECREASES_RISK" | |
| explanations.append({ | |
| "feature": feat_name, | |
| "label": label, | |
| "value": formatted_val, | |
| "raw_value": float(val), | |
| "shap_value": float(round(s_val, 4)), | |
| "impact": impact_direction, | |
| "abs_shap": abs(float(s_val)) | |
| }) | |
| # Sort by absolute SHAP impact | |
| explanations.sort(key=lambda x: x["abs_shap"], reverse=True) | |
| return explanations[:top_n] | |
| def _fallback_feature_importance(self, features_df: pd.DataFrame) -> np.ndarray: | |
| # Fallback heuristic impact calculation if SHAP tree evaluation fails | |
| row = features_df.iloc[0] | |
| impacts = [] | |
| for feat in self.feature_names: | |
| val = row[feat] | |
| if feat == "recency_days": | |
| impacts.append(0.005 * (val - 30)) | |
| elif feat == "cart_abandonments_30d": | |
| impacts.append(0.08 * val) | |
| elif feat == "support_tickets_30d": | |
| impacts.append(0.12 * val) | |
| elif feat == "engagement_change_pct": | |
| impacts.append(-0.004 * val) | |
| elif feat == "product_views_30d": | |
| impacts.append(-0.003 * val) | |
| else: | |
| impacts.append(0.01 * (val / (abs(val) + 1.0))) | |
| return np.array(impacts) | |