Merge pull request #11 from T-DAT-902-Homepedia/feat/score-quartier-iris

feat(duckpipe): maille quartier IRIS — prix et gap qualité-prix infra-communaux

Author: glopez-dev

ARCHITECTURE.md

@@ -127,8 +127,11 @@ gs://homepedia-data/
 │   └── dpe/
 ├── silver/
 │   ├── communes_geom/           (GeoParquet)
+│   ├── iris_geom/               (GeoParquet, contours IRIS — maille quartier)
 │   ├── dvf_clean/year=2024/
+│   ├── dvf_points/year=<annee>/ (mutations géolocalisées des millésimes annexes)
 │   ├── commune_agg/year=2024/
+│   ├── iris_prix/year=2024/     (agrégat prix par IRIS, fenêtre poolée)
 │   ├── transport_commune/
 │   ├── climat_commune/
 │   ├── proximite_commune/
@@ -143,6 +146,9 @@ gs://homepedia-data/
     ├── score_territoire/
     │   ├── run_date=2025-01-15/score.parquet
     │   └── latest/score.parquet   ← lu par l'API FastAPI
+    ├── score_quartier/            (gap qualité-prix à la maille IRIS)
+    │   ├── run_date=2025-01-15/score_quartier.parquet
+    │   └── latest/score_quartier.parquet
     └── dq_reports/
         ├── silver_2025-01-15.json
         └── gold_2025-01-15.json

adr/0015-maille-quartier-iris.md

@@ -0,0 +1,84 @@
+---
+status: accepted
+date: 2026-07-09
+decision-makers: équipe Homepedia
+---
+
+# Maille quartier : agrégats prix et gap qualité-prix à l'IRIS
+
+## Contexte et problème
+
+Le score global (`score_valeur`) et l'écart qualité-prix (`gap`,
+`gap_pondere`) ne sont calculés qu'à la maille commune : à l'intérieur d'une
+grande ville, la carte est aveugle à la variation entre quartiers — qui est
+d'abord une variation de prix. Le DVF est pourtant géolocalisé au grain
+mutation (lon/lat conservés en silver `dvf`), seule manque une maille
+infra-communale de restitution.
+
+## Facteurs de décision
+
+- Seul le prix est disponible partout à une maille infra-communale (points
+  DVF) ; sécurité, emploi et risques n'existent pas plus fin que la commune.
+- Le seuil de fiabilité (>= 5 ventes) devient difficile à atteindre au grain
+  quartier sur un seul millésime DVF.
+- Déterminisme entre runs (ADR-0008) : pas de simplification de géométries
+  calculée (ADR-0013), pas de tirage non reproductible.
+- Contrat front : `schema_version` reste 1, ajouts additifs uniquement
+  (ADR-0014) ; budget CDN surveillé (ADR-0013).
+
+## Options envisagées
+
+- **Maille IRIS INSEE** : découpage statistique officiel, contours nationaux
+  publiés (CONTOURS-IRIS® IGN/INSEE, Licence Ouverte 2.0), quartiers nommés,
+  et la plupart des sources INSEE (BPE, Filosofi, base logement) existent
+  déclinées à l'IRIS — seule maille qui permette d'enrichir le score plus tard.
+- Sections cadastrales (id_parcelle DVF) : écarté — aucune donnée socio-éco à
+  cette maille, pas de noms, trop de zones sous le seuil de fiabilité.
+- Carreaux INSEE 200 m/1 km : écarté — illisible produit (pas de notion de
+  quartier), volumétrie carte élevée, pas de correspondance avec les autres
+  sources.
+- Score entièrement recalculé à l'IRIS d'emblée : écarté pour le MVP — exige
+  4-5 nouvelles ingestions (BPE IRIS, Filosofi IRIS, re-fetch ADEME…) pour un
+  signal dont l'essentiel (le prix) est déjà disponible. Prévu en phase 2.
+
+## Décision
+
+MVP « gap quartier » à la maille IRIS, qualité héritée de la commune :
+
+- **Contours** : édition FlatGeoBuf 2026 de CONTOURS-IRIS® (Géoplateforme
+  `data.geopf.fr`), déjà généralisée moyenne échelle — lue par `ST_Read`,
+  aucune simplification calculée. Les IRIS PLM sont codés par arrondissement
+  (751xx/6938x/132xx), même convention que le DVF : raccord par equi-join,
+  et ne JAMAIS filtrer `iris_geom` sur `commune_geom` (Etalab ne connaît que
+  75056).
+- **Prix** : `iris_prix` poole le millésime courant (silver `dvf`) et les
+  mutations géolocalisées des millésimes annexes (`dvf_points_<annee>`,
+  seconde sortie de `prix_millesime`), médiane simple sans pondération de
+  récence. Affectation point→IRIS contrainte à la commune de la mutation
+  (equi-join puis `ST_Intersects`), point de frontière départagé au plus
+  petit `code_iris`. Millésime annexe manquant = fenêtre réduite, pas d'échec.
+- **Gold `score_quartier`** : `n_prix_iris` normalisé par le même `_norm` que
+  le communal, sur la population retenue (IRIS fiables de communes scorées) ;
+  `gap_iris = score_commune − n_prix_iris`,
+  `gap_pondere_iris = gap_iris × n_access_fin_commune`.
+- **Export web** : `choropleth/iris-high/{dept}.geojson`, communes multi-IRIS
+  uniquement (les mono-IRIS dupliqueraient `communes-high` : ×5,5 d'économie
+  CDN), clés meta additives `nb_iris`/`nb_iris_scores` pour le feature-gate
+  front.
+
+## Conséquences
+
+- Mesuré sur le run local de validation : 16 409 IRIS exportés dont 14 491
+  scorés, 104 fichiers départementaux (max 1,6 Mo brut), appariement
+  DVF→IRIS 99,95 % (15 communes orphelines, décalage COG 2026 vs DVF —
+  suivi par `iris_match` dans le rapport DQ silver).
+- Le gap quartier d'une commune mono-IRIS mesure surtout l'effet de fenêtre
+  (prix poolés vs millésime courant), pas un signal quartier — documenté,
+  et ces IRIS ne sont pas exportés côté web.
+- Phase 2 possible sans changement de schéma : recalcul à l'IRIS des
+  dimensions qui le permettent (transport GPS, BPE `GEO_OBJECT='IRIS'`, DPE
+  via champ IRIS de l'API ADEME, Filosofi IRIS avec flag secret statistique,
+  base logement déjà à l'IRIS), les autres restant héritées.
+- Repli si l'édition FlatGeoBuf disparaît : GPKG (archive .7z, extracteur à
+  écrire) ou GeoParquet IGN (encodage GeoArrow non lu par duckdb-spatial à
+  ce jour).

