1"""Phase 1.5 — coverage and attrition tables (inputs of Table 1 / Figure 1).23Outputs (data/processed/coverage_*.csv): by feed (listings, active, days observed, events, unit-price coverage,4flags), by category, by week (listings observed, feeds reporting), and the attrition ledger.5"""6import pandas as pd, numpy as np7from src.common import path, get_logger, ledger, SETTINGS89log = get_logger("coverage")1011def main():12 P = path("processed")13 prod = pd.read_parquet(P / "products.parquet"); ret = pd.read_parquet(P / "retailers.parquet")14 wk = pd.read_parquet(P / "offers_weekly.parquet"); day = pd.read_parquet(P / "offers_daily.parquet", columns=["uid", "day", "source"])15 prod = prod.merge(ret[["feed_id", "banner", "company", "channel", "store_format", "is_grocery"]], left_on="source", right_on="feed_id", how="left")16 ev = pd.read_parquet(path("interim") / "raw_price_log.parquet").groupby("uid").size().rename("n_events")17 prod = prod.merge(ev, left_on="uid", right_index=True, how="left")18 dpu = day.groupby("uid").size().rename("days_obs"); prod = prod.merge(dpu, left_on="uid", right_index=True, how="left")19 by_feed = prod.groupby(["company", "banner", "source", "channel", "store_format"]).agg(20 listings=("uid", "size"), active=("active", "sum"), on_sale_now=("on_sale", "sum"),21 median_days_obs=("days_obs", "median"), price_events=("n_events", "sum"),22 unit_price_cov=("unit_price_rp", lambda s: s.notna().mean()), brand_cov=("brand", lambda s: (s.fillna("") != "").mean()),23 gtin=("gtin_valid", "sum"), weighted=("flag_weighted", "sum"), multibuy=("flag_multibuy", "sum"),24 loyalty=("flag_loyalty", "sum"), unrealistic=("flag_unrealistic", "sum")).reset_index().sort_values("listings", ascending=False)25 by_feed.to_csv(P / "coverage_by_feed.csv", index=False)26 by_cat = prod.groupby("category").agg(listings=("uid", "size"), active=("active", "sum"), feeds=("source", "nunique"),27 unit_price_cov=("unit_price_rp", lambda s: s.notna().mean()), price_events=("n_events", "sum")).reset_index().sort_values("listings", ascending=False)28 by_cat.to_csv(P / "coverage_by_category.csv", index=False)29 wk2 = wk.merge(prod[["uid", "channel", "category", "is_grocery"]], on="uid")30 by_week = wk2.groupby("week").agg(listings=("uid", "nunique"), feeds=("source", "nunique"),31 changes=("n_changes", "sum"), online=("channel", lambda s: (s == "online_storefront").sum()),32 circular=("channel", lambda s: s.str.startswith("circular").sum())).reset_index()33 by_week.to_csv(P / "coverage_by_week.csv", index=False)34 # attrition ledger (listing level)35 steps = []36 n0 = len(prod); steps.append(("All listings in snapshot (products table)", n0))37 n = prod[~prod.source.isin(SETTINGS["scope"]["exclude_sources"])]; steps.append(("excl. SAQ (alcohol monopoly)", len(n)))38 n = n[n.is_grocery]; steps.append(("excl. pharmacy feeds", len(n)))39 n = n[~n.category.isin(SETTINGS["scope"]["non_grocery_categories"])]; steps.append(("excl. non-food categories (health & beauty, household, pets, baby, alcohol)", len(n)))40 n = n[n.price.notna() | (n.n_events.fillna(0) > 0)]; steps.append(("with at least one observed price", len(n)))41 n = n[~n.flag_unrealistic]; steps.append(("excl. unrealistic prices", len(n)))42 n1 = n[n.size_qty.notna()]; steps.append(("… with parsable package size (unit-price / matching universe)", len(n1)))43 g = pd.read_parquet(P / "match_groups_ABC.parquet"); steps.append(("… in a cross-banner match group (tiers A–C)", int(n1.uid.isin(g.uid).sum())))44 att = pd.DataFrame(steps, columns=["step", "listings"]); att["share_of_initial"] = att.listings / n045 att.to_csv(P / "attrition_listings.csv", index=False)46 log.info("\n%s", att.to_string(index=False)); log.info("weeks: %d (%s → %s)", len(by_week), by_week.week.min(), by_week.week.max())47 ledger("1", "coverage", f"attrition {n0} -> {att.listings.iloc[-1]} matched listings")4849if __name__ == "__main__":50 main()51