code.deepline.com包含需要注意的行為
SKILL DETAIL
niche-signal-discovery
code.deepline.com/niche-signal-discovery
This skill is designed to discover niche first-party signals that differentiate Closed Won vs Closed Lost accounts for ICP (Ideal Customer Profile) analysis. Use it when the user provides won/lost customer domain lists and wants differential signals (website content, job listings, tech stack, maturity markers) to build account scoring models and prospecting criteria. The pipeline includes: target company discovery, ecosystem discovery, input CSV preparation, vertical-specific config generation, multi-page website and job extraction, quality gate, differential analysis, signal interpretation, and output of top 10 net-new prospects. Signal reliability hierarchy from highest to lowest confidence: job listings, analyst validation, compliance infrastructure, buyer pain language, tech stack tools, website product/marketing content. The skill emphasizes using the Deepline CLI for enrichment and follows a Deepline-first principle.
安裝量 · 140查看來源
Installation
npx skills add https://github.com/code.deepline.com --skill niche-signal-discovery
技能檔案
SKILL.md
最近同步 · 2026年8月29日
references/dedupe.md›
# Dedupe against the user's existing list
Used in Step 1.0.5 (filter prospect candidates) and Step 7 (validate emails against the company's apex domain). The shipped helper is `scripts/dedupe_utils.py`.
## Why this exists
The common failure mode is shipping a "net-new" list that repeats companies the user has already tried. Two failure modes drive this:
1. **Raw string match misses parent-domain relationships.** `amsynergy.nikon.com` and `nikon.com` are the same buyer, but a naive set lookup treats them as unrelated. On one real run, **24 of 50 "net-new" prospects were already in the CRM** as parent-domain entries the raw-string dedupe missed. Apex normalization with a public-suffix-aware parser fixes this.
2. **Name-only matching is noisy.** "Rocket Propulsion Systems" can collide with an unrelated row that has "rocket propulsion" as a substring. Apex domain is a stronger primary key when it's available.
The fix is a layered check.
## The two-phase match
**Step A — apex domain match.** Normalize both sides to the registrable apex domain. Strip `www.`, strip subdomains to the root, handle multi-label suffixes (`co.uk`, `co.jp`, `com.au`). Use `extract_apex()` from `scripts/dedupe_utils.py`. This is the strong primary key — when it matches, you're done.
**Step B — fuzzy company name fallback.** For candidates that don't match on apex domain (either because the candidate has no website in the input, or the existing list only has names), compare normalized company names with `difflib.SequenceMatcher` ratio ≥ 0.85 after stripping corporate suffixes (Inc, LLC, Ltd, GmbH, SA, AG, Co, Corp, Holdings, Technologies, Systems, etc.). Use `norm_name()` from `scripts/dedupe_utils.py`. **Only as a fallback** — never as the primary check.
## Always ask the user first
Before running any dedupe, ask explicitly: _"Do you have an existing customer list, CRM export, or past outbound list I should dedupe the prospect output against? A CSV with a `domain` column and optionally a `name` column is enough."_ If they say no, note it as a caveat in the final report. **Don't silently skip the dedupe** — a downstream user reading the prospect section deserves to know whether one was applied.
## Usage
Sanity-test the helper after install — runs entirely in stdlib:
```bash
python3 scripts/dedupe_utils.py --selftest
```
CLI for one-shot dedupes:
```bash
python3 scripts/dedupe_utils.py \
--existing customers.csv \
--candidates prospects_raw.csv \
--out-actionable prospects_actionable.csv \
--out-matched prospects_already_known.csv \
--name-threshold 0.85
```
Library import for pipeline integration:
```python
import sys; sys.path.insert(0, "scripts")
from dedupe_utils import match_against_existing
existing = load_csv("customers.csv") # rows with 'domain' / 'name' / 'website'
candidates = load_csv("prospects_raw.csv")
actionable, matched = match_against_existing(candidates, existing, name_threshold=0.85)
```
Each row in the output carries a `dedupe_match` field — empty for actionable, populated with the match reason for matched rows (`apex:nikon.com` or `name:astura medical (0.91)`). Surface this in the final report so the user can audit what was excluded and why.
## Don't silently drop matches — categorize them
Even after dedupe, the "already in CRM" bucket often contains accounts still worth outbound, just with a different wedge or timing. Prefer to categorize:
| Category | Treatment |
| ---------------------------------------------------- | ----------------------------------------------------------------------------- |
| **Net-new** (not in existing list) | Standard outbound. |
| **Already in CRM, no opps ever** | Effectively net-new from an activity perspective; surface with a note. |
| **Previously lost / rejected, no active engagement** | Fresh outbound wedge. Flag the previous loss reason so the AE can address it. |
| **Active open opportunity** | Exclude — don't step on an active sales cycle. |
| **Current customer** | Exclude from net-new; expansion is a different motion. |
This categorization requires some signal from the existing list beyond "do they show up" — if the user's list has columns like `opp_count`, `open_count`, `won_count`, `last_opp_date`, use them. If it doesn't, just split into "net-new" vs "already known" and flag the latter for manual triage.
references/keyword-catalog.md›
# Keyword Catalog Reference
The analysis script requires three JSON config files: keywords, tools, and job roles. This reference explains each format and provides generation guidance for ANY vertical.
**CRITICAL: Do NOT copy these examples directly. Generate configs based on your target's vertical (Step 0 + 0.5 discovery).**
## How Matching Works
- **Substring matching**: `integrat` matches "integrate", "integration", "integrations"
- **Case insensitive**: All matching is lowercased
- **Category grouping**: Keywords are grouped by category for structured output
## JSON Format: Keywords (`--keywords`)
Each key is a category name, each value is a list of keyword strings.
### Universal Categories (work across all verticals)
```json
{
"business_model": [
"contact sales",
"plan",
"enterprise",
"pricing",
"demo",
"request a demo",
"subscription",
"free trial",
"annual",
"monthly"
],
"product_integration": [
"integrat",
"workflow",
"api",
"connect",
"customiz",
"automat",
"personaliz",
"sync",
"sdk",
"webhook",
"platform"
],
"company_maturity": [
"compliance",
"secur",
"leader",
"efficien",
"case stud",
"newsroom",
"gartner",
"forrester",
"training",
"resource",
"soc 2",
"gdpr",
"iso"
]
}
```
### Vertical-Specific Categories (generate per target)
**Category types:**
- **Product category terms** — What does the target sell? (e.g., "creative ops", "AR automation", "sales engagement")
- **Buyer pain points** — What problems does the buyer have? (e.g., "manual invoicing", "fragmented tools", "pipeline visibility")
- **Anti-fit signals** — Competitor names, wrong business model indicators
### Multi-Vertical Examples
**Creative Ops / DAM Tools** (e.g., Bynder, Widen, Brandfolder):
```json
{
"creative_operations": [
"creative ops",
"creative operations",
"asset management",
"DAM",
"content library",
"brand guidelines",
"creative workflow"
],
"buyer_pain_points": [
"fragmented tools",
"content discovery",
"version control",
"creative approval",
"asset organization",
"brand consistency"
],
"anti_fit": ["bynder", "widen", "brandfolder", "canto", "dam platform"]
}
```
**AR Automation / Finance Tools** (e.g., Tesorio, HighRadius):
```json
{
"finance_operations": [
"accounts receivable",
"collections",
"dunning",
"DSO",
"invoice",
"payment",
"cash flow",
"ar automation"
],
"buyer_pain_points": [
"manual invoicing",
"payment delays",
"late payments",
"cash application",
"reconciliation",
"aging report"
],
"anti_fit": ["highradius", "tesorio", "invoiced", "ar automation platform"]
}
```
**Sales Engagement Tools** (e.g., Outreach, SalesLoft):
```json
{
"sales_operations": [
"crm",
"salesforce",
"prospect",
"productiv",
"onboard",
"sequenc",
"outbound",
"pipeline",
"quota"
],
"buyer_pain_points": [
"manual outreach",
"email tracking",
"cadence management",
"activity logging",
"pipeline visibility"
],
"anti_fit": ["outreach", "salesloft", "lemlist", "sales engagement platform"]
}
```
**Developer Tools** (e.g., LaunchDarkly, Vercel):
```json
{
"developer_operations": [
"feature flag",
"deployment",
"ci/cd",
"dev experience",
"build time",
"developer productivity",
"infrastructure"
],
"buyer_pain_points": [
"slow builds",
"deploy risk",
"rollback",
"canary deploy",
"environment management",
"developer friction"
],
"anti_fit": ["launchdarkly", "vercel", "netlify", "feature flag platform"]
}
```
### Generation Pattern
**1. Product category keywords** — `deeplineagent` prompt: "Research the terminology and keywords buyers and vendors use for {target product category}."
Example for creative ops:
```bash
deeplineagent: "Research creative operations terminology, DAM phrases, and asset management keywords."
```
**2. Buyer pain point keywords** — `deeplineagent` prompt: "Research the challenges, problems, and pain points for {buyer persona}."
Example for creative teams:
```bash
deeplineagent: "Research the workflow challenges and pain points for creative teams and marketing teams."
```
**3. Anti-fit keywords** — From Step 0.5 competitor discovery
**4. Universal keywords** — Use business_model, product_integration, company_maturity (same across verticals)
## JSON Format: Tech Stack Tools (`--tools`)
Each key is a tool category, each value is a list of tool names to search for. **Niche tools are far more discriminative than generic ones.**
### Multi-Vertical Examples
**Creative / Marketing Tools Stack** (for DAM, creative ops tools):
```json
{
"creative_design": [
"figma",
"sketch",
"adobe creative cloud",
"canva",
"invision"
],
"marketing_ops": ["hubspot", "marketo", "contentful", "wordpress", "webflow"],
"project_management": ["monday.com", "asana", "jira", "clickup", "notion"],
"video_production": ["frame.io", "vimeo", "wistia", "loom"],
"anti_fit_tech": ["bynder", "widen", "brandfolder"]
}
```
**Finance / AR Tools Stack** (for AR automation, billing tools):
```json
{
"erp_accounting": ["netsuite", "quickbooks", "xero", "sage intacct"],
"crm_billing": ["salesforce", "hubspot", "chargebee", "stripe billing"],
"payment_processing": ["stripe", "ach", "paypal", "adyen"],
"reporting_bi": ["tableau", "looker", "power bi", "metabase"],
"anti_fit_tech": ["highradius", "tesorio", "invoiced"]
}
```
**Sales Tools Stack** (for sales engagement, revenue tools):
```json
{
"crm": ["salesforce", "hubspot", "pipedrive"],
"sales_engagement": ["outreach", "salesloft", "lemlist", "smartlead"],
"conversation_intelligence": ["gong", "chorus", "clari"],
"prospecting": ["zoominfo", "clearbit", "lusha", "cognism"],
"anti_fit_tech": ["outreach", "salesloft", "sales engagement"]
}
```
**Developer Tools Stack** (for dev tools, infrastructure):
```json
{
"cloud_infra": ["aws", "gcp", "azure", "vercel", "netlify"],
"ci_cd": ["github actions", "gitlab ci", "circle ci", "jenkins"],
"monitoring": ["datadog", "new relic", "pagerduty", "sentry"],
"feature_flags": ["launchdarkly", "split", "optimizely"],
"anti_fit_tech": ["launchdarkly", "vercel", "feature flag"]
}
```
### Generation Pattern
**1. Tech stack discovery** — From Step 0.5 `deeplineagent` research: "What tools are common in the {buyer persona} software stack?"
Example for creative teams:
```bash
deeplineagent: "Research the common software tools and tech stack for creative teams and marketing teams."
deeplineagent: "Research the tools, integrations, and workflows common to creative operations teams using Figma or Adobe."
```
**2. Category organization** — Group tools by function (design, marketing, project mgmt, etc.)
**3. Anti-fit tech** — Competitor products from Step 0.5
### Anti-Fit Tech
Include tools that signal the prospect is NOT a good buyer:
- **Competing products** — Bynder/Widen for DAM tools, Outreach/SalesLoft for sales tools
- **Wrong business model** — Shopify for B2B tools (indicates B2C e-commerce)
- **Substitute solutions** — Tools that solve the same problem differently
## JSON Format: Job Roles (`--job-roles`)
Each key is a role category, each value is a list of substrings to match against job titles and descriptions.
### Multi-Vertical Examples
**Creative / Marketing Roles** (for DAM, creative ops tools):
```json
{
"creative_leadership": [
"creative director",
"head of creative",
"vp creative"
],
"content_management": [
"content manager",
"content director",
"brand manager"
],
"creative_ops": [
"creative operations",
"creative ops manager",
"brand operations"
],
"marketing_ops": ["marketing operations", "marops", "marketing ops manager"],
"design": ["product designer", "brand designer", "visual designer"],
"marketing_general": ["marketing manager", "demand gen", "growth marketing"]
}
```
**Finance / Accounting Roles** (for AR automation, billing tools):
```json
{
"finance_leadership": ["cfo", "vp finance", "head of finance"],
"ar_collections": [
"accounts receivable",
"ar manager",
"collections manager"
],
"accounting": ["accountant", "controller", "staff accountant"],
"billing_ops": [
"billing manager",
"billing operations",
"revenue operations"
],
"treasury": ["treasury", "cash management", "financial analyst"]
}
```
**Sales Roles** (for sales engagement, revenue tools):
```json
{
"sales_leadership": ["cro", "vp sales", "head of sales"],
"ae": ["account executive", "ae ", "sales executive"],
"sdr_bdr": [
"sdr",
"bdr",
"sales development",
"business development representative"
],
"sales_ops": [
"sales operations",
"sales ops",
"revenue operations",
"revops"
],
"enablement": ["enablement", "sales enablement"],
"customer_success": ["customer success", "cs manager", "csm"]
}
```
**Engineering / Product Roles** (for dev tools, infrastructure):
```json
{
"engineering_leadership": ["cto", "vp engineering", "head of engineering"],
"platform_infra": [
"platform engineer",
"infrastructure engineer",
"devops",
"sre"
],
"backend": ["backend engineer", "software engineer", "full stack"],
"frontend": ["frontend engineer", "web developer"],
"product": ["product manager", "product lead", "product design"]
}
```
### Generation Pattern
**1. Job role discovery** — From Step 0.5 `deeplineagent` research: "What job titles, roles, and responsibilities are common for {buyer persona}?"
Example for creative teams:
```bash
deeplineagent: "Research creative operations job titles, including creative director and content manager variants."
deeplineagent: "Research companies hiring for creative operations or brand manager roles and extract common title variants."
```
**2. Category organization** — Group by seniority and function (leadership, IC roles, ops roles)
**3. Include adjacent roles** — Marketing ops for creative tools, sales ops for sales tools
### Customizing Job Roles for Verticals
**Buyer persona determines roles:**
- Creative/marketing tools → creative director, content manager, brand manager
- Finance tools → CFO, AR manager, controller, accountant
- Sales tools → CRO, AE, SDR, RevOps
- Dev tools → CTO, platform engineer, DevOps, SRE
**Always include:**
- Leadership roles (decision makers)
- IC roles (day-to-day users)
- Ops roles (implementation/process owners)
- Adjacent roles (related functions)
## Anti-Fit vs. Migration Opportunity Keywords
**CRITICAL DISTINCTION:**
**Anti-fit keywords** = Structural mismatches that make the company a bad buyer. These should be rare in Won companies.
**Migration keywords** = Competitor tool usage. These companies are valid targets (displacement opportunity), NOT anti-fit.
### Anti-Fit Keywords (True Exclusions)
1. **Competitor product signals** — Company SELLS the same thing the target sells (they're a vendor, not a buyer)
- Example: "sales engagement platform" for Outreach/SalesLoft target
- Example: "DAM platform" or "digital asset management platform" for a DAM tool target
- Example: "AR automation platform" for HighRadius/Tesorio target
2. **Consumer/B2C signals** (for B2B tools) — `shopper`, `checkout`, `cart`, `consumer`, `debit card`
3. **Wrong industry** — `patient` for non-healthcare tools, `student` for non-EdTech
4. **Wrong business model** — `reseller`, `distributor` for direct-sales tools
### Migration Opportunity Keywords (NOT Anti-Fit)
Competitor TOOL names (Bynder, Widen, Outreach, HighRadius, etc.) indicate companies currently using those tools. These are:
- Valid prospecting targets (migration/replacement opportunity)
- Lower priority than greenfield accounts
- Require different messaging (displacement vs. new adoption)
Add these to a separate category for segmentation, NOT exclusion.
## Generation Workflow Summary
**Step 0** — Discover target (what they sell, who they sell to)
**Step 0.5** — Discover ecosystem (competitors, tech stack, job roles)
**Step 1.5** — Generate configs using patterns above
**For keywords.json:**
1. Start with universal categories (business_model, product_integration, company_maturity)
2. Add product category terms from Step 0
3. Add buyer pain points from Step 0.5
4. Add competitor tool names for migration segment (e.g., "bynder", "widen") — NOT anti-fit
5. Add true anti-fit keywords (product signals like "DAM platform", structural mismatches)
**For tools.json:**
1. Organize tech stack from Step 0.5 into categories
2. Focus on niche tools specific to buyer persona
3. Add competitor tool products for migration segment — NOT anti_fit_tech
**For job-roles.json:**
1. Organize roles from Step 0.5 by seniority/function
2. Include leadership (decision makers), IC (users), ops (implementers)
3. Add adjacent roles related to buyer persona
**Validation (Step 3.5):**
- Check generated keywords appear in enriched data
- Verify job roles match actual job listings
- Ensure tech stack tools match integrations/tech pages
- Confirm product category keywords (what target SELLS) don't appear frequently in Won companies → if they do, those are competitors not buyers
references/pitfalls.md›
# Common pitfalls (full list)
The most critical pitfalls are kept inline in `SKILL.md`. This file is the long form — read it when you hit unexpected behavior or before shipping a report.
## Pipeline execution
1. **Skipping target discovery (Step 0)** — Without understanding what the target sells, you'll generate generic/irrelevant configs.
2. **Homepage-only scraping** — Always use multi-page discovery. Homepage alone misses pricing, integrations, security, careers.
3. **Using hardcoded examples** — Don't copy sales-focused keywords for a creative-ops tool. Generate configs per vertical.
4. **Skipping config review (Step 3.5)** — Always validate generated configs against enriched data before analysis.
5. **Running analysis immediately after enrichment** — `deepline enrich` returns to terminal before OS buffers flush. Run the file completeness check in Step 3 before executing `analyze_signals.py`. A `won_with_jobs: 0` result when you expect data is the symptom; re-running the analysis (without re-enriching) fixes it.
6. **Duplicate domains in input** — CRM exports often have the same company in both won and lost (multiple deals). Deepline only fetches job listings once per domain, so the duplicate's job data lands on one row only — silently undercounting `won_with_jobs`. Always deduplicate in Step 1.
## Signal interpretation
7. **Generic tech stack** — "AWS", "GitHub", "Slack" appear on most B2B sites and aren't differentiating. Search for niche SaaS tools specific to the buyer persona (e.g., Figma for creative teams, NetSuite for finance teams).
8. **Ignoring source context** — "prospect" on a product page = seller signal. "prospect" in a job listing = buyer signal. Same keyword, opposite meaning.
9. **Missing lost data** — Verify lost companies have content before analysis. Empty lost = meaningless lift scores.
10. **Substring false positives** — "sequenc" matches "consequences". Spot-check high-lift keywords for false matches.
11. **Treating vendor signals as buyer signals** — "accounts receivable automation" on a company's product page means they SELL AR tools (competitor). The same phrase in a job listing means they NEED AR tools (buyer). Source context is everything — see `references/signal-interpretation.md`.
12. **Trusting n=1 signals** — A signal in 1 won company with 0 lost = mathematically high lift but statistically meaningless. Require 3+ companies for Tier 1 scoring signals. Flag single-company signals in the report with a verification note.
13. **Including generic business words as signals** — "platform", "automat*", "integrat*" appear at near-identical rates in won and lost (1.0-1.1x lift). These are baseline terms, not differentiators. Focus on signals with lift > 1.5x that are specific to the target's vertical.
## Data hygiene
14. **Domain mismatches in auto-extracted lists** — When using CRM exports or automated customer discovery, domain → company name mapping can be wrong. In actual runs, up to 53% of auto-extracted domains were false positives. Always validate domains against expected company names before enrichment.
15. **Expecting website signals for back-office tools** — Companies buying AR automation, billing, or compliance tools don't discuss these needs on their marketing websites. For these verticals, rely on job listings (hiring AR Manager = budget + pain), tech stack (NetSuite, Salesforce in jobs), and firmographics (wholesale/distribution/manufacturing) instead.
## Dedupe + prospect output
16. **Raw-string dedupe that misses parent domains** — `amsynergy.nikon.com` and `nikon.com` are the same buyer. A naive set lookup treats them as unrelated and ships Nikon twice on the prospect list. Always normalize to apex domain with a public-suffix-aware parser (`scripts/dedupe_utils.py:extract_apex()`) BEFORE comparing. On one real run, **24 of 50 "net-new" prospects were already in the CRM** as parent-domain entries the raw-string dedupe missed. See `references/dedupe.md`.
17. **Shipping a signal report without a prospect list** — The signals tell you what to look for; they don't tell you who to email tomorrow morning. A report that stops at the scoring model forces the reader to do their own prospecting pass — exactly the expensive thing they were hoping to skip. Step 7 (top 10 prospects) is required, not a nice-to-have.
18. **Trusting confirmation-biased CRM fields as signals** — Catalyst note count, champions/DM counts, OCR-derived fields, MEDDPICC picklists are all downstream artifacts of AE engagement. They correlate with win-rate because AEs work the deals they think will win, not because these properties cause wins. On one real run, catalyst note count showed "109x lift" — the most extreme signal in the dataset — and was this close to making the TL;DR before we caught the direction of causality. See `references/scoring-pitfalls.md`.
references/proven-signals.md›
# Proven signal patterns (from actual runs)
These patterns have been validated across multiple customer analyses spanning creative ops, sales engagement, AR automation, legal tech, and GTM tools. Use them as a starting point when interpreting results — but always validate against the specific target's vertical.
## High-confidence positive signals
| Signal Pattern | Typical Lift | Validated For | What It Means |
| ---------------------------------------------- | ------------ | -------------------------- | ---------------------------------------------------------------------- |
| Analyst validation (Gartner, Forrester) | 4.5x-6.5x | Enterprise B2B SaaS | Company has evaluated the category, has enterprise procurement process |
| Hiring for ICP-related roles | 3.8x-5.5x | All verticals | Active budget + acknowledged pain — highest-intent signal |
| Published case studies | 3.7x | Product-led + sales-assist | Mature marketing org, values proof points, vendor-friendly |
| Compliance infrastructure (GDPR, SOC2, ISO) | 2.1x-6.5x | Enterprise tools | Formal approval processes, security reviews, higher close rates |
| Buyer pain language (e.g., "fragmented tools") | 2.9x-5.2x | Creative ops, MarTech | Operational awareness of the specific problem the target solves |
| SDK/webhook/API presence | 2.5x-3.5x | Developer-adjacent tools | Developer culture, integrates tools programmatically |
| Contact sales / sales-led GTM | 2.2x-5.5x | Enterprise sales tools | Human-led sales motion = AE-dependent = sales engagement tool buyer |
| Niche tech stack (Figma, Frame.io, NetSuite) | 1.5x-5.5x | Vertical-specific | Infrastructure readiness for the target's integration ecosystem |
## High-confidence anti-fit signals
| Signal Pattern | Typical Lift | What It Means |
| --------------------------------------------------- | ------------ | -------------------------------------------------- |
| Consumer signals (shopper, checkout, cancel, debit) | 0.2x | B2C company, not B2B sales org |
| Retention/churn language | 0.2x-0.4x | Consumer subscription model, not enterprise buying |
| Selling same product category | 0.1x-0.3x | Competitor, not buyer — they SELL the solution |
| No job listings in 12+ months | N/A | Not growing, no hiring budget |
## Scoring model guidance
From actual runs, a 0-100 point model with three tiers works well:
- **Tier 1: Core Fit (0-40 pts)** — Compliance, analyst validation, structural signals
- **Tier 2: Buying Intent (0-30 pts)** — Hiring for domain roles, pain language, tech stack
- **Tier 3: Infrastructure Readiness (0-30 pts)** — API presence, integration maturity, case studies
Score thresholds: 60+ = Tier 1 immediate outreach, 35-59 = Tier 2 trigger-based, <35 = nurture or skip.
references/quality-gate.md›
# Step 3 — Quality gate
`deepline enrich` returns control to the terminal before OS buffers fully flush to disk. Running `analyze_signals.py` immediately after enrichment can read a partially-written file where job columns for the last N rows haven't synced yet — resulting in `won_with_jobs: 0` or severely undercounted job data. Always verify file completeness before running analysis.
## Verify row count + job coverage
```bash
# 1. Check row count matches input
INPUT_ROWS=$(wc -l < output/{{company}}-icp-input.csv)
OUTPUT_ROWS=$(wc -l < output/{{company}}-enriched.csv)
echo "Input: $INPUT_ROWS rows, Output: $OUTPUT_ROWS rows"
# Output should equal input (both include header)
# 2. Spot-check job data for a known won account with job listings
python3 -c "
import csv, json, sys
csv.field_size_limit(sys.maxsize)
with open('output/{{company}}-enriched.csv') as f:
rows = list(csv.DictReader(f))
won_rows = [r for r in rows if r.get('status') == 'won']
jobs_col = 'jobs' # or use column index
has_jobs = sum(1 for r in won_rows if r.get(jobs_col, '').strip() not in ('', '{}', 'null'))
print(f'Won rows with job data: {has_jobs}/{len(won_rows)}')
# If this is 0 and you know won accounts should have listings, wait and re-run
"
```
If `won_with_jobs` is 0 but you expect job data:
1. Wait 5-10 seconds (OS buffer flush)
2. Re-run the verification check
3. If still 0, check column indices — the enriched CSV uses `website` and `jobs` column names, NOT `__dl_full_result__`. Use `--website-col N --jobs-col N` overrides.
## Coverage checks
After file verification:
- **Coverage**: >80% of companies should have website content. If <80%, check domain spelling and retry failed rows.
- **Content depth**: Average should be 6-8 pages per company, 12-20K chars.
- **Job listings**: Won companies should have more job data than lost (expected — larger/scaling companies win more).
If coverage is poor, re-run failed domains with `--rows` targeting specific rows.
## Domain validation (auto-extracted lists)
If customer domains came from automated extraction (CRM exports, Exa API, case study scraping) rather than a manually verified list, validate that domains actually belong to the named companies. **From actual runs: up to 53% of auto-extracted customers can be false positives** — competitors selling the same product, domain mismatches, and unrelated companies.
```bash
# Check for suspicious domain patterns
python3 -c "
import csv, sys
csv.field_size_limit(sys.maxsize)
with open('output/{{company}}-enriched.csv') as f:
rows = list(csv.DictReader(f))
for r in rows:
domain = r.get('domain', '')
# Flag content platforms used as source URLs, not company domains
if any(x in domain for x in ['blog.', 'medium.com', 'substack.', 'wordpress.']):
print(f'WARNING: {domain} looks like a content platform, not a company domain')
# Flag very short domains that might be generic
if len(domain.split('.')[0]) <= 2:
print(f'CHECK: {domain} — very short domain, verify it belongs to the expected company')
"
```
**Red flags for false positives:**
- Domain is a subdomain of the target company (blog.target.com)
- Domain belongs to a well-known AI/tech company but the "customer" is a different firm (domain resolution failed)
- Company appears in competitor case studies, not target's own customer list
- Company is itself a vendor in the same product category (they SELL the solution, they don't BUY it)
references/report-template.md›
# Report Template Reference
Template for the niche signals report. Follow this structure and quality rules strictly.
**Every report opens with a Quick Reference Dashboard (Sections 0.1–0.5) before the detailed data sections. This lets any reader — AE, SDR, or executive — understand key findings in under 2 minutes and take action immediately.**
---
## Section 0: Quick Reference Dashboard
**Required at the top of every report.** Generate once analysis is complete. Use actual lift scores and signal names from your dataset.
### 0.1 TLDR (5 Bullets)
Format as a prominent callout/highlight block at the very top of the report:
```
⚡ TLDR — Read This First
• #1 signal: [top signal name] on their website — [X]x more common in won accounts — [one-line reason why it indicates buying intent]
• Best-fit archetype: [ideal won customer in one sentence: size, vertical, regulatory context, maturity stage]
• Fastest path to pipeline: Deepline people search for "[title 1]" + "[title 2]" at [headcount]-person [vertical] companies — these people own the buying decision
• Hard skip flags: [signal 1], [signal 2], [signal 3] — [brief reason each signals existing solution, build culture, or procurement freeze]
• Scoring: 60+ pts → Tier 1 immediate outreach · 35–59 → Tier 2 trigger-based · <35 → nurture or skip
```
### 0.2 Signal Strength at a Glance
Two tables with visual lift bars. Sort positive signals by lift descending, anti-fit by lift ascending.
**Lift → Strength Bar scale:**
| Lift | Bar |
| ------ | ------------ |
| ≥10x | 🟩🟩🟩🟩🟩🟩 |
| ≥4x | 🟩🟩🟩🟩🟩 |
| ≥2.5x | 🟩🟩🟩🟩 |
| ≥2.0x | 🟩🟩🟩 |
| ≥1.5x | 🟩🟩 |
| ≥1.0x | 🟩 |
| ≥0.4x | 🟥🟥 |
| ≥0.25x | 🟥🟥🟥 |
| ≥0.15x | 🟥🟥🟥🟥 |
| ≥0.07x | 🟥🟥🟥🟥🟥 |
| <0.07x | 🟥🟥🟥🟥🟥🟥 |
**✅ Positive Fit Signals** — Top 10–15, sorted by lift descending:
```markdown
| Signal | Lift | Strength | Source | What to Look For |
| ------------- | ------ | -------- | ------------------------------ | --------------------------------------------- |
| [signal name] | [X.Xx] | [bar] | 🌐 Website / 💼 Jobs / 💻 Tech | [1-sentence: what to check and what it means] |
```
Source icons: `🌐 Website` = found in website content · `💼 Jobs` = found in job listings · `💻 Tech` = tech stack detection
**🚫 Anti-Fit Signals** — All signals with lift < 0.5x:
```markdown
| Signal | Lift | Risk | Why |
| ------------- | ------ | ----- | --------------------------------------------------------------------------- |
| [signal name] | [0.Xx] | [bar] | [root cause: existing solution / build culture / procurement freeze / etc.] |
```
### 0.3 Platform Search Recipes
Pre-built, click-ready search links for each buyer type.
**People Searches (find the buyers):**
```markdown
| Who You're Finding | Why They're the Buyer | Prospecting Link |
| --------------------------- | ----------------------------------------------------- | -------------------- |
| [Title 1, Title 2, Title 3] | [Signal lift + one-line reason they own the decision] | [Open Search ↗](URL) |
```
**Company Searches (find the accounts):**
```markdown
| What You're Finding | Signal It Represents | Prospecting Link |
| ------------------------------- | -------------------- | -------------------- |
| [Company type + keyword filter] | [Signal name + lift] | [Open Search ↗](URL) |
```
**Google Search Operators (verify a specific company before outreach):**
```markdown
| What to Check | Google Operator | Positive Result Looks Like |
| ------------- | ----------------------------- | ----------------------------- |
| [Signal name] | `site:domain.com "[keyword]"` | [What a positive match means] |
```
**Prospecting command format — use Deepline play-backed searches:**
```
People search:
deepline plays run prebuilt/company-to-contact \
--input '{"domain":"example.com","roles":["Title One","Title Two"],"seniority":["vp","director"],"limit":25}'
Company search:
deepline tools execute crustdata_companydb_search \
--input '{"company_keywords":["keyword-one","keyword-two"],"countries":["United States"],"headcount_ranges":["201-500"],"limit":50}'
```
Valid headcount ranges: `1-10` `11-20` `21-50` `51-200` `201-500` `501-1000` `1001-5000` `5001-10000` `10001+`
Valid seniorities: `vp` `director` `manager` `c_suite` `owner` `partner` `senior` `entry`
Use keyword-based company filtering where the selected company-search tool supports it. Do not hardcode provider-specific industry tag IDs; use portable keywords instead.
### 0.4 Buyer Persona Quick Reference
One row per key persona. Pull title patterns and pain points from job hiring signals + keyword analysis. Include 3–5 personas covering: primary decision-maker, economic buyer, technical evaluator, champion.
```markdown
| Persona | Title Pattern | Pain Point | Signal to Reference | Prospect Search |
| ------- | --------------------------- | ----------------- | -------------------------------------- | --------------- |
| [Name] | [Title 1, Title 2, Title 3] | [Core pain point] | [Top signal + lift + where to find it] | [Search ↗](URL) |
```
### 0.5 Lead Scoring Cheatsheet
Condensed scoring model — score any prospect in under 2 minutes.
```markdown
| Signal | Points | How to Check |
| --------------------------------- | ------ | ----------------------------------------------------- |
| [Top positive signal] | +[N] | `site:domain.com "[keyword]"` OR Deepline jobs/search |
| ... (8–12 positive signals total) | | |
| [Top anti-fit signal] | −[N] | [How to check] |
| ... (4–6 anti-fit signals total) | | |
```
Score tiers:
```markdown
| Score | Tier | Action |
| ------ | --------- | -------------------------------------------------------------------- |
| 60–100 | 🟢 Tier 1 | Immediate — personalized sequence referencing their specific signals |
| 35–59 | 🟡 Tier 2 | Trigger-based — sequence on funding, industry news, or hiring event |
| <35 | 🔴 Tier 3 | Nurture or skip — likely not a fit today |
```
---
## Header
```markdown
# {Company Name} ICP Niche Signals Report
**Analysis Date:** {{date}}
**Target Company:** {{company}} ({{domain}}) — {one-line description}
**Dataset:** {{won_count}} Closed Won + {{lost_count}} Closed Lost accounts
**Data Sources:** Multi-page website extraction (exa_search with contents, ~8 pages/company) + job listings (Crustdata)
**Coverage:** {{won_with_content}}/{{won_count}} won and {{lost_with_content}}/{{lost_count}} lost with website content; {{won_with_jobs}}/{{won_count}} won with job listings
```
---
## Section 1: Executive Summary
**Format:** 2-3 direct sentences profiling best-fit customers. Include top 3 differentiating signals with lift values.
**REQUIRED: Add prospective target companies** (not in dataset) that match the ICP profile:
- List 4-6 concrete companies that fit the profile but aren't current customers
- Include: company name, size, specific signals (hiring roles, tech stack, pain points mentioned)
- Shows what the ICP looks like in the wild
**Example:**
> {{Target}}'s buyers are mid-size companies (100-1000 employees) scaling {{domain}} operations. Top signals: hiring {{domain}}-related roles (3-5x lift), using {niche tools} (2-4x lift), mentioning "{buyer pain point}" (3-6x lift).
>
> **Companies that fit this profile but aren't customers yet:**
>
> - {Company A} ({{size}} employees) — {specific signal 1}, {specific signal 2}
> - {Company B} ({{size}} employees) — {specific signal 1}, {specific signal 2}
**Avoid:** Generic "perfect fit customer" descriptions. Be specific and concrete.
### Dataset Caveat (if applicable)
If the dataset has limitations, add a caveat subsection. Common caveats:
- Lookalike companies used as Won (they haven't actually purchased — signals are inferred fit, not validated)
- Small sample size (<20 won or <10 lost)
- Uneven group sizes (e.g., 8 won + 32 lost)
- Auto-extracted domains without manual verification
---
## Section 2: Website Keyword Differential
Methodology note at the top:
> Substring matching across multi-page website content for {{won_n}} won and {{lost_n}} lost companies. Lift uses Laplace smoothing: `((won + 0.5) / (won_total + 1)) / ((lost + 0.5) / (lost_total + 1))`. **Bold** = lift > 2x.
### Subsections by category (2.1, 2.2, etc.)
Table format:
```markdown
| Keyword | Won (n=X) | Lost (n=Y) | Lift | Interpretation |
```
**Quality rules:**
- Raw counts always: `15% (6)` not just `15%`
- Sample sizes in headers: `Won (n=37)`, `Lost (n=18)`
- **Bold** lift > 2x only
- Interpretation column required — explains WHY this matters for the target company
### Source Evidence (Required for top 3 keywords per table)
After each table, add a blockquote with **exact quotes** and **linked sources** for the top 3 keywords. The analysis script outputs `evidence` objects with `company`, `source_type`, `quote`, `url`, and `page_title` or `job_title`.
Format evidence as:
```markdown
> **Evidence — "keyword1":**
>
> - [company1.com](url) (page title): "...exact quote with keyword in context..."
> - [company2.com](url) (job: "Job Title"): "...exact quote from job listing..."
>
> **"keyword2":**
>
> - [company3.com](url) (page title): "...exact quote..."
```
Each evidence entry must include:
1. **Company domain** as a link to the source URL
2. **Source context** — page title for websites, job title for listings
3. **Exact quote** — the ±40 char snippet around the keyword match from the raw text
4. **Vendor-adjacent annotation** — If the evidence comes from a company that also sells a similar product (e.g., their pricing page mentions the keyword), mark with ⚠️ and note "vendor-adjacent". Clear buyer signals get ✅.
### Sales-Specific Keywords: Source Breakdown
For sales-specific keywords, add a **Source** column showing where matches came from:
```markdown
| Keyword | Won (n=X) | Lost (n=Y) | Lift | Source (website / jobs / both) | Interpretation |
```
Source format: `3w / 20j / 2both` (3 from website only, 20 from job descriptions only, 2 from both)
### Tech Stack Keywords: Niche Tool Mentions
Search for specific SaaS tools (not generic keywords like "cloud" or "security"). Group by category:
- Sales & Revenue Tools
- Data & Analytics Tools
- Customer Success & Support
- HR & ATS
- Anti-Fit Tech Stack
### Anti-Fit Keywords
Separate table for keywords with lift < 0.5x.
---
## Section 3: Structured Signal Categories
GTM motion indicators, infrastructure maturity tables with Won%, Lost%, and interpretation.
---
## Section 4: Job Hiring Signals
Role prevalence in won companies. If lost companies lack job data, present won-only with note.
---
## Section 5: Anti-Fit Signals & Competitive Tool Users
### Anti-Fit Signals Table
Website content anti-signals table for keywords with lift < 0.5x that indicate structural misfit.
### Structural Anti-Fit Patterns
Patterns indicating the company is not a fit:
- Selling the same product category (competitor, not buyer)
- No job listings in 12+ months (not growing/hiring)
- Consumer-focused business model (if target sells B2B)
- Industry/vertical mismatch
### Competitive Tool Users (Migration Opportunity Segment)
**DO NOT exclude companies using competitor tools.** Instead, create a separate prospecting segment:
```markdown
| Company Segment | Count | Approach |
| -------------------- | -------------- | ------------------------------------------ |
| Using [Competitor A] | N (X% of lost) | Displacement messaging, comparison content |
| Using [Competitor B] | N (X% of lost) | Migration case studies |
```
### Red Flag Checklist
Deprioritize if 2+ present (excluding competitive tool usage):
- ✅ Selling the same product (competitor)
- ⚠️ No relevant job listings in 12 months
- ⚠️ <50 employees
- ⚠️ Consumer-only business model
---
## Section 6: Composite Lead Scoring Model
0-100 point model organized in 3 tiers:
- Tier 1: Core Fit (0-40 points) — regulatory, compliance, or structural signals
- Tier 2: Sophistication (0-30 points) — fraud/risk/product maturity signals
- Tier 3: Developer / Integration Fit (0-30 points) — API-first, tech stack signals
Include scoring examples from the dataset (2 won, 2 lost with full point breakdown).
**CRITICAL — Scoring reconciliation:** After writing this section, cross-check every signal's point value against Section 0.5 (Lead Scoring Cheatsheet). They MUST match. Mismatches between the quick-reference and detailed sections confuse users.
---
## Section 7: Niche First-Party Signals to Pull
Actionable checklist grouped by priority:
- Highest-value (pull for every prospect)
- High-value (pull for Tier 1-2)
- Enrichment signals (context for personalization)
---
## Section 8: Won vs Lost Comparison
Side-by-side archetype profiles with concrete examples from the dataset.
---
## Section 9: Recommended Prospecting Workflow
4-step targeting guide: Build list → Enrich → Score → Personalize.
Include personalization hooks for each top signal:
```
• Regulatory trigger: "Noticed [Company] has [signal] — companies at this maturity typically [pain point]..."
• Tech stack signal: "Saw [Company] uses [tool] — teams running [tool] often face [specific friction]..."
• Hiring signal: "Saw you're hiring a [role] — we work with [comparable company] to help their [team] focus on [outcome] rather than [pain]..."
• Competitive signal: "You're using [competitor] — [specific reason a switch makes sense right now]..."
```
references/scoring-pitfalls.md›
# What NOT to use for scoring — confirmation-biased CRM fields
Some CRM fields look like predictive signals but are actually downstream artifacts of AE engagement. They get populated **after** an AE decides an opportunity is worth working — so their correlation with win-rate is measuring "did the AE work this deal" rather than any causal property of the account. **Treat them as warning flags about pipeline hygiene, never as inputs to an ICP scoring model.**
## The fields to exclude
| Field | Why it's confounded |
| ------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| Catalyst note count (or any per-opp activity count) | AEs log notes on deals they're actively working. Won deals have more notes because someone was working them; the note count doesn't cause the win. On one real run, this field showed "109x lift" as the most extreme signal in the dataset — headline material — until we noticed the direction of causality. |
| `number_of_champions_c` / `number_of_decision_makers_c` / economic-buyer count | AE-counted fields. They get filled in during deal qualification, meaning they're present when someone bothered to qualify the deal. An unworked opp shows 0 across the board regardless of the account's actual buying committee. |
| `OpportunityContactRole` derivatives — total roles, champion count, technical-evaluator count | Same failure mode: OCRs are created as a byproduct of AE effort, not as evidence about the account itself. |
| MEDDPICC picklists (champion score, decision criteria score, decision process score, identify-implicate pain score) | In practice these fields are sparsely populated or uncalibrated. On one real run, both won and lost opps averaged <1 across all four scores — the signal is noise and the correlation is zero. Even when populated, they're AE judgments formed during qualification, not independent properties of the account. |
| Any field of the form "did the AE do X on this opp" | Rule of thumb: if the field is populated by the AE during deal execution, it's downstream of engagement. |
## How to think about it
The question an ICP scoring model should answer is "should we work this account?" — which means every input has to be observable **without** already working the account. Website content, job listings, tech stack, firmographics, and third-party news or funding signals all pass that test. CRM fields populated by AE activity don't.
## Safer alternative read for loss-reason data
Loss reasons themselves (`loss_decline_reason_c` = 'Poor Fit', 'Unresponsive', 'Return on Investment', etc.) ARE useful, but as a **diagnosis of the top-of-funnel ICP** rather than as scoring inputs. If 65% of your lost pipeline is "Poor Fit" + "Unresponsive", the lever is tighter ICP gating before opps get created — which is what this whole skill produces.
references/signal-interpretation.md›
# Signal Interpretation Rules
Rules for correctly interpreting whether a signal indicates a buyer, competitor, or neutral company.
## Rule 1: Seller vs Buyer Distinction
If a company's **website content** mentions terms describing what the target product sells, they're likely a **competitor or adjacent vendor**, NOT a buyer.
**Example:** For an AR automation tool, a company whose website says "our collections automation platform" is a seller/competitor. A company whose job listing says "seeking AR Manager to reduce DSO" is a buyer.
**Application:** When a keyword has high lift but companies mentioning it are vendors in the target's space, flag it as a competitor signal, not a buying signal.
## Rule 2: Job Listings = Highest Intent
Hiring for roles related to the target's domain = very high buying signal:
- They have the pain point (they need the role)
- They are actively investing to solve it (budget allocated)
- They may not know automation exists (hiring humans instead)
**Application:** Weight job listing signals higher than website content signals. A company hiring 3 AEs is a stronger signal than a company with "sales" on their website.
## Rule 3: Tech Stack Correlation
Not all tech signals are equal. Consider whether the technology **correlates** with or is **inversely correlated** to the target's use case:
- **Positive correlation:** Technologies that create MORE complexity the target solves (ERP, CRM, payment processors for AR tools)
- **Inverse correlation:** Technologies that SOLVE the problem already (Shopify for AR tools — consumer payments are immediate, not invoiced)
## Rule 4: Source Context Matters
Same keyword means different things depending on WHERE it appears:
- **Product/features page:** Company SELLS this capability → competitor signal
- **Careers/jobs page:** Company NEEDS this capability → buyer signal
- **Blog/case study:** Could be either — evaluate if they're writing as vendor or sharing operational experience
- **Integrations page:** They connect to relevant systems → infrastructure signal
## Rule 5: n=1 Signals Need Verification
A signal appearing in only 1 won company with 0 lost companies produces mathematically high lift but is statistically unreliable. From actual runs:
- n=1 signals with 10x lift scored higher than n=4 signals with 3x lift under the original formula
- After correction: weight n=1 signals at 0.3x, n=2 at 0.6x, n=3+ at 1.0x
**Application:** Flag n=1 signals in reports with "_(single company — verify)_". Never use n=1 signals as Tier 1 scoring criteria.
## Rule 6: Website Signals Fail for Back-Office Tools
For B2B infrastructure tools (AR automation, billing, compliance, HR), buyers don't publish their pain on public websites. From actual analyses:
- 0 of 7 verified customers had AR-related website signals
- All "accounts receivable" signals came from false positives (vendors, not buyers)
- Verified customers (e.g., wholesale distributors, manufacturers, enterprise brands) talk about their own products, not back-office operations
**Application:** For back-office tool verticals, deprioritize website keyword signals. Instead rely on:
1. Job listings (hiring AR Manager, Collections Specialist = active pain + budget)
2. Tech stack signals (NetSuite, Salesforce, Stripe in job descriptions)
3. Business model indicators (B2B invoicing, wholesale distribution)
4. Firmographics (industry vertical, company size, revenue model)
## Rule 7: Anti-Fit Signals Are as Valuable as Fit Signals
From actual analyses: 80% of lost companies were disqualifiable using just 3 anti-fit signals (shopper, checkout, cancel). Identifying non-buyers early prevents wasted outreach.
**Application:** Always generate a dedicated anti-fit section. Key anti-fit patterns:
- Consumer signals (shopper, checkout, cart, debit card) → B2C, not B2B
- Retention/churn language → consumer subscription, not enterprise
- Product category language on product pages → competitor, not buyer
- No job listings → not growing, no budget
## Applying These Rules in Reports
When writing the interpretation column:
1. State what the signal indicates about the prospect's operations
2. Explain WHY it matters for the specific target company
3. Flag ambiguous signals (e.g., "Note: some companies mention this as vendors, not buyers")
4. For ALL keywords, note whether the source is primarily website or job descriptions — job-sourced signals are higher confidence
5. For tech stack tools, explain what the tool usage implies about the org's maturity and needs
6. For n=1 signals, add verification note
7. For back-office tool verticals, explicitly call out when website signals are unreliable and recommend alternative signal sources
references/step-7-prospects.md›
# Step 7 — Top 10 net-new prospects (required deliverable)
A signal report without a companion prospect list is incomplete. The signals tell you what to look for; Step 7 produces the actual "here are 10 real companies the user should pursue" list. **This is a hard requirement of the pipeline.**
## What's required vs. what's optional
| Output | Status | Why |
| ------------------------------------------------------------------------------------------ | --------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **10 net-new companies** with descriptions, signal scores, matched signals, cited evidence | **REQUIRED** — every run | A signal report without target companies forces the user to do their own prospecting pass, which is the expensive thing they wanted to skip. |
| **Contacts + corporate emails** at those companies | **OPTIONAL** — credit-aware | Contact discovery uses additional Deepline credits for the email waterfall. Always offer it; only run it if the user approves the credit spend. Default to `--no-contacts` if the user hasn't said. |
When contacts are skipped, the prospect cards still need to ship — they just include "(contacts not enriched — re-run with `--contacts` to add)" in place of the contact bullets.
## What every prospect card must contain
1. **Company identity** — name, apex domain, 1-sentence description (location, what they do)
2. **Category label** from Step 1.0.5: Net-new / Account-only / Re-engage; excluded items dropped entirely
3. **Signal score** + the 4–6 matched signals from Step 4/5, shown as inline code
4. **3–5 cited evidence quotes** from the company's own website or job listings (same format as Step 5) proving the signals actually hit
5. **(if `--contacts` is on)** 1–3 named contacts with full name (linked to LinkedIn), title, and corporate email validated against the company's apex domain. "(email not found)" annotation when the waterfall returned nothing — be honest about gaps rather than shipping false positives.
## Use the shipped orchestrator
`scripts/find_contacts.py` runs the full chain via Deepline. Don't rebuild it inline.
```bash
# Companies only (no credit spend on contacts):
python3 scripts/find_contacts.py \
--input prospects_actionable.csv \
--output top10.csv \
--top 10 \
--no-contacts
# Companies + contacts + emails (asks for credit approval).
# IMPORTANT: --roles must be the buyer-persona job titles from THIS run's
# Step 0/0.5 ecosystem discovery. Don't reuse another vertical's roles.
python3 scripts/find_contacts.py \
--input prospects_actionable.csv \
--output top10.csv \
--roles "<persona job titles from Step 0.5>" \
--top 10 \
--contacts
# Vertical examples (substitute the persona that came out of YOUR Step 0.5):
# creative ops: "Creative Director,Brand Manager,Content Operations Lead,Marketing Operations"
# AR automation: "AR Manager,Accounts Receivable Specialist,Controller,Finance Director"
# sales engagement: "SDR Manager,Sales Operations,VP Sales,Head of Sales Development"
# developer tools: "Staff Engineer,Platform Engineer,DevOps Lead,Engineering Manager"
# metal AM (the run that motivated this skill): "Design Engineer,Mechanical Design Engineer,Additive Manufacturing Engineer,DfAM Engineer"
```
## The 3-phase contact chain (when `--contacts` is on)
**Phase 1 — `company-to-contact` (FREE tier).** Dropleads → deepline_native → Icypeas → Prospeo → Crustdata. Works on >200-employee US/EU companies. Returns LinkedIn URLs + titles, often no emails. Run first because it's free.
**Phase 2 — `exa_search_people` fallback for the gaps.** For any company Phase 1 returned ZERO contacts for, fall back to `exa_search_people` with `includeDomains=['linkedin.com']` and a query like `"Design Engineer OR Mechanical Engineer OR DfAM Engineer at {{company_name}}"`. Exa neural search finds LinkedIn profiles by semantic match against the company name — far better coverage for small / non-US / niche industrial targets than the B2B provider waterfall.
**This fallback is not optional.** On the run that motivated adding it, Phase 1 returned 0 contacts on all 10 top-scoring prospects (all <200-employee industrial companies). Phase 2 found 15 real contacts at 6 of those same 10 in the same run. If you skip Phase 2 because Phase 1 "worked" (ran without error and returned 0 rows), you ship an empty list with no explanation.
**Two Exa guardrails matter:**
- **Title parse**: Exa results follow `Name | Role at Company`. Regex the name out of the leading capitalized tokens; treat the rest as the title. Fall back to de-slugging the LinkedIn URL when the title doesn't parse cleanly.
- **Company-match filter**: Require the company name (or its first ~8 characters) to appear somewhere in the result title or text. Exa neural will sometimes return profiles at COMPETING companies — e.g., searching for "plasma process engineer at Plasma Processes" returned a real plasma process engineer working at Hypertherm. Discard any result where the company name doesn't match.
**Phase 3 — `name-and-domain-to-email-waterfall` for email resolution.** For every named contact with a LinkedIn URL, resolve the company domain, then run this waterfall with both `domain` and `linkedin_url`. Returns one primary email per contact.
**Domain-match validation is mandatory.** Always check that the returned email's apex domain matches the company's apex domain before publishing. Providers return stale addresses often enough that this is the difference between a usable list and an embarrassing one. On one real run:
- A contact at X-Bow Systems came back with `@orbitalatk.com` (his previous employer, acquired into Northrop Grumman years earlier)
- Another came back with `@governors-america.com` (a completely unrelated company)
- A third came back with a personal `@googlemail.com`
Use `extract_apex()` from `scripts/dedupe_utils.py` on both the email domain and the company domain; if they don't match, mark `email_source=apex_mismatch` and publish "(email not found)" instead. Keep the raw value in `raw_email` for auditing.
## Output card skeleton
Render each prospect as a card (heading + description + signals + evidence + contacts), not a wide table. Cards make it possible to include the required 3–5 evidence quotes per prospect without blowing up layout.
```markdown
### [company name] — score [N] [category badge]
_1-sentence description of the company._
Domain: `apex.com`
Matched signals: `signal1`, `signal2`, `signal3`, `signal4`
**Cited evidence:**
- [📄 website] [page title]: "...exact quote around keyword..."
https://apex.com/source-page
- [💼 job] [job title]: "...exact quote from listing..."
https://linkedin.com/jobs/view/...
**Contacts:** (only if --contacts was on)
- **Full Name** — Role · ✉ `[email protected]`
https://linkedin.com/in/profile
- **Other Name** — Role · (email not found)
https://linkedin.com/in/profile
```
## How many is "10"
10 is a ceiling, not a floor. If the actionable pool (post-dedupe, scored) has fewer than 10 companies that pass the minimum score threshold, ship whatever you have and explain the shortfall — don't pad with low-confidence entries. If the pool has more than 10, prefer top-scoring first, then break ties on category preference (net-new > account-only > re-engage) to surface the cleanest outbound targets.
scripts/analyze_signals.py›
#!/usr/bin/env python3
"""
Differential signal analysis for ICP niche signal discovery.
Reads a Deepline-enriched CSV, parses exa_search website content and crustdata
job listings, computes Laplace-smoothed lift scores for keyword categories,
extracts tech stack tools, and outputs JSON results.
Usage:
python3 analyze_signals.py \\
--input enriched.csv \\
--keywords keywords.json \\
--tools tools.json \\
--job-roles job_roles.json \\
--output analysis.json
Options:
--input Path to enriched CSV (required)
--keywords Path to JSON file with keyword categories (required)
--tools Path to JSON file with tech stack tools (required)
--job-roles Path to JSON file with job role categories (required)
--output Path for JSON output (default: stdout)
--website-col Column index for website data (auto-detected if omitted)
--jobs-col Column index for job listings (auto-detected if omitted)
--status-col Column name for won/lost status (default: "status")
See references/keyword-catalog.md for JSON format examples and guidance on
building target-specific keyword, tool, and job role lists.
"""
import csv
import json
import sys
import re
import argparse
from collections import defaultdict
csv.field_size_limit(sys.maxsize)
def auto_detect_columns(headers):
"""Find website and jobs columns by looking for __dl_full_result__ pattern."""
website_col = None
jobs_col = None
for i, h in enumerate(headers):
if "__dl_full_result__" in h:
# Try to determine if it's website or jobs by checking position
if website_col is None:
website_col = i
elif jobs_col is None:
jobs_col = i
return website_col, jobs_col
def parse_website_content(cell_value):
"""Extract text content from exa_search results.
Returns:
combined_text: all page text concatenated (lowercased)
pages: list of {url, title, text} per page (text is lowercased)
"""
if not cell_value or cell_value.strip() == "":
return "", []
try:
data = json.loads(cell_value)
except (json.JSONDecodeError, TypeError):
return str(cell_value), []
texts = []
pages = []
# Handle various response shapes
results = []
if isinstance(data, dict):
results = data.get("data", {}).get("results", []) if isinstance(data.get("data"), dict) else []
if not results:
results = data.get("results", [])
elif isinstance(data, list):
results = data
for r in results:
if isinstance(r, dict):
text = r.get("text", "")
url = r.get("url", "")
title = r.get("title", "")
if text:
texts.append(text)
if url:
pages.append({"url": url, "title": title, "text": text.lower()})
return " ".join(texts).lower(), pages
def parse_job_listings(cell_value):
"""Extract job titles and descriptions from crustdata job listings.
Returns:
listings: list of {title, description, url, text} per listing (text is lowercased)
combined_text: all listing text concatenated (lowercased)
"""
if not cell_value or cell_value.strip() == "":
return [], ""
try:
data = json.loads(cell_value)
except (json.JSONDecodeError, TypeError):
return [], str(cell_value)
listings = []
all_text = []
# Handle various response shapes
raw_listings = []
if isinstance(data, dict):
# {"data": {"listings": [...]}} (legacy exa-like)
raw_listings = data.get("data", {}).get("listings", []) if isinstance(data.get("data"), dict) else []
# {"result": {"listings": [...]}} (Deepline/Crustdata)
if not raw_listings and isinstance(data.get("result"), dict):
raw_listings = data["result"].get("listings", [])
# {"listings": [...]} (flat)
if not raw_listings:
raw_listings = data.get("listings", [])
elif isinstance(data, list):
raw_listings = data
for entry in raw_listings:
if isinstance(entry, dict):
# Crustdata uses "title" and "description" (not "job_title"/"job_description")
title = entry.get("title", entry.get("job_title", ""))
desc = entry.get("description", entry.get("job_description", ""))
url = entry.get("url", "")
combined = f"{title} {desc}"
listings.append({"title": title, "description": desc, "url": url, "text": combined.lower()})
all_text.append(combined)
return listings, " ".join(all_text).lower()
def substring_match(text, keyword):
"""Check if keyword appears as substring in text (case-insensitive)."""
if not text:
return False
return keyword.lower().rstrip("*") in text.lower()
def laplace_lift(won_count, won_total, lost_count, lost_total):
"""Compute Laplace-smoothed lift (Bayesian posterior mean ratio with Jeffreys prior)."""
won_rate = (won_count + 0.5) / (won_total + 1)
lost_rate = (lost_count + 0.5) / (lost_total + 1)
return won_rate / lost_rate
def extract_snippet(text, keyword, context_chars=40):
"""Extract a snippet around the first occurrence of keyword in text."""
idx = text.find(keyword)
if idx == -1:
return None
start = max(0, idx - context_chars)
end = min(len(text), idx + len(keyword) + context_chars)
snippet = text[start:end].strip()
# Clean up: trim to word boundaries
if start > 0:
space = snippet.find(" ")
if space > 0 and space < context_chars // 2:
snippet = snippet[space + 1:]
snippet = "..." + snippet
if end < len(text):
space = snippet.rfind(" ")
if space > len(snippet) - context_chars // 2:
snippet = snippet[:space]
snippet = snippet + "..."
return snippet
def find_source_evidence(keyword, companies, max_evidence=5):
"""Find exact quotes with source URLs for a keyword match.
Returns list of evidence objects with company, source_type, quote, url, and page_title.
"""
evidence = []
kw = keyword.lower().rstrip("*")
for company in companies:
if len(evidence) >= max_evidence:
break
# Check website pages (per-page text has URLs)
for page in company.get("pages", []):
page_text = page.get("text", "")
if kw in page_text:
snippet = extract_snippet(page_text, kw)
if snippet:
evidence.append({
"company": company["domain"],
"source_type": "website",
"quote": snippet,
"url": page.get("url", ""),
"page_title": page.get("title", ""),
})
break # One match per company per source type
# Check job listings (per-listing text has URLs)
for listing in company.get("job_listings", []):
listing_text = listing.get("text", "")
if kw in listing_text:
snippet = extract_snippet(listing_text, kw)
job_title = listing.get("title", "")
if snippet:
evidence.append({
"company": company["domain"],
"source_type": "job_listing",
"quote": snippet,
"url": listing.get("url", ""),
"job_title": job_title,
})
break # One match per company per source type
return evidence
def analyze(input_path, keywords, tools, job_roles,
website_col=None, jobs_col=None, status_col="status"):
"""Run the full differential analysis.
Args:
input_path: Path to enriched CSV
keywords: Dict of category -> list of keyword strings
tools: Dict of category -> list of tool name strings
job_roles: Dict of role_name -> list of role keyword strings
website_col: Column index for website data (auto-detected if None)
jobs_col: Column index for job listings (auto-detected if None)
status_col: Column name for won/lost status
"""
# Read CSV
with open(input_path, "r", encoding="utf-8") as f:
reader = csv.reader(f)
headers = next(reader)
rows = list(reader)
# Auto-detect columns if not specified
if website_col is None or jobs_col is None:
auto_web, auto_jobs = auto_detect_columns(headers)
if website_col is None:
website_col = auto_web
if jobs_col is None:
jobs_col = auto_jobs
# Find status column
status_idx = None
for i, h in enumerate(headers):
if h.lower().strip() == status_col.lower():
status_idx = i
break
if status_idx is None:
raise ValueError(f"Status column '{status_col}' not found in headers: {headers}")
# Parse companies
companies = []
for row in rows:
if len(row) <= max(status_idx, website_col or 0, jobs_col or 0):
continue
status = row[status_idx].strip().lower()
if status not in ("won", "lost"):
continue
domain = row[0].strip() if row[0] else "unknown"
website_text = ""
pages = []
if website_col is not None and website_col < len(row):
website_text, pages = parse_website_content(row[website_col])
job_listings = []
jobs_text = ""
if jobs_col is not None and jobs_col < len(row):
job_listings, jobs_text = parse_job_listings(row[jobs_col])
# Combined text for general keyword matching
combined_text = f"{website_text} {jobs_text}"
companies.append({
"domain": domain,
"status": status,
"website_text": website_text,
"jobs_text": jobs_text,
"combined_text": combined_text,
"pages": pages,
"job_listings": job_listings,
"has_website": len(website_text) > 100,
"has_jobs": len(job_listings) > 0,
})
won = [c for c in companies if c["status"] == "won"]
lost = [c for c in companies if c["status"] == "lost"]
won_total = len(won)
lost_total = len(lost)
# ── Keyword analysis ──
keyword_results = {}
for category, kws in keywords.items():
category_results = []
for kw in kws:
kw_lower = kw.lower().rstrip("*")
won_count = sum(1 for c in won if kw_lower in c["combined_text"])
lost_count = sum(1 for c in lost if kw_lower in c["combined_text"])
lift = laplace_lift(won_count, won_total, lost_count, lost_total)
# Source breakdown (website vs jobs vs both)
won_web = sum(1 for c in won if kw_lower in c["website_text"] and kw_lower not in c["jobs_text"])
won_jobs = sum(1 for c in won if kw_lower not in c["website_text"] and kw_lower in c["jobs_text"])
won_both = sum(1 for c in won if kw_lower in c["website_text"] and kw_lower in c["jobs_text"])
evidence = find_source_evidence(kw, won + lost)
category_results.append({
"keyword": kw,
"won_count": won_count,
"won_pct": round(won_count / won_total * 100, 1) if won_total else 0,
"lost_count": lost_count,
"lost_pct": round(lost_count / lost_total * 100, 1) if lost_total else 0,
"lift": round(lift, 2),
"source_breakdown": {
"website_only": won_web,
"jobs_only": won_jobs,
"both": won_both
},
"evidence": evidence
})
category_results.sort(key=lambda x: x["lift"], reverse=True)
keyword_results[category] = category_results
# ── Tech stack tool analysis ──
tool_results = {}
for category, tool_list in tools.items():
category_results = []
for tool in tool_list:
tool_lower = tool.lower()
won_count = sum(1 for c in won if tool_lower in c["combined_text"])
lost_count = sum(1 for c in lost if tool_lower in c["combined_text"])
if won_count < 2 and lost_count < 1:
continue
lift = laplace_lift(won_count, won_total, lost_count, lost_total)
evidence = find_source_evidence(tool, won + lost)
category_results.append({
"tool": tool,
"won_count": won_count,
"won_pct": round(won_count / won_total * 100, 1) if won_total else 0,
"lost_count": lost_count,
"lost_pct": round(lost_count / lost_total * 100, 1) if lost_total else 0,
"lift": round(lift, 2),
"evidence": evidence
})
category_results.sort(key=lambda x: x["lift"], reverse=True)
tool_results[category] = category_results
# ── Job role analysis ──
job_role_results = {}
won_with_jobs = [c for c in won if c["has_jobs"]]
lost_with_jobs = [c for c in lost if c["has_jobs"]]
for role_name, role_keywords in job_roles.items():
won_match = sum(1 for c in won_with_jobs
if any(rk in c["jobs_text"] for rk in role_keywords))
lost_match = sum(1 for c in lost_with_jobs
if any(rk in c["jobs_text"] for rk in role_keywords))
job_role_results[role_name] = {
"won_count": won_match,
"won_with_jobs": len(won_with_jobs),
"won_pct": round(won_match / len(won_with_jobs) * 100, 1) if won_with_jobs else 0,
"lost_count": lost_match,
"lost_with_jobs": len(lost_with_jobs),
"lost_pct": round(lost_match / len(lost_with_jobs) * 100, 1) if lost_with_jobs else 0,
}
# ── Stats ──
won_with_content = sum(1 for c in won if c["has_website"])
lost_with_content = sum(1 for c in lost if c["has_website"])
avg_won_chars = (sum(len(c["website_text"]) for c in won) / won_total) if won_total else 0
avg_lost_chars = (sum(len(c["website_text"]) for c in lost) / lost_total) if lost_total else 0
stats = {
"won_total": won_total,
"lost_total": lost_total,
"won_with_content": won_with_content,
"lost_with_content": lost_with_content,
"won_with_jobs": len(won_with_jobs),
"lost_with_jobs": len(lost_with_jobs),
"won_coverage_pct": round(won_with_content / won_total * 100, 1) if won_total else 0,
"lost_coverage_pct": round(lost_with_content / lost_total * 100, 1) if lost_total else 0,
"avg_won_chars": round(avg_won_chars),
"avg_lost_chars": round(avg_lost_chars),
}
# ── Anti-fit signals (lift < 0.5) ──
anti_fit = []
for category, results in keyword_results.items():
for r in results:
if r["lift"] < 0.5 and (r["won_count"] > 0 or r["lost_count"] > 1):
anti_fit.append({**r, "category": category})
anti_fit.sort(key=lambda x: x["lift"])
return {
"keyword_results": keyword_results,
"tool_results": tool_results,
"job_results": job_role_results,
"stats": stats,
"anti_fit": anti_fit,
}
def main():
parser = argparse.ArgumentParser(
description="Differential signal analysis for ICP",
epilog="See references/keyword-catalog.md for JSON format examples."
)
parser.add_argument("--input", required=True, help="Path to enriched CSV")
parser.add_argument("--keywords", required=True, help="Path to JSON file with keyword categories")
parser.add_argument("--tools", required=True, help="Path to JSON file with tech stack tools")
parser.add_argument("--job-roles", required=True, help="Path to JSON file with job role categories")
parser.add_argument("--output", help="Path for JSON output (default: stdout)")
parser.add_argument("--website-col", type=int, help="Column index for website data")
parser.add_argument("--jobs-col", type=int, help="Column index for job listings")
parser.add_argument("--status-col", default="status", help="Column name for won/lost status")
args = parser.parse_args()
with open(args.keywords) as f:
keywords = json.load(f)
with open(args.tools) as f:
tools = json.load(f)
with open(args.job_roles) as f:
job_roles = json.load(f)
results = analyze(
input_path=args.input,
keywords=keywords,
tools=tools,
job_roles=job_roles,
website_col=args.website_col,
jobs_col=args.jobs_col,
status_col=args.status_col,
)
output = json.dumps(results, indent=2)
if args.output:
with open(args.output, "w") as f:
f.write(output)
print(f"Analysis written to {args.output}", file=sys.stderr)
print(f"Stats: {json.dumps(results['stats'], indent=2)}", file=sys.stderr)
else:
print(output)
if __name__ == "__main__":
main()
scripts/dedupe_utils.py›
#!/usr/bin/env python3
"""
Deduplication utilities for niche-signal-discovery prospect lists.
Primary match: apex-domain (public-suffix-aware, handles multi-label TLDs).
Fallback match: fuzzy company name (after stripping corporate suffixes).
The standard library only — no pip dependencies. `difflib.SequenceMatcher`
is used for the fuzzy name ratio so this runs anywhere Python 3 does.
Usage from another script:
from dedupe_utils import extract_apex, norm_name, match_against_existing
existing = load_existing_csv("customers.csv") # rows with 'domain' or 'name'
candidates = load_candidates("prospects.csv") # rows with 'domain' and 'name'
actionable, matched = match_against_existing(candidates, existing, name_threshold=0.85)
Usage from the command line:
python3 dedupe_utils.py \\
--existing customers.csv \\
--candidates prospects.csv \\
--out-actionable net_new.csv \\
--out-matched already_known.csv
Why this exists:
- Raw string match misses parent-company relationships. amsynergy.nikon.com
and nikon.com refer to the same buyer-side organization, but a naive set
lookup treats them as unrelated.
- Name matching alone is noisy: a candidate named "Rocket Propulsion Systems"
can collide with an unrelated CRM row named "Rocket Propulsion" as a
substring. Apex domain is a stronger primary key when it's available.
- The fix is a layered check: match on apex domain first (strong signal),
then fall back to normalized fuzzy name match with a high threshold only
when no domain match exists.
"""
from __future__ import annotations
import argparse
import csv
import re
import sys
from difflib import SequenceMatcher
from typing import Iterable, Sequence
from urllib.parse import urlparse
# ----------------------------------------------------------------------
# Apex domain extraction
# ----------------------------------------------------------------------
# A curated list of multi-label public suffixes that show up often in B2B
# datasets. This is NOT a complete public-suffix-list dump — a production
# system should use `tldextract` — but it covers the countries that have
# appeared in real runs (US, UK, JP, KR, AU, DE, BR, CN, etc).
MULTI_LABEL_SUFFIXES: set[str] = {
# United Kingdom
"co.uk", "org.uk", "ac.uk", "gov.uk", "ltd.uk", "plc.uk", "net.uk",
# Japan
"co.jp", "ac.jp", "or.jp", "go.jp", "ne.jp",
# Korea
"co.kr", "ac.kr", "go.kr", "or.kr", "re.kr",
# Australia / New Zealand
"com.au", "net.au", "org.au", "edu.au", "gov.au",
"co.nz", "ac.nz",
# Brazil / China / India / Israel / South Africa
"com.br", "com.cn", "net.cn", "co.il", "co.in", "ac.in", "edu.in",
"co.za", "ac.za",
# Europe — country-specific commercial
"co.it", "co.es", "com.es",
# APAC / LATAM misc
"com.mx", "edu.mx", "org.mx",
"com.hk", "com.sg", "com.tr", "com.tw", "com.ar", "com.co", "com.pe",
"com.ph", "com.my", "com.pk", "com.eg", "com.sa", "com.ua", "com.vn",
"co.th",
}
def extract_apex(url_or_host: str) -> str:
"""Normalize a URL or bare hostname to its registrable apex domain.
Returns an empty string when the input can't be parsed into something
that looks like a domain (empty input, IP literal, obvious garbage).
This is intentional — downstream code should treat "" as "skip this
row" rather than as a valid apex.
Examples:
extract_apex("amsynergy.nikon.com") -> "nikon.com"
extract_apex("industry.nikon.com") -> "nikon.com"
extract_apex("nikon.co.jp") -> "nikon.co.jp"
extract_apex("www.bbc.co.uk") -> "bbc.co.uk"
extract_apex("https://corporate.arcelormittal.com/careers") -> "arcelormittal.com"
extract_apex("") -> ""
"""
if not url_or_host:
return ""
s = url_or_host.strip().lower()
if not s:
return ""
# Prepend a scheme if missing so urlparse populates .netloc.
if not re.match(r"^https?://", s):
s = "http://" + s
try:
host = urlparse(s).netloc
except Exception:
return ""
# Strip port + leading www. variants.
host = host.split(":")[0].strip("/").split("/")[0]
while host.startswith("www."):
host = host[4:]
# Reject obvious non-domains.
if "." not in host or " " in host:
return ""
parts = host.split(".")
if len(parts) < 2:
return host
# If the last two labels form a multi-label suffix (e.g., co.uk), the
# registrable root is the last THREE labels.
if len(parts) >= 3:
last_two = ".".join(parts[-2:])
if last_two in MULTI_LABEL_SUFFIXES:
return ".".join(parts[-3:])
return ".".join(parts[-2:])
# ----------------------------------------------------------------------
# Company-name normalization + fuzzy matching
# ----------------------------------------------------------------------
# Corporate suffix tokens to strip before comparing two company names.
# Order matters only for readability — the regex compiles to a single pass.
_CORP_SUFFIX_TOKENS: Sequence[str] = (
"inc", "llc", "ltd", "gmbh", "sa", "ag", "co", "corp", "corporation",
"company", "group", "holdings", "limited", "plc", "bv", "srl", "spa",
"oy", "ab", "pte", "pty", "kg", "mbh", "cie", "sarl",
"tech", "technologies", "systems", "solutions", "industries",
"international", "global",
)
_CORP_SUFFIX_RE = re.compile(
r"\b(?:" + "|".join(re.escape(t) for t in _CORP_SUFFIX_TOKENS) + r")\b\.?",
flags=re.IGNORECASE,
)
def norm_name(company_name: str) -> str:
"""Normalize a company name for fuzzy comparison.
- Lowercase
- Strip corporate suffix tokens (Inc, LLC, Ltd, GmbH, Holdings, ...)
- Keep only [a-z0-9 ]
- Collapse whitespace
Returns "" when normalization leaves less than 3 characters — names that
short are too noisy to match reliably.
Examples:
norm_name("Astura Medical, Inc.") -> "astura medical"
norm_name("MBDA Systems Holdings Ltd") -> "mbda"
norm_name("3DMorphic") -> "3dmorphic"
norm_name("SA") -> ""
"""
if not company_name:
return ""
n = company_name.strip().lower()
n = _CORP_SUFFIX_RE.sub("", n)
n = re.sub(r"[^a-z0-9 ]", "", n)
n = re.sub(r"\s+", " ", n).strip()
if len(n) < 3:
return ""
return n
def name_similarity(a: str, b: str) -> float:
"""Return a 0..1 similarity ratio between two company names after
normalization. Uses difflib.SequenceMatcher which is stdlib and close
enough to Levenshtein ratio for typical corporate-name matching."""
na = norm_name(a)
nb = norm_name(b)
if not na or not nb:
return 0.0
return SequenceMatcher(None, na, nb).ratio()
# ----------------------------------------------------------------------
# Combined match-against-existing helper
# ----------------------------------------------------------------------
def build_existing_index(
existing_rows: Iterable[dict],
domain_field: str = "domain",
name_field: str = "name",
website_field: str | None = "website",
) -> tuple[set[str], dict[str, str]]:
"""Build an apex-domain set + normalized-name index from existing rows.
Returns:
(apex_set, name_to_apex):
apex_set is the set of apex domains present in the existing list.
name_to_apex maps every normalized company name to the apex it
came from, so a name-match can report which row it collided with.
existing_rows can be anything iterable of dicts. Pass rows from your
CRM export, a previous prospect-list CSV, a customer-list download,
or whatever the user provides as "do not re-contact".
"""
apex_set: set[str] = set()
name_to_apex: dict[str, str] = {}
for r in existing_rows:
apex = ""
# Prefer an explicit domain field, fall back to website.
for fld in (domain_field, website_field):
if fld and r.get(fld):
apex = extract_apex(r[fld])
if apex:
break
if apex:
apex_set.add(apex)
nm = norm_name(r.get(name_field, "")) if name_field else ""
if nm:
# Don't overwrite a shorter existing key with a longer one; the
# first occurrence wins so downstream messages are stable.
name_to_apex.setdefault(nm, apex or "")
return apex_set, name_to_apex
def check_duplicate(
candidate: dict,
apex_set: set[str],
name_to_apex: dict[str, str],
domain_field: str = "domain",
name_field: str = "name",
website_field: str | None = "website",
name_threshold: float = 0.85,
) -> tuple[bool, str]:
"""Check a single candidate row against the existing index.
Returns (is_duplicate, reason). reason is a short string explaining the
match (e.g. "apex:nikon.com" or "name:astura medical (0.91)") or "" when
the candidate is net-new.
The match is layered:
1. Apex domain match against apex_set (strong, preferred).
2. If no domain match, walk name_to_apex looking for a fuzzy match
above name_threshold. Only used as a fallback because name-only
matches are noisy (e.g. "rocket propulsion" as a substring).
"""
# Step A: apex domain
apex = ""
for fld in (domain_field, website_field):
if fld and candidate.get(fld):
apex = extract_apex(candidate[fld])
if apex:
break
if apex and apex in apex_set:
return True, f"apex:{apex}"
# Step B: fuzzy company name
cand_name = candidate.get(name_field, "") if name_field else ""
nc = norm_name(cand_name)
if nc:
best_key = ""
best_ratio = 0.0
for existing_name in name_to_apex:
ratio = SequenceMatcher(None, nc, existing_name).ratio()
if ratio > best_ratio:
best_ratio = ratio
best_key = existing_name
if ratio >= 0.99: # perfect match, stop early
break
if best_ratio >= name_threshold:
return True, f"name:{best_key} ({best_ratio:.2f})"
return False, ""
def match_against_existing(
candidates: Iterable[dict],
existing: Iterable[dict],
domain_field: str = "domain",
name_field: str = "name",
website_field: str | None = "website",
name_threshold: float = 0.85,
) -> tuple[list[dict], list[dict]]:
"""Split candidates into (actionable, matched) against an existing list.
Each row in the output carries a `dedupe_match` field describing why
it was classified that way — empty for actionable rows, populated with
the match reason for matched rows. This makes it trivial to surface in
reports and explain to the user why a row was excluded.
"""
apex_set, name_to_apex = build_existing_index(
existing,
domain_field=domain_field,
name_field=name_field,
website_field=website_field,
)
actionable: list[dict] = []
matched: list[dict] = []
for c in candidates:
is_dup, reason = check_duplicate(
c, apex_set, name_to_apex,
domain_field=domain_field,
name_field=name_field,
website_field=website_field,
name_threshold=name_threshold,
)
out = dict(c)
out["dedupe_match"] = reason
if is_dup:
matched.append(out)
else:
actionable.append(out)
return actionable, matched
# ----------------------------------------------------------------------
# Command-line entry point
# ----------------------------------------------------------------------
def _read_csv(path: str) -> list[dict]:
csv.field_size_limit(sys.maxsize)
with open(path) as f:
return list(csv.DictReader(f))
def _write_csv(path: str, rows: list[dict]) -> None:
if not rows:
with open(path, "w", newline="") as f:
f.write("")
return
fieldnames = list(rows[0].keys())
with open(path, "w", newline="") as f:
w = csv.DictWriter(f, fieldnames=fieldnames)
w.writeheader()
w.writerows(rows)
def _selftest() -> int:
"""Sanity-check the apex and name helpers. Exits non-zero on failure."""
apex_cases = [
("amsynergy.nikon.com", "nikon.com"),
("industry.nikon.com", "nikon.com"),
("nikon.co.jp", "nikon.co.jp"),
("www.bbc.co.uk", "bbc.co.uk"),
("corporate.arcelormittal.com", "arcelormittal.com"),
("blog.company.com", "company.com"),
("firehawkaerospace.com", "firehawkaerospace.com"),
("https://www.example.com/foo/bar", "example.com"),
("", ""),
("localhost", ""),
]
name_cases = [
("Astura Medical, Inc.", "astura medical"),
("MBDA Systems Holdings Ltd", "mbda"),
("3DMorphic", "3dmorphic"),
("Firehawk Aerospace", "firehawk aerospace"),
]
failures = 0
for inp, expected in apex_cases:
got = extract_apex(inp)
ok = got == expected
print(f"apex {'OK ' if ok else 'FAIL'} {inp!r:<45} -> {got!r} (expected {expected!r})")
if not ok:
failures += 1
for inp, expected in name_cases:
got = norm_name(inp)
ok = got == expected
print(f"name {'OK ' if ok else 'FAIL'} {inp!r:<45} -> {got!r} (expected {expected!r})")
if not ok:
failures += 1
# Layered-match sanity check
existing = [{"domain": "nikon.com", "name": "Nikon Corporation"}]
candidates = [
{"domain": "amsynergy.nikon.com", "name": "Nikon AM Synergy"},
{"domain": "ad-astra.com", "name": "Ad Astra Rocket Company"},
{"domain": "", "name": "Nikon Corp"}, # name-only fallback
]
actionable, matched = match_against_existing(candidates, existing)
print()
print("actionable:", actionable)
print("matched: ", matched)
if len(actionable) != 1 or actionable[0]["domain"] != "ad-astra.com":
print("FAIL: expected only ad-astra.com to be actionable")
failures += 1
return 0 if failures == 0 else 1
def _main() -> int:
parser = argparse.ArgumentParser(
description="Dedupe a candidate prospect list against an existing list."
)
parser.add_argument("--existing", help="CSV with the do-not-contact list")
parser.add_argument("--candidates", help="CSV with the candidate prospect list")
parser.add_argument("--out-actionable", help="Write actionable rows here")
parser.add_argument("--out-matched", help="Write matched-against-existing rows here")
parser.add_argument("--domain-field", default="domain")
parser.add_argument("--name-field", default="name")
parser.add_argument("--website-field", default="website")
parser.add_argument("--name-threshold", type=float, default=0.85)
parser.add_argument("--selftest", action="store_true",
help="Run built-in sanity tests and exit")
args = parser.parse_args()
if args.selftest:
return _selftest()
if not (args.existing and args.candidates):
parser.print_help()
return 2
existing = _read_csv(args.existing)
candidates = _read_csv(args.candidates)
actionable, matched = match_against_existing(
candidates, existing,
domain_field=args.domain_field,
name_field=args.name_field,
website_field=args.website_field,
name_threshold=args.name_threshold,
)
if args.out_actionable:
_write_csv(args.out_actionable, actionable)
if args.out_matched:
_write_csv(args.out_matched, matched)
print(f"existing: {len(existing)}")
print(f"candidates: {len(candidates)}")
print(f"actionable: {len(actionable)}")
print(f"matched: {len(matched)}")
return 0
if __name__ == "__main__":
sys.exit(_main())
scripts/find_contacts.py›
#!/usr/bin/env python3
"""
Find contacts + emails at a list of prospect companies.
This script implements the contact discovery fallback chain required by
Step 7 of the niche-signal-discovery pipeline. It runs through Deepline in
two phases:
Phase 1: company-to-contact (FREE tier).
Dropleads + Deepline native + Icypeas + Prospeo + Crustdata.
Works well for >200-employee US/EU companies with mature B2B data
coverage. Returns LinkedIn URLs + titles; often no emails.
Phase 2: For any company that Phase 1 returned ZERO contacts for (which
is the common case for <200-employee, non-US, or niche industrial
targets), fall back to exa_search_people with includeDomains=
['linkedin.com']. Exa neural search goes over public web text and
finds LinkedIn profiles that mention the company by name — far
better coverage for small companies than the B2B provider
waterfall. Parse the result titles ("Name | Role at Company") to
pull named contacts.
Phase 3: For every named contact we have a LinkedIn URL for (from either
phase), run name-and-domain-to-email-waterfall to resolve a
corporate email. Validate the result against the company's apex
domain — providers sometimes return stale emails from a previous
employer (e.g. [email protected] when Nick is now at
X-Bow Systems), and this domain-match validation filters them.
Why this fallback chain exists:
On the nTop run that motivated this skill improvement, Phase 1 (the
waterfall) returned ZERO contacts on all 10 top-scoring prospects —
Plasma Processes, Ad Astra Rocket, Avimetal, Axial3D, CubeLabs, NextAero,
Camber Spine, American Additive Mfg, 3D-Side, 3di GmbH. These are mostly
<200-employee industrial companies, many non-US, where B2B waterfall
providers have thin coverage. Exa people search found 15 real named
contacts at 6 of the 10 in the same pass. The moral: the waterfall is
cheaper (free tier) but Exa is the actual discovery engine for niche
industrial / non-US targets. Always run both.
Usage:
python3 find_contacts.py \\
--input prospects.csv \\
--output contacts.csv \\
--roles "Design Engineer,Mechanical Engineer,Additive Manufacturing Engineer" \\
--top 10
Input CSV columns: domain, name, [score], [niche]
Output CSV columns: company, domain, full_name, title, linkedin_url, email,
email_source, discovery_phase, score, niche
The script calls `deepline enrich` for each phase — it's a thin wrapper.
Nothing here bypasses Deepline.
"""
from __future__ import annotations
import argparse
import csv
import json
import os
import re
import subprocess
import sys
from typing import Iterable
# Load the apex-domain helper from the sibling dedupe_utils module.
# We add the script's directory to sys.path so imports work from anywhere.
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from dedupe_utils import extract_apex # noqa: E402
# ----------------------------------------------------------------------
# Small helpers
# ----------------------------------------------------------------------
def _enrich_run_name(output_csv: str) -> str:
stem = os.path.splitext(os.path.basename(output_csv))[0]
slug = re.sub(r"[^a-zA-Z0-9]+", "-", stem).strip("-").lower()
return f"niche-signal-{slug or 'contacts'}"
def _run_deepline_enrich(input_csv: str, output_csv: str, with_specs: list[str]) -> None:
"""Thin wrapper around `deepline enrich`. Raises on non-zero exit."""
cmd = [
"deepline", "enrich",
"--input", input_csv,
"--output", output_csv,
"--name", _enrich_run_name(output_csv),
]
for spec in with_specs:
cmd += ["--with", spec]
print(f"[find_contacts] running: {' '.join(cmd[:4])} (+{len(with_specs)} --with specs)",
file=sys.stderr)
subprocess.run(cmd, check=True)
def _read_csv(path: str) -> list[dict]:
csv.field_size_limit(sys.maxsize)
with open(path) as f:
return list(csv.DictReader(f))
def _write_csv(path: str, rows: list[dict], fieldnames: list[str]) -> None:
with open(path, "w", newline="") as f:
w = csv.DictWriter(f, fieldnames=fieldnames)
w.writeheader()
w.writerows(rows)
def _parse_json_field(val: str):
if not val:
return None
try:
return json.loads(val)
except Exception:
return None
def _slug_to_name(linkedin_url: str) -> str:
"""Best-effort first/last name extraction from a LinkedIn profile slug.
LinkedIn slugs look like `/in/first-last-deadbeef` where the trailing
hash is an account ID. Strip the hash, split on dashes, title-case.
This is a fallback for when providers don't return full_name."""
if not linkedin_url:
return ""
m = re.search(r"/in/([^/?]+)", linkedin_url)
if not m:
return ""
slug = m.group(1)
# Strip trailing hash (6+ hex chars) like `-abbbb4172` or `-342152133`
slug = re.sub(r"-[a-f0-9]{6,}$", "", slug)
parts = [p for p in slug.split("-") if p and not p.isdigit()]
return " ".join(p.capitalize() for p in parts[:3])
# ----------------------------------------------------------------------
# Phase 1: company-to-contact
# ----------------------------------------------------------------------
def phase1_waterfall(
prospects_csv: str,
out_csv: str,
roles: list[str],
seniority: list[str] | None = None,
limit: int = 3,
) -> list[dict]:
"""Run the FREE company→contact waterfall on every prospect.
Returns a list of raw contact dicts: one row per contact discovered,
with fields (company, domain, full_name, title, linkedin_url, discovery_phase).
Phase 1 rarely returns emails — email resolution is Phase 3.
"""
seniority = seniority or ["Senior", "Director", "VP", "Manager"]
spec = json.dumps({
"alias": "contact",
"tool": "company-to-contact",
"payload": {
"domain": "{{domain}}",
"company_name": "{{name}}",
"roles": roles,
"seniority": seniority,
"limit": limit,
},
})
_run_deepline_enrich(prospects_csv, out_csv, [spec])
rows = _read_csv(out_csv)
contacts: list[dict] = []
for r in rows:
parsed = _parse_json_field(r.get("contact", "") or "")
items: list = []
if isinstance(parsed, list):
items = parsed
elif isinstance(parsed, dict):
for value in parsed.values():
if isinstance(value, list):
items = value
break
if isinstance(value, dict) and isinstance(value.get("result"), list):
items = value["result"]
break
for item in items[:limit]:
li = item.get("linkedin", "") or item.get("linkedin_url", "") or item.get("profile_url", "") or ""
full = item.get("full_name") or item.get("name") or _slug_to_name(li)
contacts.append({
"company": r.get("name", ""),
"domain": r.get("domain", ""),
"full_name": full,
"title": item.get("title", "") or "",
"linkedin_url": li,
"discovery_phase": "waterfall",
})
return contacts
# ----------------------------------------------------------------------
# Phase 2: exa_search_people fallback for empty companies
# ----------------------------------------------------------------------
_TITLE_RE = re.compile(
r"^\s*([A-Z][A-Za-zÀ-ÖØ-öø-ÿ'’.\-]+(?:\s+[A-Z][A-Za-zÀ-ÖØ-öø-ÿ'’.\-]+){1,4})\s*[|\-–]\s*(.+)$"
)
# Generic role tokens used to filter out obvious noise (marketers, recruiters,
# unrelated execs) when the caller hasn't supplied vertical-specific roles. The
# vertical-specific tokens come from the --roles argument at runtime — see
# _derive_role_tokens() — so this set stays neutral.
_GENERIC_ROLE_TOKENS = (
"engineer", "designer", "principal", "director", "head", "vp",
"cto", "chief", "manager", "lead", "scientist", "researcher",
)
def _derive_role_tokens(roles: Iterable[str]) -> set[str]:
"""Tokenize the caller's --roles list into a set of single-word filter tokens.
Splits each role on whitespace, lowercases, and unions with the generic
role tokens. The point is to let Phase 2's company-match filter accept any
title that contains a word from the caller's vertical-specific role list,
without baking those words into a hardcoded constant.
"""
tokens: set[str] = set(_GENERIC_ROLE_TOKENS)
for role in roles:
for word in role.lower().split():
if len(word) >= 3: # skip short joiners like "of", "or", "in"
tokens.add(word)
return tokens
def phase2_exa_people(
prospects_csv: str,
out_csv: str,
roles: list[str],
already_covered_domains: set[str],
) -> list[dict]:
"""Run exa_search_people on every prospect that Phase 1 missed.
The Exa neural search is the workhorse for small / non-US / niche
industrial companies where the B2B provider waterfall has thin data.
Include only `linkedin.com` in the domain filter so we get profile
pages, not marketing copy.
Only processes companies in prospects_csv that are NOT already in
`already_covered_domains` (i.e., Phase 1 returned at least one contact
for them). This keeps credit usage tight.
"""
# Filter prospects down to the ones that need the fallback.
all_prospects = _read_csv(prospects_csv)
needs_fallback = [r for r in all_prospects if r["domain"] not in already_covered_domains]
if not needs_fallback:
print("[find_contacts] Phase 1 covered everything, no Phase 2 fallback needed",
file=sys.stderr)
return []
fallback_csv = out_csv.replace(".csv", "_input.csv") if out_csv.endswith(".csv") else out_csv + "_input"
fieldnames = list(all_prospects[0].keys()) if all_prospects else ["domain", "name"]
_write_csv(fallback_csv, needs_fallback, fieldnames)
# Derive the title-filter token set from the caller's --roles list. Tokens
# are matched against Exa result titles in the per-result loop below — this
# replaces a hardcoded vertical-specific keyword set so the script stays
# general across verticals.
role_tokens = _derive_role_tokens(roles)
# Build an Exa query phrase from the roles. Keep it short — Exa neural
# does better with a compact OR'd title list than with a sprawling sentence.
role_clause = " OR ".join(roles[:6])
query = f"{role_clause} at {{{{name}}}}"
spec = json.dumps({
"alias": "exa_people",
"tool": "exa_search_people",
"payload": {
"query": query,
"type": "neural",
"numResults": 10,
"includeDomains": ["linkedin.com"],
},
})
_run_deepline_enrich(fallback_csv, out_csv, [spec])
rows = _read_csv(out_csv)
contacts: list[dict] = []
for r in rows:
parsed = _parse_json_field(r.get("exa_people", "") or "")
results: list = []
if isinstance(parsed, dict):
data = parsed.get("result", {}).get("data", {}) if isinstance(parsed.get("result"), dict) else {}
if isinstance(data, dict):
results = data.get("results", []) or []
company_name = (r.get("name", "") or "").lower()
company_tail = company_name.split(",")[0].split("(")[0].strip()[:8]
for res in results:
if not isinstance(res, dict):
continue
url = res.get("url", "") or ""
if "/in/" not in url:
continue
title = res.get("title", "") or ""
text = (res.get("text", "") or "")[:300]
low_title = title.lower()
# Require the title to look role-relevant AND to mention the
# company name somewhere. The company-name requirement is the
# main false-positive filter — Exa neural sometimes returns
# profiles at COMPETING companies (e.g. on one real run, a search
# for "Plasma Processes" engineers returned a Hypertherm plasma
# process engineer, which is a different employer).
is_role_relevant = any(tok in low_title for tok in role_tokens)
is_company_match = (
company_tail and (company_tail in low_title or company_tail in text.lower())
)
if not (is_role_relevant and is_company_match):
continue
# Parse "Name | Role at Company" out of the title.
m = _TITLE_RE.match(title)
if m:
name = m.group(1).strip()
role = m.group(2).strip()
else:
name = _slug_to_name(url)
role = title
contacts.append({
"company": r.get("name", ""),
"domain": r.get("domain", ""),
"full_name": name,
"title": role,
"linkedin_url": url if url.startswith("http") else f"https://www.{url.lstrip('/')}",
"discovery_phase": "exa_people",
})
return contacts
# ----------------------------------------------------------------------
# Phase 3: email waterfall + domain validation
# ----------------------------------------------------------------------
def phase3_emails(contacts: list[dict], out_csv: str) -> list[dict]:
"""Resolve emails for every contact with a LinkedIn URL.
Uses name-and-domain-to-email-waterfall, which chains pattern validation +
deepline_native + crustdata + PDL. Then validates the returned email
against the company's apex domain — providers occasionally return a
stale email from a previous employer, and domain-mismatch is an
effective filter for that.
Returns the same contact list with `email` and `email_source` populated.
`email` is blank when the resolved address doesn't match the apex
(still kept in `raw_email` so you can inspect). `email_source` is
"corporate_validated" for clean matches, "apex_mismatch" when the
address resolved but looked like a stale/different-employer email,
or "not_found" when the waterfall returned nothing.
"""
if not contacts:
return []
# Dedupe by LinkedIn URL so we don't pay twice for the same person.
seen: set[str] = set()
dedup: list[dict] = []
for c in contacts:
li = c.get("linkedin_url", "") or ""
if not li or li in seen:
continue
if not c.get("full_name"):
continue
seen.add(li)
dedup.append(c)
# Build the Phase 3 input CSV.
input_csv = out_csv.replace(".csv", "_input.csv") if out_csv.endswith(".csv") else out_csv + "_input"
email_input_rows = []
for c in dedup:
name_parts = c["full_name"].split()
email_input_rows.append({
"first_name": name_parts[0] if name_parts else "",
"last_name": name_parts[-1] if len(name_parts) >= 2 else "",
"linkedin_url": c["linkedin_url"],
"company": c.get("company", ""),
"domain": c.get("domain", ""),
"title": c.get("title", ""),
})
_write_csv(
input_csv, email_input_rows,
["first_name", "last_name", "linkedin_url", "company", "domain", "title"],
)
spec = json.dumps({
"alias": "em",
"tool": "name-and-domain-to-email-waterfall",
"payload": {
"linkedin_url": "{{linkedin_url}}",
"first_name": "{{first_name}}",
"last_name": "{{last_name}}",
"domain": "{{domain}}",
},
})
_run_deepline_enrich(input_csv, out_csv, [spec])
rows = _read_csv(out_csv)
results_by_li: dict[str, tuple[str, str]] = {}
for r in rows:
em_col = r.get("em", "") or ""
parsed = _parse_json_field(em_col)
email = ""
if isinstance(parsed, str):
email = parsed.strip()
elif isinstance(parsed, dict):
if isinstance(parsed.get("email"), str):
email = parsed["email"].strip()
elif isinstance(parsed.get("result"), str):
email = parsed["result"].strip()
elif isinstance(parsed.get("result"), dict) and isinstance(
parsed["result"].get("email"),
str,
):
email = parsed["result"]["email"].strip()
li = r.get("linkedin_url", "")
domain = (r.get("domain", "") or "").lower()
status = "not_found"
out_email = ""
if email:
em_domain = email.split("@")[-1].lower() if "@" in email else ""
em_apex = extract_apex(em_domain) if em_domain else ""
cand_apex = extract_apex(domain)
if em_apex and cand_apex and em_apex == cand_apex:
status = "corporate_validated"
out_email = email
else:
status = "apex_mismatch"
out_email = ""
results_by_li[li] = (out_email, status)
enriched: list[dict] = []
for c in dedup:
out_email, status = results_by_li.get(c["linkedin_url"], ("", "not_found"))
enriched.append({
**c,
"email": out_email,
"email_source": status,
})
return enriched
# ----------------------------------------------------------------------
# Main entry point
# ----------------------------------------------------------------------
def _main() -> int:
parser = argparse.ArgumentParser(
description="Find prospect companies (always) and optionally their contacts + emails. "
"Companies-only mode is FREE; contact discovery costs additional Deepline credits.",
)
parser.add_argument("--input", required=True,
help="CSV of prospects (columns: domain, name, [score], [niche])")
parser.add_argument("--output", required=True,
help="Final CSV. In --no-contacts mode this is the top-N company list. "
"In --contacts mode it's one row per discovered contact.")
parser.add_argument(
"--roles",
default="",
help="Comma-separated role strings to look for (REQUIRED in --contacts mode). "
"Pass the buyer-persona job titles surfaced in Step 0/0.5 — e.g. for a "
"creative-ops tool: 'Creative Director,Brand Manager,Content Operations Lead'; "
"for an AR automation tool: 'AR Manager,Accounts Receivable Specialist,Controller'. "
"Don't reuse last run's roles for a different vertical.",
)
parser.add_argument("--seniority", default="Senior,Director,VP,Manager")
parser.add_argument("--top", type=int, default=10,
help="Only process the top N prospects by score (if present)")
parser.add_argument("--workdir", default="",
help="Directory for intermediate files (default: next to --output)")
# --contacts / --no-contacts toggle. Default is --no-contacts because contact
# discovery costs additional Deepline credits and the user should always have
# to opt in. The SKILL.md flow is: ship companies first, then ask for credit
# approval before turning on contacts.
contacts_group = parser.add_mutually_exclusive_group()
contacts_group.add_argument(
"--contacts", dest="contacts", action="store_true",
help="Run the 3-phase contact discovery chain (waterfall + Exa fallback + "
"email waterfall). Costs extra Deepline credits — get user approval first.",
)
contacts_group.add_argument(
"--no-contacts", dest="contacts", action="store_false",
help="Companies-only mode (default). Output is just the top-N company list with "
"score and niche; no contacts, no extra credit spend.",
)
parser.set_defaults(contacts=False)
args = parser.parse_args()
roles = [r.strip() for r in args.roles.split(",") if r.strip()]
seniority = [s.strip() for s in args.seniority.split(",") if s.strip()]
if args.contacts and not roles:
parser.error(
"--roles is required in --contacts mode. Pass the buyer-persona job "
"titles for THIS run's vertical (from Step 0/0.5 ecosystem discovery). "
"Example: --roles 'Design Engineer,Mechanical Engineer,DfAM Engineer'"
)
# Slice to top N if the input has a score column.
raw = _read_csv(args.input)
has_score = raw and "score" in raw[0]
if has_score:
def _score(r):
try: return int(r.get("score", "0") or 0)
except Exception: return 0
raw = sorted(raw, key=_score, reverse=True)[:args.top]
else:
raw = raw[:args.top]
workdir = args.workdir or os.path.dirname(os.path.abspath(args.output))
os.makedirs(workdir, exist_ok=True)
# ----------------------------------------------------------------
# --no-contacts mode: just write the top-N company list and exit.
# ----------------------------------------------------------------
if not args.contacts:
company_fields = ["domain", "name", "score", "niche"]
company_rows = [
{f: r.get(f, "") for f in company_fields}
for r in raw
]
_write_csv(args.output, company_rows, company_fields)
print(f"[find_contacts] Wrote top {len(company_rows)} companies to {args.output} "
f"(--no-contacts mode, no extra credits spent)", file=sys.stderr)
print(f"[find_contacts] Re-run with --contacts to add contact discovery + emails.",
file=sys.stderr)
return 0
# ----------------------------------------------------------------
# --contacts mode: stage prospects and run the 3-phase chain.
# ----------------------------------------------------------------
top_csv = os.path.join(workdir, "_top.csv")
fieldnames = list(raw[0].keys()) if raw else ["domain", "name"]
_write_csv(top_csv, raw, fieldnames)
# ---- Phase 1 ----
phase1_out = os.path.join(workdir, "_phase1_waterfall.csv")
phase1_contacts = phase1_waterfall(top_csv, phase1_out, roles=roles, seniority=seniority)
covered = {c["domain"] for c in phase1_contacts if c.get("full_name") and c.get("linkedin_url")}
print(f"[find_contacts] Phase 1 found {len(phase1_contacts)} contacts "
f"covering {len(covered)}/{len(raw)} companies", file=sys.stderr)
# ---- Phase 2 ----
phase2_out = os.path.join(workdir, "_phase2_exa_people.csv")
phase2_contacts = phase2_exa_people(top_csv, phase2_out, roles=roles,
already_covered_domains=covered)
print(f"[find_contacts] Phase 2 found {len(phase2_contacts)} additional contacts",
file=sys.stderr)
# ---- Phase 3 ----
all_contacts = phase1_contacts + phase2_contacts
phase3_out = os.path.join(workdir, "_phase3_emails.csv")
with_emails = phase3_emails(all_contacts, phase3_out)
# Merge score/niche back onto the final output if the input had them.
score_by_domain = {r["domain"]: r.get("score", "") for r in raw}
niche_by_domain = {r["domain"]: r.get("niche", "") for r in raw}
for c in with_emails:
c["score"] = score_by_domain.get(c["domain"], "")
c["niche"] = niche_by_domain.get(c["domain"], "")
# Final write.
fields = ["company", "domain", "score", "niche", "full_name", "title",
"linkedin_url", "email", "email_source", "discovery_phase"]
_write_csv(args.output, with_emails, fields)
valid = sum(1 for c in with_emails if c["email_source"] == "corporate_validated")
print(f"[find_contacts] Wrote {len(with_emails)} contacts to {args.output}",
file=sys.stderr)
print(f"[find_contacts] {valid} have corporate-validated emails", file=sys.stderr)
return 0
if __name__ == "__main__":
sys.exit(_main())
skill-metadata.json›
{
"documents": {
"SKILL.md": {
"kind": "entrypoint",
"title": "Niche Signal Discovery",
"tags": ["signals", "icp"],
"providers": []
},
"references/keyword-catalog.md": {
"kind": "guide",
"title": "Keyword Catalog",
"tags": ["signals"],
"providers": []
},
"references/report-template.md": {
"kind": "guide",
"title": "Report Template",
"tags": ["reporting"],
"providers": []
},
"references/signal-interpretation.md": {
"kind": "guide",
"title": "Signal Interpretation",
"tags": ["signals"],
"providers": []
},
"references/dedupe.md": {
"kind": "guide",
"title": "Dedupe Against Existing List",
"tags": ["prospecting", "dedupe"],
"providers": []
},
"references/quality-gate.md": {
"kind": "guide",
"title": "Quality Gate",
"tags": ["enrichment", "qa"],
"providers": []
},
"references/pitfalls.md": {
"kind": "guide",
"title": "Common Pitfalls",
"tags": ["signals", "troubleshooting"],
"providers": []
},
"references/proven-signals.md": {
"kind": "guide",
"title": "Proven Signal Patterns",
"tags": ["signals", "scoring"],
"providers": []
},
"references/step-7-prospects.md": {
"kind": "guide",
"title": "Step 7 — Top 10 Prospects",
"tags": ["prospecting", "contacts"],
"providers": []
},
"references/scoring-pitfalls.md": {
"kind": "guide",
"title": "Scoring Pitfalls — Confirmation-Biased Fields",
"tags": ["signals", "scoring"],
"providers": []
}
}
}
SKILL.md›
---
name: niche-signal-discovery
disable-model-invocation: false
description: 'Discover niche first-party signals that differentiate Closed Won vs Closed Lost accounts for ICP analysis. Use when the user provides won/lost customer domain lists and wants differential signals (website content, job listings, tech stack, maturity markers) to build account scoring models and prospecting criteria. Triggers: ICP analysis, niche signals, won vs lost analysis, differential signals, signal discovery, ICP signal report, account scoring signals, lead scoring, first-party signals, buyer signals. Before reading this file, first read deepline-gtm to understand the Deepline CLI tool and how to use it. Then read this file for guidance on the task.'
---
# Niche Signal Discovery
## Quick Start
```bash
npm install -g deepline
# Fallback for secure sandboxes: mkdir -p "$HOME/.local" && npm config set prefix "$HOME/.local" && export PATH="$HOME/.local/bin:$PATH" && npm install -g deepline --registry https://code.deepline.com/api/v2/npm/
deepline auth register --wait auto
deepline auth wait --timeout 120 # completes Cowork/browser approval; no-op if already connected
deepline auth status
deepline -h
```
## CLI resolution
Run `deepline` when it is available. If the shell reports that command is missing, use `<workspace-root>/.deepline/runtime/bin/deepline` (or the npm-created `.cmd` shim on Windows). If neither exists, follow `https://code.deepline.com/INSTALL.md` to set up Deepline.
Discover differential signals between Closed Won and Closed Lost accounts by extracting multi-page website content and job listings, then computing Laplace-smoothed lift scores to identify what distinguishes buyers from non-buyers.
## Prerequisites
- **Deepline CLI** — All enrichment runs through `deepline enrich`; route through prebuilt plays and customer-configured provider connections rather than hardcoding provider-specific prospecting tools.
- **Python 3** stdlib only — no pip dependencies for any shipped script.
- **Credits** - paid web extraction plus CrustData job search. Run a small sample or `deepline tools describe crustdata_v2_job_search --json` for current Deepline-facing pricing before scaling. Step 7 contact discovery is additional. **Always get user approval before paid steps.**
## Deepline-First Principle
Use `deepline enrich` for all enrichment and `deepline tools execute` for one-offs. Inspect CSV shape and samples with `deepline csv show`; inspect run state with the run/play URL or `deepline runs get` when a run id is available. Reruns are idempotent. Refer to `deepline-gtm` for command patterns and provider playbooks.
## Input requirements
- Won and lost customer domain lists (≥20 won + ≥10 lost for statistical significance)
- **Lookalikes can supplement Won** if Closed Won < 15. Add a Dataset Caveat to the report.
- **Target company context** from Step 0 — what they sell, who they sell to, key personas.
## Pipeline
```
0. Discover target company (what they sell, who they sell to)
0.5. Discover ecosystem (competitors, tech stack, buyer personas)
1. Prepare input CSV (deduplicate within won/lost groups)
1.0.5 Build "do not re-contact" index from user's existing list (scripts/dedupe_utils.py)
1.5. Generate vertical-specific configs (keywords, tools, job roles)
2. Multi-page website + job extraction (deepline enrich)
3. Quality gate — verify file completeness + coverage (>80%)
3.5. Review configs against enriched data
4. Differential analysis (scripts/analyze_signals.py)
5. Generate report — every top signal must include cited evidence
6. Signal interpretation review
7. Top 10 net-new prospects [REQUIRED] + contacts/emails [optional, costs credits]
```
**Step 7 is required.** A signal report without 10 actionable companies forces the reader to do their own prospecting pass — exactly the expensive thing they wanted to skip. Contacts/emails are optional only because they cost extra credits; always offer them.
## Signal reliability hierarchy
Highest → lowest confidence:
1. **Job listings** — active budget + acknowledged pain. Highest-intent.
2. **Analyst validation** (Gartner/Forrester) — typically 4-7x lift, rare in lost.
3. **Compliance infrastructure** (SOC2/GDPR/ISO) — procurement maturity.
4. **Buyer pain language** on careers/blog — operational awareness.
5. **Tech stack tools** (niche SaaS) — infrastructure readiness.
6. **Website product/marketing content** — variable; can be buyer OR competitor.
**When website signals fail:** For B2B back-office tools (AR, billing, compliance), buyers don't publish their pain on marketing pages. Prioritize jobs + tech stack + firmographics for these verticals.
## What NOT to use for scoring
CRM fields populated by AE activity — catalyst note count, OCR-derived counts (`number_of_champions_c`, `number_of_decision_makers_c`), MEDDPICC picklists, any "did the AE do X on this opp" field — correlate with win-rate as **engagement artifacts, not causal signals**. They get filled in _after_ the AE decides an opp is worth working. **Never use them as scoring inputs.** On one real run, catalyst notes showed "109x lift" — almost made the TL;DR before we caught the direction of causality.
Rule of thumb: every scoring input must be observable BEFORE the AE touches the account. Read `references/scoring-pitfalls.md` for the full list and the "safer alternative read" for loss-reason data.
## Step 0: Target company discovery
**Do this FIRST.** The entire pipeline (exa query, keywords, tech stack, job roles) adapts based on this discovery; skipping it produces generic/irrelevant signals.
```bash
deeplineagent: "Research {{company-domain}}. Summarize what the company sells, who they sell to, what makes them different, and any example customers."
```
Document: (1) product category, (2) target buyer persona, (3) key differentiation, (4) example customers.
## Step 0.5: Ecosystem discovery
Three parallel `deeplineagent` queries:
- **Competitors** — `"{product category} software alternatives competitors"` → 3-5 names
- **Tech stack** — `"{buyer persona} software stack"` → 10-15 tools by category
- **Job roles** — `"{buyer persona} job titles"` → 10-15 title variations
These feed Step 1.5 config generation.
## Step 1: Prepare input CSV
```csv
domain,status
customer1.com,won
non-customer1.com,lost
```
**Deduplicate within the input.** If a domain appears in BOTH won and lost (same company, multiple deals), Deepline only fetches job listings once — silently undercounting `won_with_jobs`. Remove ALL rows for cross-group domains:
```python
from collections import Counter
counts = Counter(r['domain'] for r in rows)
duplicate_domains = {d for d, c in counts.items() if c > 1}
# Drop every row in duplicate_domains, not just one copy.
```
## Step 1.0.5: Build "do not re-contact" index
Before any prospects ship in Step 7, dedupe candidates against whatever "already known" list the user provides — customers, CRM export, past outbound, a previous run's output. **Always ask explicitly**; if the user has no list, note it as a caveat in the final report rather than silently skipping.
**Order: apex domain first, fuzzy company name as fallback.** Use the shipped helper — it handles public-suffix multi-label TLDs (`co.uk`, `co.jp`, `com.au`) and corporate-suffix stripping:
```bash
python3 scripts/dedupe_utils.py --selftest # one-time sanity check
python3 scripts/dedupe_utils.py \
--existing customers.csv --candidates prospects_raw.csv \
--out-actionable prospects_actionable.csv --out-matched already_known.csv
```
Don't silently drop CRM matches — **categorize** them: Net-new / Account-only / Re-engage / Active-open / Current-customer.
**Read `references/dedupe.md`** for the failure modes (raw-string match missing `amsynergy.nikon.com → nikon.com` cost 24 of 50 prospects in one run), category definitions, and library usage.
## Step 1.5: Generate vertical-specific configs
Create three JSON files in `output/{{company}}/`:
```
{{company}}-keywords.json # product category, pain language, competitor names, maturity terms
{{company}}-tools.json # niche SaaS tools by category
{{company}}-job-roles.json # buyer persona job titles
```
**Read `references/keyword-catalog.md`** for the JSON schema, generation patterns, and multi-vertical examples (creative ops, AR automation, sales engagement, developer tools).
**Validation:** Do the configs match the target's vertical and buyer persona? If not, refine based on Step 0/0.5 findings.
## Step 2: Deepline enrichment
**Never scrape just the homepage.** Use Serper to discover relevant pages, Firecrawl to extract content.
**Step 2a - Discover pages with Serper (0.02 credits/company):**
```bash
deepline enrich \
--input output/{{company}}-icp-input.csv \
--output output/{{company}}-discovered.csv \
--name niche-pages-discovery \
--with '{"alias":"pages","tool":"serper_google_search","payload":{"query":"site:{{domain}} product OR features OR integrations OR customers OR security OR pricing OR careers OR about"}}' \
--json
```
Adapt the query by vertical: add `compliance OR audit` for back-office, `documentation OR api` for developer tools, `portfolio OR workflow` for creative tools.
**Step 2b - Scrape top 5 pages with Firecrawl (0.05 credits/company):**
Extract URLs from Serper results, then scrape each:
```bash
deepline enrich \
--input output/{{company}}-urls.csv \
--output output/{{company}}-scraped.csv \
--name niche-page-scrape \
--with '{"alias":"content","tool":"firecrawl_scrape","payload":{"url":"{{url}}"}}' --json
```
Aggregate scraped pages back into one row per domain, formatted as `{"data":{"results":[{url, title, text}]}}` for the analysis script.
**Step 2c - Job listings with Crustdata:**
```bash
deepline enrich \
--input output/{{company}}-aggregated.csv \
--output output/{{company}}-enriched.csv \
--name niche-job-listings \
--with '{"alias":"jobs","tool":"crustdata_v2_job_search","payload":{"filters":[{"filter_type":"company.basic_info.primary_domain","type":"=","value":"{{domain}}"}],"limit":100}}' --json
```
Estimate the paid-step total from current tool pricing before scaling. Get user approval first.
## Step 3: Quality gate
`deepline enrich` returns to terminal **before** OS buffers fully flush. Running the analysis script immediately can read a partially-written file and produce `won_with_jobs: 0` even when data is fine. Always verify:
```bash
INPUT_ROWS=$(wc -l < output/{{company}}-icp-input.csv)
OUTPUT_ROWS=$(wc -l < output/{{company}}-enriched.csv)
echo "Input: $INPUT_ROWS, Output: $OUTPUT_ROWS" # should match
```
Then spot-check that won rows have job data, that website coverage is >80%, and that average content depth is 6-8 pages / 12-20K chars per company.
**Read `references/quality-gate.md`** for the full verification script, the buffer-flush retry pattern, and the "auto-extracted domain validation" check that has caught up to **53% false-positive rates** in CRM-exported customer lists.
## Step 3.5: Review configs against enriched data
Inspect the enriched CSV before analysis:
```bash
deepline csv show --csv output/{{company}}-enriched.csv --summary
deepline csv show --csv output/{{company}}-enriched.csv --rows 0:5
```
**Red flags:**
- Keyword in <10% of enriched companies → too niche, broaden
- Keyword in >90% → too generic, refine
- Product-category keywords appear frequently in Won → wrong product category, those companies are competitors not buyers
- Job roles missing from actual listings → wrong buyer persona
Fix and regenerate configs if needed.
## Step 4: Differential analysis
```bash
python3 scripts/analyze_signals.py \
--input output/{{company}}-enriched.csv \
--keywords output/{{company}}-keywords.json \
--tools output/{{company}}-tools.json \
--job-roles output/{{company}}-job-roles.json \
--output output/{{company}}-analysis.json
```
The script computes substring-match presence, Laplace-smoothed lift, source breakdown (website/jobs/both), tech-stack mentions, job-role prevalence, anti-fit signals, and **per-keyword evidence quotes** (±40 chars with URLs) — the evidence array is what Step 5 renders.
## Step 5: Report generation
**Read `references/report-template.md`** for the full report structure (Quick Reference Dashboard at the top, then detail sections), the signal-strength visual scale, prospecting-link format, and all quality rules. Critical rules in brief:
- Raw counts always (`15% (6)`, not just `15%`); sample sizes in headers (`Won (n=37)`)
- Bold only signals with lift > 2x AND count ≥ 3 companies
- Flag n=1 signals — they're statistically meaningless
- **Source evidence is mandatory for every top signal** (lift ≥ 1.5 AND won ≥ 3) — 3-5 cited quotes per signal with source type, company, page/job title, ±40-char quote, and live URL. The analysis script outputs this; render it, don't decide whether to. Signals without 3+ citations get demoted and flagged `*(insufficient evidence)*`.
- Annotate each evidence quote with ✅ (clear buyer signal) or ⚠️ (vendor-adjacent — the company sells something similar, so the keyword on their product page isn't a buyer signal)
- Tier 1 cheatsheet point values must match the Section 6 scoring model — cross-check both before shipping
## Step 6: Signal interpretation
**Read `references/signal-interpretation.md`** before writing interpretation columns. Key rules:
- Website content mentioning what the target sells = competitor signal (not buyer)
- Job listings = highest-intent buyer signal
- Same keyword means different things on product page vs careers page vs blog
- Tech stack tools need context — do they create or solve the target's problem?
## Step 7: Top 10 net-new prospects (required)
**10 companies are required for every run; contacts + emails are optional** (additional Deepline credits). Always offer contact discovery; only run it if the user approves the spend.
```bash
# Companies only — no extra credits beyond Step 2 enrichment:
python3 scripts/find_contacts.py --input prospects_actionable.csv --output top10.csv --top 10 --no-contacts
# Companies + contacts + emails — asks for credit approval.
# --roles is REQUIRED in --contacts mode and must be the buyer-persona job
# titles surfaced in YOUR Step 0/0.5 (not a stale list from a different vertical):
python3 scripts/find_contacts.py --input prospects_actionable.csv --output top10.csv --top 10 \
--contacts --roles "<persona job titles from Step 0.5>"
```
When `--contacts` is on, the orchestrator runs a 3-phase chain via Deepline:
1. `company-to-contact` (free, mature companies)
2. **`exa_search_people` fallback for any company Phase 1 missed** — mandatory. On the run that motivated this, Phase 1 returned 0 contacts on all 10 top prospects (small/non-US industrial); Exa found 15 real contacts at 6 of those 10 in the same pass.
3. `name-and-domain-to-email-waterfall` with `linkedin_url` supplied and **apex-domain validation** — providers return stale addresses (`@orbitalatk.com` for someone now at X-Bow, personal Gmails, wrong-company false positives). Mismatched apex → publish "(email not found)", keep the raw value in `raw_email` for auditing.
**Read `references/step-7-prospects.md`** for the required vs. optional output fields, the prospect-card skeleton, the Phase 2 Exa guardrails (title parsing + company-match filter), and the "10 is a ceiling, not a floor" guidance.
## Enrichment data structure
After enrichment, each row has:
- `website` column → JSON: `{"data":{"results":[{text, url, title}]}}` (aggregated from Firecrawl scrapes)
- `jobs` column → JSON: `{"result":{"listings":[{title, description, url}]}}` (Crustdata format - note `result` not `data`, `title` not `job_title`)
`scripts/analyze_signals.py` auto-detects `__dl_full_result__` columns; override with `--website-col N --jobs-col N` for other column names.
## Common pitfalls (top 6 — full list in references/pitfalls.md)
1. **Skipping target discovery (Step 0)** → generic/irrelevant configs.
2. **Homepage-only scraping** → misses pricing, integrations, security, careers.
3. **Generic tech stack** ("AWS", "GitHub", "Slack" appear on most B2B sites) → search for niche SaaS specific to the buyer persona.
4. **Trusting n=1 signals** → require 3+ companies for Tier 1 scoring; flag single-company signals with a verification note.
5. **Raw-string dedupe missing parent domains** — `amsynergy.nikon.com ≠ nikon.com` for naive comparison. Always use `extract_apex()`. **24 of 50 "net-new" prospects in one real run were already in the CRM** as parent-domain entries the raw-string dedupe missed.
6. **Trusting confirmation-biased CRM fields** (catalyst notes, OCR counts, MEDDPICC) as signals — they're downstream of AE engagement, not causal. Read the "What NOT to use for scoring" section above.
**Read `references/pitfalls.md`** for the full 18-item list including substring false positives, vendor-vs-buyer signal context, back-office-tool interpretation, and shipping-without-prospects.
## Proven signal patterns
**Read `references/proven-signals.md`** for typical lift ranges across verticals (analyst validation 4.5-6.5x, hiring signals 3.8-5.5x, compliance infra 2.1-6.5x, etc.), high-confidence anti-fit patterns (consumer signals 0.2x, retention/churn 0.2-0.4x), and a starter 0-100 scoring model with three tiers (Core Fit / Buying Intent / Infrastructure Readiness).
## References
- **`references/keyword-catalog.md`** — JSON schema + multi-vertical examples for Step 1.5 config generation
- **`references/dedupe.md`** — Step 1.0.5 dedupe failure modes, categorization rules, library usage
- **`references/quality-gate.md`** — Step 3 verification scripts, buffer-flush retry pattern, auto-extracted-domain validation
- **`references/report-template.md`** — Step 5 full report structure, signal-strength scale, prospecting-link format, all quality rules
- **`references/signal-interpretation.md`** — Step 6 buyer-vs-seller-vs-competitor rules
- **`references/step-7-prospects.md`** — Step 7 prospect-card skeleton, Exa guardrails, Phase 3 apex validation
- **`references/scoring-pitfalls.md`** — Confirmation-biased CRM fields to exclude from scoring
- **`references/pitfalls.md`** — Full 18-item pitfalls list
- **`references/proven-signals.md`** — Typical lift ranges + scoring model guidance
- **`scripts/analyze_signals.py`** — Step 4 differential analysis. Auto-detects columns.
- **`scripts/dedupe_utils.py`** — Step 1.0.5 + Step 7 email validation. `extract_apex()`, `norm_name()`, `match_against_existing()`. Stdlib only. `--selftest` flag for one-time install verification.
- **`scripts/find_contacts.py`** — Step 7 orchestrator. `--contacts` / `--no-contacts` toggle, 3-phase Deepline chain.
## Changelog
- **2026-04-13** — Switched Step 2 from exa_search (~5 credits) to Serper + Firecrawl (~0.07 credits) for website content. Fixed analyze_signals.py to handle Crustdata's `{"result":{"listings":[]}}` wrapper. Verified E2E on 15 companies.
- **2026-04-07** — Added Step 1.0.5 (dedupe with apex helper), Step 7 (top 10 prospects required, contacts optional via `--contacts`/`--no-contacts`), `references/scoring-pitfalls.md` warning about confirmation-biased CRM fields, mandatory citation rule. Shipped `scripts/dedupe_utils.py` + `scripts/find_contacts.py`. Aggressively trimmed inline detail to references — moved Step 3 quality gate, Step 5 quality rules, Common Pitfalls (items 7-15), and Proven Signal Patterns into `references/`. SKILL.md went from 650 to ~250 lines via progressive disclosure.