"""
Pattern 239: Data Analysis Agent

Purpose: Statistical data analysis with insights, visualizations, and reporting.
Performs descriptive statistics, correlation analysis, trend detection, and generates insights.
"""

from typing import Dict, Any, List, Optional, Union
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import statistics
import re


class AnalysisType(Enum):
    """Types of data analysis."""
    DESCRIPTIVE = "descriptive"  # Mean, median, std dev
    CORRELATION = "correlation"  # Relationships between variables
    TREND = "trend"  # Time series trends
    DISTRIBUTION = "distribution"  # Data distribution
    ANOMALY = "anomaly"  # Outlier detection
    COMPARATIVE = "comparative"  # Compare groups


@dataclass
class Dataset:
    """Dataset for analysis."""
    name: str
    data: List[Dict[str, Any]]
    columns: List[str] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def __post_init__(self):
        if not self.columns and self.data:
            self.columns = list(self.data[0].keys())


@dataclass
class AnalysisResult:
    """Analysis result."""
    dataset_name: str
    analysis_type: AnalysisType
    statistics: Dict[str, Any]
    insights: List[str]
    visualizations: List[str] = field(default_factory=list)
    recommendations: List[str] = field(default_factory=list)
    confidence: float = 0.0


class DataAnalysisAgent:
    """
    Data Analysis Agent - Statistical analysis and insights.
    
    Features:
    - Descriptive statistics
    - Correlation analysis
    - Trend detection
    - Anomaly detection
    - Distribution analysis
    - AI-powered insights
    - Recommendations
    - Visualization suggestions
    """
    
    def __init__(self, name: str):
        self.name = name
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.3)
        
        # Statistics
        self.statistics: Dict[str, Any] = {
            "total_analyses": 0,
            "analysis_types": {t.value: 0 for t in AnalysisType},
            "datasets_analyzed": 0,
            "insights_generated": 0
        }
    
    def analyze(self, dataset: Dataset, analysis_type: AnalysisType = AnalysisType.DESCRIPTIVE) -> AnalysisResult:
        """Perform analysis on dataset."""
        
        # Perform statistical analysis
        if analysis_type == AnalysisType.DESCRIPTIVE:
            stats = self._descriptive_statistics(dataset)
        elif analysis_type == AnalysisType.CORRELATION:
            stats = self._correlation_analysis(dataset)
        elif analysis_type == AnalysisType.TREND:
            stats = self._trend_analysis(dataset)
        elif analysis_type == AnalysisType.DISTRIBUTION:
            stats = self._distribution_analysis(dataset)
        elif analysis_type == AnalysisType.ANOMALY:
            stats = self._anomaly_detection(dataset)
        else:  # COMPARATIVE
            stats = self._comparative_analysis(dataset)
        
        # Generate insights using LLM
        insights = self._generate_insights(dataset, stats, analysis_type)
        
        # Generate recommendations
        recommendations = self._generate_recommendations(dataset, stats, insights)
        
        # Suggest visualizations
        visualizations = self._suggest_visualizations(dataset, analysis_type)
        
        # Calculate confidence
        confidence = self._calculate_confidence(dataset, stats)
        
        # Update statistics
        self.statistics["total_analyses"] += 1
        self.statistics["analysis_types"][analysis_type.value] += 1
        self.statistics["datasets_analyzed"] = self.statistics.get("datasets_analyzed", 0) + 1
        self.statistics["insights_generated"] += len(insights)
        
        return AnalysisResult(
            dataset_name=dataset.name,
            analysis_type=analysis_type,
            statistics=stats,
            insights=insights,
            visualizations=visualizations,
            recommendations=recommendations,
            confidence=confidence
        )
    
    def _descriptive_statistics(self, dataset: Dataset) -> Dict[str, Any]:
        """Calculate descriptive statistics."""
        stats = {}
        
        for column in dataset.columns:
            # Extract numeric values
            values = []
            for row in dataset.data:
                val = row.get(column)
                if isinstance(val, (int, float)):
                    values.append(val)
            
            if values:
                stats[column] = {
                    "count": len(values),
                    "mean": statistics.mean(values),
                    "median": statistics.median(values),
                    "std_dev": statistics.stdev(values) if len(values) > 1 else 0,
                    "min": min(values),
                    "max": max(values),
                    "range": max(values) - min(values)
                }
        
        return stats
    
    def _correlation_analysis(self, dataset: Dataset) -> Dict[str, Any]:
        """Analyze correlations between variables."""
        correlations = {}
        
        # Get numeric columns
        numeric_cols = []
        for col in dataset.columns:
            values = [row.get(col) for row in dataset.data if isinstance(row.get(col), (int, float))]
            if values:
                numeric_cols.append((col, values))
        
        # Calculate pairwise correlations
        for i in range(len(numeric_cols)):
            for j in range(i + 1, len(numeric_cols)):
                col1, vals1 = numeric_cols[i]
                col2, vals2 = numeric_cols[j]
                
                if len(vals1) == len(vals2) and len(vals1) > 1:
                    corr = self._pearson_correlation(vals1, vals2)
                    correlations[f"{col1}_vs_{col2}"] = corr
        
        return {"correlations": correlations}
    
    def _trend_analysis(self, dataset: Dataset) -> Dict[str, Any]:
        """Detect trends in data."""
        trends = {}
        
        for column in dataset.columns:
            values = [row.get(column) for row in dataset.data if isinstance(row.get(column), (int, float))]
            
            if len(values) >= 3:
                # Simple trend detection
                if values[-1] > values[0]:
                    direction = "increasing"
                    change_pct = ((values[-1] - values[0]) / values[0]) * 100 if values[0] != 0 else 0
                elif values[-1]  Dict[str, Any]:
        """Analyze data distribution."""
        distributions = {}
        
        for column in dataset.columns:
            values = [row.get(column) for row in dataset.data]
            
            # Frequency distribution
            freq = defaultdict(int)
            for val in values:
                freq[str(val)] += 1
            
            # Find mode
            mode = max(freq.items(), key=lambda x: x[1]) if freq else (None, 0)
            
            distributions[column] = {
                "unique_values": len(freq),
                "mode": mode[0],
                "mode_frequency": mode[1],
                "distribution": dict(freq)
            }
        
        return {"distributions": distributions}
    
    def _anomaly_detection(self, dataset: Dataset) -> Dict[str, Any]:
        """Detect anomalies/outliers."""
        anomalies = {}
        
        for column in dataset.columns:
            values = [row.get(column) for row in dataset.data if isinstance(row.get(column), (int, float))]
            
            if len(values) > 3:
                mean = statistics.mean(values)
                std_dev = statistics.stdev(values)
                
                # Values beyond 2 standard deviations
                outliers = [v for v in values if abs(v - mean) > 2 * std_dev]
                
                anomalies[column] = {
                    "outlier_count": len(outliers),
                    "outliers": outliers[:10],  # Limit to 10
                    "threshold": 2 * std_dev
                }
        
        return {"anomalies": anomalies}
    
    def _comparative_analysis(self, dataset: Dataset) -> Dict[str, Any]:
        """Compare groups or categories."""
        comparisons = {}
        
        # Group by categorical columns and compare numeric columns
        categorical_cols = []
        numeric_cols = []
        
        for col in dataset.columns:
            sample_vals = [row.get(col) for row in dataset.data[:10]]
            if all(isinstance(v, (int, float)) for v in sample_vals if v is not None):
                numeric_cols.append(col)
            else:
                categorical_cols.append(col)
        
        if categorical_cols and numeric_cols:
            cat_col = categorical_cols[0]
            num_col = numeric_cols[0]
            
            # Group by category
            groups = defaultdict(list)
            for row in dataset.data:
                groups[str(row.get(cat_col))].append(row.get(num_col))
            
            # Calculate stats for each group
            group_stats = {}
            for group, values in groups.items():
                numeric_values = [v for v in values if isinstance(v, (int, float))]
                if numeric_values:
                    group_stats[group] = {
                        "mean": statistics.mean(numeric_values),
                        "count": len(numeric_values)
                    }
            
            comparisons[f"{cat_col}_by_{num_col}"] = group_stats
        
        return {"comparisons": comparisons}
    
    def _pearson_correlation(self, x: List[float], y: List[float]) -> float:
        """Calculate Pearson correlation coefficient."""
        if len(x) != len(y) or len(x)  List[str]:
        """Generate insights using LLM."""
        stats_summary = str(stats)[:500]  # Limit length
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are a data analysis expert. Generate 3-5 clear, actionable insights "
                      f"from the {analysis_type.value} analysis results. "
                      f"Focus on patterns, trends, and important findings."),
            ("human", f"Dataset: {dataset.name}\n\nAnalysis Results:\n{stats_summary}\n\nInsights:")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        # Parse insights
        content = str(response.content)
        insights = [line.strip() for line in content.split('\n') if line.strip()]
        insights = [re.sub(r'^\d+[\.\)]\s*', '', ins) for ins in insights]
        
        return insights[:5]
    
    def _generate_recommendations(self, dataset: Dataset, stats: Dict[str, Any], insights: List[str]) -> List[str]:
        """Generate recommendations."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Based on the analysis insights, provide 2-3 actionable recommendations."),
            ("human", f"Insights:\n" + "\n".join(f"- {ins}" for ins in insights) + "\n\nRecommendations:")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        content = str(response.content)
        recommendations = [line.strip() for line in content.split('\n') if line.strip()]
        recommendations = [re.sub(r'^\d+[\.\)]\s*', '', rec) for rec in recommendations]
        
        return recommendations[:3]
    
    def _suggest_visualizations(self, dataset: Dataset, analysis_type: AnalysisType) -> List[str]:
        """Suggest appropriate visualizations."""
        viz_map = {
            AnalysisType.DESCRIPTIVE: ["Box plot", "Histogram", "Summary table"],
            AnalysisType.CORRELATION: ["Scatter plot", "Correlation matrix", "Heatmap"],
            AnalysisType.TREND: ["Line chart", "Area chart", "Trend lines"],
            AnalysisType.DISTRIBUTION: ["Histogram", "Bar chart", "Pie chart"],
            AnalysisType.ANOMALY: ["Scatter plot with outliers", "Box plot", "Time series"],
            AnalysisType.COMPARATIVE: ["Bar chart", "Grouped bar chart", "Box plot comparison"]
        }
        
        return viz_map.get(analysis_type, ["Bar chart", "Line chart"])
    
    def _calculate_confidence(self, dataset: Dataset, stats: Dict[str, Any]) -> float:
        """Calculate confidence in analysis."""
        sample_size = len(dataset.data)
        
        # Base confidence on sample size
        if sample_size >= 100:
            confidence = 0.9
        elif sample_size >= 30:
            confidence = 0.8
        elif sample_size >= 10:
            confidence = 0.7
        else:
            confidence = 0.5
        
        return confidence
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get analysis statistics."""
        return {
            "name": self.name,
            **self.statistics
        }


