""" Portfolio Snapshot Generation Service This service handles automated daily portfolio snapshot generation, health score calculations, growth rate tracking, and performance alerts. """ from django.db import models, transaction from django.utils import timezone from django.core.mail import send_mail from django.conf import settings from decimal import Decimal from typing import Dict, List, Optional, Tuple import logging from datetime import date, timedelta from .enhanced_permissions_models import PortfolioSnapshot, ClientGrowthMetrics from .models import CustomUser, Branch from loans.models import Loan, LoanApplication logger = logging.getLogger(__name__) class PortfolioSnapshotService: """ Service for generating and managing portfolio snapshots """ def __init__(self): self.logger = logger def generate_daily_snapshots(self, target_date: Optional[date] = None) -> Dict[str, any]: """ Generate daily portfolio snapshots for all portfolio managers Args: target_date: Date to generate snapshots for (defaults to today) Returns: Dictionary with generation results and statistics """ if target_date is None: target_date = timezone.now().date() results = { 'date': target_date.isoformat(), 'snapshots_generated': 0, 'snapshots_updated': 0, 'errors': [], 'managers_processed': 0, 'success': True } try: # Get all portfolio managers (loan officers and team leaders) portfolio_managers = CustomUser.objects.filter( role__in=['loan_officer', 'team_leader'], is_active=True ).select_related('branch') with transaction.atomic(): for manager in portfolio_managers: try: snapshot_data = self._calculate_portfolio_metrics(manager, target_date) # Create or update snapshot snapshot, created = PortfolioSnapshot.objects.update_or_create( manager=manager, snapshot_date=target_date, defaults=snapshot_data ) if created: results['snapshots_generated'] += 1 else: results['snapshots_updated'] += 1 results['managers_processed'] += 1 # Calculate health score and check for alerts health_score = snapshot.get_portfolio_health_score() self._check_performance_alerts(manager, snapshot, health_score) self.logger.info(f"Generated snapshot for {manager.get_full_name()} - Health Score: {health_score}") except Exception as e: error_msg = f"Error generating snapshot for manager {manager.id}: {str(e)}" results['errors'].append(error_msg) self.logger.error(error_msg) self.logger.info(f"Daily snapshot generation completed: {results['snapshots_generated']} created, " f"{results['snapshots_updated']} updated") return results except Exception as e: self.logger.error(f"Error in daily snapshot generation: {e}") results['success'] = False results['error'] = str(e) return results def _calculate_portfolio_metrics(self, manager: CustomUser, target_date: date) -> Dict[str, any]: """ Calculate comprehensive portfolio metrics for a manager on a specific date Args: manager: Portfolio manager target_date: Date to calculate metrics for Returns: Dictionary with calculated metrics """ # Get all clients assigned to this manager manager_clients = CustomUser.objects.filter( role='borrower', branch=manager.branch ) # If manager has specific client assignments, use those # (This would require a client assignment model - for now use branch-based) # Calculate client metrics total_clients = manager_clients.count() active_clients = manager_clients.filter( status='active', loans__status='active', loans__disbursement_date__lte=target_date ).distinct().count() # Calculate new clients for this date new_clients = manager_clients.filter( date_joined__date=target_date ).count() # Calculate churned clients (clients who became inactive on this date) churned_clients = manager_clients.filter( status__in=['inactive', 'suspended'], updated_at__date=target_date ).count() # Get all loans for this manager's clients manager_loans = Loan.objects.filter( borrower__in=manager_clients, disbursement_date__lte=target_date, is_deleted=False ) # Loan metrics active_loans = manager_loans.filter(status='active').count() total_loans = manager_loans.count() # New loans disbursed on this date new_loans = manager_loans.filter( disbursement_date__date=target_date ).count() # Loans completed on this date completed_loans = manager_loans.filter( status='paid', updated_at__date=target_date ).count() # Loans that defaulted on this date defaulted_loans = manager_loans.filter( status='defaulted', updated_at__date=target_date ).count() # Financial metrics total_disbursed = manager_loans.aggregate( total=models.Sum('principal_amount') )['total'] or Decimal('0.00') total_outstanding = manager_loans.filter( status='active' ).aggregate( total=models.Sum('outstanding_amount') )['total'] or Decimal('0.00') # Calculate total collected (amount paid on all loans) total_collected = manager_loans.aggregate( total=models.Sum('amount_paid') )['total'] or Decimal('0.00') # Daily disbursements daily_disbursements = manager_loans.filter( disbursement_date__date=target_date ).aggregate( total=models.Sum('principal_amount') )['total'] or Decimal('0.00') # Daily collections (payments received on this date) from payments.models import Payment daily_collections = Payment.objects.filter( loan__borrower__in=manager_clients, payment_date__date=target_date, status='completed' ).aggregate( total=models.Sum('amount') )['total'] or Decimal('0.00') # Risk metrics - Portfolio at Risk calculations overdue_30_loans = manager_loans.filter( status='active', due_date__lt=target_date - timedelta(days=30) ) par_30 = overdue_30_loans.aggregate( total=models.Sum('outstanding_amount') )['total'] or Decimal('0.00') overdue_60_loans = manager_loans.filter( status='active', due_date__lt=target_date - timedelta(days=60) ) par_60 = overdue_60_loans.aggregate( total=models.Sum('outstanding_amount') )['total'] or Decimal('0.00') overdue_90_loans = manager_loans.filter( status='active', due_date__lt=target_date - timedelta(days=90) ) par_90 = overdue_90_loans.aggregate( total=models.Sum('outstanding_amount') )['total'] or Decimal('0.00') # Default rate calculation total_loans_count = manager_loans.count() defaulted_loans_count = manager_loans.filter(status='defaulted').count() default_rate = (defaulted_loans_count / total_loans_count * 100) if total_loans_count > 0 else Decimal('0.00') # Collection rate calculation expected_collections = manager_loans.filter( due_date__lte=target_date ).aggregate( total=models.Sum('total_amount') )['total'] or Decimal('0.00') collection_rate = (total_collected / expected_collections * 100) if expected_collections > 0 else Decimal('0.00') # Portfolio yield calculation (annualized) total_interest_earned = manager_loans.aggregate( total=models.Sum('interest_amount') )['total'] or Decimal('0.00') portfolio_yield = (total_interest_earned / total_disbursed * 100) if total_disbursed > 0 else Decimal('0.00') # Average loan size average_loan_size = (total_disbursed / total_loans_count) if total_loans_count > 0 else Decimal('0.00') return { 'branch': manager.branch, 'total_clients': total_clients, 'active_clients': active_clients, 'new_clients': new_clients, 'churned_clients': churned_clients, 'active_loans': active_loans, 'total_loans': total_loans, 'new_loans': new_loans, 'completed_loans': completed_loans, 'defaulted_loans': defaulted_loans, 'total_disbursed': total_disbursed, 'total_outstanding': total_outstanding, 'total_collected': total_collected, 'daily_disbursements': daily_disbursements, 'daily_collections': daily_collections, 'par_30': par_30, 'par_60': par_60, 'par_90': par_90, 'default_rate': default_rate, 'collection_rate': collection_rate, 'portfolio_yield': portfolio_yield, 'average_loan_size': average_loan_size, } def calculate_portfolio_health_score(self, manager: CustomUser, target_date: Optional[date] = None) -> Decimal: """ Calculate comprehensive portfolio health score for a manager Args: manager: Portfolio manager target_date: Date to calculate for (defaults to today) Returns: Health score from 0-100 """ if target_date is None: target_date = timezone.now().date() try: snapshot = PortfolioSnapshot.objects.get( manager=manager, snapshot_date=target_date ) return snapshot.get_portfolio_health_score() except PortfolioSnapshot.DoesNotExist: # Generate snapshot if it doesn't exist snapshot_data = self._calculate_portfolio_metrics(manager, target_date) snapshot = PortfolioSnapshot.objects.create( manager=manager, snapshot_date=target_date, **snapshot_data ) return snapshot.get_portfolio_health_score() def calculate_growth_rate(self, manager: CustomUser, period_days: int = 30) -> Decimal: """ Calculate portfolio growth rate over a specified period Args: manager: Portfolio manager period_days: Number of days to look back for comparison Returns: Growth rate as percentage """ try: current_date = timezone.now().date() previous_date = current_date - timedelta(days=period_days) # Get current and previous snapshots current_snapshot = PortfolioSnapshot.objects.filter( manager=manager, snapshot_date=current_date ).first() previous_snapshot = PortfolioSnapshot.objects.filter( manager=manager, snapshot_date=previous_date ).first() if not current_snapshot or not previous_snapshot: return Decimal('0.00') # Calculate growth rate based on total clients if previous_snapshot.total_clients == 0: return Decimal('100.00') if current_snapshot.total_clients > 0 else Decimal('0.00') growth_rate = ((current_snapshot.total_clients - previous_snapshot.total_clients) / previous_snapshot.total_clients) * 100 return Decimal(str(round(growth_rate, 2))) except Exception as e: self.logger.error(f"Error calculating growth rate for manager {manager.id}: {e}") return Decimal('0.00') def _check_performance_alerts(self, manager: CustomUser, snapshot: PortfolioSnapshot, health_score: Decimal): """ Check for performance alerts and send notifications if needed Args: manager: Portfolio manager snapshot: Portfolio snapshot health_score: Calculated health score """ alerts = [] # Health score alerts if health_score < 50: alerts.append({ 'type': 'critical', 'title': 'Critical Portfolio Health', 'message': f'Portfolio health score is critically low: {health_score}%', 'action_required': True }) elif health_score < 70: alerts.append({ 'type': 'warning', 'title': 'Low Portfolio Health', 'message': f'Portfolio health score needs attention: {health_score}%', 'action_required': False }) # Default rate alerts if snapshot.default_rate > 10: alerts.append({ 'type': 'critical', 'title': 'High Default Rate', 'message': f'Default rate is {snapshot.default_rate}% - immediate action required', 'action_required': True }) elif snapshot.default_rate > 5: alerts.append({ 'type': 'warning', 'title': 'Elevated Default Rate', 'message': f'Default rate is {snapshot.default_rate}% - monitor closely', 'action_required': False }) # Collection rate alerts if snapshot.collection_rate < 80: alerts.append({ 'type': 'warning', 'title': 'Low Collection Rate', 'message': f'Collection rate is {snapshot.collection_rate}% - follow up on overdue loans', 'action_required': True }) # Portfolio at Risk alerts par_30_percentage = (snapshot.par_30 / snapshot.total_outstanding * 100) if snapshot.total_outstanding > 0 else 0 if par_30_percentage > 15: alerts.append({ 'type': 'critical', 'title': 'High Portfolio at Risk', 'message': f'PAR 30+ is {par_30_percentage:.1f}% - urgent collection action needed', 'action_required': True }) elif par_30_percentage > 10: alerts.append({ 'type': 'warning', 'title': 'Elevated Portfolio at Risk', 'message': f'PAR 30+ is {par_30_percentage:.1f}% - increase collection efforts', 'action_required': False }) # Growth rate alerts growth_rate = self.calculate_growth_rate(manager, 30) if growth_rate < -10: alerts.append({ 'type': 'warning', 'title': 'Negative Portfolio Growth', 'message': f'Portfolio shrinking by {abs(growth_rate)}% - focus on client acquisition', 'action_required': False }) # Send alerts if any exist if alerts: self._send_performance_alerts(manager, alerts, snapshot) def _send_performance_alerts(self, manager: CustomUser, alerts: List[Dict], snapshot: PortfolioSnapshot): """ Send performance alerts via email and create notifications Args: manager: Portfolio manager alerts: List of alert dictionaries snapshot: Portfolio snapshot """ try: # Create notification records from utils.models import Notification for alert in alerts: Notification.objects.create( user=manager, title=alert['title'], message=alert['message'], notification_type='portfolio_alert', priority='high' if alert['type'] == 'critical' else 'medium', action_required=alert['action_required'] ) # Send email summary if there are critical alerts critical_alerts = [a for a in alerts if a['type'] == 'critical'] if critical_alerts and manager.email: self._send_alert_email(manager, critical_alerts, snapshot) self.logger.info(f"Sent {len(alerts)} alerts to {manager.get_full_name()}") except Exception as e: self.logger.error(f"Error sending alerts to manager {manager.id}: {e}") def _send_alert_email(self, manager: CustomUser, alerts: List[Dict], snapshot: PortfolioSnapshot): """ Send email alert to portfolio manager Args: manager: Portfolio manager alerts: List of critical alerts snapshot: Portfolio snapshot """ try: subject = f"Critical Portfolio Alerts - {snapshot.snapshot_date}" message = f""" Dear {manager.get_full_name()}, You have {len(alerts)} critical portfolio alerts that require immediate attention: """ for alert in alerts: message += f"• {alert['title']}: {alert['message']}\n" message += f""" Portfolio Summary for {snapshot.snapshot_date}: - Total Clients: {snapshot.total_clients} - Active Loans: {snapshot.active_loans} - Portfolio Health Score: {snapshot.get_portfolio_health_score()}% - Default Rate: {snapshot.default_rate}% - Collection Rate: {snapshot.collection_rate}% Please log into the system to review detailed metrics and take necessary actions. Best regards, Haven Grazuri Investment Limited Team """ send_mail( subject=subject, message=message, from_email=settings.DEFAULT_FROM_EMAIL, recipient_list=[manager.email], fail_silently=False ) self.logger.info(f"Sent alert email to {manager.email}") except Exception as e: self.logger.error(f"Error sending alert email to {manager.email}: {e}") def get_portfolio_trends(self, manager: CustomUser, days: int = 30) -> Dict[str, any]: """ Get portfolio trends over a specified period Args: manager: Portfolio manager days: Number of days to analyze Returns: Dictionary with trend data """ try: end_date = timezone.now().date() start_date = end_date - timedelta(days=days) snapshots = PortfolioSnapshot.objects.filter( manager=manager, snapshot_date__range=[start_date, end_date] ).order_by('snapshot_date') if not snapshots.exists(): return {'error': 'No snapshot data available for the specified period'} # Calculate trends first_snapshot = snapshots.first() last_snapshot = snapshots.last() client_growth = last_snapshot.total_clients - first_snapshot.total_clients loan_growth = last_snapshot.active_loans - first_snapshot.active_loans disbursement_growth = last_snapshot.total_disbursed - first_snapshot.total_disbursed trends = { 'period': f"{start_date} to {end_date}", 'client_growth': client_growth, 'client_growth_rate': (client_growth / first_snapshot.total_clients * 100) if first_snapshot.total_clients > 0 else 0, 'loan_growth': loan_growth, 'disbursement_growth': float(disbursement_growth), 'health_score_trend': last_snapshot.get_portfolio_health_score() - first_snapshot.get_portfolio_health_score(), 'collection_rate_trend': last_snapshot.collection_rate - first_snapshot.collection_rate, 'default_rate_trend': last_snapshot.default_rate - first_snapshot.default_rate, 'snapshots_count': snapshots.count(), 'daily_data': [] } # Add daily data points for snapshot in snapshots: trends['daily_data'].append({ 'date': snapshot.snapshot_date.isoformat(), 'total_clients': snapshot.total_clients, 'active_loans': snapshot.active_loans, 'health_score': float(snapshot.get_portfolio_health_score()), 'collection_rate': float(snapshot.collection_rate), 'default_rate': float(snapshot.default_rate), 'total_outstanding': float(snapshot.total_outstanding) }) return trends except Exception as e: self.logger.error(f"Error getting portfolio trends for manager {manager.id}: {e}") return {'error': str(e)} def generate_performance_report(self, manager: CustomUser, target_date: Optional[date] = None) -> Dict[str, any]: """ Generate comprehensive performance report for a portfolio manager Args: manager: Portfolio manager target_date: Date to generate report for (defaults to today) Returns: Dictionary with comprehensive performance data """ if target_date is None: target_date = timezone.now().date() try: # Get current snapshot snapshot = PortfolioSnapshot.objects.filter( manager=manager, snapshot_date=target_date ).first() if not snapshot: return {'error': 'No snapshot data available for the specified date'} # Get trends trends_30d = self.get_portfolio_trends(manager, 30) trends_90d = self.get_portfolio_trends(manager, 90) # Calculate health score health_score = snapshot.get_portfolio_health_score() # Get growth rate growth_rate = self.calculate_growth_rate(manager, 30) report = { 'manager': { 'id': str(manager.id), 'name': manager.get_full_name(), 'role': manager.role, 'branch': manager.branch.name if manager.branch else 'No Branch' }, 'snapshot_date': target_date.isoformat(), 'portfolio_overview': { 'total_clients': snapshot.total_clients, 'active_clients': snapshot.active_clients, 'active_loans': snapshot.active_loans, 'total_disbursed': float(snapshot.total_disbursed), 'total_outstanding': float(snapshot.total_outstanding), 'total_collected': float(snapshot.total_collected) }, 'performance_metrics': { 'health_score': float(health_score), 'collection_rate': float(snapshot.collection_rate), 'default_rate': float(snapshot.default_rate), 'portfolio_yield': float(snapshot.portfolio_yield), 'growth_rate_30d': float(growth_rate) }, 'risk_metrics': { 'par_30': float(snapshot.par_30), 'par_60': float(snapshot.par_60), 'par_90': float(snapshot.par_90), 'par_30_percentage': float(snapshot.par_30 / snapshot.total_outstanding * 100) if snapshot.total_outstanding > 0 else 0 }, 'daily_activity': { 'new_clients': snapshot.new_clients, 'new_loans': snapshot.new_loans, 'daily_disbursements': float(snapshot.daily_disbursements), 'daily_collections': float(snapshot.daily_collections), 'completed_loans': snapshot.completed_loans, 'defaulted_loans': snapshot.defaulted_loans }, 'trends': { '30_day': trends_30d, '90_day': trends_90d }, 'generated_at': timezone.now().isoformat() } return report except Exception as e: self.logger.error(f"Error generating performance report for manager {manager.id}: {e}") return {'error': str(e)}