⚡ Swarm Architecture

Prospect Segmentation — Profile Scorer + Prospect Tab Implementation Plan

📅 Published August 12, 2026 ⏱️ 11 min read 🔬 Peer Verified

# Prospect Segmentation — Profile Scorer + Prospect Tab 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: Turn the cube foundation into a per-prospect prospect_profile (fit + coverage-aware warmth + best channel + inherited offer) and surface it as a dedicated Prospects dashboard tab (Fit × Engagement quadrant → ranked list).

Architecture: A pure coverage-aware warmth function (core/warmth.py) + a scorer (workers/compute_prospect_profile.py) that joins the prospect base → rank-1 community (fit) → prospect_interaction (warmth) → community_product_offer (the 5-SKU offer), writing one row per prospect to prospect_profile. The dashboard reads that table: a quadrant panel (counts per Fit×Engagement cell) and a list/card panel (ranked prospects with offer + best channel).

Tech Stack: Python 3.12, FastAPI/HTMX (existing dashboard), pytest. Builds on plan 1 (core/shrinkage.py, core/prospect_dims.py) + plan 2 (prospect_interaction).

Scope note: Plan 3 of 3 (final). Contagion flag (Layer 4) is a nullable column populated later (droplet/OMEGA consult), not built here.

Data keys (verified this session):

  • prospect_firmographic_features(campaign_label, lyreco_118_id BIGINT, company, country_code, cro_number, sic_section, region_key, …) — the prospect universe (all GB today).
  • prospect_community_match(campaign_label, lyreco_118_id, rank, community_id, similarity, weight, …) — fit; use rank=1.
  • prospect_interaction(campaign_label, lyreco_118_id, country_code, channel, …) — warmth source (email channel populated).
  • community_product_offer(community_id, source_country, family_rank, product_rank, product_reference, product_description, penetration, …) — the inherited 5-SKU offer.

---

Task 1: Coverage-aware warmth (core/warmth.py)

Files:

  • Create: core/warmth.py
  • Test: tests/test_warmth.py

Rationale: warmth must be computed over observed channels only — an unobserved channel is "unknown," not "cold" (spec hard-constraint; GA4 dual-bridge lesson). Returns a score in [0,1] + a confidence = observed-channels / total-channels.

  • [ ] Step 1: Write the failing test

`python # tests/test_warmth.py from core.warmth import warmth_score, CHANNELS

def test_more_touches_is_warmer(): low, _ = warmth_score({"email": 1}) high, _ = warmth_score({"email": 40}) assert high > low and 0.0 <= low <= 1.0 and high <= 1.0

def test_confidence_reflects_channel_coverage(): _, c_one = warmth_score({"email": 5}) _, c_two = warmth_score({"email": 5, "web": 5}) assert c_one == round(1 / len(CHANNELS), 4) assert c_two == round(2 / len(CHANNELS), 4)

def test_unobserved_channels_do_not_drag_score_to_cold(): # email-only prospect with heavy email is HOT, not diluted by 3 missing channels score, _ = warmth_score({"email": 100}) assert score > 0.8

def test_no_interactions_is_zero_warmth_zero_confidence(): assert warmth_score({}) == (0.0, 0.0) `

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

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

  • [ ] Step 3: Write minimal implementation

`python # core/warmth.py """Coverage-aware prospect warmth. Score is the mean per-channel intensity over OBSERVED channels only -- a channel we never observe is unknown, not cold (spec hard-constraint). Confidence = observed-channels / total-channels. """ from __future__ import annotations

# Canonical observable channels (excludes UNK). Mirrors core/prospect_dims. CHANNELS = ("web", "email", "social", "offline") _SATURATION = 10.0 # touches at which a channel's intensity ~0.5

def _intensity(count: int) -> float: """Saturating touch intensity in [0,1): count/(count+SATURATION).""" return count / (count + _SATURATION)

def warmth_score(channel_counts: dict) -> tuple[float, float]: """channel_counts = {canonical_channel: n_interactions} -> (score, confidence). score = mean intensity over observed channels; confidence = coverage.""" observed = {c: n for c, n in channel_counts.items() if c in CHANNELS and n > 0} if not observed: return (0.0, 0.0) score = sum(_intensity(n) for n in observed.values()) / len(observed) confidence = len(observed) / len(CHANNELS) return (round(score, 4), round(confidence, 4)) `

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

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

  • [ ] Step 5: Commit

