1"""Cleaning, normalisation, linkage keys and sample construction (Phase 3).23Produces in data/processed:4 listings.parquet one row per listing (anonymised ids, normalised fields, sample flags)5 price_events.parquet sustained price events (entry / cut / rise / minor)6 exclusions.csv auditable exclusion log (uid, rule, value)7 sample_construction.csv attrition table for the paper8 sources_registry.csv public registry of sources with class and quality flags9 id_maps/ (interim) uid -> listing_id, dealer key -> dealer_id (restricted)10"""11from __future__ import annotations1213import csv14import hashlib15import json16import math17import re18from pathlib import Path1920import numpy as np21import polars as pl2223from . import config as C24from . import pricepaths2526# ----------------------------------------------------------------------------- helpers2728def _norm(s: str | None) -> str:29 return (s or "").strip().lower()303132def powertrain(fuel: str | None, engine: str | None, title: str | None, make: str | None) -> str:33 f = _norm(fuel)34 if "rechargeable" in f or "plug" in f:35 return "PHEV"36 if "hybride" in f or "hybrid" in f:37 return "HEV"38 if "électrique" in f or "electrique" in f or f in {"ev", "electric"}:39 return "BEV"40 if f.startswith("diesel"):41 return "Diesel"42 if f.startswith("essence") or "flex" in f or "polycarb" in f or "éthanol" in f or "propane" in f:43 return "ICE"44 if _norm(make) in {"tesla", "polestar"}:45 return "BEV"46 e = _norm(engine) + " " + _norm(title)47 if re.search(r"\b(électrique|electric|\bev\b|kwh)", e):48 return "BEV"49 return "unknown"505152_BODY = [53 (r"vus|sport utility|suv|crossover|utilitaire", "SUV"),54 (r"camionnette|pickup|pick-up|truck|camion", "Pickup"),55 (r"fourgon|minivan|van", "Van"),56 (r"berline|sedan|4dr car|4 portes|intermédiaire|compact", "Sedan"),57 (r"hayon|hatch|5 portes", "Hatchback"),58 (r"coupé|coupe|2 portes|2dr|voiture 2 portes", "Coupe"),59 (r"cabriolet|convertible|décapotable", "Convertible"),60 (r"familiale|wagon", "Wagon"),61]626364def body_segment(body: str | None) -> str | None:65 b = _norm(body)66 if not b or b in {"n/d", "autre", "other", "custom", "n.d."}:67 return None68 for pat, lab in _BODY:69 if re.search(pat, b):70 return lab71 return None727374def transmission_norm(t: str | None) -> str:75 s = _norm(t)76 if not s or s in {"n/a", "0", "null", "na", "unknown", "other", "transmission"}:77 return "unknown"78 if "manuel" in s or "manual" in s:79 return "Manual"80 return "Automatic"818283def drivetrain_norm(d: str | None) -> str:84 s = _norm(d)85 if not s or s in {"n/a", "null", "other", "not defined"}:86 return "unknown"87 if "intégrale" in s or "4x4" in s or "four" in s or "4 roues" in s or "quatre" in s or "awd" in s:88 return "AWD/4x4"89 if "traction" in s or "fwd" in s or "4x2" in s:90 return "FWD"91 if "propulsion" in s or "rwd" in s:92 return "RWD"93 return "unknown"949596def source_class(src: str) -> str:97 if src in C.SAMPLED_PORTALS:98 return "sampled_portal"99 if src in C.PRIVATE_MARKETPLACE:100 return "private_marketplace"101 if src in C.FULL_PORTALS:102 return "full_portal"103 if src in C.BANNERS:104 return "banner"105 if src in C.SMALL_PORTALS:106 return "small_portal"107 return "direct_dealer"108109110def _hid(prefix: str, key: str, n: int = 10) -> str:111 return prefix + hashlib.sha1(key.encode()).hexdigest()[:n]112113114def vehicle_age_years(ts: float, year: int) -> float:115 """Age at observation: years since 1 October of (model year - 1), floored at 0."""116 import datetime as dt117 ref = dt.datetime(year - 1, 10, 1, tzinfo=dt.timezone.utc).timestamp()118 return max(0.0, (ts - ref) / (365.25 * 86400))119120121# ----------------------------------------------------------------------------- region recovery122123def build_city_region_map(v: pl.DataFrame) -> dict[str, str]:124 d = (v.filter((pl.col("region") != "") & pl.col("region").is_not_null() & (pl.col("city") != "") & pl.col("city").is_not_null())125 .with_columns(pl.col("city").str.to_lowercase().str.strip_chars())126 .group_by(["city", "region"]).len().sort(["city", "len"], descending=[False, True])127 .unique(subset=["city"], keep="first"))128 return dict(zip(d["city"].to_list(), d["region"].to_list()))129130131def nearest_city_region(lat: float, lng: float, cities: list[tuple[str, float, float]], cmap: dict[str, str]) -> str | None:132 best, bd = None, 1e9133 for name, clat, clng in cities:134 d = (lat - clat) ** 2 + ((lng - clng) * math.cos(math.radians(lat))) ** 2135 if d < bd:136 bd, best = d, name137 if best is None or bd > (0.5 ** 2): # > ~50 km from any known city -> leave missing138 return None139 return cmap.get(best)140141142REGION_FIX = {"Saguenay-Lac-Saint-Jean": "Saguenay–Lac-Saint-Jean"}143144145# ----------------------------------------------------------------------------- main146147def run(interim: Path = C.INTERIM, out: Path = C.PROCESSED, registry: Path = C.RAW / "registry") -> dict:148 out.mkdir(parents=True, exist_ok=True)149 (interim / "id_maps").mkdir(exist_ok=True)150 v = pl.read_parquet(interim / "vehicles.parquet")151 pl_log = pl.read_parquet(interim / "price_log.parquet")152 sync = pl.read_parquet(interim / "sync_log.parquet")153 snap_ts = C.SNAPSHOT_TS154 excl: list[tuple[str, str, str]] = []155156 # --- sources registry ---------------------------------------------------157 reg = json.loads((registry / "sources.json").read_text())["sources"]158 reg_rows = []159 for s in reg:160 reg_rows.append({"source": s["id"], "name": s.get("name"), "type": s.get("type", "direct"),161 "connector": s.get("connector"), "platform": s.get("platform"), "region": s.get("region"),162 "city": s.get("city"), "status": s.get("status"), "class": source_class(s["id"]),163 "in_dynamic_sample": source_class(s["id"]) in {"direct_dealer", "full_portal", "banner"} and s["id"] not in C.FLICKER_OR_FAILED})164 # sync quality per source165 sq = (sync.group_by("source").agg([pl.len().alias("n_syncs"), pl.col("ok").sum().alias("n_ok"),166 pl.col("found").mean().alias("mean_found"), pl.col("added").sum().alias("added"),167 pl.col("removed").sum().alias("removed")]))168 inact = v.group_by("source").agg([(pl.col("active") == 0).sum().alias("n_inactive"), pl.len().alias("n_listings")])169 sq = sq.join(inact, on="source", how="left").with_columns(170 ((pl.col("removed") - pl.col("n_inactive")) / pl.col("removed").clip(lower_bound=1)).alias("reactivation_share"))171 regdf = pl.DataFrame(reg_rows).join(sq, on="source", how="left")172 regdf.write_csv(out / "sources_registry.csv")173174 # first non-empty crawl per source: entries within 24 h of it are the source's start-up stock (left-censored, D-014)175 ff = sync.filter(pl.col("found") > 0).group_by("source").agg(pl.col("ts").min().alias("first_found"))176 first_found = dict(zip(ff["source"].to_list(), ff["first_found"].to_list()))177178 # crawl interval per source (median seconds) for interval-censoring midpoints179 iv = (sync.sort(["source", "ts"]).with_columns((pl.col("ts") - pl.col("ts").shift(1).over("source")).alias("dt"))180 .filter(pl.col("dt").is_not_null()).group_by("source").agg(pl.col("dt").median().alias("crawl_median_s")))181 med_iv = dict(zip(iv["source"].to_list(), iv["crawl_median_s"].to_list()))182183 # --- price paths ---------------------------------------------------------184 events, psum = pricepaths.build(pl_log, v)185186 # --- region recovery -----------------------------------------------------187 cmap = build_city_region_map(v)188 cmap = {k: REGION_FIX.get(r, r) for k, r in cmap.items()}189 gps = json.loads((registry / "villes_gps.json").read_text())190 cities = [(name, c[0], c[1]) for name, c in gps.items() if name in cmap]191192 # --- per-listing table ---------------------------------------------------193 v = v.join(psum, on="uid", how="left")194 rows = []195 dealer_keys: dict[str, str] = {}196 vin_re = re.compile(r"^[A-HJ-NPR-Z0-9]{17}$")197 for r in v.iter_rows(named=True):198 uid, src = r["uid"], r["source"]199 cls = source_class(src)200 region = REGION_FIX.get(r["region"] or "", r["region"] or "") or None201 city = (r["city"] or "").strip() or None202 if region is None and city:203 region = cmap.get(city.lower())204 region_recovered = False205 if region is None and r["lat"] is not None and r["lng"] is not None:206 region = nearest_city_region(r["lat"], r["lng"], cities, cmap)207 region_recovered = region is not None208 dealer_key = src if cls in {"direct_dealer", "banner", "small_portal"} else f"{src}|{(r['dealer_name'] or '').strip().lower()}"209 if cls == "private_marketplace":210 dealer_key = "kijiji|private"211 dealer_id = dealer_keys.setdefault(dealer_key, _hid("d", dealer_key, 8))212 vin = (r["vin"] or "").strip().upper()213 vin_ok = bool(vin_re.match(vin))214 year = r["year"]215 entry_ts = r["first_seen"]216 entry_price = r["entry_price"] if r["entry_price"] is not None else r["price"]217 # exit variables (D-004)218 active = int(r["active"] or 0)219 last_seen = r["last_seen"]220 exit_confirmed = (active == 0) and (last_seen <= snap_ts - C.EXIT_CONFIRM_DAYS * 86400)221 half_iv = 0.5 * med_iv.get(src, 3.3 * 3600)222 if exit_confirmed:223 end_ts = min(last_seen + half_iv, snap_ts)224 else:225 end_ts = last_seen if active == 0 else snap_ts226 rows.append({227 "listing_id": _hid("l", uid), "uid": uid, "title": r["title"], "source": src, "source_class": cls,228 "dealer_id": dealer_id, "kind": r["kind"], "make": (r["make"] or "").strip() or None,229 "model": (r["model"] or "").strip() or None, "trim": (r["trim"] or "").strip() or None, "year": year,230 "price_raw": r["price"], "entry_price": entry_price, "last_price": r["last_price"] if r["last_price"] is not None else r["price"],231 "mileage_km": r["mileage_km"], "transmission": transmission_norm(r["transmission"]),232 "drivetrain": drivetrain_norm(r["drivetrain"]), "powertrain": powertrain(r["fuel"], r["engine"], r["title"], r["make"]),233 "body": body_segment(r["body_type"]), "exterior_color": (r["exterior_color"] or "").strip().lower() or None,234 "region": region, "region_recovered": region_recovered, "city": city,235 "lat_r": round(r["lat"], 2) if r["lat"] is not None else None, "lng_r": round(r["lng"], 2) if r["lng"] is not None else None,236 "lat": r["lat"], "lng": r["lng"], # exact coords stay in interim-only columns (dropped from release)237 "vin_ok": vin_ok, "vin_hash": _hid("v", vin, 12) if vin_ok else None,238 "has_stock_number": bool((r["stock_number"] or "").strip()), "is_canonical": r["dup_of"] is None,239 "n_dup_sources": len(json.loads(r["dup_sources"])) if r["dup_sources"] else 0,240 "entry_ts": entry_ts, "last_seen_ts": last_seen, "end_ts": end_ts, "active": active,241 "exit_confirmed": exit_confirmed,242 "left_censored": (entry_ts < C.LEFT_CENSOR_CUTOFF_TS) or (entry_ts < first_found.get(src, C.WINDOW_START_TS) + C.SOURCE_STARTUP_SECONDS),243 "duration_days": (end_ts - entry_ts) / 86400,244 "n_cuts": r["n_cuts"] or 0, "n_rises": r["n_rises"] or 0, "n_transients": r["n_transients"] or 0,245 "oscillating": bool(r["oscillating"]) if r["oscillating"] is not None else False,246 "first_cut_ts": r["first_cut_ts"], "first_cut_pct": r["first_cut_pct"], "total_change_pct": r["total_change_pct"],247 "age_years_entry": vehicle_age_years(entry_ts, int(year)) if year and 1980 <= year <= 2027 else None,248 })249 L = pl.DataFrame(rows)250251 # --- coordinates outside the Quebec bounding box are errors -> null (flagged) ---252 inbox = pl.col("lat").is_between(44.0, 63.0) & pl.col("lng").is_between(-80.0, -56.0)253 L = L.with_columns([254 (pl.col("lat").is_not_null() & ~inbox).alias("coords_out_of_bbox"),255 pl.when(inbox).then(pl.col("lat")).otherwise(None).alias("lat"),256 pl.when(inbox).then(pl.col("lng")).otherwise(None).alias("lng"),257 ])258259 # --- coordinates: fill missing from the dealer's median location (flagged) ---260 dm = (L.filter(pl.col("lat").is_not_null()).group_by("dealer_id")261 .agg(pl.col("lat").median().alias("_dlat"), pl.col("lng").median().alias("_dlng")))262 L = L.join(dm, on="dealer_id", how="left").with_columns([263 (pl.col("lat").is_null() & pl.col("_dlat").is_not_null()).alias("coords_from_dealer"),264 pl.when(pl.col("lat").is_null()).then(pl.col("_dlat")).otherwise(pl.col("lat")).alias("lat"),265 pl.when(pl.col("lng").is_null()).then(pl.col("_dlng")).otherwise(pl.col("lng")).alias("lng"),266 ]).drop(["_dlat", "_dlng"])267268 # --- coordinates: fill missing from city centroid (flagged) --------------269 gps_l = {k: v for k, v in gps.items()}270 L = L.with_columns([271 pl.col("city").str.to_lowercase().str.strip_chars().alias("_city_l")])272 fill_lat = pl.col("_city_l").map_elements(lambda c: gps_l.get(c, [None, None])[0] if c else None, return_dtype=pl.Float64)273 fill_lng = pl.col("_city_l").map_elements(lambda c: gps_l.get(c, [None, None])[1] if c else None, return_dtype=pl.Float64)274 L = L.with_columns([275 (pl.col("lat").is_null() & fill_lat.is_not_null()).alias("coords_from_city"),276 pl.when(pl.col("lat").is_null()).then(fill_lat).otherwise(pl.col("lat")).alias("lat"),277 pl.when(pl.col("lng").is_null()).then(fill_lng).otherwise(pl.col("lng")).alias("lng"),278 ]).drop("_city_l")279 L = L.with_columns([pl.col("lat").round(2).alias("lat_r"), pl.col("lng").round(2).alias("lng_r")])280281 # --- powertrain recovery: title keywords, then make-model(-trim) modal (>= 90 % agreement) ---282 kw = pl.col("title").fill_null("").str.to_lowercase()283 L = L.with_columns(284 pl.when(pl.col("powertrain") != "unknown").then(pl.col("powertrain"))285 .when(kw.str.contains("rechargeable|plug-in|phev")).then(pl.lit("PHEV"))286 .when(kw.str.contains("hybrid")).then(pl.lit("HEV"))287 .when(kw.str.contains(r"électrique|electric|\bev\b")).then(pl.lit("BEV"))288 .when(kw.str.contains("diesel|tdi")).then(pl.lit("Diesel"))289 .otherwise(pl.lit("unknown")).alias("powertrain"))290 known = L.filter(pl.col("powertrain") != "unknown")291 def modal(keys):292 g = known.group_by(keys + ["powertrain"]).len()293 tot = g.group_by(keys).agg(pl.col("len").sum().alias("tot"))294 g = g.join(tot, on=keys).with_columns((pl.col("len") / pl.col("tot")).alias("sh")).filter(pl.col("sh") >= 0.9)295 return g.select(keys + [pl.col("powertrain").alias("pt_fill")])296 L = L.with_columns(pl.col("powertrain").alias("_pt0"))297 for keys in (["make", "model", "trim", "year"], ["make", "model", "year"], ["make", "model"]):298 m = modal(keys)299 L = L.join(m, on=keys, how="left").with_columns(300 pl.when((pl.col("powertrain") == "unknown") & pl.col("pt_fill").is_not_null()).then(pl.col("pt_fill")).otherwise(pl.col("powertrain")).alias("powertrain")).drop("pt_fill")301 L = L.with_columns(((pl.col("_pt0") == "unknown") & (pl.col("powertrain") != "unknown")).alias("powertrain_recovered")).drop("_pt0")302303 # --- body recovery from make-model modal body ----------------------------304 mm = (L.filter(pl.col("body").is_not_null()).group_by(["make", "model", "body"]).len()305 .sort(["make", "model", "len"], descending=[False, False, True]).unique(subset=["make", "model"], keep="first")306 .rename({"body": "body_mm"}).select(["make", "model", "body_mm"]))307 L = L.join(mm, on=["make", "model"], how="left").with_columns(308 pl.when(pl.col("body").is_null()).then(pl.col("body_mm")).otherwise(pl.col("body")).alias("body_filled"),309 pl.col("body").is_null().alias("body_recovered")).drop("body_mm")310311 # --- sample flags + exclusion log ---------------------------------------312 def flag(cond: pl.Expr, name: str):313 return pl.when(cond).then(pl.lit(name)).otherwise(pl.lit(None))314315 L = L.with_columns([316 (pl.col("kind") == "auto").alias("f_auto"),317 pl.col("is_canonical").alias("f_canonical"),318 pl.col("year").is_between(C.YEAR_MIN, C.YEAR_MAX).fill_null(False).alias("f_year"),319 pl.col("mileage_km").is_between(C.KM_MIN, C.KM_MAX).fill_null(False).alias("f_km"),320 pl.col("entry_price").is_between(C.PRICE_MIN, C.PRICE_MAX).fill_null(False).alias("f_price"),321 (pl.col("make").is_not_null() & pl.col("model").is_not_null()).alias("f_mm"),322 (~pl.col("uid").is_in(list(C.UID_COLLISIONS))).alias("f_nocollision"),323 (~pl.col("source").is_in(list(C.SAMPLED_PORTALS | C.PRIVATE_MARKETPLACE))).alias("f_not_sampled_or_private"),324 (~pl.col("source").is_in(list(C.FLICKER_OR_FAILED | C.SMALL_PORTALS))).alias("f_source_reliable"),325 ])326 L = L.with_columns([327 (pl.col("f_auto") & pl.col("f_canonical") & pl.col("f_year") & pl.col("f_km") & pl.col("f_price") & pl.col("f_mm") & pl.col("f_nocollision")).alias("s1_all_sources"),328 ])329 L = L.with_columns([330 (pl.col("s1_all_sources") & pl.col("f_not_sampled_or_private")).alias("s1_hedonic"),331 ])332 L = L.with_columns([333 (pl.col("s1_hedonic") & pl.col("f_source_reliable") & ~pl.col("oscillating") & ~pl.col("left_censored")).alias("s2_dynamic"),334 (pl.col("s1_hedonic") & pl.col("f_source_reliable") & ~pl.col("oscillating")).alias("s2l_dynamic_with_stock"),335 ])336337 # exclusion log (sequential attrition, S0 -> S1 -> S2)338 steps = [339 ("kind != auto", ~pl.col("f_auto")), ("VIN duplicate (non-canonical)", ~pl.col("f_canonical")),340 (f"model year outside [{C.YEAR_MIN},{C.YEAR_MAX}] or missing", ~pl.col("f_year")),341 (f"mileage outside [{C.KM_MIN:.0f},{C.KM_MAX:.0f}] km or missing", ~pl.col("f_km")),342 (f"entry price outside [{C.PRICE_MIN:.0f},{C.PRICE_MAX:.0f}] $ or missing/placeholder", ~pl.col("f_price")),343 ("make or model missing", ~pl.col("f_mm")), ("uid identity collision", ~pl.col("f_nocollision")),344 ("sampled portal (autotrader/cargurus) or private marketplace (kijiji)", ~pl.col("f_not_sampled_or_private")),345 ("flickering/failed connector or small portal", ~pl.col("f_source_reliable")),346 ("oscillating price path", pl.col("oscillating")), ("left-censored stock (day 1 or source start-up day)", pl.col("left_censored")),347 ]348 remaining = L349 attrition = [("S0 all listings in snapshot", len(L))]350 for name, cond in steps:351 dropped = remaining.filter(cond)352 for uid in dropped["uid"].to_list():353 excl.append((uid, name, ""))354 remaining = remaining.filter(~cond)355 attrition.append((f"− {name}", len(remaining)))356 with open(out / "exclusions.csv", "w", newline="") as f:357 w = csv.writer(f); w.writerow(["uid", "rule", "value"]); w.writerows(excl)358 with open(out / "sample_construction.csv", "w", newline="") as f:359 w = csv.writer(f); w.writerow(["step", "listings_remaining"]); w.writerows(attrition)360 stats = {"S0": len(L), "S1_all_sources": int(L["s1_all_sources"].sum()), "S1_hedonic": int(L["s1_hedonic"].sum()),361 "S2_dynamic": int(L["s2_dynamic"].sum()), "S2L": int(L["s2l_dynamic_with_stock"].sum()),362 "exit_confirmed_in_S2": int(L.filter(pl.col("s2_dynamic"))["exit_confirmed"].sum()),363 "region_recovered": int(L["region_recovered"].sum()), "region_missing": int(L["region"].is_null().sum()),364 "body_recovered": int(L["body_recovered"].sum()), "body_missing_after": int(L["body_filled"].is_null().sum()),365 "oscillating": int(L["oscillating"].sum()), "n_dealer_ids": L["dealer_id"].n_unique(),366 "powertrain_recovered": int(L["powertrain_recovered"].sum()), "powertrain_unknown_after": int((L["powertrain"] == "unknown").sum()),367 "coords_from_city": int(L["coords_from_city"].sum()), "coords_from_dealer": int(L["coords_from_dealer"].sum()), "coords_out_of_bbox": int(L["coords_out_of_bbox"].sum()), "coords_missing_after": int(L["lat"].is_null().sum())}368 # restricted id maps stay in interim369 pl.DataFrame({"dealer_key": list(dealer_keys.keys()), "dealer_id": list(dealer_keys.values())}).write_csv(interim / "id_maps" / "dealers.csv")370 L.select(["uid", "listing_id"]).write_csv(interim / "id_maps" / "listings.csv")371 L.write_parquet(interim / "listings_full.parquet") # includes exact lat/lng, uid372 L.drop(["lat", "lng", "uid", "city", "title"]).write_parquet(out / "listings.parquet") # release-grade373 events = events.join(L.select(["uid", "listing_id"]), on="uid", how="left").drop("uid")374 events.write_parquet(out / "price_events.parquet")375 (out / "cleaning_stats.json").write_text(json.dumps(stats, indent=2))376 return stats377378379if __name__ == "__main__":380 print(json.dumps(run(), indent=2))381