Firmographic Cluster Semantic Labels — Implementation Plan

# Firmographic Cluster Semantic Labels — 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: Give each firmographic cluster a marketer-friendly name + one-line "what it's made of" description, generated hybrid (deterministic stat profile → batched claude -p phrasing), stored in firmographic_community_labels, and surfaced in the cluster report + Buying Profile panel.

Architecture: A pure stat profiler (core/cluster_labels.py) turns a cluster's member firmographic features into a factual profile (dominant sector + share, spend tier, size band, region, churn). A worker fetches members per cluster (Postgres-only), profiles them, sends all profiles in ONE batched claude -p call (reusing core/ai_typing.get_llm_backend — subscription billing + JSON parse), falls back to a rule-based name on failure, and writes labels. Two consumers join the labels table.

Tech Stack: Python 3.12, psycopg2, PostgreSQL, claude -p via the existing core/ai_typing CLI backend, pytest.

Spec: docs/superpowers/specs/2026-06-11-firmographic-cluster-labels-design.md

---

File structure

| File | Responsibility | New/Modify | |---|---|---| | core/cluster_labels.py | Pure profiler + spend/size banding + rule-based fallback label. No DB/LLM. | Create | | tests/test_cluster_labels.py | Hermetic tests for the profiler + fallback. | Create | | database/init_postgres_cluster_labels.sql | firmographic_community_labels DDL. | Create | | workers/label_firmographic_communities.py | Orchestrator: fetch → profile → batched LLM → write. | Create | | scripts/export_buying_profiles.py | Cluster report: join labels into the heading. | Modify | | dashboard/app.py | Buying Profile panel: add account's cluster name to context. | Modify | | dashboard/templates/_buying_profile.html | Render the cluster name in the header. | Modify |

---

Task 1: Pure profiler + bands + rule fallback

Files:

• Create: core/cluster_labels.py
• Test: tests/test_cluster_labels.py

• [ ] Step 1: Write the failing test

Create tests/test_cluster_labels.py:

`python from core.cluster_labels import profile_cluster, rule_fallback_label

def _m(sector="prof services", spend=5000, users=3, region="South-East", churn=False): return {"sector": sector, "annual_spend": spend, "n_users": users, "region": region, "is_churn": churn}

def test_profile_basic_shares_and_bands(): rows = [_m(spend=5000, users=3) for _ in range(6)] + [_m(sector="retail", spend=5000) for _ in range(4)] p = profile_cluster(rows) assert p["member_count"] == 10 assert p["top_sector"] == "prof services" assert p["top_sector_share"] == 0.6 assert p["spend_tier"] == "small" # median 5000 -> [1k,10k) assert p["size_band"] == "small" # median 3 -> [2,5] assert p["dominant_region"] == "South-East" assert p["churn_share"] == 0.0

def test_spend_tiers_and_size_bands(): assert profile_cluster([_m(spend=500, users=1)])["spend_tier"] == "micro" assert profile_cluster([_m(spend=20000, users=10)])["spend_tier"] == "mid" assert profile_cluster([_m(spend=80000, users=40)])["spend_tier"] == "large" assert profile_cluster([_m(users=1)])["size_band"] == "solo" assert profile_cluster([_m(users=40)])["size_band"] == "large"

def test_churn_share_and_nulls(): rows = [_m(churn=True), _m(churn=False), _m(churn=True), _m(churn=False)] assert profile_cluster(rows)["churn_share"] == 50.0 # tolerate missing numeric fields p = profile_cluster([{"sector": "x"}]) assert p["spend_tier"] == "unknown" and p["size_band"] == "unknown"

def test_rule_fallback_label_uses_only_facts(): p = profile_cluster([_m(spend=5000, users=3) for _ in range(8)]) lab = rule_fallback_label(p) assert lab["name"] and lab["description"] assert "prof services" in lab["description"].lower() assert "small" in lab["description"].lower() `

• [ ] Step 2: Run to verify it fails

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

• [ ] Step 3: Implement

Create core/cluster_labels.py:

`python """Pure firmographic-cluster profiling — no DB, no LLM. Turns a cluster's member feature rows into a factual profile (the deterministic 'what it's made of'), plus a rule-based fallback label used when the LLM phrasing step fails.""" from __future__ import annotations

from collections import Counter from statistics import median

# (band_name, exclusive_upper_bound); None upper = catch-all top band. SPEND_TIERS = [("micro", 1_000), ("small", 10_000), ("mid", 50_000), ("large", None)] SIZE_BANDS = [("solo", 2), ("small", 6), ("mid", 21), ("large", None)]

def _band(value: float, bands: list[tuple[str, float | None]]) -> str: for name, hi in bands: if hi is None or value < hi: return name return bands[-1][0]

def profile_cluster(rows: list[dict]) -> dict: """rows: member dicts with sector, annual_spend, n_users, region, is_churn (any numeric may be missing). Returns the deterministic profile.""" n = len(rows) if n == 0: return {"member_count": 0, "top_sector": "unknown", "top_sector_share": 0.0, "spend_tier": "unknown", "size_band": "unknown", "dominant_region": "unknown", "dominant_region_share": 0.0, "churn_share": 0.0} sectors = Counter((r.get("sector") or "unknown") for r in rows) top_sector, top_n = sectors.most_common(1)[0] regions = Counter((r.get("region") or "unknown") for r in rows) top_region, region_n = regions.most_common(1)[0] spends = [float(r["annual_spend"]) for r in rows if r.get("annual_spend") is not None] users = [float(r["n_users"]) for r in rows if r.get("n_users") is not None] churn = sum(1 for r in rows if r.get("is_churn")) return { "member_count": n, "top_sector": top_sector, "top_sector_share": round(top_n / n, 3), "spend_tier": _band(median(spends), SPEND_TIERS) if spends else "unknown", "size_band": _band(median(users), SIZE_BANDS) if users else "unknown", "dominant_region": top_region, "dominant_region_share": round(region_n / n, 3), "churn_share": round(100.0 * churn / n, 2), }

def rule_fallback_label(p: dict) -> dict: """Deterministic name + description from a profile (LLM-free fallback).""" name = f"{p['spend_tier'].title()}-spend {p['top_sector']}"[:60] desc = (f"{p['member_count']} accounts, {int(round(p['top_sector_share'] * 100))}% " f"{p['top_sector']}, {p['spend_tier']} spend, {p['size_band']} size, " f"mostly {p['dominant_region']}.") return {"name": name, "description": desc} `

• [ ] Step 4: Run to verify it passes

Run: venv\Scripts\python.exe -m pytest tests/test_cluster_labels.py -v Expected: 4 passed.

• [ ] Step 5: Commit

`bash git add core/cluster_labels.py tests/test_cluster_labels.py git commit -m "feat(cluster-labels): pure profiler + bands + rule fallback" `

Task 2: Labels table DDL

Files:

• Create: database/init_postgres_cluster_labels.sql

• [ ] Step 1: Write the DDL

Create database/init_postgres_cluster_labels.sql:

`sql -- Human-readable labels for firmographic clusters (account_firmographic_community). CREATE TABLE IF NOT EXISTS firmographic_community_labels ( source_country TEXT NOT NULL, community_id INTEGER NOT NULL, name TEXT NOT NULL, description TEXT NOT NULL, member_count INTEGER NOT NULL, top_sector TEXT, spend_tier TEXT, size_band TEXT, dominant_region TEXT, churn_share NUMERIC(5,2), generated_by TEXT NOT NULL DEFAULT 'llm', -- 'llm' | 'rule_fallback' last_computed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), PRIMARY KEY (source_country, community_id) ); `

• [ ] Step 2: Apply to the live DB (controller/Pierre — docker exec is gated)

Run: ` docker exec -i lc-postgres psql -U leadcontagion -d leadcontagion < database/init_postgres_cluster_labels.sql ` Expected: CREATE TABLE.

• [ ] Step 3: Commit

`bash git add database/init_postgres_cluster_labels.sql git commit -m "feat(cluster-labels): firmographic_community_labels DDL" `

Task 3: The labeling worker

Files:

• Create: workers/label_firmographic_communities.py

• [ ] Step 1: Write the worker

