1"""Phase 2.2 — Listing-level cleaning and the sample-flow (attrition) table.23Input : data/interim/listings_raw.parquet, data/interim/stores_raw.parquet4Output: data/interim/listings_clean.parquet (all published-active listings of live stores, with flags; nothing deleted)5 reports/sample_flow.md, experiments/sample_flow.json67Flags (all kept, so every analysis can state its own exclusion set):8 f_live_store store enabled with >= 1 published listing at the snapshot9 f_published active=1 and listing_status='published' (platform's own publication rule)10 f_cad store currency is CAD for >= 95 % of its listings and this listing is CAD11 f_price_ok price > 0, not price_on_request, price <= 50 000 CAD (p99.9 guard, see report)12 f_dup_in_store normalised title duplicated inside the same store (variants listed as separate products); the13 first-seen listing is kept as the representative (f_dup_rep)14 f_hors_mission store_kind == 'hors_mission'15 f_analysis_assort assortment sample = live & published & not f_dup_in_store-non-rep & not hors_mission16 f_analysis_price price sample = f_analysis_assort & f_cad & f_price_ok17"""18from __future__ import annotations19import json, re, unicodedata20import numpy as np, pandas as pd21from src.common import ROOT, P, get_logger, ledger, register_output, write_json, BASE_PROC, BASE_EXP2223log = get_logger("clean.listings")24PRICE_CAP = 50_000.0252627def norm_title(t: str) -> str:28 t = unicodedata.normalize("NFKD", str(t or "")).encode("ascii", "ignore").decode().lower()29 t = re.sub(r"&|&#\d+;", " ", t)30 t = re.sub(r"[^a-z0-9]+", " ", t)31 return re.sub(r"\s+", " ", t).strip()323334def main():35 pr = pd.read_parquet(P["interim"] / "listings_raw.parquet")36 st = pd.read_parquet(P["interim"] / "stores_raw.parquet")37 live_ids = set(st.loc[(st.enabled == 1) & (st.product_count > 0), "id"])38 kind = st.set_index("id")["store_kind"].fillna("")39 flow = [("DB product rows (all statuses, all stores)", len(pr))]4041 pr["f_published"] = (pr.active == 1) & (pr.listing_status == "published")42 flow.append(("active = 1 and listing_status = 'published'", int(pr.f_published.sum())))43 pr["f_live_store"] = pr.store_id.isin(live_ids)44 base = pr[pr.f_published & pr.f_live_store].copy()45 flow.append(("… of an enabled store with >= 1 published listing", len(base)))4647 base["store_kind"] = base.store_id.map(kind).fillna("")48 base["f_hors_mission"] = base.store_kind == "hors_mission"49 flow.append(("… excluding store_kind = hors_mission", int((~base.f_hors_mission).sum())))5051 base["title_norm"] = base.title.map(norm_title)52 base = base.sort_values(["store_id", "title_norm", "first_seen", "uid"])53 grp = base.groupby(["store_id", "title_norm"], sort=False)54 base["dup_group_size"] = grp["uid"].transform("size")55 base["f_dup_in_store"] = base.dup_group_size > 156 base["f_dup_rep"] = ~base.duplicated(["store_id", "title_norm"], keep="first")57 flow.append(("… keeping one representative per (store, normalised title) and non-empty title", int(((~base.f_hors_mission) & base.f_dup_rep & (base.title_norm.str.len() > 0)).sum())))58 sibp = P["interim"] / "sibling_stores.parquet"59 nonrep = set(pd.read_parquet(sibp).query("not is_rep").id) if sibp.exists() else set()60 base["f_sibling_dup"] = base.store_id.isin(nonrep)61 base["f_analysis_assort"] = (~base.f_hors_mission) & base.f_dup_rep & (base.title_norm.str.len() > 0) & (~base.f_sibling_dup)62 flow.append(("… excluding mirror domains of the same business (sibling stores, non-representative) [ASSORTMENT SAMPLE]", int(base.f_analysis_assort.sum())))6364 cur_share = base.groupby("store_id")["currency"].apply(lambda s: (s == "CAD").mean())65 base["store_cad_share"] = base.store_id.map(cur_share)66 base["f_cad"] = (base.currency == "CAD") & (base.store_cad_share >= 0.95)67 base["f_price_ok"] = base.price.notna() & (base.price > 0) & (base.price_on_request == 0) & (base.price <= PRICE_CAP)68 base["f_analysis_price"] = base.f_analysis_assort & base.f_cad & base.f_price_ok69 flow.append(("… and CAD-priced store/listing", int((base.f_analysis_assort & base.f_cad).sum())))70 flow.append(("… and a positive listed price <= 50 000 CAD, not on request [PRICE SAMPLE]", int(base.f_analysis_price.sum())))7172 base["log_price"] = np.where(base.f_price_ok, np.log(base.price.where(base.price > 0)), np.nan)73 base["on_sale"] = base.compare_at_price.notna() & (base.compare_at_price > base.price)74 base["has_variants"] = (base.price_max.notna() & (base.price_max > base.price)) | (base.n_variants.fillna(1) > 1)75 base["available"] = base.available.astype("Int64")76 base.to_parquet(P["interim"] / "listings_clean.parquet", index=False)77 register_output(P["interim"] / "listings_clean.parquet", "clean.listings")7879 # ---- diagnostics for the data-quality report ----------------------------------80 a = base[base.f_analysis_assort]81 diag = {82 "flow": flow,83 "assortment_sample": {"listings": len(a), "stores": int(a.store_id.nunique()), "categories": int(a.category.nunique()),84 "dup_in_store_listings_removed": int((base.f_dup_in_store & ~base.f_dup_rep & ~base.f_hors_mission).sum()),85 "share_autre": round(float((a.category == "autre").mean()), 4),86 "available_0_share": round(float((a.available == 0).mean()), 4),87 "on_sale_share": round(float(a.on_sale.mean()), 4), "has_variants_share": round(float(a.has_variants.mean()), 4),88 "sku_share": round(float(a.has_sku.mean()), 4), "vendor_share": round(float((a.vendor.fillna("") != "").mean()), 4)},89 "price_exclusions": {"non_cad_listing_or_store": int((base.f_analysis_assort & ~base.f_cad).sum()),90 "price_missing_or_on_request": int((base.f_analysis_assort & base.f_cad & (base.price.isna() | (base.price_on_request == 1))).sum()),91 "price_zero": int((base.f_analysis_assort & base.f_cad & (base.price == 0)).sum()),92 "price_gt_cap": int((base.f_analysis_assort & base.f_cad & (base.price > PRICE_CAP)).sum())},93 "price_quantiles_cad": {str(q): round(float(base.loc[base.f_analysis_price, "price"].quantile(q)), 2) for q in (0.001, 0.01, 0.1, 0.5, 0.9, 0.99, 0.999)},94 "top10_stores_share_of_assortment": round(float(a.store_id.value_counts().head(10).sum() / len(a)), 4),95 }96 write_json(diag, BASE_EXP / "sample_flow.json", "clean.listings")97 md = ["# Sample flow (attrition) — WP38", "", f"Snapshot 2026-11-01T17:38:30Z. Nothing is deleted; every step is a flag in `data/interim/listings_clean.parquet`.", "",98 "| step | listings |", "|---|---:|"] + [f"| {k} | {v:,} |".replace(",", " ") for k, v in flow] + ["",99 "## Assortment sample diagnostics", "```", json.dumps(diag["assortment_sample"], indent=1), "```",100 "## Price-sample exclusions (from the assortment sample)", "```", json.dumps(diag["price_exclusions"], indent=1), "```",101 "## Price quantiles in the price sample (CAD)", "```", json.dumps(diag["price_quantiles_cad"], indent=1), "```",102 f"Top-10 stores hold {diag['top10_stores_share_of_assortment']:.1%} of the assortment sample."]103 (ROOT / "reports" / "sample_flow.md").write_text("\n".join(md))104 ledger("clean.listings", **{k: v for k, v in diag.items() if k != "flow"}, flow=flow)105 log.info("\n".join(f"{k}: {v}" for k, v in flow))106107108if __name__ == "__main__":109 main()110