spb/house-ka

1# -----------------------------------------------------------------------------2# Immo-Ka — Agrégateur de maisons à vendre (province de Québec)3# Auteur : Simon-Pierre Boucher — contact@spboucher.ai4# mortgage/providers/rbc.py : RBC — API JSON publique publicrates (celle que5#   consomme leur propre page de taux, 1 requête par code de taux).6#   ⚠️ Les produits variables sont publiés en ÉCART vs prime (ex. -0.500) :7#   reconstitués avec le code prime, jamais interprétés comme des taux.8# -----------------------------------------------------------------------------9from __future__ import annotations1011import json1213from .base import RateProvider1415API = ("https://apps.royalbank.com/apps/app-services/public-rates/api/"16       "publicrates?code={code}")17PAGE_URL = "https://www.rbcroyalbank.com/mortgages/mortgage-rates.html"1819PRIME_CODE = "0006470002"2021# code -> (rate_type, terme mois, kind, insured_status, nom, code APR|None)22CODE_MAP: dict[str, tuple] = {23    "0006340010": ("fixed", 6, "posted", "unknown", "Fixe 6 mois convertible", None),24    "0006340013": ("fixed", 12, "posted", "unknown", "Fixe fermé 1 an", None),25    "0006340016": ("fixed", 24, "posted", "unknown", "Fixe fermé 2 ans", None),26    "0006340019": ("fixed", 36, "posted", "unknown", "Fixe fermé 3 ans", None),27    "0006340022": ("fixed", 48, "posted", "unknown", "Fixe fermé 4 ans", None),28    "0006340025": ("fixed", 60, "posted", "unknown", "Fixe fermé 5 ans", None),29    "0006340042": ("fixed", 84, "posted", "unknown", "Fixe fermé 7 ans", None),30    "0006340030": ("fixed", 120, "posted", "unknown", "Fixe fermé 10 ans", None),31    "0386080056": ("fixed", 12, "special", "unknown", "Fixe fermé 1 an (offre spéciale)", "0386140074"),32    "0386080057": ("fixed", 24, "special", "unknown", "Fixe fermé 2 ans (offre spéciale)", "0386140075"),33    "0386080058": ("fixed", 36, "special", "unknown", "Fixe fermé 3 ans (offre spéciale)", "0386140076"),34    "0386080059": ("fixed", 48, "special", "unknown", "Fixe fermé 4 ans (offre spéciale)", "0386140077"),35    "0386080060": ("fixed", 60, "special", "unknown", "Fixe fermé 5 ans (offre spéciale)", "0386140078"),36    "0386080061": ("fixed", 84, "special", "unknown", "Fixe fermé 7 ans (offre spéciale)", "0386140079"),37    "0385940006": ("fixed", 60, "special", "insured", "Fixe fermé 5 ans (ratio élevé)", "0386090006"),38}39# codes variables : ÉCART vs prime -> taux = prime + valeur40SPREAD_MAP: dict[str, tuple] = {41    "0236440012": ("variable", 60, "posted", "unknown", "Variable fermé 5 ans", "0166950009"),42    "0386080070": ("variable", 60, "special", "unknown", "Variable fermé 5 ans (offre spéciale)", "0386140088"),43    "0385940016": ("variable", 60, "special", "insured", "Variable fermé 5 ans (ratio élevé)", "0386090016"),44}454647def _value(responses: dict, code: str) -> float | None:48    resp = responses.get(code)49    if not isinstance(resp, dict):50        return None51    content = resp.get("result_content") or {}52    try:53        return float(content.get("Value"))54    except (TypeError, ValueError):55        return None565758class RbcProvider(RateProvider):59    provider_id = "rbc"60    institution = "RBC Banque Royale"61    source_url = PAGE_URL62    request_delay = 0.66364    def _codes(self) -> list[str]:65        codes = [PRIME_CODE] + list(CODE_MAP) + list(SPREAD_MAP)66        codes += [apr for *_, apr in CODE_MAP.values() if apr]67        codes += [apr for *_, apr in SPREAD_MAP.values() if apr]68        return codes6970    def fetch(self) -> list[dict]:71        responses: dict[str, dict] = {}72        for code in self._codes():73            try:74                responses[code] = json.loads(self.get(API.format(code=code)).text)75            except Exception:  # noqa: BLE001 — un code raté n'annule pas le reste76                continue77        return self.parse_codes(responses)7879    def parse(self, payload: str) -> list[dict]:80        """Pour les tests fixtures : payload = JSON {code: réponse API}."""81        data = json.loads(payload)82        return self.parse_codes(data.get("responses") or data)8384    def parse_codes(self, responses: dict) -> list[dict]:85        out: list[dict] = []86        prime = _value(responses, PRIME_CODE)87        for code, (rtype, term, kind, insured, name, apr_code) in CODE_MAP.items():88            rate = _value(responses, code)89            if rate is None or rate <= 0:90                continue91            apr = _value(responses, apr_code) if apr_code else None92            out.append(self.make_product(93                rate=rate, rate_type=rtype, term_months=term, kind=kind,94                product_name=name, apr=apr, insured_status=insured,95                raw={"code": code, "value": rate}))96        if prime is not None:97            for code, (rtype, term, kind, insured, name, apr_code) in SPREAD_MAP.items():98                spread = _value(responses, code)99                if spread is None or abs(spread) > 3:100                    continue  # un écart vs prime > 3 pts : structure changée101                apr = _value(responses, apr_code) if apr_code else None102                out.append(self.make_product(103                    rate=round(prime + spread, 3), rate_type=rtype,104                    term_months=term, kind=kind, product_name=name, apr=apr,105                    insured_status=insured,106                    conditions=f"Prime RBC {prime} % {spread:+} pt",107                    raw={"code": code, "spread": spread, "prime": prime}))108            out.append(self.make_product(109                rate=prime, rate_type="other", term_months=12, kind="posted",110                product_name="Taux préférentiel RBC", purpose="unknown",111                raw={"code": PRIME_CODE, "value": prime}))112        return out113