`bash git add core/warmth.py tests/test_warmth.py git commit -m "feat: coverage-aware prospect warmth score" `

---

Task 2: prospect_profile table + scorer

Files:

  • Create: database/init_postgres_prospect_profile.sql
  • Create: workers/compute_prospect_profile.py
  • Test: tests/test_compute_prospect_profile.py

  • [ ] Step 1: Create the schema

`sql -- database/init_postgres_prospect_profile.sql BEGIN; CREATE TABLE IF NOT EXISTS prospect_profile ( campaign_label TEXT NOT NULL, lyreco_118_id BIGINT NOT NULL, company TEXT, country_code TEXT, fit_community INTEGER, fit_tier TEXT, -- low / mid / high warmth_score REAL, engagement_tier TEXT, -- cold / warm / hot best_channel TEXT, n_interactions INTEGER, confidence REAL, contagion_flag BOOLEAN, -- Layer 4, populated later (nullable now) last_computed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), PRIMARY KEY (campaign_label, lyreco_118_id) ); CREATE INDEX IF NOT EXISTS idx_pp_quadrant ON prospect_profile(country_code, fit_tier, engagement_tier); COMMIT; `

Run: docker exec -e PGPASSWORD=lc_2026 -i lc-postgres psql -U leadcontagion -d leadcontagion < database/init_postgres_prospect_profile.sql Expected: CREATE TABLE / CREATE INDEX / COMMIT.

  • [ ] Step 2: Write the failing test (pure assembly helpers — no DB)

`python # tests/test_compute_prospect_profile.py from workers.compute_prospect_profile import fit_tier, best_channel

def test_fit_tier_thresholds(): assert fit_tier(0.8) == "high" assert fit_tier(0.4) == "mid" assert fit_tier(0.1) == "low" assert fit_tier(None) == "low" # unmatched prospect

def test_best_channel_picks_max_count(): assert best_channel({"email": 9, "web": 2}) == "email" assert best_channel({}) is None `

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

Run: ./venv/Scripts/python.exe -m pytest tests/test_compute_prospect_profile.py -v Expected: FAIL — ModuleNotFoundError.

  • [ ] Step 4: Write minimal implementation

