Compléments fonctionnels par IA — Implementation Plan

# Compléments fonctionnels par IA — 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: Déduire par IA (titre+marque, sales-blind) les types de produits fonctionnellement complémentaires, puis croiser avec le co-achat réel pour isoler les paires complémentaires mais rarement achetées ensemble (cross-sell sous-exploité), livrées en Excel pour Marie.

Architecture: Une couche pure et testable (core/functional_complements.py) fait tout le calcul non-IA : chargement CSV, agrégation co-achat au grain type, scoring/bucketing, assemblage des sorties. Une couche IA isolée et injectable (core/ai_typing.py) encapsule les appels LLM (typage + graphe de compléments) sur le pattern existant de core/labeling.py. Un script d'orchestration (scripts/refresh_functional_complements.py) câble le tout en réutilisant le pull de paniers et clean_baskets du refresh cross-family existant, et écrit un snapshot XLSX + JSON.

Tech Stack: Python 3.12 (.venv), pytest, SDK anthropic (modèle claude-haiku-4-5), openpyxl, psycopg2 (via database.crm_db / database.timeseries_db).

Décisions par défaut (modifiables) :

• Persistance v1 = fichiers snapshot uniquement (JSON + XLSX). Les tables postgres réutilisables (product_functional_type, type_complement_edge) sont une Task optionnelle (Task 9), non requise pour le livrable Marie.
• Vocabulaire de types = construit à la volée (réutiliser-avant-créer). Pré-seed = amélioration ultérieure.

Référence spec : docs/superpowers/specs/2026-06-02-functional-complements-ai-design.md

---

File Structure

| Fichier | Responsabilité | |---|---| | core/functional_complements.py (créer) | Fonctions pures : load_csv_products, type_copurchase, score_opportunities, build_complement_outputs. Aucun appel réseau/DB. | | core/ai_typing.py (créer) | Couche IA injectable : type_products, propose_complements, _default_llm_call. Encapsule le SDK anthropic + cache + vocab contrôlé. | | scripts/refresh_functional_complements.py (créer) | Orchestration offline : pull paniers (réutilise la requête du refresh existant) → typage → graphe → co-achat → score → XLSX + JSON. | | tests/test_functional_complements.py (créer) | Tests des 4 fonctions pures (valeurs exactes). | | tests/test_ai_typing.py (créer) | Tests de la couche IA avec llm_call stubbé (zéro réseau). |

Conventions verrouillées (utilisées par toutes les tasks) :

• Un type est une chaîne normalisée : type = label.strip().lower(). Le type EST son id (pas de dict type_id/type_label — simplifié vs spec §5.1 pour un core plus propre).
• Une paire de types est toujours stockée/consultée comme tuple(sorted((t1, t2))).
• sku_type_map: dict[str, str] = {sku: type}.
• baskets passés aux fonctions pures = list[frozenset[str]] (sortie de clean_baskets).

---

Task 1: load_csv_products — lecture du CSV de Marie (cp1252)

Files:

• Create: core/functional_complements.py
• Test: tests/test_functional_complements.py

• [ ] Step 1: Write the failing test

`python # tests/test_functional_complements.py import csv from pathlib import Path

from core.functional_complements import load_csv_products

def _write_csv(path: Path) -> None: # cp1252 + ';' delimiter + BOM, mirroring imports/marie/Datas_pour_produits_associes.csv rows = [ ["SAP Code (.)", "SAP Code", "Titre web", "Marque", "Code fabricant", "GTIN 1", "GTIN 2", "GTIN 3", "GTIN 4", "GTIN 5"], ["5.978.056", "5978056", "Assortiment de Mini Toblerone - boîte de 904 g", "TOBLERONE", "912490", "7622210418654", "", "", "", ""], ["1.516.588", "1516588", "Corbeille à courrier Cep First - noire", "CEP", "x", "1", "", "", "", ""], ] with open(path, "w", encoding="cp1252", newline="") as f: w = csv.writer(f, delimiter=";") for r in rows: w.writerow(r)

def test_load_csv_products(tmp_path): p = tmp_path / "marie.csv" _write_csv(p) out = load_csv_products(p) assert out["5978056"] == { "title": "Assortiment de Mini Toblerone - boîte de 904 g", "brand": "TOBLERONE", } assert out["1516588"]["title"] == "Corbeille à courrier Cep First - noire" # accents survive cp1252 round-trip assert "à" in out["1516588"]["title"] # keyed by SAP Code (no dots), 2 products assert set(out) == {"5978056", "1516588"} `

• [ ] Step 2: Run test to verify it fails

Run: .venv\Scripts\python.exe -m pytest tests/test_functional_complements.py::test_load_csv_products -v Expected: FAIL — ImportError: cannot import name 'load_csv_products'

• [ ] Step 3: Write minimal implementation

`python # core/functional_complements.py """Compléments fonctionnels par IA × croisement ventes — fonctions pures.

Aucun I/O réseau ni DB. Méthode + calibration : docs/superpowers/specs/2026-06-02-functional-complements-ai-design.md """ from __future__ import annotations

import csv from math import log1p from pathlib import Path

def load_csv_products(path, encoding: str = "cp1252") -> dict: """Lit le CSV de Marie (délimiteur ';', encodage cp1252).

Renvoie {sku: {"title": str, "brand": str}}, sku = colonne 'SAP Code' (sans points). Les lignes sans SAP Code ou sans titre sont ignorées. """ out: dict[str, dict] = {} with open(path, "r", encoding=encoding, newline="") as f: reader = csv.DictReader(f, delimiter=";") for row in reader: sku = (row.get("SAP Code") or "").strip() title = (row.get("Titre web") or "").strip() if not sku or not title: continue out[sku] = {"title": title, "brand": (row.get("Marque") or "").strip()} return out `

