# Prospect Segmentation — Interaction Fact + DuckDB Cube 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: Build the prospect_interaction fact + a DuckDB cube that rolls fit × engagement × channel up and applies volume-shrinkage downstream, so any slice resolves only as deep as its data supports.
Architecture: A canonical-normalized prospect_interaction fact table in Postgres (one row per prospect-interaction). A DuckDB layer ATTACHes Postgres (postgres extension), joins the fact to prospect fit/firmographics, materializes the expensive rollups, and a pure transform applies core/shrinkage.py to each rollup cell using a per-dimension K table from core/shrinkage.py:k_from_persistence. DuckDB is embedded/batch — never an API.
Tech Stack: Python 3.12, DuckDB (duckdb + bundled postgres extension), pytest. Builds on plan 1 (core/shrinkage.py, core/prospect_dims.py).
Scope note: Plan 2 of 3. The fact-builder seeds the email channel (Eloqua responders — same (campaign_label, lyreco_118_id) key as the prospect base, so the join is exact); web/social channels are additive later. Plan 3 = profile scorer + Prospect tab.
---
Task 1: Add DuckDB + a Postgres-attached connection helper
Files:
- Modify:
requirements.txt - Create:
core/cube.py - Test:
tests/test_cube_conn.py
- [ ] Step 1: Add the dependency
Append to requirements.txt:
`
duckdb>=1.1.0
`
Run: ./venv/Scripts/python.exe -m pip install "duckdb>=1.1.0"
Expected: installs successfully.
- [ ] Step 2: Write the failing test
`python
# tests/test_cube_conn.py
import duckdb
from core.cube import open_cube
def test_open_cube_returns_usable_connection(tmp_path): con = open_cube(str(tmp_path / "t.duckdb")) try: assert con.execute("SELECT 42").fetchone()[0] == 42 finally: con.close()
def test_open_cube_in_memory_default():
con = open_cube() # no path -> in-memory
try:
assert con.execute("SELECT 1").fetchone()[0] == 1
finally:
con.close()
`
- [ ] Step 3: Run test to verify it fails
Run: ./venv/Scripts/python.exe -m pytest tests/test_cube_conn.py -v
Expected: FAIL — ModuleNotFoundError: No module named 'core.cube'.
- [ ] Step 4: Write minimal implementation
`python
# core/cube.py
"""DuckDB cube layer for prospect segmentation. Embedded/batch only -- never
exposed as an API (gridiron gotcha c). ATTACHes Postgres via the bundled
postgres extension so warm CRM/prospect data is queried with no ETL copy.
"""
from __future__ import annotations
import os
import duckdb
# Lyreco local Postgres (CRM / prospect tables). Password is the dev-stack default. _PG_DSN = os.environ.get( "LC_PG_DSN", "host=127.0.0.1 port=5433 dbname=leadcontagion user=leadcontagion password=lc_2026", )
def open_cube(path: str | None = None): """Return a DuckDB connection (in-memory if path is None).""" return duckdb.connect(path or ":memory:")
def attach_postgres(con, alias: str = "pg") -> None:
"""ATTACH the Lyreco Postgres into the DuckDB connection (read warm data)."""
con.execute("INSTALL postgres; LOAD postgres;")
con.execute(f"ATTACH '{_PG_DSN}' AS {alias} (TYPE postgres, READ_ONLY);")
`
- [ ] Step 5: Run test to verify it passes
Run: ./venv/Scripts/python.exe -m pytest tests/test_cube_conn.py -v
Expected: PASS (2 passed).
- [ ] Step 6: Commit
`bash
git add requirements.txt core/cube.py tests/test_cube_conn.py
git commit -m "feat: DuckDB cube connection helper + postgres attach"
`
---
Task 2: prospect_interaction fact table + email-channel builder
Files:
- Create:
database/init_postgres_prospect_interaction.sql - Create:
workers/build_prospect_interactions.py - Test:
tests/test_build_prospect_interactions.py
Fact grain: one row per prospect-interaction (gridiron: interaction grain, not aggregate). Channel normalized to canonical via core/prospect_dims.normalize_channel.
- [ ] Step 1: Create the schema
`sql
-- database/init_postgres_prospect_interaction.sql
BEGIN;
CREATE TABLE IF NOT EXISTS prospect_interaction (
campaign_label TEXT NOT NULL,
lyreco_118_id TEXT NOT NULL,
interaction_ts TIMESTAMPTZ,
channel TEXT NOT NULL, -- canonical (web/email/social/offline/UNK)
source TEXT, -- raw source tag for audit
weight REAL NOT NULL DEFAULT 1.0,
last_computed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_pi_prospect ON prospect_interaction(campaign_label, lyreco_118_id);
CREATE INDEX IF NOT EXISTS idx_pi_channel ON prospect_interaction(channel);
COMMIT;
`
Run: docker exec -e PGPASSWORD=lc_2026 -i lc-postgres psql -U leadcontagion -d leadcontagion < database/init_postgres_prospect_interaction.sql
Expected: CREATE TABLE / CREATE INDEX / COMMIT.
- [ ] Step 2: Write the failing test (unit-test the row-shaping, no DB)
`python
# tests/test_build_prospect_interactions.py
from workers.build_prospect_interactions import shape_email_rows
def test_shape_email_rows_normalizes_channel_and_keys():
src = [
{"campaign_label": "FR_2026Q1", "lyreco_118_id": "118-1", "sent_date": "2026-01-05"},
{"campaign_label": "FR_2026Q1", "lyreco_118_id": "118-2", "sent_date": None},
]
rows = shape_email_rows(src)
assert len(rows) == 2
assert all(r["channel"] == "email" for r in rows) # canonical
assert all(r["source"] == "eloqua_campaign_responders" for r in rows)
assert rows[0]["interaction_ts"] == "2026-01-05"
assert rows[0]["campaign_label"] == "FR_2026Q1" and rows[0]["lyreco_118_id"] == "118-1"
`
- [ ] Step 3: Run test to verify it fails
Run: ./venv/Scripts/python.exe -m pytest tests/test_build_prospect_interactions.py -v
Expected: FAIL — ModuleNotFoundError.
- [ ] Step 4: Write minimal implementation
`python
# workers/build_prospect_interactions.py
"""Assemble the prospect_interaction fact (one row per prospect-interaction),
canonical-normalized. Seeds the EMAIL channel from eloqua_campaign_responders,
which shares the (campaign_label, lyreco_118_id) key with the prospect base, so
the join is exact. Additive: add web/social shapers the same way.
Run: python -m workers.build_prospect_interactions --country FR """ from __future__ import annotations
import argparse import sys from pathlib import Path
import psycopg2.extras
_REPO = Path(__file__).resolve().parent.parent if str(_REPO) not in sys.path: sys.path.insert(0, str(_REPO))
from core.prospect_dims import normalize_channel from database.crm_db import CRMDatabase
def shape_email_rows(src_rows: list[dict]) -> list[dict]: """Map raw eloqua-responder rows -> canonical prospect_interaction rows.""" out = [] for r in src_rows: out.append({ "campaign_label": r["campaign_label"], "lyreco_118_id": r["lyreco_118_id"], "interaction_ts": r.get("sent_date"), "channel": normalize_channel("eloqua"), # -> "email" "source": "eloqua_campaign_responders", "weight": 1.0, }) return out
def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--country", default="GB") # prospect base is GB; campaign_label is "lyreco_*", not country-prefixed args = ap.parse_args() crm = CRMDatabase() # Country lives on the prospect base, not in campaign_label -- join to it. src = crm.query( """SELECT r.campaign_label, r.lyreco_118_id, r.sent_date FROM eloqua_campaign_responders r JOIN prospect_firmographic_features p ON p.campaign_label = r.campaign_label AND p.lyreco_118_id = r.lyreco_118_id WHERE p.country_code = %s""", (args.country,), ) rows = shape_email_rows(src) with crm.connect() as conn, conn.cursor() as cur: # Rebuild the whole email channel each run (single-country GB today; revisit # the delete scope when a country_code lands on the fact for multi-country). cur.execute("DELETE FROM prospect_interaction WHERE source='eloqua_campaign_responders'") psycopg2.extras.execute_batch(cur, """ INSERT INTO prospect_interaction (campaign_label, lyreco_118_id, interaction_ts, channel, source, weight) VALUES (%(campaign_label)s,%(lyreco_118_id)s,%(interaction_ts)s, %(channel)s,%(source)s,%(weight)s) """, rows, page_size=500) print(f"wrote {len(rows)} email interactions ({args.country})") return 0
if __name__ == "__main__":
sys.exit(main())
`
- [ ] Step 5: Run test to verify it passes
Run: ./venv/Scripts/python.exe -m pytest tests/test_build_prospect_interactions.py -v
Expected: PASS (1 passed).
- [ ] Step 6: Integration smoke (live PG) + commit
Run: ./venv/Scripts/python.exe -m workers.build_prospect_interactions --country FR
Expected: wrote with N > 0.
`bash
git add database/init_postgres_prospect_interaction.sql workers/build_prospect_interactions.py tests/test_build_prospect_interactions.py
git commit -m "feat: prospect_interaction fact + email-channel builder"
`
---
Task 3: Rollup + shrinkage transform (workers/prospect_cube.py)
Files:
- Create:
workers/prospect_cube.py - Test:
tests/test_prospect_cube.py
Idea: roll the fact up to (fit_community × channel) counts in DuckDB, then apply norm_shrunk per fit_community toward the all-prospect channel marginal, with K from the channel dimension's persistence. The test uses an in-memory DuckDB fixture — no live PG — so it is fully runnable.
- [ ] Step 1: Write the failing test
`python
# tests/test_prospect_cube.py
from core.cube import open_cube
from workers.prospect_cube import channel_mix_shrunk
def test_channel_mix_shrunk_pulls_thin_community_to_global(tmp_path):
con = open_cube()
con.execute("CREATE TABLE fact(community_id INT, channel VARCHAR)")
# community 1: rich + email-heavy; community 2: a single 'social' touch (thin)
con.executemany("INSERT INTO fact VALUES (?, ?)",
[(1, "email")] 90 + [(1, "web")] 10 + [(2, "social")])
out = channel_mix_shrunk(con, fact="fact", K=50)
# rich community keeps its own email-heavy mix
assert out[1]["email"] > 0.8
# thin community (n=1) is pulled toward the GLOBAL marginal (mostly email/web), not 100% social
assert out[2]["social"] < 0.5
con.close()
`
- [ ] Step 2: Run test to verify it fails
Run: ./venv/Scripts/python.exe -m pytest tests/test_prospect_cube.py -v
Expected: FAIL — ModuleNotFoundError: No module named 'workers.prospect_cube'.
- [ ] Step 3: Write minimal implementation
`python
# workers/prospect_cube.py
"""DuckDB rollups for prospect segmentation, with volume-shrinkage applied
downstream (per gridiron: keep shrinkage a thin layer after the rollup).
"""
from __future__ import annotations
import sys from pathlib import Path
_REPO = Path(__file__).resolve().parent.parent if str(_REPO) not in sys.path: sys.path.insert(0, str(_REPO))
from core.shrinkage import norm_shrunk
def channel_mix_shrunk(con, fact: str = "fact", K: float = 50) -> dict:
"""Per fit-community channel distribution, shrunk toward the global channel
marginal. Returns {community_id: {channel: prob}}."""
# global channel marginal (the parent)
g = con.execute(f"SELECT channel, COUNT(*) c FROM {fact} GROUP BY channel").fetchall()
total = sum(c for _, c in g) or 1
parent = {ch: c / total for ch, c in g}
# per-community rollup
rows = con.execute(
f"SELECT community_id, channel, COUNT(*) c FROM {fact} GROUP BY community_id, channel"
).fetchall()
counts: dict = {}
for cid, ch, c in rows:
counts.setdefault(cid, {})[ch] = c
return {cid: norm_shrunk(cc, parent, n=sum(cc.values()), K=K)
for cid, cc in counts.items()}
`
- [ ] Step 4: Run test to verify it passes
Run: ./venv/Scripts/python.exe -m pytest tests/test_prospect_cube.py -v
Expected: PASS (1 passed).
- [ ] Step 5: Commit
`bash
git add workers/prospect_cube.py tests/test_prospect_cube.py
git commit -m "feat: DuckDB channel-mix rollup with downstream shrinkage"
`
---
Task 4: Per-dimension K from persistence (workers/dim_persistence.py)
Files:
- Create:
workers/dim_persistence.py - Test:
tests/test_dim_persistence.py
Idea: given a metric's value per entity across two periods, compute persistence and derive K (+ a ranking so the lowest-persistence dim collapses first). Pure wrapper over core/shrinkage.
- [ ] Step 1: Write the failing test
`python
# tests/test_dim_persistence.py
from workers.dim_persistence import k_table, collapse_order
def test_k_table_assigns_smaller_k_to_more_persistent_dim(): series = { "channel": {"e1": [1.0, 1.0], "e2": [5.0, 5.0]}, # perfectly persistent -> small K "product": {"e1": [1.0, 9.0], "e2": [9.0, 1.0]}, # noisy -> big K } kt = k_table(series, k_lo=60, k_hi=300) assert kt["channel"] < kt["product"]
def test_collapse_order_is_lowest_persistence_first():
series = {
"channel": {"e1": [1.0, 1.0], "e2": [5.0, 5.0]}, # high persistence
"product": {"e1": [1.0, 9.0], "e2": [9.0, 1.0]}, # low persistence
}
assert collapse_order(series)[0] == "product" # noisiest collapses first
`
- [ ] Step 2: Run test to verify it fails
Run: ./venv/Scripts/python.exe -m pytest tests/test_dim_persistence.py -v
Expected: FAIL — ModuleNotFoundError.
- [ ] Step 3: Write minimal implementation
`python
# workers/dim_persistence.py
"""Per-dimension K + collapse order from signal persistence. K is set inversely
to persistence (gridiron); the lowest-persistence dim is collapsed toward its
marginal first when several dims are thin at once.
"""
from __future__ import annotations
import sys from pathlib import Path
_REPO = Path(__file__).resolve().parent.parent if str(_REPO) not in sys.path: sys.path.insert(0, str(_REPO))
from core.shrinkage import persistence, k_from_persistence
def k_table(series_by_dim: dict, k_lo: float = 60, k_hi: float = 300) -> dict: """series_by_dim = {dim: {entity: [val_t, val_t+1, ...]}} -> {dim: K}.""" return {dim: k_from_persistence(persistence(s), k_lo, k_hi) for dim, s in series_by_dim.items()}
def collapse_order(series_by_dim: dict) -> list:
"""Dims ranked lowest-persistence first (= safest to collapse first)."""
return sorted(series_by_dim, key=lambda d: persistence(series_by_dim[d]))
`
- [ ] Step 4: Run test to verify it passes
Run: ./venv/Scripts/python.exe -m pytest tests/test_dim_persistence.py -v
Expected: PASS (2 passed).
- [ ] Step 5: Commit
`bash
git add workers/dim_persistence.py tests/test_dim_persistence.py
git commit -m "feat: per-dimension K + collapse order from persistence"
`
---
Self-Review
- Spec coverage: interaction fact (spec component 2) ✓ Task 2; DuckDB cube + ATTACH + rollup + downstream shrinkage (component 3) ✓ Tasks 1,3; persistence→K + walk-up order (component 5) ✓ Task 4. Profile scorer + Prospect tab (components 6,7) → plan 3.
- Placeholders: none — every step has runnable test + impl + exact command. Task 2's live-PG step is an explicit integration smoke, not a placeholder.
- Type consistency:
Kisfloatend-to-end and matchescore/shrinkage.pysignatures from plan 1;norm_shrunk(counts, parent, n, K)called in Task 3 exactly as defined in plan 1;channelvalues are the canonical strings fromcore/prospect_dims.py. - Dependency:
core/cube.open_cube(Task 1) is used by Task 3's test;core/shrinkage+core/prospect_dims(plan 1, on main) are imported by Tasks 2–4. - Risk flagged: web/social channel shapers + email open/click depth (email_results via contact_email) are deferred — Task 2 seeds the exact-key email-response channel only; the fact is additive by design.