class DataAnalysisPattern:
    """
    Data Analysis Pattern - Manage data analysis agents.
    
    Coordinates multiple analysis agents for different purposes.
    """
    
    def __init__(self):
        self.agents: Dict[str, DataAnalysisAgent] = {}
    
    def create_agent(self, name: str) -> DataAnalysisAgent:
        """Create a new data analysis agent."""
        agent = DataAnalysisAgent(name)
        self.agents[name] = agent
        print(f"   Created data analysis agent: {name}")
        return agent
    
    def run(self, agent_name: str, dataset: Dataset, analysis_type: AnalysisType = AnalysisType.DESCRIPTIVE) -> Dict[str, Any]:
        """Perform analysis using specified agent."""
        if agent_name not in self.agents:
            return {"error": f"Agent {agent_name} not found"}
        
        agent = self.agents[agent_name]
        result = agent.analyze(dataset, analysis_type)
        
        print(f"\n=== Analysis by {agent_name} ===")
        print(f"Dataset: {result.dataset_name} ({len(dataset.data)} rows)")
        print(f"Analysis Type: {result.analysis_type.value}")
        print(f"Confidence: {result.confidence:.1%}\n")
        
        print("Key Statistics:")
        for key, value in list(result.statistics.items())[:3]:
            print(f"  {key}: {str(value)[:100]}")
        
        print(f"\nInsights ({len(result.insights)}):")
        for i, insight in enumerate(result.insights, 1):
            print(f"  {i}. {insight}")
        
        if result.recommendations:
            print(f"\nRecommendations:")
            for i, rec in enumerate(result.recommendations, 1):
                print(f"  {i}. {rec}")
        
        if result.visualizations:
            print(f"\nSuggested Visualizations: {', '.join(result.visualizations)}")
        
        return {
            "agent": agent_name,
            "analysis_type": result.analysis_type.value,
            "insights_count": len(result.insights),
            "confidence": result.confidence,
            "statistics": agent.get_statistics()
        }


