""" Credit Score Calculator Utility Calculates credit scores based on actual loan history and user data """ from django.contrib.auth import get_user_model from django.utils import timezone from datetime import timedelta from decimal import Decimal from reports.models import LoanScoring from loans.models import Loan, Repayment from django.db.models import Sum, Count, Q, Avg import logging User = get_user_model() logger = logging.getLogger(__name__) class CreditScoreCalculator: """ Calculates credit scores based on multiple factors: 1. Repayment History (40 points) 2. Income Stability (25 points) 3. Rollover Frequency (20 points) 4. Employment Stability (15 points) """ MAX_REPAYMENT_SCORE = 40 MAX_INCOME_SCORE = 25 MAX_ROLLOVER_SCORE = 20 MAX_EMPLOYMENT_SCORE = 15 MAX_TOTAL_SCORE = 100 @classmethod def calculate_repayment_history_score(cls, user): """ Calculate repayment history score (0-40 points) Based on: - On-time payment percentage - Number of defaults - Recent payment behavior """ try: loans = Loan.objects.filter(borrower=user) if not loans.exists(): return 0 total_loans = loans.count() # Count loans by status paid_loans = loans.filter(status='paid').count() overdue_loans = loans.filter(status='overdue').count() defaulted_loans = loans.filter(status='defaulted').count() # Calculate on-time payment rate on_time_rate = (paid_loans / total_loans) * 100 if total_loans > 0 else 0 # Calculate recent payment behavior (last 6 months) six_months_ago = timezone.now() - timedelta(days=180) recent_loans = loans.filter(disbursed_at__gte=six_months_ago) recent_paid = recent_loans.filter(status='paid').count() recent_total = recent_loans.count() recent_rate = (recent_paid / recent_total) * 100 if recent_total > 0 else on_time_rate # Score calculation score = 0 # Base score from overall payment rate if on_time_rate >= 95: score += 25 elif on_time_rate >= 85: score += 20 elif on_time_rate >= 70: score += 15 elif on_time_rate >= 50: score += 10 else: score += 5 # Recent behavior bonus/penalty if recent_rate >= on_time_rate: score += 10 # Improving trend elif recent_rate < on_time_rate - 20: score -= 5 # Declining trend # Penalty for defaults if defaulted_loans > 0: score -= (defaulted_loans * 5) # Bonus for no overdue loans if overdue_loans == 0 and total_loans > 0: score += 5 return max(0, min(score, cls.MAX_REPAYMENT_SCORE)) except Exception as e: logger.error(f"Error calculating repayment history score for {user}: {str(e)}") return 0 @classmethod def calculate_income_score(cls, user): """ Calculate income stability score (0-25 points) Based on: - Declared income level - Income consistency - Debt-to-income ratio """ try: # Get user's declared income monthly_income = getattr(user, 'monthly_income', None) if not monthly_income: return 5 # Minimum score for no income data # Calculate total outstanding debt outstanding_loans = Loan.objects.filter( borrower=user, status='active' ) # Calculate outstanding balance for each loan (total_amount - amount_paid) total_outstanding = Decimal('0') for loan in outstanding_loans: outstanding_balance = loan.total_amount - loan.amount_paid total_outstanding += outstanding_balance # Calculate debt-to-income ratio debt_to_income = (total_outstanding / monthly_income) * 100 if monthly_income > 0 else 100 score = 0 # Base score from income level if monthly_income >= 100000: # 100K+ score += 15 elif monthly_income >= 50000: # 50K-100K score += 12 elif monthly_income >= 25000: # 25K-50K score += 10 elif monthly_income >= 15000: # 15K-25K score += 7 else: score += 5 # Debt-to-income ratio adjustment if debt_to_income <= 30: score += 10 # Excellent DTI elif debt_to_income <= 50: score += 7 # Good DTI elif debt_to_income <= 70: score += 3 # Fair DTI else: score -= 5 # Poor DTI return max(0, min(score, cls.MAX_INCOME_SCORE)) except Exception as e: logger.error(f"Error calculating income score for {user}: {str(e)}") return 5 @classmethod def calculate_rollover_frequency_score(cls, user): """ Calculate rollover frequency score (0-20 points) Lower rollover frequency = higher score """ try: loans = Loan.objects.filter(borrower=user) if not loans.exists(): return cls.MAX_ROLLOVER_SCORE # No history = full score total_loans = loans.count() rolled_over_loans = loans.filter(is_rolled_over=True).count() rollover_rate = (rolled_over_loans / total_loans) * 100 if total_loans > 0 else 0 # Score based on rollover frequency if rollover_rate == 0: return cls.MAX_ROLLOVER_SCORE elif rollover_rate <= 10: return 15 elif rollover_rate <= 25: return 10 elif rollover_rate <= 50: return 5 else: return 0 except Exception as e: logger.error(f"Error calculating rollover frequency score for {user}: {str(e)}") return 10 @classmethod def calculate_employment_stability_score(cls, user): """ Calculate employment stability score (0-15 points) Based on: - Employment status - Business type - Account age """ try: score = 0 # Employment status employment_status = getattr(user, 'employment_status', None) if employment_status == 'employed': score += 8 elif employment_status == 'self_employed': score += 6 elif employment_status == 'business_owner': score += 10 else: score += 3 # Business type stability business_type = getattr(user, 'business_type', None) stable_businesses = ['technology', 'healthcare', 'education', 'government'] if business_type in stable_businesses: score += 3 # Account age (relationship length) if hasattr(user, 'date_joined'): account_age_days = (timezone.now() - user.date_joined).days if account_age_days >= 365: # 1+ years score += 4 elif account_age_days >= 180: # 6+ months score += 2 elif account_age_days >= 90: # 3+ months score += 1 return max(0, min(score, cls.MAX_EMPLOYMENT_SCORE)) except Exception as e: logger.error(f"Error calculating employment stability score for {user}: {str(e)}") return 5 @classmethod def calculate_total_score(cls, user): """ Calculate total credit score for a user """ try: repayment_score = cls.calculate_repayment_history_score(user) income_score = cls.calculate_income_score(user) rollover_score = cls.calculate_rollover_frequency_score(user) employment_score = cls.calculate_employment_stability_score(user) total_score = repayment_score + income_score + rollover_score + employment_score return { 'repayment_history_score': repayment_score, 'income_score': income_score, 'rollover_frequency_score': rollover_score, 'employment_stability_score': employment_score, 'total_score': total_score } except Exception as e: logger.error(f"Error calculating total score for {user}: {str(e)}") return { 'repayment_history_score': 0, 'income_score': 0, 'rollover_frequency_score': 0, 'employment_stability_score': 0, 'total_score': 0 } @classmethod def update_credit_score(cls, user): """ Update or create credit score for a user """ try: scores = cls.calculate_total_score(user) # Get or create LoanScoring record credit_score, created = LoanScoring.objects.get_or_create( user=user, defaults=scores ) if not created: # Update existing record credit_score.repayment_history_score = scores['repayment_history_score'] credit_score.income_score = scores['income_score'] credit_score.rollover_frequency_score = scores['rollover_frequency_score'] credit_score.employment_stability_score = scores['employment_stability_score'] credit_score.total_score = scores['total_score'] # Calculate risk level and credit limit credit_score.calculate_score() logger.info(f"Updated credit score for {user.get_full_name()}: {credit_score.total_score}") return credit_score except Exception as e: logger.error(f"Error updating credit score for {user}: {str(e)}") return None @classmethod def bulk_update_credit_scores(cls, user_queryset=None): """ Update credit scores for multiple users """ if user_queryset is None: user_queryset = User.objects.filter(role='borrower') updated_count = 0 error_count = 0 for user in user_queryset: try: cls.update_credit_score(user) updated_count += 1 except Exception as e: logger.error(f"Error updating credit score for {user}: {str(e)}") error_count += 1 logger.info(f"Bulk update complete: {updated_count} updated, {error_count} errors") return updated_count, error_count