1"""Phase 1.3 — reconstruct the product × day (and × week) price panel from the event log.23Data-generating process (Food-Ka db.py, read-only copy): price_log holds one row at a product's first observation and4one row at every change of the displayed price (NULL = no displayed price). Between two rows the product kept the5recorded price and was re-observed at every connector pass (median 7.2 h). A product's observation span is6[first_seen, last_seen]; after two consecutive misses it is deactivated (active=0) — gaps inside the span are not7logged (limitation documented in reports/phase1_data_audit.md).89Outputs (data/processed):10 offers_daily.parquet — uid × local date: duration-weighted price statistics11 offers_weekly.parquet — uid × week (Thu→Wed 'flyer week'): same statistics12 products.parquet — canonical product table with re-parsed sizes, unit prices, flags13 snapshots.parquet — connector passes (from sync_log)14"""15from __future__ import annotations16import json17import duckdb, pandas as pd, numpy as np18from src.common import path, get_logger, ledger, write_manifest, TZ, SETTINGS19from src.cleaning.units import parse_size, parse_price_label, unit_price2021log = get_logger("panel")2223def canonical_products(prod: pd.DataFrame) -> pd.DataFrame:24 out = prod.copy()25 sizes = [parse_size(sl, nm) for sl, nm in zip(out["size_label"], out["name"])]26 labels = [parse_price_label(pl) for pl in out["price_label"]]27 out["size_qty"] = [s.qty if s else np.nan for s in sizes]28 out["size_base"] = [s.base if s else None for s in sizes]29 out["size_approx"] = [bool(s.approx) if s else False for s in sizes]30 out["size_source"] = [s.source if s else None for s in sizes]31 out["size_pack"] = [s.pack if s else np.nan for s in sizes]32 out["per_weight_label"] = [l.per_weight_unit for l in labels]33 out["multibuy_n"] = [l.multibuy_n for l in labels]34 up = [unit_price(p, s, l) for p, s, l in zip(out["price"], sizes, labels)]35 out["unit_price_rp"] = [u[0] if u else np.nan for u in up]36 out["unit_base_rp"] = [u[1] if u else None for u in up]37 det = out["details"].apply(lambda s: json.loads(s) if isinstance(s, str) and s else {})38 out["flyer_id"] = det.apply(lambda d: d.get("flyer_id"))39 out["valid_from"] = det.apply(lambda d: d.get("valid_from"))40 out["valid_to"] = det.apply(lambda d: d.get("valid_to"))41 out["fine_print"] = det.apply(lambda d: d.get("fine_print"))42 out["off_barcode"] = det.apply(lambda d: (d.get("off") or {}).get("code") if isinstance(d.get("off"), dict) else None)43 fp = out["fine_print"].fillna("").str.lower()44 out["flag_weighted"] = out["per_weight_label"].notna() | out["size_approx"]45 out["flag_multibuy"] = out["multibuy_n"].notna()46 out["flag_loyalty"] = fp.str.contains("pc optimum|moi|scène|scene|points|membre|member", regex=True)47 out["flag_min_qty"] = fp.str.contains("limite|limit|minimum|achat de|à l'achat", regex=True)48 out["flag_deposit_tax"] = fp.str.contains(r"\+\s*tx|taxes|consigne|dépôt|deposit", regex=True)49 out["flag_online_only"] = out["source"].isin(["epipresto", "bocoboco", "aliments_merci", "maturin", "costco"])50 out["flag_price_missing"] = out["price"].isna()51 out["flag_unit_missing"] = out["unit_price_rp"].isna()52 out["flag_reg_inconsistent"] = out["regular_price"].notna() & (out["regular_price"] <= out["price"])53 out["flag_unrealistic"] = out["price"].notna() & ((out["price"] < 0.05) | (out["price"] > 1500) |54 (out["unit_price_rp"].notna() & (out["unit_base_rp"] == "100 g") & (out["unit_price_rp"] > 50)))55 out["gtin_valid"] = out["external_id"].apply(_gs1_ok)56 out["first_seen_local"] = pd.to_datetime(out["first_seen"], unit="s", utc=True).dt.tz_convert(TZ)57 out["last_seen_local"] = pd.to_datetime(out["last_seen"], unit="s", utc=True).dt.tz_convert(TZ)58 out["updated_local"] = pd.to_datetime(out["updated_at"], unit="s", utc=True).dt.tz_convert(TZ)59 out["span_days"] = (out["last_seen"] - out["first_seen"]) / 86400.060 return out.drop(columns=["details", "images", "description"])6162def _gs1_ok(code: str) -> bool:63 if not isinstance(code, str) or not code.isdigit() or len(code) not in (8, 12, 13, 14):64 return False65 d = [int(c) for c in code]; body, chk = d[:-1], d[-1]66 s = sum(x * (3 if (len(body) - i) % 2 == 1 else 1) for i, x in enumerate(body))67 return (10 - s % 10) % 10 == chk6869def build(con: duckdb.DuckDBPyConnection) -> None:70 con.execute(f"SET TimeZone='{TZ}'")71 con.execute("""72 CREATE OR REPLACE TABLE ev AS73 SELECT pl.uid, p.source, to_timestamp(pl.ts) AS ts, pl.price,74 to_timestamp(p.first_seen) AS first_seen, to_timestamp(p.last_seen) AS last_seen75 FROM raw_price_log pl JOIN raw_products p USING (uid)""")76 # segments: a price holds from its event until the next event, or until last_seen77 con.execute("""78 CREATE OR REPLACE TABLE seg AS79 SELECT uid, source, price, ts AS seg_start,80 COALESCE(LEAD(ts) OVER (PARTITION BY uid ORDER BY ts), last_seen) AS seg_end81 FROM ev""")82 con.execute("DELETE FROM seg WHERE seg_end <= seg_start")83 # expand to local days84 con.execute("""85 CREATE OR REPLACE TABLE segday AS86 WITH s AS (87 SELECT *, CAST(seg_start AT TIME ZONE 'UTC' AS TIMESTAMP) AS a, CAST(seg_end AT TIME ZONE 'UTC' AS TIMESTAMP) AS b FROM seg88 ), s2 AS (89 SELECT uid, source, price, seg_start, seg_end,90 (seg_start AT TIME ZONE '%(tz)s')::DATE AS d0, (seg_end AT TIME ZONE '%(tz)s')::DATE AS d1 FROM seg91 ), g AS (92 SELECT uid, source, price, seg_start, seg_end, UNNEST(generate_series(d0, d1, INTERVAL 1 DAY))::DATE AS day FROM s293 )94 SELECT uid, source, price, day,95 GREATEST(seg_start, (day::TIMESTAMP) AT TIME ZONE '%(tz)s') AS a,96 LEAST(seg_end, ((day + INTERVAL 1 DAY)::TIMESTAMP) AT TIME ZONE '%(tz)s') AS b97 FROM g""" % {"tz": TZ})98 con.execute("DELETE FROM segday WHERE b <= a")99 con.execute("ALTER TABLE segday ADD COLUMN hours DOUBLE")100 con.execute("UPDATE segday SET hours = epoch(b - a) / 3600.0")101 _aggregate(con, "segday", "day", "offers_daily")102 # flyer week: Thursday -> Wednesday, keyed by the Thursday103 con.execute("""CREATE OR REPLACE TABLE segweek AS104 SELECT *, day - CAST((dayofweek(day) + 3) % 7 AS INTEGER) AS week FROM segday""") # dayofweek: Sun=0 ... Thu=4 -> offset 0105 _aggregate(con, "segweek", "week", "offers_weekly")106107def _aggregate(con, table: str, key: str, out: str) -> None:108 con.execute(f"""109 CREATE OR REPLACE TABLE {out} AS110 WITH byprice AS (111 SELECT uid, source, {key}, price, SUM(hours) AS h, MIN(a) AS first_a112 FROM {table} GROUP BY uid, source, {key}, price113 ), tot AS (114 SELECT uid, {key}, SUM(h) AS hours_obs, SUM(CASE WHEN price IS NULL THEN h ELSE 0 END) AS hours_null,115 COUNT(*) FILTER (WHERE price IS NOT NULL) AS n_distinct116 FROM byprice GROUP BY uid, {key}117 ), ranked AS (118 SELECT *, ROW_NUMBER() OVER (PARTITION BY uid, {key} ORDER BY h DESC, price ASC) AS rk119 FROM byprice WHERE price IS NOT NULL120 ), modep AS (SELECT uid, {key}, price AS p_mode, h AS mode_hours FROM ranked WHERE rk = 1),121 stats AS (122 SELECT uid, source, {key}, MIN(price) AS p_min, MAX(price) AS p_max,123 SUM(price * hours) / NULLIF(SUM(hours), 0) AS p_twmean,124 ARG_MAX(price, b) AS p_close, ARG_MIN(price, a) AS p_open,125 COUNT(*) - 1 AS n_changes126 FROM {table} WHERE price IS NOT NULL GROUP BY uid, source, {key}127 ), med AS (128 SELECT uid, {key}, QUANTILE_CONT(price, 0.5) AS p_median_events FROM {table} WHERE price IS NOT NULL GROUP BY uid, {key}129 )130 SELECT s.uid, s.source, s.{key}, s.p_open, s.p_close, s.p_min, s.p_max, s.p_twmean, m.p_mode, m.mode_hours,131 md.p_median_events, t.n_distinct, s.n_changes, t.hours_obs, t.hours_null,132 m.mode_hours / NULLIF(t.hours_obs - t.hours_null, 0) AS mode_share,133 (s.p_max - s.p_min) / NULLIF(m.p_mode, 0) AS rel_range134 FROM stats s JOIN modep m USING (uid, {key}) JOIN tot t USING (uid, {key}) JOIN med md USING (uid, {key})""")135 con.execute(f"COPY {out} TO '{path('processed') / (out + '.parquet')}' (FORMAT PARQUET)")136 n, = con.execute(f"SELECT COUNT(*) FROM {out}").fetchone()137 log.info("%s: %d rows", out, n)138139def main() -> None:140 interim = path("interim")141 con = duckdb.connect()142 for t in ["products", "price_log", "sync_log"]:143 con.execute(f"CREATE VIEW raw_{t} AS SELECT * FROM read_parquet('{interim / ('raw_' + t + '.parquet')}')")144 build(con)145 prod = canonical_products(pd.read_parquet(interim / "raw_products.parquet"))146 prod.to_parquet(path("processed") / "products.parquet", index=False)147 log.info("products canonical: %d rows; unit price available for %.1f%% (Food-Ka: %.1f%%)", len(prod),148 100 * prod["unit_price_rp"].notna().mean(), 100 * prod["unit_price"].notna().mean())149 sl = pd.read_parquet(interim / "raw_sync_log.parquet")150 sl["ts_utc"] = pd.to_datetime(sl["ts"], unit="s", utc=True)151 st = sl["stats"].apply(lambda s: json.loads(s) if isinstance(s, str) and s else {})152 sl["null_price_rate"] = st.apply(lambda d: d.get("null_price_rate")); sl["missed"] = st.apply(lambda d: d.get("missed"))153 sl["pipeline_commit"] = SETTINGS["snapshot"]["foodka_commit"][:7]154 sl.drop(columns=["stats"]).to_parquet(path("processed") / "snapshots.parquet", index=False)155 files = [path("processed") / f for f in ["offers_daily.parquet", "offers_weekly.parquet", "products.parquet", "snapshots.parquet"]]156 write_manifest("panel", files)157 ledger("1", "build_panel", "offers_daily/weekly, products, snapshots written")158159if __name__ == "__main__":160 main()161