Basket Bundle Association Rules — Implementation Plan

# Basket Bundle Association Rules — Implementation Plan

> For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Replace the panel section "Top cross-world SKU pairs & their bundle expansion (C/D)" with genuine 4-product basket bundles mined from cleaned baskets, plus association rules ({a,b}→{c,d}, {a,b,c}→d, a→b) showing conditional purchase confidence + lift.

Architecture: A pure-Python mining module (core/basket_bundles.py) does noise-filtering + staged Apriori (up to 4-itemsets) + association-rule generation. The snapshot refresh script (scripts/refresh_cross_family_baskets.py) calls it and persists results to JSON + XLSX. The live dashboard panel loads the latest matching snapshot JSON and renders three tables (mining is too heavy to run live).

Tech Stack: Python 3.12, psycopg2 (postgres :5433 / timescale :5434), FastAPI + Jinja2/HTMX dashboard, openpyxl, pytest (new). No new mining dependency — Apriori is hand-rolled.

---

Spec

docs/superpowers/specs/2026-05-20-basket-bundle-association-rules-design.md

Key calibrated facts (90d FR, already measured)

• 438,007 multi-line baskets; only 31,471 distinct active SKUs → SKU-grain mining is feasible.
• Noise filter size > 20 OR worlds >= 5 drops 5.4% (the "everything-reorder" baskets).
• Defaults: min_support=25, min_confidence=0.30, min_lift=1.0, max_basket_size=20, max_worlds=5.

File Structure

• Create core/basket_bundles.py — pure functions: clean_baskets, mine_frequent_itemsets, generate_rules, best_2to2_rule, build_outputs. No DB, no I/O — fully unit-testable.
• Create tests/conftest.py — put repo root on sys.path so import core... works.
• Create tests/test_basket_bundles.py — unit tests for the module.
• Modify requirements.txt — add pytest.
• Modify scripts/refresh_cross_family_baskets.py — new CLI args, Phase 5 (mining), new JSON keys, two new XLSX sheets.
• Modify dashboard/app.py — snapshot-loader helper; replace live cross_sku_pairs computation (Step 4 + Step 5, lines ~1743-1840) with snapshot bundles in the template context.
• Modify dashboard/templates/_cross_family_baskets.html — replace the C/D section with three tables (bundles-4 / rules-3→1 / rules-2→1).

Conventions to follow (from CLAUDE.md)

• Subprocess stdout that prints non-ASCII (→, ≥, €) needs PYTHONIOENCODING=utf-8; the venv is venv\Scripts\python.exe.
• Cross-DB joins are impossible in one query: order lines are on timescale, product→world on postgres — bridge in two steps (already done in the refresh script).
• Run python as a module from repo root: venv\Scripts\python.exe -m scripts..

---

Phase A — Core mining module (TDD)

Task 1: Bootstrap pytest

Files:

• Modify: requirements.txt
• Create: tests/conftest.py
• Create: tests/test_smoke.py

• [ ] Step 1: Add pytest to requirements

Append to requirements.txt:

` pytest>=8.0 `

• [ ] Step 2: Install it

Run: venv\Scripts\python.exe -m pip install pytest Expected: installs, Successfully installed pytest-...

• [ ] Step 3: Create the test path shim

Create tests/conftest.py:

`python """Put the repo root on sys.path so tests can import core... / database....""" import sys from pathlib import Path

_ROOT = Path(__file__).resolve().parent.parent if str(_ROOT) not in sys.path: sys.path.insert(0, str(_ROOT)) `

• [ ] Step 4: Add a smoke test

Create tests/test_smoke.py:

`python def test_pytest_runs(): assert 1 + 1 == 2 `

• [ ] Step 5: Run it

Run: venv\Scripts\python.exe -m pytest tests/test_smoke.py -v Expected: PASS (1 passed)

• [ ] Step 6: Commit

`bash git add requirements.txt tests/conftest.py tests/test_smoke.py git commit -m "test: bootstrap pytest harness" `

---

Task 2: clean_baskets

Files:

• Create: core/basket_bundles.py
• Test: tests/test_basket_bundles.py

