#!/usr/bin/env python3 # Author: Simon-Pierre Boucher — contact@spboucher.ai # """Step 7 — Revision extensions (editorial-board review, 2026-10-17). Addresses the identification and inference issues raised in the review: (a) language of the listing (French / English / mixed) and its effect on the similarity scores computed against French references; (b) extended specifications: language controls, FSA (forward sortation area) and municipality fixed effects, cluster-robust standard errors, unit- corrected lot size, floor-area subsample, French-only / English-only; (c) domain-level block tests of the semantic dimensions; (d) out-of-sample validation: random 5-fold and spatial (FSA-grouped) CV; (e) sensitivity of the estimates to the phrasing of the reference texts; (f) a multilingual embedding model (paraphrase-multilingual-MiniLM-L12-v2). Inputs : data/processed/houses.parquet, hedonic_maison_results.csv, embeddings_maisons.npy, data/raw/louka.db Outputs: results/revision_language.csv results/revision_specifications.csv, revision_spec_summary.csv results/revision_block_tests.csv results/revision_oos.csv results/revision_reference_sensitivity.csv (+ _summary) results/revision_multilingual.csv data/processed/embeddings_multilingual.npy (cache) """ import sys import time from pathlib import Path import numpy as np import pandas as pd import statsmodels.api as sm from sklearn.decomposition import PCA from sklearn.model_selection import GroupKFold, KFold sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from src import config from src.extensions import (block_f_test, detect_language, load_revision_fields, ols, oos_metrics, within, zscore) from src.references import ENGLISH_LABELS, REFERENCES, SIM_COLS STRUCT = config.STRUCTURAL_VARS LEN = ["remarks_length"] LANG = ["lang_en", "lang_mixed"] MULTI_MODEL = "paraphrase-multilingual-MiniLM-L12-v2" EMB_MULTI = config.DATA_PROCESSED / "embeddings_multilingual.npy" DOMAINS = { "Quality & Standing": ["sim_luxe", "sim_entree_de_gamme"], "Condition": ["sim_renove", "sim_a_renover"], "Physical Attributes": ["sim_lumineux_spacieux", "sim_garage_stationnement", "sim_sous_sol_amenage", "sim_piscine_amenagement"], "Location & Setting": ["sim_terrain_nature", "sim_vue_panoramique", "sim_localisation_premium", "sim_tranquillite", "sim_bord_eau"], "Style & Character": ["sim_moderne_contemporain", "sim_cachet_patrimonial", "sim_neuf_construction", "sim_ecoefficace", "sim_familial"], "Market Signals": ["sim_revenu_investissement", "sim_urgence_motivation"], } T0 = time.time() def log(msg): print(f"[{time.time() - T0:6.1f}s] {msg}", flush=True) # -------------------------------------------------------------------------- # Data assembly # -------------------------------------------------------------------------- def build_dataset(): df = pd.read_csv(config.ANALYSIS_CSV) houses = pd.read_parquet(config.HOUSES_PARQUET) df["id"] = df["id"].astype(str) assert (df["id"].values == houses["id"].astype(str).values).all() df["remarks"] = houses["remarks"].values df["latitude"] = houses["latitude"].values df["longitude"] = houses["longitude"].values extra = load_revision_fields(config.DB_PATH, df["id"].tolist()) for c in ["fsa", "municipality", "lot_sqm", "floor_sqm", "scraped_at"]: df[c] = extra[c].values df["lang"] = [detect_language(t) for t in df["remarks"]] df["lang_en"] = (df["lang"] == "en").astype(float) df["lang_mixed"] = (df["lang"] == "mixed").astype(float) df["log_lot_sqm"] = np.where(df["lot_sqm"] > 0, np.log(df["lot_sqm"].fillna(1.0)), 0.0) df["lot_missing"] = (~(df["lot_sqm"] > 0)).astype(float) df["log_floor_sqm"] = np.where(df["floor_sqm"] > 0, np.log(df["floor_sqm"].fillna(1.0)), np.nan) df["fsa_id"] = pd.factorize(df["fsa"])[0] df["mun_id"] = pd.factorize(df["municipality"])[0] return df # -------------------------------------------------------------------------- # Estimation helpers # -------------------------------------------------------------------------- def design(df, columns, fe=None): """Standardized design (z-scores on the estimation sample), optional FE demeaning.""" X = zscore(df, columns) if fe is not None: X = X - X.groupby(df[fe].values).transform("mean") return X # no constant after within transformation return sm.add_constant(X) def fit_spec(df, columns, sims, fe=None, cluster=None): """Fit y on columns (+sims); return model, model-without-sims and summary dict.""" y = df["log_price"] if fe is not None: y = y - y.groupby(df[fe].values).transform("mean") groups = df[cluster].values if cluster else None cov = "cluster" if cluster else "HC3" m_full = ols(y, design(df, columns + sims, fe), cov, groups) m_base = ols(y, design(df, columns, fe), cov, groups) f, p, q = block_f_test(m_full, sims) # R² relative to total variance of log price (not within variance) when FE if fe is not None: yfull = df["log_price"] sst = float(((yfull - yfull.mean()) ** 2).sum()) r2_full = 1 - float((m_full.resid ** 2).sum()) / sst r2_base = 1 - float((m_base.resid ** 2).sum()) / sst r2_within_full = m_full.rsquared else: r2_full, r2_base, r2_within_full = m_full.rsquared, m_base.rsquared, np.nan summ = {"n": int(m_full.nobs), "r2": r2_full, "r2_without_sims": r2_base, "delta_r2_sims": r2_full - r2_base, "r2_within": r2_within_full, "block_F": f, "block_p": p, "block_df": q, "n_sig_5pct": int(sum(m_full.pvalues[s] < 0.05 for s in sims))} return m_full, summ def coef_rows(spec, model, sims, label_map): rows = [] for v in sims + [c for c in model.params.index if c in STRUCT + LEN + LANG + ["log_lot_sqm", "lot_missing", "log_floor_sqm"]]: if v not in model.params.index: continue rows.append({"spec": spec, "variable": v, "label": label_map.get(v, v), "coefficient": model.params[v], "std_error": model.bse[v], "p_value": model.pvalues[v], "impact_pct": (np.exp(model.params[v]) - 1) * 100}) return rows LABELS = {**{f"sim_{k}": v for k, v in ENGLISH_LABELS.items()}, "bedrooms": "Bedrooms", "bathrooms": "Bathrooms", "half_baths": "Half-bathrooms", "parking": "Parking", "stories": "Stories", "land_size": "Lot size (raw field)", "remarks_length": "Description length", "lang_en": "English listing", "lang_mixed": "Mixed / uninformative", "log_lot_sqm": "log lot area (m2)", "lot_missing": "Lot area missing", "log_floor_sqm": "log floor area (m2)"} # -------------------------------------------------------------------------- # (a) Language # -------------------------------------------------------------------------- def language_analysis(df, S, tag, out_rows): msim = S.mean(axis=1) for lang in ["fr", "en", "mixed"]: mask = (df["lang"] == lang).values out_rows.append({"model": tag, "language": lang, "n": int(mask.sum()), "share": mask.mean(), "mean_similarity": msim[mask].mean(), "mean_log_price": df.loc[mask, "log_price"].mean(), "median_price": df.loc[mask, "price"].median(), "mean_length": df.loc[mask, "remarks_length"].mean()}) # eta² of language on mean similarity grand = msim.mean() ss_b = sum(((df["lang"] == l).sum()) * (msim[(df["lang"] == l).values].mean() - grand) ** 2 for l in ["fr", "en", "mixed"]) ss_t = ((msim - grand) ** 2).sum() out_rows.append({"model": tag, "language": "eta2_language_on_mean_similarity", "n": len(df), "share": np.nan, "mean_similarity": ss_b / ss_t, "mean_log_price": np.nan, "median_price": np.nan, "mean_length": np.nan}) # correlation of mean similarity with length and with English dummy out_rows.append({"model": tag, "language": "corr_mean_similarity_length", "n": len(df), "share": np.nan, "mean_similarity": np.corrcoef(msim, df["remarks_length"])[0, 1], "mean_log_price": np.nan, "median_price": np.nan, "mean_length": np.nan}) out_rows.append({"model": tag, "language": "corr_mean_similarity_english", "n": len(df), "share": np.nan, "mean_similarity": np.corrcoef(msim, df["lang_en"])[0, 1], "mean_log_price": np.nan, "median_price": np.nan, "mean_length": np.nan}) # first PC share of the 20 similarities pc = PCA(1).fit(S) out_rows.append({"model": tag, "language": "pc1_share_of_similarity_variance", "n": len(df), "share": np.nan, "mean_similarity": pc.explained_variance_ratio_[0], "mean_log_price": np.nan, "median_price": np.nan, "mean_length": np.nan}) # -------------------------------------------------------------------------- # (d) Out-of-sample # -------------------------------------------------------------------------- def oos_validation(df, E, sim_sets, out_rows): y = df["log_price"].values folds = {"random_5fold": list(KFold(5, shuffle=True, random_state=config.SEED).split(df)), "spatial_fsa_5fold": list(GroupKFold(5).split(df, groups=df["fsa_id"]))} # The raw lot-size field (mixed units, max 7.2e8) makes out-of-sample # predictions explode when an extreme value falls in the test fold; all # specifications below therefore use the unit-corrected lot area, except # the first row, kept to document the breakdown. STRUCT_OOS = STRUCT[:-1] + ["log_lot_sqm", "lot_missing"] specs = {"A (raw lot-size field)": (STRUCT, None, False), "A: structural": (STRUCT_OOS, None, False), "B: + length": (STRUCT_OOS + LEN, None, False), "B-L: + length + language": (STRUCT_OOS + LEN + LANG, None, False), "D: + length + 20 similarities": (STRUCT_OOS + LEN, "base", False), "D-L: D + language": (STRUCT_OOS + LEN + LANG, "base", False), "PCA-20 + length": (STRUCT_OOS + LEN, None, True), "PCA-20 + length + language": (STRUCT_OOS + LEN + LANG, None, True), "ML-D: multilingual similarities": (STRUCT_OOS + LEN, "multi", False), "ML-D-L: multilingual similarities + language": (STRUCT_OOS + LEN + LANG, "multi", False)} for scheme, splits in folds.items(): for name, (cols, simkey, use_pca) in specs.items(): if simkey == "multi" and "multi" not in sim_sets: continue preds = np.full(len(df), np.nan) for tr, te in splits: Xtr = df.iloc[tr][cols].values.astype(float) Xte = df.iloc[te][cols].values.astype(float) if simkey: Xtr = np.hstack([Xtr, sim_sets[simkey][tr]]) Xte = np.hstack([Xte, sim_sets[simkey][te]]) if use_pca: p = PCA(20, random_state=config.SEED).fit(E[tr]) Xtr = np.hstack([Xtr, p.transform(E[tr])]) Xte = np.hstack([Xte, p.transform(E[te])]) Xtr = np.hstack([np.ones((len(tr), 1)), Xtr]) Xte = np.hstack([np.ones((len(te), 1)), Xte]) beta, *_ = np.linalg.lstsq(Xtr, y[tr], rcond=None) preds[te] = Xte @ beta met = oos_metrics(pd.Series(y), pd.Series(preds)) out_rows.append({"scheme": scheme, "specification": name, **met}) log(f" OOS {scheme:18s} {name:48s} R2={met['r2_oos']:.4f} MdAPE={met['mdape_pct']:.1f}") # -------------------------------------------------------------------------- # (e) Reference phrasing sensitivity # -------------------------------------------------------------------------- def reference_variants(): """Three systematic perturbations of each reference text.""" variants = {"first_half": {}, "second_half": {}, "reversed": {}} for key, text in REFERENCES.items(): phrases = [p.strip() for p in text.replace(".", ",").split(",") if p.strip()] h = int(np.ceil(len(phrases) / 2)) variants["first_half"][key] = ", ".join(phrases[:h]) + "." variants["second_half"][key] = ", ".join(phrases[h:]) + "." variants["reversed"][key] = ", ".join(phrases[::-1]) + "." return variants def reference_sensitivity(df, E, base_model, out_rows, summ_rows): from sentence_transformers import SentenceTransformer model = SentenceTransformer(config.EMBEDDING_MODEL) base_coef = base_model.params[SIM_COLS] sig = [s for s in SIM_COLS if base_model.pvalues[s] < 0.05] for vname, refs in reference_variants().items(): R = model.encode([refs[k] for k in REFERENCES], normalize_embeddings=True) S_v = E @ R.T dv = df.copy() for i, c in enumerate(SIM_COLS): dv[c] = S_v[:, i] m, s = fit_spec(dv, STRUCT + LEN, SIM_COLS) coef = m.params[SIM_COLS] for c in SIM_COLS: out_rows.append({"variant": vname, "variable": c, "label": LABELS[c], "coefficient": coef[c], "std_error": m.bse[c], "p_value": m.pvalues[c], "corr_similarity_with_baseline": np.corrcoef(S_v[:, SIM_COLS.index(c)], df[c])[0, 1]}) summ_rows.append({"variant": vname, "r2": s["r2"], "delta_r2_sims": s["delta_r2_sims"], "block_F": s["block_F"], "corr_coef_with_baseline": np.corrcoef(coef, base_coef)[0, 1], "sign_agreement_sig16": np.mean([np.sign(coef[c]) == np.sign(base_coef[c]) for c in sig]), "n_sig_5pct": s["n_sig_5pct"], "mean_abs_coef_change": float(np.mean(np.abs(coef - base_coef))), "mean_corr_similarity": float(np.mean([np.corrcoef(S_v[:, i], df[c])[0, 1] for i, c in enumerate(SIM_COLS)]))}) log(f" variant {vname}: R2={s['r2']:.4f} corr(coef)={summ_rows[-1]['corr_coef_with_baseline']:.3f}") # -------------------------------------------------------------------------- # (f) Multilingual embeddings # -------------------------------------------------------------------------- def multilingual_similarities(df): from sentence_transformers import SentenceTransformer model = SentenceTransformer(MULTI_MODEL) R = model.encode(list(REFERENCES.values()), normalize_embeddings=True) if EMB_MULTI.exists(): E = np.load(EMB_MULTI) if len(E) != len(df): E = None else: E = None if E is None: log(f" encoding {len(df):,} descriptions with {MULTI_MODEL} ...") E = model.encode(df["remarks"].tolist(), batch_size=128, show_progress_bar=False, normalize_embeddings=True) np.save(EMB_MULTI, E) return E @ R.T, E def main(): log("assembling dataset ...") df = build_dataset() E = np.load(config.EMBEDDINGS_NPY) S = df[SIM_COLS].values log(f"n={len(df):,}; FSA={df['fsa'].nunique()}, municipalities={df['municipality'].nunique()}; " f"lot area parsed for {(df['lot_sqm'] > 0).mean():.1%}; floor area for {(df['floor_sqm'] > 0).mean():.1%}") log(f"language shares: {df['lang'].value_counts(normalize=True).round(3).to_dict()}") config.RESULTS_DIR.mkdir(parents=True, exist_ok=True) # (a) language ---------------------------------------------------------- lang_rows = [] language_analysis(df, S, "all-MiniLM-L6-v2", lang_rows) per_dim = [] for c in SIM_COLS: per_dim.append({"variable": c, "label": LABELS[c], "mean_fr": df.loc[df.lang == "fr", c].mean(), "mean_en": df.loc[df.lang == "en", c].mean(), "sd_all": df[c].std(ddof=0)}) pd.DataFrame(per_dim).to_csv(config.RESULTS_DIR / "revision_language_by_dimension.csv", index=False) # (b) specifications ---------------------------------------------------- spec_rows, summ_rows = [], [] def run(spec, data, cols, sims=SIM_COLS, fe=None, cluster=None, note=""): m, s = fit_spec(data, cols, sims, fe, cluster) s.update({"spec": spec, "fe": fe or "none", "cov": cluster or "HC3", "note": note}) summ_rows.append(s) spec_rows.extend(coef_rows(spec, m, sims, LABELS)) log(f" {spec:34s} n={s['n']:6d} R2={s['r2']:.4f} dR2(sims)={s['delta_r2_sims']:.4f} " f"F={s['block_F']:6.2f} sig={s['n_sig_5pct']:2d}/20") return m log("extended specifications ...") mD = run("D (baseline)", df, STRUCT + LEN) run("D-L: + language", df, STRUCT + LEN + LANG) run("D-L, FSA-clustered SE", df, STRUCT + LEN + LANG, cluster="fsa_id") run("D-L + FSA FE", df, STRUCT + LEN + LANG, fe="fsa_id", cluster="fsa_id") run("D-L + municipality FE", df, STRUCT + LEN + LANG, fe="mun_id", cluster="mun_id") run("D-L + FSA FE, lot area (m2)", df, STRUCT[:-1] + ["log_lot_sqm", "lot_missing"] + LEN + LANG, fe="fsa_id", cluster="fsa_id", note="land_size replaced by unit-corrected lot area") dff = df[df["floor_sqm"] > 0].copy() run("D-L + FSA FE, floor-area subsample", dff, STRUCT[:-1] + ["log_lot_sqm", "lot_missing", "log_floor_sqm"] + LEN + LANG, fe="fsa_id", cluster="fsa_id", note="listings reporting interior floor area") dfr = df[df["lang"] == "fr"].copy() run("French-only, D + FSA FE", dfr, STRUCT + LEN, fe="fsa_id", cluster="fsa_id") dfe = df[df["lang"] == "en"].copy() run("English-only, D + FSA FE", dfe, STRUCT + LEN, fe="fsa_id", cluster="fsa_id") run("D (baseline), FSA-clustered SE", df, STRUCT + LEN, cluster="fsa_id") # (c) domain block tests under the preferred specification --------------- log("domain block tests ...") y_w = df["log_price"] - df["log_price"].groupby(df["fsa_id"].values).transform("mean") X_w = design(df, STRUCT + LEN + LANG + SIM_COLS, fe="fsa_id") m_fe = ols(y_w, X_w, "cluster", df["fsa_id"].values) m_hc = ols(df["log_price"], design(df, STRUCT + LEN + LANG + SIM_COLS), "HC3") block_rows = [] for dom, vars_ in DOMAINS.items(): for spec, m in [("D-L (HC3)", m_hc), ("D-L + FSA FE (clustered)", m_fe)]: f, p, q = block_f_test(m, vars_) block_rows.append({"spec": spec, "domain": dom, "n_dims": q, "F": f, "p_value": p}) f, p, q = block_f_test(m_fe, SIM_COLS) block_rows.append({"spec": "D-L + FSA FE (clustered)", "domain": "All 20 dimensions", "n_dims": q, "F": f, "p_value": p}) f, p, q = block_f_test(m_hc, SIM_COLS) block_rows.append({"spec": "D-L (HC3)", "domain": "All 20 dimensions", "n_dims": q, "F": f, "p_value": p}) pd.DataFrame(block_rows).to_csv(config.RESULTS_DIR / "revision_block_tests.csv", index=False) # (e) reference phrasing sensitivity ------------------------------------- log("reference phrasing sensitivity ...") sens_rows, sens_summ = [], [] reference_sensitivity(df, E, mD, sens_rows, sens_summ) pd.DataFrame(sens_rows).to_csv(config.RESULTS_DIR / "revision_reference_sensitivity.csv", index=False) pd.DataFrame(sens_summ).to_csv(config.RESULTS_DIR / "revision_reference_sensitivity_summary.csv", index=False) # (f) multilingual model ------------------------------------------------- log("multilingual embeddings ...") S_multi, E_multi = multilingual_similarities(df) language_analysis(df, S_multi, MULTI_MODEL, lang_rows) pd.DataFrame(lang_rows).to_csv(config.RESULTS_DIR / "revision_language.csv", index=False) ML_COLS = [c.replace("sim_", "ml_") for c in SIM_COLS] for i, c in enumerate(ML_COLS): df[c] = S_multi[:, i] ml_labels = {c.replace("sim_", "ml_"): v for c, v in LABELS.items() if c.startswith("sim_")} LABELS.update(ml_labels) corr_models = pd.DataFrame({"variable": SIM_COLS, "label": [LABELS[c] for c in SIM_COLS], "corr_minilm_multilingual": [np.corrcoef(df[c], df[m])[0, 1] for c, m in zip(SIM_COLS, ML_COLS)], "mean_fr_multi": [df.loc[df.lang == "fr", m].mean() for m in ML_COLS], "mean_en_multi": [df.loc[df.lang == "en", m].mean() for m in ML_COLS]}) corr_models.to_csv(config.RESULTS_DIR / "revision_multilingual.csv", index=False) run("ML-D: multilingual similarities", df, STRUCT + LEN, sims=ML_COLS) run("ML-D-L: + language", df, STRUCT + LEN + LANG, sims=ML_COLS) run("ML-D-L + FSA FE", df, STRUCT + LEN + LANG, sims=ML_COLS, fe="fsa_id", cluster="fsa_id") run("ML-D-L + FSA FE, lot area (m2)", df, STRUCT[:-1] + ["log_lot_sqm", "lot_missing"] + LEN + LANG, sims=ML_COLS, fe="fsa_id", cluster="fsa_id") pd.DataFrame(spec_rows).to_csv(config.RESULTS_DIR / "revision_specifications.csv", index=False) pd.DataFrame(summ_rows).to_csv(config.RESULTS_DIR / "revision_spec_summary.csv", index=False) # (d) out-of-sample ------------------------------------------------------- log("out-of-sample validation ...") oos_rows = [] oos_validation(df, E, {"base": S, "multi": S_multi}, oos_rows) pd.DataFrame(oos_rows).to_csv(config.RESULTS_DIR / "revision_oos.csv", index=False) df[["id", "lang", "fsa", "municipality", "lot_sqm", "floor_sqm"]].to_csv( config.DATA_PROCESSED / "revision_fields.csv", index=False) log("done.") if __name__ == "__main__": main()