""" Growth Forecasting and Projections Service This service provides predictive models for client growth, seasonal trend analysis, capacity planning tools, and early warning systems for growth anomalies. """ from django.db import models from django.db.models import Count, Sum, Avg, Q, F from django.utils import timezone from datetime import datetime, timedelta, date from decimal import Decimal from typing import Dict, List, Optional, Any, Tuple import calendar import numpy as np from collections import defaultdict import math from users.models import CustomUser, Branch from loans.models import Loan, LoanApplication from utils.models import Notification class GrowthForecastingService: """ Service class for growth forecasting and projections Provides predictive models, seasonal analysis, capacity planning, and anomaly detection """ def __init__(self): self.current_date = timezone.now().date() def create_predictive_models(self, branch_id: Optional[str] = None, forecast_months: int = 12) -> Dict[str, Any]: """ Create predictive models for client growth using historical data Args: branch_id: Optional branch ID to filter by forecast_months: Number of months to forecast ahead Returns: Dictionary containing forecast models and predictions """ # Get historical data for the last 24 months end_date = self.current_date start_date = end_date - timedelta(days=24 * 30) # Base queryset for clients clients_query = CustomUser.objects.filter( role='borrower', created_at__date__gte=start_date, created_at__date__lte=end_date ) if branch_id: clients_query = clients_query.filter(branch_id=branch_id) # Get monthly historical data historical_data = [] for i in range(24): month_start = end_date.replace(day=1) - timedelta(days=i*30) month_end = (month_start + timedelta(days=32)).replace(day=1) - timedelta(days=1) month_clients = clients_query.filter( created_at__date__gte=month_start, created_at__date__lte=month_end ) new_clients = month_clients.count() approved_clients = month_clients.filter(status='active').count() historical_data.append({ 'month': month_start.strftime('%Y-%m'), 'new_clients': new_clients, 'approved_clients': approved_clients, 'month_index': i }) historical_data.reverse() # Oldest to newest # Extract time series data new_clients_series = [data['new_clients'] for data in historical_data] approved_clients_series = [data['approved_clients'] for data in historical_data] # Simple linear regression for trend def linear_regression(y_values): n = len(y_values) x_values = list(range(n)) sum_x = sum(x_values) sum_y = sum(y_values) sum_xy = sum(x * y for x, y in zip(x_values, y_values)) sum_x2 = sum(x * x for x in x_values) if n * sum_x2 - sum_x * sum_x == 0: slope = 0 intercept = sum_y / n if n > 0 else 0 else: slope = (n * sum_xy - sum_x * sum_y) / (n * sum_x2 - sum_x * sum_x) intercept = (sum_y - slope * sum_x) / n return slope, intercept # Calculate trends new_clients_slope, new_clients_intercept = linear_regression(new_clients_series) approved_clients_slope, approved_clients_intercept = linear_regression(approved_clients_series) # Generate forecasts forecasts = [] for i in range(forecast_months): future_month_index = len(historical_data) + i future_date = end_date + timedelta(days=i*30) predicted_new = max(0, int(new_clients_slope * future_month_index + new_clients_intercept)) predicted_approved = max(0, int(approved_clients_slope * future_month_index + approved_clients_intercept)) forecasts.append({ 'month': future_date.strftime('%Y-%m'), 'predicted_new_clients': predicted_new, 'predicted_approved_clients': predicted_approved, 'confidence_level': max(0.5, 1.0 - (i * 0.05)) # Decreasing confidence over time }) # Calculate model accuracy metrics def calculate_accuracy(actual, predicted): if not actual or not predicted: return 0 errors = [abs(a - p) for a, p in zip(actual, predicted)] mae = sum(errors) / len(errors) # Mean Absolute Error mape = sum(abs(a - p) / max(a, 1) for a, p in zip(actual, predicted)) / len(actual) * 100 # MAPE return { 'mae': mae, 'mape': mape, 'accuracy': max(0, 100 - mape) } # Test model on last 6 months of historical data test_data = historical_data[-6:] test_predictions = [] for i, data in enumerate(test_data): month_index = len(historical_data) - 6 + i pred_new = max(0, int(new_clients_slope * month_index + new_clients_intercept)) pred_approved = max(0, int(approved_clients_slope * month_index + approved_clients_intercept)) test_predictions.append({'new': pred_new, 'approved': pred_approved}) actual_new = [data['new_clients'] for data in test_data] predicted_new = [pred['new'] for pred in test_predictions] model_accuracy = calculate_accuracy(actual_new, predicted_new) return { 'historical_data': historical_data, 'forecasts': forecasts, 'model_metrics': { 'new_clients_trend': { 'slope': new_clients_slope, 'intercept': new_clients_intercept, 'direction': 'increasing' if new_clients_slope > 0 else 'decreasing' if new_clients_slope < 0 else 'stable' }, 'approved_clients_trend': { 'slope': approved_clients_slope, 'intercept': approved_clients_intercept, 'direction': 'increasing' if approved_clients_slope > 0 else 'decreasing' if approved_clients_slope < 0 else 'stable' }, 'accuracy': model_accuracy }, 'forecast_period': forecast_months, 'generated_at': timezone.now().isoformat() } def implement_seasonal_trend_analysis(self, branch_id: Optional[str] = None, years: int = 2) -> Dict[str, Any]: """ Implement seasonal trend analysis to identify patterns in client acquisition Args: branch_id: Optional branch ID to filter by years: Number of years of historical data to analyze Returns: Dictionary containing seasonal analysis data """ end_date = self.current_date start_date = end_date - timedelta(days=years * 365) # Base queryset clients_query = CustomUser.objects.filter( role='borrower', created_at__date__gte=start_date, created_at__date__lte=end_date ) if branch_id: clients_query = clients_query.filter(branch_id=branch_id) # Group by month of year for seasonal analysis monthly_patterns = {} for month in range(1, 13): month_clients = clients_query.filter(created_at__month=month) monthly_patterns[calendar.month_name[month]] = { 'total_clients': month_clients.count(), 'avg_per_year': month_clients.count() / years, 'approved_clients': month_clients.filter(status='active').count() } # Group by quarter quarterly_patterns = {} quarters = { 'Q1': [1, 2, 3], 'Q2': [4, 5, 6], 'Q3': [7, 8, 9], 'Q4': [10, 11, 12] } for quarter, months in quarters.items(): quarter_clients = clients_query.filter(created_at__month__in=months) quarterly_patterns[quarter] = { 'total_clients': quarter_clients.count(), 'avg_per_year': quarter_clients.count() / years, 'approved_clients': quarter_clients.filter(status='active').count() } # Identify peak and low seasons monthly_totals = {month: data['total_clients'] for month, data in monthly_patterns.items()} peak_month = max(monthly_totals, key=monthly_totals.get) low_month = min(monthly_totals, key=monthly_totals.get) quarterly_totals = {quarter: data['total_clients'] for quarter, data in quarterly_patterns.items()} peak_quarter = max(quarterly_totals, key=quarterly_totals.get) low_quarter = min(quarterly_totals, key=quarterly_totals.get) # Calculate seasonality index (month average / overall average) overall_monthly_avg = sum(monthly_totals.values()) / 12 seasonality_index = {} for month, total in monthly_totals.items(): if overall_monthly_avg > 0: seasonality_index[month] = (total / years) / overall_monthly_avg else: seasonality_index[month] = 1.0 # Seasonal forecast for next 12 months seasonal_forecast = [] base_monthly_avg = overall_monthly_avg for i in range(12): future_date = end_date + timedelta(days=i*30) month_name = calendar.month_name[future_date.month] seasonal_multiplier = seasonality_index.get(month_name, 1.0) predicted_clients = int(base_monthly_avg * seasonal_multiplier) seasonal_forecast.append({ 'month': future_date.strftime('%Y-%m'), 'month_name': month_name, 'predicted_clients': predicted_clients, 'seasonal_multiplier': seasonal_multiplier, 'is_peak_season': month_name == peak_month, 'is_low_season': month_name == low_month }) return { 'monthly_patterns': monthly_patterns, 'quarterly_patterns': quarterly_patterns, 'seasonality_index': seasonality_index, 'peak_periods': { 'peak_month': peak_month, 'peak_quarter': peak_quarter, 'peak_month_clients': monthly_totals[peak_month], 'peak_quarter_clients': quarterly_totals[peak_quarter] }, 'low_periods': { 'low_month': low_month, 'low_quarter': low_quarter, 'low_month_clients': monthly_totals[low_month], 'low_quarter_clients': quarterly_totals[low_quarter] }, 'seasonal_forecast': seasonal_forecast, 'analysis_period': f"{start_date.strftime('%Y-%m')} to {end_date.strftime('%Y-%m')}", 'years_analyzed': years } def build_capacity_planning_tools(self, branch_id: Optional[str] = None) -> Dict[str, Any]: """ Build capacity planning tools based on growth projections Args: branch_id: Optional branch ID to filter by Returns: Dictionary containing capacity planning recommendations """ # Get current capacity metrics current_staff = CustomUser.objects.filter( role__in=['loan_officer', 'team_leader'], is_active=True ) if branch_id: current_staff = current_staff.filter(branch_id=branch_id) total_staff = current_staff.count() # Get current client load current_clients = CustomUser.objects.filter( role='borrower', status='active' ) if branch_id: current_clients = current_clients.filter(branch_id=branch_id) total_clients = current_clients.count() # Calculate current ratios clients_per_officer = total_clients / total_staff if total_staff > 0 else 0 # Get growth projections growth_forecast = self.create_predictive_models(branch_id, 12) # Calculate projected client growth projected_new_clients = sum(forecast['predicted_new_clients'] for forecast in growth_forecast['forecasts']) projected_total_clients = total_clients + projected_new_clients # Industry benchmarks (configurable) optimal_clients_per_officer = 50 # Industry standard max_clients_per_officer = 75 # Maximum sustainable load # Calculate capacity requirements optimal_staff_needed = math.ceil(projected_total_clients / optimal_clients_per_officer) minimum_staff_needed = math.ceil(projected_total_clients / max_clients_per_officer) additional_staff_optimal = max(0, optimal_staff_needed - total_staff) additional_staff_minimum = max(0, minimum_staff_needed - total_staff) # Calculate hiring timeline hiring_timeline = [] monthly_growth = projected_new_clients / 12 for i in range(12): month_date = self.current_date + timedelta(days=i*30) cumulative_new_clients = monthly_growth * (i + 1) projected_clients_at_month = total_clients + cumulative_new_clients staff_needed_at_month = math.ceil(projected_clients_at_month / optimal_clients_per_officer) additional_staff_at_month = max(0, staff_needed_at_month - total_staff) hiring_timeline.append({ 'month': month_date.strftime('%Y-%m'), 'projected_clients': int(projected_clients_at_month), 'staff_needed': staff_needed_at_month, 'additional_staff_needed': additional_staff_at_month, 'clients_per_officer': projected_clients_at_month / max(staff_needed_at_month, 1) }) # Resource planning resource_requirements = { 'staff_expansion': { 'current_staff': total_staff, 'optimal_staff_needed': optimal_staff_needed, 'minimum_staff_needed': minimum_staff_needed, 'additional_staff_optimal': additional_staff_optimal, 'additional_staff_minimum': additional_staff_minimum }, 'infrastructure': { 'current_capacity': total_clients, 'projected_capacity_needed': projected_total_clients, 'capacity_utilization': (projected_total_clients / (total_staff * max_clients_per_officer)) * 100 if total_staff > 0 else 0 }, 'budget_estimates': { 'monthly_salary_per_officer': 50000, # Configurable 'additional_monthly_cost_optimal': additional_staff_optimal * 50000, 'additional_monthly_cost_minimum': additional_staff_minimum * 50000, 'annual_cost_optimal': additional_staff_optimal * 50000 * 12, 'annual_cost_minimum': additional_staff_minimum * 50000 * 12 } } # Capacity alerts capacity_alerts = [] if clients_per_officer > max_clients_per_officer: capacity_alerts.append({ 'type': 'critical', 'message': f'Current client-to-officer ratio ({clients_per_officer:.1f}) exceeds maximum sustainable load ({max_clients_per_officer})', 'recommendation': 'Immediate hiring required' }) elif clients_per_officer > optimal_clients_per_officer: capacity_alerts.append({ 'type': 'warning', 'message': f'Current client-to-officer ratio ({clients_per_officer:.1f}) exceeds optimal level ({optimal_clients_per_officer})', 'recommendation': 'Consider hiring additional staff' }) if additional_staff_optimal > 0: capacity_alerts.append({ 'type': 'info', 'message': f'Projected growth will require {additional_staff_optimal} additional staff members over the next 12 months', 'recommendation': 'Begin recruitment planning' }) return { 'current_metrics': { 'total_staff': total_staff, 'total_clients': total_clients, 'clients_per_officer': clients_per_officer }, 'projections': { 'projected_new_clients': projected_new_clients, 'projected_total_clients': projected_total_clients, 'growth_rate': (projected_new_clients / total_clients * 100) if total_clients > 0 else 0 }, 'resource_requirements': resource_requirements, 'hiring_timeline': hiring_timeline, 'capacity_alerts': capacity_alerts, 'benchmarks': { 'optimal_clients_per_officer': optimal_clients_per_officer, 'max_clients_per_officer': max_clients_per_officer }, 'generated_at': timezone.now().isoformat() } def add_early_warning_systems(self, branch_id: Optional[str] = None) -> Dict[str, Any]: """ Add early warning systems for growth anomalies Args: branch_id: Optional branch ID to filter by Returns: Dictionary containing anomaly detection results and alerts """ # Get recent data for anomaly detection end_date = self.current_date # Analyze last 3 months vs previous 3 months current_period_start = end_date - timedelta(days=90) previous_period_start = end_date - timedelta(days=180) previous_period_end = end_date - timedelta(days=90) # Base queryset base_query = CustomUser.objects.filter(role='borrower') if branch_id: base_query = base_query.filter(branch_id=branch_id) # Current period metrics current_clients = base_query.filter( created_at__date__gte=current_period_start, created_at__date__lte=end_date ) current_new_clients = current_clients.count() current_approved = current_clients.filter(status='active').count() current_conversion_rate = (current_approved / current_new_clients * 100) if current_new_clients > 0 else 0 # Previous period metrics previous_clients = base_query.filter( created_at__date__gte=previous_period_start, created_at__date__lte=previous_period_end ) previous_new_clients = previous_clients.count() previous_approved = previous_clients.filter(status='active').count() previous_conversion_rate = (previous_approved / previous_new_clients * 100) if previous_new_clients > 0 else 0 # Calculate changes client_growth_change = ((current_new_clients - previous_new_clients) / previous_new_clients * 100) if previous_new_clients > 0 else 0 conversion_rate_change = current_conversion_rate - previous_conversion_rate # Anomaly detection thresholds growth_decline_threshold = -20 # 20% decline conversion_decline_threshold = -10 # 10 percentage point decline growth_spike_threshold = 100 # 100% increase # Detect anomalies anomalies = [] alerts = [] # Growth anomalies if client_growth_change <= growth_decline_threshold: anomalies.append({ 'type': 'growth_decline', 'severity': 'high', 'metric': 'client_acquisition', 'change': client_growth_change, 'description': f'Client acquisition declined by {abs(client_growth_change):.1f}% compared to previous period' }) alerts.append({ 'type': 'critical', 'title': 'Significant Growth Decline Detected', 'message': f'Client acquisition has dropped by {abs(client_growth_change):.1f}% in the last 3 months', 'recommendation': 'Review marketing strategies and market conditions', 'action_required': True }) elif client_growth_change >= growth_spike_threshold: anomalies.append({ 'type': 'growth_spike', 'severity': 'medium', 'metric': 'client_acquisition', 'change': client_growth_change, 'description': f'Client acquisition increased by {client_growth_change:.1f}% compared to previous period' }) alerts.append({ 'type': 'warning', 'title': 'Unusual Growth Spike Detected', 'message': f'Client acquisition has increased by {client_growth_change:.1f}% in the last 3 months', 'recommendation': 'Ensure adequate capacity to handle increased demand', 'action_required': True }) # Conversion rate anomalies if conversion_rate_change <= conversion_decline_threshold: anomalies.append({ 'type': 'conversion_decline', 'severity': 'high', 'metric': 'conversion_rate', 'change': conversion_rate_change, 'description': f'Conversion rate declined by {abs(conversion_rate_change):.1f} percentage points' }) alerts.append({ 'type': 'critical', 'title': 'Conversion Rate Decline', 'message': f'Client approval rate has dropped by {abs(conversion_rate_change):.1f} percentage points', 'recommendation': 'Review approval criteria and application quality', 'action_required': True }) # Weekly trend analysis for more granular anomaly detection weekly_trends = [] for i in range(12): # Last 12 weeks week_start = end_date - timedelta(days=(i+1)*7) week_end = end_date - timedelta(days=i*7) week_clients = base_query.filter( created_at__date__gte=week_start, created_at__date__lte=week_end ).count() weekly_trends.append({ 'week': week_start.strftime('%Y-W%U'), 'new_clients': week_clients, 'week_start': week_start.isoformat(), 'week_end': week_end.isoformat() }) weekly_trends.reverse() # Oldest to newest # Detect weekly anomalies if len(weekly_trends) >= 4: recent_weeks = [trend['new_clients'] for trend in weekly_trends[-4:]] previous_weeks = [trend['new_clients'] for trend in weekly_trends[-8:-4]] recent_avg = sum(recent_weeks) / len(recent_weeks) previous_avg = sum(previous_weeks) / len(previous_weeks) weekly_change = ((recent_avg - previous_avg) / previous_avg * 100) if previous_avg > 0 else 0 if weekly_change <= -30: # 30% weekly decline alerts.append({ 'type': 'warning', 'title': 'Weekly Acquisition Trend Decline', 'message': f'Weekly client acquisition has declined by {abs(weekly_change):.1f}% in recent weeks', 'recommendation': 'Monitor closely and investigate causes', 'action_required': False }) # Market saturation analysis total_active_clients = base_query.filter(status='active').count() # Estimate market size (this would typically come from external data) estimated_market_size = 10000 # Configurable based on market research market_penetration = (total_active_clients / estimated_market_size * 100) if estimated_market_size > 0 else 0 if market_penetration > 80: alerts.append({ 'type': 'info', 'title': 'High Market Penetration', 'message': f'Market penetration is at {market_penetration:.1f}%', 'recommendation': 'Consider market expansion or new product offerings', 'action_required': False }) # Competitive analysis alerts (based on growth patterns) if client_growth_change < -10 and conversion_rate_change < -5: alerts.append({ 'type': 'warning', 'title': 'Potential Competitive Pressure', 'message': 'Both acquisition and conversion rates are declining', 'recommendation': 'Analyze competitive landscape and adjust strategy', 'action_required': True }) # Generate notifications for critical alerts for alert in alerts: if alert.get('action_required') and alert['type'] == 'critical': # Create notification for administrators admin_users = CustomUser.objects.filter(role='admin', is_active=True) for admin in admin_users: Notification.objects.create( user=admin, notification_type='portfolio_alert', title=alert['title'], message=alert['message'], priority='high', action_required=True, alert_data={ 'alert_type': alert['type'], 'recommendation': alert['recommendation'], 'branch_id': branch_id } ) return { 'period_comparison': { 'current_period': { 'start_date': current_period_start.isoformat(), 'end_date': end_date.isoformat(), 'new_clients': current_new_clients, 'approved_clients': current_approved, 'conversion_rate': current_conversion_rate }, 'previous_period': { 'start_date': previous_period_start.isoformat(), 'end_date': previous_period_end.isoformat(), 'new_clients': previous_new_clients, 'approved_clients': previous_approved, 'conversion_rate': previous_conversion_rate }, 'changes': { 'client_growth_change': client_growth_change, 'conversion_rate_change': conversion_rate_change } }, 'anomalies': anomalies, 'alerts': alerts, 'weekly_trends': weekly_trends, 'market_analysis': { 'total_active_clients': total_active_clients, 'estimated_market_size': estimated_market_size, 'market_penetration': market_penetration }, 'thresholds': { 'growth_decline_threshold': growth_decline_threshold, 'conversion_decline_threshold': conversion_decline_threshold, 'growth_spike_threshold': growth_spike_threshold }, 'generated_at': timezone.now().isoformat() } def generate_comprehensive_forecast_report(self, branch_id: Optional[str] = None) -> Dict[str, Any]: """ Generate a comprehensive forecast report combining all forecasting components Args: branch_id: Optional branch ID to filter by Returns: Dictionary containing comprehensive forecast analysis """ # Get all forecasting components predictive_models = self.create_predictive_models(branch_id, 12) seasonal_analysis = self.implement_seasonal_trend_analysis(branch_id, 2) capacity_planning = self.build_capacity_planning_tools(branch_id) early_warnings = self.add_early_warning_systems(branch_id) # Generate executive summary executive_summary = { 'forecast_period': '12 months', 'projected_growth': { 'new_clients': sum(f['predicted_new_clients'] for f in predictive_models['forecasts']), 'growth_trend': predictive_models['model_metrics']['new_clients_trend']['direction'], 'confidence': predictive_models['model_metrics']['accuracy']['accuracy'] }, 'seasonal_insights': { 'peak_month': seasonal_analysis['peak_periods']['peak_month'], 'low_month': seasonal_analysis['low_periods']['low_month'], 'seasonality_strength': max(seasonal_analysis['seasonality_index'].values()) - min(seasonal_analysis['seasonality_index'].values()) }, 'capacity_requirements': { 'additional_staff_needed': capacity_planning['resource_requirements']['staff_expansion']['additional_staff_optimal'], 'budget_impact': capacity_planning['resource_requirements']['budget_estimates']['annual_cost_optimal'] }, 'risk_alerts': len([alert for alert in early_warnings['alerts'] if alert['type'] == 'critical']) } # Generate recommendations recommendations = [] # Growth recommendations if predictive_models['model_metrics']['new_clients_trend']['direction'] == 'decreasing': recommendations.append({ 'category': 'growth', 'priority': 'high', 'title': 'Address Declining Growth Trend', 'description': 'Client acquisition is trending downward. Consider reviewing marketing strategies and market conditions.', 'actions': [ 'Analyze competitor activities', 'Review marketing campaign effectiveness', 'Conduct customer satisfaction surveys', 'Explore new market segments' ] }) # Seasonal recommendations peak_month = seasonal_analysis['peak_periods']['peak_month'] recommendations.append({ 'category': 'seasonal', 'priority': 'medium', 'title': f'Prepare for Peak Season ({peak_month})', 'description': f'{peak_month} typically shows highest client acquisition. Ensure adequate preparation.', 'actions': [ 'Increase marketing budget during peak months', 'Ensure adequate staffing levels', 'Prepare promotional campaigns', 'Stock up on necessary resources' ] }) # Capacity recommendations if capacity_planning['resource_requirements']['staff_expansion']['additional_staff_optimal'] > 0: recommendations.append({ 'category': 'capacity', 'priority': 'high', 'title': 'Staff Expansion Required', 'description': f"Projected growth requires {capacity_planning['resource_requirements']['staff_expansion']['additional_staff_optimal']} additional staff members.", 'actions': [ 'Begin recruitment process immediately', 'Develop training programs for new staff', 'Plan office space expansion', 'Budget for additional salary costs' ] }) # Risk recommendations for alert in early_warnings['alerts']: if alert['type'] == 'critical': recommendations.append({ 'category': 'risk', 'priority': 'critical', 'title': alert['title'], 'description': alert['message'], 'actions': [alert['recommendation']] }) return { 'executive_summary': executive_summary, 'detailed_analysis': { 'predictive_models': predictive_models, 'seasonal_analysis': seasonal_analysis, 'capacity_planning': capacity_planning, 'early_warnings': early_warnings }, 'recommendations': recommendations, 'report_metadata': { 'generated_at': timezone.now().isoformat(), 'branch_id': branch_id, 'report_type': 'comprehensive_growth_forecast', 'validity_period': '30 days' } }