""" build_prix_evolution.py ======================= Calcule les statistiques de prix immobiliers par année / code postal / type de bien pour les villes sélectionnées de la métropole lilloise, puis insère les résultats dans la table `prix_evolution` de immobilier.db. Usage : python build_prix_evolution.py Sources : data/2021/59.csv … data/2025/59.csv (DVF géocodé, département 59) """ from __future__ import annotations import csv import os import statistics import unicodedata from collections import defaultdict from domain.core.mysql_db import get_connection # --------------------------------------------------------------------------- # Chemins # --------------------------------------------------------------------------- ROOT = os.path.dirname(__file__) DATA_DIR = os.path.join(ROOT, "data") # --------------------------------------------------------------------------- # Villes cibles (formes originales — normalisées au runtime) # --------------------------------------------------------------------------- VILLES_CIBLES_RAW = [ "Wasquehal", "Marcq-en-Barœul", "Lille", "Croix", "Roubaix", "Tourcoing", "Villeneuve-d'Ascq", "Lezennes", "Mons-en-Barœul", "Hem", "Saint-André-lez-Lille", "Mouvaux", "Lambersart", "Marquette-lez-Lille", "Wambrechies", "Linselles", "Quesnoy-sur-Deûle", ] ANNEES = [2021, 2022, 2023, 2024, 2025] TYPES_VALIDES = {"Maison", "Appartement"} # --------------------------------------------------------------------------- # Normalisation des noms de communes # --------------------------------------------------------------------------- def normaliser(nom: str) -> str: """ Normalise un nom de commune pour la comparaison : ligatures (œ→oe, æ→ae), accents supprimés, tirets/apostrophes→espace, majuscules, espaces multiples fusionnés. """ nom = ( nom .replace("œ", "oe").replace("Œ", "OE") .replace("æ", "ae").replace("Æ", "AE") ) n = unicodedata.normalize("NFD", nom).encode("ascii", "ignore").decode() return ( n.upper() .replace("-", " ") .replace("'", " ") # apostrophe U+0027 .replace("’", " ") # RIGHT SINGLE QUOTATION MARK .replace(" ", " ") .strip() ) VILLES_CIBLES = {normaliser(v) for v in VILLES_CIBLES_RAW} # --------------------------------------------------------------------------- # Lecture des transactions DVF géocodées # --------------------------------------------------------------------------- def lire_mutations(annee: int) -> list[dict]: """ Lit data/{annee}/59.csv et retourne la liste des transactions résidentielles pour les villes cibles. Agrégation par id_mutation pour gérer les multi-lots (maison + dépendance…) : - valeur_fonciere : unique par mutation (prise sur la première ligne) - surface_reelle_bati : somme des lots Maison/Appartement de la mutation - type_local : type du lot avec la plus grande surface """ path = os.path.join(DATA_DIR, str(annee), "59.csv") if not os.path.exists(path): print(f" [!] Fichier introuvable : {path}") return [] by_mutation: dict[str, dict] = {} with open(path, encoding="utf-8", newline="") as f: for row in csv.DictReader(f): if row.get("nature_mutation", "").strip() != "Vente": continue type_local = row.get("type_local", "").strip() if type_local not in TYPES_VALIDES: continue if normaliser(row.get("nom_commune", "")) not in VILLES_CIBLES: continue try: surface = float(row["surface_reelle_bati"] or 0) valeur = float(row["valeur_fonciere"].replace(",", ".") or 0) except (ValueError, KeyError): continue if surface <= 0 or valeur <= 0: continue mid = row["id_mutation"] if mid not in by_mutation: by_mutation[mid] = { "annee": annee, "code_postal": row["code_postal"].strip(), "commune": row["nom_commune"].strip(), "valeur": valeur, "surface": 0.0, "type_local": type_local, "max_surface": 0.0, } m = by_mutation[mid] m["surface"] += surface if surface > m["max_surface"]: m["max_surface"] = surface m["type_local"] = type_local # type du lot principal result = [] for m in by_mutation.values(): if m["surface"] > 0: m["prix_m2"] = m["valeur"] / m["surface"] if not (500.0 <= m["prix_m2"] <= 15_000.0): continue result.append(m) print(f" {annee} : {len(result):>6,} transactions retenues") return result # --------------------------------------------------------------------------- # Calcul des statistiques agrégées # --------------------------------------------------------------------------- def calculer_stats(transactions: list[dict]) -> list[dict]: """ Agrège les transactions par (annee, code_postal, commune, type_local) et calcule : nb, prix_m2_moyen, prix_m2_median, prix_moyen, prix_median, surface_moyenne. """ groupes: dict[tuple, list[dict]] = defaultdict(list) for t in transactions: key = (t["annee"], t["code_postal"], t["commune"], t["type_local"]) groupes[key].append(t) rows = [] for (annee, cp, commune, type_local), txs in sorted(groupes.items()): prix_m2 = [t["prix_m2"] for t in txs] prix_tot = [t["valeur"] for t in txs] surfaces = [t["surface"] for t in txs] rows.append({ "annee": annee, "code_postal": cp, "nom_commune": commune, "type_local": type_local, "nb_transactions": len(txs), "prix_m2_moyen": round(statistics.mean(prix_m2), 2), "prix_m2_median": round(statistics.median(prix_m2), 2), "prix_moyen": round(statistics.mean(prix_tot), 2), "prix_median": round(statistics.median(prix_tot), 2), "surface_moyenne": round(statistics.mean(surfaces), 2), }) return rows # --------------------------------------------------------------------------- # Calcul de l'évolution annuelle du prix médian au m² # --------------------------------------------------------------------------- def ajouter_evolution(stats: list[dict]) -> list[dict]: """ Pour chaque ligne, calcule l'évolution en % du prix_m2_median par rapport à l'année précédente (même code_postal + type_local). La première année disponible reçoit evolution_m2_pct = None. Formule : (prix_n - prix_n-1) / prix_n-1 * 100 (peut être négatif). """ # index : (code_postal, type_local, annee) → prix_m2_median index: dict[tuple, float] = { (r["code_postal"], r["type_local"], r["annee"]): r["prix_m2_median"] for r in stats } for r in stats: annee_prec = r["annee"] - 1 median_prec = index.get((r["code_postal"], r["type_local"], annee_prec)) if median_prec and median_prec != 0: r["evolution_m2_pct"] = round( 100.0 * (r["prix_m2_median"] - median_prec) / median_prec, 2 ) else: r["evolution_m2_pct"] = None return stats # --------------------------------------------------------------------------- # Base de données # --------------------------------------------------------------------------- DDL = """ CREATE TABLE IF NOT EXISTS prix_evolution ( id INT AUTO_INCREMENT PRIMARY KEY, annee INT NOT NULL, code_postal VARCHAR(10) NOT NULL, nom_commune VARCHAR(100) NOT NULL, type_local VARCHAR(20) NOT NULL, nb_transactions INT NOT NULL, prix_m2_moyen DOUBLE NOT NULL, prix_m2_median DOUBLE NOT NULL, evolution_m2_pct DOUBLE, prix_moyen DOUBLE NOT NULL, prix_median DOUBLE NOT NULL, surface_moyenne DOUBLE NOT NULL, UNIQUE (annee, code_postal, type_local) ) """ INSERT_SQL = """ REPLACE INTO prix_evolution (annee, code_postal, nom_commune, type_local, nb_transactions, prix_m2_moyen, prix_m2_median, evolution_m2_pct, prix_moyen, prix_median, surface_moyenne) VALUES (%(annee)s, %(code_postal)s, %(nom_commune)s, %(type_local)s, %(nb_transactions)s, %(prix_m2_moyen)s, %(prix_m2_median)s, %(evolution_m2_pct)s, %(prix_moyen)s, %(prix_median)s, %(surface_moyenne)s) """ def inserer(stats: list[dict], con) -> None: with con.cursor() as cur: cur.executemany(INSERT_SQL, stats) con.commit() # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main() -> None: print("=== build_prix_evolution.py ===") print(f"Villes : {len(VILLES_CIBLES)} communes cibles") print() con = get_connection() with con.cursor() as cur: cur.execute("DROP TABLE IF EXISTS prix_evolution") cur.execute(DDL) con.commit() all_transactions: list[dict] = [] print("Lecture des fichiers DVF géocodés :") for annee in ANNEES: all_transactions.extend(lire_mutations(annee)) print(f"\nTotal transactions : {len(all_transactions):,}") stats = calculer_stats(all_transactions) stats = ajouter_evolution(stats) print(f"Lignes agrégées : {len(stats)}") inserer(stats, con) con.close() print("\nInsertion terminée dans MySQL.") # ------------------------------------------------------------------ # Affichage de la requête SELECT # ------------------------------------------------------------------ print() print("=" * 70) print("REQUÊTE SELECT — évolution des prix par commune / type / année") print("=" * 70) print(""" -- Vue d'ensemble : prix médian au m² + évolution annuelle par commune, type et année SELECT nom_commune, type_local, annee, nb_transactions, ROUND(prix_m2_median) AS prix_m2_median, ROUND(prix_m2_moyen) AS prix_m2_moyen, CASE WHEN evolution_m2_pct IS NULL THEN 'ref' ELSE ROUND(evolution_m2_pct, 1) || '%' END AS evol_vs_an_prec, ROUND(surface_moyenne) AS surface_moy_m2, ROUND(prix_median) AS prix_median FROM prix_evolution ORDER BY nom_commune, type_local, annee; -- Pivot condensé : prix médian au m² + évolution YoY par colonne SELECT nom_commune, type_local, MAX(CASE WHEN annee = 2021 THEN ROUND(prix_m2_median) END) AS "€/m²_2021", MAX(CASE WHEN annee = 2022 THEN ROUND(prix_m2_median) END) AS "€/m²_2022", MAX(CASE WHEN annee = 2022 THEN ROUND(evolution_m2_pct, 1) END) AS "evol_2022_%", MAX(CASE WHEN annee = 2023 THEN ROUND(prix_m2_median) END) AS "€/m²_2023", MAX(CASE WHEN annee = 2023 THEN ROUND(evolution_m2_pct, 1) END) AS "evol_2023_%", MAX(CASE WHEN annee = 2024 THEN ROUND(prix_m2_median) END) AS "€/m²_2024", MAX(CASE WHEN annee = 2024 THEN ROUND(evolution_m2_pct, 1) END) AS "evol_2024_%", MAX(CASE WHEN annee = 2025 THEN ROUND(prix_m2_median) END) AS "€/m²_2025", MAX(CASE WHEN annee = 2025 THEN ROUND(evolution_m2_pct, 1) END) AS "evol_2025_%", ROUND( 100.0 * ( MAX(CASE WHEN annee = 2025 THEN prix_m2_median END) - MAX(CASE WHEN annee = 2021 THEN prix_m2_median END) ) / NULLIF(MAX(CASE WHEN annee = 2021 THEN prix_m2_median END), 0), 1 ) AS "evol_2021_2025_%" FROM prix_evolution GROUP BY nom_commune, type_local ORDER BY nom_commune, type_local; -- Filtre sur une ville et un type spécifique SELECT annee, code_postal, nb_transactions, ROUND(prix_m2_median) AS prix_m2_median, ROUND(prix_m2_moyen) AS prix_m2_moyen, CASE WHEN evolution_m2_pct IS NULL THEN 'ref' ELSE ROUND(evolution_m2_pct, 1) || '%' END AS evol_vs_an_prec, ROUND(surface_moyenne) AS surface_moy_m2, ROUND(prix_median) AS prix_median FROM prix_evolution WHERE nom_commune = 'Wasquehal' -- remplacer par la commune souhaitée AND type_local = 'Appartement' -- 'Appartement' | 'Maison' ORDER BY annee; """) if __name__ == "__main__": main()