Lending Tech: Credit Scoring and Loan Origination

class CreditBureauData:
    """Represents credit bureau report data."""
    
    def __init__(self, bureau: str, report: dict):
        self.bureau = bureau
        self.report_date = report.get('report_date')
        
        # Account history
        self.total_accounts = report.get('total_accounts', 0)
        self.credit_limit = report.get('credit_limit', 0)
        self.balance = report.get('balance', 0)
        self.utilization = self.balance / max(1, self.credit_limit)
        
        # Payment history
        self.on_time_payments = report.get('on_time_pct', 0)
        self.delinquencies = report.get('delinquencies', 0)
        self.collections = report.get('collections', 0)
        
        # Derogatory marks
        self.bankruptcies = report.get('bankruptcies', 0)
        self.foreclosures = report.get('foreclosures', 0)
        self.judgments = report.get('judgments', 0)
        
        # Inquiries
        self.hard_inquiries = report.get('hard_inquiries', 0)
        self.inquiries_last_6m = report.get('inquiries_6m', 0)
        
        # Age of credit
        self.oldest_account_months = report.get('oldest_account_months', 0)
        self.average_account_age = report.get('avg_account_age', 0)

class BankTransactionAnalyzer:
    def __init__(self, categorization_model):
        self.categorizer = categorization_model
    
    def analyze_transactions(self, transactions: List[Transaction]) -> dict:
        """Analyze bank transactions for credit assessment."""
        
        # Income analysis
        income_transactions = [t for t in transactions if t.amount > 0]
        monthly_income = self._calculate_monthly_income(income_transactions)
        
        # Expense analysis
        expense_categories = self.categorizer.categorize(transactions)
        
        # Stability metrics
        income_stability = self._calculate_income_stability(income_transactions)
        
        # Behavioral indicators
        cash_flow_patterns = self._analyze_cash_flow(transactions)
        
        return {
            'monthly_income': monthly_income,
            'monthly_expenses': sum(expense_categories.values()),
            'income_stability_score': income_stability,
            'cash_flow_positive': cash_flow_patterns['positive_months'] > 8,
            'savings_rate': cash_flow_patterns['savings_rate'],
            'expense_categories': expense_categories,
            'overdraft_count': len([t for t in transactions if t.is_overdraft]),
            'insufficient_funds_count': len([
                t for t in transactions if t.is_insufficient_funds
            ])
        }
    
    def _calculate_income_stability(self, income_transactions: List[Transaction]) -> float:
        """Calculate income stability score (0-1)."""
        if not income_transactions:
            return 0.0
        
        # Count income sources
        unique_sources = len(set(t.source for t in income_transactions))
        
        # Calculate coefficient of variation
        monthly_incomes = self._group_by_month(income_transactions)
        if len(monthly_incomes) < 3:
            return 0.5
        
        mean = sum(monthly_incomes.values()) / len(monthly_incomes)
        variance = sum((v - mean) ** 2 for v in monthly_incomes.values()) / len(monthly_incomes)
        std_dev = variance ** 0.5
        cv = std_dev / max(1, mean)
        
        # Lower CV = higher stability (max 1.0)
        return max(0, 1 - cv)

class CreditScoringModel:
    def __init__(self, features: dict, weights: dict):
        self.features = features
        self.weights = weights
    
    def calculate_score(
        self,
        bureau_data: CreditBureauData,
        bank_data: dict,
        application_data: dict
    ) -> CreditScore:
        """Calculate overall credit score from multiple data sources."""
        
        # Component scores
        bureau_score = self._score_bureau_data(bureau_data)
        bank_score = self._score_bank_data(bank_data)
        application_score = self._score_application(application_data)
        
        # Weighted combination
        final_score = (
            bureau_score * self.weights['bureau'] +
            bank_score * self.weights['bank'] +
            application_score * self.weights['application']
        )
        
        # Convert to standard score
        standard_score = self._to_standard_score(final_score)
        
        return CreditScore(
            score=standard_score,
            score_range='exceptional' if standard_score >= 800
                else 'very_good' if standard_score >= 740
                else 'good' if standard_score >= 670
                else 'fair' if standard_score >= 580
                else 'poor',
            components={
                'bureau': bureau_score,
                'bank': bank_score,
                'application': application_score
            },
            risk_factors=self._identify_risk_factors(
                bureau_data, bank_data, application_data
            ),
            positive_factors=self._identify_positive_factors(
                bureau_data, bank_data, application_data
            )
        )
    
    def _score_bureau_data(self, data: CreditBureauData) -> float:
        """Score based on credit bureau data."""
        score = 0
        factors = []
        
        # Utilization (30% weight)
        if data.utilization < 0.1:
            score += 30
        elif data.utilization < 0.3:
            score += 25
        elif data.utilization < 0.5:
            score += 15
        elif data.utilization < 0.75:
            score += 5
        
        # Payment history (35% weight)
        score += min(35, data.on_time_payments * 0.35)
        
        # Derogatory marks (15% weight)
        if data.bankruptcies > 0:
            score -= 15
        if data.foreclosures > 0:
            score -= 10
        if data.collections > 3:
            score -= 10
        
        # Credit age (10% weight)
        score += min(10, data.oldest_account_months / 24)
        
        # Inquiries (10% weight)
        if data.inquiries_last_6m <= 2:
            score += 10
        elif data.inquiries_last_6m <= 5:
            score += 5
        
        return max(0, min(100, score))