`python # workers/compute_prospect_profile.py """Assemble prospect_profile: one row per prospect joining fit (rank-1 community), coverage-aware warmth (prospect_interaction), best channel, and confidence. The 5-SKU offer is read at display time from community_product_offer via fit_community, so it is not duplicated here.

Run: python -m workers.compute_prospect_profile --country GB """ from __future__ import annotations

import argparse import sys from collections import defaultdict 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_engagement_tier from core.warmth import warmth_score from database.crm_db import CRMDatabase

def fit_tier(similarity: float | None) -> str: if similarity is None: return "low" if similarity >= 0.66: return "high" if similarity >= 0.33: return "mid" return "low"

def best_channel(channel_counts: dict) -> str | None: return max(channel_counts, key=channel_counts.get) if channel_counts else None

def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--country", default="GB") args = ap.parse_args() crm = CRMDatabase()

base = crm.query( """SELECT campaign_label, lyreco_118_id, company, country_code FROM prospect_firmographic_features WHERE country_code = %s""", (args.country,))

fit = {(r["campaign_label"], r["lyreco_118_id"]): r for r in crm.query( """SELECT campaign_label, lyreco_118_id, community_id, similarity FROM prospect_community_match WHERE rank = 1""")}

inter: dict = defaultdict(lambda: defaultdict(int)) for r in crm.query( """SELECT campaign_label, lyreco_118_id, channel, COUNT(*) c FROM prospect_interaction WHERE country_code = %s GROUP BY 1,2,3""", (args.country,)): inter[(r["campaign_label"], r["lyreco_118_id"])][r["channel"]] = r["c"]

rows = [] for b in base: key = (b["campaign_label"], b["lyreco_118_id"]) f = fit.get(key) counts = dict(inter.get(key, {})) score, conf = warmth_score(counts) rows.append({ "campaign_label": b["campaign_label"], "lyreco_118_id": b["lyreco_118_id"], "company": b["company"], "country_code": b["country_code"], "fit_community": f["community_id"] if f else None, "fit_tier": fit_tier(f["similarity"] if f else None), "warmth_score": score, "engagement_tier": normalize_engagement_tier(score), "best_channel": best_channel(counts), "n_interactions": sum(counts.values()), "confidence": conf, "contagion_flag": None, }) with crm.connect() as conn, conn.cursor() as cur: cur.execute("DELETE FROM prospect_profile WHERE country_code = %s", (args.country,)) psycopg2.extras.execute_batch(cur, """ INSERT INTO prospect_profile (campaign_label, lyreco_118_id, company, country_code, fit_community, fit_tier, warmth_score, engagement_tier, best_channel, n_interactions, confidence, contagion_flag) VALUES (%(campaign_label)s,%(lyreco_118_id)s,%(company)s,%(country_code)s, %(fit_community)s,%(fit_tier)s,%(warmth_score)s,%(engagement_tier)s, %(best_channel)s,%(n_interactions)s,%(confidence)s,%(contagion_flag)s) """, rows, page_size=1000) print(f"wrote {len(rows)} prospect profiles ({args.country})") return 0

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

  • [ ] Step 5: Run the unit test, verify PASS (2 passed).

Run: ./venv/Scripts/python.exe -m pytest tests/test_compute_prospect_profile.py -v

  • [ ] Step 6: Integration smoke (live PG) + commit

Run: ./venv/Scripts/python.exe -m workers.compute_prospect_profile --country GB Expected: wrote prospect profiles (GB) with N ≈ 25,762 (the prospect base count). Then verify: docker exec -e PGPASSWORD=lc_2026 lc-postgres psql -U leadcontagion -d leadcontagion -c "SELECT fit_tier, engagement_tier, COUNT(*) FROM prospect_profile GROUP BY 1,2 ORDER BY 1,2;" Expected: a small grid of counts (the 9 quadrant cells, some possibly empty).

`bash git add database/init_postgres_prospect_profile.sql workers/compute_prospect_profile.py tests/test_compute_prospect_profile.py git commit -m "feat: prospect_profile scorer (fit + coverage-aware warmth + best channel)" `

---

Task 3: Prospects tab — Fit × Engagement quadrant panel

Files:

  • Modify: dashboard/app.py (add route /panel/prospect-quadrant)
  • Create: dashboard/templates/_prospect_quadrant.html
  • Modify: dashboard/templates/home.html (new "Prospects" tab + nav)
  • Test: HTTP smoke (manual command in Step 5)

  • [ ] Step 1: Add the route — insert near the other /panel/... routes in dashboard/app.py:

`python @app.get("/panel/prospect-quadrant", response_class=HTMLResponse) def prospect_quadrant_panel(request: Request, country: str = "GB") -> HTMLResponse: """Fit x Engagement quadrant counts from prospect_profile.""" country = country.upper() if country.upper() in ("FR", "GB") else "GB" rows = crm.query( """SELECT fit_tier, engagement_tier, COUNT(*) n, ROUND(AVG(warmth_score)::numeric, 3) avg_warmth FROM prospect_profile WHERE country_code = %s GROUP BY fit_tier, engagement_tier""", (country,)) grid = {(r["fit_tier"], r["engagement_tier"]): r for r in rows} total = sum(r["n"] for r in rows) return templates.TemplateResponse(request, "_prospect_quadrant.html", {"country": country, "grid": grid, "total": total}) `

  • [ ] Step 2: Create the template dashboard/templates/_prospect_quadrant.html:

`html

{{ '{:,}'.format(total) }} {{ country }} prospects · Fit (rows) × Engagement (cols)
cold
warm
hot
{% for ft in ['high','mid','low'] %}
{{ ft }} fit
{% for et in ['cold','warm','hot'] %} {% set c = grid.get((ft,et)) %}
{{ '{:,}'.format(c.n) if c else 0 }}
{% if c %}avg {{ c.avg_warmth }}{% else %}—{% endif %}
{% endfor %} {% endfor %}
`

  • [ ] Step 3: Add the Prospects tab in dashboard/templates/home.html. Add a tab button alongside the others (
    Prospects
    ) and a tab page section:

`html

Prospect Quadrant — Fit × Engagement

