Spaces:
Sleeping
Sleeping
File size: 8,357 Bytes
0846ca3 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 |
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..."
|