adr/README.md

@@ -31,3 +31,4 @@ Records).
 | [0012](0012-ci-cd-github-actions-wif.md) | CI/CD GitHub Actions avec Workload Identity Federation | accepted | 2026-07-02 |
 | [0013](0013-serving-statique-cdn.md) | Serving statique : artefacts web pré-générés sur bucket public | accepted | 2026-07-02 |
 | [0014](0014-consolidation-export-web.md) | Consolidation de l'export web : publish-web chemin unique | accepted | 2026-07-03 |
+| [0015](0015-maille-quartier-iris.md) | Maille quartier : agrégats prix et gap qualité-prix à l'IRIS | accepted | 2026-07-09 |

duckpipe/deploy/homepedia-pipeline.yaml

@@ -53,6 +53,7 @@ main:
               - "geometries_departements_100m"
               - "geometries_departements_1000m"
               - "geometries_regions_1000m"
+              - "geometries_iris"
               - "transport"
               - "revenus"
               - "risques"
@@ -88,6 +89,7 @@ main:
             value: pipeline
             in:
               - "dvf"
+              - "iris_geom"
               - "transport"
               - "revenus"
               - "risques"
@@ -125,6 +127,15 @@ main:
                     job_name: ${job_name}
                     cli_args: ${["run", "prix_millesime", "--env", "prod", "--year", string(millesime)]}
 
+    # Agrégat prix à la maille IRIS : pool du millésime courant (silver dvf)
+    # et des mutations géolocalisées des millésimes annexes (dvf_points,
+    # produits par l'étape millesimes) — d'où le placement séquentiel après.
+    - iris_prix:
+        call: run_duckpipe
+        args:
+          job_name: ${job_name}
+          cli_args: ${["run", "iris_prix", "--env", "prod", "--year", string(year), "--run-date", run_date]}
+
     - validate_silver:
         call: run_duckpipe
         args:
@@ -137,6 +148,14 @@ main:
           job_name: ${job_name}
           cli_args: ${["run", "score", "--env", "prod", "--year", string(year), "--run-date", run_date]}
 