• [ ] Step 4: Run test to verify it passes

Run: .venv\Scripts\python.exe -m pytest tests/test_functional_complements.py::test_load_csv_products -v Expected: PASS

• [ ] Step 5: Commit

`bash git add core/functional_complements.py tests/test_functional_complements.py git commit -m "feat(complements-ia): load_csv_products (cp1252 reader)" `

---

Task 2: type_copurchase — co-achat au grain type

Files:

• Modify: core/functional_complements.py
• Test: tests/test_functional_complements.py

• [ ] Step 1: Write the failing test

`python # tests/test_functional_complements.py (append) from core.functional_complements import type_copurchase

def test_type_copurchase_exact(): baskets = [ frozenset({"s1", "s2"}), # souris, pile frozenset({"s1", "s3"}), # souris, tapis frozenset({"s1", "s2", "s4"}), # souris, pile, pile ] sku_type = {"s1": "souris", "s2": "pile", "s3": "tapis", "s4": "pile"} cp = type_copurchase(baskets, sku_type, min_type_support=1)

assert cp["N"] == 3 assert cp["n"] == {"souris": 3, "pile": 2, "tapis": 1} # pairs stored sorted assert cp["co"][("pile", "souris")] == 2 assert cp["co"][("souris", "tapis")] == 1 assert ("pile", "tapis") not in cp["co"] # never co-occur assert cp["lift"][("pile", "souris")] == 1.0 assert cp["cond"][("pile", "souris")] == 1.0 assert cp["cond"][("souris", "tapis")] == 1.0 `

• [ ] Step 2: Run test to verify it fails

Run: .venv\Scripts\python.exe -m pytest tests/test_functional_complements.py::test_type_copurchase_exact -v Expected: FAIL — ImportError: cannot import name 'type_copurchase'

• [ ] Step 3: Write minimal implementation

`python # core/functional_complements.py (append) from collections import Counter from itertools import combinations

def type_copurchase(baskets, sku_type_map: dict, min_type_support: int = 25) -> dict: """Agrège le co-achat au grain type sur des paniers nettoyés.

baskets : list[frozenset[str]] (SKU). sku_type_map : {sku: type}. Renvoie {N, n, co, lift, cond} où : n[t] = nb de paniers contenant >=1 SKU de type t co[(a,b)] = nb de paniers contenant a la fois >=1 SKU de a et de b (a0). min_type_support n'est PAS filtré ici (le scoring s'en charge) — il est accepté pour homogénéité de signature. """ N = len(baskets) n: Counter = Counter() co: Counter = Counter() for b in baskets: types = {sku_type_map.get(s) for s in b} types.discard(None) for t in types: n[t] += 1 for a, c in combinations(sorted(types), 2): co[(a, c)] += 1

lift: dict = {} cond: dict = {} for (a, c), k in co.items(): if N and n[a] and n[c]: lift[(a, c)] = (k / N) / ((n[a] / N) * (n[c] / N)) cond[(a, c)] = k / min(n[a], n[c]) return {"N": N, "n": dict(n), "co": dict(co), "lift": lift, "cond": cond} `

• [ ] Step 4: Run test to verify it passes

Run: .venv\Scripts\python.exe -m pytest tests/test_functional_complements.py::test_type_copurchase_exact -v Expected: PASS

• [ ] Step 5: Commit

`bash git add core/functional_complements.py tests/test_functional_complements.py git commit -m "feat(complements-ia): type_copurchase (n/co/lift/cond at type grain)" `

---

Task 3: score_opportunities — score + buckets

Files:

• Modify: core/functional_complements.py
• Test: tests/test_functional_complements.py

Buckets (spec §4.5) : co == 0 → ia_seule ; co > 0 et lift < lift_low → sous_exploite ; co > 0 et lift >= lift_low → confirme. Garde-fou : on écarte toute arête dont un type a n < min_type_support.

• [ ] Step 1: Write the failing test

`python # tests/test_functional_complements.py (append) import pytest from core.functional_complements import score_opportunities

def test_score_opportunities_buckets_and_order(): copurchase = { "N": 100, "n": {"souris": 40, "pile": 50, "tapis": 30, "clavier": 26}, "co": {("pile", "souris"): 25, ("souris", "tapis"): 2}, "lift": {("pile", "souris"): 1.25, ("souris", "tapis"): 0.17}, "cond": {("pile", "souris"): 0.625, ("souris", "tapis"): 0.067}, } edges = [ {"t1": "pile", "t2": "souris", "relation": "consommable-de", "ai_strength": 0.9, "rationale": "la souris sans-fil a besoin de piles"}, {"t1": "souris", "t2": "tapis", "relation": "accessoire-de", "ai_strength": 0.8, "rationale": "tapis pour la souris"}, {"t1": "clavier", "t2": "pile", "relation": "consommable-de", "ai_strength": 0.7, "rationale": "clavier sans-fil a besoin de piles"}, {"t1": "souris", "t2": "rare", "relation": "accessoire-de", "ai_strength": 0.6, "rationale": "type sans assez d'acheteurs"}, ] out = score_opportunities(edges, copurchase, min_type_support=25, lift_low=1.0)

by_pair = {tuple(sorted((r["t1"], r["t2"]))): r for r in out} # 'rare' has n=0 < support -> dropped assert ("rare", "souris") not in by_pair assert by_pair[("pile", "souris")]["bucket"] == "confirme" # lift 1.25 >= 1 assert by_pair[("souris", "tapis")]["bucket"] == "sous_exploite" # lift 0.17 < 1 assert by_pair[("clavier", "pile")]["bucket"] == "ia_seule" # co absent -> 0

# scores assert by_pair[("souris", "tapis")]["score"] == pytest.approx( 0.8 (3.4339872) (1 - 0.067), rel=1e-3) # sorted by score desc scores = [r["score"] for r in out] assert scores == sorted(scores, reverse=True) `

