"""Phase 4.0 — analysis-ready weekly panels (preregistration §1).

Outputs (data/processed):
  panel_weekly.parquet            — grocery-universe listing × complete circular week, posted price = weekly mode, unit prices,
                                    banner/company/channel/category, size, flags
  panel_matched_<TIERS>.parquet   — panel_weekly restricted to listings in cross-banner match groups (A, AB, ABC, ABCD),
                                    with group_id and `ambiguous` flag (group has >1 simultaneously posted listing from the
                                    same banner-channel in that week)
"""
import numpy as np, pandas as pd
from src.common import path, get_logger, ledger, SETTINGS, write_manifest
log = get_logger("panel4")
P = path("processed")

def main():
    wk = pd.read_parquet(P / "offers_weekly.parquet")
    prod = pd.read_parquet(P / "products.parquet")
    ret = pd.read_parquet(P / "retailers.parquet")
    prod = prod.merge(ret[["feed_id", "banner", "company", "channel", "store_format", "is_grocery"]], left_on="source", right_on="feed_id", how="left")
    sc = SETTINGS["scope"]
    uni = prod[(~prod.source.isin(sc["exclude_sources"])) & prod.is_grocery & (~prod.category.isin(sc["non_grocery_categories"])) & (~prod.flag_unrealistic)]
    log.info("grocery universe listings: %d (of %d)", len(uni), len(prod))
    weeks = sorted(wk.week.unique())
    complete = [w for w in weeks if w >= pd.Timestamp("2026-08-13").date() and w <= pd.Timestamp("2026-10-22").date()]
    pw = wk[wk.week.isin(complete) & wk.uid.isin(uni.uid)].copy()
    cols = ["uid", "source", "banner", "company", "channel", "store_format", "category", "name", "brand", "size_label", "size_qty", "size_base", "size_pack",
            "per_weight_label", "flag_weighted", "flag_multibuy", "flag_loyalty", "flag_online_only", "gtin_valid", "active", "on_sale", "regular_price", "price"]
    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"])
    pw["week"] = pd.to_datetime(pw.week); pw["week_idx"] = pw.week.rank(method="dense").astype(int)
    pw["log_p"] = np.log(pw.p_mode)
    qty = pw.size_qty.where(pw.size_base != "un", pw.size_qty)
    pw["unit_p"] = np.where(pw.size_base == "un", pw.p_mode / pw.size_qty, pw.p_mode / pw.size_qty * 100.0)
    pw.loc[pw.per_weight_label.eq("kg"), "unit_p"] = pw.p_mode / 10.0
    pw.loc[pw.per_weight_label.eq("lb"), "unit_p"] = pw.p_mode / 4.53592
    pw["unit_base"] = np.where(pw.size_base == "un", "un", np.where(pw.size_base == "ml", "100 ml", "100 g"))
    pw.loc[pw.per_weight_label.isin(["kg", "lb"]), "unit_base"] = "100 g"
    pw["log_unit_p"] = np.log(pw.unit_p)
    pw["multi_price_week"] = pw.n_distinct > 1
    pw.to_parquet(P / "panel_weekly.parquet", index=False)
    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())
    files = [P / "panel_weekly.parquet"]
    for tiers in ["A", "AB", "ABC", "ABCD"]:
        g = pd.read_parquet(P / f"match_groups_{tiers}.parquet")[["group_id", "uid"]]
        pm = pw.merge(g, on="uid", how="inner")
        # keep group-weeks with >= 2 banners posting
        nb = pm.groupby(["group_id", "week"]).banner.transform("nunique"); pm = pm[nb >= 2].copy()
        dup = pm.groupby(["group_id", "week", "banner", "channel"]).uid.transform("size") > 1
        amb = pm.assign(d=dup).groupby(["group_id", "week"]).d.transform("max"); pm["ambiguous"] = amb.astype(bool)
        pm["same_company_group"] = pm.groupby(["group_id", "week"]).company.transform("nunique") == 1
        pm.to_parquet(P / f"panel_matched_{tiers}.parquet", index=False); files.append(P / f"panel_matched_{tiers}.parquet")
        log.info("panel_matched_%-4s: %d listing-weeks, %d groups, %d group-weeks, ambiguous group-weeks %.1f%%, same-company-only group-weeks %.1f%%",
                 tiers, len(pm), pm.group_id.nunique(), pm.groupby(["group_id", "week"]).ngroups,
                 100 * pm.drop_duplicates(["group_id", "week"]).ambiguous.mean(), 100 * pm.drop_duplicates(["group_id", "week"]).same_company_group.mean())
    write_manifest("panel4", files)
    ledger("4", "build_matched_panel", f"{len(pw)} listing-weeks; weeks {pw.week.min().date()}..{pw.week.max().date()}")

if __name__ == "__main__":
    main()