Automated Data Enrichment: Beyond Clearbit and Apollo

// B2B data decay rates from industry benchmarks
const dataDecayRates = {
  jobTitleChange: 0.658,      // 65.8% of contacts change roles annually
  emailChurn: 0.373,          // 28-37.3% of emails become invalid
  phoneNumberTurnover: 0.429, // 42.9% of direct dials go stale
  companyDataDecay: 0.225,    // 22.5% of firmographic data changes
};
 
// For a database of 10,000 contacts:
// - 6,580 will have different job titles within 12 months
// - 3,730 email addresses will bounce
// - 4,290 phone numbers will be wrong

// n8n workflow structure for waterfall enrichment
const waterfallWorkflow = {
  trigger: "Webhook - New lead from CRM",
  nodes: [
    {
      name: "Check Cache",
      type: "Postgres",
      operation: "SELECT * FROM enrichment_cache WHERE domain = $domain AND updated_at > NOW() - INTERVAL '7 days'",
    },
    {
      name: "Provider A - Clearbit",
      type: "HTTP Request",
      condition: "Cache miss OR missing required fields",
      onError: "Continue to next provider",
    },
    {
      name: "Evaluate Completeness",
      type: "Code",
      operation: "Check if minimum fields populated",
    },
    {
      name: "Provider B - Apollo",
      type: "HTTP Request",
      condition: "Completeness < 80%",
      onError: "Continue to next provider",
    },
    {
      name: "Provider C - ZoomInfo",
      type: "HTTP Request",
      condition: "Completeness < 80% AND high_value_lead = true",
      // Only burn expensive credits on high-value targets
    },
    {
      name: "Merge & Resolve Conflicts",
      type: "Code",
      operation: "Apply field-level merge rules",
    },
    {
      name: "Update CRM",
      type: "HubSpot/Salesforce",
      operation: "Update contact/company record",
    },
    {
      name: "Update Cache",
      type: "Postgres",
      operation: "INSERT INTO enrichment_cache",
    },
  ],
};

// Field-level conflict resolution
interface EnrichmentResult {
  source: string;
  data: Record<string, unknown>;
  timestamp: Date;
  confidence: number;
}
 
interface MergeRule {
  field: string;
  strategy: "most_recent" | "highest_confidence" | "prefer_source" | "consensus";
  preferredSource?: string;
}
 
const mergeRules: MergeRule[] = [
  { field: "employee_count", strategy: "most_recent" },
  { field: "industry", strategy: "consensus" },
  { field: "annual_revenue", strategy: "highest_confidence" },
  { field: "job_title", strategy: "prefer_source", preferredSource: "linkedin" },
  { field: "email", strategy: "highest_confidence" },
  { field: "phone", strategy: "prefer_source", preferredSource: "zoominfo" },
];
 
function mergeEnrichmentResults(
  results: EnrichmentResult[],
  rules: MergeRule[]
): Record<string, unknown> {
  const merged: Record<string, unknown> = {};
 
  for (const rule of rules) {
    const values = results
      .filter((r) => r.data[rule.field] !== undefined)
      .map((r) => ({
        value: r.data[rule.field],
        source: r.source,
        timestamp: r.timestamp,
        confidence: r.confidence,
      }));
 
    if (values.length === 0) continue;
 
    switch (rule.strategy) {
      case "most_recent":
        merged[rule.field] = values.sort(
          (a, b) => b.timestamp.getTime() - a.timestamp.getTime()
        )[0].value;
        break;
 
      case "highest_confidence":
        merged[rule.field] = values.sort(
          (a, b) => b.confidence - a.confidence
        )[0].value;
        break;
 
      case "prefer_source":
        const preferred = values.find((v) => v.source === rule.preferredSource);
        merged[rule.field] = preferred?.value ?? values[0].value;
        break;
 
      case "consensus":
        // Use the value that appears most frequently
        const counts = new Map<unknown, number>();
        values.forEach((v) => counts.set(v.value, (counts.get(v.value) || 0) + 1));
        merged[rule.field] = [...counts.entries()].sort((a, b) => b[1] - a[1])[0][0];
        break;
    }
  }
 
  return merged;
}