• [ ] Step 2: Run test to verify it fails

Run: .venv\Scripts\python.exe -m pytest tests/test_functional_complements.py::test_score_opportunities_buckets_and_order -v Expected: FAIL — ImportError: cannot import name 'score_opportunities'

• [ ] Step 3: Write minimal implementation

`python # core/functional_complements.py (append) def score_opportunities(complement_edges, copurchase: dict, min_type_support: int = 25, lift_low: float = 1.0) -> list: """Jointe les arêtes IA × co-achat → score + bucket. Pur.

Écarte toute arête dont un type a n < min_type_support. Pour chaque arête conservée : co/lift/cond lus dans copurchase (défaut 0 si absent), volume = min(n1, n2), score = ai_strength log1p(volume) (1 - cond). Bucket : co==0 -> 'ia_seule' ; lift 'sous_exploite' ; sinon 'confirme'. Renvoie la liste triée par score décroissant. """ n = copurchase["n"] co = copurchase["co"] lift = copurchase["lift"] cond = copurchase["cond"] out: list = [] for e in complement_edges: key = tuple(sorted((e["t1"], e["t2"]))) n1, n2 = n.get(key[0], 0), n.get(key[1], 0) if n1 < min_type_support or n2 < min_type_support: continue k = co.get(key, 0) lf = lift.get(key, 0.0) cd = cond.get(key, 0.0) volume = min(n1, n2) score = e["ai_strength"] log1p(volume) (1.0 - cd) if k == 0: bucket = "ia_seule" elif lf < lift_low: bucket = "sous_exploite" else: bucket = "confirme" out.append({ "t1": key[0], "t2": key[1], "relation": e.get("relation"), "ai_strength": e["ai_strength"], "rationale": e.get("rationale"), "n1": n1, "n2": n2, "co": k, "lift": round(lf, 3), "cond": round(cd, 3), "score": score, "bucket": bucket, }) out.sort(key=lambda r: -r["score"]) return out `

• [ ] Step 4: Run test to verify it passes

Run: .venv\Scripts\python.exe -m pytest tests/test_functional_complements.py::test_score_opportunities_buckets_and_order -v Expected: PASS

• [ ] Step 5: Commit

`bash git add core/functional_complements.py tests/test_functional_complements.py git commit -m "feat(complements-ia): score_opportunities (score + 3 buckets)" `

---

Task 4: build_complement_outputs — sections d'affichage enrichies

Files:

• Modify: core/functional_complements.py
• Test: tests/test_functional_complements.py

• [ ] Step 1: Write the failing test

`python # tests/test_functional_complements.py (append) from core.functional_complements import build_complement_outputs

def test_build_complement_outputs_groups_and_reps(): scored = [ {"t1": "souris", "t2": "tapis", "relation": "accessoire-de", "ai_strength": 0.8, "rationale": "r", "n1": 40, "n2": 30, "co": 2, "lift": 0.17, "cond": 0.067, "score": 2.5, "bucket": "sous_exploite"}, {"t1": "pile", "t2": "souris", "relation": "consommable-de", "ai_strength": 0.9, "rationale": "r2", "n1": 50, "n2": 40, "co": 25, "lift": 1.25, "cond": 0.625, "score": 1.2, "bucket": "confirme"}, ] sku_type_map = {"s1": "souris", "s2": "pile", "s3": "tapis", "s4": "souris"} sku_sales = {"s1": 100, "s2": 50, "s3": 30, "s4": 5} prod_meta = { "s1": {"name": "Souris sans fil Logitech"}, "s2": {"name": "Pile AA Energizer"}, "s3": {"name": "Tapis souris"}, "s4": {"name": "Souris filaire Lyreco"}, } out = build_complement_outputs(scored, sku_type_map, sku_sales, prod_meta, top_per_type=2)

assert len(out["sous_exploite"]) == 1 assert len(out["confirme"]) == 1 assert out["ia_seule"] == [] # representative SKUs for 'souris' = top-2 by sales: s1 (100) then s4 (5) row = out["sous_exploite"][0] rep_skus = [r["sku"] for r in row["reps_t1"]] # t1 == 'souris' assert rep_skus == ["s1", "s4"] assert row["reps_t1"][0]["name"] == "Souris sans fil Logitech" # typage + graphe present assert len(out["typage"]) == 4 assert len(out["graphe"]) == 2 `

• [ ] Step 2: Run test to verify it fails

Run: .venv\Scripts\python.exe -m pytest tests/test_functional_complements.py::test_build_complement_outputs_groups_and_reps -v Expected: FAIL — ImportError: cannot import name 'build_complement_outputs'

• [ ] Step 3: Write minimal implementation