• [ ] Step 1: Write the failing test

Create tests/test_basket_bundles.py:

`python from core.basket_bundles import clean_baskets

# world map: a..d -> distinct worlds; e..z -> "OFFICE" WORLD = {chr(c): "OFFICE" for c in range(ord("a"), ord("z") + 1)} WORLD.update({"a": "IT", "b": "BUREAUTIQUE", "c": "HYGIENE", "d": "MOBILIER", "e": "EHS"})

def test_clean_keeps_normal_basket(): baskets = [{"skus": ["a", "b", "c"]}] # 3 SKU, 3 worlds kept, stats = clean_baskets(baskets, WORLD, max_basket_size=20, max_worlds=5) assert kept == [frozenset({"a", "b", "c"})] assert stats["n_kept"] == 1 and stats["n_dropped"] == 0

def test_clean_drops_oversize_basket(): big = {"skus": [f"x{i}" for i in range(21)]} # 21 SKU, unmapped -> 0 worlds kept, stats = clean_baskets([big], WORLD, max_basket_size=20, max_worlds=5) assert kept == [] assert stats["dropped_size"] == 1 and stats["dropped_worlds"] == 0

def test_clean_drops_wide_basket(): wide = {"skus": ["a", "b", "c", "d", "e", "f"]} # 6 SKU, 5 distinct worlds kept, stats = clean_baskets([wide], WORLD, max_basket_size=20, max_worlds=5) assert kept == [] assert stats["dropped_worlds"] == 1 and stats["dropped_size"] == 0

def test_clean_dedups_skus(): dupe = {"skus": ["a", "a", "b"]} kept, _ = clean_baskets([dupe], WORLD, max_basket_size=20, max_worlds=5) assert kept == [frozenset({"a", "b"})] `

• [ ] Step 2: Run to verify it fails

Run: venv\Scripts\python.exe -m pytest tests/test_basket_bundles.py -v Expected: FAIL — ModuleNotFoundError: No module named 'core.basket_bundles'

• [ ] Step 3: Implement clean_baskets

Create core/basket_bundles.py:

`python """Functional basket bundles: noise filter + staged Apriori + association rules.

Pure functions, no DB / no I/O — see scripts/refresh_cross_family_baskets.py for the wiring. Method + calibration in docs/superpowers/specs/2026-05-20-basket-bundle-association-rules-design.md. """ from __future__ import annotations

from collections import Counter from itertools import combinations

def clean_baskets( baskets, # iterable of {"skus": [...]} prod_world: dict, # {sku: world} max_basket_size: int = 20, max_worlds: int = 5, ): """Drop "everything-reorder" baskets. A basket is dropped if it has more than max_basket_size distinct SKUs OR spans max_worlds or more distinct worlds. Returns (kept, stats) where kept is list[frozenset[str]].""" kept: list[frozenset] = [] n_in = dropped_size = dropped_worlds = 0 for b in baskets: n_in += 1 skus = set(b["skus"]) n_worlds = len({prod_world.get(s) for s in skus} - {None}) if len(skus) > max_basket_size: dropped_size += 1 continue if n_worlds >= max_worlds: dropped_worlds += 1 continue kept.append(frozenset(skus)) stats = { "n_in": n_in, "n_kept": len(kept), "n_dropped": n_in - len(kept), "dropped_size": dropped_size, "dropped_worlds": dropped_worlds, } return kept, stats `

• [ ] Step 4: Run to verify it passes

Run: venv\Scripts\python.exe -m pytest tests/test_basket_bundles.py -v Expected: PASS (4 passed)

• [ ] Step 5: Commit

`bash git add core/basket_bundles.py tests/test_basket_bundles.py git commit -m "feat: clean_baskets noise filter for bundle mining" `

---

Task 3: mine_frequent_itemsets

Files:

• Modify: core/basket_bundles.py
• Test: tests/test_basket_bundles.py

• [ ] Step 1: Add the failing test

Append to tests/test_basket_bundles.py:

