# Prospect Segmentation — Foundation (Shrinkage + Dimensions) 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 pure, unit-tested foundation for the all-channels prospect segmentation: the volume-shrinkage functions (the load-bearing 80%) and the canonical dimension taxonomy + normalizers.
Architecture: Two dependency-free modules. core/shrinkage.py holds gridiron's four shrinkage primitives (distribution shrink, deviation shrink, volume floor, persistence→K). core/prospect_dims.py holds the canonical coarse value sets + deterministic per-source normalizers that every later stage maps into BEFORE cubing. No DB, no I/O — so both are trivially testable and reusable by the cube/scorer plans that follow.
Tech Stack: Python 3.12, pytest. (DuckDB + the cube/fact/scorer/tab come in follow-on plans 2–3.)
Scope note: This is plan 1 of 3. Plan 2 = interaction-fact assembly + DuckDB cube. Plan 3 = profile scorer + Prospect tab. Contagion overlay is deferred (droplet/OMEGA consult).
---
Task 1: Distribution-cell shrinkage (norm_shrunk)
Files:
- Create:
core/shrinkage.py - Test:
tests/test_shrinkage.py
- [ ] Step 1: Write the failing test
`python
# tests/test_shrinkage.py
from core.shrinkage import norm_shrunk
def test_norm_shrunk_thin_cell_collapses_to_parent(): parent = {"a": 0.7, "b": 0.3} out = norm_shrunk(counts={"a": 1, "b": 0}, parent=parent, n=1, K=50) # n << K -> result hugs the parent, not the 1-sample cell assert abs(out["a"] - 0.70) < 0.02 and abs(out["b"] - 0.30) < 0.02
def test_norm_shrunk_rich_cell_trusts_itself(): parent = {"a": 0.5, "b": 0.5} out = norm_shrunk(counts={"a": 900, "b": 100}, parent=parent, n=1000, K=50) assert out["a"] > 0.85 # n >> K -> own counts dominate
def test_norm_shrunk_renormalizes_to_one():
out = norm_shrunk(counts={"a": 3, "b": 1}, parent={"a": 0.5, "b": 0.5}, n=4, K=10)
assert abs(sum(out.values()) - 1.0) < 1e-6
`
- [ ] Step 2: Run test to verify it fails
Run: ./venv/Scripts/python.exe -m pytest tests/test_shrinkage.py -v
Expected: FAIL — ModuleNotFoundError: No module named 'core.shrinkage' (or ImportError).
- [ ] Step 3: Write minimal implementation
`python
# core/shrinkage.py
"""Volume-shrinkage primitives for sparse multi-dimensional segmentation.
Pattern + reference code from gridiron (mesh DM #1418); reframe from lab-ovh (#1412). A thin cell is shrunk toward its parent marginal by volume so the combinatorial blow-up of fit x engagement x channel x product never produces noise that surfaces as signal. Pure functions, no I/O — keep them that way. """ from __future__ import annotations
def norm_shrunk(counts: dict, parent: dict, n: int, K: float) -> dict:
"""Shrink a distribution cell toward a parent PROBABILITY distribution:
out[k] = (count_k + K*parent_k) / (n + K), then renormalize.
thin cell (n<`
- [ ] Step 4: Run test to verify it passes
Run: ./venv/Scripts/python.exe -m pytest tests/test_shrinkage.py -v
Expected: PASS (3 passed).
- [ ] Step 5: Commit
`bash
git add core/shrinkage.py tests/test_shrinkage.py
git commit -m "feat: norm_shrunk distribution shrinkage primitive"
`
---
Task 2: Continuous deviation shrinkage (value_oa)
Files:
- Modify:
core/shrinkage.py - Test:
tests/test_shrinkage.py
- [ ] Step 1: Write the failing test
`python
from core.shrinkage import value_oa
def test_value_oa_thin_cell_shrinks_to_zero_deviation(): # n -> small vs K: deviation from base collapses toward 0 (== toward parent mean) assert abs(value_oa(cell_mean=10.0, base=4.0, n=1, K=99)) < 0.1
def test_value_oa_rich_cell_keeps_deviation():
# n >> K: keeps almost the full (cell_mean - base) deviation
assert value_oa(cell_mean=10.0, base=4.0, n=9000, K=100) > 5.9
`
- [ ] Step 2: Run test to verify it fails
Run: ./venv/Scripts/python.exe -m pytest tests/test_shrinkage.py::test_value_oa_thin_cell_shrinks_to_zero_deviation -v
Expected: FAIL — ImportError: cannot import name 'value_oa'.
- [ ] Step 3: Write minimal implementation
`python
# append to core/shrinkage.py
def value_oa(cell_mean: float, base: float, n: int, K: float) -> float:
"""Shrink a continuous metric's deviation from its parent (base) mean:
(cell_mean - base) * n/(n+K). Shrinking toward 0 == toward the parent."""
return round((cell_mean - base) * n / (n + K), 4)
`
- [ ] Step 4: Run test to verify it passes
Run: ./venv/Scripts/python.exe -m pytest tests/test_shrinkage.py -v
Expected: PASS (5 passed).
- [ ] Step 5: Commit
`bash
git add core/shrinkage.py tests/test_shrinkage.py
git commit -m "feat: value_oa continuous deviation shrinkage"
`
---
Task 3: Persistence → per-dimension K (persistence, k_from_persistence, carry)
Files:
- Modify:
core/shrinkage.py - Test:
tests/test_shrinkage.py
- [ ] Step 1: Write the failing test
`python
from core.shrinkage import persistence, k_from_persistence, carry
def test_persistence_perfect_autocorrelation_is_one(): series = {"e1": [1.0, 2.0, 3.0], "e2": [2.0, 4.0, 6.0]} # consecutive pairs perfectly correlated assert persistence(series) > 0.99
def test_k_inverse_to_persistence(): # high persistence -> trust the cell (small K); low persistence -> regress hard (big K) assert k_from_persistence(1.0, k_lo=60, k_hi=300) == 60 assert k_from_persistence(0.0, k_lo=60, k_hi=300) == 300 assert k_from_persistence(0.5, k_lo=60, k_hi=300) == 180
def test_carry_blends_prior_toward_parent_by_rho():
assert carry(prior=10.0, parent=4.0, rho=0.0) == 4.0 # no persistence -> all parent
assert carry(prior=10.0, parent=4.0, rho=1.0) == 10.0 # full persistence -> keep prior
`
- [ ] Step 2: Run test to verify it fails
Run: ./venv/Scripts/python.exe -m pytest tests/test_shrinkage.py -k "persistence or carry or k_inverse" -v
Expected: FAIL — ImportError: cannot import name 'persistence'.
- [ ] Step 3: Write minimal implementation
`python
# append to core/shrinkage.py
import statistics as _st
def persistence(series_by_entity: dict) -> float: """Signal persistence per dimension = correlation of consecutive-period values across entities (your YoY / split-half). Feeds K and the walk-up order.""" xs, ys = [], [] for v in series_by_entity.values(): for a, b in zip(v, v[1:]): xs.append(a); ys.append(b) if len(xs) < 2: return 0.0 try: return _st.correlation(xs, ys) except _st.StatisticsError: # zero variance return 0.0
def k_from_persistence(rho: float, k_lo: float = 60, k_hi: float = 300) -> float: """Set K INVERSELY to persistence. rho->1 : small K (trust the cell); rho->0 : big K (regress hard toward parent).""" rho = max(0.0, min(1.0, rho)) return k_hi - (k_hi - k_lo) * rho
def carry(prior: float, parent: float, rho: float) -> float:
"""Pull a prior-period value toward the parent by (1-rho) before blending."""
return parent + rho * (prior - parent)
`
- [ ] Step 4: Run test to verify it passes
Run: ./venv/Scripts/python.exe -m pytest tests/test_shrinkage.py -v
Expected: PASS (8 passed).
- [ ] Step 5: Commit
`bash
git add core/shrinkage.py tests/test_shrinkage.py
git commit -m "feat: persistence-driven per-dimension K + period carry"
`
---
Task 4: Volume floor (passes_floor)
Files:
- Modify:
core/shrinkage.py - Test:
tests/test_shrinkage.py
- [ ] Step 1: Write the failing test
`python
from core.shrinkage import passes_floor
def test_passes_floor_gates_thin_cells():
floors = {"web": 30, "email": 20}
assert passes_floor("web", n=45, floors=floors) is True
assert passes_floor("web", n=12, floors=floors) is False
# unknown dim falls back to the provided default
assert passes_floor("social", n=5, floors=floors, default=10) is False
assert passes_floor("social", n=11, floors=floors, default=10) is True
`
- [ ] Step 2: Run test to verify it fails
Run: ./venv/Scripts/python.exe -m pytest tests/test_shrinkage.py -k passes_floor -v
Expected: FAIL — ImportError: cannot import name 'passes_floor'.
- [ ] Step 3: Write minimal implementation
`python
# append to core/shrinkage.py
def passes_floor(dim_key: str, n: int, floors: dict, default: int = 0) -> bool:
"""Volume gate applied BEFORE shrinking — without it, thin cells still surface
spuriously on top charts (gridiron gotcha d)."""
return n >= floors.get(dim_key, default)
`
- [ ] Step 4: Run test to verify it passes
Run: ./venv/Scripts/python.exe -m pytest tests/test_shrinkage.py -v
Expected: PASS (9 passed).
- [ ] Step 5: Commit
`bash
git add core/shrinkage.py tests/test_shrinkage.py
git commit -m "feat: passes_floor volume gate"
`
---
Task 5: Canonical dimension taxonomy + normalizers (core/prospect_dims.py)
Files:
- Create:
core/prospect_dims.py - Test:
tests/test_prospect_dims.py
Rationale: gridiron's #1 lesson — lock a canonical coarse taxonomy + deterministic per-source normalizer and map everything into it BEFORE cubing; retrofitting is the expensive path. Start with the four core dims we already have signal for; high-cardinality drift gets mapped to canonical buckets, unknowns to an explicit "UNK" (never silently dropped).
- [ ] Step 1: Write the failing test
`python
# tests/test_prospect_dims.py
from core.prospect_dims import normalize_channel, normalize_engagement_tier, CANONICAL
def test_channel_synonyms_map_to_canonical(): assert normalize_channel("eloqua") == "email" assert normalize_channel("E-mail") == "email" assert normalize_channel("organic") == "web" assert normalize_channel("linkedin") == "social" assert normalize_channel("rep visit") == "offline"
def test_unknown_channel_is_explicit_unk_not_dropped(): assert normalize_channel("carrier-pigeon") == "UNK" assert normalize_channel(None) == "UNK"
def test_engagement_tier_buckets_by_score(): assert normalize_engagement_tier(0.0) == "cold" assert normalize_engagement_tier(0.5) == "warm" assert normalize_engagement_tier(0.9) == "hot"
def test_canonical_sets_are_declared():
assert set(CANONICAL["channel"]) == {"web", "email", "social", "offline", "UNK"}
`
- [ ] Step 2: Run test to verify it fails
Run: ./venv/Scripts/python.exe -m pytest tests/test_prospect_dims.py -v
Expected: FAIL — ModuleNotFoundError: No module named 'core.prospect_dims'.
- [ ] Step 3: Write minimal implementation
`python
# core/prospect_dims.py
"""Canonical coarse dimension taxonomy + deterministic per-source normalizers
for the prospect segmentation cube. Everything maps into these BEFORE cubing.
Unknown/unresolvable values map to explicit "UNK" — modelled honestly at the collapsed level, never silently dropped (gridiron lesson; coverage-aware). """ from __future__ import annotations
CANONICAL = { "channel": ["web", "email", "social", "offline", "UNK"], "engagement_tier": ["cold", "warm", "hot"], "fit_tier": ["low", "mid", "high"], }
_CHANNEL_MAP = { "web": "web", "organic": "web", "direct": "web", "site": "web", "ga4": "web", "email": "email", "e-mail": "email", "eloqua": "email", "mail": "email", "social": "social", "linkedin": "social", "meta": "social", "facebook": "social", "instagram": "social", "offline": "offline", "rep": "offline", "rep visit": "offline", "call": "offline", "edi": "offline", "sales": "offline", }
def normalize_channel(raw: str | None) -> str: if not raw: return "UNK" return _CHANNEL_MAP.get(raw.strip().lower(), "UNK")
def normalize_engagement_tier(score: float) -> str:
"""Blended warmth score in [0,1] -> coarse tier."""
if score >= 0.66:
return "hot"
if score >= 0.33:
return "warm"
return "cold"
`
- [ ] Step 4: Run test to verify it passes
Run: ./venv/Scripts/python.exe -m pytest tests/test_prospect_dims.py -v
Expected: PASS (4 passed).
- [ ] Step 5: Commit
`bash
git add core/prospect_dims.py tests/test_prospect_dims.py
git commit -m "feat: canonical prospect dimension taxonomy + channel normalizer"
`
---
Self-Review
- Spec coverage (foundation slice): shrinkage primitives (spec component 4) ✓ Tasks 1–4; persistence→K + walk-up basis (component 5) ✓ Task 3; canonical taxonomy + normalizers (component 1) ✓ Task 5. Cube/fact/scorer/tab (components 2,3,6,7) are explicitly out of scope → plans 2–3.
- Placeholders: none — every step has runnable test + impl code + exact command.
- Type consistency:
Kisfloatacrossnorm_shrunk/value_oa/k_from_persistence;norm_shrunkparent = probability dist (sums to 1),value_oabase = scalar mean — distinct by design, documented in docstrings. - Engagement tier thresholds (Task 5) align with the warmth score [0,1] the Plan-3 scorer will emit; restated here so Task 5 is self-contained.