Social Media Analyzer

{
  "campaign_metrics": {
    "platform": "instagram",
    "total_posts": 3,
    "total_engagements": 1521,
    "total_reach": 18200,
    "total_impressions": 27700,
    "total_clicks": 430,
    "avg_engagement_rate": 8.36,
    "ctr": 1.55
  },
  "roi_metrics": {
    "total_spend": 500.0,
    "cost_per_engagement": 0.33,
    "cost_per_click": 1.16,
    "estimated_value": 3802.5,
    "roi_percentage": 660.5
  },
  "top_posts": [
    {
      "post_id": "post_002",
      "content_type": "video",
      "engagement_rate": 8.18,
      "likes": 587,
      "reach": 8900
    },
    {
      "post_id": "post_001",
      "content_type": "image",
      "engagement_rate": 8.27,
      "likes": 342,
      "reach": 5200
    },
    {
      "post_id": "post_003",
      "content_type": "carousel",
      "engagement_rate": 8.85,
      "likes": 298,
      "reach": 4100
    }
  ],
  "insights": {
    "overall_health": "excellent",
    "benchmark_comparison": {
      "engagement_status": "excellent",
      "engagement_benchmark": "1.22%",
      "engagement_actual": "8.36%",
      "ctr_status": "excellent",
      "ctr_benchmark": "0.22%",
      "ctr_actual": "1.55%"
    },
    "recommendations": [
      "Excellent ROI (660.5%)! Consider: (1) Scaling this campaign with increased budget, (2) Replicating successful elements to other campaigns, (3) Testing similar audiences"
    ],
    "key_strengths": [
      "Strong audience engagement",
      "Excellent return on investment",
      "High click-through rate"
    ]
  }
}

{
  "platform": "instagram",
  "total_spend": 500,
  "posts": [
    {
      "post_id": "post_001",
      "content_type": "image",
      "likes": 342,
      "comments": 28,
      "shares": 15,
      "saves": 45,
      "reach": 5200,
      "impressions": 8500,
      "clicks": 120,
      "posted_at": "2025-10-15T14:30:00Z"
    },
    {
      "post_id": "post_002",
      "content_type": "video",
      "likes": 587,
      "comments": 42,
      "shares": 31,
      "saves": 68,
      "reach": 8900,
      "impressions": 12400,
      "clicks": 215,
      "posted_at": "2025-10-16T18:45:00Z"
    },
    {
      "post_id": "post_003",
      "content_type": "carousel",
      "likes": 298,
      "comments": 19,
      "shares": 12,
      "saves": 34,
      "reach": 4100,
      "impressions": 6800,
      "clicks": 95,
      "posted_at": "2025-10-18T12:15:00Z"
    }
  ]
}

"""
Performance analysis and recommendation module.
Provides insights and optimization recommendations.
"""

from typing import Dict, List, Any