`python from core.basket_bundles import mine_frequent_itemsets

# 4 baskets; A,B in all 4; C,D in 2; E in 1 MINE_BASKETS = [ frozenset("ABCD"), frozenset("ABCD"), frozenset("AB"), frozenset("ABE"), ]

def test_mine_supports_and_min_support(): sup = mine_frequent_itemsets(MINE_BASKETS, min_support=2, max_len=4) assert sup[frozenset("A")] == 4 assert sup[frozenset("B")] == 4 assert sup[frozenset("C")] == 2 assert sup[frozenset("D")] == 2 assert frozenset("E") not in sup # support 1 < 2 assert sup[frozenset("AB")] == 4 assert sup[frozenset("CD")] == 2 assert sup[frozenset("ABCD")] == 2

def test_mine_respects_max_len(): sup = mine_frequent_itemsets(MINE_BASKETS, min_support=2, max_len=2) assert all(len(k) <= 2 for k in sup) assert frozenset("ABCD") not in sup

def test_mine_min_support_prunes_all_pairs(): sup = mine_frequent_itemsets(MINE_BASKETS, min_support=5, max_len=4) assert sup == {} # nothing reaches support 5 `

• [ ] Step 2: Run to verify it fails

Run: venv\Scripts\python.exe -m pytest tests/test_basket_bundles.py -k mine -v Expected: FAIL — cannot import name 'mine_frequent_itemsets'

• [ ] Step 3: Implement mine_frequent_itemsets

Append to core/basket_bundles.py:

`python def mine_frequent_itemsets( baskets: list[frozenset], min_support: int = 25, max_len: int = 4, ) -> dict: """Staged Apriori. Returns {frozenset: support} for itemsets of size 1..max_len whose support (number of baskets containing all members) is >= min_support. Counts every r-combination among frequent singletons — support values are exact; no candidate pruning needed at this scale.""" c1: Counter = Counter() for b in baskets: for s in b: c1[s] += 1 keep1 = {s for s, n in c1.items() if n >= min_support} support: dict = {frozenset([s]): n for s, n in c1.items() if n >= min_support}

reduced = [sorted(b & keep1) for b in baskets] for r in range(2, max_len + 1): cr: Counter = Counter() for items in reduced: if len(items) >= r: for combo in combinations(items, r): cr[combo] += 1 any_kept = False for combo, n in cr.items(): if n >= min_support: support[frozenset(combo)] = n any_kept = True if not any_kept: break return support `

• [ ] Step 4: Run to verify it passes

Run: venv\Scripts\python.exe -m pytest tests/test_basket_bundles.py -k mine -v Expected: PASS (3 passed)

• [ ] Step 5: Commit

`bash git add core/basket_bundles.py tests/test_basket_bundles.py git commit -m "feat: staged Apriori frequent-itemset miner (up to 4-itemsets)" `

---

Task 4: generate_rules

Files:

• Modify: core/basket_bundles.py
• Test: tests/test_basket_bundles.py

• [ ] Step 1: Add the failing test

Append to tests/test_basket_bundles.py:

`python from core.basket_bundles import generate_rules

def _find(rules, ant, con): ant, con = frozenset(ant), frozenset(con) return next((r for r in rules if r["antecedent"] == ant and r["consequent"] == con), None)

def test_rules_confidence_and_lift(): sup = mine_frequent_itemsets(MINE_BASKETS, min_support=2, max_len=4) rules = generate_rules(sup, n_baskets=4, min_confidence=0.3, min_lift=1.0) # C -> D: support(CD)=2, support(C)=2 -> conf 1.0; support(D)=2 -> lift 1/(2/4)=2.0 r = _find(rules, "C", "D") assert r is not None assert abs(r["confidence"] - 1.0) < 1e-9 assert abs(r["lift"] - 2.0) < 1e-9

def test_rules_filtered_by_confidence(): sup = mine_frequent_itemsets(MINE_BASKETS, min_support=2, max_len=4) # A -> C: support(AC)=2 / support(A)=4 = 0.5 ; raise the floor above it rules = generate_rules(sup, n_baskets=4, min_confidence=0.6, min_lift=1.0) assert _find(rules, "A", "C") is None rules_lo = generate_rules(sup, n_baskets=4, min_confidence=0.3, min_lift=1.0) assert _find(rules_lo, "A", "C") is not None `