+    # Gap qualité-prix à la maille quartier : hérite du score communal du run,
+    # d'où le placement après l'étape score.
+    - score_quartier:
+        call: run_duckpipe
+        args:
+          job_name: ${job_name}
+          cli_args: ${["run", "score_quartier", "--env", "prod", "--year", string(year), "--run-date", run_date]}
+
     # Agrège le silver avis (produit hors-DAG par le workflow GitHub nlp-avis.yml,
     # chemins stables non millésimés) en gold avis_commune. Indépendant du score,
     # placé séquentiellement pour garder le DAG simple. Toléré si le silver avis

duckpipe/src/duckpipe/__main__.py

@@ -26,6 +26,7 @@
 from duckpipe.fetch_climat import CLIMAT_BRONZE_PATH, build_stations_csv
 from duckpipe.fetch_dpe import DPE_BRONZE_PATH, build_dpe_sample
 from duckpipe.pipeline_registry import register_pipelines
+from duckpipe.pipelines.iris import make_iris_prix_pipeline
 from duckpipe.pipelines.prix_millesime import make_prix_millesime_pipeline
 
 logger = logging.getLogger(__name__)
@@ -61,18 +62,39 @@ def cmd_ingest(args: argparse.Namespace) -> None:
         logger.info("[ok] ingest %s", name)
 
 
+def _uri_exists(uri: str) -> bool:
+    return fetch.gcs_exists(uri) if fetch.is_gcs_uri(uri) else Path(uri).exists()
+
+
+def _annees_points_disponibles(env: catalogs.Environment, year: int) -> list[int]:
+    """Millésimes annexes dont les dvf_points existent en silver : un millésime
+    manquant réduit la fenêtre poolée d'iris_prix au lieu d'échouer le run
+    (même tolérance que l'évolution des fiches dans publish_web)."""
+    annees: list[int] = []
+    for annee in catalogs.WEB_MILLESIMES:
+        if annee == year:
+            continue
+        if _uri_exists(catalogs.dvf_points_path(env, annee)):
+            annees.append(annee)
+        else:
+            logger.warning("[warn] dvf_points_%s absent, fenêtre poolée réduite", annee)
+    return annees
+
+
 def cmd_run(args: argparse.Namespace) -> None:
     env = catalogs.get_environment(args.env, local_root=args.local_root)
     catalog = catalogs.build_catalog(env, year=args.year, run_date=args.run_date)
 
     if args.pipeline == "prix_millesime":
         pipeline = make_prix_millesime_pipeline(args.year)
+    elif args.pipeline == "iris_prix":
+        pipeline = make_iris_prix_pipeline(args.year, _annees_points_disponibles(env, args.year))
     else:
         pipelines = register_pipelines()
         if args.pipeline not in pipelines:
             raise SystemExit(
                 f"pipeline inconnu : {args.pipeline!r} "
-                f"(disponibles : {', '.join(sorted(pipelines))}, prix_millesime)"
+                f"(disponibles : {', '.join(sorted(pipelines))}, prix_millesime, iris_prix)"
             )
         pipeline = pipelines[args.pipeline]
 
@@ -120,10 +142,7 @@ def cmd_validate_gold(args: argparse.Namespace) -> None:
             )
         previous_top: list[str] | None = None
         latest = catalogs.gold_latest_path(env)
-        latest_exists = (
-            fetch.gcs_exists(latest) if fetch.is_gcs_uri(latest) else Path(latest).exists()
-        )
-        if latest_exists:
+        if _uri_exists(latest):
             with fetch.local_read_path(latest) as latest_path:
                 previous_top = [
                     row[0]
@@ -138,12 +157,26 @@ def cmd_validate_gold(args: argparse.Namespace) -> None:
             report_dest=catalogs.dq_report_path(env, "gold", args.run_date),
         )
 
+        # Contrôles gold du quartier, si l'étape score_quartier a produit la
+        # table (facultatif : le run doit rester rejouable sans la maille IRIS).
+        quartier_uri = catalogs.gold_quartier_path(env, args.run_date)
+        if _uri_exists(quartier_uri):
+            with fetch.local_read_path(quartier_uri) as quartier_path:
+                con.execute(
+                    "CREATE TABLE score_quartier AS SELECT * FROM "
+                    f"read_parquet('{quartier_path}')"
+                )
+            validation.validate_gold_quartier(
+                con,
+                report_dest=catalogs.dq_report_path(env, "gold_quartier", args.run_date),
+            )
+        else:
+            logger.warning("[warn] score_quartier absent, contrôle gold quartier ignoré")
+
         # Contrôles gold des avis, si l'étape NLP a produit la table (facultatif :
         # la couverture avis est partielle et le pipeline peut tourner sans).
         avis_uri = catalogs.gold_avis_path(env, args.run_date)
