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/validate.py : couche de validation des observations de taux.5#   Règle absolue : ne jamais inventer ni laisser passer un taux impossible.6#   4.19 % -> 419 % doit être rejeté ; une bonne donnée n'est jamais écrasée7#   par une donnée manifestement erronée (garde-fou complémentaire dans8#   store.record_observations).9# -----------------------------------------------------------------------------10from __future__ import annotations1112RATE_MIN = 0.25      # sous 0,25 % : impossible pour une hypothèque canadienne13RATE_MAX = 20.0      # au-delà de 20 % : aberrant (même les prêts privés)14TERM_MIN_MONTHS = 315TERM_MAX_MONTHS = 1201617VALID_RATE_TYPES = {"fixed", "variable", "adjustable", "other"}18VALID_KINDS = {"posted", "special"}19VALID_INSURED = {"insured", "insurable", "uninsured", "unknown"}20VALID_PURPOSES = {"purchase", "renewal", "refinance", "unknown"}2122REQUIRED_FIELDS = ("provider", "institution", "rate_type", "term_months",23                   "kind", "rate", "source_url")242526def validate_product(p: dict) -> tuple[bool, list[str]]:27    """Valide UNE observation normalisée. Retourne (ok, problèmes)."""28    issues: list[str] = []29    for f in REQUIRED_FIELDS:30        if p.get(f) in (None, ""):31            issues.append(f"champ manquant: {f}")32    if issues:33        return False, issues34    rate = p["rate"]35    if not isinstance(rate, (int, float)):36        issues.append(f"taux non numérique: {rate!r}")37    elif rate <= 0:38        issues.append(f"taux nul ou négatif: {rate}")39    elif rate < RATE_MIN or rate > RATE_MAX:40        issues.append(f"taux impossible: {rate}")41    term = p["term_months"]42    if not isinstance(term, int) or not (TERM_MIN_MONTHS <= term <= TERM_MAX_MONTHS):43        issues.append(f"terme invalide: {term!r}")44    if p["rate_type"] not in VALID_RATE_TYPES:45        issues.append(f"rate_type invalide: {p['rate_type']}")46    if p["kind"] not in VALID_KINDS:47        issues.append(f"kind invalide: {p['kind']}")48    if p.get("insured_status", "unknown") not in VALID_INSURED:49        issues.append(f"insured_status invalide: {p.get('insured_status')}")50    if p.get("purpose", "purchase") not in VALID_PURPOSES:51        issues.append(f"purpose invalide: {p.get('purpose')}")52    apr = p.get("apr")53    if apr is not None:54        if not isinstance(apr, (int, float)) or apr < RATE_MIN or apr > RATE_MAX:55            issues.append(f"APR impossible: {apr!r}")56        elif isinstance(rate, (int, float)) and apr < rate - 0.02:57            issues.append(f"APR ({apr}) inférieur au taux contractuel ({rate})")58    return not issues, issues596061def validate_batch(products: list[dict]) -> tuple[list[dict], list[str]]:62    """Valide une passe complète : élimine invalides et duplicatas exacts.63    Retourne (produits valides, problèmes)."""64    from .store import product_key65    ok_products: list[dict] = []66    problems: list[str] = []67    seen: set[tuple[str, float]] = set()68    for p in products:69        valid, issues = validate_product(p)70        if not valid:71            name = p.get("product_name") or "?"72            problems.extend(f"{name}: {i}" for i in issues)73            continue74        key = (product_key(p), round(float(p["rate"]), 4))75        if key in seen:76            continue  # duplicata exact silencieusement ignoré77        seen.add(key)78        ok_products.append(p)79    return ok_products, problems80