"""Chapitre 14 — Nettoyage, géocodage et qualité des données : reproductibilité du jeu brut, identités des diagnostics, chiffres cités dans le texte et dans les solutions (tolérances déclarées).""" import sys, pathlib, math import numpy as np, pandas as pd, pytest sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1] / "cases")) import ch14_nettoyage as c14 from traite import cas1, hedonique as h @pytest.fixture(scope="module") def R(): return c14.main(ecrire=False, verbose=False) # ----------------------------------------------------------------------------- jeu brut def test_brut_reproductible_et_identique_au_fichier(R): g1 = c14.generer_brut(); g2 = c14.generer_brut() pd.testing.assert_frame_equal(g1["brut"], g2["brut"]) disque = c14.charger_brut() assert len(disque["brut"]) == 618 and len(disque["verite"]) == 127 and len(disque["role"]) == 600 b = g1["brut"]; d = disque["brut"].copy() for c in ("date_promesse", "date_acte"): d[c] = d[c].dt.strftime("%Y-%m-%d") pd.testing.assert_frame_equal(b.reset_index(drop=True), d[b.columns].reset_index(drop=True), check_dtype=False) def test_anomalies_injectees(R): v = R["verite"]["anomalie"].value_counts() attendu = {"terrain_manquant": 20, "doublon": 18, "qualite_manquante": 15, "superficie_pi2": 12, "dates_inversees": 10, "annee_manquante": 10, "coord_nulles": 9, "meubles_inclus": 8, "vente_liee_declaree": 8, "coord_hors_zone": 6, "vente_liee_non_declaree": 6, "prix_symbolique": 5} assert v.to_dict() == attendu b = R["brut"] assert (b["prix_vente"] == 1).sum() == 5 and (b["superficie_habitable_m2"] > 450).sum() == 12 assert b["superficie_terrain_m2"].isna().sum() == 20 and b["qualite_1a5"].isna().sum() == 15 and b["annee_construction"].isna().sum() == 10 assert ((b["lat"] == 0) & (b["lon"] == 0)).sum() == 9 assert (pd.to_datetime(b["date_acte"]) < pd.to_datetime(b["date_promesse"])).sum() == 10 # les lignes touchées (hors doublons) sont disjointes sans_dup = R["verite"][R["verite"]["anomalie"] != "doublon"] assert sans_dup["id_inscription"].is_unique and len(sans_dup) == 109 # ----------------------------------------------------------------------------- pipeline def test_entonnoir_et_journal(R): P = R["pipeline"] assert P["etapes"] == [("0 lecture", 618), ("R1 doublons", 600), ("R4 prix symbolique", 595), ("R7 non représentatives", 587), ("I1 imputation", 587), ("S1 scores robustes", 581), ("E1 régression", 579)] j = P["journal"]; g = j.groupby(["etape", "action"]).size().to_dict() assert g == {("E1", "exclusion"): 2, ("G1", "imputation"): 15, ("I1", "imputation"): 45, ("R1", "exclusion"): 18, ("R2", "correction"): 10, ("R3", "correction"): 8, ("R4", "exclusion"): 5, ("R5", "correction"): 12, ("R6", "correction"): 15, ("R7", "exclusion"): 8, ("S1", "exclusion"): 6} assert len(j) == 144 # chaque anomalie est traitée par l'étape prévue ; les deux ventes liées non déclarées restantes sont prises par la régression jv = j.merge(R["verite"], on="id_inscription", how="left"); ct = pd.crosstab(jv["etape"], jv["anomalie"].fillna("propre")) # R1 conserve l'inscription la plus récente (la copie) et exclut l'originale : on vérifie par l'unité d'origine orig = dict(zip(R["brut"]["id_inscription"], R["brut"]["id_vente_origine"])) dup_units = set(R["verite"].loc[R["verite"]["anomalie"] == "doublon", "id_vente_origine"]) assert {orig[i] for i in j.loc[j["etape"] == "R1", "id_inscription"]} == dup_units and len(dup_units) == 18 assert ct.loc["R4", "prix_symbolique"] == 5 and ct.loc["R5", "superficie_pi2"] == 12 and ct.loc["R7", "vente_liee_declaree"] == 8 assert ct.loc["S1", "vente_liee_non_declaree"] == 4 and ct.loc["E1", "vente_liee_non_declaree"] == 2 and ct.loc["S1", "propre"] == 2 assert set(P["regression"]["exclus"]["id_inscription"]) == {"B0039", "B0410"} assert set(P["scores"]["exclus"]["id_inscription"]) == {"B0003", "B0006", "B0021", "B0105", "B0173", "B0558"} def test_corrections_restituent_les_valeurs_propres(R): """Après R2, R3, R5 les valeurs corrigées coïncident avec le jeu propre pour les lignes concernées.""" P = R["pipeline"]; d = P["propre"].set_index("id_vente_origine"); v = cas1.charger()["ventes"].set_index("id_vente") ver = R["verite"] for code, col in [("superficie_pi2", "superficie_habitable_m2"), ("meubles_inclus", "prix_vente")]: ids = [i for i in ver.loc[ver["anomalie"] == code, "id_vente_origine"] if i in d.index] assert len(ids) >= 7 np.testing.assert_allclose(d.loc[ids, col].astype(float), v.loc[ids, col].astype(float), rtol=0, atol=0.51) ids = [i for i in ver.loc[ver["anomalie"] == "dates_inversees", "id_vente_origine"] if i in d.index] assert (pd.to_datetime(d.loc[ids, "date_vente"]).values == pd.to_datetime(v.loc[ids, "date_vente"]).values).all() def test_iterations_regression(R): it = R["pipeline"]["regression"]["iterations"] assert [x["n"] for x in it] == [581, 579] and [x["exclus"] for x in it] == [2, 0] assert it[0]["sigma"] == pytest.approx(0.0897, abs=5e-4) and it[1]["sigma"] == pytest.approx(0.0837, abs=5e-4) assert it[0]["cook_max"] < 1 and it[0]["n_cook_4n"] == 25 and it[0]["n_levier_2kn"] == 18 # ----------------------------------------------------------------------------- identités des diagnostics def test_identites_levier_cook(): v = cas1.charger()["ventes"]; X = cas1.matrice(v).values; y = np.log(v["prix_vente"].values) dg = c14.diagnostics(X, y) assert dg["somme_h"] == pytest.approx(dg["k"], rel=1e-9) # tr(H) = k assert (dg["h"] > 0).all() and (dg["h"] < 1).all() i = int(np.argmax(dg["cook"])) assert dg["cook"][i] == pytest.approx(c14.cook_direct(X, y, i), rel=1e-6) # D_i = r_i² h/(k(1−h)) = (β−β_(i))'X'X(β−β_(i))/(k s²) # résidu externe = résidu interne × sqrt((n−k−1)/(n−k−r²)) n, k = dg["n"], dg["k"]; r = dg["r_int"] np.testing.assert_allclose(dg["t_ext"], r * np.sqrt((n - k - 1) / (n - k - r ** 2)), rtol=1e-9) def test_constante_mad_et_score(): from scipy.stats import norm assert 1 / norm.ppf(0.75) == pytest.approx(1.4826, abs=1e-4) rng = np.random.default_rng(1); x = rng.normal(0, 2, 200000) mad = np.median(np.abs(x - np.median(x))) assert 1.4826 * mad == pytest.approx(2.0, rel=0.01) # convergence vers σ z = c14.z_robuste(np.array([100.0, 101, 99, 100, 102, 98, 100, 1000])) assert z[-1] > 100 and abs(z[:-1]).max() < 2.5 # une valeur extrême n'écrase pas les autres def test_masquage(R): t = R["masquage"]["tableau"] assert t.loc["Hull", "n"] == 193 and t.loc["Hull", "n_classique"] == 2 and t.loc["Hull", "n_robuste"] == 10 and t.loc["Hull", "n_prix_1"] == 2 assert t.loc["Hull", "ecart_type"] == pytest.approx(1.395, abs=2e-3) and t.loc["Hull", "mad_x1.4826"] == pytest.approx(0.181, abs=2e-3) assert (t["n_robuste"] >= t["n_classique"]).all() # ----------------------------------------------------------------------------- effet sur le modèle def test_modeles_avant_apres(R): t = R["modeles"]["tableau"] brut = t.loc["brut (lignes complètes)"]; net = t.loc["après régression E1 (nettoyé)"]; ref = t.loc["jeu propre de référence"] assert brut["n"] == 573 and brut["R2"] == pytest.approx(0.039, abs=2e-3) and brut["sigma"] == pytest.approx(1.231, abs=2e-3) assert t.loc["après règles R1–R7 (+G1, I1)", "sigma"] == pytest.approx(0.1132, abs=5e-4) and t.loc["après scores robustes S1", "sigma"] == pytest.approx(0.0897, abs=5e-4) assert net["n"] == 579 and net["sigma"] == pytest.approx(0.0837, abs=5e-4) and net["R2"] == pytest.approx(0.850, abs=2e-3) assert ref["n"] == 600 and ref["sigma"] == pytest.approx(0.0830, abs=5e-4) and ref["R2"] == pytest.approx(0.853, abs=2e-3) assert net["ln_S"] == pytest.approx(0.5335, abs=5e-4) and ref["ln_S"] == pytest.approx(0.5391, abs=5e-4) assert net["prediction_sujet"] == pytest.approx(637017, abs=5) and ref["prediction_sujet"] == pytest.approx(637953, abs=5) assert brut["prediction_sujet"] == pytest.approx(565471, abs=5) # le modèle nettoyé retrouve la référence : tous les coefficients à moins de 0,5 erreur-type e = R["modeles"]["ecarts"]["ecart_en_se"].abs() assert e.max() < 0.5 and e.idxmax() in ("ln_S", "piscine") # ----------------------------------------------------------------------------- géocodage, appariement, imputation, sensibilité def test_geocodage(R): G = R["geocodage"]; t = G["tableau"] assert G["n_dans_rayon"] == 103 assert list(t["erreur_mediane_m"].round(0)) == [6, 18, 47, 177, 1199] assert t.loc["point d'adresse", "pct_voisins_conserves"] == pytest.approx(99.1, abs=0.1) assert t.loc["centroïde de code postal (urbain)", "pct_voisins_conserves"] == pytest.approx(76.1, abs=0.1) assert t.loc["centroïde de code postal (périurbain)", "pct_voisins_conserves"] == pytest.approx(17.6, abs=0.1) assert t.loc["centroïde de code postal (périurbain)", "pct_rayon_change"] == pytest.approx(6.7, abs=0.1) assert t["erreur_mediane_m"].is_monotonic_increasing and t["pct_voisins_conserves"].is_monotonic_decreasing # erreur médiane théorique d'une loi de Rayleigh : σ·sqrt(2 ln 2) = 1,1774 σ np.testing.assert_allclose(t["erreur_mediane_m"] / (1000 * t["sigma_km"]), 1.1774, rtol=0.06) assert abs(t.loc["point d'adresse", "beta_dist"] - G["beta_dist_ref"]) < 1e-3 def test_appariement(R): A = R["appariement"] assert A["taux"].to_dict() == {"1 exact": 367, "2 normalisé": 198, "3 attributs": 45, "non apparié": 8} assert A["pct"].to_dict() == {"1 exact": 59.4, "2 normalisé": 32.0, "3 attributs": 7.3, "non apparié": 1.3} assert A["ambigus"] == 3 and A["faux_appariements"] == 0 assert c14.normaliser_matricule("05 94 6466 4") == "059464664" and c14.normaliser_matricule(float("nan")) == "" def test_imputations(R): I = R["imputations"] assert list(I.index) == ["suppression", "mediane", "regression", "indicateur"] assert I.loc["suppression", "n"] == 535 and (I.loc[["mediane", "regression", "indicateur"], "n"] == 579).all() assert I.loc["suppression", "ln_T"] == pytest.approx(0.0969, abs=5e-4) and I.loc["mediane", "ln_T"] == pytest.approx(0.0941, abs=5e-4) assert I.loc["regression", "ln_T"] == pytest.approx(0.0940, abs=5e-4) and I.loc["indicateur", "ln_T"] == pytest.approx(0.0945, abs=5e-4) assert I.loc["suppression", "prediction_sujet"] == pytest.approx(640921, abs=5) and I.loc["indicateur", "prediction_sujet"] == pytest.approx(638354, abs=5) assert I.loc["indicateur", "coef_manq_terrain"] == pytest.approx(-0.0215, abs=5e-4) def test_sensibilite_seuils(R): T = R["sensibilite"].set_index(["seuil_mad", "seuil_t"]) assert len(T) == 12 and (T["liees_detectees"] == 6).all() assert T.loc[(3.5, 3.0), "faux_positifs"] == 2 and T.loc[(2.5, 2.5), "faux_positifs"] == 22 and T.loc[(4.0, 3.0), "faux_positifs"] == 0 assert T.loc[(3.5, 3.0), "n_final"] == 579 and T.loc[(2.5, 2.5), "n_final"] == 546 and T.loc[(4.0, 3.5), "n_final"] == 581 assert T.loc[(2.5, 2.5), "sigma"] == pytest.approx(0.0783, abs=5e-4) and T.loc[(4.0, 3.0), "sigma"] == pytest.approx(0.0838, abs=5e-4) assert T["prediction_sujet"].min() > 633000 and T["prediction_sujet"].max() < 639000 # ----------------------------------------------------------------------------- solutions des exercices def test_solutions_exercices(): # ex. 14.1 : scores robustes de neuf prix au m² x = np.array([3150.0, 3320, 3080, 3410, 3260, 3190, 3370, 3240, 5980]) med = np.median(x); mad = np.median(np.abs(x - med)) assert med == 3260 and mad == 110 z = (x - med) / (1.4826 * mad) assert z[-1] == pytest.approx(16.68, abs=0.01) and abs(z[:-1]).max() == pytest.approx(1.10, abs=0.01) zc = (x - x.mean()) / x.std(ddof=1) assert x.mean() == pytest.approx(3555.6, abs=0.1) and x.std(ddof=1) == pytest.approx(915.2, abs=0.5) and zc[-1] == pytest.approx(2.65, abs=0.01) # ex. 14.2 : conversion 1 776 pi² assert 1776 / 10.7639 == pytest.approx(165.0, abs=0.05) # ex. 14.3 : levier et Cook n, k, s = 600, 16, 0.083; e, hii = -0.25, 0.04 r = e / (s * math.sqrt(1 - hii)); D = r ** 2 / k * hii / (1 - hii) assert r == pytest.approx(-3.074, abs=0.002) and D == pytest.approx(0.0246, abs=0.0002) and D > 4 / n t = r * math.sqrt((n - k - 1) / (n - k - r ** 2)); assert t == pytest.approx(-3.097, abs=0.002) # ex. 14.4 : contamination à 50 % du marché eps, delta = 0.05, math.log(0.5) assert eps * delta == pytest.approx(-0.0347, abs=1e-4) and (math.exp(eps * delta) - 1) == pytest.approx(-0.0341, abs=1e-4) assert math.sqrt(0.085 ** 2 + eps * (1 - eps) * delta ** 2) == pytest.approx(0.173, abs=1e-3) # ex. 14.5 : poids de Fellegi–Sunter m = [0.98, 0.95, 0.90]; u = [0.001, 0.05, 0.20] w_acc = [math.log2(a / b) for a, b in zip(m, u)]; w_dis = [math.log2((1 - a) / (1 - b)) for a, b in zip(m, u)] assert sum(w_acc) == pytest.approx(16.35, abs=0.05) and (w_acc[0] + w_acc[1] + w_dis[2]) == pytest.approx(11.2, abs=0.05) assert (w_dis[0] + w_acc[1] + w_acc[2]) == pytest.approx(0.78, abs=0.05)