Coverage for domain/pricing.py: 92%
250 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-04 17:24 +0000
1"""Product & Pricing Engine (PPE) + Rate Stack + Agentic Recommendation.
3Port of ``arive-spec/src/domain/pricing.ts`` (spec §16.2, §13.1).
4Pure, side-effect-free financial domain logic using Decimal exactness with half-up rounding.
5Zero dependencies on Django or external databases.
6"""
8from __future__ import annotations
10from collections.abc import Sequence
11from dataclasses import dataclass, field
12from datetime import date
13from decimal import ROUND_HALF_UP, Decimal
15from domain.calc import monthly_pi
16from domain.enums import (
17 CompType,
18 MortgageType,
19 OccupancyType,
20 PricingChannel,
21 PricingGoal,
22 RateType,
23)
25Number = Decimal | int | str | float
27_CENTS = Decimal("0.01")
28_MILLS = Decimal("0.001")
29_HUNDRED = Decimal("100")
30ZERO = Decimal("0")
33def _d(value: Number) -> Decimal:
34 """Coerce number to Decimal safely, routing floats through str to prevent binary float inaccuracies."""
35 if isinstance(value, Decimal):
36 return value
37 if isinstance(value, float): 37 ↛ 38line 37 didn't jump to line 38 because the condition on line 37 was never true
38 return Decimal(str(value))
39 return Decimal(value)
42def round2(n: Number) -> Decimal:
43 """Round to 2 decimal places using standard ROUND_HALF_UP."""
44 return _d(n).quantize(_CENTS, rounding=ROUND_HALF_UP)
47def round3(n: Number) -> Decimal:
48 """Round to 3 decimal places using standard ROUND_HALF_UP (mortgage price/rate convention)."""
49 return _d(n).quantize(_MILLS, rounding=ROUND_HALF_UP)
52@dataclass(frozen=True)
53class RateProductInput:
54 """Specification of an investor rate product."""
56 id: str
57 investor: str
58 name: str
59 mortgage_type: MortgageType | str
60 term_months: int
61 base_rate: Decimal
62 base_price: Decimal # price in points of par (100.000 = par)
63 min_fico: int
64 max_ltv: Decimal
65 occupancies: tuple[str, ...]
66 comp_type: CompType | str = CompType.LENDER_PAID
67 comp_pct: Decimal = Decimal("1.25")
69 def __post_init__(self) -> None:
70 object.__setattr__(self, "base_rate", round3(self.base_rate))
71 object.__setattr__(self, "base_price", round3(self.base_price))
72 object.__setattr__(self, "max_ltv", round2(self.max_ltv))
73 object.__setattr__(self, "comp_pct", round2(self.comp_pct))
74 object.__setattr__(
75 self,
76 "occupancies",
77 tuple(str(o) for o in self.occupancies),
78 )
81@dataclass(frozen=True)
82class PricingScenario:
83 """Pricing scenario inputs from loan parameters."""
85 loan_amount: Decimal
86 property_value: Decimal
87 fico: int
88 occupancy: OccupancyType | str
89 term_months: int = 360
90 monthly_taxes: Decimal = ZERO
91 monthly_hazard: Decimal = ZERO
92 monthly_hoa: Decimal = ZERO
93 monthly_mi: Decimal = ZERO
94 channel: PricingChannel | str = PricingChannel.WHOLESALE
95 comp_type: CompType | str | None = None
96 rate_type: RateType | str = RateType.FIXED
97 lock_period_days: int = 30
98 number_of_units: int = 1
99 non_occupant_co_borrower: bool = False
100 first_time_homebuyer: bool = False
101 self_employed: bool = False
103 def __post_init__(self) -> None:
104 object.__setattr__(self, "loan_amount", round2(self.loan_amount))
105 object.__setattr__(self, "property_value", round2(self.property_value))
106 object.__setattr__(self, "monthly_taxes", round2(self.monthly_taxes))
107 object.__setattr__(self, "monthly_hazard", round2(self.monthly_hazard))
108 object.__setattr__(self, "monthly_hoa", round2(self.monthly_hoa))
109 object.__setattr__(self, "monthly_mi", round2(self.monthly_mi))
112@dataclass(frozen=True)
113class PriceAdjustment:
114 code: str
115 description: str
116 adjustment_type: str # "PRICE" | "RATE"
117 amount: Decimal
120@dataclass(frozen=True)
121class ClosingCostItem:
122 label: str
123 amount: Decimal
126@dataclass(frozen=True)
127class ClosingCostBreakdown:
128 origination_amt: Decimal
129 third_party_amt: Decimal
130 prepaids_amt: Decimal
131 total: Decimal
132 items: list[ClosingCostItem] = field(default_factory=list)
134 # Aliases for backward parity with arive-spec
135 @property
136 def lender_fees(self) -> Decimal:
137 return self.origination_amt
139 @property
140 def title(self) -> Decimal:
141 return self.third_party_amt
143 @property
144 def prepaids(self) -> Decimal:
145 return self.prepaids_amt
148@dataclass(frozen=True)
149class LoCompDetail:
150 comp_type: str
151 pct: Decimal
152 amt: Decimal
155@dataclass(frozen=True)
156class RateStackItem:
157 rate: Decimal
158 price: Decimal
159 points_pct: Decimal
162@dataclass(frozen=True)
163class PricedProduct:
164 product: RateProductInput
165 rate: Decimal
166 price: Decimal
167 pi: Decimal
168 apr: Decimal
169 piti: Decimal
170 lo_comp: Decimal
171 lo_comp_detail: LoCompDetail
172 closing_costs: ClosingCostBreakdown
173 adjustments: list[PriceAdjustment]
174 rate_stack: list[RateStackItem]
175 lock_period_days: int
176 rate_sheet_id: str
177 eligible: bool
178 reasons: list[str]
181@dataclass(frozen=True)
182class IneligibleProduct:
183 product_id: str
184 product_name: str
185 reasons: list[str]
188@dataclass(frozen=True)
189class PricingResult:
190 count: int
191 products: list[PricedProduct]
192 ineligible_products: list[IneligibleProduct]
195@dataclass(frozen=True)
196class Recommendation:
197 best: PricedProduct
198 rationale: str
199 alternatives: list[PricedProduct]
202def approximate_apr(
203 loan_amount: Number,
204 note_rate: Number,
205 term_months: int,
206 finance_charge_dollars: Number,
207) -> Decimal:
208 """Solve for the exact APR via bisection numerical method over payment annuity stream."""
209 amt = _d(loan_amount)
210 rate = _d(note_rate)
211 charges = _d(finance_charge_dollars)
212 n = term_months or 360
214 pmt = monthly_pi(amt, rate, n)
215 financed = amt - charges
217 if financed <= 0 or pmt <= 0: 217 ↛ 218line 217 didn't jump to line 218 because the condition on line 217 was never true
218 return round2(rate)
220 def pv(i: float) -> float:
221 if i == 0: 221 ↛ 222line 221 didn't jump to line 222 because the condition on line 221 was never true
222 return float(pmt * n)
223 return float(pmt) * (1.0 - (1.0 + i) ** (-n)) / i
225 lo, hi = 0.0, 0.02 # up to 24% annual APR
226 target = float(financed)
228 for _ in range(60):
229 mid = (lo + hi) / 2.0
230 if pv(mid) > target:
231 lo = mid
232 else:
233 hi = mid
235 annual_apr = ((lo + hi) / 2.0) * 12.0 * 100.0
236 return round2(Decimal(str(annual_apr)))
239def price_adjustments(s: PricingScenario) -> list[PriceAdjustment]:
240 """Calculate Fannie Mae / Freddie Mac LLPA price adjustments based on scenario risks."""
241 if s.property_value <= 0: 241 ↛ 242line 241 didn't jump to line 242 because the condition on line 241 was never true
242 return []
244 ltv = (s.loan_amount / s.property_value) * _HUNDRED
245 adj: list[PriceAdjustment] = []
247 # FICO Score adjustments
248 if s.fico < 680:
249 adj.append(
250 PriceAdjustment(
251 code="LLPA_FICO",
252 description=f"Credit score {s.fico} < 680",
253 adjustment_type="PRICE",
254 amount=Decimal("1.25"),
255 )
256 )
257 elif s.fico < 740:
258 adj.append(
259 PriceAdjustment(
260 code="LLPA_FICO",
261 description=f"Credit score {s.fico} (680–739)",
262 adjustment_type="PRICE",
263 amount=Decimal("0.50"),
264 )
265 )
267 # LTV adjustment
268 if ltv > Decimal("80.00"):
269 adj.append(
270 PriceAdjustment(
271 code="LLPA_LTV",
272 description=f"LTV {round2(ltv)}% > 80%",
273 adjustment_type="PRICE",
274 amount=Decimal("0.75"),
275 )
276 )
278 # Occupancy adjustments
279 occ_str = str(s.occupancy)
280 if occ_str == OccupancyType.INVESTMENT.value:
281 adj.append(
282 PriceAdjustment(
283 code="LLPA_OCC",
284 description="Investment property",
285 adjustment_type="PRICE",
286 amount=Decimal("2.125"),
287 )
288 )
289 elif occ_str == OccupancyType.SECOND_HOME.value:
290 adj.append(
291 PriceAdjustment(
292 code="LLPA_OCC",
293 description="Second home",
294 adjustment_type="PRICE",
295 amount=Decimal("1.125"),
296 )
297 )
299 # Multi-unit property adjustment
300 if s.number_of_units >= 2:
301 adj.append(
302 PriceAdjustment(
303 code="LLPA_UNITS",
304 description=f"{s.number_of_units}-unit property",
305 adjustment_type="PRICE",
306 amount=Decimal("1.00"),
307 )
308 )
310 # Non-occupant co-borrower
311 if s.non_occupant_co_borrower:
312 adj.append(
313 PriceAdjustment(
314 code="LLPA_NONOCC",
315 description="Non-occupant co-borrower",
316 adjustment_type="PRICE",
317 amount=Decimal("0.375"),
318 )
319 )
321 return adj
324def price_scenario(
325 s: PricingScenario, products: Sequence[RateProductInput]
326) -> list[PricedProduct]:
327 """Price all candidate rate products against a borrower pricing scenario."""
328 if s.loan_amount <= 0 or s.property_value <= 0:
329 return []
331 ltv = round2((s.loan_amount / s.property_value) * _HUNDRED)
332 adjustments = price_adjustments(s)
333 total_adj = sum((a.amount for a in adjustments), ZERO)
334 escrow = s.monthly_taxes + s.monthly_hazard + s.monthly_hoa + s.monthly_mi
336 results: list[PricedProduct] = []
338 for p in products:
339 reasons: list[str] = []
340 if s.fico < p.min_fico:
341 reasons.append(f"FICO {s.fico} below {p.min_fico}")
342 if ltv > p.max_ltv:
343 reasons.append(f"LTV {ltv}% above {p.max_ltv}%")
344 if str(s.occupancy) not in p.occupancies:
345 reasons.append(f"Occupancy {s.occupancy} not allowed")
347 term = p.term_months or s.term_months
348 price = round3(p.base_price - total_adj)
349 pi = monthly_pi(s.loan_amount, p.base_rate, term)
351 # Points cost relative to par (100.00). Below par = borrower cost (positive discount points)
352 points_cost = round2(((Decimal("100") - price) / _HUNDRED) * s.loan_amount)
353 origination_amt = round2(max(ZERO, points_cost) + Decimal("1295"))
354 title_amt = round2(s.loan_amount * Decimal("0.004") + Decimal("650"))
355 prepaids_amt = round2(escrow * Decimal("3") + s.monthly_hazard * Decimal("12"))
357 items = [
358 ClosingCostItem(label="Origination charges (A)", amount=origination_amt),
359 ClosingCostItem(label="Third-party services (B/C)", amount=title_amt),
360 ClosingCostItem(label="Prepaids & escrow (F/G)", amount=prepaids_amt),
361 ]
362 total_closing_costs = round2(origination_amt + title_amt + prepaids_amt)
363 closing_costs = ClosingCostBreakdown(
364 origination_amt=origination_amt,
365 third_party_amt=title_amt,
366 prepaids_amt=prepaids_amt,
367 total=total_closing_costs,
368 items=items,
369 )
371 comp_pct = p.comp_pct if p.comp_pct > ZERO else Decimal("1.25")
372 lo_comp = round2(s.loan_amount * (comp_pct / _HUNDRED))
373 comp_type_val = str(s.comp_type or p.comp_type)
374 lo_comp_detail = LoCompDetail(
375 comp_type=comp_type_val,
376 pct=comp_pct,
377 amt=lo_comp,
378 )
380 finance_charge = origination_amt + (
381 lo_comp if comp_type_val == CompType.BORROWER_PAID.value else ZERO
382 )
383 apr = approximate_apr(s.loan_amount, p.base_rate, term, finance_charge)
384 piti = round2(pi + escrow)
386 # Rate stack: 9 rungs around base note rate in 0.125% steps (slope: 4 pts price / 1% rate)
387 deltas = (
388 Decimal("-0.500"),
389 Decimal("-0.375"),
390 Decimal("-0.250"),
391 Decimal("-0.125"),
392 Decimal("0.000"),
393 Decimal("0.125"),
394 Decimal("0.250"),
395 Decimal("0.375"),
396 Decimal("0.500"),
397 )
398 rate_stack = [
399 RateStackItem(
400 rate=round3(p.base_rate + d),
401 price=round3(price - d * Decimal("4")),
402 points_pct=round3(Decimal("100") - (price - d * Decimal("4"))),
403 )
404 for d in deltas
405 ]
407 today_str = date.today().isoformat()
408 rate_sheet_id = f"RS-{p.id}-{today_str}"
410 results.append(
411 PricedProduct(
412 product=p,
413 rate=p.base_rate,
414 price=price,
415 pi=pi,
416 apr=apr,
417 piti=piti,
418 lo_comp=lo_comp,
419 lo_comp_detail=lo_comp_detail,
420 closing_costs=closing_costs,
421 adjustments=adjustments,
422 rate_stack=rate_stack,
423 lock_period_days=s.lock_period_days,
424 rate_sheet_id=rate_sheet_id,
425 eligible=len(reasons) == 0,
426 reasons=reasons,
427 )
428 )
430 return results
433def to_pricing_result(priced: Sequence[PricedProduct]) -> PricingResult:
434 """Format priced products into a full PricingResult object."""
435 eligible = [p for p in priced if p.eligible]
436 ineligible = [
437 IneligibleProduct(
438 product_id=p.product.id,
439 product_name=p.product.name,
440 reasons=p.reasons,
441 )
442 for p in priced
443 if not p.eligible
444 ]
445 return PricingResult(
446 count=len(eligible),
447 products=eligible,
448 ineligible_products=ineligible,
449 )
452def price_manual(
453 s: PricingScenario,
454 investor: str,
455 name: str,
456 rate: Number,
457 price: Number,
458 term_months: int | None = None,
459 comp_pct: Number = Decimal("1.25"),
460) -> PricedProduct:
461 """Price a manual / non-integrated lender scenario."""
462 clean_investor = investor.replace(" ", "_")
463 product = RateProductInput(
464 id=f"MANUAL-{clean_investor}",
465 investor=investor,
466 name=name,
467 mortgage_type=MortgageType.CONVENTIONAL,
468 term_months=term_months or s.term_months,
469 base_rate=_d(rate),
470 base_price=_d(price),
471 min_fico=0,
472 max_ltv=Decimal("100.00"),
473 occupancies=(str(s.occupancy),),
474 comp_type=CompType.BORROWER_PAID,
475 comp_pct=_d(comp_pct),
476 )
477 return price_scenario(s, [product])[0]
480def recommend(
481 priced: Sequence[PricedProduct], goal: PricingGoal | str
482) -> Recommendation | None:
483 """Agentic recommendation pick: rank eligible products according to borrower goal."""
484 eligible = [p for p in priced if p.eligible]
485 if not eligible: 485 ↛ 486line 485 didn't jump to line 486 because the condition on line 485 was never true
486 return None
488 goal_str = str(goal)
490 def score(p: PricedProduct) -> Decimal:
491 if goal_str == PricingGoal.LOWEST_PAYMENT.value:
492 return -p.pi
493 if goal_str == PricingGoal.BEST_LONG_TERM.value: 493 ↛ 494line 493 didn't jump to line 494 because the condition on line 493 was never true
494 return -(p.rate * Decimal(p.product.term_months or 360))
495 if goal_str == PricingGoal.LEAST_CASH_TO_CLOSE.value:
496 return p.price
497 if goal_str == PricingGoal.FASTEST_CLOSE.value: 497 ↛ 498line 497 didn't jump to line 498 because the condition on line 497 was never true
498 is_conv = Decimal("1") if str(p.product.mortgage_type) == MortgageType.CONVENTIONAL.value else ZERO
499 return is_conv - (p.rate / _HUNDRED)
500 if goal_str == PricingGoal.MAX_CASH_OUT.value: 500 ↛ 502line 500 didn't jump to line 502 because the condition on line 500 was always true
501 return p.product.max_ltv
502 if goal_str == PricingGoal.CREDIT_SENSITIVE.value:
503 return -Decimal(p.product.min_fico)
504 return ZERO
506 ranked = sorted(eligible, key=score, reverse=True)
507 best = ranked[0]
509 rationales: dict[str, str] = {
510 PricingGoal.LOWEST_PAYMENT.value: f"Lowest monthly P&I (${best.pi}) among eligible products.",
511 PricingGoal.BEST_LONG_TERM.value: f"Lowest lifetime interest at {best.rate}% over {(best.product.term_months or 360) // 12} years.",
512 PricingGoal.LEAST_CASH_TO_CLOSE.value: f"Best price ({best.price}) applies the most credit toward closing costs.",
513 PricingGoal.FASTEST_CLOSE.value: f"{best.product.mortgage_type} product with Day-1 Certainty eligibility for a faster close.",
514 PricingGoal.MAX_CASH_OUT.value: f"Allows up to {best.product.max_ltv}% — the most equity for cash-out.",
515 PricingGoal.CREDIT_SENSITIVE.value: f"Most forgiving FICO floor ({best.product.min_fico}).",
516 }
518 return Recommendation(
519 best=best,
520 rationale=rationales.get(goal_str, "Best matching product for your scenario."),
521 alternatives=ranked[1:3],
522 )