Python 67%
TypeScript 18.2%
CSS 14.4%
1# -----------------------------------------------------------------------------2# Immo-Ka — Agrégateur de maisons à vendre (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# mortgage/api.py : routes /api/mortgage/* — taux courants, meilleur taux,5# historique, santé des providers, calculateur canadien, abordabilité.6# Lecture seule sur mortgage.db (le scheduler écrit) ; cache applicatif7# 5 min sur les lectures ; jamais d'internals de scraping exposés (le champ8# raw est retiré au niveau du store).9# -----------------------------------------------------------------------------10from __future__ import annotations1112import time1314from fastapi import APIRouter, Body, HTTPException, Query1516from . import calc, cmhc, store1718router = APIRouter(prefix="/api/mortgage", tags=["mortgage"])1920CACHE_TTL_S = 30021_cache: dict[str, tuple[float, object]] = {}2223RATE_TYPES = {"fixed", "variable", "adjustable", "other"}24KINDS = {"posted", "special"}25INSURED = {"insured", "insurable", "uninsured", "unknown"}262728def _cached(key: str, builder):29 hit = _cache.get(key)30 now = time.time()31 if hit and now - hit[0] < CACHE_TTL_S:32 return hit[1]33 value = builder()34 _cache[key] = (now, value)35 if len(_cache) > 512: # borne mémoire : purge des entrées expirées36 for k in [k for k, (t, _) in _cache.items() if now - t >= CACHE_TTL_S]:37 _cache.pop(k, None)38 return value394041def _check(value, allowed: set, label: str):42 if value is not None and value not in allowed:43 raise HTTPException(422, f"{label} invalide : {value}")44 return value454647@router.get("/rates")48def rates(rate_type: str | None = None, term_months: int | None = None,49 kind: str | None = None, insured_status: str | None = None,50 purpose: str | None = Query("purchase"),51 provider: str | None = None):52 """Taux courants (dernière donnée valide par produit), filtrables."""53 _check(rate_type, RATE_TYPES, "rate_type")54 _check(kind, KINDS, "kind")55 _check(insured_status, INSURED, "insured_status")56 if purpose in ("", "all"):57 purpose = None58 key = f"rates|{rate_type}|{term_months}|{kind}|{insured_status}|{purpose}|{provider}"5960 def build():61 con = store.connect()62 rows = store.current_rates(63 con, rate_type=rate_type, term_months=term_months, kind=kind,64 insured_status=insured_status, purpose=purpose, provider=provider)65 con.close()66 return {"count": len(rows), "rates": rows,67 "stale_hours_threshold": store.STALE_H}68 return _cached(key, build)697071@router.get("/rates/best")72def rates_best(rate_type: str = "fixed", term_months: int = 60,73 insured_status: str | None = None, purpose: str = "purchase"):74 """Meilleur taux courant par produit comparable + comparateur par banque."""75 _check(rate_type, RATE_TYPES, "rate_type")76 _check(insured_status, INSURED, "insured_status")77 key = f"best|{rate_type}|{term_months}|{insured_status}|{purpose}"7879 def build():80 con = store.connect()81 best = store.best_rate(con, rate_type=rate_type,82 term_months=term_months,83 insured_status=insured_status, purpose=purpose)84 con.close()85 if best is None:86 raise HTTPException(404, "Aucun taux courant pour ces critères")87 return best88 return _cached(key, build)899091@router.get("/rates/history")92def rates_history(rate_type: str = "fixed", term_months: int = 60,93 provider: str | None = None, kind: str | None = None,94 days: int = 365):95 """Périodes de validité par produit — reconstruit « le taux X à date ». """96 _check(rate_type, RATE_TYPES, "rate_type")97 _check(kind, KINDS, "kind")98 days = max(1, min(days, 730))99 key = f"hist|{rate_type}|{term_months}|{provider}|{kind}|{days}"100101 def build():102 con = store.connect()103 rows = store.history(con, rate_type, term_months,104 provider=provider, kind=kind, days=days)105 con.close()106 return {"count": len(rows), "days": days, "history": rows}107 return _cached(key, build)108109110@router.get("/providers")111def providers():112 """Santé des connecteurs de taux (OK / WARNING / ERROR + fraîcheur)."""113 def build():114 from .providers import PROVIDERS115 con = store.connect()116 health = store.provider_health(con)117 con.close()118 names = {slug: cls.institution for slug, cls in PROVIDERS.items()}119 urls = {slug: cls.source_url for slug, cls in PROVIDERS.items()}120 out = []121 for h in health:122 out.append({123 "provider": h["provider"],124 "institution": names.get(h["provider"], h["provider"]),125 "source_url": urls.get(h["provider"]),126 "level": h["level"],127 "status": h["status"],128 "age_minutes": h["age_minutes"],129 "current_products": h["current_products"],130 "last_data_at": h["last_data_at"],131 })132 return {"providers": out,133 "registered": sorted(names),134 "stale_hours_threshold": store.STALE_H}135 return _cached("providers", build)136137138@router.get("/market")139def market(rate_type: str = "fixed", term_months: int = 60):140 """Métriques Mortgage Intelligence pour un produit donné."""141 _check(rate_type, RATE_TYPES, "rate_type")142143 def build():144 con = store.connect()145 stats = store.market_stats(con, rate_type=rate_type,146 term_months=term_months)147 con.close()148 if stats is None:149 raise HTTPException(404, "Aucun taux courant pour ces critères")150 return stats151 return _cached(f"market|{rate_type}|{term_months}", build)152153154@router.get("/intelligence")155def intelligence():156 """Vue d'ensemble du marché : les produits phares en un appel."""157 def build():158 con = store.connect()159 combos = [("fixed", 12), ("fixed", 36), ("fixed", 48),160 ("fixed", 60), ("fixed", 120), ("variable", 60)]161 grid = []162 for rtype, term in combos:163 s = store.market_stats(con, rate_type=rtype, term_months=term)164 if s:165 grid.append(s)166 prime_rows = store.current_rates(con, rate_type="other")167 con.close()168 prime = [{"institution": r["institution"], "rate": r["rate"],169 "product_name": r["product_name"],170 "age_minutes": r["age_minutes"]}171 for r in prime_rows if "préférentiel" in172 (r["product_name"] or "").lower() or "prime" in173 (r["product_name"] or "").lower()]174 return {"products": grid, "prime_rates": prime}175 return _cached("intelligence", build)176177178# ---------------------------------------------------------------------------179# Calculateur180# ---------------------------------------------------------------------------181182def _pick_rate(rate_type: str, term_months: int,183 insured_status: str | None) -> dict | None:184 con = store.connect()185 best = store.best_rate(con, rate_type=rate_type, term_months=term_months,186 insured_status=insured_status)187 con.close()188 return best189190191def _validate_scenario(price: float, down: float, amort: int, term: int,192 frequency: str) -> None:193 if price <= 0 or price > 100_000_000:194 raise HTTPException(422, "Prix invalide")195 if down < 0 or down >= price:196 raise HTTPException(422, "Mise de fonds invalide")197 if amort not in calc.AMORTIZATIONS_YEARS and not (5 <= amort <= 30):198 raise HTTPException(422, "Amortissement invalide (5–30 ans)")199 if not 3 <= term <= 120:200 raise HTTPException(422, "Terme invalide (3–120 mois)")201 if frequency not in calc.FREQUENCIES:202 raise HTTPException(422, f"Fréquence invalide : {frequency}")203204205@router.post("/calculate")206def calculate(body: dict = Body(...)):207 """Calcul hypothécaire canadien complet pour un scénario.208209 Entrées : price, down_payment (ou down_payment_pct), amortization_years,210 term_months, frequency, rate_type, rate (sinon meilleur taux observé),211 insured_status?, include_schedule?, income?, other_debts_monthly?,212 property_tax_monthly?, heating_monthly?, condo_fees_monthly?.213 """214 try:215 price = float(body.get("price") or 0)216 if body.get("down_payment") is not None:217 down = float(body["down_payment"])218 else:219 down = price * float(body.get("down_payment_pct") or 20) / 100220 amort = int(body.get("amortization_years") or 25)221 term = int(body.get("term_months") or 60)222 frequency = str(body.get("frequency") or "monthly")223 rate_type = str(body.get("rate_type") or "fixed")224 except (TypeError, ValueError):225 raise HTTPException(422, "Paramètres numériques invalides")226 _check(rate_type, {"fixed", "variable"}, "rate_type")227 _validate_scenario(price, down, amort, term, frequency)228229 quote = cmhc.insurance_quote(price, down, amort)230 rate_source = None231 rate = body.get("rate")232 if rate is None:233 insured = "insured" if quote["required"] and quote["eligible"] else None234 best = _pick_rate(rate_type, term, insured)235 if best is None:236 raise HTTPException(237 503, "Aucun taux courant disponible — réessayez plus tard")238 rate = best["rate"]239 rate_source = {k: best[k] for k in240 ("provider", "institution", "product_name", "kind",241 "rate", "apr", "insured_status", "source_url",242 "last_checked", "age_minutes", "stale")}243 rate = float(rate)244 if not 0 < rate <= 25:245 raise HTTPException(422, "Taux invalide")246 if quote["required"] and not quote["eligible"]:247 principal = price - down # non assurable : calcul quand même, signalé248 else:249 principal = quote["total_mortgage"] if quote["required"] else price - down250251 compounding = "semi-annual" if rate_type == "fixed" else "monthly"252 pay = calc.payment(principal, rate, amort, frequency, compounding)253 monthly_eq = calc.payment(principal, rate, amort, "monthly", compounding)254 q_rate = calc.qualifying_rate(rate)255 q_pay = calc.payment(principal, q_rate, amort, frequency, compounding)256 out = {257 "inputs": {"price": price, "down_payment": round(down, 2),258 "down_payment_pct": round(down / price * 100, 2),259 "rate": rate, "rate_type": rate_type,260 "term_months": term, "amortization_years": amort,261 "frequency": frequency, "compounding": compounding},262 "insurance": quote,263 "principal": round(principal, 2),264 "payment": pay,265 "payment_monthly_equivalent": monthly_eq,266 "qualifying": {"rate": q_rate, "payment": q_pay,267 "note": "Test de résistance : max(taux + 2, 5,25 %)"},268 "term": calc.term_summary(principal, rate, amort, term,269 frequency, compounding),270 "stress": calc.stress_scenarios(principal, rate, amort,271 frequency, compounding),272 "renewal": calc.renewal_scenarios(principal, rate, amort, term,273 frequency, compounding),274 "payoff_years": calc.payoff_years(principal, rate, amort,275 frequency, compounding),276 "rate_source": rate_source,277 }278 rows = calc.schedule(principal, rate, amort, frequency, compounding)279 out["annual"] = calc.annual_rollup(rows, frequency)280 if body.get("include_schedule"):281 out["schedule"] = rows282 income = body.get("income")283 if income:284 out["ratios"] = calc.gds_tds(285 float(income), monthly_eq,286 float(body.get("property_tax_monthly") or 0),287 float(body.get("heating_monthly") or 0),288 float(body.get("condo_fees_monthly") or 0),289 float(body.get("other_debts_monthly") or 0))290 return out291292293@router.post("/affordability")294def affordability(body: dict = Body(...)):295 """Capacité d'achat : prix max selon un versement cible OU selon les296 revenus (ABD/ATD au taux de qualification) ; taux requis pour un297 versement cible vs meilleur taux observé."""298 try:299 amort = int(body.get("amortization_years") or 25)300 term = int(body.get("term_months") or 60)301 frequency = str(body.get("frequency") or "monthly")302 rate_type = str(body.get("rate_type") or "fixed")303 down = float(body.get("down_payment") or 0)304 except (TypeError, ValueError):305 raise HTTPException(422, "Paramètres numériques invalides")306 _check(rate_type, {"fixed", "variable"}, "rate_type")307 if frequency not in calc.FREQUENCIES:308 raise HTTPException(422, f"Fréquence invalide : {frequency}")309310 rate = body.get("rate")311 rate_source = None312 if rate is None:313 best = _pick_rate(rate_type, term, None)314 if best is None:315 raise HTTPException(316 503, "Aucun taux courant disponible — réessayez plus tard")317 rate = best["rate"]318 rate_source = {k: best[k] for k in319 ("provider", "institution", "product_name", "kind",320 "rate", "source_url", "age_minutes", "stale")}321 rate = float(rate)322 if not 0 < rate <= 25:323 raise HTTPException(422, "Taux invalide")324 compounding = "semi-annual" if rate_type == "fixed" else "monthly"325 q_rate = calc.qualifying_rate(rate)326 out: dict = {"rate": rate, "qualifying_rate": q_rate,327 "rate_source": rate_source}328329 target = body.get("target_payment_monthly")330 if target:331 target = float(target)332 loan = calc.max_loan(target, rate, amort, "monthly", compounding)333 loan_q = calc.max_loan(target, q_rate, amort, "monthly", compounding)334 out["from_payment"] = {335 "target_payment_monthly": target,336 "max_loan": loan, "max_price": round(loan + down, 2),337 "max_loan_stress_tested": loan_q,338 "max_price_stress_tested": round(loan_q + down, 2),339 }340 income = body.get("income")341 if income:342 income = float(income)343 gds_room = income / 12 * 0.39 \344 - float(body.get("property_tax_monthly") or 0) \345 - float(body.get("heating_monthly") or 0) \346 - float(body.get("condo_fees_monthly") or 0) * 0.5347 tds_room = income / 12 * 0.44 \348 - float(body.get("property_tax_monthly") or 0) \349 - float(body.get("heating_monthly") or 0) \350 - float(body.get("condo_fees_monthly") or 0) * 0.5 \351 - float(body.get("other_debts_monthly") or 0)352 room = max(0.0, min(gds_room, tds_room))353 loan_q = calc.max_loan(room, q_rate, amort, "monthly", compounding)354 out["from_income"] = {355 "income": income, "max_payment_monthly": round(room, 2),356 "max_loan_stress_tested": loan_q,357 "max_price_estimate": round(loan_q + down, 2),358 "note": ("Indicatif seulement (ABD 39 % / ATD 44 % au taux de "359 "qualification) — ne constitue pas une préapprobation."),360 }361 principal = body.get("principal")362 target_rate_payment = body.get("required_rate_for_payment")363 if principal and target_rate_payment:364 req = calc.required_rate(float(principal), float(target_rate_payment),365 amort, "monthly", compounding)366 out["required_rate"] = {367 "principal": float(principal),368 "target_payment_monthly": float(target_rate_payment),369 "rate": req,370 "achievable_now": req is not None and req >= rate,371 "best_observed": rate,372 }373 return out374