Analytics & Intelligence Service

import asyncio
import pandas as pd
import numpy as np
from typing import Dict, List, Any, Optional, Union
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
import json
from sqlalchemy import create_engine
import plotly.graph_objects as go
import plotly.express as px
 
class AnalyticsType(Enum):
    DESCRIPTIVE = "descriptive"
    DIAGNOSTIC = "diagnostic"
    PREDICTIVE = "predictive"
    PRESCRIPTIVE = "prescriptive"
 
class ChartType(Enum):
    LINE = "line"
    BAR = "bar"
    PIE = "pie"
    SCATTER = "scatter"
    HEATMAP = "heatmap"
    HISTOGRAM = "histogram"
    BOX_PLOT = "box_plot"
    FUNNEL = "funnel"
 
@dataclass
class AnalyticsQuery:
    dataset: str
    metrics: List[str]
    dimensions: List[str] = field(default_factory=list)
    filters: Dict[str, Any] = field(default_factory=dict)
    date_range: Dict[str, str] = field(default_factory=dict)
    aggregation: str = "sum"
    group_by: List[str] = field(default_factory=list)
    order_by: str = ""
    limit: int = 1000
 
@dataclass
class AnalyticsResult:
    data: pd.DataFrame
    metadata: Dict[str, Any] = field(default_factory=dict)
    insights: List[str] = field(default_factory=list)
    visualizations: List[Dict[str, Any]] = field(default_factory=list)
    query_time_ms: float = 0
 
