┌──────────────┐   JSONL brut   ┌──────────────┐   Parquet    ┌──────────────┐
  │   COLLECTE   │ ─────────────► │  NETTOYAGE   │ ───────────► │  INGESTION   │
  │ httpx async  │  (immuable)    │   polars     │  (validé)    │  PostgreSQL  │
  │  §3          │                │   §4         │              │   §5         │
  └──────────────┘                └──────────────┘              └──────┬───────┘
                                                                       │ SQL
                                                                       ▼
  ┌──────────────┐   image        ┌──────────────┐   pipeline   ┌──────────────┐
  │   CI/CD      │ ◄───────────── │  API FastAPI │ ◄─────────── │ ENTRAÎNEMENT │
  │ GitHub Act.  │   Docker       │   + Docker   │  .joblib     │  LightGBM    │
  │  §8          │                │   §7         │              │   §6         │
  └──────────────┘                └──────────────┘              └──────────────┘

immo-prix/
├── pyproject.toml
├── uv.lock
├── compose.yaml                 # API + PostgreSQL          (module 5)
├── Dockerfile
├── .dockerignore  .gitignore  .pre-commit-config.yaml
├── .github/workflows/ci.yml     # intégration continue      (module 26)
├── sql/
│   ├── 001_schema.sql           # tables, index, vues       (module 4)
│   └── 002_vues.sql
├── src/immo/
│   ├── config.py                # pydantic-settings         (module 6)
│   ├── schemas.py               # modèles Pydantic partagés
│   ├── scrape.py                # §3
│   ├── clean.py                 # §4
│   ├── ingest.py                # §5
│   ├── features.py              # §6
│   ├── train.py                 # §6
│   └── api.py                   # §7
├── tests/
└── data/                        # ignoré par Git
    ├── raw/                     # JSONL horodatés, jamais modifiés
    └── clean/                   # Parquet

import httpx, polars as pl

def transactions_dvf(code_commune: str, annee: int) -> pl.DataFrame:
    """Transactions réelles issues de l'open data DVF (licence ouverte)."""
    r = httpx.get("https://api.cquest.org/dvf",
                  params={"code_commune": code_commune}, timeout=30)
    r.raise_for_status()
    return (
        pl.DataFrame(r.json()["resultats"])
          .filter(pl.col("date_mutation").str.starts_with(str(annee)))
          .select("date_mutation", "valeur_fonciere", "surface_relle_bati",
                  "nombre_pieces_principales", "type_local", "code_postal", "commune")
    )

from urllib.robotparser import RobotFileParser
from urllib.parse import urljoin

AGENT = "immo-prix-bot/1.0 (+https://nils-malmberg.fr/project7-m28.html)"

def autorise(base_url: str, chemin: str) -> tuple[bool, float]:
    """Lit robots.txt : (le chemin est-il autorisé, délai demandé entre requêtes)."""
    rp = RobotFileParser()
    rp.set_url(urljoin(base_url, "/robots.txt"))
    rp.read()
    delai = rp.crawl_delay(AGENT) or 1.0        # délai demandé, 1 s par défaut
    return rp.can_fetch(AGENT, urljoin(base_url, chemin)), float(delai)

ok, delai = autorise("https://immo-exemple.fr", "/annonces/")
if not ok:
    raise SystemExit("robots.txt interdit cette section : on s'arrête ici.")

import asyncio, hashlib, random
from pathlib import Path
import httpx

CACHE = Path("data/cache")
CACHE.mkdir(parents=True, exist_ok=True)

async def recuperer(client: httpx.AsyncClient, url: str,
                    sem: asyncio.Semaphore, delai: float) -> str | None:
    """Renvoie le HTML d'une page, avec cache disque, reprise et politesse."""
    fichier = CACHE / (hashlib.sha256(url.encode()).hexdigest()[:16] + ".html")
    if fichier.exists():                       # cache : on ne redemande jamais deux fois
        return fichier.read_text(encoding="utf-8")

    async with sem:                            # au plus N requêtes simultanées
        for essai in range(4):
            try:
                r = await client.get(url, timeout=20, follow_redirects=True)
                if r.status_code == 429:       # trop de requêtes : on ralentit
                    await asyncio.sleep(float(r.headers.get("Retry-After", 30)))
                    continue
                r.raise_for_status()
                fichier.write_text(r.text, encoding="utf-8")
                # politesse : délai demandé + un peu d'aléa pour ne pas être régulier
                await asyncio.sleep(delai + random.uniform(0, 0.5))
                return r.text
            except (httpx.TimeoutException, httpx.HTTPStatusError) as e:
                if essai == 3:
                    print(f"abandon {url}: {e!r}")
                    return None
                await asyncio.sleep(2 ** essai)      # attente exponentielle : 1, 2, 4 s