# Example usage
if __name__ == "__main__":
    print("=== Data Analysis Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = DataAnalysisPattern()
    
    # Create agents
    print("1. Setting up data analysis agents...")
    
    analyst = pattern.create_agent("DataAnalyst")
    
    # Create sample datasets
    print("\n2. Creating sample datasets...")
    
    sales_data = Dataset(
        name="Monthly Sales",
        data=[
            {"month": "Jan", "sales": 45000, "costs": 30000, "customers": 120},
            {"month": "Feb", "sales": 52000, "costs": 32000, "customers": 145},
            {"month": "Mar", "sales": 48000, "costs": 31000, "customers": 135},
            {"month": "Apr", "sales": 61000, "costs": 35000, "customers": 170},
            {"month": "May", "sales": 58000, "costs": 34000, "customers": 160},
            {"month": "Jun", "sales": 65000, "costs": 36000, "customers": 180},
        ]
    )
    
    # Perform analyses
    print("\n3. Performing analyses...")
    
    # Descriptive statistics
    result1 = pattern.run("DataAnalyst", sales_data, AnalysisType.DESCRIPTIVE)
    
    print("\n" + "="*60 + "\n")
    
    # Trend analysis
    result2 = pattern.run("DataAnalyst", sales_data, AnalysisType.TREND)
    
    print("\n" + "="*60 + "\n")
    
    # Correlation analysis
    result3 = pattern.run("DataAnalyst", sales_data, AnalysisType.CORRELATION)
    
    # Print statistics
    print("\n" + "="*60)
    print("\n4. Analysis Statistics:")
    stats = result3["statistics"]
    for key, value in stats.items():
        print(f"   {key}: {value}")
    
    print("\n✓ Comprehensive data analysis with AI-powered insights complete!")

    

Agentic Architecture Pattern: Data Analysis Agent

Pattern 239: 239 Data Analysis Agent – Line-by-Line Explanation

Overview

Pattern 239: Data Analysis Agent

Purpose: Statistical data analysis with insights, visualizations, and reporting. Performs descriptive statistics, correlation analysis, trend detection, and generates insights.

Code Structure Analysis

File Information

• Pattern Number: 239
• Pattern Name: 239 Data Analysis Agent
• File Path: patterns\domain_specific\239_data_analysis_agent.py
• Total Lines: 483

Detailed Code Explanation

Module Docstring

"""
Pattern 239: Data Analysis Agent

Purpose: Statistical data analysis with insights, visualizations, and reporting.
Performs descriptive statistics, correlation analysis, trend detection, and generates insights.
"""

Explanation: This module docstring describes the purpose and key features of the 239 Data Analysis Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional, Union
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import statistics
import re  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Union: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

AnalysisType (Line 19)

class AnalysisType(Enum):
    DESCRIPTIVE
    CORRELATION
    TREND
    DISTRIBUTION
    ANOMALY
    COMPARATIVE

Purpose: Types of data analysis.

Values: – DESCRIPTIVE: Represents descriptive – CORRELATION: Represents correlation – TREND: Represents trend – DISTRIBUTION: Represents distribution – ANOMALY: Represents anomaly – COMPARATIVE: Represents comparative

Classes

AnalysisType (Line 19)

Description: Types of data analysis.

Dataset (Line 30)

Description: Dataset for analysis.

Methods:

__post_init__ (Line 37)

def __post_init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

AnalysisResult (Line 43)

Description: Analysis result.

DataAnalysisAgent (Line 54)

Description: Data Analysis Agent – Statistical analysis and insights.

Features: – Descriptive statistics – Correlation analysis – Trend detection – Anomaly detection – Distribution analysis – AI-powered insights – Recommendations – Visualization suggestions

Methods:

__init__ (Line 69)

def __init__(self, name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label

analyze (Line 81)

def analyze(self, dataset, analysis_type):
    """Perform analysis on dataset."""

Purpose: Perform analysis on dataset.

Parameters: – self: Instance reference – dataset: Parameter for dataset – analysis_type: Parameter for analysis_type

_descriptive_statistics (Line 126)

def _descriptive_statistics(self, dataset):
    """Calculate descriptive statistics."""

Purpose: Calculate descriptive statistics.

Parameters: – self: Instance reference – dataset: Parameter for dataset

_correlation_analysis (Line 151)

def _correlation_analysis(self, dataset):
    """Analyze correlations between variables."""

Purpose: Analyze correlations between variables.

Parameters: – self: Instance reference – dataset: Parameter for dataset

_trend_analysis (Line 174)

def _trend_analysis(self, dataset):
    """Detect trends in data."""

Purpose: Detect trends in data.

Parameters: – self: Instance reference – dataset: Parameter for dataset

_distribution_analysis (Line 202)

def _distribution_analysis(self, dataset):
    """Analyze data distribution."""

Purpose: Analyze data distribution.

Parameters: – self: Instance reference – dataset: Parameter for dataset

_anomaly_detection (Line 226)

def _anomaly_detection(self, dataset):
    """Detect anomalies/outliers."""

Purpose: Detect anomalies/outliers.

Parameters: – self: Instance reference – dataset: Parameter for dataset

_comparative_analysis (Line 248)

def _comparative_analysis(self, dataset):
    """Compare groups or categories."""

Purpose: Compare groups or categories.

Parameters: – self: Instance reference – dataset: Parameter for dataset

_pearson_correlation (Line 286)

def _pearson_correlation(self, x, y):
    """Calculate Pearson correlation coefficient."""

Purpose: Calculate Pearson correlation coefficient.

Parameters: – self: Instance reference – x: Parameter for x – y: Parameter for y

_generate_insights (Line 303)

def _generate_insights(self, dataset, stats, analysis_type):
    """Generate insights using LLM."""

Purpose: Generate insights using LLM.

Parameters: – self: Instance reference – dataset: Parameter for dataset – stats: Parameter for stats – analysis_type: Parameter for analysis_type

_generate_recommendations (Line 324)

def _generate_recommendations(self, dataset, stats, insights):
    """Generate recommendations."""

Purpose: Generate recommendations.

Parameters: – self: Instance reference – dataset: Parameter for dataset – stats: Parameter for stats – insights: Parameter for insights

_suggest_visualizations (Line 340)

def _suggest_visualizations(self, dataset, analysis_type):
    """Suggest appropriate visualizations."""

Purpose: Suggest appropriate visualizations.

Parameters: – self: Instance reference – dataset: Parameter for dataset – analysis_type: Parameter for analysis_type

_calculate_confidence (Line 353)

def _calculate_confidence(self, dataset, stats):
    """Calculate confidence in analysis."""

Purpose: Calculate confidence in analysis.

Parameters: – self: Instance reference – dataset: Parameter for dataset – stats: Parameter for stats

get_statistics (Line 369)

def get_statistics(self):
    """Get analysis statistics."""

Purpose: Get analysis statistics.

Parameters: – self: Instance reference

DataAnalysisPattern (Line 377)

Description: Data Analysis Pattern – Manage data analysis agents.

Coordinates multiple analysis agents for different purposes.

Methods:

__init__ (Line 384)

def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

create_agent (Line 387)

def create_agent(self, name):
    """Create a new data analysis agent."""

Purpose: Create a new data analysis agent.

Parameters: – self: Instance reference – name: Identifier or label

run (Line 394)

def run(self, agent_name, dataset, analysis_type):
    """Perform analysis using specified agent."""

Purpose: Perform analysis using specified agent.

Parameters: – self: Instance reference – agent_name: Parameter for agent_name – dataset: Parameter for dataset – analysis_type: Parameter for analysis_type

Design Patterns Used

• Strategy Pattern: Different execution strategies for various scenarios
• State Pattern: Managing different states in the workflow
• Chain of Responsibility: Sequential processing through workflow nodes
• Observer Pattern: Tracking and monitoring pattern execution
• Factory Pattern: Creating instances with proper configuration

Usage Example

# Initialize the pattern
pattern = AnalysisType(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

• LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
• Workflow Orchestration: LangGraph manages execution flow through nodes
• State Management: TypedDict defines workflow state structure
• Error Handling: Comprehensive error handling and fallback mechanisms
• Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities