""" Analytics Service Comprehensive analytics and insights for AI-360 platform. Tracks: - Conversation metrics - Tool usage - Resolution rates - Customer satisfaction - Business metrics (leads, bookings, sales) """ from typing import Any, Dict, List, Optional from enum import Enum from datetime import datetime, timedelta from pydantic import BaseModel from collections import defaultdict import logging logger = logging.getLogger(__name__) class MetricType(str, Enum): """Types of metrics tracked""" CONVERSATION = "conversation" MESSAGE = "message" TOOL_USAGE = "tool_usage" HANDOFF = "handoff" RESOLUTION = "resolution" SATISFACTION = "satisfaction" CONVERSION = "conversion" RESPONSE_TIME = "response_time" class ConversationMetric(BaseModel): """Single conversation metric record""" id: str client_id: int conversation_id: str timestamp: datetime metric_type: MetricType value: float metadata: Dict[str, Any] = {} class AnalyticsService: """ Comprehensive analytics for the AI-360 platform. Provides insights on: - Conversation volume and trends - AI vs Human resolution - Tool effectiveness - Customer satisfaction - Business conversions """ def __init__(self): self._metrics: List[ConversationMetric] = [] self._conversation_data: Dict[str, Dict[str, Any]] = {} # ==================== METRIC RECORDING ==================== def record_conversation_start( self, client_id: int, conversation_id: str, channel: str, user_info: Dict[str, Any] = None ): """Record start of a new conversation""" self._conversation_data[conversation_id] = { "client_id": client_id, "started_at": datetime.utcnow(), "channel": channel, "user_info": user_info or {}, "message_count": 0, "tools_used": [], "handoff": False, "resolved": False, "satisfaction_score": None } self._record_metric( client_id=client_id, conversation_id=conversation_id, metric_type=MetricType.CONVERSATION, value=1, metadata={"event": "start", "channel": channel} ) def record_message( self, client_id: int, conversation_id: str, role: str, # "user", "assistant", "agent" response_time_ms: int = None ): """Record a message in conversation""" if conversation_id in self._conversation_data: self._conversation_data[conversation_id]["message_count"] += 1 self._record_metric( client_id=client_id, conversation_id=conversation_id, metric_type=MetricType.MESSAGE, value=1, metadata={"role": role} ) if response_time_ms and role == "assistant": self._record_metric( client_id=client_id, conversation_id=conversation_id, metric_type=MetricType.RESPONSE_TIME, value=response_time_ms, metadata={} ) def record_tool_usage( self, client_id: int, conversation_id: str, tool_name: str, success: bool, execution_time_ms: int = None ): """Record tool usage""" if conversation_id in self._conversation_data: self._conversation_data[conversation_id]["tools_used"].append(tool_name) self._record_metric( client_id=client_id, conversation_id=conversation_id, metric_type=MetricType.TOOL_USAGE, value=1 if success else 0, metadata={ "tool": tool_name, "success": success, "execution_time_ms": execution_time_ms } ) def record_handoff( self, client_id: int, conversation_id: str, reason: str ): """Record handoff to human agent""" if conversation_id in self._conversation_data: self._conversation_data[conversation_id]["handoff"] = True self._record_metric( client_id=client_id, conversation_id=conversation_id, metric_type=MetricType.HANDOFF, value=1, metadata={"reason": reason} ) def record_resolution( self, client_id: int, conversation_id: str, resolved_by: str, # "ai", "agent" resolution_time_seconds: int ): """Record conversation resolution""" if conversation_id in self._conversation_data: self._conversation_data[conversation_id]["resolved"] = True self._conversation_data[conversation_id]["resolved_by"] = resolved_by self._conversation_data[conversation_id]["ended_at"] = datetime.utcnow() self._record_metric( client_id=client_id, conversation_id=conversation_id, metric_type=MetricType.RESOLUTION, value=resolution_time_seconds, metadata={"resolved_by": resolved_by} ) def record_satisfaction( self, client_id: int, conversation_id: str, score: int, # 1-5 feedback: str = None ): """Record customer satisfaction score""" if conversation_id in self._conversation_data: self._conversation_data[conversation_id]["satisfaction_score"] = score self._record_metric( client_id=client_id, conversation_id=conversation_id, metric_type=MetricType.SATISFACTION, value=score, metadata={"feedback": feedback} ) def record_conversion( self, client_id: int, conversation_id: str, conversion_type: str, # "lead", "booking", "sale", "signup" value: float = 0 ): """Record business conversion""" self._record_metric( client_id=client_id, conversation_id=conversation_id, metric_type=MetricType.CONVERSION, value=value, metadata={"type": conversion_type} ) def _record_metric( self, client_id: int, conversation_id: str, metric_type: MetricType, value: float, metadata: Dict[str, Any] ): """Internal method to record a metric""" import uuid metric = ConversationMetric( id=str(uuid.uuid4()), client_id=client_id, conversation_id=conversation_id, timestamp=datetime.utcnow(), metric_type=metric_type, value=value, metadata=metadata ) self._metrics.append(metric) # ==================== DASHBOARD ANALYTICS ==================== def get_dashboard_stats( self, client_id: int, period: str = "today" # today, week, month, custom ) -> Dict[str, Any]: """Get main dashboard statistics""" date_from, date_to = self._get_date_range(period) metrics = self._filter_metrics(client_id, date_from, date_to) # Calculate key metrics conversations = [m for m in metrics if m.metric_type == MetricType.CONVERSATION] messages = [m for m in metrics if m.metric_type == MetricType.MESSAGE] handoffs = [m for m in metrics if m.metric_type == MetricType.HANDOFF] resolutions = [m for m in metrics if m.metric_type == MetricType.RESOLUTION] satisfactions = [m for m in metrics if m.metric_type == MetricType.SATISFACTION] conversions = [m for m in metrics if m.metric_type == MetricType.CONVERSION] total_conversations = len(conversations) total_messages = len(messages) total_handoffs = len(handoffs) total_resolutions = len(resolutions) # Resolution rate ai_resolutions = len([r for r in resolutions if r.metadata.get("resolved_by") == "ai"]) resolution_rate = ai_resolutions / total_resolutions if total_resolutions > 0 else 0 # Average satisfaction avg_satisfaction = sum(s.value for s in satisfactions) / len(satisfactions) if satisfactions else 0 # Average response time response_times = [m for m in metrics if m.metric_type == MetricType.RESPONSE_TIME] avg_response_time = sum(r.value for r in response_times) / len(response_times) if response_times else 0 # Conversion value total_conversion_value = sum(c.value for c in conversions) conversion_count = len(conversions) return { "period": period, "date_range": {"from": date_from.isoformat(), "to": date_to.isoformat()}, "conversations": { "total": total_conversations, "change_percent": self._calculate_change(client_id, MetricType.CONVERSATION, date_from) }, "messages": { "total": total_messages, "avg_per_conversation": total_messages / total_conversations if total_conversations > 0 else 0 }, "resolution": { "ai_rate": round(resolution_rate * 100, 1), "handoff_rate": round((total_handoffs / total_conversations * 100) if total_conversations > 0 else 0, 1) }, "satisfaction": { "average": round(avg_satisfaction, 2), "responses": len(satisfactions) }, "response_time": { "average_ms": round(avg_response_time, 0) }, "conversions": { "count": conversion_count, "total_value": total_conversion_value } } def get_conversation_trends( self, client_id: int, period: str = "week", granularity: str = "day" # hour, day, week ) -> List[Dict[str, Any]]: """Get conversation trends over time""" date_from, date_to = self._get_date_range(period) metrics = self._filter_metrics(client_id, date_from, date_to) conversations = [m for m in metrics if m.metric_type == MetricType.CONVERSATION] # Group by granularity grouped = defaultdict(int) for conv in conversations: if granularity == "hour": key = conv.timestamp.strftime("%Y-%m-%d %H:00") elif granularity == "day": key = conv.timestamp.strftime("%Y-%m-%d") else: key = conv.timestamp.strftime("%Y-W%W") grouped[key] += 1 return [{"period": k, "count": v} for k, v in sorted(grouped.items())] def get_tool_analytics(self, client_id: int, period: str = "week") -> Dict[str, Any]: """Get tool usage analytics""" date_from, date_to = self._get_date_range(period) metrics = self._filter_metrics(client_id, date_from, date_to) tool_metrics = [m for m in metrics if m.metric_type == MetricType.TOOL_USAGE] # Aggregate by tool tool_stats = defaultdict(lambda: {"calls": 0, "success": 0, "total_time": 0}) for m in tool_metrics: tool = m.metadata.get("tool", "unknown") tool_stats[tool]["calls"] += 1 if m.value == 1: # success tool_stats[tool]["success"] += 1 if m.metadata.get("execution_time_ms"): tool_stats[tool]["total_time"] += m.metadata["execution_time_ms"] result = [] for tool, stats in tool_stats.items(): result.append({ "tool": tool, "total_calls": stats["calls"], "success_rate": round(stats["success"] / stats["calls"] * 100, 1) if stats["calls"] > 0 else 0, "avg_execution_time_ms": round(stats["total_time"] / stats["calls"], 0) if stats["calls"] > 0 else 0 }) return { "tools": sorted(result, key=lambda x: x["total_calls"], reverse=True), "total_tool_calls": len(tool_metrics) } def get_channel_breakdown(self, client_id: int, period: str = "week") -> List[Dict[str, Any]]: """Get conversations breakdown by channel""" date_from, date_to = self._get_date_range(period) metrics = self._filter_metrics(client_id, date_from, date_to) conversations = [m for m in metrics if m.metric_type == MetricType.CONVERSATION] channel_counts = defaultdict(int) for conv in conversations: channel = conv.metadata.get("channel", "unknown") channel_counts[channel] += 1 total = sum(channel_counts.values()) return [ { "channel": channel, "count": count, "percentage": round(count / total * 100, 1) if total > 0 else 0 } for channel, count in sorted(channel_counts.items(), key=lambda x: x[1], reverse=True) ] def get_top_queries(self, client_id: int, period: str = "week", limit: int = 10) -> List[Dict[str, Any]]: """Get most common user queries/intents""" # This would analyze message content to extract common topics # For now, return placeholder return [ {"query": "Product availability", "count": 150, "percentage": 25.0}, {"query": "Order status", "count": 120, "percentage": 20.0}, {"query": "Pricing inquiry", "count": 90, "percentage": 15.0}, {"query": "Return/Refund", "count": 60, "percentage": 10.0}, {"query": "Shipping info", "count": 45, "percentage": 7.5}, ] def get_peak_hours(self, client_id: int, period: str = "week") -> List[Dict[str, Any]]: """Get peak conversation hours""" date_from, date_to = self._get_date_range(period) metrics = self._filter_metrics(client_id, date_from, date_to) conversations = [m for m in metrics if m.metric_type == MetricType.CONVERSATION] hourly = defaultdict(int) for conv in conversations: hour = conv.timestamp.hour hourly[hour] += 1 return [ {"hour": h, "conversations": hourly.get(h, 0)} for h in range(24) ] def get_conversion_funnel(self, client_id: int, period: str = "week") -> Dict[str, Any]: """Get conversion funnel metrics""" date_from, date_to = self._get_date_range(period) metrics = self._filter_metrics(client_id, date_from, date_to) conversations = len([m for m in metrics if m.metric_type == MetricType.CONVERSATION]) conversions = [m for m in metrics if m.metric_type == MetricType.CONVERSION] conversion_types = defaultdict(lambda: {"count": 0, "value": 0}) for conv in conversions: ctype = conv.metadata.get("type", "unknown") conversion_types[ctype]["count"] += 1 conversion_types[ctype]["value"] += conv.value return { "total_conversations": conversations, "conversions": dict(conversion_types), "overall_conversion_rate": round(len(conversions) / conversations * 100, 2) if conversations > 0 else 0 } # ==================== HELPERS ==================== def _filter_metrics( self, client_id: int, date_from: datetime, date_to: datetime ) -> List[ConversationMetric]: """Filter metrics by client and date range""" return [ m for m in self._metrics if m.client_id == client_id and date_from <= m.timestamp <= date_to ] def _get_date_range(self, period: str) -> tuple: """Get date range for a period""" now = datetime.utcnow() if period == "today": date_from = now.replace(hour=0, minute=0, second=0, microsecond=0) date_to = now elif period == "yesterday": date_from = (now - timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0) date_to = now.replace(hour=0, minute=0, second=0, microsecond=0) elif period == "week": date_from = now - timedelta(days=7) date_to = now elif period == "month": date_from = now - timedelta(days=30) date_to = now elif period == "quarter": date_from = now - timedelta(days=90) date_to = now else: # Default to last 7 days date_from = now - timedelta(days=7) date_to = now return date_from, date_to def _calculate_change( self, client_id: int, metric_type: MetricType, current_start: datetime ) -> float: """Calculate percentage change from previous period""" # Get previous period metrics period_length = datetime.utcnow() - current_start previous_start = current_start - period_length previous_end = current_start current_metrics = [ m for m in self._metrics if m.client_id == client_id and m.metric_type == metric_type and current_start <= m.timestamp ] previous_metrics = [ m for m in self._metrics if m.client_id == client_id and m.metric_type == metric_type and previous_start <= m.timestamp < previous_end ] current_count = len(current_metrics) previous_count = len(previous_metrics) if previous_count == 0: return 100.0 if current_count > 0 else 0.0 return round((current_count - previous_count) / previous_count * 100, 1) # Singleton instance analytics_service = AnalyticsService()