class PerformanceAnalyzer:
    """Analyze campaign performance and generate recommendations."""

    # Industry benchmark ranges
    BENCHMARKS = {
        'facebook': {'engagement_rate': 0.09, 'ctr': 0.90},
        'instagram': {'engagement_rate': 1.22, 'ctr': 0.22},
        'twitter': {'engagement_rate': 0.045, 'ctr': 1.64},
        'linkedin': {'engagement_rate': 0.54, 'ctr': 0.39},
        'tiktok': {'engagement_rate': 5.96, 'ctr': 1.00}
    }

    def __init__(self, campaign_metrics: Dict[str, Any], roi_metrics: Dict[str, Any]):
        """
        Initialize with calculated metrics.

        Args:
            campaign_metrics: Dictionary of campaign performance metrics
            roi_metrics: Dictionary of ROI and cost metrics
        """
        self.campaign_metrics = campaign_metrics
        self.roi_metrics = roi_metrics
        self.platform = campaign_metrics.get('platform', 'unknown').lower()

    def benchmark_performance(self) -> Dict[str, str]:
        """Compare metrics against industry benchmarks."""
        benchmarks = self.BENCHMARKS.get(self.platform, {})

        if not benchmarks:
            return {'status': 'no_benchmark_available'}

        engagement_rate = self.campaign_metrics.get('avg_engagement_rate', 0)
        ctr = self.campaign_metrics.get('ctr', 0)

        benchmark_engagement = benchmarks.get('engagement_rate', 0)
        benchmark_ctr = benchmarks.get('ctr', 0)

        engagement_status = 'excellent' if engagement_rate >= benchmark_engagement * 1.5 else \
                          'good' if engagement_rate >= benchmark_engagement else \
                          'below_average'

        ctr_status = 'excellent' if ctr >= benchmark_ctr * 1.5 else \
                    'good' if ctr >= benchmark_ctr else \
                    'below_average'

        return {
            'engagement_status': engagement_status,
            'engagement_benchmark': f"{benchmark_engagement}%",
            'engagement_actual': f"{engagement_rate:.2f}%",
            'ctr_status': ctr_status,
            'ctr_benchmark': f"{benchmark_ctr}%",
            'ctr_actual': f"{ctr:.2f}%"
        }

    def generate_recommendations(self) -> List[str]:
        """Generate actionable recommendations based on performance."""
        recommendations = []

        # Analyze engagement rate
        engagement_rate = self.campaign_metrics.get('avg_engagement_rate', 0)
        if engagement_rate < 1.0:
            recommendations.append(
                "Low engagement rate detected. Consider: (1) Posting during peak audience activity times, "
                "(2) Using more interactive content formats (polls, questions), "
                "(3) Improving visual quality of posts"
            )

        # Analyze CTR
        ctr = self.campaign_metrics.get('ctr', 0)
        if ctr < 0.5:
            recommendations.append(
                "Click-through rate is below average. Try: (1) Stronger call-to-action statements, "
                "(2) More compelling headlines, (3) Better alignment between content and audience interests"
            )

        # Analyze cost efficiency
        cpc = self.roi_metrics.get('cost_per_click', 0)
        if cpc > 1.00:
            recommendations.append(
                f"Cost per click (${cpc:.2f}) is high. Optimize by: (1) Refining audience targeting, "
                "(2) Testing different ad creatives, (3) Adjusting bidding strategy"
            )

        # Analyze ROI
        roi = self.roi_metrics.get('roi_percentage', 0)
        if roi < 100:
            recommendations.append(
                f"ROI ({roi:.1f}%) needs improvement. Focus on: (1) Conversion rate optimization, "
                "(2) Reducing cost per acquisition, (3) Better audience segmentation"
            )
        elif roi > 200:
            recommendations.append(
                f"Excellent ROI ({roi:.1f}%)! Consider: (1) Scaling this campaign with increased budget, "
                "(2) Replicating successful elements to other campaigns, (3) Testing similar audiences"
            )

        # Post frequency analysis
        total_posts = self.campaign_metrics.get('total_posts', 0)
        if total_posts < 10:
            recommendations.append(
                "Limited post volume may affect insights accuracy. Consider increasing posting frequency "
                "to gather more performance data"
            )

        # Default positive recommendation if performing well
        if not recommendations:
            recommendations.append(
                "Campaign is performing well across all metrics. Continue current strategy while "
                "testing minor variations to optimize further"
            )

        return recommendations

    def generate_insights(self) -> Dict[str, Any]:
        """Generate comprehensive performance insights."""
        benchmark_results = self.benchmark_performance()
        recommendations = self.generate_recommendations()

        # Determine overall campaign health
        engagement_status = benchmark_results.get('engagement_status', 'unknown')
        ctr_status = benchmark_results.get('ctr_status', 'unknown')

        if engagement_status == 'excellent' and ctr_status == 'excellent':
            overall_health = 'excellent'
        elif engagement_status in ['good', 'excellent'] and ctr_status in ['good', 'excellent']:
            overall_health = 'good'
        else:
            overall_health = 'needs_improvement'

        return {
            'overall_health': overall_health,
            'benchmark_comparison': benchmark_results,
            'recommendations': recommendations,
            'key_strengths': self._identify_strengths(),
            'areas_for_improvement': self._identify_weaknesses()
        }

    def _identify_strengths(self) -> List[str]:
        """Identify campaign strengths."""
        strengths = []

        engagement_rate = self.campaign_metrics.get('avg_engagement_rate', 0)
        if engagement_rate > 1.0:
            strengths.append("Strong audience engagement")

        roi = self.roi_metrics.get('roi_percentage', 0)
        if roi > 150:
            strengths.append("Excellent return on investment")

        ctr = self.campaign_metrics.get('ctr', 0)
        if ctr > 1.0:
            strengths.append("High click-through rate")

        return strengths if strengths else ["Campaign shows baseline performance"]

    def _identify_weaknesses(self) -> List[str]:
        """Identify areas needing improvement."""
        weaknesses = []

        engagement_rate = self.campaign_metrics.get('avg_engagement_rate', 0)
        if engagement_rate < 0.5:
            weaknesses.append("Low engagement rate - content may not resonate with audience")

        roi = self.roi_metrics.get('roi_percentage', 0)
        if roi < 50:
            weaknesses.append("ROI below target - need to improve conversion or reduce costs")

        cpc = self.roi_metrics.get('cost_per_click', 0)
        if cpc > 2.00:
            weaknesses.append("High cost per click - targeting or bidding needs optimization")

        return weaknesses if weaknesses else ["No critical weaknesses identified"]

"""
Social media metrics calculation module.
Provides functions to calculate engagement, reach, and ROI metrics.
"""

from typing import Dict, List, Any, Optional
from datetime import datetime


