import pandas as pd import numpy as np from typing import Dict, List, Any import streamlit as st class ProcurementAgent: def __init__(self): self.insights = [] def analyze_spend_trends(self, df: pd.DataFrame) -> Dict[str, Any]: """Analyze spending trends and patterns""" try: # Monthly spend analysis df['PO_Date'] = pd.to_datetime(df['PO_Date']) monthly_spend = df.groupby(df['PO_Date'].dt.to_period('M'))['Total_Value'].sum() # Calculate trend if len(monthly_spend) > 1: trend = "increasing" if monthly_spend.iloc[-1] > monthly_spend.iloc[-2] else "decreasing" else: trend = "stable" # Top categories category_spend = df.groupby('Category')['Total_Value'].sum().sort_values(ascending=False) insights = { 'total_spend': df['Total_Value'].sum(), 'monthly_trend': trend, 'top_category': category_spend.index[0], 'top_category_spend': category_spend.iloc[0], 'avg_po_value': df['Total_Value'].mean(), 'recommendations': self._generate_spend_recommendations(df) } return insights except Exception as e: st.error(f"Error in spend analysis: {str(e)}") return {} def analyze_supplier_performance(self, df: pd.DataFrame) -> Dict[str, Any]: """Analyze supplier performance metrics""" try: supplier_metrics = df.groupby('Supplier').agg({ 'Total_Value': 'sum', 'Delivery_Performance': 'mean', 'PO_Number': 'count' }).round(2) # Best and worst performers best_supplier = supplier_metrics.loc[supplier_metrics['Delivery_Performance'].idxmax()] worst_supplier = supplier_metrics.loc[supplier_metrics['Delivery_Performance'].idxmin()] insights = { 'best_performer': { 'name': best_supplier.name, 'performance': best_supplier['Delivery_Performance'] }, 'worst_performer': { 'name': worst_supplier.name, 'performance': worst_supplier['Delivery_Performance'] }, 'recommendations': self._generate_supplier_recommendations(supplier_metrics) } return insights except Exception as e: st.error(f"Error in supplier analysis: {str(e)}") return {} def detect_anomalies(self, df: pd.DataFrame) -> List[Dict[str, Any]]: """Detect procurement anomalies""" anomalies = [] try: # High value orders threshold = df['Total_Value'].quantile(0.95) high_value_orders = df[df['Total_Value'] > threshold] for _, order in high_value_orders.iterrows(): anomalies.append({ 'type': 'High Value Order', 'po_number': order['PO_Number'], 'value': order['Total_Value'], 'supplier': order['Supplier'], 'risk_level': 'Medium' if order['Total_Value'] < threshold * 1.5 else 'High' }) # Overdue deliveries df['PO_Date'] = pd.to_datetime(df['PO_Date']) df['Delivery_Date'] = pd.to_datetime(df['Delivery_Date']) overdue = df[ (df['Delivery_Date'] < pd.Timestamp.now()) & (df['Status'] == 'Open') ] for _, order in overdue.iterrows(): days_overdue = (pd.Timestamp.now() - order['Delivery_Date']).days anomalies.append({ 'type': 'Overdue Delivery', 'po_number': order['PO_Number'], 'days_overdue': days_overdue, 'supplier': order['Supplier'], 'risk_level': 'High' if days_overdue > 30 else 'Medium' }) except Exception as e: st.error(f"Error in anomaly detection: {str(e)}") return anomalies[:10] # Return top 10 anomalies def _generate_spend_recommendations(self, df: pd.DataFrame) -> List[str]: """Generate AI-powered spending recommendations""" recommendations = [] # Category concentration analysis category_spend = df.groupby('Category')['Total_Value'].sum() total_spend = category_spend.sum() for category, spend in category_spend.items(): percentage = (spend / total_spend) * 100 if percentage > 30: recommendations.append(f"🎯 Consider diversifying suppliers in {category} (represents {percentage:.1f}% of total spend)") # Supplier dependency supplier_spend = df.groupby('Supplier')['Total_Value'].sum() for supplier, spend in supplier_spend.items(): percentage = (spend / total_spend) * 100 if percentage > 25: recommendations.append(f"⚠️ High dependency on {supplier} ({percentage:.1f}% of spend) - consider risk mitigation") # Price optimization avg_unit_prices = df.groupby('Category')['Unit_Price'].mean() recommendations.append("💡 Implement category-specific negotiation strategies for cost optimization") return recommendations[:5] def _generate_supplier_recommendations(self, supplier_metrics: pd.DataFrame) -> List[str]: """Generate supplier performance recommendations""" recommendations = [] # Performance-based recommendations poor_performers = supplier_metrics[supplier_metrics['Delivery_Performance'] < 90] if not poor_performers.empty: recommendations.append(f"📈 Develop improvement plans for {len(poor_performers)} underperforming suppliers") # Volume-based recommendations high_volume_suppliers = supplier_metrics[supplier_metrics['PO_Number'] > supplier_metrics['PO_Number'].quantile(0.8)] recommendations.append(f"🤝 Consider strategic partnerships with top {len(high_volume_suppliers)} high-volume suppliers") recommendations.append("🔍 Implement regular supplier audits and performance reviews") recommendations.append("📊 Set up automated alerts for delivery performance degradation") return recommendations[:4] def generate_insights(self, po_data: pd.DataFrame, supplier_data: pd.DataFrame) -> Dict[str, Any]: """Generate comprehensive procurement insights""" spend_insights = self.analyze_spend_trends(po_data) supplier_insights = self.analyze_supplier_performance(po_data) anomalies = self.detect_anomalies(po_data) return { 'spend_analysis': spend_insights, 'supplier_analysis': supplier_insights, 'anomalies': anomalies, 'summary': self._generate_executive_summary(spend_insights, supplier_insights, anomalies) } def _generate_executive_summary(self, spend_insights: Dict, supplier_insights: Dict, anomalies: List) -> str: """Generate executive summary""" try: total_spend = spend_insights.get('total_spend', 0) trend = spend_insights.get('monthly_trend', 'stable') best_supplier = supplier_insights.get('best_performer', {}).get('name', 'N/A') anomaly_count = len(anomalies) summary = f""" 📊 **Procurement Analytics Summary** • Total Spend: ${total_spend:,.2f} • Spending Trend: {trend.title()} • Best Performing Supplier: {best_supplier} • Critical Issues Detected: {anomaly_count} • Overall Health: {'Good' if anomaly_count < 5 else 'Needs Attention'} """ return summary except: return "📊 **Procurement Analytics Summary**\n\nData processing in progress..."