Coverage for apps/underwriting/aus_api.py: 94%

1"""AUS router: eligibility pre-flight + dual DU/LPA runs + comparison (under /api/loans).""" 

2 

3import uuid 

4from decimal import Decimal 

5from uuid import UUID 

6 

7from ninja import Router 

8 

9from apps.accounts.auth import ensure_staff_scope, requires 

10from apps.application_1003.models import Declaration, Demographics, SubjectProperty 

11from apps.core_los.models import Loan, LoanTransaction 

12from apps.underwriting.models import AusRun 

13from apps.underwriting.providers import get_aus_engine 

14from apps.underwriting.schemas import AusComparisonOut, AusEligibilityOut, AusRunOut 

15from domain.aus import AusBorrower, AusContext, assert_aus_ready, check_aus_eligibility 

16from domain.calc import monthly_pi 

17from domain.qualification import HousingItem, IncomeItem, LiabilityItem, qualify 

18from domain.result import AppError, DomainError 

19 

20router = Router(tags=["aus"]) 

21 

22 

23def _scoped_loan(request, guid: UUID) -> Loan: 

24 loan = Loan.objects.filter(guid=guid).first() 

25 if loan is None: 25 ↛ 26line 25 didn't jump to line 26 because the condition on line 25 was never true

26 raise DomainError([AppError("NOT_FOUND", "Loan not found")]) 

27 ensure_staff_scope(request.principal, loan) 

28 return loan 

29 

30 

31def _txn(loan: Loan) -> LoanTransaction | None: 

32 try: 

33 return loan.transaction 

34 except LoanTransaction.DoesNotExist: 

35 return None 

36 

37 

38def _build_ctx(loan: Loan) -> AusContext: 

39 txn = _txn(loan) 

40 value = (txn.appraised_value_amt or txn.purchase_price_amt) if txn else None 

41 borrowers = [ 

42 AusBorrower( 

43 first_name=b.first_name, 

44 last_name=b.last_name, 

45 is_primary=b.is_primary, 

46 active=True, 

47 has_ssn=b.ssn_encrypted is not None, 

48 has_dob=b.birth_date is not None, 

49 credit_auth=b.credit_auth_ind, 

50 has_demographics=Demographics.objects.filter(borrower=b).exists(), 

51 has_declarations=Declaration.objects.filter(borrower=b).exists(), 

52 has_current_residence=b.residences.filter(residence_type="Current").exists(), 

53 ) 

54 for b in loan.borrowers.all() 

55 ] 

56 sp = SubjectProperty.objects.filter(loan=loan).first() 

57 return AusContext( 

58 credit_pulled=loan.credit_pulled, 

59 stage=loan.stage, 

60 purpose=loan.purpose, 

61 base_loan_amt=txn.base_loan_amt if txn else None, 

62 property_value=value, 

63 borrowers=borrowers, 

64 income_count=loan.incomes.count(), 

65 asset_count=loan.assets.count(), 

66 has_subject_address=bool(sp and sp.address_line), 

67 ) 

68 

69 

70def _ratios(loan: Loan) -> dict: 

71 txn = _txn(loan) 

72 incomes = [IncomeItem(i.monthly_amt, i.current_ind) for i in loan.incomes.all()] 

73 liabilities = [ 

74 LiabilityItem(liab.monthly_payment_amt, liab.exclude_from_dti_ind, liab.omit_ind) 

75 for liab in loan.liabilities.all() 

76 ] 

77 housing = [HousingItem(h.payment_amt, h.timing_type) for h in loan.housing_expenses.all()] 

78 fallback = ( 

79 monthly_pi(txn.base_loan_amt, txn.note_rate_percent or 0, txn.term_months) if txn else None 

80 ) 

81 q = qualify(incomes, liabilities, housing, fallback_pi=fallback) 

82 ltv = txn.ltv_ratio_percent if (txn and txn.ltv_ratio_percent is not None) else Decimal("0.00") 

83 cltv = txn.cltv_ratio_percent if (txn and txn.cltv_ratio_percent is not None) else ltv 

84 return { 

85 "total_dti_pct": q.total_dti_pct, 

86 "housing_dti_pct": q.housing_dti_pct, 

87 "ltv_pct": ltv, 

88 "cltv_pct": cltv, 

89 } 

90 

91 

92def _eligibility_payload(elig) -> dict: 

93 return { 

94 "du_eligible": elig.du_eligible, 

95 "lpa_eligible": elig.lpa_eligible, 

96 "issues": [ 

97 { 

98 "category": i.category, 

99 "code": i.code, 

100 "message": i.message, 

101 "engines": i.engines, 

102 "severity": i.severity, 

103 } 

104 for i in elig.issues 

105 ], 

106 } 

107 

108 

109@router.get("/{guid}/aus/check", response=AusEligibilityOut, auth=requires("LOAN_VIEW")) 

110def aus_check(request, guid: UUID): 

111 loan = _scoped_loan(request, guid) 

112 return _eligibility_payload(check_aus_eligibility(_build_ctx(loan))) 

113 

114 

115def _run(request, guid: UUID, engine: str): 

116 loan = _scoped_loan(request, guid) 

117 ready = assert_aus_ready(_build_ctx(loan), engine) 

118 if not ready.ok: 

119 raise DomainError(ready.errors) 

120 result = get_aus_engine(org=loan.org_unit).run(engine, _ratios(loan)) 

121 casefile_prefix = "DU" if engine.lower() == "du" else "LPA" 

122 casefile_id = f"{casefile_prefix}-{uuid.uuid4().hex[:8].upper()}" 

123 run = AusRun.objects.create( 

124 loan=loan, 

125 engine=engine, 

126 casefile_id=casefile_id, 

127 recommendation=result.recommendation, 

128 risk_class=result.risk_class, 

129 housing_pct=result.housing_pct, 

130 dti_pct=result.dti_pct, 

131 ltv_pct=result.ltv_pct, 

132 cltv_pct=result.cltv_pct, 

133 messages=result.messages, 

134 ) 

135 return 201, run 

136 

137 

138@router.post("/{guid}/aus/du", response={201: AusRunOut}, auth=requires("LOAN_EDIT")) 

139def run_du(request, guid: UUID): 

140 return _run(request, guid, "du") 

141 

142 

143@router.post("/{guid}/aus/lpa", response={201: AusRunOut}, auth=requires("LOAN_EDIT")) 

144def run_lpa(request, guid: UUID): 

145 return _run(request, guid, "lpa") 

146 

147 

148@router.get("/{guid}/aus", response=AusComparisonOut, auth=requires("LOAN_VIEW")) 

149def aus_results(request, guid: UUID): 

150 loan = _scoped_loan(request, guid) 

151 runs = list(loan.aus_runs.all().order_by("-submitted_at")) 

152 du = next((r for r in runs if r.engine == "du"), None) 

153 lpa = next((r for r in runs if r.engine == "lpa"), None) 

154 return {"du": du, "lpa": lpa, "runs": runs} 

155 

156 

157@router.get("/{guid}/aus/compare", response=AusComparisonOut, auth=requires("LOAN_VIEW")) 

158def aus_compare(request, guid: UUID): 

159 return aus_results(request, guid)