#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Nettoyage des aberrations + modèle hédonique de prix des maisons (Realtor + DuProprio combinés). Log-linéaire, effets fixes ville, SE robustes. """ import json, re, warnings import numpy as np import pandas as pd import statsmodels.formula.api as smf warnings.filterwarnings("ignore") SRC = "data/processed/maisons_avec_prix.json" CUR_YEAR = 2026 # ---------- Chargement ---------- with open(SRC, encoding="utf-8") as f: data = json.load(f) print(f"Chargé : {len(data)} enregistrements") # ---------- Parsers ---------- def parse_area(s): """Retourne une superficie en pieds carrés (ft²).""" if not s or not isinstance(s, str): return np.nan txt = s.lower() is_m2 = ("m2" in txt) or ("m²" in txt) # garder chiffres, points, virgules, espaces ; retirer unités num = re.sub(r"[^0-9.,\s]", "", txt).strip() num = num.replace(" ", "").replace(",", "") # séparateurs de milliers try: v = float(num) except ValueError: return np.nan if v <= 0: return np.nan return v * 10.7639 if is_m2 else v def clean_city(city, source): if not city or not isinstance(city, str): return None c = city.split("(")[0].strip() # retire quartiers realtor c = re.sub(r"\s+", " ", c).lower().strip() return c or None def to_int(x): try: return int(x) except (TypeError, ValueError): return np.nan # amenity keywords -> recherche dans features + description + title (bilingue) AMENITIES = { "pool": ["swimming pool", "piscine", "inground pool", "pool"], "waterfront": ["waterfront", "bord de l'eau", "navigable", "water access", "lakefront", "riverfront", "au bord de", "waterfront"], "garage": ["garage"], "fireplace": ["fireplace", "foyer", "wood stove", "poêle"], "central_air": ["central air", "air conditionné", "climatisation", "a/c", "heat-pump", "thermopompe", "air exchanger"], "finished_bsmt":["totally finished", "finished basement", "sous-sol fini", "basement"], "near_water": ["near water", "waterfront", "navigable"], "cul_de_sac": ["cul-de-sac", "cul de sac"], "golf": ["golf"], "ski": ["ski"], "near_park": ["park", "parc"], "near_school": ["school", "école", "cegep", "university", "université", "elementary", "high school"], "near_highway":["highway", "autoroute"], "transit": ["public transit", "public transportation", "transport"], "no_rear_neighbor":["no backyard neighbors", "no rear neighbor", "sans voisin arrière"], } # ---------- Parsing des pièces (Realtor) ---------- def _side_ft(side): s = side.lower() if "ft" in s: ft = re.search(r"([\d.]+)\s*ft", s) inch = re.search(r"([\d.]+)\s*in", s) return (float(ft.group(1)) if ft else 0) + \ (float(inch.group(1)) / 12 if inch else 0) m = re.search(r"([\d.,]+)\s*m", s) if m: try: return float(m.group(1).replace(",", ".")) * 3.28084 except ValueError: return None return None def _room_area(dim): if not dim or "x" not in dim: return None a, _, b = dim.partition("x") la, lb = _side_ft(a), _side_ft(b) if la and lb and 1 < la < 200 and 1 < lb < 200: return la * lb return None ROOM_FLAGS = { # pièces à valeur potentielle "r_office": ["office"], "r_family": ["family room"], "r_workshop": ["workshop"], "r_winecellar": ["wine cellar"], "r_sunroom": ["sunroom", "solarium"], "r_mezzanine": ["mezzanine"], "r_library": ["library"], "r_laundry": ["laundry"], } def parse_rooms(rooms): out = {"room_area": np.nan, "n_rooms": 0, "n_bed_r": 0, "n_bath_r": 0, "has_primary_bed": 0, "has_basement": 0, "n_levels_r": 0} for k in ROOM_FLAGS: out[k] = 0 if not rooms: return out area, levels = 0.0, set() for r in rooms: nm = (r.get("name") or "").lower() lv = (r.get("level") or "").lower() out["n_rooms"] += 1 if lv: levels.add(lv) if "basement" in lv or "sous-sol" in lv: out["has_basement"] = 1 if "primary bedroom" in nm: out["has_primary_bed"] = 1 out["n_bed_r"] += 1 elif "bedroom" in nm: out["n_bed_r"] += 1 if "bathroom" in nm: out["n_bath_r"] += 1 for k, kws in ROOM_FLAGS.items(): if any(w in nm for w in kws): out[k] = 1 a = _room_area(r.get("dimensions")) if a: area += a out["room_area"] = area if area > 0 else np.nan out["n_levels_r"] = len(levels) return out rows = [] for d in data: prop = d.get("property") or {} addr = d.get("address") or {} feats = d.get("features") or [] blob = (" ".join(feats) + " " + (d.get("description") or "") + " " + (d.get("title") or "")).lower() yb = to_int(prop.get("yearBuilt")) age = (CUR_YEAR - yb) if (not pd.isna(yb) and 1700 <= yb <= CUR_YEAR) else np.nan ptype = prop.get("buildingType") or prop.get("type") # realtor/duproprio storeys = prop.get("storeys") or prop.get("levels") try: storeys = float(str(storeys).replace(",", ".")) except (TypeError, ValueError): storeys = np.nan rec = { "price": d.get("priceNumber"), "source": d.get("source"), "sqft": parse_area(prop.get("squareFeet")), "lot_sqft": parse_area(prop.get("lotSize")), "beds": to_int(prop.get("bedrooms")), "baths": to_int(prop.get("bathrooms")), "parking": to_int(prop.get("parkingSpaces")), # realtor seulement "storeys": storeys, "age": age, "ptype": (ptype or "Unknown").strip(), "city": clean_city(addr.get("city"), d.get("source")), "region": (addr.get("region") or "").strip().lower() or None, } for name, kws in AMENITIES.items(): rec[name] = int(any(k in blob for k in kws)) # signaux "autre chose" : photos + visite virtuelle rec["n_images"] = len(d.get("images") or []) rec["virtual_tour"] = int(bool(d.get("virtualTourUrl"))) # détail des pièces rec.update(parse_rooms(d.get("rooms"))) rows.append(rec) df = pd.DataFrame(rows) print(f"DataFrame brut : {df.shape}") # ---------- Nettoyage des aberrations ---------- n0 = len(df) df = df[df["price"].notna()] # prix plausibles pour une maison au Québec df = df[(df["price"] >= 50_000) & (df["price"] <= 15_000_000)] # superficie habitable : borner si présente, garder les NA (imputés + flag) df.loc[(df["sqft"] < 300) | (df["sqft"] > 20_000), "sqft"] = np.nan # terrain plausible (garder NA autorisé -> mais on impute plus bas) df.loc[(df["lot_sqft"] < 200) | (df["lot_sqft"] > 5_000_000), "lot_sqft"] = np.nan # chambres / sdb plausibles df = df[(df["beds"].between(0, 15)) | (df["beds"].isna())] df = df[(df["baths"].between(0, 12)) | (df["baths"].isna())] df.loc[df["beds"] > 15, "beds"] = np.nan df.loc[df["baths"] > 12, "baths"] = np.nan # prix au pied carré aberrant (sur les lignes qui ont une superficie réelle) df["ppsf"] = df["price"] / df["sqft"] lo, hi = df["ppsf"].quantile([0.005, 0.995]) bad = df["ppsf"].notna() & ((df["ppsf"] < lo) | (df["ppsf"] > hi)) df = df[~bad] print(f"Après nettoyage : {len(df)} (retirés : {n0 - len(df)}) " f"[filtre ppsf {lo:.0f}–{hi:.0f} $/ft² sur superficies réelles]") # ---------- Superficie : réelle > reconstruite via pièces > imputée ---------- # calibration : ratio médian (superficie réelle / somme des pièces) sur recouvrement ov = df[df["sqft"].notna() & df["room_area"].notna() & (df["room_area"] > 200)] ratio = (ov["sqft"] / ov["room_area"]).median() print(f"Calibration surface pièces : ratio réel/pièces = {ratio:.3f} " f"(n={len(ov)})") room_est = df["room_area"] * ratio room_ok = df["room_area"].notna() & (room_est.between(300, 20_000)) df["sqft_final"] = df["sqft"] fill = df["sqft_final"].isna() & room_ok df.loc[fill, "sqft_final"] = room_est[fill] df["sqft_measured"] = df["sqft_final"].notna().astype(int) # réelle ou pièces n_room_fill = int(fill.sum()) print(f"Superficies : réelles={int(df['sqft'].notna().sum())}, " f"complétées via pièces=+{n_room_fill}, " f"mesurées au total={int(df['sqft_measured'].sum())}/{len(df)}") # imputation du reste = médiane par type df["has_sqft"] = df["sqft_measured"] df["sqft_imp"] = df.groupby("ptype")["sqft_final"].transform( lambda s: s.fillna(s.median())) df["sqft_imp"] = df["sqft_imp"].fillna(df["sqft_final"].median()) df["has_lot"] = df["lot_sqft"].notna().astype(int) df["lot_sqft"] = df["lot_sqft"].fillna(df["lot_sqft"].median()) df["has_age"] = df["age"].notna().astype(int) df["age"] = df["age"].fillna(df["age"].median()).clip(lower=0) df["beds"] = df["beds"].fillna(df["beds"].median()) df["baths"] = df["baths"].fillna(df["baths"].median()) df["parking"] = df["parking"].fillna(0) df["storeys"] = df["storeys"].fillna(df["storeys"].median()) df["ln_price"] = np.log(df["price"]) df["ln_sqft"] = np.log(df["sqft_imp"]) df["ln_lot"] = np.log(df["lot_sqft"]) df["age2"] = df["age"] ** 2 # Effets fixes ville : regrouper les villes rares vc = df["city"].value_counts() keep = set(vc[vc >= 20].index) df["city_fe"] = df["city"].where(df["city"].isin(keep), other="__other__") df["city_fe"] = df["city_fe"].fillna("__other__") print(f"Villes en effets fixes : {df['city_fe'].nunique()} " f"(seuil ≥20 listings ; {len(keep)} villes nommées)") # Type de propriété harmonisé (regrouper rares) tc = df["ptype"].value_counts() keept = set(tc[tc >= 30].index) df["ptype_fe"] = df["ptype"].where(df["ptype"].isin(keept), other="Other") # ---------- Modèles hédoniques ---------- amen = list(AMENITIES.keys()) def pct(b): return (np.exp(b) - 1) * 100 # variables "détail des pièces" + autres signaux (Realtor surtout) df["n_images_c"] = df["n_images"].clip(upper=60) # écrête les extrêmes room_vars = ["n_rooms", "n_levels_r", "has_primary_bed", "has_basement", "r_office", "r_family", "r_workshop", "r_winecellar", "r_sunroom", "r_mezzanine", "r_library", "r_laundry"] extra = ["n_images_c", "virtual_tour"] + room_vars base = ("ln_lot + has_lot + beds + baths + parking + storeys " "+ age + age2 + has_age + C(source) + C(ptype_fe) + " + " + ".join(amen) + " + " + " + ".join(extra) + " + C(city_fe)") fA = "ln_price ~ ln_sqft + " + base # riche (superficie réelle) fB = "ln_price ~ " + base # large (sans superficie) dfA = df[df["has_sqft"] == 1].copy() print(f"\nModèle A (avec superficie réelle) : N={len(dfA)}") mA = smf.ols(fA, data=dfA).fit(cov_type="HC3") print(f"Modèle B (sans superficie, couverture large) : N={len(df)}") mB = smf.ols(fB, data=df).fit(cov_type="HC3") sdA, sdB = np.std(mA.resid), np.std(mB.resid) def report(m, sd, name): print("\n" + "=" * 70) print(f"{name} N={int(m.nobs)} R²={m.rsquared:.4f} " f"R²_adj={m.rsquared_adj:.4f} " f"erreur typique ±{(np.exp(sd)-1)*100:.1f}%") print("=" * 70) pr, pv = m.params, m.pvalues show = [p for p in pr.index if not p.startswith("C(city_fe)")] print(f"{'variable':<30}{'coef':>10}{'~%impact':>11}{'p':>9}") for p in show: st = "***" if pv[p]<.01 else "**" if pv[p]<.05 else "*" if pv[p]<.1 else "" pc = f"{pct(pr[p]):+.1f}%" if abs(pr[p]) < 3 else "" print(f"{p:<30}{pr[p]:>10.4f}{pc:>11}{pv[p]:>9.3g} {st}") report(mA, sdA, "MODÈLE A — hédonique riche (superficie incluse)") report(mB, sdB, "MODÈLE B — couverture large (sans superficie)") # ---------- Prédictions combinées ---------- # A pour les biens à superficie réelle, B sinon predA = np.exp(mA.predict(df)) * np.exp(sdA**2 / 2) predB = np.exp(mB.predict(df)) * np.exp(sdB**2 / 2) df["pred_price"] = np.where(df["has_sqft"] == 1, predA, predB) df["model_used"] = np.where(df["has_sqft"] == 1, "A", "B") df["resid_pct"] = (df["price"] / df["pred_price"] - 1) * 100 out_cols = (["source", "city", "ptype", "price", "pred_price", "resid_pct", "model_used", "sqft", "room_area", "sqft_final", "sqft_measured", "lot_sqft", "beds", "baths", "parking", "storeys", "age", "n_images", "virtual_tour"] + amen + room_vars) df[out_cols].to_csv("data/processed/maisons_hedonic.csv", index=False) print(f"\nDataset final + prédictions -> maisons_hedonic.csv ({len(df)} lignes)") # top villes (effets fixes du modèle A) cityfe = {k.split("T.")[1].rstrip("]"): v for k, v in mA.params.items() if k.startswith("C(city_fe)")} cf = pd.Series(cityfe).sort_values(ascending=False) print("\nTop 12 villes — prime de localisation (Modèle A, vs référence):") for c, b in cf.head(12).items(): print(f" {c:<30}{pct(b):+.1f}%") print("Bottom 6:") for c, b in cf.tail(6).items(): print(f" {c:<30}{pct(b):+.1f}%") with open("results/hedonic_summary.txt", "w", encoding="utf-8") as fh: fh.write("=== MODELE A (avec superficie) ===\n") fh.write(str(mA.summary())) fh.write("\n\n=== MODELE B (sans superficie) ===\n") fh.write(str(mB.summary())) print("Résumés complets -> hedonic_summary.txt")