""" Business Analytics Service Provides insights, KPIs, and data aggregation """ import logging from datetime import datetime, timedelta from typing import List, Dict, Any, Optional import numpy as np from collections import Counter logger = logging.getLogger(__name__) class AnalyticsService: """Business intelligence and analytics""" async def calculate_kpis( self, conversations: List[Dict[str, Any]], period_days: int = 30 ) -> Dict[str, Any]: """ Calculate key performance indicators Args: conversations: List of conversation data period_days: Analysis period in days Returns: KPI metrics """ if not conversations: return self._empty_kpis() # Basic counts total_conversations = len(conversations) total_messages = sum(c.get('message_count', 0) for c in conversations) # Resolution rates resolved = sum(1 for c in conversations if c.get('status') == 'resolved') escalated = sum(1 for c in conversations if c.get('status') == 'escalated') resolution_rate = (resolved / total_conversations * 100) if total_conversations > 0 else 0 escalation_rate = (escalated / total_conversations * 100) if total_conversations > 0 else 0 # Response times response_times = [ c.get('first_response_time', 0) for c in conversations if c.get('first_response_time') is not None ] avg_response_time = np.mean(response_times) if response_times else 0 # Conversation durations durations = [ c.get('duration_seconds', 0) for c in conversations if c.get('duration_seconds') is not None ] avg_duration = np.mean(durations) if durations else 0 # Satisfaction scores scores = [ c.get('satisfaction_score', 0) for c in conversations if c.get('satisfaction_score') is not None ] avg_satisfaction = np.mean(scores) if scores else 0 # Channel distribution channels = [c.get('channel', 'unknown') for c in conversations] channel_distribution = dict(Counter(channels)) # Daily averages daily_conversations = total_conversations / period_days if period_days > 0 else 0 daily_messages = total_messages / period_days if period_days > 0 else 0 return { "period_days": period_days, "total_conversations": total_conversations, "total_messages": total_messages, "daily_avg_conversations": round(daily_conversations, 1), "daily_avg_messages": round(daily_messages, 1), "resolution_rate": round(resolution_rate, 1), "escalation_rate": round(escalation_rate, 1), "avg_response_time_seconds": round(avg_response_time, 2), "avg_conversation_duration_seconds": round(avg_duration, 2), "avg_satisfaction_score": round(avg_satisfaction, 2), "channel_distribution": channel_distribution, "messages_per_conversation": round(total_messages / total_conversations, 1) if total_conversations > 0 else 0 } async def get_trends( self, data: List[Dict[str, Any]], metric: str = "count", granularity: str = "daily" ) -> Dict[str, Any]: """ Calculate trends for a metric Args: data: List of data points with 'date' and metric value metric: The metric to analyze granularity: daily, weekly, or monthly Returns: Trend analysis """ if not data: return {"trend": "stable", "change": 0, "data": []} # Sort by date sorted_data = sorted(data, key=lambda x: x.get('date', '')) values = [d.get(metric, 0) for d in sorted_data] if len(values) < 2: return {"trend": "stable", "change": 0, "data": sorted_data} # Calculate trend first_half = np.mean(values[:len(values)//2]) second_half = np.mean(values[len(values)//2:]) if first_half > 0: change = ((second_half - first_half) / first_half) * 100 else: change = 0 if change > 5: trend = "increasing" elif change < -5: trend = "decreasing" else: trend = "stable" # Calculate moving average window = min(7, len(values) // 2) if len(values) >= 4 else 1 if window > 1: moving_avg = np.convolve(values, np.ones(window)/window, mode='valid') moving_avg_data = [ {"date": sorted_data[i + window - 1]["date"], "moving_avg": round(float(v), 2)} for i, v in enumerate(moving_avg) ] else: moving_avg_data = [] return { "trend": trend, "change_percent": round(change, 1), "first_half_avg": round(first_half, 2), "second_half_avg": round(second_half, 2), "min_value": min(values), "max_value": max(values), "avg_value": round(np.mean(values), 2), "moving_average": moving_avg_data, "data_points": len(values) } async def analyze_user_behavior( self, user_sessions: List[Dict[str, Any]] ) -> Dict[str, Any]: """ Analyze user behavior patterns Args: user_sessions: List of user session data Returns: User behavior insights """ if not user_sessions: return {"insights": []} # Peak hours hours = [s.get('hour', 0) for s in user_sessions if 'hour' in s] hour_counts = Counter(hours) peak_hours = sorted(hour_counts.items(), key=lambda x: x[1], reverse=True)[:3] # Day of week patterns days = [s.get('day_of_week', 0) for s in user_sessions if 'day_of_week' in s] day_counts = Counter(days) day_names = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday'] busiest_days = [ {"day": day_names[d], "count": c} for d, c in sorted(day_counts.items(), key=lambda x: x[1], reverse=True)[:3] ] # Returning vs new users user_ids = [s.get('user_id') for s in user_sessions if s.get('user_id')] user_session_counts = Counter(user_ids) returning_users = sum(1 for count in user_session_counts.values() if count > 1) new_users = sum(1 for count in user_session_counts.values() if count == 1) # Session lengths session_lengths = [s.get('duration_seconds', 0) for s in user_sessions if 'duration_seconds' in s] avg_session_length = np.mean(session_lengths) if session_lengths else 0 # Bounce rate (sessions with only 1 message) single_message_sessions = sum( 1 for s in user_sessions if s.get('message_count', 0) == 1 ) bounce_rate = (single_message_sessions / len(user_sessions) * 100) if user_sessions else 0 return { "peak_hours": [ {"hour": f"{h:02d}:00", "sessions": c} for h, c in peak_hours ], "busiest_days": busiest_days, "unique_users": len(user_session_counts), "returning_users": returning_users, "new_users": new_users, "returning_user_rate": round(returning_users / len(user_session_counts) * 100, 1) if user_session_counts else 0, "avg_session_length_seconds": round(avg_session_length, 2), "bounce_rate": round(bounce_rate, 1) } async def get_topic_analysis( self, conversations: List[Dict[str, Any]] ) -> Dict[str, Any]: """ Analyze conversation topics Args: conversations: List of conversations with messages Returns: Topic analysis results """ # This would use NLP in production # For now, simple keyword extraction all_messages = [] for conv in conversations: messages = conv.get('messages', []) user_messages = [m.get('content', '') for m in messages if m.get('role') == 'user'] all_messages.extend(user_messages) if not all_messages: return {"topics": [], "keywords": []} # Simple word frequency from collections import Counter import re words = [] for msg in all_messages: # Extract words msg_words = re.findall(r'\b[a-zA-Z]{3,}\b', msg.lower()) words.extend(msg_words) # Remove common stop words stop_words = {'the', 'and', 'for', 'that', 'this', 'with', 'you', 'are', 'have', 'what', 'how', 'can', 'not', 'was', 'but'} words = [w for w in words if w not in stop_words] word_counts = Counter(words) top_keywords = word_counts.most_common(20) return { "top_keywords": [ {"keyword": word, "count": count} for word, count in top_keywords ], "total_messages_analyzed": len(all_messages) } async def generate_insights( self, kpis: Dict[str, Any], trends: Dict[str, Any] ) -> List[Dict[str, Any]]: """ Generate actionable insights Args: kpis: KPI metrics trends: Trend data Returns: List of insights """ insights = [] # Resolution rate insight resolution_rate = kpis.get('resolution_rate', 0) if resolution_rate < 70: insights.append({ "type": "warning", "category": "resolution", "title": "Low Resolution Rate", "description": f"Resolution rate is {resolution_rate}%, which is below the recommended 70%. Consider reviewing common unresolved queries.", "priority": "high" }) elif resolution_rate > 90: insights.append({ "type": "success", "category": "resolution", "title": "Excellent Resolution Rate", "description": f"Your chatbot is resolving {resolution_rate}% of conversations successfully!", "priority": "low" }) # Response time insight avg_response_time = kpis.get('avg_response_time_seconds', 0) if avg_response_time > 5: insights.append({ "type": "warning", "category": "performance", "title": "Slow Response Time", "description": f"Average response time is {avg_response_time:.1f}s. Consider optimizing your AI model or reducing context size.", "priority": "medium" }) # Volume trend insight trend = trends.get('trend', 'stable') change = trends.get('change_percent', 0) if trend == 'increasing' and change > 20: insights.append({ "type": "info", "category": "growth", "title": "Rapid Growth", "description": f"Message volume has increased by {change:.1f}%. Consider upgrading your plan to handle increased demand.", "priority": "medium" }) elif trend == 'decreasing' and change < -20: insights.append({ "type": "warning", "category": "engagement", "title": "Declining Engagement", "description": f"Message volume has decreased by {abs(change):.1f}%. Review your chatbot's performance and user feedback.", "priority": "high" }) # Escalation rate insight escalation_rate = kpis.get('escalation_rate', 0) if escalation_rate > 10: insights.append({ "type": "warning", "category": "quality", "title": "High Escalation Rate", "description": f"{escalation_rate}% of conversations are being escalated. Review and improve responses for common escalation triggers.", "priority": "high" }) return insights def _empty_kpis(self) -> Dict[str, Any]: return { "period_days": 0, "total_conversations": 0, "total_messages": 0, "daily_avg_conversations": 0, "daily_avg_messages": 0, "resolution_rate": 0, "escalation_rate": 0, "avg_response_time_seconds": 0, "avg_conversation_duration_seconds": 0, "avg_satisfaction_score": 0, "channel_distribution": {}, "messages_per_conversation": 0 } # Singleton instance analytics_service = AnalyticsService()