-        avis_exists = (
-            fetch.gcs_exists(avis_uri) if fetch.is_gcs_uri(avis_uri) else Path(avis_uri).exists()
-        )
+        avis_exists = _uri_exists(avis_uri)
         if avis_exists:
             with fetch.local_read_path(avis_uri) as avis_path:
                 con.execute(
@@ -165,6 +198,13 @@ def cmd_publish(args: argparse.Namespace) -> None:
     validation.publish(
         catalogs.gold_score_path(env, args.run_date), catalogs.gold_latest_path(env)
     )
+    # Score quartier : publié s'il existe (toléré absent, comme les avis — les
+    # runs antérieurs à la maille IRIS restent rejouables).
+    quartier_uri = catalogs.gold_quartier_path(env, args.run_date)
+    if _uri_exists(quartier_uri):
+        validation.publish(quartier_uri, catalogs.gold_quartier_latest_path(env))
+    else:
+        logger.warning("[warn] score_quartier absent, publication quartier ignorée")
 
 
 def cmd_publish_web(args: argparse.Namespace) -> None:

duckpipe/src/duckpipe/catalogs.py

@@ -75,6 +75,10 @@ def get_environment(name: str, *, local_root: str = "data") -> Environment:
     "dept_geom_100m": "dept_geom/dept_geom_100m.parquet",
     "dept_geom_1000m": "dept_geom/dept_geom_1000m.parquet",
     "region_geom_1000m": "region_geom/region_geom_1000m.parquet",
+    # Maille quartier (IRIS) : contours stables, agrégat prix millésimé (la
+    # fenêtre poolée multi-millésimes glisse avec l'année de run).
+    "iris_geom": "iris_geom/iris_geom.parquet",  # GeoParquet
+    "iris_prix": "iris_prix/year={year}/iris_prix.parquet",
     "dvf": "dvf_clean/year={year}/dvf.parquet",
     "commune_agg": "commune_agg/year={year}/commune_agg.parquet",
     "commune_agg_type": "commune_agg_type/year={year}/commune_agg_type.parquet",
@@ -111,6 +115,21 @@ def gold_avis_path(env: Environment, run_date: str) -> str:
     return f"{env.gold_root}/avis_commune/run_date={run_date}/avis_commune.parquet"
 
 
+def gold_quartier_path(env: Environment, run_date: str) -> str:
+    return f"{env.gold_root}/score_quartier/run_date={run_date}/score_quartier.parquet"
+
+
+def gold_quartier_latest_path(env: Environment) -> str:
+    return f"{env.gold_root}/score_quartier/latest/score_quartier.parquet"
+
+
+def dvf_points_path(env: Environment, annee: int) -> str:
+    """Chemin silver des mutations géolocalisées d'un millésime annexe (les
+    points du millésime courant restent dans la table silver `dvf`). Exposé
+    pour que le CLI vérifie l'existence avant de construire iris_prix."""
+    return f"{env.silver_root}/dvf_points/year={annee}/dvf_points.parquet"
+
+
 def dq_report_path(env: Environment, kind: str, run_date: str) -> str:
     return f"{env.gold_root}/dq_reports/{kind}_{run_date}.json"
 
@@ -166,6 +185,10 @@ def build_catalog(env: Environment, *, year: int, run_date: str) -> Catalog:
         "regions_1000m_raw",
         GeoJsonDataset(f"{bronze}/{SOURCES['geometries_regions_1000m'].bronze_path}"),
     )