import re
from selectolax.parser import HTMLParser
from pydantic import BaseModel, Field, field_validator

class AnnonceBrute(BaseModel):
    """Ce qu'on extrait d'une page. Pydantic (module 6) fait la validation ici."""
    url: str
    titre: str
    prix_texte: str
    surface_texte: str | None = None
    pieces_texte: str | None = None
    ville: str
    code_postal: str = Field(pattern=r"^\d{5}$")
    description: str = ""
    collecte_le: str                        # horodatage ISO

    @field_validator("titre", "ville")
    @classmethod
    def non_vide(cls, v: str) -> str:
        v = v.strip()
        if not v:
            raise ValueError("champ vide")
        return v

def texte(noeud, selecteur: str, defaut: str = "") -> str:
    """Renvoie le texte d'un sélecteur, ou une valeur par défaut s'il est absent."""
    trouve = noeud.css_first(selecteur)
    return trouve.text(strip=True) if trouve else defaut

def extraire(html: str, url: str) -> AnnonceBrute | None:
    arbre = HTMLParser(html)
    try:
        return AnnonceBrute(
            url=url,
            titre=texte(arbre, "h1.annonce-titre"),
            prix_texte=texte(arbre, "[data-testid='prix']"),
            surface_texte=texte(arbre, "li.critere-surface") or None,
            pieces_texte=texte(arbre, "li.critere-pieces") or None,
            ville=texte(arbre, "span.localisation-ville"),
            code_postal=re.search(r"\b\d{5}\b", texte(arbre, "span.localisation")).group(),
            description=texte(arbre, "div.description"),
            collecte_le=datetime.now(timezone.utc).isoformat(),
        )
    except (ValidationError, AttributeError) as e:
        print(f"page ignorée {url}: {e}")     # structure inattendue → on saute, sans planter
        return None

import json
from datetime import datetime, timezone

async def collecter(urls: list[str], sortie: Path, concurrence: int = 4) -> int:
    ok, delai = autorise("https://immo-exemple.fr", "/annonces/")
    if not ok:
        raise SystemExit("collecte non autorisée")

    sem = asyncio.Semaphore(concurrence)     # 4 requêtes simultanées, pas 400
    n = 0
    async with httpx.AsyncClient(headers={"User-Agent": AGENT}, http2=True) as client:
        # JSONL : une ligne = un objet JSON. Écriture au fil de l'eau, sans tout
        # garder en mémoire — le principe des générateurs du module 1.
        with sortie.open("w", encoding="utf-8") as f:
            taches = [recuperer(client, u, sem, delai) for u in urls]
            for future, url in zip(asyncio.as_completed(taches), urls):
                html = await future
                if html and (annonce := extraire(html, url)):
                    f.write(annonce.model_dump_json() + "\n")
                    n += 1
    return n

if __name__ == "__main__":
    horodatage = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S")
    sortie = Path(f"data/raw/annonces_{horodatage}.jsonl")   # fichier HORODATÉ, immuable
    print(f"{asyncio.run(collecter(lire_urls(), sortie))} annonces écrites dans {sortie}")

import re
import polars as pl                       # module 3 : lazy, rapide, expressif

def nettoyer(chemin_jsonl: str) -> pl.DataFrame:
    brut = pl.read_ndjson(chemin_jsonl)

    return (
        brut
        # ── 1. extraction numérique depuis le texte ──
        .with_columns(
            prix=pl.col("prix_texte")
                   .str.replace_all(r"[^\d]", "")      # ne garder que les chiffres
                   .cast(pl.Float64, strict=False),    # "Nous consulter" → null
            surface=pl.col("surface_texte")
                      .str.extract(r"(\d+[,.]?\d*)")   # "72,5 m²" → "72,5"
                      .str.replace(",", ".")
                      .cast(pl.Float64, strict=False),
            n_pieces=pl.col("pieces_texte")
                       .str.extract(r"(\d+)")
                       .cast(pl.Int8, strict=False),
        )
        # ── 2. normalisation des chaînes ──
        .with_columns(
            ville=pl.col("ville").str.to_uppercase()
                    .str.replace_all(r"[-']", " ")
                    .str.replace_all(r"\s+", " ").str.strip_chars(),
            departement=pl.col("code_postal").str.slice(0, 2),
            collecte_le=pl.col("collecte_le").str.to_datetime(),
        )
        # ── 3. features dérivées utiles au contrôle qualité ──
        .with_columns(
            prix_m2=pl.col("prix") / pl.col("surface"),
            surface_par_piece=pl.col("surface") / pl.col("n_pieces"),
            a_ascenseur=pl.col("description").str.contains("(?i)ascenseur"),
            a_balcon=pl.col("description").str.contains("(?i)balcon|terrasse"),
        )
    )

