from decimal import Decimal, ROUND_HALF_UP from sqlalchemy.orm import Session from sqlalchemy.exc import IntegrityError from fastapi import HTTPException, status from . import repository from .schema import LoancalculationCreate, LoancalculationUpdate from .models import Amortizationscheduleentry def calculate_monthly_payment(principal: Decimal, annual_rate: Decimal, term_months: int) -> Decimal: """ Calculate monthly payment using standard amortization formula. M = P[r(1+r)^n]/[(1+r)^n-1] Special case: if rate is 0, M = P/n """ if annual_rate == Decimal("0"): return (principal / Decimal(term_months)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) monthly_rate = annual_rate / Decimal("1200") factor = (Decimal("1") + monthly_rate) ** term_months monthly_payment = (principal * monthly_rate * factor) / (factor - Decimal("1")) return monthly_payment.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) def generate_amortization_schedule( principal: Decimal, annual_rate: Decimal, term_months: int, monthly_payment: Decimal ) -> list: """ Generate complete amortization schedule entries. Returns list of dicts with payment breakdown for each period. """ schedule = [] remaining_balance = principal cumulative_interest = Decimal("0") cumulative_principal = Decimal("0") monthly_rate = annual_rate / Decimal("1200") for payment_num in range(1, term_months + 1): if annual_rate == Decimal("0"): interest_portion = Decimal("0") else: interest_portion = (remaining_balance * monthly_rate).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) principal_portion = monthly_payment - interest_portion if payment_num == term_months: principal_portion = remaining_balance payment_amount = principal_portion + interest_portion else: payment_amount = monthly_payment remaining_balance = remaining_balance - principal_portion if remaining_balance < Decimal("0.01") and remaining_balance > Decimal("-0.01"): remaining_balance = Decimal("0") cumulative_interest += interest_portion cumulative_principal += principal_portion schedule.append({ "payment_number": payment_num, "payment_date": None, "payment_amount": payment_amount.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP), "principal_portion": principal_portion.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP), "interest_portion": interest_portion.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP), "remaining_balance": remaining_balance.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP), "cumulative_interest": cumulative_interest.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP), "cumulative_principal": cumulative_principal.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP), }) return schedule def create_loan_calculation_with_schedule(db: Session, data: LoancalculationCreate): """ Create loan calculation and generate complete amortization schedule in single transaction. Batch write service implementing business rules for loan calculation and schedule generation. """ principal = data.principal_amount annual_rate = data.annual_interest_rate term_months = data.loan_term_months monthly_payment = calculate_monthly_payment(principal, annual_rate, term_months) total_amount_paid = (monthly_payment * Decimal(term_months)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) total_interest_paid = (total_amount_paid - principal).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) calculation_data = { "principal_amount": principal, "annual_interest_rate": annual_rate, "loan_term_months": term_months, "monthly_payment": monthly_payment, "total_amount_paid": total_amount_paid, "total_interest_paid": total_interest_paid, } try: calculation = repository.create(db, calculation_data) db.flush() schedule_entries = generate_amortization_schedule(principal, annual_rate, term_months, monthly_payment) for entry_data in schedule_entries: entry_data["calculation_id"] = calculation.id schedule_entry = Amortizationscheduleentry(**entry_data) db.add(schedule_entry) db.commit() db.refresh(calculation) return calculation except IntegrityError as exc: db.rollback() msg = str(exc.orig).lower() if exc.orig else "" if "unique" in msg or "duplicate" in msg: raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Resource already exists.") if "foreign key" in msg: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Referenced resource does not exist.") raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Data integrity error.") except HTTPException: db.rollback() raise except Exception: db.rollback() raise def get_loan_calculation(db: Session, calculation_id: str): """ Retrieve loan calculation summary by ID. """ calculation = repository.get_by_id(db, calculation_id) if not calculation: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Loan calculation not found.") return calculation def get_loan_calculation_with_schedule(db: Session, calculation_id: str): """ Retrieve loan calculation with complete amortization schedule using eager loading. """ calculation = repository.get_with_details(db, calculation_id) if not calculation: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Loan calculation not found.") return calculation def list_loan_calculations(db: Session, limit: int, offset: int, **filters) -> dict: """ List all loan calculations with pagination. """ items = repository.list_all(db, limit=limit, offset=offset, **filters) total = repository.count_all(db, **filters) return {"items": items, "total": total, "limit": limit, "offset": offset} def update_loan_calculation(db: Session, calculation_id: str, data: LoancalculationUpdate): """ Update loan calculation fields. """ update_data = data.model_dump(exclude_unset=True) if not update_data: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="No fields provided for update.") try: calculation = repository.update(db, calculation_id, update_data) if not calculation: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Loan calculation not found.") db.commit() db.refresh(calculation) return calculation except IntegrityError as exc: db.rollback() msg = str(exc.orig).lower() if exc.orig else "" if "unique" in msg or "duplicate" in msg: raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Resource already exists.") if "foreign key" in msg: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Referenced resource does not exist.") raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Data integrity error.") except HTTPException: db.rollback() raise except Exception: db.rollback() raise def delete_loan_calculation(db: Session, calculation_id: str): """ Delete loan calculation and cascade delete all associated amortization schedule entries. """ try: deleted = repository.delete(db, calculation_id) if not deleted: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Loan calculation not found.") db.commit() return {"message": "Loan calculation deleted successfully."} except IntegrityError as exc: db.rollback() msg = str(exc.orig).lower() if exc.orig else "" if "foreign key" in msg: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="Cannot delete loan calculation while referenced by other resources." ) raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Data integrity error.") except HTTPException: db.rollback() raise except Exception: db.rollback() raise