`python # core/functional_complements.py (append) from collections import defaultdict

def _type_reps(sku_type_map: dict, sku_sales: dict, prod_meta: dict, top_per_type: int) -> dict: """type -> [{sku, name}] des top_per_type SKU les plus vendus du type.""" by_type: dict[str, list] = defaultdict(list) for sku, t in sku_type_map.items(): by_type[t].append(sku) reps: dict[str, list] = {} for t, skus in by_type.items(): ranked = sorted(skus, key=lambda s: -sku_sales.get(s, 0))[:top_per_type] reps[t] = [{"sku": s, "name": (prod_meta.get(s) or {}).get("name") or "—"} for s in ranked] return reps

def build_complement_outputs(scored, sku_type_map: dict, sku_sales: dict, prod_meta: dict, top_per_type: int = 3) -> dict: """Groupe scored par bucket, attache les SKU représentatifs par type, et ajoute les sections d'audit typage et graphe. Pur.""" reps = _type_reps(sku_type_map, sku_sales, prod_meta, top_per_type) buckets: dict[str, list] = {"sous_exploite": [], "confirme": [], "ia_seule": []} graphe: list = [] for r in scored: row = dict(r) row["reps_t1"] = reps.get(r["t1"], []) row["reps_t2"] = reps.get(r["t2"], []) buckets.setdefault(r["bucket"], []).append(row) graphe.append({"t1": r["t1"], "t2": r["t2"], "relation": r["relation"], "ai_strength": r["ai_strength"], "rationale": r["rationale"]}) typage = [{"sku": s, "type": t, "name": (prod_meta.get(s) or {}).get("name") or "—"} for s, t in sorted(sku_type_map.items())] return {buckets, "typage": typage, "graphe": graphe} `

• [ ] Step 4: Run test to verify it passes

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

• [ ] Step 5: Commit

`bash git add core/functional_complements.py tests/test_functional_complements.py git commit -m "feat(complements-ia): build_complement_outputs (buckets + reps + audit)" `

---

Task 5: core/ai_typing.py — type_products (batché, cache, vocab contrôlé)

Files:

• Create: core/ai_typing.py
• Test: tests/test_ai_typing.py

llm_call(prompt: str) -> str est injectable (renvoie du JSON brut). En test on injecte un faux ; en prod, _default_llm_call utilise le SDK anthropic (Task 7). Le cache est un dict {hash(title+brand): type} persistable en JSON.

• [ ] Step 1: Write the failing test

`python # tests/test_ai_typing.py import json from core.ai_typing import type_products

def test_type_products_reuses_vocab_and_caches(): calls = []

def fake_llm(prompt: str) -> str: calls.append(prompt) # the fake returns one type per sku found in the prompt's JSON payload # (the impl passes products as a JSON list under a known marker) return json.dumps({"s1": "Souris", "s2": "pile aa"})

products = { "s1": {"title": "Souris sans fil Logitech", "brand": "LOGITECH"}, "s2": {"title": "Pile AA Energizer", "brand": "ENERGIZER"}, } out = type_products(products, llm_call=fake_llm, batch_size=10) # types normalised to lowercase/stripped assert out == {"s1": "souris", "s2": "pile aa"} assert len(calls) == 1 # one batch

# second run with a cache dict pre-filled -> no llm call cache = {} type_products(products, llm_call=fake_llm, batch_size=10, cache=cache) n_after_first = len(calls) type_products(products, llm_call=fake_llm, batch_size=10, cache=cache) assert len(calls) == n_after_first # served from cache, no new calls `

• [ ] Step 2: Run test to verify it fails

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

• [ ] Step 3: Write minimal implementation

`python # core/ai_typing.py """Couche IA pour le typage fonctionnel + le graphe de compléments.

Appels LLM isolés et injectables (pattern de core/labeling.py : SDK anthropic, ANTHROPIC_API_KEY, modèle Haiku, fallback gracieux). Tout le calcul non-IA vit dans core/functional_complements.py. """ from __future__ import annotations

import hashlib import json import os

HAIKU_MODEL = "claude-haiku-4-5"

TYPING_PROMPT = """Tu classes des produits B2B (catalogue Lyreco) en un TYPE \ FONCTIONNEL court et générique, à partir de leur titre et marque uniquement.

Règles :

• Le type décrit la FONCTION (ex: "souris", "pile aa", "cartouche encre", \

• RÉUTILISE un type de la liste existante si un colle ; n'en crée un nouveau \

• Ne regroupe PAS des produits substituables sous un même type s'ils ont des \

Types déjà existants (réutilise en priorité) : {vocab}

Produits (JSON) : {products}

Réponds UNIQUEMENT par un objet JSON {{"": "", ...}} couvrant tous \ les SKU fournis, sans texte autour."""

def _norm_type(s: str) -> str: return (s or "").strip().lower()

def _key(p: dict) -> str: raw = f"{p.get('title','')}|{p.get('brand','')}" return hashlib.sha1(raw.encode("utf-8")).hexdigest()

def _parse_json_obj(text: str) -> dict: """Tolère un éventuel fence json ... autour de l'objet.""" t = text.strip() if t.startswith("`"): t = t.split("`", 2)[1] if t.startswith("json"): t = t[4:] return json.loads(t.strip())

def type_products(products: dict, llm_call, batch_size: int = 50, cache: dict | None = None) -> dict: """{sku: type} pour chaque produit de products ({sku:{title,brand}}).

Batché par batch_size. Vocabulaire contrôlé incrémental (les types déjà attribués sont passés au LLM pour réutilisation). cache (si fourni) = {hash(title+brand): type} muté en place pour éviter de retyper. """ cache = cache if cache is not None else {} result: dict[str, str] = {} vocab: set[str] = set(_norm_type(v) for v in cache.values())

pending = [] for sku, p in products.items(): k = _key(p) if k in cache: result[sku] = cache[k] else: pending.append((sku, p, k))

for start in range(0, len(pending), batch_size): chunk = pending[start:start + batch_size] payload = json.dumps({sku: p for sku, p, _ in chunk}, ensure_ascii=False) prompt = TYPING_PROMPT.format( vocab="\n".join(f"- {v}" for v in sorted(vocab)) or "(aucun)", products=payload, ) raw = llm_call(prompt) parsed = _parse_json_obj(raw) for sku, p, k in chunk: t = _norm_type(parsed.get(sku, "")) if not t: t = "(non typé)" result[sku] = t cache[k] = t vocab.add(t) return result `

