Coverage for apps/pricing_ppe/api.py: 89%

1"""Pricing router: product pricing, ad-hoc quote, and rate lock (mounted under /api/loans). 

2 

3Thin ORM<->domain adapter over ``domain/pricing.py`` — the engine stays pure. Follows the 

4credit/AUS router shape: ``_scoped_loan`` + ``ensure_staff_scope`` + ``requires(...)`` guards. 

5""" 

6 

7from __future__ import annotations 

8 

9from datetime import timedelta 

10from decimal import Decimal 

11from uuid import UUID 

12 

13from django.utils import timezone 

14from ninja import Router 

15 

16from apps.accounts.auth import ensure_staff_scope, requires 

17from apps.core_los.models import Loan, LoanTransaction 

18from apps.pricing_ppe.models import RateLock, RateProduct 

19from apps.pricing_ppe.schemas import ( 

20 PricingResponseOut, 

21 PricingResultOut, 

22 QuoteIn, 

23 RateLockCreate, 

24 RateLockOut, 

25 RateLockReadOut, 

26) 

27from domain.pricing import ( 

28 PricingScenario, 

29 RateProductInput, 

30 price_scenario, 

31 recommend, 

32 to_pricing_result, 

33) 

34from domain.result import AppError, DomainError 

35 

36router = Router(tags=["pricing"]) 

37 

38_CENTS = Decimal("0.01") 

39_HUNDRED = Decimal("100") 

40 

41 

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

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

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

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

46 ensure_staff_scope(request.principal, loan) 

47 return loan 

48 

49 

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

51 try: 

52 return loan.transaction 

53 except LoanTransaction.DoesNotExist: 

54 return None 

55 

56 

57def representative_fico(loan: Loan) -> int | None: 

58 """Loan-level representative FICO: the lowest per-borrower mid score across the loan's 

59 most recent credit order (the standard 'representative credit score' rule).""" 

60 order = loan.credit_orders.order_by("-created_at").first() 

61 if order is None: 

62 return None 

63 mids = [s.mid for s in order.scores.all() if s.mid] 

64 return min(mids) if mids else None 

65 

66 

67def _active_products() -> list[RateProductInput]: 

68 return [p.to_domain() for p in RateProduct.objects.filter(is_active=True)] 

69 

70 

71def _build_scenario(loan: Loan, overrides: QuoteIn | None = None) -> PricingScenario: 

72 txn = _txn(loan) 

73 o = overrides or QuoteIn() 

74 txn_amount = txn.base_loan_amt if txn else None 

75 loan_amount = o.loan_amount if o.loan_amount is not None else txn_amount 

76 stored_value = (txn.appraised_value_amt or txn.purchase_price_amt) if txn else None 

77 property_value = o.property_value if o.property_value is not None else stored_value 

78 fico = o.fico if o.fico is not None else representative_fico(loan) 

79 

80 if loan_amount is None or property_value is None: 80 ↛ 81line 80 didn't jump to line 81 because the condition on line 80 was never true

81 raise DomainError( 

82 [AppError("PRICING_INCOMPLETE", "Loan amount and property value are required to price")] 

83 ) 

84 if fico is None: 

85 raise DomainError( 

86 [AppError("PRICING_NO_FICO", "No representative FICO; pull credit or supply a score")] 

87 ) 

88 

89 kwargs: dict = { 

90 "loan_amount": loan_amount, 

91 "property_value": property_value, 

92 "fico": fico, 

93 "occupancy": o.occupancy or loan.occupancy, 

94 "term_months": o.term_months or (txn.term_months if txn else 360), 

95 } 

96 for fld in ("monthly_taxes", "monthly_hazard", "monthly_hoa", "monthly_mi"): 

97 val = getattr(o, fld) 

98 if val is not None: 98 ↛ 99line 98 didn't jump to line 99 because the condition on line 98 was never true

99 kwargs[fld] = val 

100 return PricingScenario(**kwargs) 

101 

102 

103@router.get("/{guid}/pricing/products", response=PricingResultOut, auth=requires("LOAN_VIEW")) 

104def pricing_products(request, guid: UUID): 

105 loan = _scoped_loan(request, guid) 

106 priced = price_scenario(_build_scenario(loan), _active_products()) 

107 return to_pricing_result(priced) 

108 

109 

110@router.post("/{guid}/pricing/quote", response=PricingResponseOut, auth=requires("LOAN_VIEW")) 

111def pricing_quote(request, guid: UUID, payload: QuoteIn): 

112 loan = _scoped_loan(request, guid) 

113 priced = price_scenario(_build_scenario(loan, payload), _active_products()) 

114 result = to_pricing_result(priced) 

115 return { 

116 "count": result.count, 

117 "products": result.products, 

118 "ineligible_products": result.ineligible_products, 

119 "recommendation": recommend(priced, payload.goal) if payload.goal else None, 

120 } 

121 

122 

123@router.get("/{guid}/pricing/lock", response=RateLockReadOut, auth=requires("LOAN_VIEW")) 

124def get_lock(request, guid: UUID): 

125 loan = _scoped_loan(request, guid) 

126 try: 

127 return {"lock": loan.rate_lock} 

128 except RateLock.DoesNotExist: 

129 return {"lock": None} 

130 

131 

132@router.post("/{guid}/pricing/lock", response={201: RateLockOut}, auth=requires("LOAN_EDIT")) 

133def create_lock(request, guid: UUID, payload: RateLockCreate): 

134 loan = _scoped_loan(request, guid) 

135 if RateLock.objects.filter(loan=loan).exists(): 

136 raise DomainError([AppError("CONFLICT", "Loan already has a rate lock")]) 

137 product = RateProduct.objects.filter(id=payload.product_id, is_active=True).first() 

138 if product is None: 138 ↛ 139line 138 didn't jump to line 139 because the condition on line 138 was never true

139 raise DomainError([AppError("NOT_FOUND", "Rate product not found")]) 

140 

141 lock_days = payload.lock_period_days or 30 

142 txn = _txn(loan) 

143 loan_amt = txn.base_loan_amt if txn else Decimal("0") 

144 lo_comp_amt = (loan_amt * (product.comp_pct / _HUNDRED)).quantize(_CENTS) 

145 locked_rate = payload.rate if payload.rate is not None else product.base_rate 

146 locked_price = payload.price if payload.price is not None else product.base_price 

147 

148 lock = RateLock.objects.create( 

149 loan=loan, 

150 product=product, 

151 locked_rate=locked_rate, 

152 locked_price=locked_price, 

153 lock_period_days=lock_days, 

154 expires_at=timezone.now() + timedelta(days=lock_days), 

155 lo_comp_amt=lo_comp_amt, 

156 lo_comp_type=product.comp_type, 

157 ) 

158 return 201, lock 

159 

160 

161@router.get("/{guid}/lock", response=RateLockReadOut, auth=requires("LOAN_VIEW")) 

162def get_lock_alias(request, guid: UUID): 

163 return get_lock(request, guid) 

164 

165 

166@router.post("/{guid}/lock", response={201: RateLockOut}, auth=requires("LOAN_EDIT")) 

167def create_lock_alias(request, guid: UUID, payload: RateLockCreate): 

168 return create_lock(request, guid, payload)