# Python retry pattern for enrichment APIs
import asyncio
from typing import Optional, Dict, Any
from dataclasses import dataclass
from enum import Enum
 
class RetryStrategy(Enum):
    EXPONENTIAL = "exponential"
    LINEAR = "linear"
    NONE = "none"
 
@dataclass
class ProviderConfig:
    name: str
    base_url: str
    api_key: str
    rate_limit: int  # requests per minute
    retry_strategy: RetryStrategy
    max_retries: int
    timeout_seconds: int
 
async def call_provider_with_retry(
    config: ProviderConfig,
    payload: Dict[str, Any]
) -> Optional[Dict[str, Any]]:
    """Call enrichment provider with configurable retry logic."""
 
    for attempt in range(config.max_retries + 1):
        try:
            async with aiohttp.ClientSession() as session:
                async with session.post(
                    config.base_url,
                    json=payload,
                    headers={"Authorization": f"Bearer {config.api_key}"},
                    timeout=aiohttp.ClientTimeout(total=config.timeout_seconds)
                ) as response:
                    if response.status == 200:
                        return await response.json()
 
                    if response.status == 429:  # Rate limited
                        retry_after = int(response.headers.get("Retry-After", 60))
                        await asyncio.sleep(retry_after)
                        continue
 
                    if response.status >= 500:  # Server error, retry
                        raise Exception(f"Server error: {response.status}")
 
                    # Client error (4xx), don't retry
                    return None
 
        except asyncio.TimeoutError:
            pass  # Will retry
        except Exception as e:
            if attempt == config.max_retries:
                log_enrichment_failure(config.name, payload, str(e))
                return None
 
        # Calculate retry delay
        if config.retry_strategy == RetryStrategy.EXPONENTIAL:
            delay = min(2 ** attempt, 60)  # Cap at 60 seconds
        elif config.retry_strategy == RetryStrategy.LINEAR:
            delay = 5 * (attempt + 1)
        else:
            break
 
        await asyncio.sleep(delay)
 
    return None

# LLM-based entity extraction for enrichment
from openai import OpenAI
from pydantic import BaseModel
from typing import List, Optional
 
class CompanyProfile(BaseModel):
    company_name: str
    industry: Optional[str]
    employee_range: Optional[str]
    technologies: List[str]
    funding_stage: Optional[str]
    headquarters: Optional[str]
 