from dataclasses import dataclass
from datetime import datetime, date
from typing import Optional, List
from enum import Enum
import uuid

class LoanPurpose(Enum):
    PERSONAL = "personal"
    DEBT_CONSOLIDATION = "debt_consolidation"
    HOME_IMPROVEMENT = "home_improvement"
    AUTO = "auto"
    SMALL_BUSINESS = "small_business"
    MEDICAL = "medical"

class LoanStatus(Enum):
    APPLICATION = "application"
    SUBMITTED = "submitted"
    IN_REVIEW = "in_review"
    APPROVED = "approved"
    DENIED = "denied"
    FUNDED = "funded"
    IN_REPAYMENT = "in_repayment"
    DEFAULTED = "defaulted"
    PAID_OFF = "paid_off"

class DecisionType(Enum):
    AUTO_APPROVE = "auto_approve"
    MANUAL_REVIEW = "manual_review"
    AUTO_DENY = "auto_deny"

@dataclass
class LoanApplication:
    application_id: str
    applicant: dict
    requested_amount: float
    requested_term: int  # months
    purpose: LoanPurpose
    employment: dict
    income: dict
    assets: dict
    submitted_at: datetime
    
    def __post_init__(self):
        if not self.application_id:
            self.application_id = str(uuid.uuid4())


class LoanOriginationSystem:
    def __init__(
        self,
        credit_scoring: CreditScoringModel,
        fraud_detector,
        underwriting_rules,
        document_processor,
        pricing_engine
    ):
        self.credit_scoring = credit_scoring
        self.fraud_detector = fraud_detector
        self.underwriting_rules = underwriting_rules
        self.document_processor = document_processor
        self.pricing_engine = pricing_engine
    
    async def submit_application(self, application: LoanApplication) -> ApplicationResult:
        """Process loan application through origination workflow."""
        
        # Step 1: Initial validation
        validation_result = self._validate_application(application)
        if not validation_result.valid:
            return ApplicationResult(
                application_id=application.application_id,
                status=LoanStatus.DENIED,
                decision=DecisionType.AUTO_DENY,
                reasons=validation_result.errors
            )
        
        # Step 2: Credit check
        credit_data = await self._fetch_credit_data(application)
        credit_score = self.credit_scoring.calculate_score(
            credit_data['bureau'],
            credit_data['bank'],
            application.__dict__
        )
        
        # Step 3: Fraud detection
        fraud_result = await self.fraud_detector.screen(application)
        if fraud_result.high_risk:
            return ApplicationResult(
                application_id=application.application_id,
                status=LoanStatus.DENIED,
                decision=DecisionType.AUTO_DENY,
                reasons=['Fraud risk detected']
            )
        
        # Step 4: Underwriting rules
        decision = self.underwriting_rules.evaluate(
            application=application,
            credit_score=credit_score,
            fraud_result=fraud_result
        )
        
        if decision.decision == DecisionType.AUTO_DENY:
            return ApplicationResult(
                application_id=application.application_id,
                status=LoanStatus.DENIED,
                decision=DecisionType.AUTO_DENY,
                reasons=decision.reasons,
                credit_score=credit_score.score
            )
        
        # Step 5: Pricing
        pricing = self.pricing_engine.calculate_pricing(
            amount=application.requested_amount,
            term=application.requested_term,
            credit_score=credit_score.score,
            purpose=application.purpose
        )
        
        if decision.decision == DecisionType.MANUAL_REVIEW:
            return ApplicationResult(
                application_id=application.application_id,
                status=LoanStatus.IN_REVIEW,
                decision=DecisionType.MANUAL_REVIEW,
                credit_score=credit_score.score,
                pricing=pricing,
                requires_documents=decision.required_documents
            )
        
        # Auto-approve
        return ApplicationResult(
            application_id=application.application_id,
            status=LoanStatus.APPROVED,
            decision=DecisionType.AUTO_APPROVE,
            credit_score=credit_score.score,
            pricing=pricing
        )
    
    def _validate_application(self, application: LoanApplication) -> ValidationResult:
        """Validate application completeness and eligibility."""
        errors = []
        
        # Required fields
        if not application.applicant.get('ssn'):
            errors.append("SSN required")
        if not application.applicant.get('address'):
            errors.append("Address required")
        
        # Amount limits
        if application.requested_amount < 1000:
            errors.append("Minimum loan amount is $1,000")
        if application.requested_amount > 50000:
            errors.append("Maximum loan amount is $50,000")
        
        # Income requirements
        monthly_income = application.income.get('monthly', 0)
        requested_payment = self._estimate_payment(
            application.requested_amount,
            application.requested_term,
            0.36  # Assume worst-case rate for estimation
        )
        
        if monthly_income < requested_payment * 3:
            errors.append("Income does not support requested loan")
        
        # Employment requirements
        employment_length = application.employment.get('months_employed', 0)
        if employment_length < 3:
            errors.append("Minimum 3 months employment required")
        
        return ValidationResult(
            valid=len(errors) == 0,
            errors=errors
        )


