1"""Phase 9 — independent recomputation of central statistics with different tools/code paths.23Reads only the raw parquet tables (interim) and the match groups, NOT the model outputs, and recomputes:4 (1) weekly modal price per listing via DuckDB SQL from the raw event log (independent of build_panel.py)5 (2) H1 median cross-banner gap (tiers A+B, storefront, non-ambiguous) via DuckDB6 (3) AR(1) of banner-pair gaps via statsmodels OLS (vs pyfixest in Model A)7 (4) banner fixed effects for Super C and Maxi via statsmodels OLS with dummies on the Metro Inc. and Loblaw sub-samples8 (5) promotion share and volatility ratio via an independent pandas implementation9 (6) chained Jevons via DuckDB10Writes reports/audit_recompute.json with both values side by side.11"""12import json, numpy as np, pandas as pd, duckdb, statsmodels.api as sm13from src.common import ROOT, path, get_logger, TZ14log = get_logger("audit"); P = path("processed"); I = path("interim"); E = ROOT / "experiments"1516def main():17 con = duckdb.connect(); con.execute(f"SET TimeZone='{TZ}'")18 con.execute(f"CREATE VIEW pl AS SELECT * FROM read_parquet('{I / 'raw_price_log.parquet'}')")19 con.execute(f"CREATE VIEW pr AS SELECT * FROM read_parquet('{I / 'raw_products.parquet'}')")20 con.execute(f"CREATE VIEW g AS SELECT * FROM read_parquet('{P / 'match_groups_AB.parquet'}')")21 con.execute(f"CREATE VIEW ret AS SELECT * FROM read_parquet('{P / 'retailers.parquet'}')")22 # (1) independent weekly mode: segment durations clipped to flyer weeks (Thu->Wed)23 con.execute(f"""24 CREATE TABLE seg AS25 SELECT pl.uid, pl.price, to_timestamp(pl.ts) AS a, COALESCE(LEAD(to_timestamp(pl.ts)) OVER (PARTITION BY pl.uid ORDER BY pl.ts), to_timestamp(p.last_seen)) AS b26 FROM pl JOIN pr p ON p.uid = pl.uid WHERE pl.price IS NOT NULL""")27 con.execute("DELETE FROM seg WHERE b <= a")28 con.execute(f"""29 CREATE TABLE segw AS30 WITH d AS (SELECT uid, price, a, b, (a AT TIME ZONE '{TZ}')::DATE AS da, (b AT TIME ZONE '{TZ}')::DATE AS db FROM seg),31 x AS (SELECT uid, price, a, b, UNNEST(generate_series(da - CAST((dayofweek(da)+3)%7 AS INTEGER), db, INTERVAL 7 DAY))::DATE AS wk FROM d)32 SELECT uid, price, wk, GREATEST(a, (wk::TIMESTAMP) AT TIME ZONE '{TZ}') AS aa, LEAST(b, ((wk + INTERVAL 7 DAY)::TIMESTAMP) AT TIME ZONE '{TZ}') AS bb FROM x""")33 con.execute("DELETE FROM segw WHERE bb <= aa")34 con.execute("""CREATE TABLE wmode AS35 SELECT uid, wk, price AS p_mode FROM (36 SELECT uid, wk, price, SUM(epoch(bb-aa)) AS h, ROW_NUMBER() OVER (PARTITION BY uid, wk ORDER BY SUM(epoch(bb-aa)) DESC, price ASC) AS rk37 FROM segw GROUP BY uid, wk, price) WHERE rk = 1""")38 ref = pd.read_parquet(P / "offers_weekly.parquet", columns=["uid", "week", "p_mode"]); ref["week"] = pd.to_datetime(ref.week)39 mine = con.execute("SELECT uid, wk AS week, p_mode AS p_mode_sql FROM wmode").df(); mine["week"] = pd.to_datetime(mine.week)40 cmp = ref.merge(mine, on=["uid", "week"], how="inner")41 out = {"weekly_mode_agreement": {"rows_compared": int(len(cmp)), "share_equal": float((np.isclose(cmp.p_mode, cmp.p_mode_sql)).mean())}}42 # (2) H1 median gap, storefront, AB, non-ambiguous, >=2 banners43 con.execute("CREATE VIEW wm AS SELECT w.uid, w.wk, w.p_mode, p.source, p.category, r.banner, r.company, r.channel FROM wmode w JOIN pr p USING (uid) JOIN ret r ON r.feed_id = p.source")44 gaps = con.execute("""45 WITH m AS (SELECT g.group_id, wm.* FROM wm JOIN g USING (uid) WHERE channel='online_storefront' AND wk BETWEEN DATE '2026-08-13' AND DATE '2026-10-22'46 AND source NOT IN ('saq') AND category NOT IN ('Santé et beauté','Maison et entretien','Animaux','Bébé','Alcool')),47 amb AS (SELECT group_id, wk, MAX(c) AS mx FROM (SELECT group_id, wk, banner, COUNT(*) c FROM m GROUP BY 1,2,3) GROUP BY 1,2),48 ok AS (SELECT m.* FROM m JOIN amb USING (group_id, wk) WHERE mx = 1)49 SELECT group_id, wk, MAX(LN(p_mode)) - MIN(LN(p_mode)) AS gap, COUNT(DISTINCT banner) nb, COUNT(DISTINCT company) nc FROM ok GROUP BY 1,2 HAVING nb >= 2""").df()50 A = json.load(open(E / "modelA_summary.json"))51 out["H1_median_gap"] = {"independent_sql": float(gaps.gap.median()), "modelA": A["gap_all"]["median"], "group_weeks_independent": int(len(gaps)), "group_weeks_modelA": A["gap_all"]["n"]}52 out["H1_cross_company_median"] = {"independent": float(gaps[gaps.nc >= 2].gap.median()), "modelA": A["gap_cross_company"]["median"]}53 # (3) AR(1) via statsmodels on pair gaps built here54 pairs = con.execute("""55 WITH m AS (SELECT g.group_id, wm.* FROM wm JOIN g USING (uid) WHERE channel='online_storefront' AND wk BETWEEN DATE '2026-08-13' AND DATE '2026-10-22'56 AND source NOT IN ('saq') AND category NOT IN ('Santé et beauté','Maison et entretien','Animaux','Bébé','Alcool')),57 amb AS (SELECT group_id, wk, MAX(c) AS mx FROM (SELECT group_id, wk, banner, COUNT(*) c FROM m GROUP BY 1,2,3) GROUP BY 1,2),58 ok AS (SELECT m.* FROM m JOIN amb USING (group_id, wk) WHERE mx = 1)59 SELECT a.group_id, a.wk, a.banner AS br, b.banner AS bs, LN(a.p_mode) - LN(b.p_mode) AS gap60 FROM ok a JOIN ok b ON a.group_id=b.group_id AND a.wk=b.wk AND a.banner < b.banner""").df()61 pairs = pairs.sort_values(["group_id", "br", "bs", "wk"]); pairs["lag"] = pairs.groupby(["group_id", "br", "bs"]).gap.shift(1)62 pairs["dw"] = (pd.to_datetime(pairs.wk) - pd.to_datetime(pairs.groupby(["group_id", "br", "bs"]).wk.shift(1))).dt.days63 d = pairs[(pairs.dw == 7)].dropna(subset=["lag"])64 ols = sm.OLS(d.gap.values, sm.add_constant(d.lag.values)).fit(cov_type="cluster", cov_kwds={"groups": pd.factorize(d.group_id)[0]})65 out["AR1"] = {"independent_statsmodels": float(ols.params[1]), "se": float(ols.bse[1]), "modelA": A["ar1_rho"], "modelA_se": A["ar1_se"]}66 # (4) banner effects via OLS with group-week dummies on sub-samples (within-transform by group-week)67 fe = pd.read_csv(P / "modelA_banner_effects.csv").set_index("banner")68 sub = con.execute("""69 WITH m AS (SELECT g.group_id, wm.* FROM wm JOIN g USING (uid) WHERE channel='online_storefront' AND wk BETWEEN DATE '2026-08-13' AND DATE '2026-10-22'70 AND source NOT IN ('saq') AND category NOT IN ('Santé et beauté','Maison et entretien','Animaux','Bébé','Alcool')),71 amb AS (SELECT group_id, wk, MAX(c) AS mx FROM (SELECT group_id, wk, banner, COUNT(*) c FROM m GROUP BY 1,2,3) GROUP BY 1,2),72 ok AS (SELECT m.* FROM m JOIN amb USING (group_id, wk) WHERE mx = 1), nb AS (SELECT group_id, wk FROM ok GROUP BY 1,2 HAVING COUNT(DISTINCT banner) >= 2)73 SELECT ok.group_id, ok.wk, ok.banner, LN(ok.p_mode) AS lp FROM ok JOIN nb USING (group_id, wk)""").df()74 res = {}75 for pair in [("Metro", "Super C"), ("Metro", "Maxi"), ("Metro", "Club Entrepôt"), ("Metro", "Provigo")]:76 s2 = sub[sub.banner.isin(pair)]; s2 = s2[s2.groupby(["group_id", "wk"]).banner.transform("nunique") == 2]77 s2["lp_dm"] = s2.lp - s2.groupby(["group_id", "wk"]).lp.transform("mean"); x = (s2.banner == pair[1]).astype(float); x = x - x.groupby([s2.group_id, s2.wk]).transform("mean")78 o = sm.OLS(s2.lp_dm.values, x.values).fit(cov_type="cluster", cov_kwds={"groups": pd.factorize(s2.group_id)[0]})79 res[pair[1]] = {"independent_pairwise_within": float(o.params[0]), "se": float(o.bse[0]), "modelA_full_FE": float(fe.loc[pair[1], "coef"]) if pair[1] in fe.index else None, "n": int(len(s2))}80 out["banner_effects_vs_Metro"] = res81 # (5) promotion share & volatility ratio (independent pandas)82 w = mine.merge(pd.read_parquet(P / "products.parquet", columns=["uid", "source", "category", "flag_unrealistic"]), on="uid").merge(pd.read_parquet(P / "retailers.parquet")[["feed_id", "channel", "is_grocery"]], left_on="source", right_on="feed_id")83 w = w[(w.channel == "online_storefront") & w.is_grocery & (~w.flag_unrealistic) & (~w.category.isin(["Santé et beauté", "Maison et entretien", "Animaux", "Bébé", "Alcool"])) & (w.week >= "2026-08-13") & (w.week <= "2026-10-22")]84 w = w[w.groupby("uid").week.transform("nunique") >= 8].sort_values(["uid", "week"])85 def roll(s):86 v = s.tolist(); o = []87 for i in range(len(v)):88 vc = pd.Series(v[max(0, i - 7): i + 1]).value_counts(); o.append(max(vc[vc == vc.max()].index))89 return pd.Series(o, index=s.index)90 w["ref"] = w.groupby("uid").p_mode_sql.transform(roll); w["promo"] = w.p_mode_sql <= 0.9 * w.ref91 w["dlp"] = np.log(w.p_mode_sql).groupby(w.uid).diff(); w["dlr"] = np.log(w.ref).groupby(w.uid).diff()92 B = json.load(open(E / "modelB_summary.json"))93 out["H2"] = {"promo_share_independent": float(w.promo.mean()), "modelB": B["promo_share_listing_weeks"], "ratio_independent": float(w.dlp.std() / w.dlr.std()), "modelB_ratio": B["ratio_all"], "listing_weeks": int(len(w))}94 # (6) chained Jevons via SQL on wmode (storefront grocery universe)95 u = w[["uid", "week", "p_mode_sql"]].copy(); piv = u.pivot_table(index="uid", columns="week", values="p_mode_sql"); weeks = sorted(piv.columns); lj = 0.096 for a, b in zip(weeks, weeks[1:]):97 both = piv[[a, b]].dropna(); lj += (np.log(both[b]) - np.log(both[a])).mean()98 D = json.load(open(E / "modelD_summary.json"))99 out["D_jevons_chained"] = {"independent_on_ge8week_sample": float(np.exp(lj) - 1), "modelD_full_storefront": D["storefront_posted"]["jevons_chained_change"], "note": "samples differ (>=8 weeks vs all); orders of magnitude should agree"}100 (ROOT / "reports" / "audit_recompute.json").write_text(json.dumps(out, indent=2, default=float))101 log.info("%s", json.dumps(out, indent=1, default=float))102103if __name__ == "__main__":104 main()105