class AnalyticsEngine:
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.data_warehouse = self._initialize_data_warehouse(config)
        self.ml_models = {}
        self.insight_generator = InsightGenerator()
        self.visualization_engine = VisualizationEngine()
        
        # Initialize analytics modules
        self.descriptive_analytics = DescriptiveAnalytics(self.data_warehouse)
        self.predictive_analytics = PredictiveAnalytics(self.data_warehouse)
        self.diagnostic_analytics = DiagnosticAnalytics(self.data_warehouse)
    
    def _initialize_data_warehouse(self, config: Dict[str, Any]):
        """Initialize connection to data warehouse"""
        connection_string = config['data_warehouse']['connection_string']
        return create_engine(connection_string)
    
    async def execute_query(self, query: AnalyticsQuery) -> AnalyticsResult:
        """Execute analytics query and return results with insights"""
        start_time = datetime.utcnow()
        
        try:
            # Build and execute SQL query
            sql_query = self._build_sql_query(query)
            df = pd.read_sql(sql_query, self.data_warehouse)
            
            # Generate insights
            insights = await self.insight_generator.generate_insights(df, query)
            
            # Create visualizations
            visualizations = await self.visualization_engine.create_visualizations(df, query)
            
            # Calculate query time
            query_time = (datetime.utcnow() - start_time).total_seconds() * 1000
            
            return AnalyticsResult(
                data=df,
                metadata={
                    'row_count': len(df),
                    'columns': list(df.columns),
                    'query': sql_query
                },
                insights=insights,
                visualizations=visualizations,
                query_time_ms=query_time
            )
            
        except Exception as e:
            print(f"Analytics query error: {e}")
            return AnalyticsResult(
                data=pd.DataFrame(),
                metadata={'error': str(e)}
            )
    
    def _build_sql_query(self, query: AnalyticsQuery) -> str:
        """Build SQL query from analytics query"""
        
        # Select clause
        select_parts = []
        
        # Add dimensions
        for dim in query.dimensions:
            select_parts.append(dim)
        
        # Add metrics with aggregation
        for metric in query.metrics:
            if query.aggregation == "count":
                select_parts.append(f"COUNT({metric}) as {metric}")
            elif query.aggregation == "sum":
                select_parts.append(f"SUM({metric}) as {metric}")
            elif query.aggregation == "avg":
                select_parts.append(f"AVG({metric}) as {metric}")
            elif query.aggregation == "max":
                select_parts.append(f"MAX({metric}) as {metric}")
            elif query.aggregation == "min":
                select_parts.append(f"MIN({metric}) as {metric}")
            else:
                select_parts.append(metric)
        
        select_clause = "SELECT " + ", ".join(select_parts)
        
        # From clause
        from_clause = f"FROM {query.dataset}"
        
        # Where clause
        where_conditions = []
        
        # Add filters
        for field, value in query.filters.items():
            if isinstance(value, list):
                value_list = "', '".join(str(v) for v in value)
                where_conditions.append(f"{field} IN ('{value_list}')")
            elif isinstance(value, dict):
                # Range filter
                if 'gte' in value:
                    where_conditions.append(f"{field} >= '{value['gte']}'")
                if 'lte' in value:
                    where_conditions.append(f"{field} <= '{value['lte']}'")
            else:
                where_conditions.append(f"{field} = '{value}'")
        
        # Add date range filter
        if query.date_range:
            date_field = query.date_range.get('field', 'created_at')
            if 'start' in query.date_range:
                where_conditions.append(f"{date_field} >= '{query.date_range['start']}'")
            if 'end' in query.date_range:
                where_conditions.append(f"{date_field} <= '{query.date_range['end']}'")
        
        where_clause = ""
        if where_conditions:
            where_clause = "WHERE " + " AND ".join(where_conditions)
        
        # Group by clause
        group_by_clause = ""
        if query.group_by or query.dimensions:
            group_fields = query.group_by if query.group_by else query.dimensions
            group_by_clause = "GROUP BY " + ", ".join(group_fields)
        
        # Order by clause
        order_by_clause = ""
        if query.order_by:
            order_by_clause = f"ORDER BY {query.order_by}"
        
        # Limit clause
        limit_clause = f"LIMIT {query.limit}" if query.limit > 0 else ""
        
        # Combine all clauses
        sql_parts = [select_clause, from_clause, where_clause, group_by_clause, order_by_clause, limit_clause]
        return " ".join(part for part in sql_parts if part)
    
    async def create_dashboard(self, dashboard_config: Dict[str, Any]) -> Dict[str, Any]:
        """Create interactive dashboard with multiple visualizations"""
        
        dashboard = {
            'id': dashboard_config['id'],
            'title': dashboard_config['title'],
            'widgets': [],
            'filters': dashboard_config.get('filters', {}),
            'refresh_interval': dashboard_config.get('refresh_interval', 300)  # 5 minutes
        }
        
        # Process each widget
        for widget_config in dashboard_config['widgets']:
            widget = await self._create_widget(widget_config)
            dashboard['widgets'].append(widget)
        
        return dashboard
    
    async def _create_widget(self, widget_config: Dict[str, Any]) -> Dict[str, Any]:
        """Create individual dashboard widget"""
        
        # Execute analytics query for widget
        query = AnalyticsQuery(**widget_config['query'])
        result = await self.execute_query(query)
        
        # Create visualization
        chart_type = ChartType(widget_config.get('chart_type', 'line'))
        visualization = await self.visualization_engine.create_chart(
            result.data, chart_type, widget_config.get('chart_options', {})
        )
        
        return {
            'id': widget_config['id'],
            'title': widget_config['title'],
            'type': widget_config['type'],
            'data': result.data.to_dict('records'),
            'visualization': visualization,
            'insights': result.insights,
            'last_updated': datetime.utcnow().isoformat()
        }
 