`

(Match the exact tab-button markup of the sibling tabs — copy one and change data-tab/label.)

  • [ ] Step 4: Restart the dashboard service

Run: nssm restart LeadContagionDashboard

  • [ ] Step 5: HTTP smoke + commit

Run (PowerShell): build Basic-auth header for lyreco:ryu4dPIvg_-WGYz8, GET http://127.0.0.1:8001/panel/prospect-quadrant?country=GB. Expected: HTTP 200, HTML containing the 3×3 grid with non-zero total.

`bash git add dashboard/app.py dashboard/templates/_prospect_quadrant.html dashboard/templates/home.html git commit -m "feat: Prospects tab + Fit x Engagement quadrant panel" `

---

Task 4: Prospects tab — ranked list / offer card panel

Files:

  • Modify: dashboard/app.py (add route /panel/prospect-list)
  • Create: dashboard/templates/_prospect_list.html
  • Modify: dashboard/templates/home.html (add the list panel to the Prospects tab)
  • Test: HTTP smoke (Step 5)

  • [ ] Step 1: Add the route in dashboard/app.py:

`python @app.get("/panel/prospect-list", response_class=HTMLResponse) def prospect_list_panel(request: Request, country: str = "GB", fit: str = "high", eng: str = "hot", limit: int = 50) -> HTMLResponse: """Ranked prospects in a chosen Fit×Engagement cell, with the inherited offer.""" country = country.upper() if country.upper() in ("FR", "GB") else "GB" prospects = crm.query( """SELECT campaign_label, lyreco_118_id, company, fit_community, warmth_score, best_channel, n_interactions, confidence FROM prospect_profile WHERE country_code=%s AND fit_tier=%s AND engagement_tier=%s ORDER BY warmth_score DESC NULLS LAST LIMIT %s""", (country, fit, eng, limit)) # the inherited 5-SKU offer for the communities in view cids = sorted({p["fit_community"] for p in prospects if p["fit_community"] is not None}) offers: dict = {} if cids: for r in crm.query( """SELECT community_id, product_description, family_rank, product_rank FROM community_product_offer WHERE source_country=%s AND community_id = ANY(%s) ORDER BY community_id, family_rank, product_rank""", (country, cids)): offers.setdefault(r["community_id"], []).append(r["product_description"]) return templates.TemplateResponse(request, "_prospect_list.html", {"country": country, "fit": fit, "eng": eng, "prospects": prospects, "offers": offers}) `

  • [ ] Step 2: Create the template dashboard/templates/_prospect_list.html:

`html

{{ country }} · fit={{ fit }} · engagement={{ eng }} · {{ prospects|length }} shown (top by warmth)
{% for p in prospects %} {% else %} {% endfor %}
CompanyWarmthBest channelTouchesInherited offer (top SKUs)
{{ p.company or '—' }} {{ '%.2f'|format(p.warmth_score or 0) }} {{ p.best_channel or '—' }} {{ p.n_interactions }} {{ (offers.get(p.fit_community) or [])[:5]|join(' · ') or '—' }}
No prospects in this cell.
`

  • [ ] Step 3: Add the list panel to the Prospects tab section in home.html (below the quadrant panel):

`html

Prospect List — ranked by warmth, with inherited offer

`

  • [ ] Step 4: Restart the dashboard service

Run: nssm restart LeadContagionDashboard

  • [ ] Step 5: HTTP smoke + commit

Run (PowerShell): GET http://127.0.0.1:8001/panel/prospect-list?country=GB&fit=high&eng=hot with the Basic-auth header. Expected: HTTP 200 (a table, or the "No prospects in this cell" row if that cell is empty — both are valid).

`bash git add dashboard/app.py dashboard/templates/_prospect_list.html dashboard/templates/home.html git commit -m "feat: Prospect list/offer panel on Prospects tab" `

---

Self-Review

  • Spec coverage: profile scorer (spec component 6) ✓ Tasks 1–2 (coverage-aware warmth ✓ Task 1); Prospect tab (component 7) ✓ Tasks 3–4 (quadrant + list/card). Contagion flag is a nullable column, populated later (Layer 4) — out of scope by design.
  • Placeholders: none — every step has runnable code + exact command. Dashboard HTTP smokes accept an empty cell as valid (data-dependent), which is correct, not a placeholder.
  • Type consistency: warmth_score(channel_counts) -> (score, confidence) used identically in Task 2; normalize_engagement_tier (plan 1) maps the same [0,1] score the scorer emits; fit_tier/best_channel defined in Task 2 and not re-defined elsewhere; fit_community (INTEGER) joins community_product_offer.community_id.
  • Pattern adherence: routes follow the existing @app.get("/panel/X") + templates.TemplateResponse(request, "_X.html", {...}) + data-panel auto-load + refresh-button convention used by sibling panels; nssm restart LeadContagionDashboard per CLAUDE.md.
  • Coverage-aware: email-only prospects (today's only populated channel) get confidence = 1/4 and a warmth driven purely by email intensity — honest, not falsely cold.