@dataclass
class ApplicationResult:
    application_id: str
    status: LoanStatus
    decision: DecisionType
    reasons: List[str] = None
    credit_score: int = None
    pricing: dict = None
    requires_documents: List[str] = None

class LoanPricingEngine:
    def __init__(self, base_rates: dict, risk_adjustments: dict):
        self.base_rates = base_rates
        self.risk_adjustments = risk_adjustments
    
    def calculate_pricing(
        self,
        amount: float,
        term: int,
        credit_score: int,
        purpose: LoanPurpose
    ) -> PricingResult:
        """Calculate loan pricing based on risk assessment."""
        
        # Get base rate for term
        base_rate = self._get_base_rate(term)
        
        # Adjust for credit score
        score_adjustment = self._get_score_adjustment(credit_score)
        
        # Adjust for loan purpose
        purpose_adjustment = self.risk_adjustments['purpose'].get(
            purpose.value, 0
        )
        
        # Adjust for loan amount
        amount_adjustment = self._get_amount_adjustment(amount)
        
        # Calculate final rate
        final_rate = base_rate + score_adjustment + purpose_adjustment + amount_adjustment
        final_rate = max(self.base_rates['minimum'], final_rate)
        
        # Calculate payment
        monthly_payment = self._calculate_payment(amount, final_rate, term)
        
        # Calculate APR (includes fees)
        fees = self._calculate_fees(amount)
        apr = self._calculate_apr(amount, fees, monthly_payment, term)
        
        return PricingResult(
            interest_rate=final_rate,
            apr=apr,
            monthly_payment=monthly_payment,
            total_interest=monthly_payment * term - amount,
            total_cost=monthly_payment * term + fees,
            fees=fees
        )
    
    def _calculate_payment(
        self,
        principal: float,
        annual_rate: float,
        term_months: int
    ) -> float:
        """Calculate monthly payment using standard amortization formula."""
        monthly_rate = annual_rate / 12 / 100
        
        if monthly_rate == 0:
            return principal / term_months
        
        payment = principal * (monthly_rate * (1 + monthly_rate) ** term_months)
        payment /= ((1 + monthly_rate) ** term_months - 1)
        
        return round(payment, 2)

