spb/traite-evaluation-qc

1"""Chapitre 14 — Nettoyage, géocodage et qualité des données : reproductibilité du jeu brut, identités des diagnostics,2chiffres cités dans le texte et dans les solutions (tolérances déclarées)."""3import sys, pathlib, math4import numpy as np, pandas as pd, pytest5sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1] / "cases"))6import ch14_nettoyage as c147from traite import cas1, hedonique as h89@pytest.fixture(scope="module")10def R():11    return c14.main(ecrire=False, verbose=False)1213# ----------------------------------------------------------------------------- jeu brut1415def test_brut_reproductible_et_identique_au_fichier(R):16    g1 = c14.generer_brut(); g2 = c14.generer_brut()17    pd.testing.assert_frame_equal(g1["brut"], g2["brut"])18    disque = c14.charger_brut()19    assert len(disque["brut"]) == 618 and len(disque["verite"]) == 127 and len(disque["role"]) == 60020    b = g1["brut"]; d = disque["brut"].copy()21    for c in ("date_promesse", "date_acte"): d[c] = d[c].dt.strftime("%Y-%m-%d")22    pd.testing.assert_frame_equal(b.reset_index(drop=True), d[b.columns].reset_index(drop=True), check_dtype=False)2324def test_anomalies_injectees(R):25    v = R["verite"]["anomalie"].value_counts()26    attendu = {"terrain_manquant": 20, "doublon": 18, "qualite_manquante": 15, "superficie_pi2": 12, "dates_inversees": 10, "annee_manquante": 10,27               "coord_nulles": 9, "meubles_inclus": 8, "vente_liee_declaree": 8, "coord_hors_zone": 6, "vente_liee_non_declaree": 6, "prix_symbolique": 5}28    assert v.to_dict() == attendu29    b = R["brut"]30    assert (b["prix_vente"] == 1).sum() == 5 and (b["superficie_habitable_m2"] > 450).sum() == 1231    assert b["superficie_terrain_m2"].isna().sum() == 20 and b["qualite_1a5"].isna().sum() == 15 and b["annee_construction"].isna().sum() == 1032    assert ((b["lat"] == 0) & (b["lon"] == 0)).sum() == 933    assert (pd.to_datetime(b["date_acte"]) < pd.to_datetime(b["date_promesse"])).sum() == 1034    # les lignes touchées (hors doublons) sont disjointes35    sans_dup = R["verite"][R["verite"]["anomalie"] != "doublon"]36    assert sans_dup["id_inscription"].is_unique and len(sans_dup) == 1093738# ----------------------------------------------------------------------------- pipeline3940def test_entonnoir_et_journal(R):41    P = R["pipeline"]42    assert P["etapes"] == [("0 lecture", 618), ("R1 doublons", 600), ("R4 prix symbolique", 595), ("R7 non représentatives", 587),43                           ("I1 imputation", 587), ("S1 scores robustes", 581), ("E1 régression", 579)]44    j = P["journal"]; g = j.groupby(["etape", "action"]).size().to_dict()45    assert g == {("E1", "exclusion"): 2, ("G1", "imputation"): 15, ("I1", "imputation"): 45, ("R1", "exclusion"): 18, ("R2", "correction"): 10,46                 ("R3", "correction"): 8, ("R4", "exclusion"): 5, ("R5", "correction"): 12, ("R6", "correction"): 15, ("R7", "exclusion"): 8, ("S1", "exclusion"): 6}47    assert len(j) == 14448    # 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égression49    jv = j.merge(R["verite"], on="id_inscription", how="left"); ct = pd.crosstab(jv["etape"], jv["anomalie"].fillna("propre"))50    # R1 conserve l'inscription la plus récente (la copie) et exclut l'originale : on vérifie par l'unité d'origine51    orig = dict(zip(R["brut"]["id_inscription"], R["brut"]["id_vente_origine"]))52    dup_units = set(R["verite"].loc[R["verite"]["anomalie"] == "doublon", "id_vente_origine"])53    assert {orig[i] for i in j.loc[j["etape"] == "R1", "id_inscription"]} == dup_units and len(dup_units) == 1854    assert ct.loc["R4", "prix_symbolique"] == 5 and ct.loc["R5", "superficie_pi2"] == 12 and ct.loc["R7", "vente_liee_declaree"] == 855    assert ct.loc["S1", "vente_liee_non_declaree"] == 4 and ct.loc["E1", "vente_liee_non_declaree"] == 2 and ct.loc["S1", "propre"] == 256    assert set(P["regression"]["exclus"]["id_inscription"]) == {"B0039", "B0410"}57    assert set(P["scores"]["exclus"]["id_inscription"]) == {"B0003", "B0006", "B0021", "B0105", "B0173", "B0558"}5859def test_corrections_restituent_les_valeurs_propres(R):60    """Après R2, R3, R5 les valeurs corrigées coïncident avec le jeu propre pour les lignes concernées."""61    P = R["pipeline"]; d = P["propre"].set_index("id_vente_origine"); v = cas1.charger()["ventes"].set_index("id_vente")62    ver = R["verite"]63    for code, col in [("superficie_pi2", "superficie_habitable_m2"), ("meubles_inclus", "prix_vente")]:64        ids = [i for i in ver.loc[ver["anomalie"] == code, "id_vente_origine"] if i in d.index]65        assert len(ids) >= 766        np.testing.assert_allclose(d.loc[ids, col].astype(float), v.loc[ids, col].astype(float), rtol=0, atol=0.51)67    ids = [i for i in ver.loc[ver["anomalie"] == "dates_inversees", "id_vente_origine"] if i in d.index]68    assert (pd.to_datetime(d.loc[ids, "date_vente"]).values == pd.to_datetime(v.loc[ids, "date_vente"]).values).all()6970def test_iterations_regression(R):71    it = R["pipeline"]["regression"]["iterations"]72    assert [x["n"] for x in it] == [581, 579] and [x["exclus"] for x in it] == [2, 0]73    assert it[0]["sigma"] == pytest.approx(0.0897, abs=5e-4) and it[1]["sigma"] == pytest.approx(0.0837, abs=5e-4)74    assert it[0]["cook_max"] < 1 and it[0]["n_cook_4n"] == 25 and it[0]["n_levier_2kn"] == 187576# ----------------------------------------------------------------------------- identités des diagnostics7778def test_identites_levier_cook():79    v = cas1.charger()["ventes"]; X = cas1.matrice(v).values; y = np.log(v["prix_vente"].values)80    dg = c14.diagnostics(X, y)81    assert dg["somme_h"] == pytest.approx(dg["k"], rel=1e-9)            # tr(H) = k82    assert (dg["h"] > 0).all() and (dg["h"] < 1).all()83    i = int(np.argmax(dg["cook"]))84    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²)85    # résidu externe = résidu interne × sqrt((n−k−1)/(n−k−r²))86    n, k = dg["n"], dg["k"]; r = dg["r_int"]87    np.testing.assert_allclose(dg["t_ext"], r * np.sqrt((n - k - 1) / (n - k - r ** 2)), rtol=1e-9)8889def test_constante_mad_et_score():90    from scipy.stats import norm91    assert 1 / norm.ppf(0.75) == pytest.approx(1.4826, abs=1e-4)92    rng = np.random.default_rng(1); x = rng.normal(0, 2, 200000)93    mad = np.median(np.abs(x - np.median(x)))94    assert 1.4826 * mad == pytest.approx(2.0, rel=0.01)                  # convergence vers σ95    z = c14.z_robuste(np.array([100.0, 101, 99, 100, 102, 98, 100, 1000]))96    assert z[-1] > 100 and abs(z[:-1]).max() < 2.5                        # une valeur extrême n'écrase pas les autres9798def test_masquage(R):99    t = R["masquage"]["tableau"]100    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"] == 2101    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)102    assert (t["n_robuste"] >= t["n_classique"]).all()103104# ----------------------------------------------------------------------------- effet sur le modèle105106def test_modeles_avant_apres(R):107    t = R["modeles"]["tableau"]108    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"]109    assert brut["n"] == 573 and brut["R2"] == pytest.approx(0.039, abs=2e-3) and brut["sigma"] == pytest.approx(1.231, abs=2e-3)110    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)111    assert net["n"] == 579 and net["sigma"] == pytest.approx(0.0837, abs=5e-4) and net["R2"] == pytest.approx(0.850, abs=2e-3)112    assert ref["n"] == 600 and ref["sigma"] == pytest.approx(0.0830, abs=5e-4) and ref["R2"] == pytest.approx(0.853, abs=2e-3)113    assert net["ln_S"] == pytest.approx(0.5335, abs=5e-4) and ref["ln_S"] == pytest.approx(0.5391, abs=5e-4)114    assert net["prediction_sujet"] == pytest.approx(637017, abs=5) and ref["prediction_sujet"] == pytest.approx(637953, abs=5)115    assert brut["prediction_sujet"] == pytest.approx(565471, abs=5)116    # le modèle nettoyé retrouve la référence : tous les coefficients à moins de 0,5 erreur-type117    e = R["modeles"]["ecarts"]["ecart_en_se"].abs()118    assert e.max() < 0.5 and e.idxmax() in ("ln_S", "piscine")119120# ----------------------------------------------------------------------------- géocodage, appariement, imputation, sensibilité121122def test_geocodage(R):123    G = R["geocodage"]; t = G["tableau"]124    assert G["n_dans_rayon"] == 103125    assert list(t["erreur_mediane_m"].round(0)) == [6, 18, 47, 177, 1199]126    assert t.loc["point d'adresse", "pct_voisins_conserves"] == pytest.approx(99.1, abs=0.1)127    assert t.loc["centroïde de code postal (urbain)", "pct_voisins_conserves"] == pytest.approx(76.1, abs=0.1)128    assert t.loc["centroïde de code postal (périurbain)", "pct_voisins_conserves"] == pytest.approx(17.6, abs=0.1)129    assert t.loc["centroïde de code postal (périurbain)", "pct_rayon_change"] == pytest.approx(6.7, abs=0.1)130    assert t["erreur_mediane_m"].is_monotonic_increasing and t["pct_voisins_conserves"].is_monotonic_decreasing131    # erreur médiane théorique d'une loi de Rayleigh : σ·sqrt(2 ln 2) = 1,1774 σ132    np.testing.assert_allclose(t["erreur_mediane_m"] / (1000 * t["sigma_km"]), 1.1774, rtol=0.06)133    assert abs(t.loc["point d'adresse", "beta_dist"] - G["beta_dist_ref"]) < 1e-3134135def test_appariement(R):136    A = R["appariement"]137    assert A["taux"].to_dict() == {"1 exact": 367, "2 normalisé": 198, "3 attributs": 45, "non apparié": 8}138    assert A["pct"].to_dict() == {"1 exact": 59.4, "2 normalisé": 32.0, "3 attributs": 7.3, "non apparié": 1.3}139    assert A["ambigus"] == 3 and A["faux_appariements"] == 0140    assert c14.normaliser_matricule("05 94 6466 4") == "059464664" and c14.normaliser_matricule(float("nan")) == ""141142def test_imputations(R):143    I = R["imputations"]144    assert list(I.index) == ["suppression", "mediane", "regression", "indicateur"]145    assert I.loc["suppression", "n"] == 535 and (I.loc[["mediane", "regression", "indicateur"], "n"] == 579).all()146    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)147    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)148    assert I.loc["suppression", "prediction_sujet"] == pytest.approx(640921, abs=5) and I.loc["indicateur", "prediction_sujet"] == pytest.approx(638354, abs=5)149    assert I.loc["indicateur", "coef_manq_terrain"] == pytest.approx(-0.0215, abs=5e-4)150151def test_sensibilite_seuils(R):152    T = R["sensibilite"].set_index(["seuil_mad", "seuil_t"])153    assert len(T) == 12 and (T["liees_detectees"] == 6).all()154    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"] == 0155    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"] == 581156    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)157    assert T["prediction_sujet"].min() > 633000 and T["prediction_sujet"].max() < 639000158159# ----------------------------------------------------------------------------- solutions des exercices160161def test_solutions_exercices():162    # ex. 14.1 : scores robustes de neuf prix au m²163    x = np.array([3150.0, 3320, 3080, 3410, 3260, 3190, 3370, 3240, 5980])164    med = np.median(x); mad = np.median(np.abs(x - med))165    assert med == 3260 and mad == 110166    z = (x - med) / (1.4826 * mad)167    assert z[-1] == pytest.approx(16.68, abs=0.01) and abs(z[:-1]).max() == pytest.approx(1.10, abs=0.01)168    zc = (x - x.mean()) / x.std(ddof=1)169    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)170    # ex. 14.2 : conversion 1 776 pi²171    assert 1776 / 10.7639 == pytest.approx(165.0, abs=0.05)172    # ex. 14.3 : levier et Cook173    n, k, s = 600, 16, 0.083; e, hii = -0.25, 0.04174    r = e / (s * math.sqrt(1 - hii)); D = r ** 2 / k * hii / (1 - hii)175    assert r == pytest.approx(-3.074, abs=0.002) and D == pytest.approx(0.0246, abs=0.0002) and D > 4 / n176    t = r * math.sqrt((n - k - 1) / (n - k - r ** 2)); assert t == pytest.approx(-3.097, abs=0.002)177    # ex. 14.4 : contamination à 50 % du marché178    eps, delta = 0.05, math.log(0.5)179    assert eps * delta == pytest.approx(-0.0347, abs=1e-4) and (math.exp(eps * delta) - 1) == pytest.approx(-0.0341, abs=1e-4)180    assert math.sqrt(0.085 ** 2 + eps * (1 - eps) * delta ** 2) == pytest.approx(0.173, abs=1e-3)181    # ex. 14.5 : poids de Fellegi–Sunter182    m = [0.98, 0.95, 0.90]; u = [0.001, 0.05, 0.20]183    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)]184    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)185    assert (w_dis[0] + w_acc[1] + w_acc[2]) == pytest.approx(0.78, abs=0.05)186187def test_variante_mnar(R):188    """Exercice 14.7 : terrains manquants = les 20 plus petits (MNAR)."""189    V = R["mnar"]190    assert V.loc["suppression", "n"] == 536 and V.loc["suppression", "ln_T"] == pytest.approx(0.1007, abs=5e-4)191    assert V.loc["mediane", "ln_T"] == pytest.approx(0.0977, abs=5e-4) and V.loc["regression", "ln_T"] == pytest.approx(0.0972, abs=5e-4)192    assert V.loc["indicateur", "ln_T"] == pytest.approx(0.0978, abs=5e-4)193    assert V.loc["suppression", "prediction_sujet"] == pytest.approx(638976, abs=5) and V.loc["mediane", "prediction_sujet"] == pytest.approx(636195, abs=5)194    assert V.loc["regression", "prediction_sujet"] == pytest.approx(636224, abs=5) and V.loc["indicateur", "prediction_sujet"] == pytest.approx(638201, abs=5)195    assert V.loc["indicateur", "coef_manq_terrain"] == pytest.approx(-0.0502, abs=5e-4) and V.loc["indicateur", "se_manq_terrain"] == pytest.approx(0.0143, abs=5e-4)196    assert abs(V.loc["indicateur", "coef_manq_terrain"] / V.loc["indicateur", "se_manq_terrain"]) > 3197