• [ ] Step 2: Run to verify it fails

Run: venv\Scripts\python.exe -m pytest tests/test_basket_bundles.py -k rules -v Expected: FAIL — cannot import name 'generate_rules'

• [ ] Step 3: Implement generate_rules

Append to core/basket_bundles.py:

`python def generate_rules( support: dict, n_baskets: int, min_confidence: float = 0.30, min_lift: float = 1.0, ) -> list: """For each frequent itemset S (size >= 2) and each non-empty proper subset A (antecedent), consequent = S - A. confidence = sup(S)/sup(A); lift = confidence / (sup(consequent)/n_baskets). Keeps rules meeting both floors. All subsets of a frequent itemset are themselves frequent (anti- monotonicity), so sup(A) and sup(consequent) are always present.""" rules: list = [] for S, sup_S in support.items(): if len(S) < 2: continue items = sorted(S) for r in range(1, len(S)): for ant_t in combinations(items, r): A = frozenset(ant_t) C = S - A sup_A = support.get(A) sup_C = support.get(C) if not sup_A or not sup_C: continue conf = sup_S / sup_A lift = conf / (sup_C / n_baskets) if conf >= min_confidence and lift >= min_lift: rules.append({ "antecedent": A, "consequent": C, "support": sup_S, "confidence": conf, "lift": lift, }) return rules `

• [ ] Step 4: Run to verify it passes

Run: venv\Scripts\python.exe -m pytest tests/test_basket_bundles.py -k rules -v Expected: PASS (2 passed)

• [ ] Step 5: Commit

`bash git add core/basket_bundles.py tests/test_basket_bundles.py git commit -m "feat: association-rule generation with confidence + lift" `

---

Task 5: best_2to2_rule + build_outputs

Files:

• Modify: core/basket_bundles.py
• Test: tests/test_basket_bundles.py

• [ ] Step 1: Add the failing test

Append to tests/test_basket_bundles.py:

`python from core.basket_bundles import best_2to2_rule, build_outputs

META = { "A": {"name": "Cutter blade", "section": "S_CUT", "world": "EHS"}, "B": {"name": "Cutter holder", "section": "S_CUT", "world": "EHS"}, "C": {"name": "Cardboard box", "section": "S_BOX", "world": "LOGISTIQUE"}, "D": {"name": "Tape", "section": "S_TAPE", "world": "LOGISTIQUE"}, }

def test_best_2to2_picks_max_confidence(): sup = mine_frequent_itemsets(MINE_BASKETS, min_support=2, max_len=4) br = best_2to2_rule(frozenset("ABCD"), sup, n_baskets=4) assert br is not None assert len(br["antecedent"]) == 2 and len(br["consequent"]) == 2 # max possible 2->2 confidence here is 1.0 (e.g. {A,C}->{B,D}) assert abs(br["confidence"] - 1.0) < 1e-9

def test_build_outputs_shapes_and_enrichment(): sup = mine_frequent_itemsets(MINE_BASKETS, min_support=2, max_len=4) rules = generate_rules(sup, n_baskets=4, min_confidence=0.3, min_lift=1.0) out = build_outputs(sup, rules, META, n_baskets=4) assert set(out) == {"bundles_4", "rules_3to1", "rules_2to1"} assert len(out["bundles_4"]) == 1 bundle = out["bundles_4"][0] assert [it["sku"] for it in bundle["items"]] == ["A", "B", "C", "D"] assert bundle["items"][0]["name"] == "Cutter blade" assert bundle["items"][0]["world"] == "EHS" assert bundle["support"] == 2 assert bundle["best_rule"] is not None # secondary table = 3->1 rules; tertiary = 2->1 rules assert all(len(r["antecedent"]) == 3 and len(r["consequent"]) == 1 for r in out["rules_3to1"]) assert all(len(r["antecedent"]) == 2 and len(r["consequent"]) == 1 for r in out["rules_2to1"]) `

• [ ] Step 2: Run to verify it fails