• [ ] Step 4: Run test to verify it passes

Run: .venv\Scripts\python.exe -m pytest tests/test_ai_typing.py::test_type_products_reuses_vocab_and_caches -v Expected: PASS

• [ ] Step 5: Commit

`bash git add core/ai_typing.py tests/test_ai_typing.py git commit -m "feat(complements-ia): ai_typing.type_products (batched, cached, vocab)" `

---

Task 6: core/ai_typing.py — propose_complements (graphe par type)

Files:

• Modify: core/ai_typing.py
• Test: tests/test_ai_typing.py

Pour chaque type-ancre, le LLM propose des types compléments choisis dans le vocabulaire. On symétrise en arêtes non orientées (t1,t2) avec max(ai_strength).

• [ ] Step 1: Write the failing test

`python # tests/test_ai_typing.py (append) import json from core.ai_typing import propose_complements

def test_propose_complements_symmetrises_and_filters_vocab(): def fake_llm(prompt: str) -> str: if '"souris"' in prompt or "souris" in prompt.split("ANCRE:")[-1][:20]: return json.dumps([ {"type": "pile aa", "relation": "consommable-de", "strength": 0.9, "rationale": "souris sans-fil -> piles"}, {"type": "inconnu", "relation": "x", "strength": 0.5, "rationale": "hors vocab, doit être ignoré"}, ]) return json.dumps([ {"type": "souris", "relation": "consommable-pour", "strength": 0.6, "rationale": "piles pour souris"}, ])

vocab = ["souris", "pile aa"] edges = propose_complements(vocab, llm_call=fake_llm)

# one undirected edge (souris, pile aa); 'inconnu' dropped (not in vocab); # strength = max(0.9, 0.6) = 0.9 assert len(edges) == 1 e = edges[0] assert (e["t1"], e["t2"]) == ("pile aa", "souris") # sorted assert e["ai_strength"] == 0.9 assert "souris" in e["rationale"] or "piles" in e["rationale"] `

• [ ] Step 2: Run test to verify it fails

Run: .venv\Scripts\python.exe -m pytest tests/test_ai_typing.py::test_propose_complements_symmetrises_and_filters_vocab -v Expected: FAIL — ImportError: cannot import name 'propose_complements'

• [ ] Step 3: Write minimal implementation

`python # core/ai_typing.py (append) COMPLEMENT_PROMPT = """Tu raisonnes sur la COMPLÉMENTARITÉ FONCTIONNELLE entre \ types de produits B2B (bureau, EPI/sécurité, informatique, hygiène…). NE te base \ PAS sur des ventes : seulement sur l'usage réel des produits.

ANCRE: {anchor}

Parmi la liste de types ci-dessous UNIQUEMENT, lesquels sont fonctionnellement \ complémentaires de l'ancre (utilisés/nécessaires ensemble, pas des substituts) ?

Types disponibles : {vocab}

Réponds UNIQUEMENT par un tableau JSON \ [{{"type": "", "relation": "", "strength": <0..1>, \ "rationale": ""}}], vide [] si aucun. N'invente aucun type."""

def propose_complements(vocab, llm_call) -> list: """Graphe de compléments non orienté à partir du vocabulaire de types.

Pour chaque type-ancre, demande au LLM ses compléments (choisis dans vocab). Filtre les types hors-vocab, symétrise (a,b)/(b,a) en une arête tuple(sorted) avec strength = max et la justification de plus forte strength. Renvoie list[{t1, t2, relation, ai_strength, rationale}]. """ vocab_set = set(_norm_type(v) for v in vocab) vocab_lines = "\n".join(f"- {v}" for v in sorted(vocab_set)) edges: dict[tuple, dict] = {} for anchor in sorted(vocab_set): prompt = COMPLEMENT_PROMPT.format(anchor=anchor, vocab=vocab_lines) try: items = _parse_json_obj_list(llm_call(prompt)) except Exception: continue for it in items: other = _norm_type(it.get("type", "")) if other == anchor or other not in vocab_set: continue key = tuple(sorted((anchor, other))) strength = float(it.get("strength", 0.0) or 0.0) cur = edges.get(key) if cur is None or strength > cur["ai_strength"]: edges[key] = { "t1": key[0], "t2": key[1], "relation": it.get("relation") or "autre", "ai_strength": strength, "rationale": it.get("rationale") or "", } return list(edges.values())

def _parse_json_obj_list(text: str) -> list: t = text.strip() if t.startswith("`"): t = t.split("`", 2)[1] if t.startswith("json"): t = t[4:] data = json.loads(t.strip()) return data if isinstance(data, list) else [] `

• [ ] Step 4: Run test to verify it passes

Run: .venv\Scripts\python.exe -m pytest tests/test_ai_typing.py -v Expected: PASS (both ai_typing tests)

• [ ] Step 5: Commit

`bash git add core/ai_typing.py tests/test_ai_typing.py git commit -m "feat(complements-ia): ai_typing.propose_complements (undirected graph)" `

---

Task 7: _default_llm_call — backend anthropic réel (Haiku, température 0)

Files:

• Modify: core/ai_typing.py
• Test: tests/test_ai_typing.py

• [ ] Step 1: Write the failing test (le défaut sans clé API renvoie une erreur explicite, jamais un appel réseau en test)

`python # tests/test_ai_typing.py (append) import pytest from core.ai_typing import _default_llm_call

def test_default_llm_call_requires_key(monkeypatch): monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) with pytest.raises(RuntimeError, match="ANTHROPIC_API_KEY"): _default_llm_call("hello") `