def extract_company_data(website_text: str) -> CompanyProfile:
    """Extract structured company data from website content."""
 
    client = OpenAI()
 
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": """Extract company information from the provided website text.
                Return structured data with these fields:
                - company_name: Official company name
                - industry: Primary industry (use standard SIC categories)
                - employee_range: Estimated employee count (1-10, 11-50, 51-200, 201-500, 501-1000, 1000+)
                - technologies: List of technologies/tools mentioned
                - funding_stage: If mentioned (Seed, Series A/B/C, Public, Bootstrapped)
                - headquarters: City, Country if mentioned
 
                Only include fields you can confidently extract. Use null for uncertain fields."""
            },
            {
                "role": "user",
                "content": website_text[:8000]  # Truncate to fit context
            }
        ],
        response_format={"type": "json_object"}
    )
 
    data = json.loads(response.choices[0].message.content)
    return CompanyProfile(**data)
 
 
async def enrich_from_website(domain: str) -> Optional[CompanyProfile]:
    """Fetch and extract company data from their website."""
 
    # Fetch website content
    website_text = await fetch_website_text(domain)
    if not website_text:
        return None
 
    # Extract structured data
    try:
        profile = extract_company_data(website_text)
 
        # Log extraction for quality monitoring
        await log_llm_extraction(domain, profile)
 
        return profile
    except Exception as e:
        log_extraction_error(domain, str(e))
        return None

# Validation layer for LLM-extracted data
def validate_extraction(
    extracted: CompanyProfile,
    known_data: Dict[str, Any]
) -> Dict[str, float]:
    """Score confidence for each extracted field."""
 
    confidence_scores = {}
 
    # Company name: Check against domain
    if extracted.company_name:
        domain_words = known_data.get("domain", "").replace(".com", "").split("-")
        name_words = extracted.company_name.lower().split()
        overlap = len(set(domain_words) & set(name_words))
        confidence_scores["company_name"] = min(overlap / max(len(domain_words), 1), 1.0)
 
    # Industry: Validate against known SIC codes
    if extracted.industry:
        if extracted.industry in VALID_SIC_INDUSTRIES:
            confidence_scores["industry"] = 0.8
        else:
            confidence_scores["industry"] = 0.3
 
    # Employee range: Cross-reference with LinkedIn if available
    if extracted.employee_range and known_data.get("linkedin_employees"):
        linkedin_range = categorize_employee_count(known_data["linkedin_employees"])
        if extracted.employee_range == linkedin_range:
            confidence_scores["employee_range"] = 0.95
        elif abs(RANGE_ORDER.index(extracted.employee_range) -
                 RANGE_ORDER.index(linkedin_range)) <= 1:
            confidence_scores["employee_range"] = 0.7
        else:
            confidence_scores["employee_range"] = 0.3
 
    return confidence_scores

// Tiered enrichment by lead value
interface LeadEnrichmentTier {
  minScore: number;
  maxScore: number;
  providers: string[];
  maxSpend: number;
}
 
const enrichmentTiers: LeadEnrichmentTier[] = [
  {
    minScore: 0,
    maxScore: 30,
    providers: ["clearbit_free_tier"],
    maxSpend: 0,
  },
  {
    minScore: 31,
    maxScore: 60,
    providers: ["clearbit", "apollo"],
    maxSpend: 0.5,
  },
  {
    minScore: 61,
    maxScore: 85,
    providers: ["clearbit", "apollo", "zoominfo"],
    maxSpend: 2.0,
  },
  {
    minScore: 86,
    maxScore: 100,
    providers: ["clearbit", "apollo", "zoominfo", "linkedin", "llm_extraction"],
    maxSpend: 5.0,
  },
];
 
function getEnrichmentTier(leadScore: number): LeadEnrichmentTier {
  return enrichmentTiers.find(
    (tier) => leadScore >= tier.minScore && leadScore <= tier.maxScore
  )!;
}

-- Enrichment cache with field-specific TTL
CREATE TABLE enrichment_cache (
    domain VARCHAR(255) PRIMARY KEY,
    company_name VARCHAR(255),
    company_name_updated_at TIMESTAMP,
    employee_count INTEGER,
    employee_count_updated_at TIMESTAMP,
    industry VARCHAR(100),
    industry_updated_at TIMESTAMP,
    annual_revenue BIGINT,
    annual_revenue_updated_at TIMESTAMP,
    technologies JSONB,
    technologies_updated_at TIMESTAMP
);
 
-- Query with field-specific freshness checks
SELECT
    domain,
    CASE WHEN company_name_updated_at > NOW() - INTERVAL '90 days'
         THEN company_name ELSE NULL END as company_name,
    CASE WHEN employee_count_updated_at > NOW() - INTERVAL '30 days'
         THEN employee_count ELSE NULL END as employee_count,
    CASE WHEN industry_updated_at > NOW() - INTERVAL '180 days'
         THEN industry ELSE NULL END as industry,
    CASE WHEN technologies_updated_at > NOW() - INTERVAL '14 days'
         THEN technologies ELSE NULL END as technologies
FROM enrichment_cache
WHERE domain = $1;
 
-- Company names rarely change: 90 days
-- Employee counts change quarterly: 30 days
-- Industry is stable: 180 days
-- Tech stack changes frequently: 14 days

# Batch enrichment scheduler
from datetime import datetime, time
import asyncio
 
BATCH_WINDOW_START = time(2, 0)  # 2 AM
BATCH_WINDOW_END = time(6, 0)    # 6 AM
 
async def schedule_batch_enrichment(records: List[Dict]) -> None:
    """Queue records for off-peak batch enrichment."""
 
    # Separate high-priority (enrich now) from batch-eligible
    high_priority = [r for r in records if r.get("lead_score", 0) > 80]
    batch_eligible = [r for r in records if r.get("lead_score", 0) <= 80]
 
    # Enrich high-priority immediately
    for record in high_priority:
        await enrich_single_record(record)
 
    # Queue rest for batch window
    await add_to_batch_queue(batch_eligible)
 
async def run_batch_enrichment():
    """Execute batch enrichment during off-peak window."""
 
    current_time = datetime.now().time()
    if not (BATCH_WINDOW_START <= current_time <= BATCH_WINDOW_END):
        return
 
    batch = await get_batch_queue(limit=1000)
 
    # Use batch API endpoints where available
    # Clearbit: /v2/companies/batch
    # Apollo: /v1/people/bulk_match
 
    clearbit_batch = [r for r in batch if needs_clearbit(r)]
    apollo_batch = [r for r in batch if needs_apollo(r)]
 
    await asyncio.gather(
        enrich_clearbit_batch(clearbit_batch),
        enrich_apollo_batch(apollo_batch)
    )

// Compliance checks in enrichment pipeline
interface ComplianceConfig {
  region: "EU" | "US" | "OTHER";
  consentStatus: "explicit" | "legitimate_interest" | "none";
  dataCategories: string[];
}
 
function canEnrich(
  record: LeadRecord,
  compliance: ComplianceConfig
): { allowed: boolean; restrictions: string[] } {
  const restrictions: string[] = [];
 
  // GDPR: EU residents require consent or legitimate interest
  if (compliance.region === "EU") {
    if (compliance.consentStatus === "none") {
      return { allowed: false, restrictions: ["GDPR: No consent basis"] };
    }
 
    // Even with consent, limit to necessary data
    if (compliance.consentStatus === "legitimate_interest") {
      restrictions.push("Limit to business contact data only");
      restrictions.push("No personal social profiles");
    }
  }
 
  // CCPA: California residents can opt out
  if (record.state === "CA" && record.ccpaOptOut) {
    return { allowed: false, restrictions: ["CCPA: User opted out"] };
  }
 
  // Log compliance decision for audit trail
  logComplianceDecision(record.id, compliance, restrictions);
 
  return { allowed: true, restrictions };
}

-- Enrichment quality dashboard metrics
WITH enrichment_stats AS (
    SELECT
        DATE_TRUNC('week', enriched_at) as week,
        COUNT(*) as total_records,
        COUNT(CASE WHEN company_name IS NOT NULL THEN 1 END) as has_company,
        COUNT(CASE WHEN industry IS NOT NULL THEN 1 END) as has_industry,
        COUNT(CASE WHEN employee_count IS NOT NULL THEN 1 END) as has_employees,
        COUNT(CASE WHEN email_verified = true THEN 1 END) as verified_emails,
        AVG(enrichment_cost) as avg_cost_per_record,
        AVG(providers_queried) as avg_providers_used
    FROM enriched_leads
    WHERE enriched_at > NOW() - INTERVAL '90 days'
    GROUP BY DATE_TRUNC('week', enriched_at)
)
SELECT
    week,
    total_records,
    ROUND(100.0 * has_company / total_records, 1) as company_coverage_pct,
    ROUND(100.0 * has_industry / total_records, 1) as industry_coverage_pct,
    ROUND(100.0 * has_employees / total_records, 1) as employee_coverage_pct,
    ROUND(100.0 * verified_emails / total_records, 1) as email_verified_pct,
    ROUND(avg_cost_per_record, 3) as cost_per_record,
    ROUND(avg_providers_used, 1) as providers_per_record
FROM enrichment_stats
ORDER BY week DESC;