Run: venv\Scripts\python.exe -m pytest tests/test_basket_bundles.py -k "best_2to2 or build_outputs" -v Expected: FAIL — cannot import name 'best_2to2_rule'

• [ ] Step 3: Implement both functions

Append to core/basket_bundles.py:

`python def best_2to2_rule(quad: frozenset, support: dict, n_baskets: int): """Among the 2->2 splits of a 4-itemset, return the rule with the highest confidence (dict with antecedent/consequent/support/confidence/lift), or None if quad is not a frequent 4-itemset.""" if len(quad) != 4: return None sup_S = support.get(quad) if sup_S is None: return None items = sorted(quad) best = None for ant_t in combinations(items, 2): A = frozenset(ant_t) C = quad - A sup_A = support.get(A) sup_C = support.get(C) if not sup_A or not sup_C: continue conf = sup_S / sup_A lift = conf / (sup_C / n_baskets) if best is None or conf > best["confidence"]: best = {"antecedent": A, "consequent": C, "support": sup_S, "confidence": conf, "lift": lift} return best

def build_outputs( support: dict, rules: list, prod_meta: dict, n_baskets: int, top_bundles: int = 50, top_rules: int = 50, ) -> dict: """Assemble the three display sections, each enriched with name/section/ world per SKU:

• bundles_4 : 4-itemsets ranked by support, each with its best 2->2 rule
• rules_3to1: {a,b,c} -> d, filtered rules, ranked by lift
• rules_2to1: {a,b} -> c, filtered rules, ranked by lift

def rule_obj(rule: dict) -> dict: return { "antecedent": [item_obj(s) for s in sorted(rule["antecedent"])], "consequent": [item_obj(s) for s in sorted(rule["consequent"])], "support": rule["support"], "confidence": round(rule["confidence"], 3), "lift": round(rule["lift"], 2), }

quads = sorted(((S, n) for S, n in support.items() if len(S) == 4), key=lambda x: -x[1])[:top_bundles] bundles_4 = [] for S, sup in quads: br = best_2to2_rule(S, support, n_baskets) bundles_4.append({ "items": [item_obj(s) for s in sorted(S)], "support": sup, "support_pct": round(sup / n_baskets * 100, 3) if n_baskets else 0.0, "best_rule": None if br is None else { "antecedent": [item_obj(s) for s in sorted(br["antecedent"])], "consequent": [item_obj(s) for s in sorted(br["consequent"])], "confidence": round(br["confidence"], 3), "lift": round(br["lift"], 2), }, })

r3 = sorted((r for r in rules if len(r["antecedent"]) == 3 and len(r["consequent"]) == 1), key=lambda r: -r["lift"])[:top_rules] r2 = sorted((r for r in rules if len(r["antecedent"]) == 2 and len(r["consequent"]) == 1), key=lambda r: -r["lift"])[:top_rules] return { "bundles_4": bundles_4, "rules_3to1": [rule_obj(r) for r in r3], "rules_2to1": [rule_obj(r) for r in r2], } `

• [ ] Step 4: Run the full module test suite

Run: venv\Scripts\python.exe -m pytest tests/test_basket_bundles.py -v Expected: PASS (all tests, ~11 passed)

• [ ] Step 5: Commit

`bash git add core/basket_bundles.py tests/test_basket_bundles.py git commit -m "feat: bundle output assembly (best 2->2 + 3->1/2->1 rule tables)" `

---

Phase B — Snapshot refresh integration

Task 6: Wire mining into refresh_cross_family_baskets.py (compute + JSON)

Files:

• Modify: scripts/refresh_cross_family_baskets.py

• [ ] Step 1: Add CLI args

In main() (after the existing --min-co arg, around line 583), add:

`python ap.add_argument("--min-support", type=int, default=25, dest="min_support", help="min baskets for a frequent itemset (default 25)") ap.add_argument("--min-confidence", type=float, default=0.30, dest="min_confidence", help="min rule confidence (default 0.30)") ap.add_argument("--min-lift", type=float, default=1.0, dest="min_lift", help="min rule lift (default 1.0)") ap.add_argument("--max-basket-size", type=int, default=20, dest="max_basket_size", help="drop baskets with more distinct SKUs (default 20)") ap.add_argument("--max-worlds", type=int, default=5, dest="max_worlds", help="drop baskets spanning >= this many worlds (default 5)") `