• [ ] Step 2: Run test to verify it fails

Run: .venv\Scripts\python.exe -m pytest tests/test_ai_typing.py::test_default_llm_call_requires_key -v Expected: FAIL — ImportError: cannot import name '_default_llm_call'

• [ ] Step 3: Write minimal implementation

`python # core/ai_typing.py (append) def _default_llm_call(prompt: str, model: str = HAIKU_MODEL, max_tokens: int = 4096) -> str: """Backend par défaut : SDK anthropic, température 0 (déterministe).

Lève RuntimeError si la clé/SDK manque — l'orchestrateur décide quoi faire (on ne veut PAS de typage silencieusement vide). """ api_key = os.environ.get("ANTHROPIC_API_KEY") if not api_key: raise RuntimeError("ANTHROPIC_API_KEY manquante pour le typage IA") try: from anthropic import Anthropic except ImportError as exc: # pragma: no cover raise RuntimeError("SDK 'anthropic' non installé") from exc client = Anthropic(api_key=api_key) msg = client.messages.create( model=model, max_tokens=max_tokens, temperature=0, messages=[{"role": "user", "content": prompt}], ) return msg.content[0].text `

• [ ] Step 4: Run test to verify it passes

Run: .venv\Scripts\python.exe -m pytest tests/test_ai_typing.py -v Expected: PASS (all 3 ai_typing tests)

• [ ] Step 5: Commit

`bash git add core/ai_typing.py tests/test_ai_typing.py git commit -m "feat(complements-ia): _default_llm_call (anthropic Haiku, temp 0)" `

---

Task 8: scripts/refresh_functional_complements.py — orchestration + XLSX/JSON

Files:

• Create: scripts/refresh_functional_complements.py

Intégration (DB + LLM réels) : pas de test unitaire — vérifiée par un run réel --limit. La logique calculatoire est déjà couverte par les Tasks 1-7.

• [ ] Step 1: Write the script

