Create healthcare_analysis.py

healthcare_analysis.py

@@ -0,0 +1,400 @@
+# healthcare_analysis.py
+import pandas as pd
+import numpy as np
+from typing import Dict, List, Any, Optional, Tuple
+import logging
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+class HealthcareAnalyzer:
+    def __init__(self, data_registry):
+        self.data_registry = data_registry
+        self.analysis_results = {}
+    
+    def comprehensive_analysis(self, scenario_text: str) -> Dict[str, Any]:
+        """Perform comprehensive healthcare scenario analysis"""
+        logger.info("Starting comprehensive healthcare analysis")
+        
+        # Extract tasks and requirements
+        tasks = self._extract_tasks(scenario_text)
+        requirements = self._extract_requirements(scenario_text)
+        
+        # Identify relevant datasets
+        relevant_data = self._identify_relevant_data(scenario_text)
+        
+        # Perform analyses based on tasks
+        results = {}
+        
+        if "facility_distribution" in tasks:
+            results["facility_distribution"] = self.analyze_facility_distribution(relevant_data)
+        
+        if "capacity_analysis" in tasks:
+            results["capacity_analysis"] = self.analyze_capacity(relevant_data)
+        
+        if "resource_allocation" in tasks:
+            results["resource_allocation"] = self.analyze_resource_allocation(relevant_data)
+        
+        if "trends" in tasks:
+            results["trends"] = self.analyze_trends(relevant_data)
+        
+        # Generate recommendations
+        results["recommendations"] = self.generate_recommendations(results, requirements)
+        
+        # Future integration opportunities
+        results["future_integration"] = self.identify_integration_opportunities(results)
+        
+        logger.info("Comprehensive analysis completed")
+        return results
+    
+    def _extract_tasks(self, scenario_text: str) -> List[str]:
+        """Extract specific tasks from scenario text"""
+        tasks = []
+        task_keywords = {
+            "facility_distribution": ["facility", "distribution", "location", "sites"],
+            "capacity_analysis": ["capacity", "beds", "occupancy", "utilization"],
+            "resource_allocation": ["resource", "allocation", "staffing", "equipment"],
+            "trends": ["trend", "change", "growth", "decline", "pattern"]
+        }
+        
+        for task_type, keywords in task_keywords.items():
+            if any(kw in scenario_text.lower() for kw in keywords):
+                tasks.append(task_type)
+        
+        return tasks
+    
+    def _extract_requirements(self, scenario_text: str) -> Dict[str, Any]:
+        """Extract specific requirements from scenario text"""
+        return {
+            "geographic_scope": self._extract_geographic_scope(scenario_text),
+            "time_period": self._extract_time_period(scenario_text),
+            "facility_types": self._extract_facility_types(scenario_text),
+            "metrics_needed": self._extract_metrics(scenario_text)
+        }
+    
+    def analyze_facility_distribution(self, relevant_data: List[str]) -> Dict[str, Any]:
+        """Enhanced facility distribution analysis"""
+        results = {}
+        
+        for data_name in relevant_data:
+            df = self.data_registry.get(data_name)
+            if df is None:
+                continue
+                
+            # Geographic distribution
+            geo_col = self._find_column(df, ['province', 'state', 'region', 'zone'])
+            if geo_col:
+                geo_dist = df[geo_col].value_counts().to_dict()
+                results["geographic_distribution"] = geo_dist
+                
+                # Calculate Gini coefficient for inequality
+                gini = self._calculate_gini(list(geo_dist.values()))
+                results["geographic_inequality"] = gini
+            
+            # Facility type distribution
+            type_col = self._find_column(df, ['type', 'category', 'facility_type'])
+            if type_col:
+                type_dist = df[type_col].value_counts().to_dict()
+                results["facility_type_distribution"] = type_dist
+                
+                # Calculate diversity index
+                diversity = self._calculate_diversity_index(type_dist)
+                results["facility_diversity"] = diversity
+            
+            # Urban vs rural distribution
+            urban_col = self._find_column(df, ['urban', 'rural', 'location_type'])
+            if urban_col:
+                urban_rural = df[urban_col].value_counts().to_dict()
+                results["urban_rural_distribution"] = urban_rural
+        
+        return results
+    
+    def analyze_capacity(self, relevant_data: List[str]) -> Dict[str, Any]:
+        """Enhanced capacity analysis"""
+        results = {}
+        
+        for data_name in relevant_data:
+            df = self.data_registry.get(data_name)
+            if df is None:
+                continue
+                
+            # Current capacity
+            capacity_col = self._find_column(df, ['capacity', 'beds', 'current_capacity'])
+            if capacity_col:
+                total_capacity = df[capacity_col].sum()
+                results["total_capacity"] = total_capacity
+                
+                # Capacity by facility type
+                type_col = self._find_column(df, ['type', 'facility_type'])
+                if type_col:
+                    capacity_by_type = df.groupby(type_col)[capacity_col].sum().to_dict()
+                    results["capacity_by_type"] = capacity_by_type
+                
+                # Capacity utilization
+                utilization_col = self._find_column(df, ['utilization', 'occupancy', 'occupancy_rate'])
+                if utilization_col:
+                    avg_utilization = df[utilization_col].mean()
+                    results["average_utilization"] = avg_utilization
+                    
+                    # Utilization by facility type
+                    if type_col:
+                        utilization_by_type = df.groupby(type_col)[utilization_col].mean().to_dict()
+                        results["utilization_by_type"] = utilization_by_type
+                
+                # Capacity trends
+                time_cols = [col for col in df.columns if any(year in col.lower() for year in ['2020', '2021', '2022', '2023', '2024'])]
+                if len(time_cols) >= 2:
+                    trend_data = {}
+                    for col in time_cols:
+                        trend_data[col] = df[col].sum()
+                    results["capacity_trends"] = trend_data
+                    
+                    # Calculate growth rate
+                    if len(time_cols) >= 2:
+                        latest = time_cols[-1]
+                        earliest = time_cols[0]
+                        growth_rate = (trend_data[latest] - trend_data[earliest]) / trend_data[earliest] * 100
+                        results["capacity_growth_rate"] = growth_rate
+        
+        return results
+    
+    def analyze_resource_allocation(self, relevant_data: List[str]) -> Dict[str, Any]:
+        """Analyze resource allocation patterns"""
+        results = {}
+        
+        for data_name in relevant_data:
+            df = self.data_registry.get(data_name)
+            if df is None:
+                continue
+                
+            # Staff analysis
+            staff_col = self._find_column(df, ['staff', 'employees', 'fte'])
+            if staff_col:
+                total_staff = df[staff_col].sum()
+                results["total_staff"] = total_staff
+                
+                # Staff per bed ratio
+                capacity_col = self._find_column(df, ['capacity', 'beds'])
+                if capacity_col:
+                    df['staff_per_bed'] = df[staff_col] / df[capacity_col]
+                    avg_staff_per_bed = df['staff_per_bed'].mean()
+                    results["staff_per_bed_ratio"] = avg_staff_per_bed
+            
+            # Equipment analysis
+            equipment_cols = [col for col in df.columns if 'equipment' in col.lower()]
+            if equipment_cols:
+                equipment_summary = {}
+                for col in equipment_cols:
+                    equipment_summary[col] = df[col].sum()
+                results["equipment_summary"] = equipment_summary
+        
+        return results
+    
+    def analyze_trends(self, relevant_data: List[str]) -> Dict[str, Any]:
+        """Analyze trends in healthcare data"""
+        results = {}
+        
+        for data_name in relevant_data:
+            df = self.data_registry.get(data_name)
+            if df is None:
+                continue
+                
+            # Find time-based columns
+            time_cols = [col for col in df.columns if any(year in col.lower() for year in ['2020', '2021', '2022', '2023', '2024'])]
+            
+            if len(time_cols) >= 2:
+                trends = {}
+                
+                # Calculate year-over-year changes
+                for i in range(1, len(time_cols)):
+                    prev_year = time_cols[i-1]
+                    curr_year = time_cols[i]
+                    
+                    prev_total = df[prev_year].sum()
+                    curr_total = df[curr_year].sum()
+                    
+                    if prev_total > 0:
+                        change_pct = (curr_total - prev_total) / prev_total * 100
+                        trends[f"{prev_year}_to_{curr_year}"] = {
+                            "absolute_change": curr_total - prev_total,
+                            "percentage_change": change_pct
+                        }
+                
+                results["year_over_year_trends"] = trends
+        
+        return results
+    
+    def generate_recommendations(self, analysis_results: Dict[str, Any], requirements: Dict[str, Any]) -> List[Dict[str, str]]:
+        """Generate data-driven operational recommendations"""
+        recommendations = []
+        
+        # Capacity-related recommendations
+        if "capacity_analysis" in analysis_results:
+            capacity = analysis_results["capacity_analysis"]
+            
+            # Low utilization recommendations
+            if "average_utilization" in capacity and capacity["average_utilization"] < 0.7:
+                recommendations.append({
+                    "title": "Optimize Underutilized Capacity",
+                    "description": f"Average utilization is {capacity['average_utilization']:.1%}. Consider repurposing underutilized facilities or consolidating services.",
+                    "priority": "Medium",
+                    "data_source": "Capacity utilization analysis"
+                })
+            
+            # Capacity growth recommendations
+            if "capacity_growth_rate" in capacity and capacity["capacity_growth_rate"] < 2:
+                recommendations.append({
+                    "title": "Expand Capacity Strategically",
+                    "description": f"Capacity growth rate is only {capacity['capacity_growth_rate']:.1f}%. Invest in new facilities or expand existing ones to meet demand.",
+                    "priority": "High",
+                    "data_source": "Capacity trend analysis"
+                })
+        
+        # Geographic distribution recommendations
+        if "facility_distribution" in analysis_results:
+            dist = analysis_results["facility_distribution"]
+            
+            if "geographic_inequality" in dist and dist["geographic_inequality"] > 0.4:
+                recommendations.append({
+                    "title": "Address Geographic Inequity",
+                    "description": f"High geographic inequality (Gini: {dist['geographic_inequality']:.2f}). Consider targeted investments in underserved areas.",
+                    "priority": "High",
+                    "data_source": "Geographic distribution analysis"
+                })
+        
+        # Resource allocation recommendations
+        if "resource_allocation" in analysis_results:
+            resources = analysis_results["resource_allocation"]
+            
+            if "staff_per_bed_ratio" in resources and resources["staff_per_bed_ratio"] < 1.5:
+                recommendations.append({
+                    "title": "Increase Staffing Levels",
+                    "description": f"Staff per bed ratio is {resources['staff_per_bed_ratio']:.2f}, which may be insufficient. Consider hiring additional staff.",
+                    "priority": "High",
+                    "data_source": "Resource allocation analysis"
+                })
+        
+        # Sort by priority
+        priority_order = {"High": 0, "Medium": 1, "Low": 2}
+        recommendations.sort(key=lambda x: priority_order.get(x["priority"], 3))
+        
+        return recommendations
+    
+    def identify_integration_opportunities(self, analysis_results: Dict[str, Any]) -> Dict[str, Any]:
+        """Identify opportunities for AI integration and data enhancement"""
+        opportunities = {
+            "data_integration": [],
+            "ai_applications": [],
+            "enhanced_metrics": []
+        }
+        
+        # Data integration opportunities
+        opportunities["data_integration"].append({
+            "opportunity": "Integrate real-time occupancy data",
+            "description": "Combine current facility data with real-time occupancy monitoring systems",
+            "benefit": "Enable dynamic resource allocation and surge planning"
+        })
+        
+        opportunities["data_integration"].append({
+            "opportunity": "Incorporate demographic data",
+            "description": "Add population demographics and health needs data",
+            "benefit": "Improve demand forecasting and service planning"
+        })
+        
+        # AI application opportunities
+        opportunities["ai_applications"].append({
+            "opportunity": "Predictive capacity modeling",
+            "description": "Use ML to forecast capacity needs based on trends and external factors",
+            "benefit": "Proactive resource planning and reduced wait times"
+        })
+        
+        opportunities["ai_applications"].append({
+            "opportunity": "Optimization algorithms",
+            "description": "Implement AI for staff scheduling and resource allocation",
+            "benefit": "Improved efficiency and reduced operational costs"
+        })
+        
+        # Enhanced metrics
+        opportunities["enhanced_metrics"].append({
+            "metric": "Patient flow efficiency",
+            "description": "Measure time from admission to discharge across facilities",
+            "benefit": "Identify bottlenecks and improve patient experience"
+        })
+        
+        opportunities["enhanced_metrics"].append({
+            "metric": "Resource utilization index",
+            "description": "Composite metric combining staff, equipment, and space utilization",
+            "benefit": "Holistic view of operational efficiency"
+        })
+        
+        return opportunities
+    
+    # Helper methods
+    def _find_column(self, df, patterns):
+        """Find the first column matching any pattern"""
+        for col in df.columns:
+            if any(pattern.lower() in col.lower() for pattern in patterns):
+                return col
+        return None
+    
+    def _calculate_gini(self, values):
+        """Calculate Gini coefficient for inequality measurement"""
+        values = sorted(values)
+        n = len(values)
+        index = np.arange(1, n + 1)
+        gini = (np.sum((2 * index - n - 1) * values)) / (n * np.sum(values))
+        return gini
+    
+    def _calculate_diversity_index(self, distribution):
+        """Calculate Shannon diversity index"""
+        total = sum(distribution.values())
+        if total == 0:
+            return 0
+        proportions = [count/total for count in distribution.values()]
+        return -sum(p * np.log(p) for p in proportions if p > 0)
+    
+    def _extract_geographic_scope(self, text):
+        """Extract geographic scope from text"""
+        # Simple keyword-based extraction
+        if "alberta" in text.lower():
+            return "Alberta"
+        elif "canada" in text.lower():
+            return "Canada"
+        return "Unknown"
+    
+    def _extract_time_period(self, text):
+        """Extract time period from text"""
+        # Look for year patterns
+        import re
+        years = re.findall(r'\b(20\d{2})\b', text)
+        if len(years) >= 2:
+            return f"{min(years)}-{max(years)}"
+        return "Unknown"
+    
+    def _extract_facility_types(self, text):
+        """Extract facility types from text"""
+        types = []
+        if "hospital" in text.lower():
+            types.append("Hospitals")
+        if "nursing" in text.lower() or "long-term" in text.lower():
+            types.append("Nursing homes")
+        if "clinic" in text.lower():
+            types.append("Clinics")
+        return types
+    
+    def _extract_metrics(self, text):
+        """Extract required metrics from text"""
+        metrics = []
+        if "bed" in text.lower():
+            metrics.append("Bed capacity")
+        if "occupancy" in text.lower():
+            metrics.append("Occupancy rates")
+        if "staff" in text.lower():
+            metrics.append("Staffing levels")
+        return metrics
+    
+    def _identify_relevant_data(self, text):
+        """Identify relevant datasets for the scenario"""
+        # Use data registry's find_related_datasets method
+        keywords = ["facility", "bed", "capacity", "healthcare", "hospital"]
+        return [item["name"] for item in self.data_registry.find_related_datasets(keywords)]