1"""Phase 4.0 — analysis-ready weekly panels (preregistration §1).23Outputs (data/processed):4 panel_weekly.parquet — grocery-universe listing × complete circular week, posted price = weekly mode, unit prices,5 banner/company/channel/category, size, flags6 panel_matched_<TIERS>.parquet — panel_weekly restricted to listings in cross-banner match groups (A, AB, ABC, ABCD),7 with group_id and `ambiguous` flag (group has >1 simultaneously posted listing from the8 same banner-channel in that week)9"""10import numpy as np, pandas as pd11from src.common import path, get_logger, ledger, SETTINGS, write_manifest12log = get_logger("panel4")13P = path("processed")1415def main():16 wk = pd.read_parquet(P / "offers_weekly.parquet")17 prod = pd.read_parquet(P / "products.parquet")18 ret = pd.read_parquet(P / "retailers.parquet")19 prod = prod.merge(ret[["feed_id", "banner", "company", "channel", "store_format", "is_grocery"]], left_on="source", right_on="feed_id", how="left")20 sc = SETTINGS["scope"]21 uni = prod[(~prod.source.isin(sc["exclude_sources"])) & prod.is_grocery & (~prod.category.isin(sc["non_grocery_categories"])) & (~prod.flag_unrealistic)]22 log.info("grocery universe listings: %d (of %d)", len(uni), len(prod))23 weeks = sorted(wk.week.unique())24 complete = [w for w in weeks if w >= pd.Timestamp("2026-08-13").date() and w <= pd.Timestamp("2026-10-22").date()]25 pw = wk[wk.week.isin(complete) & wk.uid.isin(uni.uid)].copy()26 cols = ["uid", "source", "banner", "company", "channel", "store_format", "category", "name", "brand", "size_label", "size_qty", "size_base", "size_pack",27 "per_weight_label", "flag_weighted", "flag_multibuy", "flag_loyalty", "flag_online_only", "gtin_valid", "active", "on_sale", "regular_price", "price"]28 pw = pw.merge(uni[cols].rename(columns={"source": "source_p", "price": "price_now", "on_sale": "on_sale_now", "regular_price": "regular_now"}), on="uid", how="left").drop(columns=["source_p"])29 pw["week"] = pd.to_datetime(pw.week); pw["week_idx"] = pw.week.rank(method="dense").astype(int)30 pw["log_p"] = np.log(pw.p_mode)31 qty = pw.size_qty.where(pw.size_base != "un", pw.size_qty)32 pw["unit_p"] = np.where(pw.size_base == "un", pw.p_mode / pw.size_qty, pw.p_mode / pw.size_qty * 100.0)33 pw.loc[pw.per_weight_label.eq("kg"), "unit_p"] = pw.p_mode / 10.034 pw.loc[pw.per_weight_label.eq("lb"), "unit_p"] = pw.p_mode / 4.5359235 pw["unit_base"] = np.where(pw.size_base == "un", "un", np.where(pw.size_base == "ml", "100 ml", "100 g"))36 pw.loc[pw.per_weight_label.isin(["kg", "lb"]), "unit_base"] = "100 g"37 pw["log_unit_p"] = np.log(pw.unit_p)38 pw["multi_price_week"] = pw.n_distinct > 139 pw.to_parquet(P / "panel_weekly.parquet", index=False)40 log.info("panel_weekly: %d listing-weeks, %d listings, %d weeks (%s → %s), %d banners", len(pw), pw.uid.nunique(), pw.week.nunique(), pw.week.min().date(), pw.week.max().date(), pw.banner.nunique())41 files = [P / "panel_weekly.parquet"]42 for tiers in ["A", "AB", "ABC", "ABCD"]:43 g = pd.read_parquet(P / f"match_groups_{tiers}.parquet")[["group_id", "uid"]]44 pm = pw.merge(g, on="uid", how="inner")45 # keep group-weeks with >= 2 banners posting46 nb = pm.groupby(["group_id", "week"]).banner.transform("nunique"); pm = pm[nb >= 2].copy()47 dup = pm.groupby(["group_id", "week", "banner", "channel"]).uid.transform("size") > 148 amb = pm.assign(d=dup).groupby(["group_id", "week"]).d.transform("max"); pm["ambiguous"] = amb.astype(bool)49 pm["same_company_group"] = pm.groupby(["group_id", "week"]).company.transform("nunique") == 150 pm.to_parquet(P / f"panel_matched_{tiers}.parquet", index=False); files.append(P / f"panel_matched_{tiers}.parquet")51 log.info("panel_matched_%-4s: %d listing-weeks, %d groups, %d group-weeks, ambiguous group-weeks %.1f%%, same-company-only group-weeks %.1f%%",52 tiers, len(pm), pm.group_id.nunique(), pm.groupby(["group_id", "week"]).ngroups,53 100 * pm.drop_duplicates(["group_id", "week"]).ambiguous.mean(), 100 * pm.drop_duplicates(["group_id", "week"]).same_company_group.mean())54 write_manifest("panel4", files)55 ledger("4", "build_matched_panel", f"{len(pw)} listing-weeks; weeks {pw.week.min().date()}..{pw.week.max().date()}")5657if __name__ == "__main__":58 main()59