Create workers/label_firmographic_communities.py:

`python """Label firmographic clusters: profile each community's members, then a single batched claude -p call writes a marketer name + 1-line description. Rule-based fallback on LLM failure. Idempotent per source_country. Postgres-only.

Reproducibility: KMeans community_id is not stable across rebuilds — re-run this after workers.community_match_soft.""" from __future__ import annotations

import argparse import json import logging 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.ai_typing import get_llm_backend, _parse_json_obj_list from core.cluster_labels import profile_cluster, rule_fallback_label from database.crm_db import CRMDatabase

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") log = logging.getLogger("cluster_labels")

LABEL_PROMPT = """You name firmographic customer clusters for B2B marketers at Lyreco. Each cluster has a FACTUAL profile (dominant sector + share, spend tier, size band, region, churn). Give each cluster a short NAME (<=4 words, Title Case) and a one-sentence DESCRIPTION in plain marketer language. Use ONLY the facts provided — never invent attributes.

Clusters (JSON): {clusters}

Respond ONLY with a JSON array [{{"community_id": , "name": "", "description": ""}}] covering EVERY community_id, with no text around it."""

def _fetch_members(crm: CRMDatabase, country: str) -> dict[int, list[dict]]: rows = crm.query( """ SELECT a.community_id, COALESCE(NULLIF(f.accounts_category, ''), sec.meta_sector, f.sic_group) AS sector, f.annual_spend, f.n_users, COALESCE(NULLIF(f.postcode_area, ''), f.region_key) AS region, f.is_churn FROM account_firmographic_community a JOIN customer_firmographic_features f ON f.account_number = a.account_number AND f.source_country = a.source_country LEFT JOIN LATERAL ( SELECT meta_sector FROM sic_meta_sectors s WHERE s.sic_2digit = f.sic_2digit LIMIT 1 ) sec ON TRUE WHERE a.source_country = %s """, (country,), ) by_cluster: dict[int, list[dict]] = {} for r in rows: by_cluster.setdefault(r["community_id"], []).append(dict(r)) return by_cluster

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

by_cluster = _fetch_members(crm, args.country) if not by_cluster: log.warning("no clusters for %s — run community_match_soft first", args.country) return 0 profiles = {cid: profile_cluster(rows) for cid, rows in by_cluster.items()} log.info("%s: %d clusters profiled", args.country, len(profiles))

# one batched LLM call; rule-based fallback per missing/failed cluster llm_input = [{"community_id": cid, p} for cid, p in profiles.items()] named: dict[int, dict] = {} try: llm_call = get_llm_backend(prefer="cli", model=args.model) out = _parse_json_obj_list(llm_call( LABEL_PROMPT.format(clusters=json.dumps(llm_input, ensure_ascii=False)))) for item in out: cid = item.get("community_id") if cid in profiles and item.get("name") and item.get("description"): named[cid] = {"name": str(item["name"])[:80], "description": str(item["description"])[:300], "generated_by": "llm"} except Exception as exc: log.warning("LLM labeling failed (%s) — using rule fallback for all", exc)

rows_out = [] for cid, p in profiles.items(): lab = named.get(cid) if lab is None: fb = rule_fallback_label(p) lab = {"name": fb["name"], "description": fb["description"], "generated_by": "rule_fallback"} rows_out.append({ "source_country": args.country, "community_id": cid, "name": lab["name"], "description": lab["description"], "member_count": p["member_count"], "top_sector": p["top_sector"], "spend_tier": p["spend_tier"], "size_band": p["size_band"], "dominant_region": p["dominant_region"], "churn_share": p["churn_share"], "generated_by": lab["generated_by"], })

with crm.connect() as conn, conn.cursor() as cur: cur.execute("DELETE FROM firmographic_community_labels WHERE source_country=%s", (args.country,)) psycopg2.extras.execute_batch(cur, """ INSERT INTO firmographic_community_labels (source_country, community_id, name, description, member_count, top_sector, spend_tier, size_band, dominant_region, churn_share, generated_by) VALUES (%(source_country)s, %(community_id)s, %(name)s, %(description)s, %(member_count)s, %(top_sector)s, %(spend_tier)s, %(size_band)s, %(dominant_region)s, %(churn_share)s, %(generated_by)s) """, rows_out, page_size=200) n_llm = sum(1 for r in rows_out if r["generated_by"] == "llm") log.info("wrote %d labels (%d llm, %d fallback)", len(rows_out), n_llm, len(rows_out) - n_llm) return 0

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

• [ ] Step 2: Parse + import check (no DB / no LLM)

Run: venv\Scripts\python.exe -c "import ast; ast.parse(open(r'workers/label_firmographic_communities.py',encoding='utf-8').read()); print('parse OK')" Expected: parse OK.

• [ ] Step 3: Live run (controller/Pierre — claude -p is subscription-billed + DB write)

Run: venv\Scripts\python.exe -m workers.label_firmographic_communities --country GB Expected: log GB: 60 clusters profiled then wrote 60 labels (N llm, M fallback).

• [ ] Step 4: Sanity-check the labels

Run: ` docker exec lc-postgres psql -U leadcontagion -d leadcontagion -c "SELECT community_id, name, left(description,60), member_count, generated_by FROM firmographic_community_labels WHERE source_country='GB' ORDER BY member_count DESC LIMIT 8;" ` Expected: readable names + descriptions, mostly generated_by='llm'.

• [ ] Step 5: Commit

`bash git add workers/label_firmographic_communities.py git commit -m "feat(cluster-labels): batched LLM labeling worker (rule fallback)" `

Task 4: Surface labels in the report + panel

Files:

• Modify: scripts/export_buying_profiles.py
• Modify: dashboard/app.py (buying_profile_panel)
• Modify: dashboard/templates/_buying_profile.html

• [ ] Step 1: Report — join labels into the cluster heading

In scripts/export_buying_profiles.py, the cluster query selects c.cluster_id ... and the render prints ### Firmographic cluster {cur_cluster} (...). Change the query to also fetch the label and use it. Replace the rows = crm.query("""... """) block's SELECT/JOIN with:

`python rows = crm.query( """ SELECT c.cluster_id, c.product_reference, c.rank_freq, c.orders_count, c.units_qty, c.n_accounts, p.product_description, l.name AS cluster_name, l.description AS cluster_desc FROM cluster_buying_profile_weekly c LEFT JOIN ecom_products p ON p.product_reference = c.product_reference AND p.source_country = c.source_country LEFT JOIN firmographic_community_labels l ON l.community_id = c.cluster_id AND l.source_country = c.source_country WHERE c.cluster_source='firmographic' AND c.snapshot_week=%s AND c.rank_freq <= 10 ORDER BY c.cluster_id, c.rank_freq """, (latest,)) `

Then change the cluster heading print from: `python print(f"\n### Firmographic cluster {cur_cluster} ({r['n_accounts']} accounts)\n") ` to: `python _title = r["cluster_name"] or f"Firmographic cluster {cur_cluster}" _desc = f" — {r['cluster_desc']}" if r["cluster_desc"] else "" print(f"\n### {_title} ({r['n_accounts']} accounts){_desc}\n") `

• [ ] Step 2: Verify the report still runs

Run: venv\Scripts\python.exe scripts\export_buying_profiles.py Expected: a RESULT_MARKDOWN block where cluster headings show names (once labels exist; before that they fall back to "Firmographic cluster N").

• [ ] Step 3: Panel — fetch the account's cluster label

In dashboard/app.py, inside buying_profile_panel, after acc_no = account["account_number"] and before building the context, add:

`python cluster_label = crm.query( """ SELECT l.name, l.description FROM account_firmographic_community a JOIN firmographic_community_labels l ON l.community_id = a.community_id AND l.source_country = a.source_country WHERE a.account_number=%s AND a.source_country=%s LIMIT 1 """, (acc_no, country), ) cluster_label = cluster_label[0] if cluster_label else None ` Then add "cluster_label": cluster_label, to BOTH the success TemplateResponse context dict AND (as "cluster_label": None,) the ctx_empty dict and the no_match return, so the key always exists.

• [ ] Step 4: Panel template — show the cluster name

In dashboard/templates/_buying_profile.html, inside the