class InsightGenerator:
    """Generate automated insights from analytics data"""
    
    def __init__(self):
        self.insight_rules = self._load_insight_rules()
    
    async def generate_insights(self, df: pd.DataFrame, query: AnalyticsQuery) -> List[str]:
        """Generate insights from analytics results"""
        insights = []
        
        if df.empty:
            return ["No data available for the selected criteria"]
        
        # Basic statistical insights
        insights.extend(self._generate_statistical_insights(df))
        
        # Trend insights
        insights.extend(self._generate_trend_insights(df))
        
        # Anomaly insights
        insights.extend(self._generate_anomaly_insights(df))
        
        # Comparative insights
        insights.extend(self._generate_comparative_insights(df))
        
        return insights[:5]  # Return top 5 insights
    
    def _generate_statistical_insights(self, df: pd.DataFrame) -> List[str]:
        """Generate statistical insights"""
        insights = []
        
        # Get numeric columns
        numeric_cols = df.select_dtypes(include=[np.number]).columns
        
        for col in numeric_cols:
            if col in df.columns:
                mean_val = df[col].mean()
                median_val = df[col].median()
                std_val = df[col].std()
                
                if std_val > mean_val * 0.5:  # High variability
                    insights.append(f"High variability detected in {col} (std: {std_val:.2f})")
                
                if abs(mean_val - median_val) > mean_val * 0.2:  # Skewed distribution
                    skew_direction = "right" if mean_val > median_val else "left"
                    insights.append(f"{col} shows {skew_direction}-skewed distribution")
        
        return insights
    
    def _generate_trend_insights(self, df: pd.DataFrame) -> List[str]:
        """Generate trend-based insights"""
        insights = []
        
        # Look for date columns
        date_cols = df.select_dtypes(include=['datetime64']).columns
        numeric_cols = df.select_dtypes(include=[np.number]).columns
        
        if len(date_cols) > 0 and len(numeric_cols) > 0:
            date_col = date_cols[0]
            
            for metric_col in numeric_cols:
                # Sort by date and calculate trend
                df_sorted = df.sort_values(date_col)
                
                if len(df_sorted) >= 3:
                    # Calculate simple linear trend
                    x = range(len(df_sorted))
                    y = df_sorted[metric_col].values
                    trend = np.polyfit(x, y, 1)[0]
                    
                    if abs(trend) > np.std(y) * 0.1:
                        direction = "increasing" if trend > 0 else "decreasing"
                        insights.append(f"{metric_col} is {direction} over time (trend: {trend:.2f})")
        
        return insights
    
    def _generate_anomaly_insights(self, df: pd.DataFrame) -> List[str]:
        """Generate anomaly detection insights"""
        insights = []
        
        numeric_cols = df.select_dtypes(include=[np.number]).columns
        
        for col in numeric_cols:
            if col in df.columns and len(df) > 5:
                # Simple outlier detection using IQR
                Q1 = df[col].quantile(0.25)
                Q3 = df[col].quantile(0.75)
                IQR = Q3 - Q1
                
                lower_bound = Q1 - 1.5 * IQR
                upper_bound = Q3 + 1.5 * IQR
                
                outliers = df[(df[col] < lower_bound) | (df[col] > upper_bound)]
                
                if len(outliers) > 0:
                    outlier_percentage = (len(outliers) / len(df)) * 100
                    insights.append(f"{outlier_percentage:.1f}% outliers detected in {col}")
        
        return insights
 
