Coverage for domain/fees.py: 92%
150 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-04 14:06 +0000
1"""TRID fee tolerance curing and QM points-and-fees engine.
3Port of ``arive-spec/src/domain/fees.ts`` (spec §16.1, CFPB Reg Z 12 CFR § 1026.43(e)(3)).
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
12from decimal import ROUND_HALF_UP, Decimal
13from typing import Any
15from domain.enums import (
16 SECTION_TOLERANCE,
17 FeeSection,
18 PaidByType,
19 PaidToType,
20 QmType,
21 ToleranceBucket,
22)
24Number = Decimal | int | str | float
26_CENTS = Decimal("0.01")
27_HUNDRED = Decimal("100")
28_TEN_PERCENT_FACTOR = Decimal("1.10")
29ZERO = Decimal("0")
32def _d(value: Number) -> Decimal:
33 """Coerce number to Decimal safely, routing floats through str to prevent binary float inaccuracies."""
34 if isinstance(value, Decimal):
35 return value
36 if isinstance(value, float): 36 ↛ 37line 36 didn't jump to line 37 because the condition on line 36 was never true
37 return Decimal(str(value))
38 return Decimal(value)
41def round2(n: Number) -> Decimal:
42 """Round to 2 decimal places using standard ROUND_HALF_UP (US mortgage convention)."""
43 return _d(n).quantize(_CENTS, rounding=ROUND_HALF_UP)
46@dataclass(frozen=True)
47class FeeInput:
48 """Snapshot representation of a fee item for pure domain calculations."""
50 section: FeeSection | str
51 total_amt: Number
52 tolerance_bucket: ToleranceBucket | str | None = None
53 poc_ind: bool = False
54 name: str = ""
55 paid_by: PaidByType | str = PaidByType.BORROWER
56 paid_to: PaidToType | str = PaidToType.THIRD_PARTY
59def _get_field(obj: Any, *keys: str, default: Any = None) -> Any:
60 for key in keys:
61 if isinstance(obj, dict) and key in obj: 61 ↛ 62line 61 didn't jump to line 62 because the condition on line 61 was never true
62 return obj[key]
63 if hasattr(obj, key):
64 val = getattr(obj, key)
65 if val is not None:
66 return val
67 return default
70def bucket_of(fee: Any) -> ToleranceBucket:
71 """Derive the tolerance bucket for a fee item.
73 If an explicit `tolerance_bucket` is present, it is returned.
74 Otherwise, the default bucket for the TRID section (`SECTION_TOLERANCE`) is returned.
75 """
76 bucket_val = _get_field(fee, "tolerance_bucket", "ToleranceBucket")
77 if bucket_val:
78 return ToleranceBucket(str(bucket_val))
80 section_val = _get_field(fee, "section", "Section")
81 if section_val: 81 ↛ 88line 81 didn't jump to line 88 because the condition on line 81 was always true
82 try:
83 sec = FeeSection(str(section_val))
84 return SECTION_TOLERANCE.get(sec, ToleranceBucket.UNLIMITED)
85 except ValueError:
86 return ToleranceBucket.UNLIMITED
88 return ToleranceBucket.UNLIMITED
91@dataclass(frozen=True)
92class BucketCure:
93 """Baseline, revised, variance, and required cure for a single tolerance bucket."""
95 baseline: Decimal
96 revised: Decimal
97 variance: Decimal
98 cure: Decimal
101@dataclass(frozen=True)
102class ToleranceResult:
103 """Result of TRID tolerance cure comparison across all three buckets."""
105 by_bucket: dict[ToleranceBucket | str, BucketCure]
106 total_cure: Decimal
109def compute_tolerance_cure(
110 baseline: Sequence[Any], revised: Sequence[Any]
111) -> ToleranceResult:
112 """Compare baseline (disclosed) vs revised fees on a Change of Circumstance (CoC).
114 - ZERO bucket: Any aggregate increase is cured in full (cure = max(0, variance)).
115 - TEN_PERCENT bucket: Cures only the aggregate increase beyond 10% (cure = max(0, rev - base * 1.10)).
116 - UNLIMITED bucket: No tolerance cure (cure = 0).
117 - Fees flagged with `poc_ind = True` (Paid Outside of Closing) are excluded from tolerance totals.
118 """
119 def _sum_bucket(fees: Sequence[Any], bucket: ToleranceBucket) -> Decimal:
120 total = ZERO
121 for f in fees:
122 if bucket_of(f) != bucket:
123 continue
124 is_poc = bool(_get_field(f, "poc_ind", "PocInd", default=False))
125 if is_poc:
126 continue
127 amt = _get_field(f, "total_amt", "TotalAmt", default=ZERO)
128 total += _d(amt)
129 return total
131 buckets: tuple[ToleranceBucket, ...] = (
132 ToleranceBucket.ZERO,
133 ToleranceBucket.TEN_PERCENT,
134 ToleranceBucket.UNLIMITED,
135 )
136 by_bucket: dict[ToleranceBucket | str, BucketCure] = {}
137 total_cure = ZERO
139 for b in buckets:
140 base = round2(_sum_bucket(baseline, b))
141 rev = round2(_sum_bucket(revised, b))
142 variance = round2(rev - base)
144 if b == ToleranceBucket.ZERO:
145 cure = max(ZERO, variance)
146 elif b == ToleranceBucket.TEN_PERCENT:
147 cure = max(ZERO, round2(rev - round2(base * _TEN_PERCENT_FACTOR)))
148 else:
149 cure = ZERO
151 cure_rounded = round2(cure)
152 by_bucket[b] = BucketCure(
153 baseline=base,
154 revised=rev,
155 variance=variance,
156 cure=cure_rounded,
157 )
158 total_cure += cure_rounded
160 return ToleranceResult(by_bucket=by_bucket, total_cure=round2(total_cure))
163# ── QM points-and-fees test (Reg Z general QM thresholds, 2024 tiers) ──
165@dataclass(frozen=True)
166class QmTier:
167 min_amount: Decimal
168 fixed_cap: Decimal | None
169 pct_cap: Decimal | None
172QM_TIERS: tuple[QmTier, ...] = (
173 QmTier(min_amount=Decimal("130000"), fixed_cap=None, pct_cap=Decimal("0.03")),
174 QmTier(min_amount=Decimal("78000"), fixed_cap=Decimal("3900"), pct_cap=None),
175 QmTier(min_amount=Decimal("26000"), fixed_cap=None, pct_cap=Decimal("0.05")),
176 QmTier(min_amount=Decimal("13000"), fixed_cap=Decimal("1300"), pct_cap=None),
177 QmTier(min_amount=Decimal("0"), fixed_cap=None, pct_cap=Decimal("0.08")),
178)
181@dataclass(frozen=True)
182class QmFinding:
183 code: str
184 severity: str # "pass" | "warn" | "fail"
185 message: str
188@dataclass(frozen=True)
189class QmResult:
190 loan_amount: Decimal
191 points_and_fees: Decimal
192 threshold: Decimal
193 pct: Decimal
194 passed: bool
195 qm_type: QmType
196 apr_spread: Decimal
197 findings: list[QmFinding]
199 @property
200 def pass_status(self) -> bool:
201 return self.passed
203 def __getattr__(self, name: str) -> Any:
204 if name == "pass":
205 return self.passed
206 raise AttributeError(f"'{type(self).__name__}' object has no attribute '{name}'")
208 def __getitem__(self, item: str) -> Any:
209 if item == "pass": 209 ↛ 211line 209 didn't jump to line 211 because the condition on line 209 was always true
210 return self.passed
211 return getattr(self, item)
214def qm_points_and_fees_test(
215 loan_amount: Number,
216 points_and_fees: Number,
217 apr: Number | None = None,
218 apor: Number | None = None,
219) -> QmResult:
220 """CFPB Reg Z ATR/QM points-and-fees test and APR-spread price cap test.
222 - Evaluates points & fees against statutory 5-tier thresholds based on loan amount.
223 - Evaluates APR spread over APOR:
224 - spread <= 1.50% -> Safe Harbor QM
225 - spread <= 2.25% -> Rebuttable Presumption QM (HPML)
226 - spread > 2.25% -> Non-QM (fails General QM price cap)
227 """
228 loan_amt = _d(loan_amount)
229 pf = round2(points_and_fees)
231 tier = next(t for t in QM_TIERS if loan_amt >= t.min_amount)
232 if tier.pct_cap is not None:
233 threshold = round2(loan_amt * tier.pct_cap)
234 else:
235 assert tier.fixed_cap is not None
236 threshold = round2(tier.fixed_cap)
238 pf_pass = pf <= threshold
239 apr_spread = round2(_d(apr) - _d(apor)) if apr is not None and apor is not None else ZERO
241 findings: list[QmFinding] = []
242 if pf_pass:
243 findings.append(
244 QmFinding(
245 code="QM-PF",
246 severity="pass",
247 message=f"Points & fees ${pf} within the ${threshold} cap.",
248 )
249 )
250 else:
251 findings.append(
252 QmFinding(
253 code="QM-PF",
254 severity="fail",
255 message=f"Points & fees ${pf} exceed the ${threshold} cap.",
256 )
257 )
259 if not pf_pass:
260 qm_type = QmType.NON_QM
261 findings.append(
262 QmFinding(
263 code="QM-TYPE",
264 severity="fail",
265 message="Loan is Non-QM: points-and-fees cap exceeded.",
266 )
267 )
268 elif apr is None or apor is None:
269 qm_type = QmType.GENERAL
270 findings.append(
271 QmFinding(
272 code="QM-APR",
273 severity="warn",
274 message="APR/APOR not supplied — price-based Safe Harbor tier undetermined.",
275 )
276 )
277 elif apr_spread <= Decimal("1.50"):
278 qm_type = QmType.SAFE_HARBOR
279 findings.append(
280 QmFinding(
281 code="QM-APR",
282 severity="pass",
283 message=f"APR spread {apr_spread}% ≤ 1.5% — Safe Harbor.",
284 )
285 )
286 elif apr_spread <= Decimal("2.25"):
287 qm_type = QmType.REBUTTABLE_PRESUMPTION
288 findings.append(
289 QmFinding(
290 code="QM-APR",
291 severity="warn",
292 message=f"APR spread {apr_spread}% in (1.5%, 2.25%] — Rebuttable Presumption (HPML).",
293 )
294 )
295 else:
296 qm_type = QmType.NON_QM
297 findings.append(
298 QmFinding(
299 code="QM-APR",
300 severity="fail",
301 message=f"APR spread {apr_spread}% > 2.25% — exceeds General QM price cap.",
302 )
303 )
305 pct = round2((pf / loan_amt) * _HUNDRED) if loan_amt > 0 else ZERO
306 overall_pass = pf_pass and qm_type != QmType.NON_QM
308 return QmResult(
309 loan_amount=loan_amt,
310 points_and_fees=pf,
311 threshold=threshold,
312 pct=pct,
313 passed=overall_pass,
314 qm_type=qm_type,
315 apr_spread=apr_spread,
316 findings=findings,
317 )