def filtrer(df: pl.DataFrame) -> tuple[pl.DataFrame, dict]:
    """Renvoie les données retenues ET le compte de ce qui a été écarté."""
    n0 = len(df)
    rapport = {"initial": n0}

    # ── doublons ──
    df = df.unique(subset=["url"], keep="last")           # même annonce recollectée
    rapport["doublons_url"] = n0 - len(df); n1 = len(df)

    # quasi-doublons : même bien republié sous une autre URL
    df = df.unique(subset=["ville", "surface", "n_pieces", "prix"], keep="first")
    rapport["doublons_contenu"] = n1 - len(df); n2 = len(df)

    # ── valeurs manquantes sur les colonnes indispensables ──
    df = df.drop_nulls(["prix", "surface"])
    rapport["sans_prix_ou_surface"] = n2 - len(df); n3 = len(df)

    # ── aberrants : bornes MÉTIER, pas statistiques ──
    df = df.filter(
        (pl.col("surface").is_between(8, 1000))           # < 8 m² : impossible en France
        & (pl.col("prix").is_between(10_000, 5_000_000))
        & (pl.col("prix_m2").is_between(300, 25_000))     # hors de ces bornes = erreur
        & (pl.col("n_pieces").is_null() | pl.col("n_pieces").is_between(1, 20))
    )
    rapport["aberrants"] = n3 - len(df)
    rapport["retenu"] = len(df)
    rapport["taux_retenu"] = round(len(df) / n0, 3)
    return df, rapport

ATTENDU = {"prix": pl.Float64, "surface": pl.Float64, "n_pieces": pl.Int8,
           "ville": pl.String, "code_postal": pl.String, "collecte_le": pl.Datetime}

def verifier(df: pl.DataFrame) -> None:
    for colonne, dtype in ATTENDU.items():
        assert colonne in df.columns, f"colonne manquante : {colonne}"
        assert df[colonne].dtype == dtype, f"{colonne}: {df[colonne].dtype} ≠ {dtype}"
    assert df["prix"].min() > 0
    assert df["code_postal"].str.contains(r"^\d{5}$").all()
    assert df["prix"].null_count() == 0
    taux_manquants = df["n_pieces"].null_count() / len(df)
    assert taux_manquants < 0.30, f"trop de n_pieces manquants : {taux_manquants:.0%}"

df_propre, rapport = filtrer(nettoyer("data/raw/annonces_20260316T090000.jsonl"))
verifier(df_propre)
print(rapport)   # {'initial': 4820, 'doublons_url': 112, 'doublons_contenu': 340,
                 #  'sans_prix_ou_surface': 87, 'aberrants': 61,
                 #  'retenu': 4220, 'taux_retenu': 0.876}
df_propre.write_parquet("data/clean/annonces_20260316.parquet")

CREATE TABLE IF NOT EXISTS annonces (
    annonce_id     BIGSERIAL PRIMARY KEY,
    url            TEXT        NOT NULL UNIQUE,   -- la clé métier : une URL = un bien
    titre          TEXT        NOT NULL,
    prix           NUMERIC(12,2) NOT NULL CHECK (prix > 0),
    surface        REAL        NOT NULL CHECK (surface BETWEEN 8 AND 1000),
    n_pieces       SMALLINT    CHECK (n_pieces BETWEEN 1 AND 20),
    ville          TEXT        NOT NULL,
    code_postal    CHAR(5)     NOT NULL CHECK (code_postal ~ '^\d{5}$'),
    departement    CHAR(2)     GENERATED ALWAYS AS (LEFT(code_postal, 2)) STORED,
    a_ascenseur    BOOLEAN     NOT NULL DEFAULT FALSE,
    a_balcon       BOOLEAN     NOT NULL DEFAULT FALSE,
    description    TEXT,
    collecte_le    TIMESTAMPTZ NOT NULL,
    maj_le         TIMESTAMPTZ NOT NULL DEFAULT now(),
    prix_m2        NUMERIC(10,2) GENERATED ALWAYS AS (prix / surface) STORED
);