class SocialMediaMetricsCalculator:
    """Calculate social media performance metrics."""

    def __init__(self, campaign_data: Dict[str, Any]):
        """
        Initialize with campaign data.

        Args:
            campaign_data: Dictionary containing platform, posts, and cost data
        """
        self.platform = campaign_data.get('platform', 'unknown')
        self.posts = campaign_data.get('posts', [])
        self.total_spend = campaign_data.get('total_spend', 0)
        self.metrics = {}

    def safe_divide(self, numerator: float, denominator: float, default: float = 0.0) -> float:
        """Safely divide two numbers, returning default if denominator is zero."""
        if denominator == 0:
            return default
        return numerator / denominator

    def calculate_engagement_rate(self, post: Dict[str, Any]) -> float:
        """
        Calculate engagement rate for a post.

        Args:
            post: Dictionary with likes, comments, shares, and reach

        Returns:
            Engagement rate as percentage
        """
        likes = post.get('likes', 0)
        comments = post.get('comments', 0)
        shares = post.get('shares', 0)
        saves = post.get('saves', 0)
        reach = post.get('reach', 0)

        total_engagements = likes + comments + shares + saves
        engagement_rate = self.safe_divide(total_engagements, reach) * 100

        return round(engagement_rate, 2)

    def calculate_ctr(self, clicks: int, impressions: int) -> float:
        """
        Calculate click-through rate.

        Args:
            clicks: Number of clicks
            impressions: Number of impressions

        Returns:
            CTR as percentage
        """
        ctr = self.safe_divide(clicks, impressions) * 100
        return round(ctr, 2)

    def calculate_campaign_metrics(self) -> Dict[str, Any]:
        """Calculate overall campaign metrics."""
        total_likes = sum(post.get('likes', 0) for post in self.posts)
        total_comments = sum(post.get('comments', 0) for post in self.posts)
        total_shares = sum(post.get('shares', 0) for post in self.posts)
        total_reach = sum(post.get('reach', 0) for post in self.posts)
        total_impressions = sum(post.get('impressions', 0) for post in self.posts)
        total_clicks = sum(post.get('clicks', 0) for post in self.posts)

        total_engagements = total_likes + total_comments + total_shares

        return {
            'platform': self.platform,
            'total_posts': len(self.posts),
            'total_engagements': total_engagements,
            'total_reach': total_reach,
            'total_impressions': total_impressions,
            'total_clicks': total_clicks,
            'avg_engagement_rate': self.safe_divide(total_engagements, total_reach) * 100,
            'ctr': self.calculate_ctr(total_clicks, total_impressions)
        }

    def calculate_roi_metrics(self) -> Dict[str, float]:
        """Calculate ROI and cost efficiency metrics."""
        campaign_metrics = self.calculate_campaign_metrics()

        total_engagements = campaign_metrics['total_engagements']
        total_clicks = campaign_metrics['total_clicks']

        cost_per_engagement = self.safe_divide(self.total_spend, total_engagements)
        cost_per_click = self.safe_divide(self.total_spend, total_clicks)

        # Assuming average value per engagement (can be customized)
        avg_value_per_engagement = 2.50  # Example: $2.50 value per engagement
        total_value = total_engagements * avg_value_per_engagement
        roi_percentage = self.safe_divide(total_value - self.total_spend, self.total_spend) * 100

        return {
            'total_spend': round(self.total_spend, 2),
            'cost_per_engagement': round(cost_per_engagement, 2),
            'cost_per_click': round(cost_per_click, 2),
            'estimated_value': round(total_value, 2),
            'roi_percentage': round(roi_percentage, 2)
        }

    def identify_top_posts(self, metric: str = 'engagement_rate', limit: int = 5) -> List[Dict[str, Any]]:
        """
        Identify top performing posts.

        Args:
            metric: Metric to sort by (engagement_rate, likes, shares, etc.)
            limit: Number of top posts to return

        Returns:
            List of top performing posts with metrics
        """
        posts_with_metrics = []

        for post in self.posts:
            post_copy = post.copy()
            post_copy['engagement_rate'] = self.calculate_engagement_rate(post)
            posts_with_metrics.append(post_copy)

        # Sort by specified metric
        if metric == 'engagement_rate':
            sorted_posts = sorted(posts_with_metrics,
                                key=lambda x: x['engagement_rate'],
                                reverse=True)
        else:
            sorted_posts = sorted(posts_with_metrics,
                                key=lambda x: x.get(metric, 0),
                                reverse=True)

        return sorted_posts[:limit]

    def analyze_all(self) -> Dict[str, Any]:
        """Run complete analysis."""
        return {
            'campaign_metrics': self.calculate_campaign_metrics(),
            'roi_metrics': self.calculate_roi_metrics(),
            'top_posts': self.identify_top_posts()
        }