class VisualizationEngine:
    """Create interactive visualizations from analytics data"""
    
    async def create_visualizations(self, df: pd.DataFrame, 
                                   query: AnalyticsQuery) -> List[Dict[str, Any]]:
        """Create appropriate visualizations based on data"""
        visualizations = []
        
        if df.empty:
            return visualizations
        
        # Determine best chart types based on data
        chart_suggestions = self._suggest_chart_types(df, query)
        
        for chart_type in chart_suggestions[:3]:  # Create top 3 suggestions
            chart = await self.create_chart(df, chart_type)
            if chart:
                visualizations.append(chart)
        
        return visualizations
    
    def _suggest_chart_types(self, df: pd.DataFrame, 
                           query: AnalyticsQuery) -> List[ChartType]:
        """Suggest appropriate chart types based on data characteristics"""
        suggestions = []
        
        numeric_cols = df.select_dtypes(include=[np.number]).columns
        categorical_cols = df.select_dtypes(include=['object', 'category']).columns
        date_cols = df.select_dtypes(include=['datetime64']).columns
        
        # Time series data
        if len(date_cols) > 0 and len(numeric_cols) > 0:
            suggestions.append(ChartType.LINE)
        
        # Categorical data
        if len(categorical_cols) > 0 and len(numeric_cols) > 0:
            suggestions.extend([ChartType.BAR, ChartType.PIE])
        
        # Multiple numeric columns
        if len(numeric_cols) > 1:
            suggestions.extend([ChartType.SCATTER, ChartType.HEATMAP])
        
        # Single numeric column
        if len(numeric_cols) == 1:
            suggestions.extend([ChartType.HISTOGRAM, ChartType.BOX_PLOT])
        
        return suggestions
    
    async def create_chart(self, df: pd.DataFrame, chart_type: ChartType, 
                          options: Dict[str, Any] = None) -> Dict[str, Any]:
        """Create specific chart type"""
        options = options or {}
        
        try:
            if chart_type == ChartType.LINE:
                fig = self._create_line_chart(df, options)
            elif chart_type == ChartType.BAR:
                fig = self._create_bar_chart(df, options)
            elif chart_type == ChartType.PIE:
                fig = self._create_pie_chart(df, options)
            elif chart_type == ChartType.SCATTER:
                fig = self._create_scatter_chart(df, options)
            elif chart_type == ChartType.HEATMAP:
                fig = self._create_heatmap(df, options)
            elif chart_type == ChartType.HISTOGRAM:
                fig = self._create_histogram(df, options)
            else:
                return None
            
            return {
                'type': chart_type.value,
                'config': fig.to_dict(),
                'options': options
            }
            
        except Exception as e:
            print(f"Visualization error: {e}")
            return None
    
    def _create_line_chart(self, df: pd.DataFrame, options: Dict[str, Any]):
        """Create line chart"""
        x_col = options.get('x_column', df.columns[0])
        y_col = options.get('y_column', df.select_dtypes(include=[np.number]).columns[0])
        
        fig = px.line(df, x=x_col, y=y_col, 
                     title=options.get('title', f'{y_col} over {x_col}'))
        return fig
    
    def _create_bar_chart(self, df: pd.DataFrame, options: Dict[str, Any]):
        """Create bar chart"""
        x_col = options.get('x_column', df.columns[0])
        y_col = options.get('y_column', df.select_dtypes(include=[np.number]).columns[0])
        
        # Aggregate data if needed
        if len(df) > 20:  # Too many bars, aggregate
            df_agg = df.groupby(x_col)[y_col].sum().reset_index()
            df_agg = df_agg.sort_values(y_col, ascending=False).head(20)
        else:
            df_agg = df
        
        fig = px.bar(df_agg, x=x_col, y=y_col,
                    title=options.get('title', f'{y_col} by {x_col}'))
        return fig
 
# Usage example
analytics_config = {
    'data_warehouse': {
        'connection_string': 'postgresql://analytics:password@warehouse.sindhan.ai:5432/analytics'
    },
    'cache': {
        'type': 'redis',
        'host': 'redis.sindhan.ai',
        'port': 6379
    },
    'ml_models': {
        'anomaly_detection': 'isolation_forest',
        'forecasting': 'arima'
    }
}
 
# Initialize analytics engine
analytics = AnalyticsEngine(analytics_config)
 
# Execute analytics query
query = AnalyticsQuery(
    dataset='user_events',
    metrics=['session_duration', 'page_views', 'conversions'],
    dimensions=['user_segment', 'traffic_source'],
    date_range={'start': '2024-01-01', 'end': '2024-01-31'},
    aggregation='avg',
    group_by=['user_segment', 'traffic_source'],
    order_by='conversions DESC'
)
 
result = await analytics.execute_query(query)
print(f"Query returned {len(result.data)} rows")
print("Insights:", result.insights)
 
# Create dashboard
dashboard_config = {
    'id': 'user_analytics_dashboard',
    'title': 'User Analytics Dashboard',
    'widgets': [
        {
            'id': 'user_growth',
            'title': 'User Growth Trend',
            'type': 'chart',
            'chart_type': 'line',
            'query': {
                'dataset': 'users',
                'metrics': ['user_count'],
                'dimensions': ['signup_date'],
                'date_range': {'start': '2024-01-01', 'end': '2024-01-31'},
                'aggregation': 'count',
                'group_by': ['signup_date']
            }
        },
        {
            'id': 'conversion_funnel',
            'title': 'Conversion Funnel',
            'type': 'chart',
            'chart_type': 'funnel',
            'query': {
                'dataset': 'user_events',
                'metrics': ['event_count'],
                'dimensions': ['event_type'],
                'aggregation': 'count',
                'group_by': ['event_type']
            }
        }
    ]
}
 
dashboard = await analytics.create_dashboard(dashboard_config)
print(f"Created dashboard with {len(dashboard['widgets'])} widgets")