-- index : jointures et filtres les plus fréquents (module 4)
CREATE INDEX IF NOT EXISTS idx_annonces_cp     ON annonces(code_postal);
CREATE INDEX IF NOT EXISTS idx_annonces_date   ON annonces(collecte_le DESC);
CREATE INDEX IF NOT EXISTS idx_annonces_comp   ON annonces(code_postal, surface);

-- historique des prix : une annonce republiée moins cher est un signal fort
CREATE TABLE IF NOT EXISTS historique_prix (
    annonce_id  BIGINT      NOT NULL REFERENCES annonces(annonce_id) ON DELETE CASCADE,
    prix        NUMERIC(12,2) NOT NULL,
    constate_le TIMESTAMPTZ NOT NULL DEFAULT now(),
    PRIMARY KEY (annonce_id, constate_le)
);

-- suivi des exécutions : indispensable pour le monitoring (module 26)
CREATE TABLE IF NOT EXISTS executions (
    execution_id  BIGSERIAL PRIMARY KEY,
    lance_le      TIMESTAMPTZ NOT NULL DEFAULT now(),
    fichier_source TEXT,
    n_collectees  INTEGER,
    n_retenues    INTEGER,
    taux_retenu   REAL,
    rapport       JSONB                  -- le dict de la section 4, tel quel
);

import json
import polars as pl
from sqlalchemy import create_engine, text
from immo.config import parametres          # pydantic-settings (module 6)

moteur = create_engine(parametres.database_url, pool_pre_ping=True)

UPSERT = text("""
    INSERT INTO annonces (url, titre, prix, surface, n_pieces, ville, code_postal,
                          a_ascenseur, a_balcon, description, collecte_le)
    VALUES (:url, :titre, :prix, :surface, :n_pieces, :ville, :code_postal,
            :a_ascenseur, :a_balcon, :description, :collecte_le)
    ON CONFLICT (url) DO UPDATE SET          -- l'annonce existe déjà : on met à jour
        prix   = EXCLUDED.prix,
        maj_le = now()
    RETURNING annonce_id, (xmax = 0) AS insertion    -- xmax = 0 → c'était un INSERT
""")

def ingerer(chemin_parquet: str, rapport: dict) -> dict:
    df = pl.read_parquet(chemin_parquet)
    lignes = df.to_dicts()

    with moteur.begin() as con:              # TRANSACTION : tout ou rien (module 4)
        n_nouvelles = 0
        for lot in (lignes[i:i + 500] for i in range(0, len(lignes), 500)):
            for ligne in lot:
                res = con.execute(UPSERT, ligne).one()
                n_nouvelles += int(res.insertion)
                # on garde la trace du prix à chaque passage
                con.execute(text("""INSERT INTO historique_prix (annonce_id, prix)
                                    VALUES (:id, :prix)
                                    ON CONFLICT DO NOTHING"""),
                            {"id": res.annonce_id, "prix": ligne["prix"]})

        con.execute(text("""INSERT INTO executions
                            (fichier_source, n_collectees, n_retenues, taux_retenu, rapport)
                            VALUES (:f, :c, :r, :t, CAST(:j AS jsonb))"""),
                    {"f": chemin_parquet, "c": rapport["initial"], "r": rapport["retenu"],
                     "t": rapport["taux_retenu"], "j": json.dumps(rapport)})

    return {"nouvelles": n_nouvelles, "mises_a_jour": len(lignes) - n_nouvelles}

-- prix médian au m² par commune : c'est la BASELINE du modèle (section 1)
CREATE OR REPLACE VIEW v_reference_commune AS
SELECT code_postal,
       COUNT(*)                                             AS n_annonces,
       PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY prix_m2) AS prix_m2_median,
       PERCENTILE_CONT(0.1) WITHIN GROUP (ORDER BY prix_m2) AS prix_m2_p10,
       PERCENTILE_CONT(0.9) WITHIN GROUP (ORDER BY prix_m2) AS prix_m2_p90