+    catalog.add(
+        "iris_raw",
+        GeoJsonDataset(f"{bronze}/{SOURCES['geometries_iris'].bronze_path}"),
+    )
     catalog.add("arrets_raw", CsvDataset(f"{bronze}/{SOURCES['transport'].bronze_path}"))
     catalog.add(
         "revenus_raw",
@@ -219,17 +242,20 @@ def build_catalog(env: Environment, *, year: int, run_date: str) -> Catalog:
         catalog.add(name, ParquetDataset(path))
     catalog.add("arrets", MemoryDataset())  # intermédiaire transport, jamais persisté
     # Millésimes DVF annexes : l'année du run + ceux consommés par l'export web
-    # (l'évolution des prix des fiches communes, cf. WEB_MILLESIMES).
+    # (l'évolution des prix des fiches communes, cf. WEB_MILLESIMES) et par
+    # l'agrégat quartier (mutations géolocalisées poolées, cf. iris_prix).
     for annee in {year, *WEB_MILLESIMES}:
         catalog.add(
             f"commune_prix_{annee}",
             ParquetDataset(
                 f"{env.silver_root}/commune_prix/year={annee}/commune_prix.parquet"
             ),
         )
+        catalog.add(f"dvf_points_{annee}", ParquetDataset(dvf_points_path(env, annee)))
 
     # --- Gold ----------------------------------------------------------------
     catalog.add("score_territoire", ParquetDataset(gold_score_path(env, run_date)))
     catalog.add("avis_commune", ParquetDataset(gold_avis_path(env, run_date)))
+    catalog.add("score_quartier", ParquetDataset(gold_quartier_path(env, run_date)))
 
     return catalog

duckpipe/src/duckpipe/datasets/geojson.py

@@ -10,14 +10,16 @@
 
 
 class GeoJsonDataset(Dataset):
-    """Dataset GeoJSON, lu via `ST_Read` (extension `spatial` DuckDB).
+    """Dataset vectoriel lu via `ST_Read` (extension `spatial` DuckDB) : tout
+    format géré par les drivers GDAL embarqués — GeoJSON (contours Etalab)
+    comme FlatGeoBuf (CONTOURS-IRIS® IGN).
 
     `ST_Read` passe par GDAL et non par `httpfs` : contrairement aux CSV et
     Parquet, un chemin `gs://` n'est pas lisible directement — on télécharge
     alors l'objet vers un fichier temporaire avant lecture.
 
-    Écriture non supportée : les GeoJSON de ce pipeline sont uniquement des
-    sources en entrée (contours communaux/départementaux), jamais des sorties.
+    Écriture non supportée : ces fichiers sont uniquement des sources en
+    entrée (contours communaux/départementaux/IRIS), jamais des sorties.
     """
 
     def __init__(self, path: str) -> None:

duckpipe/src/duckpipe/export_web.py

@@ -310,6 +310,48 @@ def build_choropleth_regions(
     return out_table
 
 
+def build_choropleth_iris(
+    con: duckdb.DuckDBPyConnection,
+    iris_geom: str,
+    score_quartier: str,
+    *,
+    out_table: str = "web_choropleth_iris",
+) -> str:
+    """Table choroplèthe quartier (IRIS) : uniquement les communes multi-IRIS —
+    un IRIS de commune mono-IRIS duplique le contour communal déjà servi par
+    communes-high, le front retombe alors sur la maille commune. LEFT JOIN :
+    un IRIS non fiable (ou de commune non scorée) reste affiché « pas de
+    donnée », comme les communes. Les properties tiennent lieu de
+    fiche/tooltip : pas d'artefact fiche séparé au MVP quartier.
+    """
+    con.execute(
+        f"""
+        CREATE OR REPLACE TABLE {out_table} AS
+        SELECT
+            g.code_iris,
+            g.code_commune,
+            g.nom_iris AS nom,
+            g.nom_commune,
+            g.type_iris,
+            {DEPT_EXPR.replace("code_commune", "g.code_commune")} AS code_departement,
+            CAST(round(q.prix_m2_median) AS INTEGER) AS prix_m2_median,
+            coalesce(q.nb_transactions, 0) AS nb_transactions,
+            q.code_iris IS NOT NULL AS fiable,
+            round(q.n_prix_iris, 3) AS n_prix_iris,
+            round(q.score_commune, 3) AS score_commune,
+            round(q.gap_iris, 3) AS gap_iris,
+            round(q.gap_pondere_iris, 3) AS gap_pondere_iris,
+            q.annee_min,
+            q.annee_max,
+            g.geom
+        FROM {iris_geom} g
+        LEFT JOIN {score_quartier} q USING (code_iris)
+        WHERE g.nb_iris_commune > 1
+        """
+    )
+    return out_table
+
+
 def build_evolution(
     con: duckdb.DuckDBPyConnection,
     commune_agg: str,