SPB Git
spboucher.ai

spb/wp37_uqo

Public

phase1: ingestion to parquet, unit parser (+tests), retailers hierarchy, daily/weekly panel reconstruction (duckdb), StatCan benchmarks, panel invariant tests

Simon-Pierre Boucher committed (Nov 1, 2026) parent 0ecd954 Browse files

12 changed files +549 −1

12 changed files
Unified Split
modified reports/phase0_feasibility.md +1 −1
@@ -34,7 +34,7 @@ Public cross-check (2026-11-01 ~08:00 UTC): homepage shows 50 760 products, 54 b
34 34
35 ### 3.2 Data-generating process (from the connector code, read-only copy in `data/raw/foodka-src/`) 35 ### 3.2 Data-generating process (from the connector code, read-only copy in `data/raw/foodka-src/`)
36 36
37−- 57 active source feeds = **54 with active products**, belonging to far fewer **retail companies**: Loblaw (maxi, provigo, club_entrepot + 3 flyers + inter_marche, axep? no: Axep/Richelieu/Marché Ami are Metro affiliates), Metro Inc. (metro, superc + 2 flyers + richelieu, marche_ami, axep), Sobeys (iga + flyer, tradition, bonichoix, rachelle_bery), Walmart, Costco, Adonis (Metro Inc.), Avril, independents, 4 pharmacies, SAQ. A `retailers` table (company → banner → feed → channel) will be built in Phase 1. 37+- 57 active source feeds = **54 with active products**, belonging to far fewer **retail companies** (Loblaw, Metro Inc., Sobeys/Empire, Walmart, Costco, independents, four pharmacy banners, the SAQ monopoly); several banners appear twice (online storefront feed + Flipp circular feed). A `retailers` table (company → banner → feed → channel) is built in Phase 1 and ownership is verified against public corporate information in Phase 2.
38 - **Two channels**: (a) *online storefront* feeds (Shopify/WooCommerce JSON, Loblaw/Metro/Walmart/Costco/IGA sites scraped through residential proxy pools without store selection); (b) *weekly circulars* via Flipp for postal code **H2X 1Y4 (Montréal)** with `valid_from`/`valid_to`, `original_price`, `dollars_off`, fine print. 38 - **Two channels**: (a) *online storefront* feeds (Shopify/WooCommerce JSON, Loblaw/Metro/Walmart/Costco/IGA sites scraped through residential proxy pools without store selection); (b) *weekly circulars* via Flipp for postal code **H2X 1Y4 (Montréal)** with `valid_from`/`valid_to`, `original_price`, `dollars_off`, fine print.
39 - **Geography**: no store identifier anywhere. Circulars are regional (Montréal zone); online prices are whatever store the retailer's site assigned to the proxy session. → **Store-level geography is unavailable** (H4 and travel-cost components of Model E cannot be estimated from Food-Ka). 39 - **Geography**: no store identifier anywhere. Circulars are regional (Montréal zone); online prices are whatever store the retailer's site assigned to the proxy session. → **Store-level geography is unavailable** (H4 and travel-cost components of Model E cannot be estimated from Food-Ka).
40 - **Identifiers**: Metro and Super C `external_id` are valid GS1 UPC-A codes (100 % check-digit valid, n = 1 954 active). Shopify/Walmart/Loblaw ids are internal SKUs (check-digit validity ≈ 10 %, i.e. chance). Exact GTIN matching is therefore only possible within Metro Inc.; everything else requires validated probabilistic matching. 40 - **Identifiers**: Metro and Super C `external_id` are valid GS1 UPC-A codes (100 % check-digit valid, n = 1 954 active). Shopify/Walmart/Loblaw ids are internal SKUs (check-digit validity ≈ 10 %, i.e. chance). Exact GTIN matching is therefore only possible within Metro Inc.; everything else requires validated probabilistic matching.
added src/__init__.py +0 −0
added src/cleaning/__init__.py +0 −0
added src/cleaning/build_panel.py +159 −0
@@ -0,0 +1,159 @@
1+"""Phase 1.3 — reconstruct the product × day (and × week) price panel from the event log.
2+
3+Data-generating process (Food-Ka db.py, read-only copy): price_log holds one row at a product's first observation and
4+one row at every change of the displayed price (NULL = no displayed price). Between two rows the product kept the
5+recorded price and was re-observed at every connector pass (median 7.2 h). A product's observation span is
6+[first_seen, last_seen]; after two consecutive misses it is deactivated (active=0) — gaps inside the span are not
7+logged (limitation documented in reports/phase1_data_audit.md).
8+
9+Outputs (data/processed):
10+ offers_daily.parquet — uid × local date: duration-weighted price statistics
11+ offers_weekly.parquet — uid × week (Thu→Wed 'flyer week'): same statistics
12+ products.parquet — canonical product table with re-parsed sizes, unit prices, flags
13+ snapshots.parquet — connector passes (from sync_log)
14+"""
15+from __future__ import annotations
16+import json
17+import duckdb, pandas as pd, numpy as np
18+from src.common import path, get_logger, ledger, write_manifest, TZ, SETTINGS
19+from src.cleaning.units import parse_size, parse_price_label, unit_price
20+
21+log = get_logger("panel")
22+
23+def 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.0
59+ return out.drop(columns=["details", "images", "description"])
60+
61+def _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 False
64+ 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 == chk
67+
68+def build(con: duckdb.DuckDBPyConnection) -> None:
69+ con.execute(f"SET TimeZone='{TZ}'")
70+ con.execute("""
71+ CREATE OR REPLACE TABLE ev AS
72+ 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_seen
74+ 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_seen
76+ con.execute("""
77+ CREATE OR REPLACE TABLE seg AS
78+ SELECT uid, source, price, ts AS seg_start,
79+ COALESCE(LEAD(ts) OVER (PARTITION BY uid ORDER BY ts), last_seen) AS seg_end
80+ FROM ev""")
81+ con.execute("DELETE FROM seg WHERE seg_end <= seg_start")
82+ # expand to local days
83+ con.execute("""
84+ CREATE OR REPLACE TABLE segday AS
85+ 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 seg
87+ ), 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 seg
90+ ), g AS (
91+ SELECT uid, source, price, seg_start, seg_end, UNNEST(generate_series(d0, d1, INTERVAL 1 DAY))::DATE AS day FROM s2
92+ )
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 b
96+ 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 Thursday
102+ con.execute("""CREATE OR REPLACE TABLE segweek AS
103+ SELECT *, day - CAST((dayofweek(day) + 3) % 7 AS INTEGER) AS week FROM segday""") # dayofweek: Sun=0 ... Thu=4 -> offset 0
104+ _aggregate(con, "segweek", "week", "offers_weekly")
105+
106+def _aggregate(con, table: str, key: str, out: str) -> None:
107+ con.execute(f"""
108+ CREATE OR REPLACE TABLE {out} AS
109+ WITH byprice AS (
110+ SELECT uid, source, {key}, price, SUM(hours) AS h, MIN(a) AS first_a
111+ FROM {table} GROUP BY uid, source, {key}, price
112+ ), 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_distinct
115+ 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 rk
118+ FROM byprice WHERE price IS NOT NULL
119+ ), 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_changes
125+ 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_range
133+ 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)
137+
138+def 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")
157+
158+if __name__ == "__main__":
159+ main()
added src/cleaning/retailers.py +96 −0
@@ -0,0 +1,96 @@
1+"""Phase 1.2 — canonical `retailers` table: company → banner → source feed → channel.
2+
3+Ownership/affiliation is taken from the Food-Ka registry text (`region`, `tech`) and from public corporate
4+information; the column `ownership_basis` records which. Items marked 'public-corporate' are re-verified in
5+Phase 2 (institutional context) and any correction is logged.
6+"""
7+from __future__ import annotations
8+import pandas as pd
9+from src.common import path, get_logger, ledger
10+
11+log = get_logger("retailers")
12+
13+# feed id -> (company, banner, channel, format, ownership_basis)
14+MAP = {
15+ "metro": ("Metro Inc.", "Metro", "online_storefront", "supermarket", "public-corporate"),
16+ "metro_flyer": ("Metro Inc.", "Metro", "circular", "supermarket", "public-corporate"),
17+ "superc": ("Metro Inc.", "Super C", "online_storefront", "discount", "public-corporate"),
18+ "superc_flyer": ("Metro Inc.", "Super C", "circular", "discount", "public-corporate"),
19+ "adonis": ("Metro Inc.", "Marché Adonis", "online_storefront", "ethnic_supermarket", "public-corporate"),
20+ "adonis_flyer": ("Metro Inc.", "Marché Adonis", "circular", "ethnic_supermarket", "public-corporate"),
21+ "richelieu": ("Metro Inc. (affiliate)", "Marché Richelieu", "circular", "proximity", "registry"),
22+ "marche_ami": ("Metro Inc. (affiliate)", "Marché Ami", "circular", "proximity", "registry"),
23+ "axep": ("Metro Inc. (affiliate)", "Axep", "circular", "proximity", "registry"),
24+ "jean_coutu": ("Metro Inc.", "Jean Coutu", "circular", "pharmacy", "public-corporate"),
25+ "brunet": ("Metro Inc.", "Brunet", "circular", "pharmacy", "public-corporate"),
26+ "maxi": ("Loblaw Companies", "Maxi", "online_storefront", "discount", "public-corporate"),
27+ "maxi_flyer": ("Loblaw Companies", "Maxi", "circular", "discount", "public-corporate"),
28+ "provigo": ("Loblaw Companies", "Provigo", "online_storefront", "supermarket", "public-corporate"),
29+ "provigo_flyer": ("Loblaw Companies", "Provigo", "circular", "supermarket", "public-corporate"),
30+ "club_entrepot": ("Loblaw Companies", "Club Entrepôt", "online_storefront", "wholesale_club", "registry"),
31+ "pharmaprix": ("Loblaw Companies", "Pharmaprix", "circular", "pharmacy", "public-corporate"),
32+ "tt": ("Loblaw Companies", "T&T Supermarket", "online_storefront", "ethnic_supermarket", "public-corporate"),
33+ "inter_marche": ("Loblaw Companies (affiliate)", "L'Inter-Marché", "circular", "proximity", "registry"),
34+ "inter_marche_intl": ("Loblaw Companies (affiliate)", "L'Inter-Marché International", "circular", "ethnic_supermarket", "registry"),
35+ "iga": ("Sobeys / Empire", "IGA", "online_storefront", "supermarket", "public-corporate"),
36+ "iga_flyer": ("Sobeys / Empire", "IGA", "circular", "supermarket", "public-corporate"),
37+ "tradition": ("Sobeys / Empire", "Les Marchés Tradition", "circular", "proximity", "registry"),
38+ "bonichoix": ("Sobeys / Empire", "Marché Bonichoix", "circular", "proximity", "registry"),
39+ "rachelle_bery": ("Sobeys / Empire", "Rachelle-Béry", "circular", "health_specialty", "public-corporate"),
40+ "walmart": ("Walmart Canada", "Walmart", "online_storefront", "mass_merchant", "public-corporate"),
41+ "walmart_flyer": ("Walmart Canada", "Walmart", "circular", "mass_merchant", "public-corporate"),
42+ "costco": ("Costco Wholesale Canada", "Costco", "online_storefront", "warehouse_club", "public-corporate"),
43+ "costco_flyer": ("Costco Wholesale Canada", "Costco", "circular", "warehouse_club", "public-corporate"),
44+ "giant_tiger": ("Giant Tiger Stores", "Giant Tiger", "online_storefront", "discount_general", "public-corporate"),
45+ "avril": ("Avril Supermarché Santé", "Avril", "online_storefront", "health_specialty", "registry"),
46+ "avril_flyer": ("Avril Supermarché Santé", "Avril", "circular", "health_specialty", "registry"),
47+ "pa": ("Supermarché PA", "PA Supermarché", "online_storefront", "independent_supermarket", "registry"),
48+ "pa_nature": ("Supermarché PA", "PA Nature", "circular", "health_specialty", "registry"),
49+ "uniprix": ("McKesson Canada", "Uniprix", "circular", "pharmacy", "public-corporate"),
50+ "saq": ("Société des alcools du Québec", "SAQ", "online_storefront", "alcohol_monopoly", "registry"),
51+ "mayrand": ("Groupe Mayrand", "Mayrand", "online_storefront", "cash_and_carry", "registry"),
52+ "epipresto": ("Epipresto", "Epipresto", "online_storefront", "online_only", "registry"),
53+ "nuvo": ("Marché Nuvo", "Marché Nuvo", "online_storefront", "independent_supermarket", "registry"),
54+ "boite_a_grains": ("La Boîte à Grains", "La Boîte à Grains", "online_storefront", "health_specialty", "registry"),
55+ "bocoboco": ("BocoBoco", "BocoBoco", "online_storefront", "online_only", "registry"),
56+ "aliments_merci": ("Aliments Merci", "Aliments Merci", "online_storefront", "online_only", "registry"),
57+ "akhavan": ("Akhavan", "Akhavan", "online_storefront", "ethnic_supermarket", "registry"),
58+ "aubut": ("Aubut", "Aubut", "online_storefront", "cash_and_carry", "registry"),
59+ "maturin": ("Maturin", "Maturin", "online_storefront", "online_only", "registry"),
60+ "tau": ("Marché Tau", "Marché Tau", "online_storefront", "health_specialty", "registry"),
61+ "loco": ("Épiceries LOCO", "LOCO", "online_storefront", "zero_waste", "registry"),
62+ "kim_phat": ("Kim Phat", "Kim Phat", "circular", "ethnic_supermarket", "registry"),
63+ "fu_tai": ("Marché Fu Tai", "Fu Tai", "circular", "ethnic_supermarket", "registry"),
64+ "marche_ct": ("Marché C&T", "C&T", "circular", "ethnic_supermarket", "registry"),
65+ "aures": ("Supermarché Aurès", "Aurès", "circular", "ethnic_supermarket", "registry"),
66+ "euro_marche": ("Euro Marché", "Euro Marché", "circular_hybrid", "ethnic_supermarket", "registry"),
67+ "val_mont": ("Marché Val-Mont", "Val-Mont", "circular", "independent_supermarket", "registry"),
68+ "bonanza": ("Marché Bonanza", "Bonanza", "circular", "independent_supermarket", "registry"),
69+ "pasquier": ("Pasquier", "Pasquier", "circular", "independent_supermarket", "registry"),
70+ "marche_vegetarien": ("Le Marché Végétarien", "Marché Végétarien", "circular", "health_specialty", "registry"),
71+ "aliments_mm": ("Les Aliments M&M", "M&M", "circular", "frozen_specialty", "registry"),
72+ "bulk_barn": ("Bulk Barn Foods", "Bulk Barn", "none", "bulk_specialty", "public-corporate"),
73+ "dollarama": ("Dollarama", "Dollarama", "none", "discount_general", "public-corporate"),
74+ "lufa": ("Fermes Lufa", "Lufa", "none", "online_only", "registry"),
75+ "frenco": ("Frenco", "Frenco", "none", "bulk_specialty", "registry"),
76+}
77+
78+def main() -> None:
79+ src = pd.read_parquet(path("interim") / "sources.parquet")
80+ rows = []
81+ for _, r in src.iterrows():
82+ comp, banner, channel, fmt, basis = MAP[r["id"]]
83+ rows.append(dict(feed_id=r["id"], feed_name=r["name"], company=comp, banner=banner, channel=channel,
84+ store_format=fmt, ownership_basis=basis, registry_status=r["status"], registry_region=r["region"],
85+ tech=r["tech"], url=r["url"],
86+ circular_postal_zone=("H2X 1Y4 (Montréal)" if channel.startswith("circular") else None),
87+ store_identified=False, is_grocery=(fmt not in ("pharmacy", "alcohol_monopoly"))))
88+ ret = pd.DataFrame(rows)
89+ ret.to_parquet(path("processed") / "retailers.parquet", index=False)
90+ ret.to_csv(path("processed") / "retailers.csv", index=False)
91+ log.info("retailers: %d feeds, %d banners, %d companies; channels=%s", len(ret), ret.banner.nunique(), ret.company.nunique(),
92+ ret.channel.value_counts().to_dict())
93+ ledger("1", "retailers", f"{len(ret)} feeds -> {ret.banner.nunique()} banners -> {ret.company.nunique()} companies")
94+
95+if __name__ == "__main__":
96+ main()
added src/cleaning/units.py +90 −0
@@ -0,0 +1,90 @@
1+"""Unit normalisation for grocery listings (CAD; g / ml / unit).
2+
3+Re-implements and extends Food-Ka's parser (data/raw/foodka-src/foodka/normalize.py) so that the research
4+pipeline does not depend on production code. Differences are documented in reports/phase1_data_audit.md.
5+"""
6+from __future__ import annotations
7+import re, unicodedata
8+from dataclasses import dataclass
9+
10+_UNITS = {"kg": (1000.0, "g"), "g": (1.0, "g"), "mg": (0.001, "g"), "lb": (453.592, "g"), "lbs": (453.592, "g"),
11+ "oz": (28.3495, "g"), "l": (1000.0, "ml"), "ml": (1.0, "ml"), "cl": (10.0, "ml"),
12+ "un": (1.0, "un"), "unite": (1.0, "un"), "unites": (1.0, "un"), "ea": (1.0, "un"), "each": (1.0, "un"),
13+ "pk": (1.0, "un"), "ch": (1.0, "un"), "pc": (1.0, "un"), "pcs": (1.0, "un"), "units": (1.0, "un"), "unit": (1.0, "un")}
14+_SIZE_RE = re.compile(r"(?:(\d+)\s*[x×*]\s*)?(\d+(?:[.,]\d+)?)\s*(kg|g|mg|lb|lbs|oz|ml|cl|l|un|unites?|ea|each|pk|ch|pcs?|units?)\b", re.I)
15+_RANGE_RE = re.compile(r"(\d+(?:[.,]\d+)?)\s*(?:-|–|a|à)\s*(\d+(?:[.,]\d+)?)\s*(kg|g|lb|lbs|oz|ml|l)\b", re.I)
16+_PERKG_RE = re.compile(r"/\s*(kg|lb|100\s*g|100\s*ml|l|ch|un)\b", re.I)
17+_MULTI_RE = re.compile(r"^\s*(\d+)\s*(?:/|pour|for)\s*(?:(\d{1,4}(?:[.,]\d{1,2})?)\s*\$)?")
18+_APPROX_RE = re.compile(r"environ|approx|moyenne|avg|about|~", re.I)
19+_SOLD_INDIV_RE = re.compile(r"vendu individuellement|sold individually|a l'unite|à l'unité|l'unite|unite$", re.I)
20+
21+def strip_accents(s: str) -> str:
22+ return "".join(c for c in unicodedata.normalize("NFKD", s or "") if not unicodedata.combining(c))
23+
24+@dataclass
25+class Size:
26+ qty: float # in base unit (g, ml, un)
27+ base: str # 'g' | 'ml' | 'un'
28+ approx: bool = False
29+ source: str = "size_label" # where it was parsed from
30+
31+def parse_size(label: str | None, name: str | None = None) -> Size | None:
32+ """Parse '450 g', '2 L', '12 x 355 ml', '6 un', 'Vendu individuellement', '1,36 kg'; falls back to the name."""
33+ for text, src in ((label, "size_label"), (name, "name")):
34+ if not text:
35+ continue
36+ t = strip_accents(text).lower().replace("\u00a0", " ")
37+ m = _RANGE_RE.search(t)
38+ if m: # "1,2 - 1,5 kg" -> midpoint, approximate (checked before the simple pattern)
39+ a, b = float(m.group(1).replace(",", ".")), float(m.group(2).replace(",", "."))
40+ factor, base = _UNITS[m.group(3).lower()]
41+ return Size((a + b) / 2 * factor, base, True, src)
42+ m = _SIZE_RE.search(t)
43+ if m:
44+ mult = int(m.group(1)) if m.group(1) else 1
45+ qty = float(m.group(2).replace(",", "."))
46+ factor, base = _UNITS[m.group(3).lower()]
47+ total = mult * qty * factor
48+ if total > 0:
49+ return Size(total, base, bool(_APPROX_RE.search(t)), src)
50+ if src == "size_label" and _SOLD_INDIV_RE.search(t):
51+ return Size(1.0, "un", False, src)
52+ return None
53+
54+@dataclass
55+class PriceLabel:
56+ per_weight_unit: str | None = None # 'kg','lb','100 g','100 ml','l' when the displayed price is per weight/volume
57+ multibuy_n: int | None = None # 'n / x $'
58+ multibuy_total: float | None = None
59+
60+def parse_price_label(label: str | None) -> PriceLabel:
61+ pl = PriceLabel()
62+ if not label:
63+ return pl
64+ t = strip_accents(label).lower().replace(" ", " ")
65+ m = _PERKG_RE.search(t)
66+ if m:
67+ pl.per_weight_unit = re.sub(r"\s+", " ", m.group(1))
68+ m = _MULTI_RE.match(t)
69+ if m:
70+ pl.multibuy_n = int(m.group(1))
71+ if m.group(2):
72+ pl.multibuy_total = float(m.group(2).replace(",", "."))
73+ return pl
74+
75+def unit_price(price: float | None, size: Size | None, label: PriceLabel | None = None) -> tuple[float, str] | None:
76+ """Comparable price: $/100 g, $/100 ml or $/unit. Returns None when not computable."""
77+ if price is None or price <= 0:
78+ return None
79+ if label and label.per_weight_unit:
80+ u = label.per_weight_unit
81+ if u == "kg": return (round(price / 10.0, 6), "100 g")
82+ if u == "lb": return (round(price / 4.53592, 6), "100 g")
83+ if u == "100 g": return (price, "100 g")
84+ if u == "100 ml": return (price, "100 ml")
85+ if u == "l": return (round(price / 10.0, 6), "100 ml")
86+ if size is None:
87+ return None
88+ if size.base == "un":
89+ return (round(price / size.qty, 6), "un")
90+ return (round(price / size.qty * 100.0, 6), f"100 {size.base}")
added src/common.py +59 −0
@@ -0,0 +1,59 @@
1+"""Shared helpers: settings, paths, logging, deterministic seeds."""
2+from __future__ import annotations
3+import hashlib, json, logging, os, random, sys, time
4+from pathlib import Path
5+import numpy as np
6+import yaml
7+
8+ROOT = Path(__file__).resolve().parents[1]
9+SETTINGS = yaml.safe_load((ROOT / "config" / "settings.yaml").read_text())
10+SEED = int(SETTINGS["seed"])
11+TZ = SETTINGS.get("timezone_display", "America/Toronto")
12+
13+def path(key: str) -> Path:
14+ p = ROOT / SETTINGS["paths"][key]
15+ p.mkdir(parents=True, exist_ok=True)
16+ return p
17+
18+def snapshot_db() -> Path:
19+ return ROOT / SETTINGS["snapshot"]["path"]
20+
21+def verify_snapshot() -> str:
22+ h = hashlib.sha256()
23+ with open(snapshot_db(), "rb") as f:
24+ for chunk in iter(lambda: f.read(1 << 20), b""):
25+ h.update(chunk)
26+ digest = h.hexdigest()
27+ if digest != SETTINGS["snapshot"]["sha256"]:
28+ raise SystemExit(f"snapshot sha256 mismatch: {digest}")
29+ return digest
30+
31+def seed_everything(seed: int = SEED) -> None:
32+ random.seed(seed); np.random.seed(seed)
33+ os.environ["PYTHONHASHSEED"] = str(seed)
34+
35+def get_logger(name: str) -> logging.Logger:
36+ log = logging.getLogger(name)
37+ if not log.handlers:
38+ fmt = logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s", "%Y-%m-%dT%H:%M:%S")
39+ sh = logging.StreamHandler(sys.stdout); sh.setFormatter(fmt); log.addHandler(sh)
40+ fh = logging.FileHandler(path("logs") / "pipeline.log"); fh.setFormatter(fmt); log.addHandler(fh)
41+ log.setLevel(logging.INFO)
42+ return log
43+
44+def ledger(phase: str, action: str, result: str) -> None:
45+ """Append a machine-readable line to the structured run ledger."""
46+ rec = {"utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "phase": phase, "action": action, "result": result}
47+ with open(path("logs") / "run_ledger.jsonl", "a") as f:
48+ f.write(json.dumps(rec, ensure_ascii=False) + "\n")
49+
50+def write_manifest(name: str, files: list[Path], extra: dict | None = None) -> None:
51+ man = {"name": name, "utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "seed": SEED,
52+ "snapshot": SETTINGS["snapshot"]["id"], "files": {}}
53+ for f in files:
54+ f = Path(f)
55+ if f.exists():
56+ man["files"][str(f.relative_to(ROOT))] = {"bytes": f.stat().st_size,
57+ "sha256": hashlib.sha256(f.read_bytes()).hexdigest() if f.stat().st_size < 400_000_000 else None}
58+ if extra: man.update(extra)
59+ (path("logs") / f"manifest_{name}.json").write_text(json.dumps(man, indent=2, ensure_ascii=False))
added src/ingestion/__init__.py +0 −0
added src/ingestion/load_snapshot.py +29 −0
@@ -0,0 +1,29 @@
1+"""Phase 1.1 — Load the authorised Food-Ka SQLite snapshot (read-only) into Parquet tables.
2+
3+Nothing is modified in the snapshot. Output: data/interim/raw_{products,price_log,sync_log,product_links,off_cache}.parquet
4+plus sources.parquet from the JSON registry.
5+"""
6+from __future__ import annotations
7+import json, sqlite3
8+import pandas as pd
9+from src.common import ROOT, SETTINGS, path, verify_snapshot, snapshot_db, get_logger, ledger, write_manifest
10+
11+log = get_logger("ingest")
12+
13+def main() -> None:
14+ digest = verify_snapshot()
15+ log.info("snapshot verified sha256=%s", digest[:16])
16+ con = sqlite3.connect(f"file:{snapshot_db()}?mode=ro", uri=True)
17+ out = path("interim"); files = []
18+ for t in ["products", "price_log", "sync_log", "product_links", "off_cache"]:
19+ df = pd.read_sql_query(f"SELECT * FROM {t}", con)
20+ f = out / f"raw_{t}.parquet"; df.to_parquet(f, index=False); files.append(f)
21+ log.info("%s: %d rows x %d cols", t, *df.shape)
22+ reg = json.loads((ROOT / SETTINGS["snapshot"]["sources_registry"]).read_text())["sources"]
23+ src = pd.DataFrame(reg); f = out / "sources.parquet"; src.to_parquet(f, index=False); files.append(f)
24+ log.info("sources registry: %d entries (%s)", len(src), src["status"].value_counts().to_dict())
25+ write_manifest("ingest", files, {"snapshot_sha256": digest})
26+ ledger("1", "load_snapshot", f"{len(files)} parquet files written to data/interim")
27+
28+if __name__ == "__main__":
29+ main()
added src/ingestion/statcan.py +54 −0
@@ -0,0 +1,54 @@
1+"""Download Statistics Canada benchmarks via the WDS API (public, no key).
2+
3+- 18-10-0004-01 CPI monthly NSA: Quebec (geo 11), Montréal (13), Canada (2) × food aggregates (members 3,4,5,21,22,28,29,30,34,39,49,58)
4+- 18-10-0245-01 Monthly average retail prices for selected products: Quebec, all 110 products
5+Raw JSON kept in data/raw/statcan/ with download timestamp; tidy CSVs in data/processed/.
6+"""
7+from __future__ import annotations
8+import json, time, requests, pandas as pd
9+from src.common import ROOT, path, get_logger, ledger
10+
11+log = get_logger("statcan")
12+WDS = "https://www150.statcan.gc.ca/t1/wds/rest/"
13+RAW = ROOT / "data" / "raw" / "statcan"; RAW.mkdir(parents=True, exist_ok=True)
14+
15+def coords(product_id: int, pairs: list[tuple[int, int]], periods: int) -> pd.DataFrame:
16+ body = [{"productId": product_id, "coordinate": f"{g}.{p}.0.0.0.0.0.0.0.0", "latestN": periods} for g, p in pairs]
17+ r = requests.post(WDS + "getDataFromCubePidCoordAndLatestNPeriods", json=body, timeout=120); r.raise_for_status()
18+ (RAW / f"{product_id}_{int(time.time())}.json").write_text(r.text)
19+ rows = []
20+ for item in r.json():
21+ if item["status"] != "SUCCESS":
22+ log.warning("WDS item failed: %s", item); continue
23+ o = item["object"]
24+ for dp in o["vectorDataPoint"]:
25+ rows.append(dict(productId=product_id, coordinate=o["coordinate"], vectorId=o["vectorId"],
26+ ref_date=dp["refPer"], value=dp["value"], scalar=dp["scalarFactorCode"], status=dp.get("statusCode")))
27+ return pd.DataFrame(rows)
28+
29+def main() -> None:
30+ meta = {}
31+ for pid in (18100004, 18100245):
32+ r = requests.post(WDS + "getCubeMetadata", json=[{"productId": pid}], timeout=120); r.raise_for_status()
33+ meta[pid] = r.json()[0]["object"]; (RAW / f"meta_{pid}.json").write_text(r.text)
34+ # CPI
35+ m = meta[18100004]; geo = {x["memberId"]: x["memberNameEn"] for x in m["dimension"][0]["member"]}
36+ prodn = {x["memberId"]: x["memberNameEn"] for x in m["dimension"][1]["member"]}
37+ food_members = [2, 3, 4, 5, 16, 21, 22, 28, 29, 30, 34, 39, 49, 58]
38+ cpi = coords(18100004, [(g, p) for g in (2, 11, 13) for p in food_members], periods=60)
39+ cpi["geo"] = cpi.coordinate.str.split(".").str[0].astype(int).map(geo)
40+ cpi["product"] = cpi.coordinate.str.split(".").str[1].astype(int).map(prodn)
41+ cpi.to_csv(path("processed") / "statcan_cpi_food.csv", index=False)
42+ log.info("CPI: %d points, %s → %s, latest release %s", len(cpi), cpi.ref_date.min(), cpi.ref_date.max(), m.get("releaseTime"))
43+ # Average retail prices
44+ m2 = meta[18100245]; geo2 = {x["memberId"]: x["memberNameEn"] for x in m2["dimension"][0]["member"]}
45+ prod2 = {x["memberId"]: x["memberNameEn"] for x in m2["dimension"][1]["member"]}
46+ qc = [k for k, v in geo2.items() if v == "Quebec"][0]
47+ arp = coords(18100245, [(qc, p) for p in prod2], periods=36)
48+ arp["geo"] = "Quebec"; arp["product"] = arp.coordinate.str.split(".").str[1].astype(int).map(prod2)
49+ arp.to_csv(path("processed") / "statcan_avg_retail_prices_qc.csv", index=False)
50+ log.info("Average retail prices QC: %d points, %d products, %s → %s", len(arp), arp["product"].nunique(), arp.ref_date.min(), arp.ref_date.max())
51+ ledger("1", "statcan", f"CPI {len(cpi)} pts to {cpi.ref_date.max()}; ARP {len(arp)} pts to {arp.ref_date.max()}")
52+
53+if __name__ == "__main__":
54+ main()
added tests/test_panel.py +39 −0
@@ -0,0 +1,39 @@
1+"""Invariants of the reconstructed panel against the raw event log."""
2+import duckdb, pandas as pd, numpy as np, pytest
3+from src.common import ROOT
4+
5+P = ROOT / "data" / "processed"; I = ROOT / "data" / "interim"
6+
7+@pytest.fixture(scope="module")
8+def d():
9+ return pd.read_parquet(P / "offers_daily.parquet")
10+
11+def test_daily_bounds(d):
12+ assert (d.hours_obs <= 24.0 + 1e-6).all()
13+ assert (d.p_min <= d.p_mode + 1e-9).all() and (d.p_mode <= d.p_max + 1e-9).all()
14+ assert (d.p_min <= d.p_twmean + 1e-9).all() and (d.p_twmean <= d.p_max + 1e-9).all()
15+ assert (d.n_distinct >= 1).all() and (d.mode_share <= 1.0 + 1e-9).all()
16+
17+def test_total_hours_equal_span(d):
18+ prod = pd.read_parquet(I / "raw_products.parquet", columns=["uid", "first_seen", "last_seen"])
19+ pl = pd.read_parquet(I / "raw_price_log.parquet")
20+ # products with >=1 priced event: covered hours == span from first event to last_seen (NULL segments excluded)
21+ tot = d.groupby("uid").hours_obs.sum()
22+ first_ev = pl.groupby("uid").ts.min()
23+ span = ((prod.set_index("uid").last_seen - first_ev) / 3600.0).dropna()
24+ common = tot.index.intersection(span.index)
25+ diff = (tot.loc[common] - span.loc[common]).abs()
26+ assert diff.quantile(0.99) < 1e-3, diff.describe()
27+
28+def test_known_sequence(d):
29+ # metro:068400662600 (mayonnaise): raw events on 2026-08-13 local = 7.49 carried, then 5.77 at 07h UTC... bounds must hold
30+ x = d[d.uid == "metro:068400662600"].sort_values("day")
31+ assert x.day.min() == pd.Timestamp("2026-08-12").date()
32+ assert set(x.loc[x.day == pd.Timestamp("2026-08-13").date(), "p_min"]) == {5.77}
33+ assert (x.hours_obs.iloc[1:-1] == 24.0).all()
34+
35+def test_single_price_products_have_one_distinct(d):
36+ pl = pd.read_parquet(I / "raw_price_log.parquet")
37+ one = pl.groupby("uid").size(); one = one[one == 1].index
38+ sub = d[d.uid.isin(one)]
39+ assert (sub.n_distinct == 1).all() and (sub.n_changes == 0).all()
added tests/test_units.py +22 −0
@@ -0,0 +1,22 @@
1+from src.cleaning.units import parse_size, parse_price_label, unit_price
2+
3+def test_sizes():
4+ assert parse_size("450 g").qty == 450 and parse_size("450 g").base == "g"
5+ assert parse_size("2 L").qty == 2000 and parse_size("2 L").base == "ml"
6+ assert parse_size("12 x 355 ml").qty == 12 * 355
7+ assert parse_size("1,36 kg").qty == 1360
8+ assert parse_size("6 un").base == "un" and parse_size("6 un").qty == 6
9+ assert parse_size("Vendu individuellement").base == "un"
10+ s = parse_size("Une barquette contient en moyenne 1250 g"); assert s.qty == 1250 and s.approx
11+ s = parse_size("", "Lait 2% 2 L"); assert s.qty == 2000 and s.source == "name"
12+ s = parse_size("1,2 - 1,5 kg"); assert abs(s.qty - 1350) < 1e-6 and s.approx
13+ assert parse_size("") is None and parse_size(None, None) is None
14+
15+def test_labels_and_unit_price():
16+ pl = parse_price_label("6,57 $ / kg"); assert pl.per_weight_unit == "kg"
17+ assert unit_price(6.57, None, pl) == (0.657, "100 g")
18+ pl = parse_price_label("2 / 5,00 $"); assert pl.multibuy_n == 2 and pl.multibuy_total == 5.0
19+ assert unit_price(4.99, parse_size("500 g")) == (0.998, "100 g")
20+ assert unit_price(3.0, parse_size("6 un")) == (0.5, "un")
21+ assert unit_price(None, parse_size("500 g")) is None
22+ assert unit_price(4.99, None) is None
23
Menu

Theme
↑↓ navigate ↵ open esc close / or ⌘K anywhere