# ----------------------------------------------------------------------------- # Immo-Ka — Agrégateur de maisons à vendre (province de Québec) # Auteur : Simon-Pierre Boucher — contact@spboucher.ai # mortgage/api.py : routes /api/mortgage/* — taux courants, meilleur taux, # historique, santé des providers, calculateur canadien, abordabilité. # Lecture seule sur mortgage.db (le scheduler écrit) ; cache applicatif # 5 min sur les lectures ; jamais d'internals de scraping exposés (le champ # raw est retiré au niveau du store). # ----------------------------------------------------------------------------- from __future__ import annotations import time from fastapi import APIRouter, Body, HTTPException, Query from . import calc, cmhc, store router = APIRouter(prefix="/api/mortgage", tags=["mortgage"]) CACHE_TTL_S = 300 _cache: dict[str, tuple[float, object]] = {} RATE_TYPES = {"fixed", "variable", "adjustable", "other"} KINDS = {"posted", "special"} INSURED = {"insured", "insurable", "uninsured", "unknown"} def _cached(key: str, builder): hit = _cache.get(key) now = time.time() if hit and now - hit[0] < CACHE_TTL_S: return hit[1] value = builder() _cache[key] = (now, value) if len(_cache) > 512: # borne mémoire : purge des entrées expirées for k in [k for k, (t, _) in _cache.items() if now - t >= CACHE_TTL_S]: _cache.pop(k, None) return value def _check(value, allowed: set, label: str): if value is not None and value not in allowed: raise HTTPException(422, f"{label} invalide : {value}") return value @router.get("/rates") def rates(rate_type: str | None = None, term_months: int | None = None, kind: str | None = None, insured_status: str | None = None, purpose: str | None = Query("purchase"), provider: str | None = None): """Taux courants (dernière donnée valide par produit), filtrables.""" _check(rate_type, RATE_TYPES, "rate_type") _check(kind, KINDS, "kind") _check(insured_status, INSURED, "insured_status") if purpose in ("", "all"): purpose = None key = f"rates|{rate_type}|{term_months}|{kind}|{insured_status}|{purpose}|{provider}" def build(): con = store.connect() rows = store.current_rates( con, rate_type=rate_type, term_months=term_months, kind=kind, insured_status=insured_status, purpose=purpose, provider=provider) con.close() return {"count": len(rows), "rates": rows, "stale_hours_threshold": store.STALE_H} return _cached(key, build) @router.get("/rates/best") def rates_best(rate_type: str = "fixed", term_months: int = 60, insured_status: str | None = None, purpose: str = "purchase"): """Meilleur taux courant par produit comparable + comparateur par banque.""" _check(rate_type, RATE_TYPES, "rate_type") _check(insured_status, INSURED, "insured_status") key = f"best|{rate_type}|{term_months}|{insured_status}|{purpose}" def build(): con = store.connect() best = store.best_rate(con, rate_type=rate_type, term_months=term_months, insured_status=insured_status, purpose=purpose) con.close() if best is None: raise HTTPException(404, "Aucun taux courant pour ces critères") return best return _cached(key, build) @router.get("/rates/history") def rates_history(rate_type: str = "fixed", term_months: int = 60, provider: str | None = None, kind: str | None = None, days: int = 365): """Périodes de validité par produit — reconstruit « le taux X à date ». """ _check(rate_type, RATE_TYPES, "rate_type") _check(kind, KINDS, "kind") days = max(1, min(days, 730)) key = f"hist|{rate_type}|{term_months}|{provider}|{kind}|{days}" def build(): con = store.connect() rows = store.history(con, rate_type, term_months, provider=provider, kind=kind, days=days) con.close() return {"count": len(rows), "days": days, "history": rows} return _cached(key, build) @router.get("/providers") def providers(): """Santé des connecteurs de taux (OK / WARNING / ERROR + fraîcheur).""" def build(): from .providers import PROVIDERS con = store.connect() health = store.provider_health(con) con.close() names = {slug: cls.institution for slug, cls in PROVIDERS.items()} urls = {slug: cls.source_url for slug, cls in PROVIDERS.items()} out = [] for h in health: out.append({ "provider": h["provider"], "institution": names.get(h["provider"], h["provider"]), "source_url": urls.get(h["provider"]), "level": h["level"], "status": h["status"], "age_minutes": h["age_minutes"], "current_products": h["current_products"], "last_data_at": h["last_data_at"], }) return {"providers": out, "registered": sorted(names), "stale_hours_threshold": store.STALE_H} return _cached("providers", build) @router.get("/market") def market(rate_type: str = "fixed", term_months: int = 60): """Métriques Mortgage Intelligence pour un produit donné.""" _check(rate_type, RATE_TYPES, "rate_type") def build(): con = store.connect() stats = store.market_stats(con, rate_type=rate_type, term_months=term_months) con.close() if stats is None: raise HTTPException(404, "Aucun taux courant pour ces critères") return stats return _cached(f"market|{rate_type}|{term_months}", build) @router.get("/intelligence") def intelligence(): """Vue d'ensemble du marché : les produits phares en un appel.""" def build(): con = store.connect() combos = [("fixed", 12), ("fixed", 36), ("fixed", 48), ("fixed", 60), ("fixed", 120), ("variable", 60)] grid = [] for rtype, term in combos: s = store.market_stats(con, rate_type=rtype, term_months=term) if s: grid.append(s) prime_rows = store.current_rates(con, rate_type="other") con.close() prime = [{"institution": r["institution"], "rate": r["rate"], "product_name": r["product_name"], "age_minutes": r["age_minutes"]} for r in prime_rows if "préférentiel" in (r["product_name"] or "").lower() or "prime" in (r["product_name"] or "").lower()] return {"products": grid, "prime_rates": prime} return _cached("intelligence", build) # --------------------------------------------------------------------------- # Calculateur # --------------------------------------------------------------------------- def _pick_rate(rate_type: str, term_months: int, insured_status: str | None) -> dict | None: con = store.connect() best = store.best_rate(con, rate_type=rate_type, term_months=term_months, insured_status=insured_status) con.close() return best def _validate_scenario(price: float, down: float, amort: int, term: int, frequency: str) -> None: if price <= 0 or price > 100_000_000: raise HTTPException(422, "Prix invalide") if down < 0 or down >= price: raise HTTPException(422, "Mise de fonds invalide") if amort not in calc.AMORTIZATIONS_YEARS and not (5 <= amort <= 30): raise HTTPException(422, "Amortissement invalide (5–30 ans)") if not 3 <= term <= 120: raise HTTPException(422, "Terme invalide (3–120 mois)") if frequency not in calc.FREQUENCIES: raise HTTPException(422, f"Fréquence invalide : {frequency}") @router.post("/calculate") def calculate(body: dict = Body(...)): """Calcul hypothécaire canadien complet pour un scénario. Entrées : price, down_payment (ou down_payment_pct), amortization_years, term_months, frequency, rate_type, rate (sinon meilleur taux observé), insured_status?, include_schedule?, income?, other_debts_monthly?, property_tax_monthly?, heating_monthly?, condo_fees_monthly?. """ try: price = float(body.get("price") or 0) if body.get("down_payment") is not None: down = float(body["down_payment"]) else: down = price * float(body.get("down_payment_pct") or 20) / 100 amort = int(body.get("amortization_years") or 25) term = int(body.get("term_months") or 60) frequency = str(body.get("frequency") or "monthly") rate_type = str(body.get("rate_type") or "fixed") except (TypeError, ValueError): raise HTTPException(422, "Paramètres numériques invalides") _check(rate_type, {"fixed", "variable"}, "rate_type") _validate_scenario(price, down, amort, term, frequency) quote = cmhc.insurance_quote(price, down, amort) rate_source = None rate = body.get("rate") if rate is None: insured = "insured" if quote["required"] and quote["eligible"] else None best = _pick_rate(rate_type, term, insured) if best is None: raise HTTPException( 503, "Aucun taux courant disponible — réessayez plus tard") rate = best["rate"] rate_source = {k: best[k] for k in ("provider", "institution", "product_name", "kind", "rate", "apr", "insured_status", "source_url", "last_checked", "age_minutes", "stale")} rate = float(rate) if not 0 < rate <= 25: raise HTTPException(422, "Taux invalide") if quote["required"] and not quote["eligible"]: principal = price - down # non assurable : calcul quand même, signalé else: principal = quote["total_mortgage"] if quote["required"] else price - down compounding = "semi-annual" if rate_type == "fixed" else "monthly" pay = calc.payment(principal, rate, amort, frequency, compounding) monthly_eq = calc.payment(principal, rate, amort, "monthly", compounding) q_rate = calc.qualifying_rate(rate) q_pay = calc.payment(principal, q_rate, amort, frequency, compounding) out = { "inputs": {"price": price, "down_payment": round(down, 2), "down_payment_pct": round(down / price * 100, 2), "rate": rate, "rate_type": rate_type, "term_months": term, "amortization_years": amort, "frequency": frequency, "compounding": compounding}, "insurance": quote, "principal": round(principal, 2), "payment": pay, "payment_monthly_equivalent": monthly_eq, "qualifying": {"rate": q_rate, "payment": q_pay, "note": "Test de résistance : max(taux + 2, 5,25 %)"}, "term": calc.term_summary(principal, rate, amort, term, frequency, compounding), "stress": calc.stress_scenarios(principal, rate, amort, frequency, compounding), "renewal": calc.renewal_scenarios(principal, rate, amort, term, frequency, compounding), "payoff_years": calc.payoff_years(principal, rate, amort, frequency, compounding), "rate_source": rate_source, } rows = calc.schedule(principal, rate, amort, frequency, compounding) out["annual"] = calc.annual_rollup(rows, frequency) if body.get("include_schedule"): out["schedule"] = rows income = body.get("income") if income: out["ratios"] = calc.gds_tds( float(income), monthly_eq, float(body.get("property_tax_monthly") or 0), float(body.get("heating_monthly") or 0), float(body.get("condo_fees_monthly") or 0), float(body.get("other_debts_monthly") or 0)) return out @router.post("/affordability") def affordability(body: dict = Body(...)): """Capacité d'achat : prix max selon un versement cible OU selon les revenus (ABD/ATD au taux de qualification) ; taux requis pour un versement cible vs meilleur taux observé.""" try: amort = int(body.get("amortization_years") or 25) term = int(body.get("term_months") or 60) frequency = str(body.get("frequency") or "monthly") rate_type = str(body.get("rate_type") or "fixed") down = float(body.get("down_payment") or 0) except (TypeError, ValueError): raise HTTPException(422, "Paramètres numériques invalides") _check(rate_type, {"fixed", "variable"}, "rate_type") if frequency not in calc.FREQUENCIES: raise HTTPException(422, f"Fréquence invalide : {frequency}") rate = body.get("rate") rate_source = None if rate is None: best = _pick_rate(rate_type, term, None) if best is None: raise HTTPException( 503, "Aucun taux courant disponible — réessayez plus tard") rate = best["rate"] rate_source = {k: best[k] for k in ("provider", "institution", "product_name", "kind", "rate", "source_url", "age_minutes", "stale")} rate = float(rate) if not 0 < rate <= 25: raise HTTPException(422, "Taux invalide") compounding = "semi-annual" if rate_type == "fixed" else "monthly" q_rate = calc.qualifying_rate(rate) out: dict = {"rate": rate, "qualifying_rate": q_rate, "rate_source": rate_source} target = body.get("target_payment_monthly") if target: target = float(target) loan = calc.max_loan(target, rate, amort, "monthly", compounding) loan_q = calc.max_loan(target, q_rate, amort, "monthly", compounding) out["from_payment"] = { "target_payment_monthly": target, "max_loan": loan, "max_price": round(loan + down, 2), "max_loan_stress_tested": loan_q, "max_price_stress_tested": round(loan_q + down, 2), } income = body.get("income") if income: income = float(income) gds_room = income / 12 * 0.39 \ - float(body.get("property_tax_monthly") or 0) \ - float(body.get("heating_monthly") or 0) \ - float(body.get("condo_fees_monthly") or 0) * 0.5 tds_room = income / 12 * 0.44 \ - float(body.get("property_tax_monthly") or 0) \ - float(body.get("heating_monthly") or 0) \ - float(body.get("condo_fees_monthly") or 0) * 0.5 \ - float(body.get("other_debts_monthly") or 0) room = max(0.0, min(gds_room, tds_room)) loan_q = calc.max_loan(room, q_rate, amort, "monthly", compounding) out["from_income"] = { "income": income, "max_payment_monthly": round(room, 2), "max_loan_stress_tested": loan_q, "max_price_estimate": round(loan_q + down, 2), "note": ("Indicatif seulement (ABD 39 % / ATD 44 % au taux de " "qualification) — ne constitue pas une préapprobation."), } principal = body.get("principal") target_rate_payment = body.get("required_rate_for_payment") if principal and target_rate_payment: req = calc.required_rate(float(principal), float(target_rate_payment), amort, "monthly", compounding) out["required_rate"] = { "principal": float(principal), "target_payment_monthly": float(target_rate_payment), "rate": req, "achievable_now": req is not None and req >= rate, "best_observed": rate, } return out