• [ ] Step 2: Thread the args into compute(...)

Change the compute signature (line 57) from:

`python def compute(days: int, channel: str, min_co: int) -> dict: `

to:

`python def compute(days: int, channel: str, min_co: int, min_support: int = 25, min_confidence: float = 0.30, min_lift: float = 1.0, max_basket_size: int = 20, max_worlds: int = 5) -> dict: `

And update the call in main() (line 591) from compute(args.days, args.channel, args.min_co) to:

`python data = compute(args.days, args.channel, args.min_co, min_support=args.min_support, min_confidence=args.min_confidence, min_lift=args.min_lift, max_basket_size=args.max_basket_size, max_worlds=args.max_worlds) `

• [ ] Step 3: Add the import

Near the top imports of the file (after the from database... imports, ~line 43), add:

`python from core.basket_bundles import ( clean_baskets, mine_frequent_itemsets, generate_rules, build_outputs, ) `

• [ ] Step 4: Add Phase 5 (mining) before the return in compute

Immediately before the return { statement (line 388), insert:

`python step("Phase 5/5 — mining functional bundles + association rules…") kept, clean_stats = clean_baskets( order_baskets, prod_world, max_basket_size=max_basket_size, max_worlds=max_worlds) n_kept = len(kept) support = mine_frequent_itemsets(kept, min_support=min_support, max_len=4) rules = generate_rules(support, n_kept, min_confidence=min_confidence, min_lift=min_lift)

# Enrich SKUs that appear in any frequent itemset with name + section. # prod_world already covers world; we still need description + section_code. bundle_skus = sorted({s for itemset in support for s in itemset}) prod_meta: dict[str, dict] = {} for start in range(0, len(bundle_skus), 10000): for r in crm.query(""" SELECT p.product_reference, p.product_description, p.section_code, t.world FROM ecom_products p JOIN section_taxonomy t ON p.section_code = t.section_code AND p.source_country = t.source_country WHERE p.source_country = %s AND p.product_reference = ANY(%s) """, (COUNTRY, bundle_skus[start:start + 10000])): prod_meta[r["product_reference"]] = { "name": r["product_description"], "section": r["section_code"], "world": r["world"], } bundle_outputs = build_outputs(support, rules, prod_meta, n_kept) step(f" → {len(bundle_outputs['bundles_4'])} bundles · " f"{len(bundle_outputs['rules_3to1'])} 3→1 · " f"{len(bundle_outputs['rules_2to1'])} 2→1 " f"(kept {n_kept:,}/{clean_stats['n_in']:,} baskets)") `

• [ ] Step 5: Add the new keys to the returned dict

Inside the return { dict, after the existing "cross_sku_pairs": [...] entry (line ~418-421), add:

`python "bundle_params": { "min_support": min_support, "min_confidence": min_confidence, "min_lift": min_lift, "max_basket_size": max_basket_size, "max_worlds": max_worlds, }, "bundle_clean_stats": clean_stats, "bundles_4": bundle_outputs["bundles_4"], "rules_3to1": bundle_outputs["rules_3to1"], "rules_2to1": bundle_outputs["rules_2to1"], `

• [ ] Step 6: Smoke-test the import + arg parsing (no DB hit yet)

Run: venv\Scripts\python.exe -c "import scripts.refresh_cross_family_baskets as m; print('import ok')" Expected: import ok (verifies the core.basket_bundles import resolves)

• [ ] Step 7: Commit

`bash git add scripts/refresh_cross_family_baskets.py git commit -m "feat: mine bundles + rules in cross-family refresh (Phase 5 + JSON)" `

---

Task 7: New XLSX sheets (bundles_4, bundle_rules)

Files:

• Modify: scripts/refresh_cross_family_baskets.py (the write_xlsx function)

• [ ] Step 1: Add the two sheets

In write_xlsx(...), just before wb.save(out_path) (line 576), insert:

`python # 7. functional bundles of 4 (with best 2->2 decomposition rule) def _names(items): return " + ".join((it["name"] or "")[:22] for it in items)

def _rule_str(rule): if not rule: return "" a = " + ".join(it["sku"] for it in rule["antecedent"]) c = " + ".join(it["sku"] for it in rule["consequent"]) return f"{a} → {c} (conf {rule['confidence']:.2f}, lift {rule['lift']:.2f})"

add("bundles_4", ["sku_a", "sku_b", "sku_c", "sku_d", "products", "worlds", "support", "support_%", "best_2to2_rule"], [[*[it["sku"] for it in b["items"]], _names(b["items"]), " / ".join(it["world"] for it in b["items"]), b["support"], b["support_pct"], _rule_str(b["best_rule"])] for b in data["bundles_4"]], [14, 14, 14, 14, 50, 30, 10, 10, 52])

# 8. association rules (3->1 then 2->1) def _rule_rows(rows, kind): out = [] for r in rows: out.append([ kind, " + ".join(it["sku"] for it in r["antecedent"]), " + ".join((it["name"] or "")[:22] for it in r["antecedent"]), r["consequent"][0]["sku"], (r["consequent"][0]["name"] or "")[:30], r["support"], round(r["confidence"], 3), round(r["lift"], 2), ]) return out

add("bundle_rules", ["type", "antecedent_skus", "antecedent_names", "consequent_sku", "consequent_name", "support", "confidence", "lift"], _rule_rows(data["rules_3to1"], "3→1") + _rule_rows(data["rules_2to1"], "2→1"), [8, 26, 40, 16, 32, 10, 12, 10]) `

• [ ] Step 2: Run the full refresh on real data (90d FR)

Run: set PYTHONIOENCODING=utf-8 && venv\Scripts\python.exe -m scripts.refresh_cross_family_baskets --days 90 (PowerShell: $env:PYTHONIOENCODING="utf-8"; venv\Scripts\python.exe -m scripts.refresh_cross_family_baskets --days 90) Expected: completes; prints the Phase 5 line with non-zero bundles; writes exports/cross_family_baskets_FR__d90_ALL.xlsx + .json.

• [ ] Step 3: Sanity-check the JSON output

Run: venv\Scripts\python.exe -c "import json,glob,os; f=max(glob.glob('exports/cross_family_baskets_FR_*_d90_ALL.json'), key=os.path.getmtime); d=json.load(open(f,encoding='utf-8')); print('bundles', len(d['bundles_4'])); print('3to1', len(d['rules_3to1'])); print('2to1', len(d['rules_2to1'])); print('kept', d['bundle_clean_stats'])" Expected: non-zero bundles, and kept showing ~5% dropped. Eyeball the first bundle: venv\Scripts\python.exe -c "import json,glob,os; f=max(glob.glob('exports/cross_family_baskets_FR_*_d90_ALL.json'), key=os.path.getmtime); d=json.load(open(f,encoding='utf-8')); import pprint; pprint.pprint(d['bundles_4'][0])" — confirm 4 items with names/worlds and a best_rule.

• [ ] Step 4: Note the Phase 5 runtime

If the refresh's Phase 5 step takes more than ~3 minutes, record it in the spec's "risque perf" note and consider adding anti-monotone pruning to mine_frequent_itemsets (count an r-combo only when all its (r-1)-subsets are frequent). Otherwise leave as-is.

• [ ] Step 5: Commit

`bash git add scripts/refresh_cross_family_baskets.py git commit -m "feat: bundles_4 + bundle_rules XLSX sheets" `

---

Phase C — Dashboard panel

Task 8: Load latest snapshot bundles in the panel route

Files:

• Modify: dashboard/app.py

• [ ] Step 1: Add a snapshot-loader helper

Just above the @app.get("/panel/cross-family-baskets"...) decorator (line 1398), add:

`python import glob as _glob import json as _json

def _load_latest_cfb_bundles(days: int, channel: str) -> dict: """Load the bundles/rules section from the most recent matching cross-family snapshot. Mining is offline (refresh script); the panel only reads it. Returns {} when no snapshot exists for this (days, channel).""" pattern = str(_AC_EXPORTS_DIR / f"cross_family_baskets_FR_*_d{days}_{channel}.json") matches = _glob.glob(pattern) if not matches: return {} latest = max(matches, key=_os.path.getmtime) try: d = _json.loads(Path(latest).read_text(encoding="utf-8")) except Exception: return {} return { "bundles_4": d.get("bundles_4", []), "rules_3to1": d.get("rules_3to1", []), "rules_2to1": d.get("rules_2to1", []), "bundle_params": d.get("bundle_params", {}), "bundle_clean_stats": d.get("bundle_clean_stats", {}), "bundle_snapshot_at": (d.get("computed_at") or "")[:19], } `

Note: _AC_EXPORTS_DIR is defined at line ~4891 — if the helper is placed before that definition, instead inline the path: Path(__file__).resolve().parent.parent / "exports". Verify which symbol is in scope; prefer the existing _AC_EXPORTS_DIR if available, else use the inline Path(...). Confirm _os / os import name in the file (import os → use os.path.getmtime).

• [ ] Step 2: Remove the live cross_sku_pairs computation

Delete Step 4 + Step 5 (lines ~1743-1840): the cross_sku_pairs = crm.query(...) block, the sku_baskets companion loop, pair_companions, companion_meta, and cross_sku_pairs_enriched. These are replaced by the snapshot bundles.

• [ ] Step 3: Swap the template context

In the templates.TemplateResponse(...) context dict (line ~1844), remove the "cross_sku_pairs": cross_sku_pairs, entry and add:

`python _load_latest_cfb_bundles(days, channel), `

• [ ] Step 4: Restart the dashboard service + verify it boots

Run: nssm restart LeadContagionDashboard Then: venv\Scripts\python.exe -c "import urllib.request; print(urllib.request.urlopen('http://127.0.0.1:8001/healthz').status)" (If there is no /healthz, request / — expect 401 from Basic Auth, which still proves the app booted.) Expected: the panel route imports cleanly (no 500 on /panel/cross-family-baskets).

• [ ] Step 5: Commit

`bash git add dashboard/app.py git commit -m "feat: panel loads bundles from latest snapshot, drops live pair query" `

---

Task 9: Render the three tables in the template

Files:

• Modify: dashboard/templates/_cross_family_baskets.html

• [ ] Step 1: Replace the C/D section

Replace the entire

`html {% if bundles_4 %}

• [ ] Step 3: Update the footer note

In the closing

footer (line ~631-638), replace the sentence beginning "Pair grain = product_co_occurrence..." with:

`html Bundles & règles = itemsets de 1 à 4 SKU minés (Apriori) sur les paniers FR nettoyés de la fenêtre, snapshot via scripts/refresh_cross_family_baskets.py. `

• [ ] Step 4: Restart + visual check

Run: nssm restart LeadContagionDashboard Open http://127.0.0.1:8001/ → Customers/Journey tab containing the cross-family-baskets panel → confirm: the "Bundles fonctionnels — 4 produits" table renders with pills + support + 2→2 rule, and the 3→1 / 2→1 tables appear below. Toggle the 90d window to confirm it matches the snapshot just generated.

• [ ] Step 5: Commit

`bash git add dashboard/templates/_cross_family_baskets.html git commit -m "feat: render bundles-4 + 3->1/2->1 rule tables in panel" `

---

Final verification

• [ ] Full test suite green

Run: venv\Scripts\python.exe -m pytest tests/ -v Expected: all pass.

• [ ] End-to-end snapshot + panel match

Re-run the refresh for 90d, reload the panel, confirm the displayed bundles match the JSON (bundles_4[0]). Confirm the noise-filter caption shows the params.

• [ ] No regressions in the rest of the panel

Confirm the world matrix, top world pairs/triples/quads, party-dimension tables, and sold-to/ship-to tables all still render (they were untouched).

---

Self-review notes (addressed)

• Spec coverage: grain SKU-exact ✓ (no cross-world constraint applied in mining);

• §5.3 open item resolved: panel reads latest snapshot JSON (no live mining);

• Type consistency: rule dicts always carry antecedent/consequent as