`python # scripts/refresh_functional_complements.py """Refresh 'Compléments IA' (FR) -> snapshot XLSX + JSON.

IA sales-blind (typage fonctionnel depuis titre+marque) -> graphe de compléments par type -> croisement avec le co-achat réel -> 3 buckets (sous-exploité / confirmé / IA-seule). Méthode : docs/superpowers/specs/2026-06-02-functional-complements-ai-design.md

Run : .venv\\Scripts\\python.exe -m scripts.refresh_functional_complements --limit 300 .venv\\Scripts\\python.exe -m scripts.refresh_functional_complements --days 90 """ from __future__ import annotations

import argparse import json import sys from collections import Counter from datetime import datetime, timezone from pathlib import Path

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

from database.crm_db import CRMDatabase from database.timeseries_db import TimeseriesDatabase from core.basket_bundles import clean_baskets from core.functional_complements import ( load_csv_products, type_copurchase, score_opportunities, build_complement_outputs, ) from core.ai_typing import type_products, propose_complements, _default_llm_call

COUNTRY = "FR" EXPORT_DIR = _REPO / "exports" EXPORT_DIR.mkdir(parents=True, exist_ok=True) CSV_PATH = _REPO / "imports" / "marie" / "Datas_pour_produits_associes.csv" CACHE_PATH = EXPORT_DIR / "complements_ia_typing_cache.json"

def step(msg: str) -> None: print(f" [{datetime.now().strftime('%H:%M:%S')}] {msg}", flush=True)

def compute(days: int, channel: str, min_type_support: int, limit: int | None) -> dict: crm = CRMDatabase() ts = TimeseriesDatabase() channel_filter = "AND order_channel_code = 'W'" if channel == "WEB" else ""

step(f"Phase 1/6 — pull paniers FR multi-lignes ({days}j, {channel})…") order_baskets = ts.query( f""" SELECT order_number, ARRAY_AGG(DISTINCT product_reference) AS skus FROM ecom_order_lines WHERE source_country = %s AND order_date >= CURRENT_DATE - (%s || ' days')::INTERVAL AND product_reference IS NOT NULL {channel_filter} GROUP BY order_number HAVING COUNT(DISTINCT product_reference) >= 2 """, (COUNTRY, days), ) step(f" → {len(order_baskets):,} paniers")

# product->world map (réutilise le pattern du refresh cross-family) all_skus = sorted({s for r in order_baskets for s in r["skus"]}) prod_world: dict[str, str] = {} prod_meta: dict[str, dict] = {} for start in range(0, len(all_skus), 10000): for r in crm.query(""" SELECT p.product_reference, p.product_description, 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, all_skus[start:start + 10000])): prod_world[r["product_reference"]] = r["world"] prod_meta[r["product_reference"]] = {"name": r["product_description"]}

step("Phase 2/6 — nettoyage paniers (anti-réassort)…") kept, clean_stats = clean_baskets(order_baskets, prod_world) sku_sales: Counter = Counter() for b in kept: for s in b: sku_sales[s] += 1

step("Phase 3/6 — chargement CSV Marie + texte produit pour typage…") csv_products = load_csv_products(CSV_PATH) if limit: csv_products = dict(list(csv_products.items())[:limit]) # univers de typage = SKU du CSV (graphe) ∪ SKU actifs en paniers (co-achat) prod_text: dict[str, dict] = dict(csv_products) for s in sku_sales: if s not in prod_text: nm = (prod_meta.get(s) or {}).get("name") or "" prod_text[s] = {"title": nm, "brand": ""} prod_meta.setdefault(s, {"name": prod_text[s]["title"]}) # enrichir prod_meta des SKU CSV (titre comme nom si absent en base) for s, p in csv_products.items(): prod_meta.setdefault(s, {"name": p["title"]}) step(f" → {len(csv_products):,} SKU CSV · {len(prod_text):,} SKU à typer")

step("Phase 4/6 — typage fonctionnel IA (Haiku, caché)…") cache = json.loads(CACHE_PATH.read_text(encoding="utf-8")) if CACHE_PATH.exists() else {} t0 = datetime.now() sku_type_map = type_products(prod_text, llm_call=_default_llm_call, cache=cache) CACHE_PATH.write_text(json.dumps(cache, ensure_ascii=False), encoding="utf-8") vocab = sorted({t for t in sku_type_map.values() if t and t != "(non typé)"}) step(f" → {len(vocab)} types · typage en {(datetime.now()-t0).seconds}s")

step("Phase 5/6 — graphe de compléments par type (IA, sales-blind)…") # le graphe ne raisonne que sur les types présents dans le CSV csv_types = sorted({sku_type_map[s] for s in csv_products if s in sku_type_map}) t1 = datetime.now() edges = propose_complements(csv_types, llm_call=_default_llm_call) step(f" → {len(edges)} arêtes de compléments · {(datetime.now()-t1).seconds}s")

step("Phase 6/6 — croisement co-achat + scoring…") cp = type_copurchase(kept, sku_type_map, min_type_support=min_type_support) scored = score_opportunities(edges, cp, min_type_support=min_type_support) outputs = build_complement_outputs(scored, sku_type_map, dict(sku_sales), prod_meta, top_per_type=3) step(f" → {len(outputs['sous_exploite'])} sous-exploités · " f"{len(outputs['confirme'])} confirmés · " f"{len(outputs['ia_seule'])} IA-seule")

return { "computed_at": datetime.now(timezone.utc).isoformat(), "country": COUNTRY, "days": days, "channel": channel, "min_type_support": min_type_support, "n_baskets_kept": len(kept), "clean_stats": clean_stats, "n_types": len(vocab), "n_edges": len(edges), outputs, }

def write_xlsx(out_path: Path, data: dict) -> None: from openpyxl import Workbook from openpyxl.styles import Alignment, Font, PatternFill from openpyxl.utils import get_column_letter

wb = Workbook() head_fill = PatternFill("solid", fgColor="2A2A38") head_font = Font(bold=True, color="FFFFFF")

def add(name, headers, rows, widths): ws = wb.active if wb.sheetnames == ["Sheet"] else wb.create_sheet(name) ws.title = name ws.append(headers) for c in ws[1]: c.fill = head_fill; c.font = head_font c.alignment = Alignment(vertical="center", wrap_text=True) ws.freeze_panes = "A2" for r in rows: ws.append(r) for i, w in enumerate(widths, 1): ws.column_dimensions[get_column_letter(i)].width = w

def _reps(row, key): return " / ".join(f"{r['sku']} {r['name'][:24]}" for r in row.get(key, []))

def bucket_rows(rows): return [[r["t1"], r["t2"], r["relation"], round(r["ai_strength"], 2), r["n1"], r["n2"], r["co"], r["lift"], r["cond"], round(r["score"], 3), _reps(r, "reps_t1"), _reps(r, "reps_t2"), (r["rationale"] or "")[:120]] for r in rows]

cols = ["type_a", "type_b", "relation", "force_ia", "n_a", "n_b", "co_paniers", "lift", "cond", "score", "skus_a", "skus_b", "justification"] widths = [22, 22, 16, 8, 8, 8, 10, 8, 8, 8, 40, 40, 50]

add("Sous-exploités", cols, bucket_rows(data["sous_exploite"]), widths) add("Confirmés", cols, bucket_rows(data["confirme"]), widths) add("IA-seule à vérifier", cols, bucket_rows(data["ia_seule"]), widths) add("Typage", ["sku", "type", "name"], [[r["sku"], r["type"], r["name"][:60]] for r in data["typage"]], [16, 28, 60]) add("Graphe types", ["type_a", "type_b", "relation", "force_ia", "justification"], [[e["t1"], e["t2"], e["relation"], round(e["ai_strength"], 2), (e["rationale"] or "")[:120]] for e in data["graphe"]], [22, 22, 16, 8, 50]) wb.save(out_path)

def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--days", type=int, default=90) ap.add_argument("--channel", default="ALL", choices=["ALL", "WEB"]) ap.add_argument("--min-type-support", type=int, default=25, dest="min_type_support") ap.add_argument("--limit", type=int, default=None, help="sous-échantillonne le CSV (dev/smoke)") args = ap.parse_args()

today = datetime.now(timezone.utc).strftime("%Y-%m-%d") out_xlsx = EXPORT_DIR / f"complements_ia_{COUNTRY}_{today}_d{args.days}.xlsx" out_json = out_xlsx.with_suffix(".json")

data = compute(args.days, args.channel, args.min_type_support, args.limit) step(f"écriture {out_json.name}…") out_json.write_text(json.dumps(data, indent=2, default=str, ensure_ascii=False), encoding="utf-8") step(f"écriture {out_xlsx.name}…") write_xlsx(out_xlsx, data)

print("\n" + "=" * 70) print("COMPLÉMENTS IA — FR — SNAPSHOT") print("=" * 70) print(f" Types: {data['n_types']} · Arêtes: {data['n_edges']}") print(f" Sous-exploités: {len(data['sous_exploite'])} · " f"Confirmés: {len(data['confirme'])} · IA-seule: {len(data['ia_seule'])}") print(f" exports/{out_xlsx.name}") return 0

if __name__ == "__main__": sys.exit(main()) `

• [ ] Step 2: Smoke-run on a small sample

