""" KPI Generator Tool - Generate business KPIs from data """ import logging from typing import Dict, Any, List import sys import os # Add parent directory to path for imports sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from utils.helpers import parse_json_safe, safe_divide logger = logging.getLogger(__name__) def generate_kpis(data: str, metrics: List[str] = None) -> Dict[str, Any]: """ Generate KPI report from business data. Args: data: JSON string containing business data metrics: List of metrics to calculate (revenue, growth, efficiency, etc.) Returns: Dictionary with calculated KPIs and insights """ try: import json # Parse input data try: business_data = json.loads(data) except json.JSONDecodeError as e: raise ValueError(f"Invalid JSON data: {e}") if metrics is None: metrics = ["revenue", "growth", "efficiency"] kpis = {} trends = [] # Calculate different KPIs based on requested metrics for metric in metrics: if metric == "revenue": revenue_kpis = _calculate_revenue_kpis(business_data) kpis.update(revenue_kpis) elif metric == "growth": growth_kpis = _calculate_growth_kpis(business_data) kpis.update(growth_kpis) elif metric == "efficiency": efficiency_kpis = _calculate_efficiency_kpis(business_data) kpis.update(efficiency_kpis) elif metric == "customer": customer_kpis = _calculate_customer_kpis(business_data) kpis.update(customer_kpis) elif metric == "operational": operational_kpis = _calculate_operational_kpis(business_data) kpis.update(operational_kpis) # Generate trends trends = _identify_trends(kpis, business_data) # Generate executive summary summary = _generate_summary(kpis, trends) return { "kpis": kpis, "summary": summary, "trends": trends, "metrics_analyzed": metrics, "data_points": len(business_data) if isinstance(business_data, list) else len(business_data.keys()) } except Exception as e: logger.error(f"Error generating KPIs: {e}") raise def _calculate_revenue_kpis(data: Dict[str, Any]) -> Dict[str, Any]: """Calculate revenue-related KPIs""" kpis = {} try: # Total Revenue if "revenue" in data: if isinstance(data["revenue"], list): kpis["total_revenue"] = sum(data["revenue"]) kpis["average_revenue"] = sum(data["revenue"]) / len(data["revenue"]) kpis["min_revenue"] = min(data["revenue"]) kpis["max_revenue"] = max(data["revenue"]) else: kpis["total_revenue"] = data["revenue"] # Revenue per customer if "revenue" in data and "customers" in data: revenue = data["revenue"] if not isinstance(data["revenue"], list) else sum(data["revenue"]) customers = data["customers"] if not isinstance(data["customers"], list) else sum(data["customers"]) kpis["revenue_per_customer"] = safe_divide(revenue, customers) # Profit margin if "revenue" in data and "costs" in data: revenue = data["revenue"] if not isinstance(data["revenue"], list) else sum(data["revenue"]) costs = data["costs"] if not isinstance(data["costs"], list) else sum(data["costs"]) profit = revenue - costs kpis["profit"] = profit kpis["profit_margin_percent"] = safe_divide(profit * 100, revenue) except Exception as e: logger.warning(f"Error calculating revenue KPIs: {e}") return kpis def _calculate_growth_kpis(data: Dict[str, Any]) -> Dict[str, Any]: """Calculate growth-related KPIs""" kpis = {} try: # Year-over-year growth if "current_revenue" in data and "previous_revenue" in data: growth = data["current_revenue"] - data["previous_revenue"] growth_rate = safe_divide(growth * 100, data["previous_revenue"]) kpis["revenue_growth"] = growth kpis["revenue_growth_rate_percent"] = growth_rate # Customer growth if "current_customers" in data and "previous_customers" in data: customer_growth = data["current_customers"] - data["previous_customers"] customer_growth_rate = safe_divide(customer_growth * 100, data["previous_customers"]) kpis["customer_growth"] = customer_growth kpis["customer_growth_rate_percent"] = customer_growth_rate # Monthly growth rate (if time series data provided) if "monthly_revenue" in data and isinstance(data["monthly_revenue"], list): revenues = data["monthly_revenue"] if len(revenues) >= 2: recent_growth = safe_divide((revenues[-1] - revenues[-2]) * 100, revenues[-2]) kpis["recent_monthly_growth_percent"] = recent_growth except Exception as e: logger.warning(f"Error calculating growth KPIs: {e}") return kpis def _calculate_efficiency_kpis(data: Dict[str, Any]) -> Dict[str, Any]: """Calculate efficiency-related KPIs""" kpis = {} try: # Cost per acquisition if "marketing_costs" in data and "new_customers" in data: kpis["cost_per_acquisition"] = safe_divide(data["marketing_costs"], data["new_customers"]) # Operational efficiency if "revenue" in data and "operational_costs" in data: revenue = data["revenue"] if not isinstance(data["revenue"], list) else sum(data["revenue"]) kpis["operational_efficiency_ratio"] = safe_divide(revenue, data["operational_costs"]) # Employee productivity if "revenue" in data and "employees" in data: revenue = data["revenue"] if not isinstance(data["revenue"], list) else sum(data["revenue"]) kpis["revenue_per_employee"] = safe_divide(revenue, data["employees"]) # ROI if "revenue" in data and "investment" in data: revenue = data["revenue"] if not isinstance(data["revenue"], list) else sum(data["revenue"]) roi = safe_divide((revenue - data["investment"]) * 100, data["investment"]) kpis["roi_percent"] = roi except Exception as e: logger.warning(f"Error calculating efficiency KPIs: {e}") return kpis def _calculate_customer_kpis(data: Dict[str, Any]) -> Dict[str, Any]: """Calculate customer-related KPIs""" kpis = {} try: # Customer lifetime value if "average_purchase_value" in data and "purchase_frequency" in data and "customer_lifespan" in data: clv = data["average_purchase_value"] * data["purchase_frequency"] * data["customer_lifespan"] kpis["customer_lifetime_value"] = clv # Churn rate if "churned_customers" in data and "total_customers" in data: kpis["churn_rate_percent"] = safe_divide(data["churned_customers"] * 100, data["total_customers"]) # Retention rate if "retained_customers" in data and "total_customers" in data: kpis["retention_rate_percent"] = safe_divide(data["retained_customers"] * 100, data["total_customers"]) # Net Promoter Score (if provided) if "nps_score" in data: kpis["net_promoter_score"] = data["nps_score"] except Exception as e: logger.warning(f"Error calculating customer KPIs: {e}") return kpis def _calculate_operational_kpis(data: Dict[str, Any]) -> Dict[str, Any]: """Calculate operational KPIs""" kpis = {} try: # Inventory turnover if "cost_of_goods_sold" in data and "average_inventory" in data: kpis["inventory_turnover"] = safe_divide(data["cost_of_goods_sold"], data["average_inventory"]) # Order fulfillment rate if "orders_fulfilled" in data and "total_orders" in data: kpis["fulfillment_rate_percent"] = safe_divide(data["orders_fulfilled"] * 100, data["total_orders"]) # Average response time if "total_response_time" in data and "ticket_count" in data: kpis["average_response_time"] = safe_divide(data["total_response_time"], data["ticket_count"]) except Exception as e: logger.warning(f"Error calculating operational KPIs: {e}") return kpis def _identify_trends(kpis: Dict[str, Any], data: Dict[str, Any]) -> List[str]: """Identify key trends from KPIs""" trends = [] try: # Check growth trends if "revenue_growth_rate_percent" in kpis: rate = kpis["revenue_growth_rate_percent"] if rate > 20: trends.append(f"Strong revenue growth of {rate:.1f}%") elif rate > 0: trends.append(f"Positive revenue growth of {rate:.1f}%") else: trends.append(f"Revenue decline of {abs(rate):.1f}%") # Check profitability if "profit_margin_percent" in kpis: margin = kpis["profit_margin_percent"] if margin > 20: trends.append(f"Healthy profit margin at {margin:.1f}%") elif margin > 0: trends.append(f"Modest profit margin at {margin:.1f}%") else: trends.append(f"Operating at a loss with {abs(margin):.1f}% negative margin") # Check efficiency if "roi_percent" in kpis: roi = kpis["roi_percent"] if roi > 100: trends.append(f"Excellent ROI of {roi:.1f}%") elif roi > 0: trends.append(f"Positive ROI of {roi:.1f}%") # Check customer metrics if "churn_rate_percent" in kpis: churn = kpis["churn_rate_percent"] if churn > 10: trends.append(f"High customer churn rate of {churn:.1f}%") else: trends.append(f"Healthy churn rate of {churn:.1f}%") except Exception as e: logger.warning(f"Error identifying trends: {e}") return trends if trends else ["Insufficient data for trend analysis"] def _generate_summary(kpis: Dict[str, Any], trends: List[str]) -> str: """Generate executive summary""" summary_parts = [] summary_parts.append("Executive KPI Summary:") summary_parts.append(f"- Analyzed {len(kpis)} key performance indicators") if trends: summary_parts.append("- Key insights:") for trend in trends[:3]: # Top 3 trends summary_parts.append(f" • {trend}") return "\n".join(summary_parts)