class FraudDetectionSystem:
    def __init__(
        self,
        identity_verifier,
        device_fingerprint,
        behavioral_analysis,
        network_analyzer
    ):
        self.identity = identity_verifier
        self.device = device_fingerprint
        self.behavioral = behavioral_analysis
        self.network = network_analyzer
    
    async def screen(self, application: LoanApplication) -> FraudScreeningResult:
        """Perform comprehensive fraud screening."""
        
        fraud_signals = []
        risk_score = 0
        
        # Identity verification
        identity_result = await self.identity.verify(application.applicant)
        if not identity_result.verified:
            risk_score += 40
            fraud_signals.append({
                'type': 'identity',
                'severity': 'high',
                'description': 'Identity verification failed'
            })
        
        if identity_result.risk_flags:
            risk_score += 20
            fraud_signals.extend([
                {'type': 'identity', 'severity': 'medium', 'description': flag}
                for flag in identity_result.risk_flags
            ])
        
        # Device fingerprinting
        device_result = self.device.analyze(application.device_info)
        if device_result.risk_score > 0.7:
            risk_score += 30
            fraud_signals.append({
                'type': 'device',
                'severity': 'high',
                'description': f"High-risk device: {device_result.risk_factors}"
            })
        
        # Behavioral analysis
        behavioral_result = await self.behavioral.analyze(
            application.applicant_id,
            application.device_info
        )
        if behavioral_result.anomaly_detected:
            risk_score += 25
            fraud_signals.append({
                'type': 'behavioral',
                'severity': 'medium',
                'description': 'Behavioral anomaly detected'
            })
        
        # Network analysis
        network_result = await self.network.analyze(
            application.applicant,
            application.device_info
        )
        
        # Check for fraud rings
        if network_result.fraud_ring_detected:
            risk_score += 50
            fraud_signals.append({
                'type': 'network',
                'severity': 'critical',
                'description': 'Connection to known fraud ring'
            })
        
        # Velocity checks
        if network_result.application_velocity > 5:
            risk_score += 20
            fraud_signals.append({
                'type': 'velocity',
                'severity': 'medium',
                'description': 'High application velocity'
            })
        
        return FraudScreeningResult(
            risk_score=min(100, risk_score),
            high_risk=risk_score >= 70,
            signals=fraud_signals,
            recommendation='decline' if risk_score >= 70
                else 'review' if risk_score >= 40
                else 'approve'
        )

class LendingCompliance:
    def __init__(self, rules_engine):
        self.rules = rules_engine
    
    def validate_pricing_disclosure(
        self,
        pricing: PricingResult,
        application: LoanApplication
    ) -> DisclosureResult:
        """Validate required disclosures are provided."""
        
        disclosures = []
        
        # APR disclosure
        disclosures.append({
            'type': 'APR',
            'value': pricing.apr,
            'required': True,
            'timing': 'before consummation'
        })
        
        # Payment schedule
        disclosures.append({
            'type': 'payment_schedule',
            'value': {
                'monthly': pricing.monthly_payment,
                'total': pricing.total_cost,
                'total_interest': pricing.total_interest
            },
            'required': True,
            'timing': 'before consummation'
        })
        
        # Fee disclosures
        if pricing.fees['origination'] > 0:
            disclosures.append({
                'type': 'origination_fee',
                'value': pricing.fees['origination'],
                'required': True,
                'timing': 'before consummation'
            })
        
        return DisclosureResult(
            disclosures=disclosures,
            complete=len([d for d in disclosures if d['required']]) == len(disclosures)
        )
    
    def check_adverse_action(
        self,
        application: LoanApplication,
        result: ApplicationResult
    ) -> AdverseActionNotice:
        """Generate adverse action notice if application is denied."""
        
        if result.status != LoanStatus.DENIED:
            return None
        
        # Identify reason for denial
        reasons = []
        
        if 'credit' in [r.lower() for r in result.reasons]:
            reasons.append({
                'factor': 'credit_history',
                'code': '001',
                'description': 'Credit history insufficient'
            })
        
        if 'income' in [r.lower() for r in result.reasons]:
            reasons.append({
                'factor': 'income',
                'code': '002',
                'description': 'Income does not meet requirements'
            })
        
        # Include credit score if adverse action
        if result.credit_score:
            reasons.append({
                'factor': 'credit_score',
                'code': '003',
                'description': f'Credit score: {result.credit_score}'
            })
        
        return AdverseActionNotice(
            application_id=application.application_id,
            applicant=application.applicant,
            reasons=reasons,
            credit_report_provider='Experian',  # Example
            date_sent=datetime.now()
        )