Run: .venv\Scripts\python.exe -m scripts.refresh_functional_complements --limit 300 --days 30 Expected: termine sans exception ; écrit exports/complements_ia_FR__d30.xlsx + .json ; le récap imprime des compteurs non nuls pour Types/Arêtes. (Nécessite ANTHROPIC_API_KEY dans l'env + DBs Docker up.)

• [ ] Step 3: Eyeball the output

Ouvrir l'XLSX : vérifier que la feuille Sous-exploités contient des paires plausibles (ex. un consommable face à son équipement) avec lift < 1 ou co faible, et que la feuille Typage n'a pas de types absurdes en masse. Noter dans le commit le runtime de la passe IA (1er run).

• [ ] Step 4: Run the full pure-function suite once more

Run: .venv\Scripts\python.exe -m pytest tests/test_functional_complements.py tests/test_ai_typing.py -v Expected: PASS (7 tests)

• [ ] Step 5: Commit

`bash git add scripts/refresh_functional_complements.py git commit -m "feat(complements-ia): refresh script (pull->type->graph->cross-check->xlsx/json)" `

---

Task 9 (OPTIONNELLE — non requise pour le livrable Marie): persistance DB réutilisable

À ne faire que si on veut alimenter les recos en aval. Ajoute deux tables postgres et un upsert en fin de compute.

Files:

• Create: database/init_postgres_functional_complements.sql
• Modify: scripts/refresh_functional_complements.py (ajout d'une Phase 7 d'upsert, derrière un flag --persist)

• [ ] Step 1: Write the migration

`sql -- database/init_postgres_functional_complements.sql CREATE TABLE IF NOT EXISTS product_functional_type ( source_country TEXT NOT NULL, product_reference TEXT NOT NULL, type_label TEXT NOT NULL, computed_at TIMESTAMPTZ NOT NULL DEFAULT now(), PRIMARY KEY (source_country, product_reference) );

CREATE TABLE IF NOT EXISTS type_complement_edge ( source_country TEXT NOT NULL, type_a TEXT NOT NULL, type_b TEXT NOT NULL, relation TEXT, ai_strength DOUBLE PRECISION, rationale TEXT, co_baskets INTEGER, lift DOUBLE PRECISION, cond DOUBLE PRECISION, bucket TEXT, computed_at TIMESTAMPTZ NOT NULL DEFAULT now(), PRIMARY KEY (source_country, type_a, type_b) ); `

• [ ] Step 2: Apply the migration

Run: Get-Content database/init_postgres_functional_complements.sql | docker exec -i psql -U -d Expected: CREATE TABLE ×2 (ou aucun message si déjà présentes).

• [ ] Step 3: Add --persist upsert in the script (derrière le flag, après le scoring)

`python # dans main(): ap.add_argument("--persist", action="store_true") # dans compute(): si persist, après build_complement_outputs : # crm.execute_many( # "INSERT INTO product_functional_type (source_country, product_reference, type_label) " # "VALUES (%s,%s,%s) ON CONFLICT (source_country, product_reference) " # "DO UPDATE SET type_label=EXCLUDED.type_label, computed_at=now()", # [(COUNTRY, s, t) for s, t in sku_type_map.items()]) # crm.execute_many( # "INSERT INTO type_complement_edge (source_country, type_a, type_b, relation, " # "ai_strength, rationale, co_baskets, lift, cond, bucket) VALUES (%s,...,%s) " # "ON CONFLICT (source_country, type_a, type_b) DO UPDATE SET ...", # [(COUNTRY, r["t1"], r["t2"], r["relation"], r["ai_strength"], r["rationale"], # r["co"], r["lift"], r["cond"], r["bucket"]) for r in scored]) ` > Avant d'écrire ce code, vérifier la vraie API de database.crm_db.CRMDatabase > (présence d'un execute_many / execute) et adapter — ne pas présumer.

• [ ] Step 4: Verify

Run: .venv\Scripts\python.exe -m scripts.refresh_functional_complements --limit 300 --persist puis docker exec ... psql -c "SELECT bucket, count(*) FROM type_complement_edge GROUP BY bucket;" Expected: des lignes par bucket.

• [ ] Step 5: Commit

`bash git add database/init_postgres_functional_complements.sql scripts/refresh_functional_complements.py git commit -m "feat(complements-ia): optional DB persistence (type map + complement edges)" `

---

Self-Review (effectuée)

• Spec coverage : §3 CSV→Task 1 ; §4.4 co-achat→Task 2 ; §4.5 score/buckets→Task 3 ; §4.6 sorties→Task 4 + Task 8 ; §4.2 typage→Tasks 5/7 ; §4.3 graphe→Task 6 ; §5.3 orchestration→Task 8 ; §6 défauts/CLI→Task 8 ; persistance §4.6→Task 9 (optionnelle, défaut = fichiers). Section panel §8 = hors périmètre (phase 2) — non planifiée, conforme.
• Placeholders : aucun TODO/TBD ; tout le code des steps est complet. La seule note « vérifier l'API CRMDatabase » est dans la Task optionnelle 9 et est une consigne de prudence, pas un placeholder de logique livrée.
• Type consistency : type = chaîne normalisée partout ; paires = tuple(sorted) dans type_copurchase / score_opportunities / propose_complements ; sku_type_map: dict[str,str], baskets: list[frozenset], copurchase keys {N,n,co,lift,cond} cohérents entre Tasks 2↔3 ; score_opportunities consomme exactement la sortie de type_copurchase ; build_complement_outputs consomme la sortie de score_opportunities (+ sku_sales, prod_meta). Signatures llm_call identiques Tasks 5/6/7.