FROM annonces
WHERE collecte_le > now() - INTERVAL '90 days'
GROUP BY code_postal
HAVING COUNT(*) >= 20;            -- pas de référence sur 3 annonces

-- baisses de prix : fonction de fenêtre (module 4)
CREATE OR REPLACE VIEW v_baisses AS
SELECT a.url, a.ville, h.prix AS prix_actuel,
       FIRST_VALUE(h.prix) OVER (PARTITION BY h.annonce_id
                                 ORDER BY h.constate_le) AS prix_initial,
       h.prix / FIRST_VALUE(h.prix) OVER (PARTITION BY h.annonce_id
                                          ORDER BY h.constate_le) - 1 AS variation
FROM historique_prix h JOIN annonces a USING (annonce_id);

import numpy as np, polars as pl

def construire_features(df: pl.DataFrame, reference: pl.DataFrame) -> pl.DataFrame:
    """reference = la vue v_reference_commune, calculée sur le passé uniquement."""
    return (
        df.join(reference, on="code_postal", how="left")
          .with_columns(
              # la variable la plus prédictive : le niveau de prix local
              prix_m2_median=pl.col("prix_m2_median").fill_null(
                  pl.col("prix_m2_median").median()),          # commune inconnue
              # non-linéarités simples, utiles aux modèles linéaires (module 14)
              log_surface=pl.col("surface").log(),
              surface_par_piece=pl.col("surface") / pl.col("n_pieces").fill_null(2),
              # indicateur de manquant : le FAIT de manquer est informatif (module 12)
              pieces_manquant=pl.col("n_pieces").is_null().cast(pl.Int8),
              n_pieces=pl.col("n_pieces").fill_null(-1),       # sentinelle pour les arbres
              # saisonnalité : le marché n'est pas le même en août (module 17)
              mois=pl.col("collecte_le").dt.month(),
              mois_sin=(2 * np.pi * pl.col("collecte_le").dt.month() / 12).sin(),
              mois_cos=(2 * np.pi * pl.col("collecte_le").dt.month() / 12).cos(),
          )
    )

COLONNES = ["surface", "log_surface", "n_pieces", "pieces_manquant",
            "surface_par_piece", "prix_m2_median", "a_ascenseur", "a_balcon",
            "mois_sin", "mois_cos", "departement"]

import joblib, numpy as np, polars as pl, mlflow
from lightgbm import LGBMRegressor, early_stopping
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OrdinalEncoder
from sklearn.metrics import mean_absolute_error

VERSION = "1.0.0"

def preparer(df: pl.DataFrame, date_coupure: str):
    train = df.filter(pl.col("collecte_le") <  pl.lit(date_coupure).str.to_datetime())
    test  = df.filter(pl.col("collecte_le") >= pl.lit(date_coupure).str.to_datetime())
    # cible : le LOGARITHME du prix. L'erreur devient relative, et la distribution
    # fortement asymétrique des prix (module 10) devient à peu près symétrique.
    return (train[COLONNES].to_pandas(), np.log(train["prix"].to_numpy()),
            test[COLONNES].to_pandas(),  np.log(test["prix"].to_numpy()))

def entrainer(X_tr, y_tr, X_te, y_te):
    pre = ColumnTransformer(
        [("cat", OrdinalEncoder(handle_unknown="use_encoded_value", unknown_value=-1),
          ["departement"])],
        remainder="passthrough",
    )
    modele = Pipeline([                       # pipeline COMPLET : protège des fuites
        ("pre", pre),
        ("gbm", LGBMRegressor(n_estimators=3000, learning_rate=0.03, num_leaves=31,
                              min_child_samples=30, subsample=0.8, colsample_bytree=0.8,
                              random_state=42)),
    ])
    modele.fit(X_tr, y_tr,
               gbm__eval_set=[(pre.fit(X_tr).transform(X_te), y_te)],
               gbm__callbacks=[early_stopping(100, verbose=False)])
    return modele

def evaluer(modele, X_te, y_te, baseline_m2, surfaces) -> dict:
    pred_log = modele.predict(X_te)
    prix_reel, prix_pred = np.exp(y_te), np.exp(pred_log)

    return {
        "mae_log":   mean_absolute_error(y_te, pred_log),        # ≈ erreur relative
        "mape_med":  float(np.median(np.abs(prix_pred / prix_reel - 1))),
        "dans_10pct": float(np.mean(np.abs(prix_pred / prix_reel - 1) < 0.10)),
        # LA comparaison qui compte : bat-on la baseline ? (module 12)
        "mape_med_baseline": float(np.median(np.abs(baseline_m2 * surfaces / prix_reel - 1))),
    }

