WISE Market-Location Prospect-Conversion Report — Implementation Plan

# WISE Market-Location Prospect-Conversion Report — 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 v1 "spine" of a WISE-only report that tracks each market-location cold-email company from send → dual-journey activation → 1st/2nd purchase (timing, amounts, product mix) with its pre-conversion prospect model level, delivered as a refreshable HTML report.

Architecture: Four scripts run in sequence — SFTP-fetch+ingest the daily xlsx into eloqua_campaign_responders; refresh the CH/BvD/community match; compute the cross-DB per-company spine + cohort rollups; render a self-contained HTML report (per-company table + cohort rollups + data-driven journey-map). Pure logic (parse/map/timeline/rollup/SVG) is unit-tested; DB steps are integration-verified against the live Docker DBs.

Tech Stack: Python 3.12 venv, openpyxl (xlsx), paramiko (SFTP), psycopg2 via database/crm_db.py::CRMDatabase (Postgres leadcontagion) and database/timeseries_db.py::TimeseriesDatabase (TimescaleDB lc_timeseries), pytest.

Spec: docs/superpowers/specs/2026-07-07-wise-marketloc-conversion-report-design.md

Global Constraints

• WISE-only: every account/order query filters source_country = 'GB'.
• Grain: one report row per company = lyreco_118_id.
• Cross-DB, never joined in one query: responders/bridge/accounts/hubspot/prospect-model in Postgres; ecom_order_lines in TimescaleDB. Pull account_numbers from PG, then query TS with WHERE soldto_number = ANY(%s).
• Account key: soldto_number = account_number, LPAD-10 (LPAD(x,10,'0')) on both sides of any join.
• Order hygiene: order_channel_code = 'W' and sales_amount BETWEEN 0 AND 1e7 (excludes the corrupt channel-C mega-lines).
• Idempotent: ingest and match refresh are safe to re-run daily (dedupe / upsert, no duplicate rows).
• Match honesty: every matched company records match_source ∈ {bvd, companies_house, community} + confidence; unmatched companies are kept and reported as "reached, unlinked," never dropped.
• Encoding: scripts read/write UTF-8; run with PYTHONIOENCODING=utf-8 when invoked as a report subprocess.
• Repo-root bootstrap at top of every scripts/*.py: import sys, pathlib; sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent)).

---

File Structure

• scripts/marketloc_parse.py — pure parse/normalize of a market-location xlsx → responder-column dicts (+ derive_campaign_label, derive_email_version, derive_program). No I/O beyond reading the given path. (Task 1)
• integrations/market_location_sftp.py — fetch_new_files(local_dir): pull *.xlsx from MARKET_LOCATION_{HOST,USER,PASSWORD} into local_dir, skipping already-present names. (Task 2)
• scripts/ingest_market_location.py — orchestrate fetch → parse each file → upsert into eloqua_campaign_responders. (Task 3)
• scripts/refresh_marketloc_match.py — for the ingested lyreco_118_ids, populate eloqua_responder_bridge (BvD) and fall back to CH/community, stamping match_source. (Task 4)
• scripts/marketloc_spine.py — pure per-company assembly helpers: extract_purchases, compute_timeline, classify_activation. (Task 5)
• scripts/compute_marketloc_conversion.py — cross-DB orchestration using Task-5 helpers → per-company rows + rollup_cohorts. (Task 5)
• scripts/marketloc_report.py — render_journey_map (SVG) + build_report_html + write to exports/ + register report. (Task 6)
• Tests: tests/test_marketloc_parse.py, tests/test_market_location_sftp.py, tests/test_marketloc_spine.py, tests/test_marketloc_report.py.

---

Task 1: Parse & normalize the xlsx (pure logic)

Files:

• Create: scripts/marketloc_parse.py
• Test: tests/test_marketloc_parse.py

Interfaces:

• Produces: parse_market_location_file(path: str) -> list[dict] (each dict keyed by eloqua_campaign_responders column names); derive_campaign_label(campaign_name: str) -> str; derive_program(campaign_name: str) -> str | None; derive_email_version(campaign_name: str, subject: str) -> str | None.

Column mapping (xlsx header → table column): Sent Date→sent_date, Campaign Name→campaign_name, Subject→subject, TemplateSubject→template_subject, Contact Urn→lead_tracker_id, Town→town, Employees Band Desc→employees_band, National Emp Band Desc→national_emp_band, Sic 07 2 Digit→sic_07_2digit, Sic 07 2 Digit Desc→sic_07_2digit_desc, Sic 07→sic_07, Sic 07 Desc→sic_07_desc, Reference→reference, Title→title, First Name→first_name, Last Name→last_name, Job Title→job_title, Company→company, Telephone→telephone, Address Line 1/2/3→address_line_1/2/3, City→city, Post Code→post_code, Market Sector→market_sector, Employees→employees, Web Address→web_address, Open Count→open_count, Open Last→open_last, TPS→tps_flag, CTPS→ctps_flag, OptOut Date→optout_date, Click Count→click_count, Click Last→click_last, Easy Desc→easy_desc, 118 ID→lyreco_118_id. Derived: campaign_label, program, email_version.

• [ ] Step 1: Write the failing tests

Create a tiny fixture xlsx in the test (via openpyxl) with the 38 headers + one row mirroring the real data, then:

`python # tests/test_marketloc_parse.py import openpyxl, pathlib from scripts.marketloc_parse import ( parse_market_location_file, derive_campaign_label, derive_program, derive_email_version)

HEADERS = ['Sent Date','Campaign Name','Subject','Template Desc','TemplateSubject','Email Domain', 'Contact Urn','Town','Employees Band Desc','National Emp Band Desc','Sic 07 2 Digit', 'Sic 07 2 Digit Desc','Sic 07','Sic 07 Desc','Reference','Title','First Name','Last Name', 'Job Title','Company','Telephone','Address Line 1','Address Line 2','Address Line 3','City', 'Post Code','Market Sector','Employees','Web Address','Open Count','Open Last','TPS','CTPS', 'OptOut Date','Click Count','Click Last','Easy Desc','118 ID'] ROW = ['2026-06-05 11:16:30','[supplies4work-email] AD_B_2026_06_05_10','Welcome','[supplies4work-email] AD_B', 'Welcome','carnegiesteel.co.uk','6690618','Westerham','C: 10 to 19 employees','A: 1 to 20 nat. emps.', '25','Manufacture','25110','Metal structures','3887154|6690618','Mr','David','Mindham','MD', 'Carnegie Steel','01959572501','The Beeches','Old Lane','Tatsfield','Kent','TN16 2LH','Manufacturing', '12','www.carnegie.co.uk','6','2026-07-02 09:19:08','False','False',None,'6','2026-07-02 09:19:08', 'Steel Buildings','3887154']

def _fixture(tmp_path): wb = openpyxl.Workbook(); ws = wb.active; ws.title = 'Data' ws.append(HEADERS); ws.append(ROW) p = tmp_path / 'ml.xlsx'; wb.save(p); return str(p)

def test_parse_maps_columns_and_types(tmp_path): rows = parse_market_location_file(_fixture(tmp_path)) assert len(rows) == 1 r = rows[0] assert r['lyreco_118_id'] == 3887154 # int, from '118 ID' assert r['lead_tracker_id'] == 6690618 # int, from 'Contact Urn' assert r['company'] == 'Carnegie Steel' assert r['post_code'] == 'TN16 2LH' assert r['open_count'] == 6 and r['click_count'] == 6 assert r['tps_flag'] is False and r['ctps_flag'] is False assert r['sent_date'].year == 2026 and r['sent_date'].month == 6 assert r['optout_date'] is None assert r['market_sector'] == 'Manufacturing'

def test_derivations(): cn = '[supplies4work-email] AD_B_2026_06_05_10' assert derive_program(cn) == 'supplies4work-email' assert derive_campaign_label(cn) == cn.strip() assert derive_email_version('x','V1_EMAIL_F_ML_OFFICE_FINANCE promo') == 'V1_EMAIL_F_ML_OFFICE_FINANCE' assert derive_email_version('no version here','plain subject') is None

def test_blank_118id_row_skipped(tmp_path): wb = openpyxl.load_workbook(_fixture(tmp_path)); ws = wb.active bad = list(ROW); bad[-1] = None; ws.append(bad); wb.save(tmp_path / 'ml.xlsx') rows = parse_market_location_file(str(tmp_path / 'ml.xlsx')) assert len(rows) == 1 # the no-118-ID row is dropped `

• [ ] Step 2: Run to verify they fail — venv/Scripts/python.exe -m pytest tests/test_marketloc_parse.py -v → FAIL (module not found).

• [ ] Step 3: Implement scripts/marketloc_parse.py

`python # -- coding: utf-8 -- """Pure parse/normalize of a market-location responder xlsx -> eloqua_campaign_responders dicts.""" from __future__ import annotations import re, datetime, pathlib, sys sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent)) import openpyxl

_HDR_TO_COL = { 'Sent Date':'sent_date','Campaign Name':'campaign_name','Subject':'subject', 'TemplateSubject':'template_subject','Contact Urn':'lead_tracker_id','Town':'town', 'Employees Band Desc':'employees_band','National Emp Band Desc':'national_emp_band', 'Sic 07 2 Digit':'sic_07_2digit','Sic 07 2 Digit Desc':'sic_07_2digit_desc','Sic 07':'sic_07', 'Sic 07 Desc':'sic_07_desc','Reference':'reference','Title':'title','First Name':'first_name', 'Last Name':'last_name','Job Title':'job_title','Company':'company','Telephone':'telephone', 'Address Line 1':'address_line_1','Address Line 2':'address_line_2','Address Line 3':'address_line_3', 'City':'city','Post Code':'post_code','Market Sector':'market_sector','Employees':'employees', 'Web Address':'web_address','Open Count':'open_count','Open Last':'open_last','TPS':'tps_flag', 'CTPS':'ctps_flag','OptOut Date':'optout_date','Click Count':'click_count','Click Last':'click_last', 'Easy Desc':'easy_desc','118 ID':'lyreco_118_id', } _INT_COLS = {'lyreco_118_id','lead_tracker_id','employees','open_count','click_count'} _DT_COLS = {'sent_date','open_last','click_last','optout_date'} _BOOL_COLS = {'tps_flag','ctps_flag'} _VERSION_RE = re.compile(r'V\d+_EMAIL_[A-Z]_ML_[A-Z_]+') _PROGRAM_RE = re.compile(r'^\[([^\]]+)\]')

def derive_program(campaign_name: str | None) -> str | None: if not campaign_name: return None m = _PROGRAM_RE.match(campaign_name.strip()) return m.group(1) if m else None

def derive_campaign_label(campaign_name: str | None) -> str: return (campaign_name or '').strip()

def derive_email_version(campaign_name: str | None, subject: str | None) -> str | None: for s in (campaign_name, subject): if s: m = _VERSION_RE.search(s) if m: return m.group(0) return None

def _to_int(v): if v in (None, ''): return None try: return int(str(v).strip()) except (TypeError, ValueError): return None

def _to_dt(v): if v in (None, ''): return None if isinstance(v, datetime.datetime): return v for fmt in ('%Y-%m-%d %H:%M:%S', '%Y-%m-%d'): try: return datetime.datetime.strptime(str(v).strip(), fmt) except ValueError: continue return None

def _to_bool(v): if v in (None, ''): return None return str(v).strip().lower() in ('true','1','yes')

def parse_market_location_file(path: str) -> list[dict]: wb = openpyxl.load_workbook(path, read_only=True, data_only=True) ws = wb.active it = ws.iter_rows(values_only=True) headers = [str(h).strip() if h is not None else '' for h in next(it)] rows: list[dict] = [] for raw in it: rec = {} for h, val in zip(headers, raw): col = _HDR_TO_COL.get(h) if not col: continue if col in _INT_COLS: rec[col] = _to_int(val) elif col in _DT_COLS: rec[col] = _to_dt(val) elif col in _BOOL_COLS: rec[col] = _to_bool(val) else: rec[col] = (str(val).strip() if val is not None else None) if rec.get('lyreco_118_id') is None: continue rec['campaign_label'] = derive_campaign_label(rec.get('campaign_name')) rec['program'] = derive_program(rec.get('campaign_name')) rec['email_version'] = derive_email_version(rec.get('campaign_name'), rec.get('subject')) rows.append(rec) wb.close() return rows `

• [ ] Step 4: Run tests — venv/Scripts/python.exe -m pytest tests/test_marketloc_parse.py -v → PASS (3 tests).
• [ ] Step 5: Commit — git add scripts/marketloc_parse.py tests/test_marketloc_parse.py && git commit -m "feat(marketloc): parse+normalize market-location xlsx"

---

Task 2: SFTP fetch of daily files

Files:

• Create: integrations/market_location_sftp.py
• Test: tests/test_market_location_sftp.py
• Modify: requirements.txt (add paramiko)

Interfaces:

• Produces: fetch_new_files(local_dir: str, creds: dict | None = None) -> list[str] — connects to MARKET_LOCATION_{HOST,USER,PASSWORD} (or creds), lists remote *.xlsx, downloads any whose basename is not already in local_dir, returns the list of newly-downloaded local paths. Missing creds → returns [] and logs (so the ingest still runs on already-local files).

• [ ] Step 1: Write the failing test (mock paramiko — no real network):

`python # tests/test_market_location_sftp.py import types, pathlib from integrations import market_location_sftp as m

class _FakeSFTP: def __init__(self, remote): self._remote = remote def listdir(self): return list(self._remote) def get(self, r, l): pathlib.Path(l).write_text('x') def close(self): pass

def test_downloads_only_new(monkeypatch, tmp_path): (tmp_path / 'already.xlsx').write_text('old') fake = _FakeSFTP(['already.xlsx', 'new1.xlsx', 'skip.txt']) monkeypatch.setattr(m, '_open_sftp', lambda creds: (fake, lambda: None)) got = m.fetch_new_files(str(tmp_path), creds={'host':'h','user':'u','password':'p'}) assert [pathlib.Path(g).name for g in got] == ['new1.xlsx'] # .txt skipped, existing skipped

def test_missing_creds_returns_empty(monkeypatch, tmp_path): monkeypatch.delenv('MARKET_LOCATION_HOST', raising=False) assert m.fetch_new_files(str(tmp_path), creds=None) == [] `

• [ ] Step 2: Run to verify fail — venv/Scripts/python.exe -m pytest tests/test_market_location_sftp.py -v → FAIL.

• [ ] Step 3: Install paramiko + implement

Run: venv/Scripts/python.exe -m pip install paramiko and add paramiko to requirements.txt.

`python # -- coding: utf-8 -- """Pull market-location responder xlsx from the MARKET_LOCATION_* SFTP into a local dir.""" from __future__ import annotations import os, pathlib

def _creds_from_env() -> dict | None: h, u, p = (os.environ.get('MARKET_LOCATION_HOST'), os.environ.get('MARKET_LOCATION_USER'), os.environ.get('MARKET_LOCATION_PASSWORD')) if not (h and u and p): return None return {'host': h, 'user': u, 'password': p, 'port': int(os.environ.get('MARKET_LOCATION_PORT', '22')), 'remote_dir': os.environ.get('MARKET_LOCATION_DIR', '.')}

def _open_sftp(creds: dict): import paramiko t = paramiko.Transport((creds['host'], creds.get('port', 22))) t.connect(username=creds['user'], password=creds['password']) sftp = paramiko.SFTPClient.from_transport(t) if creds.get('remote_dir'): sftp.chdir(creds['remote_dir']) return sftp, t.close

def fetch_new_files(local_dir: str, creds: dict | None = None) -> list[str]: creds = creds or _creds_from_env() if not creds: print(' MARKET_LOCATION_* creds unset — skipping SFTP fetch (using local files)', flush=True) return [] pathlib.Path(local_dir).mkdir(parents=True, exist_ok=True) have = {p.name for p in pathlib.Path(local_dir).glob('*.xlsx')} sftp, close = _open_sftp(creds) got = [] try: for name in sftp.listdir(): if not name.lower().endswith('.xlsx') or name in have: continue dest = str(pathlib.Path(local_dir) / name) sftp.get(name, dest); got.append(dest) print(f' fetched {name}', flush=True) finally: sftp.close(); close() return got `

• [ ] Step 4: Run tests → PASS (2 tests).
• [ ] Step 5: Commit — git add integrations/market_location_sftp.py tests/test_market_location_sftp.py requirements.txt && git commit -m "feat(marketloc): SFTP fetch of daily files"

---

Task 3: Ingest/upsert into eloqua_campaign_responders

Files:

• Create: scripts/ingest_market_location.py
• Test: tests/test_marketloc_parse.py (add an integration test guarded by DB availability)

Interfaces:

• Consumes: parse_market_location_file (Task 1), fetch_new_files (Task 2), CRMDatabase (database/crm_db.py).
• Produces: upsert_responders(db, rows: list[dict]) -> int (returns rows affected); main(local_dir, do_fetch=True) -> int.

Dedupe key = (lyreco_118_id, campaign_label, sent_date). Use INSERT ... ON CONFLICT DO UPDATE on a unique index; add the index if absent.

• [ ] Step 1: Write the failing test (integration; skip if DB down):

`python # add to tests/test_marketloc_parse.py import os, pytest from database.crm_db import CRMDatabase from scripts.ingest_market_location import upsert_responders

def _db_or_skip(): try: db = CRMDatabase(); db.query('SELECT 1'); return db except Exception as e: pytest.skip(f'postgres unavailable: {e}')

def test_upsert_is_idempotent(): db = _db_or_skip() row = {'lyreco_118_id': 999000001, 'campaign_label': '[test_ml] unit', 'campaign_name': '[test_ml] unit', 'sent_date': '2026-06-01 00:00:00', 'company': 'UnitCo', 'open_count': 1, 'click_count': 0, 'program': 'test_ml', 'email_version': None} try: upsert_responders(db, [row]); upsert_responders(db, [row]) # twice n = db.query("SELECT count(*) AS c FROM eloqua_campaign_responders WHERE lyreco_118_id=999000001")[0]['c'] assert n == 1 finally: db.query("DELETE FROM eloqua_campaign_responders WHERE lyreco_118_id=999000001") `

• [ ] Step 2: Run to verify fail → FAIL (import error).

• [ ] Step 3: Implement scripts/ingest_market_location.py

`python # -- coding: utf-8 -- """Fetch (SFTP) + ingest market-location xlsx into eloqua_campaign_responders. Idempotent.""" from __future__ import annotations import argparse, pathlib, sys sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent)) from database.crm_db import CRMDatabase from scripts.marketloc_parse import parse_market_location_file from integrations.market_location_sftp import fetch_new_files

_REPO = pathlib.Path(__file__).resolve().parent.parent _DEFAULT_DIR = str(_REPO / 'imports' / 'marketlocation') # Columns we write (subset of the table that the parser produces + program/email_version are NOT # table columns, so drop them before insert). _NON_TABLE = {'program', 'email_version'} _UNIQUE_DDL = ("CREATE UNIQUE INDEX IF NOT EXISTS ux_ecr_118_campaign_sent " "ON eloqua_campaign_responders (lyreco_118_id, campaign_label, sent_date)")

def upsert_responders(db: CRMDatabase, rows: list[dict]) -> int: if not rows: return 0 db.query(_UNIQUE_DDL) cols = sorted({k for r in rows for k in r.keys()} - _NON_TABLE) collist = ', '.join(cols) ph = ', '.join(['%s'] * len(cols)) updates = ', '.join(f"{c}=EXCLUDED.{c}" for c in cols if c not in ('lyreco_118_id','campaign_label','sent_date')) sql = (f"INSERT INTO eloqua_campaign_responders ({collist}) VALUES ({ph}) " f"ON CONFLICT (lyreco_118_id, campaign_label, sent_date) DO UPDATE SET {updates}") n = 0 for r in rows: db.query(sql, tuple(r.get(c) for c in cols)); n += 1 return n

def main(local_dir: str = _DEFAULT_DIR, do_fetch: bool = True) -> int: if do_fetch: fetch_new_files(local_dir) total = 0 db = CRMDatabase() for f in sorted(pathlib.Path(local_dir).glob('*.xlsx')): rows = parse_market_location_file(str(f)) total += upsert_responders(db, rows) print(f' {f.name}: {len(rows)} rows upserted', flush=True) print(f'DONE ingested {total} responder rows', flush=True) return total

if __name__ == '__main__': ap = argparse.ArgumentParser() ap.add_argument('--dir', default=_DEFAULT_DIR); ap.add_argument('--no-fetch', action='store_true') a = ap.parse_args(); main(a.dir, do_fetch=not a.no_fetch) `

• [ ] Step 4: Run tests → PASS (or skip if DB down). Then smoke-run: venv/Scripts/python.exe -m scripts.ingest_market_location --no-fetch → expect "DONE ingested N responder rows" with N≈54k+ from the three local files; verify SELECT count(*) FROM eloqua_campaign_responders WHERE campaign_label ILIKE '%supplies4work%' > 0.
• [ ] Step 5: Commit — git add scripts/ingest_market_location.py tests/test_marketloc_parse.py && git commit -m "feat(marketloc): idempotent ingest into eloqua_campaign_responders"

---

Task 4: Refresh the CH/BvD/community match

Files:

• Create: scripts/refresh_marketloc_match.py
• Test: tests/test_marketloc_spine.py (integration test of match coverage)

Interfaces:

• Consumes: CRMDatabase; existing tables eloqua_responder_bridge, companies_house_profiles, prospect_community_match, ecom_accounts.
• Produces: refresh_match(db, campaign_labels: list[str]) -> dict returning {'bvd':n1,'companies_house':n2,'community':n3,'unmatched':n4}; writes/updates eloqua_responder_bridge rows for the given campaigns' lyreco_118_ids, each with a match_source column (add column if missing).

Match precedence per lyreco_118_id: (1) existing eloqua_responder_bridge.payer_number (source bvd); else (2) companies_house_profiles join on cro_number → ecom_accounts (source companies_house); else (3) prospect_community_match (source community); else unmatched.

• [ ] Step 1: Write failing test — assert that after refresh, every responder for a campaign has a bridge row (matched or explicitly match_source='unmatched' / null payer), and matched count ≥ BvD-only count:

`python # tests/test_marketloc_spine.py (part 1) import pytest from database.crm_db import CRMDatabase from scripts.refresh_marketloc_match import refresh_match

def _db(): try: db = CRMDatabase(); db.query('SELECT 1'); return db except Exception as e: pytest.skip(f'pg down: {e}')

def test_match_covers_all_and_records_source(): db = _db() labels = [r['campaign_label'] for r in db.query( "SELECT DISTINCT campaign_label FROM eloqua_campaign_responders WHERE campaign_label ILIKE '%supplies4work%' LIMIT 3")] if not labels: pytest.skip('no market-loc data ingested yet') res = refresh_match(db, labels) assert set(res) == {'bvd','companies_house','community','unmatched'} # every responder id has a bridge row with a source orphan = db.query("""SELECT count(*) c FROM eloqua_campaign_responders r WHERE r.campaign_label = ANY(%s) AND NOT EXISTS (SELECT 1 FROM eloqua_responder_bridge b WHERE b.lyreco_118_id=r.lyreco_118_id)""", (labels,))[0]['c'] assert orphan == 0 `

• [ ] Step 2: Run to verify fail → FAIL.
• [ ] Step 3: Implement scripts/refresh_marketloc_match.py — add match_source TEXT to eloqua_responder_bridge if missing; run the three-tier match as INSERT ... ON CONFLICT (lyreco_118_id) DO UPDATE filling payer_number + match_source; insert unmatched rows (null payer) for the remainder so no responder is orphaned. (Full SQL: BvD rows already exist; CH tier joins eloqua_campaign_responders.cro_number = companies_house_profiles.company_number then to ecom_accounts on the CH-linked account; community tier from prospect_community_match. Each tier WHERE NOT EXISTS a higher-precedence match.)
• [ ] Step 4: Run tests → PASS/skip. Smoke: print the returned counts; expect bvd+companies_house+community > BvD-only baseline.
• [ ] Step 5: Commit — git add scripts/refresh_marketloc_match.py tests/test_marketloc_spine.py && git commit -m "feat(marketloc): CH/BvD/community match refresh with match_source"

---

Task 5: Per-company spine + cohort rollups (cross-DB)

Files:

• Create: scripts/marketloc_spine.py (pure helpers), scripts/compute_marketloc_conversion.py (orchestration)
• Test: tests/test_marketloc_spine.py (part 2 — pure-logic unit tests)

Interfaces:

• Produces (pure, in marketloc_spine.py):
• extract_purchases(orders: list[dict]) -> dict — orders are {order_number, order_date, sales_amount, product_family} rows for one account; returns {first_order_date, first_order_amount, first_order_top_families, second_order_date, second_order_amount, second_order_top_families, total_orders}.
• compute_timeline(sent_date, activation_date, first_order_date, second_order_date) -> dict — returns day-gaps sent_to_activation, activation_to_first, first_to_second, sent_to_first, sent_to_second (None where an endpoint is missing).
• classify_activation(ecom_creation_date, hubspot_created, first_sent_date) -> dict — returns {activation_journey, activation_date, activated, net_new}.
• rollup_cohorts(company_rows: list[dict]) -> list[dict] — group by (dimension, value) for dims program, market_sector, engagement_tier, fit_tier, activation_journey, match_source; each rollup: reached, activated, converted, conversion_rate, median_days_to_first, avg_first_amount, avg_second_amount, top_families.
• Produces (orchestration, in compute_marketloc_conversion.py): compute_company_rows(campaign_labels) -> list[dict].

• [ ] Step 1: Write failing unit tests for the pure helpers

`python # tests/test_marketloc_spine.py (part 2) import datetime as dt from scripts.marketloc_spine import extract_purchases, compute_timeline, classify_activation

def D(s): return dt.date.fromisoformat(s)

def test_extract_first_and_second_order(): orders = [ {'order_number':'A','order_date':D('2026-06-10'),'sales_amount':100.0,'product_family':'Paper'}, {'order_number':'A','order_date':D('2026-06-10'),'sales_amount':50.0,'product_family':'Ink'}, {'order_number':'B','order_date':D('2026-07-01'),'sales_amount':200.0,'product_family':'Paper'}, ] r = extract_purchases(orders) assert r['first_order_date'] == D('2026-06-10') assert r['first_order_amount'] == 150.0 assert 'Paper' in r['first_order_top_families'] assert r['second_order_date'] == D('2026-07-01') assert r['second_order_amount'] == 200.0 assert r['total_orders'] == 2

def test_extract_single_order_has_no_second(): r = extract_purchases([{'order_number':'A','order_date':D('2026-06-10'),'sales_amount':10.0,'product_family':'X'}]) assert r['second_order_date'] is None and r['second_order_amount'] is None

def test_timeline_legs(): t = compute_timeline(D('2026-06-01'), D('2026-06-05'), D('2026-06-10'), D('2026-07-01')) assert t['sent_to_activation'] == 4 and t['activation_to_first'] == 5 assert t['first_to_second'] == 21 and t['sent_to_first'] == 9

def test_timeline_missing_endpoints(): t = compute_timeline(D('2026-06-01'), None, None, None) assert t['sent_to_activation'] is None and t['sent_to_first'] is None

def test_classify_activation_journeys(): a = classify_activation(D('2026-06-05'), None, D('2026-06-01')) assert a['activation_journey'] == 'webshop' and a['net_new'] is True b = classify_activation(None, D('2026-06-03'), D('2026-06-01')) assert b['activation_journey'] == 'hubspot' c = classify_activation(D('2026-06-05'), D('2026-06-03'), D('2026-06-01')) assert c['activation_journey'] == 'both' d = classify_activation(None, None, D('2026-06-01')) assert d['activation_journey'] == 'none' and d['activated'] is False e = classify_activation(D('2026-05-01'), None, D('2026-06-01')) # account predates send assert e['net_new'] is False `

• [ ] Step 2: Run to verify fail → FAIL.

• [ ] Step 3: Implement scripts/marketloc_spine.py (pure functions):

`python # -- coding: utf-8 -- """Pure per-company assembly + cohort rollup helpers for the market-location report.""" from __future__ import annotations import statistics from collections import Counter, defaultdict

def _days(a, b): return (b - a).days if (a is not None and b is not None) else None

def extract_purchases(orders: list[dict]) -> dict: by_order = defaultdict(lambda: {'date': None, 'amount': 0.0, 'fams': Counter()}) for o in orders: g = by_order[o['order_number']] g['date'] = o['order_date']; g['amount'] += float(o['sales_amount'] or 0) if o.get('product_family'): g['fams'][o['product_family']] += 1 ordered = sorted(by_order.values(), key=lambda g: g['date']) out = {'total_orders': len(ordered), 'first_order_date': None, 'first_order_amount': None, 'first_order_top_families': [], 'second_order_date': None, 'second_order_amount': None, 'second_order_top_families': []} if ordered: f = ordered[0] out.update(first_order_date=f['date'], first_order_amount=round(f['amount'], 2), first_order_top_families=[k for k, _ in f['fams'].most_common(3)]) if len(ordered) > 1: s = ordered[1] out.update(second_order_date=s['date'], second_order_amount=round(s['amount'], 2), second_order_top_families=[k for k, _ in s['fams'].most_common(3)]) return out

def compute_timeline(sent_date, activation_date, first_order_date, second_order_date) -> dict: return {'sent_to_activation': _days(sent_date, activation_date), 'activation_to_first': _days(activation_date, first_order_date), 'first_to_second': _days(first_order_date, second_order_date), 'sent_to_first': _days(sent_date, first_order_date), 'sent_to_second': _days(sent_date, second_order_date)}

def classify_activation(ecom_creation_date, hubspot_created, first_sent_date) -> dict: has_web, has_hs = ecom_creation_date is not None, hubspot_created is not None journey = ('both' if has_web and has_hs else 'webshop' if has_web else 'hubspot' if has_hs else 'none') dates = [d for d in (ecom_creation_date, hubspot_created) if d is not None] activation_date = min(dates) if dates else None net_new = (activation_date is not None and first_sent_date is not None and activation_date >= first_sent_date) return {'activation_journey': journey, 'activation_date': activation_date, 'activated': bool(dates), 'net_new': net_new}

def rollup_cohorts(company_rows: list[dict]) -> list[dict]: dims = ('program', 'market_sector', 'engagement_tier', 'fit_tier', 'activation_journey', 'match_source') buckets = defaultdict(list) for r in company_rows: for d in dims: buckets[(d, r.get(d) or 'unknown')].append(r) out = [] for (dim, val), rows in buckets.items(): reached = len(rows) converted = [r for r in rows if r.get('first_order_date')] d2f = [r['timeline']['sent_to_first'] for r in converted if r['timeline'].get('sent_to_first') is not None] a1 = [r['first_order_amount'] for r in converted if r.get('first_order_amount')] a2 = [r['second_order_amount'] for r in rows if r.get('second_order_amount')] fams = Counter() for r in converted: fams.update(r.get('first_order_top_families') or []) out.append({'dimension': dim, 'value': val, 'reached': reached, 'activated': sum(1 for r in rows if r.get('activated')), 'converted': len(converted), 'conversion_rate': round(len(converted) / reached, 4) if reached else 0.0, 'median_days_to_first': statistics.median(d2f) if d2f else None, 'avg_first_amount': round(statistics.mean(a1), 2) if a1 else None, 'avg_second_amount': round(statistics.mean(a2), 2) if a2 else None, 'top_families': [k for k, _ in fams.most_common(3)]}) return out `

• [ ] Step 4: Run unit tests → PASS (6 tests).

• [ ] Step 5: Implement scripts/compute_marketloc_conversion.py — the cross-DB orchestration. Pseudocode-precise (real SQL, uses the pure helpers):

• [ ] Step 6: Smoke-run — venv/Scripts/python.exe -m scripts.compute_marketloc_conversion --dry prints counts: companies, matched, activated, converted, and a sample company row; verify converted>0 and timeline legs populate.
• [ ] Step 7: Commit — git add scripts/marketloc_spine.py scripts/compute_marketloc_conversion.py tests/test_marketloc_spine.py && git commit -m "feat(marketloc): cross-DB per-company spine + cohort rollups"

---

Task 6: Journey-map viz + HTML report + registration

Files:

• Create: scripts/marketloc_report.py
• Modify: scripts/compute_marketloc_conversion.py (call the report builder at the end of main)
• Test: tests/test_marketloc_report.py

Interfaces:

• Consumes: company_rows, cohorts (Task 5).
• Produces: render_journey_map(stage_stats: list[dict]) -> str (inline SVG, CSP-safe); build_report_html(company_rows, cohorts, generated_at: str) -> str; write_and_register(html: str, out_dir, report_id='wisemlconv01') -> str.

Journey stages (from spine aggregates): Reached, Engaged (open/click≥1), Activated, 1st purchase, 2nd purchase, each carrying count + step conversion %.

• [ ] Step 1: Write failing tests (pure render — no DB):

`python # tests/test_marketloc_report.py from scripts.marketloc_report import render_journey_map, build_report_html

def test_journey_map_svg_has_all_stages(): stages = [{'label':'Reached','count':1000},{'label':'Engaged','count':300}, {'label':'Activated','count':120},{'label':'1st purchase','count':80},{'label':'2nd purchase','count':40}] svg = render_journey_map(stages) assert svg.startswith('') for s in stages: assert s['label'] in svg assert '8.0%' in svg or '8%' in svg # 80/1000 cumulative label rendered

def test_report_html_self_contained(): rows = [{'lyreco_118_id':1,'company':'C','program':'supplies4work','market_sector':'Manufacturing', 'engagement_tier':'T2_clicked','fit_tier':'A','match_source':'bvd','activation_journey':'webshop', 'activated':True,'net_new':True,'first_order_date':'2026-06-10','first_order_amount':150.0, 'second_order_date':None,'second_order_amount':None,'first_order_top_families':['Paper'], 'timeline':{'sent_to_first':9,'sent_to_activation':4,'activation_to_first':5,'first_to_second':None}}] cohorts = [{'dimension':'program','value':'supplies4work','reached':1,'activated':1,'converted':1, 'conversion_rate':1.0,'median_days_to_first':9,'avg_first_amount':150.0,'avg_second_amount':None, 'top_families':['Paper']}] html = build_report_html(rows, cohorts, '2026-07-07') assert '')[0] or 'xmlns' in html # no external refs in the svg `

• [ ] Step 2: Run to verify fail → FAIL.
• [ ] Step 3: Implement scripts/marketloc_report.py — render_journey_map builds an SVG wave with one node per stage (x = stage index, y alternating above/mid), the stage label, count, and cumulative conversion % vs Reached; build_report_html emits a self-contained page (inline