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["per_weight_label"] = [l.per_weight_unit for l in labels]32 out["multibuy_n"] = [l.multibuy_n for l in labels]33 up = [unit_price(p, s, l) for p, s, l in zip(out["price"], sizes, labels)]34 out["unit_price_rp"] = [u[0] if u else np.nan for u in up]35 out["unit_base_rp"] = [u[1] if u else None for u in up]36 det = out["details"].apply(lambda s: json.loads(s) if isinstance(s, str) and s else {})37 out["flyer_id"] = det.apply(lambda d: d.get("flyer_id"))38 out["valid_from"] = det.apply(lambda d: d.get("valid_from"))39 out["valid_to"] = det.apply(lambda d: d.get("valid_to"))40 out["fine_print"] = det.apply(lambda d: d.get("fine_print"))41 out["off_barcode"] = det.apply(lambda d: (d.get("off") or {}).get("code") if isinstance(d.get("off"), dict) else None)42 fp = out["fine_print"].fillna("").str.lower()43 out["flag_weighted"] = out["per_weight_label"].notna() | out["size_approx"]44 out["flag_multibuy"] = out["multibuy_n"].notna()45 out["flag_loyalty"] = fp.str.contains("pc optimum|moi|scène|scene|points|membre|member", regex=True)46 out["flag_min_qty"] = fp.str.contains("limite|limit|minimum|achat de|à l'achat", regex=True)47 out["flag_deposit_tax"] = fp.str.contains(r"\+\s*tx|taxes|consigne|dépôt|deposit", regex=True)48 out["flag_online_only"] = out["source"].isin(["epipresto", "bocoboco", "aliments_merci", "maturin", "costco"])49 out["flag_price_missing"] = out["price"].isna()50 out["flag_unit_missing"] = out["unit_price_rp"].isna()51 out["flag_reg_inconsistent"] = out["regular_price"].notna() & (out["regular_price"] <= out["price"])52 out["flag_unrealistic"] = out["price"].notna() & ((out["price"] < 0.05) | (out["price"] > 1500) |53 (out["unit_price_rp"].notna() & (out["unit_base_rp"] == "100 g") & (out["unit_price_rp"] > 50)))54 out["gtin_valid"] = out["external_id"].apply(_gs1_ok)55 out["first_seen_local"] = pd.to_datetime(out["first_seen"], unit="s", utc=True).dt.tz_convert(TZ)56 out["last_seen_local"] = pd.to_datetime(out["last_seen"], unit="s", utc=True).dt.tz_convert(TZ)57 out["updated_local"] = pd.to_datetime(out["updated_at"], unit="s", utc=True).dt.tz_convert(TZ)58 out["span_days"] = (out["last_seen"] - out["first_seen"]) / 86400.059 return out.drop(columns=["details", "images", "description"])6061def _gs1_ok(code: str) -> bool:62 if not isinstance(code, str) or not code.isdigit() or len(code) not in (8, 12, 13, 14):63 return False64 d = [int(c) for c in code]; body, chk = d[:-1], d[-1]65 s = sum(x * (3 if (len(body) - i) % 2 == 1 else 1) for i, x in enumerate(body))66 return (10 - s % 10) % 10 == chk6768def build(con: duckdb.DuckDBPyConnection) -> None:69 con.execute(f"SET TimeZone='{TZ}'")70 con.execute("""71 CREATE OR REPLACE TABLE ev AS72 SELECT pl.uid, p.source, to_timestamp(pl.ts) AS ts, pl.price,73 to_timestamp(p.first_seen) AS first_seen, to_timestamp(p.last_seen) AS last_seen74 FROM raw_price_log pl JOIN raw_products p USING (uid)""")75 # segments: a price holds from its event until the next event, or until last_seen76 con.execute("""77 CREATE OR REPLACE TABLE seg AS78 SELECT uid, source, price, ts AS seg_start,79 COALESCE(LEAD(ts) OVER (PARTITION BY uid ORDER BY ts), last_seen) AS seg_end80 FROM ev""")81 con.execute("DELETE FROM seg WHERE seg_end <= seg_start")82 # expand to local days83 con.execute("""84 CREATE OR REPLACE TABLE segday AS85 WITH s AS (86 SELECT *, CAST(seg_start AT TIME ZONE 'UTC' AS TIMESTAMP) AS a, CAST(seg_end AT TIME ZONE 'UTC' AS TIMESTAMP) AS b FROM seg87 ), s2 AS (88 SELECT uid, source, price, seg_start, seg_end,89 (seg_start AT TIME ZONE '%(tz)s')::DATE AS d0, (seg_end AT TIME ZONE '%(tz)s')::DATE AS d1 FROM seg90 ), g AS (91 SELECT uid, source, price, seg_start, seg_end, UNNEST(generate_series(d0, d1, INTERVAL 1 DAY))::DATE AS day FROM s292 )93 SELECT uid, source, price, day,94 GREATEST(seg_start, (day::TIMESTAMP) AT TIME ZONE '%(tz)s') AS a,95 LEAST(seg_end, ((day + INTERVAL 1 DAY)::TIMESTAMP) AT TIME ZONE '%(tz)s') AS b96 FROM g""" % {"tz": TZ})97 con.execute("DELETE FROM segday WHERE b <= a")98 con.execute("ALTER TABLE segday ADD COLUMN hours DOUBLE")99 con.execute("UPDATE segday SET hours = epoch(b - a) / 3600.0")100 _aggregate(con, "segday", "day", "offers_daily")101 # flyer week: Thursday -> Wednesday, keyed by the Thursday102 con.execute("""CREATE OR REPLACE TABLE segweek AS103 SELECT *, day - CAST((dayofweek(day) + 3) % 7 AS INTEGER) AS week FROM segday""") # dayofweek: Sun=0 ... Thu=4 -> offset 0104 _aggregate(con, "segweek", "week", "offers_weekly")105106def _aggregate(con, table: str, key: str, out: str) -> None:107 con.execute(f"""108 CREATE OR REPLACE TABLE {out} AS109 WITH byprice AS (110 SELECT uid, source, {key}, price, SUM(hours) AS h, MIN(a) AS first_a111 FROM {table} GROUP BY uid, source, {key}, price112 ), tot AS (113 SELECT uid, {key}, SUM(h) AS hours_obs, SUM(CASE WHEN price IS NULL THEN h ELSE 0 END) AS hours_null,114 COUNT(*) FILTER (WHERE price IS NOT NULL) AS n_distinct115 FROM byprice GROUP BY uid, {key}116 ), ranked AS (117 SELECT *, ROW_NUMBER() OVER (PARTITION BY uid, {key} ORDER BY h DESC, price ASC) AS rk118 FROM byprice WHERE price IS NOT NULL119 ), modep AS (SELECT uid, {key}, price AS p_mode, h AS mode_hours FROM ranked WHERE rk = 1),120 stats AS (121 SELECT uid, source, {key}, MIN(price) AS p_min, MAX(price) AS p_max,122 SUM(price * hours) / NULLIF(SUM(hours), 0) AS p_twmean,123 ARG_MAX(price, b) AS p_close, ARG_MIN(price, a) AS p_open,124 COUNT(*) - 1 AS n_changes125 FROM {table} WHERE price IS NOT NULL GROUP BY uid, source, {key}126 ), med AS (127 SELECT uid, {key}, QUANTILE_CONT(price, 0.5) AS p_median_events FROM {table} WHERE price IS NOT NULL GROUP BY uid, {key}128 )129 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,130 md.p_median_events, t.n_distinct, s.n_changes, t.hours_obs, t.hours_null,131 m.mode_hours / NULLIF(t.hours_obs - t.hours_null, 0) AS mode_share,132 (s.p_max - s.p_min) / NULLIF(m.p_mode, 0) AS rel_range133 FROM stats s JOIN modep m USING (uid, {key}) JOIN tot t USING (uid, {key}) JOIN med md USING (uid, {key})""")134 con.execute(f"COPY {out} TO '{path('processed') / (out + '.parquet')}' (FORMAT PARQUET)")135 n, = con.execute(f"SELECT COUNT(*) FROM {out}").fetchone()136 log.info("%s: %d rows", out, n)137138def main() -> None:139 interim = path("interim")140 con = duckdb.connect()141 for t in ["products", "price_log", "sync_log"]:142 con.execute(f"CREATE VIEW raw_{t} AS SELECT * FROM read_parquet('{interim / ('raw_' + t + '.parquet')}')")143 build(con)144 prod = canonical_products(pd.read_parquet(interim / "raw_products.parquet"))145 prod.to_parquet(path("processed") / "products.parquet", index=False)146 log.info("products canonical: %d rows; unit price available for %.1f%% (Food-Ka: %.1f%%)", len(prod),147 100 * prod["unit_price_rp"].notna().mean(), 100 * prod["unit_price"].notna().mean())148 sl = pd.read_parquet(interim / "raw_sync_log.parquet")149 sl["ts_utc"] = pd.to_datetime(sl["ts"], unit="s", utc=True)150 st = sl["stats"].apply(lambda s: json.loads(s) if isinstance(s, str) and s else {})151 sl["null_price_rate"] = st.apply(lambda d: d.get("null_price_rate")); sl["missed"] = st.apply(lambda d: d.get("missed"))152 sl["pipeline_commit"] = SETTINGS["snapshot"]["foodka_commit"][:7]153 sl.drop(columns=["stats"]).to_parquet(path("processed") / "snapshots.parquet", index=False)154 files = [path("processed") / f for f in ["offers_daily.parquet", "offers_weekly.parquet", "products.parquet", "snapshots.parquet"]]155 write_manifest("panel", files)156 ledger("1", "build_panel", "offers_daily/weekly, products, snapshots written")157158if __name__ == "__main__":159 main()160