baseline (prix médian au m² × surface)   erreur médiane : 18,4 %
LightGBM                                 erreur médiane :  9,1 %   ← division par deux
                                         62 % des biens estimés à ±10 %
                                         MAE sur log(prix) : 0,118

# jeu de calibration disjoint de l'entraînement
scores = np.abs(y_calib - modele.predict(X_calib))
n, alpha = len(scores), 0.1
q = np.quantile(scores, np.ceil((n + 1) * (1 - alpha)) / n, method="higher")
# en log : un écart constant devient un FACTEUR multiplicatif sur le prix
print(f"intervalle à 90 % : ×{np.exp(-q):.2f} à ×{np.exp(q):.2f}")   # ×0,82 à ×1,22

mlflow.set_experiment("immo-prix")
with mlflow.start_run(run_name=f"lgbm-{VERSION}"):
    mlflow.log_params({"n_estimators": 3000, "learning_rate": 0.03,
                       "date_coupure": DATE_COUPURE, "n_train": len(X_tr)})
    mlflow.log_param("git_sha", sha_courant())
    mlflow.log_metrics(metriques)
    mlflow.sklearn.log_model(modele, "modele", registered_model_name="immo-prix")

joblib.dump({"pipeline": modele, "version": VERSION, "colonnes": COLONNES,
             "q_conforme": q, "entraine_le": datetime.now(timezone.utc).isoformat()},
            f"models/immo-{VERSION}.joblib")     # le pipeline ENTIER, pas l'estimateur seul

from contextlib import asynccontextmanager
import anyio, joblib, numpy as np, pandas as pd
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from sqlalchemy import create_engine, text

etat: dict = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    artefact = joblib.load("models/immo-1.0.0.joblib")     # chargé UNE fois (module 6)
    etat.update(artefact)
    etat["moteur"] = create_engine(parametres.database_url, pool_pre_ping=True)
    yield
    etat.clear()

app = FastAPI(title="Estimation de prix immobilier", version="1.0.0", lifespan=lifespan)

class Bien(BaseModel):
    surface: float = Field(gt=8, le=1000, description="surface habitable en m²")
    n_pieces: int | None = Field(default=None, ge=1, le=20)
    code_postal: str = Field(pattern=r"^\d{5}$")
    a_ascenseur: bool = False
    a_balcon: bool = False

class Estimation(BaseModel):
    prix_estime: float
    prix_bas: float
    prix_haut: float
    prix_m2_estime: float
    reference_locale_m2: float | None
    n_comparables: int
    version_modele: str

@app.get("/health")
def health() -> dict:
    return {"status": "ok", "modele": etat.get("version")}

@app.post("/estimer", response_model=Estimation)
async def estimer(bien: Bien) -> Estimation:
    # référence locale : une requête SQL indexée, quelques millisecondes (module 4)
    with etat["moteur"].connect() as con:
        ref = con.execute(text("""SELECT prix_m2_median, n_annonces
                                  FROM v_reference_commune WHERE code_postal = :cp"""),
                          {"cp": bien.code_postal}).one_or_none()
    if ref is None:
        raise HTTPException(404, "commune sans référence : pas assez d'annonces")

    X = construire_ligne(bien, prix_m2_median=float(ref.prix_m2_median))
    log_prix = await anyio.to_thread.run_sync(etat["pipeline"].predict, X)   # module 2

    prix = float(np.exp(log_prix[0]))
    q = etat["q_conforme"]
    return Estimation(
        prix_estime=round(prix, -2),
        prix_bas=round(prix * float(np.exp(-q)), -2),
        prix_haut=round(prix * float(np.exp(q)), -2),
        prix_m2_estime=round(prix / bien.surface, 1),
        reference_locale_m2=float(ref.prix_m2_median),
        n_comparables=int(ref.n_annonces),
        version_modele=etat["version"],
    )

