Spaces:
Build error
Build error
| import os | |
| import re | |
| import json | |
| import numpy as np | |
| import pandas as pd | |
| import matplotlib.pyplot as plt | |
| import seaborn as sns | |
| from typing import List, Dict, Any | |
| from datetime import datetime | |
| from openai import OpenAI | |
| from agent import Agent | |
| from sklearn.cluster import KMeans | |
| from sklearn.preprocessing import RobustScaler | |
| from sklearn.metrics import silhouette_score | |
| from sklearn.decomposition import PCA | |
| class SegmentationAgent(Agent): | |
| """ | |
| Agent that uses K-means clustering and DeepSeek LLM to segment customers based on RFM data | |
| """ | |
| name = "Segmentation Agent" | |
| color = Agent.CYAN | |
| MODEL = "deepseek-r1:1.5b" | |
| def __init__(self, output_dir="output"): | |
| """ | |
| Set up this instance by connecting to DeepSeek and initializing segmentation state | |
| """ | |
| self.log("Initializing Segmentation Agent") | |
| # Store output directory | |
| self.output_dir = output_dir | |
| # Create output directory | |
| os.makedirs(self.output_dir, exist_ok=True) | |
| # Set up the DeepSeek client via Ollama | |
| self.client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama") | |
| # Segmentation state | |
| self.rfm_data = None | |
| self.segments = None | |
| self.segment_descriptions = None | |
| self.segment_strategies = None | |
| self.kmeans_model = None | |
| self.scaler = None | |
| self.min_clusters = 3 # Minimum number of clusters to ensure | |
| self.log(f"Segmentation Agent is ready with {self.MODEL}") | |
| def load_data(self, data_path: str) -> bool: | |
| """ | |
| Load the RFM data for segmentation | |
| """ | |
| self.log(f"Loading data from {data_path}") | |
| try: | |
| # Determine file type and load accordingly | |
| if data_path.endswith('.parquet'): | |
| self.rfm_data = pd.read_parquet(data_path) | |
| elif data_path.endswith('.csv'): | |
| self.rfm_data = pd.read_csv(data_path) | |
| else: | |
| self.log(f"Unsupported file format: {data_path}") | |
| return False | |
| # Ensure required columns exist | |
| required_columns = ['Customer ID', 'Recency', 'Frequency', 'Monetary'] | |
| for col in required_columns: | |
| if col not in self.rfm_data.columns: | |
| self.log(f"Missing required column: {col}") | |
| return False | |
| self.log(f"Data loaded successfully with {len(self.rfm_data)} customers") | |
| return True | |
| except Exception as e: | |
| self.log(f"Error loading data: {str(e)}") | |
| return False | |
| def _get_rfm_summary(self) -> Dict[str, Any]: | |
| """Analyze RFM data and create a summary for the LLM""" | |
| summary = {} | |
| # Calculate basic statistics | |
| for metric in ['Recency', 'Frequency', 'Monetary']: | |
| summary[metric] = { | |
| "min": float(self.rfm_data[metric].min()), | |
| "max": float(self.rfm_data[metric].max()), | |
| "mean": float(self.rfm_data[metric].mean()), | |
| "median": float(self.rfm_data[metric].median()), | |
| "p25": float(self.rfm_data[metric].quantile(0.25)), | |
| "p75": float(self.rfm_data[metric].quantile(0.75)) | |
| } | |
| # Calculate correlations | |
| corr_matrix = self.rfm_data[['Recency', 'Frequency', 'Monetary']].corr() | |
| summary["correlations"] = { | |
| "Recency_Frequency": float(corr_matrix.loc['Recency', 'Frequency']), | |
| "Recency_Monetary": float(corr_matrix.loc['Recency', 'Monetary']), | |
| "Frequency_Monetary": float(corr_matrix.loc['Frequency', 'Monetary']) | |
| } | |
| return summary | |
| def messages_for_cluster_interpretation(self, clusters: Dict[str, Any]) -> List[Dict[str, str]]: | |
| """ | |
| Create the message list for generating cluster interpretations with improved prompting | |
| """ | |
| system_message = """You are an expert in customer segmentation and marketing analytics. | |
| Your task is to interpret K-means clusters from RFM (Recency, Frequency, Monetary) data | |
| and provide meaningful segment names and descriptions. Follow the standard RFM segmentation | |
| framework used in marketing analytics.""" | |
| user_message = f""" | |
| I have performed K-means clustering on customer RFM data and need your help interpreting the clusters. | |
| Here are the clusters with their average RFM values and other statistics: | |
| {json.dumps(clusters, indent=2)} | |
| Remember that: | |
| - Recency: Days since last purchase (LOWER is better) | |
| - Frequency: Number of purchases (HIGHER is better) | |
| - Monetary: Average spending per customer (HIGHER is better) | |
| Use these standard segment names when appropriate: | |
| - Champions: Recent purchases, frequent buyers, high spending | |
| - Loyal Customers: Buy regularly but not as recently as Champions | |
| - New Customers: Recent first purchase, low frequency/spend | |
| - At-Risk Customers: Above-average recency, good past purchase frequency | |
| - Big Spenders: High monetary values but lower frequency | |
| - Lost Customers: High recency (haven't purchased in a long time) | |
| For each cluster, please: | |
| 1. Provide a descriptive segment name from the standard list above when applicable, or create a custom name if needed | |
| 2. Write a detailed description of the customer behavior and characteristics | |
| 3. Suggest specific marketing strategies tailored to this customer segment | |
| Format your response as a JSON object with the following structure: | |
| {{ | |
| "renamed_clusters": {{ | |
| "0": "Descriptive name for cluster 0", | |
| "1": "Descriptive name for cluster 1", | |
| ... | |
| }}, | |
| "descriptions": {{ | |
| "0": "Detailed description for cluster 0...", | |
| "1": "Detailed description for cluster 1...", | |
| ... | |
| }}, | |
| "strategies": {{ | |
| "0": "Marketing strategies for cluster 0...", | |
| "1": "Marketing strategies for cluster 1...", | |
| ... | |
| }} | |
| }} | |
| Be specific and insightful in your analysis. Use the RFM values to inform your interpretation. | |
| Return only the JSON object. | |
| """ | |
| return [ | |
| {"role": "system", "content": system_message}, | |
| {"role": "user", "content": user_message}, | |
| {"role": "assistant", "content": "{"} | |
| ] | |
| def messages_for_report(self, segments: Dict[str, Any], descriptions: Dict[str, str], strategies: Dict[str, str]) -> List[Dict[str, str]]: | |
| """ | |
| Create the message list for generating a report | |
| """ | |
| system_message = """You are a marketing analytics expert who creates comprehensive business reports. | |
| Create a detailed customer segmentation report in Markdown format.""" | |
| user_message = f""" | |
| Create a comprehensive customer segmentation report based on the provided segmentation data. The report should include: | |
| 1. Executive Summary | |
| 2. Introduction and Methodology (mention that K-means clustering was used on RFM data) | |
| 3. Overview of Customer Segments | |
| 4. Detailed Analysis of Each Segment | |
| 5. Marketing Recommendations | |
| 6. Conclusion | |
| Here's the segmentation data: | |
| {json.dumps(segments)} | |
| Additional insights about each segment: | |
| Descriptions: | |
| {json.dumps(descriptions)} | |
| Marketing Strategies: | |
| {json.dumps(strategies)} | |
| Format the report in Markdown with proper headings, bullet points, and any tables/charts you think would be useful (described in Markdown). | |
| Make the report professional, insightful, and actionable for business stakeholders. | |
| """ | |
| return [ | |
| {"role": "system", "content": system_message}, | |
| {"role": "user", "content": user_message} | |
| ] | |
| def extract_json(self, content: str) -> Any: | |
| """Extract JSON from LLM response""" | |
| try: | |
| # Try to parse as-is first | |
| return json.loads(content) | |
| except: | |
| # Try to extract JSON if embedded in other text | |
| try: | |
| # For objects | |
| json_match = re.search(r'\{.*\}', content, re.DOTALL) | |
| if json_match: | |
| return json.loads(json_match.group(0)) | |
| # For arrays | |
| json_match = re.search(r'\[.*\]', content, re.DOTALL) | |
| if json_match: | |
| return json.loads(json_match.group(0)) | |
| return None | |
| except: | |
| return None | |
| def _find_optimal_k(self, X_scaled, max_k=10) -> int: | |
| """Find optimal number of clusters using silhouette score and elbow method""" | |
| self.log("Finding optimal number of clusters") | |
| if len(X_scaled) < self.min_clusters: | |
| return self.min_clusters | |
| max_k = min(max_k, len(X_scaled) - 1) | |
| silhouette_scores = [] | |
| inertia_values = [] | |
| k_range = range(2, max_k + 1) | |
| for k in k_range: | |
| kmeans = KMeans(n_clusters=k, random_state=42, n_init=15, init='k-means++') | |
| cluster_labels = kmeans.fit_predict(X_scaled) | |
| silhouette_avg = silhouette_score(X_scaled, cluster_labels) | |
| silhouette_scores.append(silhouette_avg) | |
| inertia_values.append(kmeans.inertia_) | |
| self.log(f"K={k}, Silhouette Score={silhouette_avg:.4f}, Inertia={kmeans.inertia_:.2f}") | |
| # Find the best k using both silhouette score and elbow method | |
| # Calculate inertia differences (for elbow method) | |
| inertia_diffs = np.diff(inertia_values) | |
| inertia_diffs = np.append(inertia_diffs, inertia_diffs[-1]) # pad last element | |
| # Normalize both metrics between 0 and 1 | |
| norm_silhouette = np.array(silhouette_scores) / max(silhouette_scores) | |
| norm_inertia_diffs = inertia_diffs / max(inertia_diffs) | |
| # Combined score: high silhouette and significant inertia drop | |
| combined_score = norm_silhouette + norm_inertia_diffs | |
| optimal_k = k_range[np.argmax(combined_score)] | |
| self.log(f"Optimal number of clusters: {optimal_k}") | |
| # Force a minimum of desired clusters for better segmentation | |
| optimal_k = max(optimal_k, self.min_clusters) | |
| self.log(f"Final number of clusters (after enforcing minimum): {optimal_k}") | |
| return optimal_k | |
| def perform_segmentation(self, num_segments: int = 0) -> Dict[str, Any]: | |
| """ | |
| Perform customer segmentation using K-means clustering | |
| """ | |
| self.log(f"Starting K-means segmentation") | |
| if self.rfm_data is None: | |
| self.log("No data loaded. Please load data first.") | |
| return {} | |
| # Filter outliers first, similar to notebook approach | |
| filtered_data = self._filter_outliers(self.rfm_data) | |
| # Get RFM features for clustering | |
| X = filtered_data[['Recency', 'Frequency', 'Monetary']].copy() | |
| # Scale features using RobustScaler (same as in notebook) | |
| self.scaler = RobustScaler() | |
| X_scaled = self.scaler.fit_transform(X) | |
| # Find optimal number of clusters if not specified | |
| if num_segments <= 0: | |
| num_segments = self._find_optimal_k(X_scaled, max_k=8) | |
| # Perform K-means clustering with improved parameters | |
| self.log(f"Performing K-means clustering with {num_segments} clusters") | |
| self.kmeans_model = KMeans( | |
| n_clusters=num_segments, | |
| random_state=42, | |
| n_init=20, # Increase number of initializations for better stability | |
| init='k-means++', | |
| max_iter=1000, # Increased from 500 for better convergence | |
| tol=1e-5 # Tighter tolerance for better convergence | |
| ) | |
| # Fit model on filtered data | |
| cluster_labels = self.kmeans_model.fit_predict(X_scaled) | |
| filtered_data['segment_id'] = cluster_labels | |
| # Now propagate cluster assignments to the original dataset | |
| # Create a mapping from customer ID to cluster | |
| customer_to_cluster = dict(zip(filtered_data['Customer ID'], filtered_data['segment_id'])) | |
| # Apply to the entire original dataset | |
| self.rfm_data['segment_id'] = self.rfm_data['Customer ID'].map(customer_to_cluster) | |
| # Fill any unassigned customers (those that were outliers) | |
| if self.rfm_data['segment_id'].isna().any(): | |
| # For unassigned customers, predict their cluster | |
| missing_mask = self.rfm_data['segment_id'].isna() | |
| missing_customers = self.rfm_data[missing_mask] | |
| if len(missing_customers) > 0: | |
| missing_X = missing_customers[['Recency', 'Frequency', 'Monetary']].values | |
| missing_X_scaled = self.scaler.transform(missing_X) | |
| missing_labels = self.kmeans_model.predict(missing_X_scaled) | |
| self.rfm_data.loc[missing_mask, 'segment_id'] = missing_labels | |
| # Ensure segment_id is integer type | |
| self.rfm_data['segment_id'] = self.rfm_data['segment_id'].astype(int) | |
| # Debug clustering results | |
| unique, counts = np.unique(self.rfm_data['segment_id'], return_counts=True) | |
| cluster_distribution = dict(zip(unique, counts)) | |
| self.log(f"Cluster distribution: {cluster_distribution}") | |
| # Create segment information | |
| segments = {} | |
| for i in range(num_segments): | |
| segment_mask = self.rfm_data['segment_id'] == i | |
| segment_data = self.rfm_data[segment_mask] | |
| # Add temporary generic names | |
| self.rfm_data.loc[segment_mask, 'segment_name'] = f"Cluster {i}" | |
| segments[str(i)] = { | |
| "segment_id": i, | |
| "segment_name": f"Cluster {i}", | |
| "count": int(segment_mask.sum()), | |
| "percentage": float(segment_mask.sum() / len(self.rfm_data) * 100), | |
| "center_recency": float(self.kmeans_model.cluster_centers_[i, 0]), | |
| "center_frequency": float(self.kmeans_model.cluster_centers_[i, 1]), | |
| "center_monetary": float(self.kmeans_model.cluster_centers_[i, 2]), | |
| "avg_recency": float(segment_data['Recency'].mean()), | |
| "avg_frequency": float(segment_data['Frequency'].mean()), | |
| "avg_monetary": float(segment_data['Monetary'].mean()), | |
| "total_monetary": float(segment_data['Monetary'].sum()), | |
| "std_recency": float(segment_data['Recency'].std()), | |
| "std_frequency": float(segment_data['Frequency'].std()), | |
| "std_monetary": float(segment_data['Monetary'].std()), | |
| "min_recency": float(segment_data['Recency'].min()), | |
| "max_recency": float(segment_data['Recency'].max()), | |
| "min_frequency": float(segment_data['Frequency'].min()), | |
| "max_frequency": float(segment_data['Frequency'].max()), | |
| "min_monetary": float(segment_data['Monetary'].min()), | |
| "max_monetary": float(segment_data['Monetary'].max()) | |
| } | |
| self.segments = segments | |
| # Use LLM to interpret the clusters with improved prompting | |
| try: | |
| self.log(f"Calling {self.MODEL} to interpret clusters") | |
| response = self.client.chat.completions.create( | |
| model=self.MODEL, | |
| messages=self.messages_for_cluster_interpretation(self.segments), | |
| max_tokens=2048 | |
| ) | |
| content = response.choices[0].message.content | |
| cluster_insights = self.extract_json(content) | |
| if (not cluster_insights or | |
| "renamed_clusters" not in cluster_insights or | |
| "descriptions" not in cluster_insights or | |
| "strategies" not in cluster_insights): | |
| self.log("Failed to parse cluster insights from LLM response") | |
| insights = self._get_fallback_insights(self.segments) | |
| self.segment_descriptions = insights["descriptions"] | |
| self.segment_strategies = insights["strategies"] | |
| # Use fallback names based on RFM patterns instead of generic names | |
| self._apply_rfm_based_names(self.segments) | |
| else: | |
| # Update segment names | |
| for segment_id, new_name in cluster_insights["renamed_clusters"].items(): | |
| if segment_id in self.segments: | |
| self.segments[segment_id]["segment_name"] = new_name | |
| # Update in the dataframe too | |
| self.rfm_data.loc[self.rfm_data['segment_id'] == int(segment_id), 'segment_name'] = new_name | |
| self.segment_descriptions = cluster_insights["descriptions"] | |
| self.segment_strategies = cluster_insights["strategies"] | |
| except Exception as e: | |
| self.log(f"Error using LLM for cluster interpretation: {str(e)}") | |
| insights = self._get_fallback_insights(self.segments) | |
| self.segment_descriptions = insights["descriptions"] | |
| self.segment_strategies = insights["strategies"] | |
| self._apply_rfm_based_names(self.segments) | |
| # Compile results | |
| results = { | |
| "num_customers": len(self.rfm_data), | |
| "num_segments": len(self.segments), | |
| "segments": self.segments, | |
| "segment_descriptions": self.segment_descriptions, | |
| "segment_strategies": self.segment_strategies, | |
| "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S") | |
| } | |
| # Save results | |
| results_file = os.path.join(self.output_dir, "kmeans_segmentation_results.json") | |
| with open(results_file, "w") as f: | |
| json.dump(results, f, indent=2) | |
| self.log(f"Segmentation complete - created {len(self.segments)} segments with K-means") | |
| return results | |
| def _apply_rfm_based_names(self, segments): | |
| """Apply RFM-based names to clusters based on their characteristics""" | |
| # Sort segments by their RFM metrics for consistent naming | |
| sorted_segments = [] | |
| for seg_id, segment in segments.items(): | |
| # Calculate RFM scores (1-5 scale) | |
| r_score = 5 - min(int(segment['avg_recency'] / 60) + 1, 5) # Lower recency is better | |
| f_score = min(int(segment['avg_frequency'] / 5) + 1, 5) # Higher frequency is better | |
| m_score = min(int(segment['avg_monetary'] / 200) + 1, 5) # Higher monetary is better | |
| rfm_score = r_score * 100 + f_score * 10 + m_score # Combine for sorting | |
| sorted_segments.append((seg_id, segment, rfm_score, r_score, f_score, m_score)) | |
| # Sort by RFM score descending | |
| sorted_segments.sort(key=lambda x: x[2], reverse=True) | |
| # Name segments based on their RFM patterns and relative position | |
| for i, (seg_id, segment, _, r_score, f_score, m_score) in enumerate(sorted_segments): | |
| if r_score >= 4 and f_score >= 4 and m_score >= 4: | |
| name = "Champions" | |
| elif r_score >= 4 and f_score >= 3: | |
| name = "Loyal Customers" | |
| elif r_score >= 4 and f_score <= 2: | |
| name = "New Customers" | |
| elif r_score <= 2 and f_score >= 3: | |
| name = "At-Risk Customers" | |
| elif r_score <= 2 and f_score <= 2 and m_score <= 2: | |
| name = "Lost Customers" | |
| elif m_score >= 4 and f_score <= 2: | |
| name = "Big Spenders" | |
| elif r_score <= 2 and m_score >= 4: | |
| name = "At-Risk High Value" | |
| else: | |
| name = "Regular Customers" | |
| # Update segment name | |
| segment["segment_name"] = name | |
| # Update in dataframe | |
| self.rfm_data.loc[self.rfm_data['segment_id'] == int(seg_id), 'segment_name'] = name | |
| def _filter_outliers(self, dataframe): | |
| """Filter outliers using IQR method, similar to notebook approach""" | |
| self.log("Filtering outliers using IQR method") | |
| df = dataframe.copy() | |
| for column in ['Recency', 'Frequency', 'Monetary']: | |
| Q1 = df[column].quantile(0.01) # Use 1st percentile instead of 0.25 for less aggressive filtering | |
| Q3 = df[column].quantile(0.99) # Use 99th percentile instead of 0.75 | |
| IQR = Q3 - Q1 | |
| lower_bound = Q1 - 1.5 * IQR | |
| upper_bound = Q3 + 1.5 * IQR | |
| # Filter outliers | |
| df = df[(df[column] >= lower_bound) & (df[column] <= upper_bound)] | |
| self.log(f"After outlier removal: {len(df)} customers (removed {len(dataframe) - len(df)} outliers)") | |
| return df | |
| def _get_fallback_insights(self, segments: Dict[str, Any]) -> Dict[str, Any]: | |
| """Generate fallback insights if LLM fails""" | |
| descriptions = {} | |
| strategies = {} | |
| for segment_id, segment in segments.items(): | |
| # Create basic descriptions | |
| descriptions[segment_id] = ( | |
| f"Cluster {segment_id}: This segment contains {segment['count']} customers ({segment['percentage']:.1f}% of total). " | |
| f"On average, they made their last purchase {segment['avg_recency']:.1f} days ago, " | |
| f"have made {segment['avg_frequency']:.1f} purchases, and " | |
| f"spent ${segment['avg_monetary']:.2f} per customer." | |
| ) | |
| # Create basic strategies based on RFM values | |
| recency = segment['avg_recency'] | |
| frequency = segment['avg_frequency'] | |
| monetary = segment['avg_monetary'] | |
| if recency < segments["0"]["avg_recency"] and frequency > segments["0"]["avg_frequency"]: | |
| strategies[segment_id] = ( | |
| "These are likely your best customers. Reward them with loyalty programs, exclusive offers, " | |
| "and early access to new products. Focus on building relationships and turning them into brand advocates." | |
| ) | |
| elif recency > segments["0"]["avg_recency"] and frequency > segments["0"]["avg_frequency"]: | |
| strategies[segment_id] = ( | |
| "These customers used to shop frequently but haven't returned recently. Implement a win-back campaign " | |
| "with personalized offers based on their past purchases." | |
| ) | |
| elif recency < segments["0"]["avg_recency"] and frequency < segments["0"]["avg_frequency"]: | |
| strategies[segment_id] = ( | |
| "These are likely new customers. Focus on onboarding and education about your products and services. " | |
| "Cross-sell related products to increase their engagement." | |
| ) | |
| else: | |
| strategies[segment_id] = ( | |
| "Develop personalized communication and offers based on their purchasing patterns. " | |
| "Analyze their behavior to find opportunities to increase engagement and spending." | |
| ) | |
| return { | |
| "descriptions": descriptions, | |
| "strategies": strategies | |
| } | |
| def generate_report(self, output_file: str = "customer_segmentation_report.md") -> str: | |
| """ | |
| Generate a comprehensive customer segmentation report | |
| """ | |
| self.log(f"Generating customer segmentation report") | |
| if self.segments is None: | |
| self.log("No segmentation results. Please perform segmentation first.") | |
| return "" | |
| # Call DeepSeek to generate the report | |
| try: | |
| response = self.client.chat.completions.create( | |
| model=self.MODEL, | |
| messages=self.messages_for_report( | |
| self.segments, | |
| self.segment_descriptions, | |
| self.segment_strategies | |
| ), | |
| max_tokens=4096 | |
| ) | |
| report_content = response.choices[0].message.content.strip() | |
| except Exception as e: | |
| self.log(f"Error generating report: {str(e)}") | |
| report_content = self._generate_fallback_report() | |
| # Save the report | |
| output_path = os.path.join(self.output_dir, output_file) | |
| with open(output_path, "w") as f: | |
| f.write(report_content) | |
| self.log(f"Report saved to {output_path}") | |
| return output_path | |
| def _generate_fallback_report(self) -> str: | |
| """Generate a simple fallback report if LLM fails""" | |
| report = [] | |
| report.append("# Customer Segmentation Report (K-means Clustering)\n") | |
| report.append("## Executive Summary\n") | |
| report.append("This report presents a customer segmentation analysis based on K-means clustering of RFM (Recency, Frequency, Monetary) data.\n") | |
| report.append("## Overview of Customer Segments\n") | |
| segment_table = "| Segment ID | Segment Name | Customer Count | Percentage | Avg. Recency | Avg. Frequency | Avg. Monetary |\n" | |
| segment_table += "|------------|--------------|---------------|------------|-------------|---------------|---------------|\n" | |
| for segment_id, segment in self.segments.items(): | |
| segment_table += f"| {segment['segment_id']} | {segment['segment_name']} | {segment['count']} | {segment['percentage']:.1f}% | {segment['avg_recency']:.1f} | {segment['avg_frequency']:.1f} | ${segment['avg_monetary']:.2f} |\n" | |
| report.append(segment_table) | |
| report.append("\n## Detailed Analysis of Each Segment\n") | |
| for segment_id, segment in self.segments.items(): | |
| report.append(f"### {segment['segment_name']}\n") | |
| if segment_id in self.segment_descriptions: | |
| report.append(f"{self.segment_descriptions[segment_id]}\n") | |
| else: | |
| report.append(f"- Customer count: {segment['count']} ({segment['percentage']:.1f}%)\n") | |
| report.append(f"- Average recency: {segment['avg_recency']:.1f} days\n") | |
| report.append(f"- Average frequency: {segment['avg_frequency']:.1f} purchases\n") | |
| report.append(f"- Average monetary value: ${segment['avg_monetary']:.2f}\n") | |
| report.append("\n**Marketing Strategies:**\n") | |
| if segment_id in self.segment_strategies: | |
| report.append(f"{self.segment_strategies[segment_id]}\n") | |
| report.append("\n") | |
| report.append("## Conclusion\n") | |
| report.append("This segmentation using K-means clustering has identified distinct customer groups that can be targeted with specific marketing strategies.\n") | |
| return "\n".join(report) | |
| def plot_segments(self, output_file: str = "segment_visualization.png") -> str: | |
| """ | |
| Generate a visualization of the customer segments | |
| """ | |
| self.log("Generating segment visualization") | |
| if self.rfm_data is None or 'segment_name' not in self.rfm_data.columns: | |
| self.log("No segmentation results. Please perform segmentation first.") | |
| return "" | |
| # Create a figure with 4 subplots | |
| fig, axes = plt.subplots(2, 2, figsize=(15, 12)) | |
| # Use a consistent color palette | |
| unique_segments = self.rfm_data['segment_name'].unique() | |
| color_palette = sns.color_palette('tab10', len(unique_segments)) | |
| segment_colors = dict(zip(unique_segments, color_palette)) | |
| # 1. Pie chart of segment sizes | |
| segment_counts = self.rfm_data['segment_name'].value_counts() | |
| axes[0, 0].pie( | |
| segment_counts, | |
| labels=segment_counts.index, | |
| autopct='%1.1f%%', | |
| startangle=90, | |
| colors=[segment_colors[seg] for seg in segment_counts.index] | |
| ) | |
| axes[0, 0].set_title('Customer Segment Distribution') | |
| # 2. Scatter plot of Recency vs Frequency | |
| sns.scatterplot( | |
| data=self.rfm_data, | |
| x='Recency', | |
| y='Frequency', | |
| hue='segment_name', | |
| palette=segment_colors, | |
| alpha=0.7, | |
| ax=axes[0, 1] | |
| ) | |
| axes[0, 1].set_title('Recency vs Frequency by Segment') | |
| # 3. Scatter plot of Frequency vs Monetary | |
| sns.scatterplot( | |
| data=self.rfm_data, | |
| x='Frequency', | |
| y='Monetary', | |
| hue='segment_name', | |
| palette=segment_colors, | |
| alpha=0.7, | |
| ax=axes[1, 0] | |
| ) | |
| axes[1, 0].set_title('Frequency vs Monetary by Segment') | |
| # 4. PCA visualization | |
| # Apply PCA for visualization | |
| X = self.rfm_data[['Recency', 'Frequency', 'Monetary']] | |
| pca = PCA(n_components=2) | |
| X_pca = pca.fit_transform(self.scaler.transform(X)) | |
| # Create a dataframe with PCA results | |
| pca_df = pd.DataFrame(data=X_pca, columns=['PC1', 'PC2']) | |
| pca_df['segment_name'] = self.rfm_data['segment_name'].values | |
| # Plot PCA results | |
| sns.scatterplot( | |
| data=pca_df, | |
| x='PC1', | |
| y='PC2', | |
| hue='segment_name', | |
| palette=segment_colors, | |
| alpha=0.7, | |
| ax=axes[1, 1] | |
| ) | |
| var_explained = pca.explained_variance_ratio_ * 100 | |
| axes[1, 1].set_title(f'PCA Visualization (Explained Variance: {var_explained[0]:.1f}%, {var_explained[1]:.1f}%)') | |
| # Adjust layout and save | |
| plt.tight_layout() | |
| output_path = os.path.join(self.output_dir, output_file) | |
| plt.savefig(output_path, dpi=300) | |
| plt.close() | |
| self.log(f"Visualization saved to {output_path}") | |
| return output_path | |
| def segment_customer(self, customer_data: Dict[str, Any]) -> Dict[str, Any]: | |
| """ | |
| Segment a single customer based on their RFM data | |
| """ | |
| self.log(f"Segmenting individual customer") | |
| if self.kmeans_model is None or self.scaler is None: | |
| self.log("No segmentation model. Please perform segmentation first.") | |
| return {"error": "No segmentation model available"} | |
| # Validate input | |
| required_fields = ['Recency', 'Frequency', 'Monetary'] | |
| for field in required_fields: | |
| if field not in customer_data: | |
| self.log(f"Missing required field: {field}") | |
| return {"error": f"Missing required field: {field}"} | |
| # Prepare the customer data for prediction | |
| customer_array = np.array([[ | |
| customer_data['Recency'], | |
| customer_data['Frequency'], | |
| customer_data['Monetary'] | |
| ]]) | |
| # Scale the data | |
| customer_scaled = self.scaler.transform(customer_array) | |
| # Predict the segment | |
| segment_id = int(self.kmeans_model.predict(customer_scaled)[0]) | |
| segment_id_str = str(segment_id) | |
| # Get segment details | |
| if segment_id_str in self.segments: | |
| segment = self.segments[segment_id_str] | |
| result = { | |
| "segment_id": segment_id, | |
| "segment_name": segment["segment_name"], | |
| "description": self.segment_descriptions.get(segment_id_str, ""), | |
| "recommended_strategy": self.segment_strategies.get(segment_id_str, "") | |
| } | |
| return result | |
| else: | |
| self.log(f"Unexpected segment ID: {segment_id}") | |
| return { | |
| "segment_id": segment_id, | |
| "segment_name": f"Cluster {segment_id}", | |
| "description": "Unknown segment", | |
| "recommended_strategy": "Further analysis needed" | |
| } |