services:
  api:
    build: .
    ports: ["8000:8000"]
    environment:
      DATABASE_URL: postgresql+psycopg://immo:${POSTGRES_PASSWORD}@base:5432/immo
    volumes:
      - ./models:/app/models:ro          # le modèle, monté en lecture seule
    depends_on:
      base: { condition: service_healthy }
    restart: unless-stopped

  base:
    image: postgres:16
    environment:
      POSTGRES_USER: immo
      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
      POSTGRES_DB: immo
    volumes:
      - donnees-pg:/var/lib/postgresql/data
      - ./sql:/docker-entrypoint-initdb.d:ro     # schéma appliqué au 1er démarrage
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U immo"]
      interval: 5s
      retries: 10

volumes:
  donnees-pg:

docker compose up -d --build
curl -X POST localhost:8000/estimer -H "Content-Type: application/json" \
     -d '{"surface": 72.5, "n_pieces": 3, "code_postal": "38000", "a_ascenseur": true}'
# {"prix_estime":248300.0,"prix_bas":203600.0,"prix_haut":302900.0,
#  "prix_m2_estime":3425.5,"reference_locale_m2":3180.0,"n_comparables":412,
#  "version_modele":"1.0.0"}

name: CI

on:                                   # QUAND ce workflow se déclenche
  push:
    branches: [main]
  pull_request:                       # sur chaque proposition de modification

jobs:
  qualite:
    runs-on: ubuntu-latest
    services:                         # PostgreSQL éphémère pour les tests d'intégration
      postgres:
        image: postgres:16
        env:
          POSTGRES_USER: immo
          POSTGRES_PASSWORD: test
          POSTGRES_DB: immo_test
        ports: ["5432:5432"]
        options: >-
          --health-cmd pg_isready --health-interval 5s --health-retries 10

    steps:
      - uses: actions/checkout@v4                   # récupère le code

      - name: Installer uv                          # module 1
        uses: astral-sh/setup-uv@v3
        with:
          enable-cache: true                        # cache des dépendances entre exécutions

      - name: Installer les dépendances
        run: uv sync --frozen                       # --frozen : échoue si uv.lock a dérivé

      - name: Lint et formatage
        run: |
          uv run ruff check .
          uv run ruff format --check .

      - name: Types
        run: uv run mypy src/

      - name: Tests
        env:
          DATABASE_URL: postgresql+psycopg://immo:test@localhost:5432/immo_test
        run: uv run pytest -q --cov=src --cov-report=term-missing

      - name: Vérifier que le schéma SQL s'applique
        env:
          PGPASSWORD: test
        run: psql -h localhost -U immo -d immo_test -f sql/001_schema.sql

  image:
    needs: qualite                                  # ne démarre que si la qualité passe
    if: github.ref == 'refs/heads/main'             # uniquement sur main
    runs-on: ubuntu-latest
    permissions:
      contents: read
      packages: write                               # droit de publier l'image
    steps:
      - uses: actions/checkout@v4
      - uses: docker/setup-buildx-action@v3
      - uses: docker/login-action@v3
        with:
          registry: ghcr.io
          username: ${{ github.actor }}
          password: ${{ secrets.GITHUB_TOKEN }}     # fourni automatiquement
      - uses: docker/build-push-action@v6
        with:
          push: true
          tags: |
            ghcr.io/${{ github.repository }}:latest
            ghcr.io/${{ github.repository }}:${{ github.sha }}
          cache-from: type=gha                      # cache des couches Docker
          cache-to: type=gha,mode=max

name: Collecte quotidienne

on:
  schedule:
    - cron: "17 4 * * *"          # tous les jours à 4 h 17 UTC (heure creuse)
  workflow_dispatch:              # et déclenchable à la main depuis l'interface

jobs:
  collecter:
    runs-on: ubuntu-latest
    timeout-minutes: 45           # garde-fou : un scraper bloqué ne tourne pas 6 h
    steps:
      - uses: actions/checkout@v4
      - uses: astral-sh/setup-uv@v3
      - run: uv sync --frozen

      - name: Collecte, nettoyage, ingestion
        env:
          DATABASE_URL: ${{ secrets.DATABASE_URL }}    # SECRET, jamais dans le code
        run: |
          uv run python -m immo.scrape
          uv run python -m immo.clean
          uv run python -m immo.ingest

      - name: Contrôle qualité              # échoue si le taux retenu s'effondre
        run: uv run python -m immo.checks --seuil-taux-retenu 0.75

      - name: Alerter en cas d'échec
        if: failure()
        run: echo "::error::La collecte a échoué — vérifier la structure du site"