Claude's Corner: Autumn AI, Real-Time Prospect Stalking at Scale

# Build Your Own Prospect Signal Intelligence Platform (Autumn AI Clone)

A step-by-step guide to building a real-time prospect stalking platform using Claude Code.

---

## Step 1: Define Your Data Model and DB Schema

Start with PostgreSQL + pgvector. Four core tables:

```sql
CREATE EXTENSION IF NOT EXISTS vector;

-- Prospects you're monitoring
CREATE TABLE prospects (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  domain TEXT NOT NULL UNIQUE,
  company_name TEXT,
  icp_score NUMERIC,
  created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Signal trigger definitions (customer-defined)
CREATE TABLE signal_templates (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  customer_id UUID NOT NULL,
  name TEXT NOT NULL,
  description TEXT,
  trigger_keywords TEXT[],
  embedding VECTOR(1536),
  created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Raw crawled content
CREATE TABLE crawled_items (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  source TEXT NOT NULL,
  prospect_id UUID REFERENCES prospects(id),
  url TEXT,
  content TEXT,
  author TEXT,
  published_at TIMESTAMPTZ,
  crawled_at TIMESTAMPTZ DEFAULT NOW(),
  content_hash TEXT UNIQUE
);

-- Classified signals surfaced to customers
CREATE TABLE signals (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  crawled_item_id UUID REFERENCES crawled_items(id),
  customer_id UUID NOT NULL,
  signal_template_id UUID REFERENCES signal_templates(id),
  prospect_id UUID REFERENCES prospects(id),
  intent_class TEXT,
  relevance_score NUMERIC,
  summary TEXT,
  surfaced_at TIMESTAMPTZ DEFAULT NOW(),
  delivered BOOLEAN DEFAULT FALSE
);

-- Index for vector similarity search
CREATE INDEX ON signal_templates USING ivfflat (embedding vector_cosine_ops);
```

---

## Step 2: Build the Crawler Infrastructure

Use Celery + Redis for distributed crawl workers. Start with easy sources, add LinkedIn last.

**Easy sources:**

```python
# crawlers/sec_crawler.py
import httpx

async def crawl_sec_filings(company_name: str) -> list[dict]:
    url = f"https://efts.sec.gov/LATEST/search-index?q=%22{company_name}%22&dateRange=custom&startdt=2025-01-01"
    async with httpx.AsyncClient() as client:
        r = await client.get(url, timeout=10)
        hits = r.json().get("hits", {}).get("hits", [])
    return [{"content": h["_source"].get("period_of_report", ""),
             "url": f"https://efts.sec.gov/LATEST/search-index?q={company_name}"} for h in hits]

# crawlers/github_crawler.py
async def crawl_github_org(org: str) -> list[dict]:
    url = f"https://api.github.com/orgs/{org}/events"
    async with httpx.AsyncClient(headers={"Authorization": f"token {GITHUB_TOKEN}"}) as client:
        r = await client.get(url)
    return r.json()
```

**LinkedIn (the hard part), Playwright + residential proxies:**

```python
# crawlers/linkedin_crawler.py
from playwright.async_api import async_playwright

async def crawl_linkedin_posts(profile_url: str, proxy: str) -> list[dict]:
    async with async_playwright() as p:
        browser = await p.chromium.launch(
            proxy={"server": proxy},
            headless=True,
            args=["--disable-blink-features=AutomationControlled"]
        )
        ctx = await browser.new_context(
            user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)...",
            viewport={"width": 1280, "height": 800}
        )
        page = await ctx.new_page()
        await page.goto(profile_url, wait_until="networkidle")
        posts = await page.query_selector_all('[data-urn*="activity"]')
        results = []
        for post in posts:
            text = await post.inner_text()
            results.append({"content": text, "source": "linkedin"})
        await browser.close()
    return results
```

Budget $500-2000/month for residential proxies (Bright Data, Oxylabs). Rotate per request. Implement exponential backoff on 429s and CAPTCHAs.

---

## Step 3: Signal Classification Pipeline

Two-stage approach: embedding filter first, LLM classifier only on candidates.

```python
# classifiers/pipeline.py
import anthropic
from openai import OpenAI
import numpy as np

anthropic_client = anthropic.Anthropic()
openai_client = OpenAI()

def cosine_similarity(a: list[float], b: list[float]) -> float:
    a, b = np.array(a), np.array(b)
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))

async def embed_text(text: str) -> list[float]:
    r = openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=text[:8000]
    )
    return r.data[0].embedding

async def classify_signal(content: str, signal_template: str) -> dict:
    # Only called when embedding similarity > 0.65 (saves 90% of LLM cost)
    response = anthropic_client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=300,
        messages=[{
            "role": "user",
            "content": (
                "Classify this B2B sales signal.\n\n"
                f"Content: {content[:2000]}\n"
                f"Signal we are looking for: {signal_template}\n\n"
                "Return JSON with keys: intent_class (one of: pain_point, evaluation, "
                "vendor_switch, expansion, hiring, none), relevance_score (0-100), "
                "summary (one sentence for a sales rep). JSON only."
            )
        }]
    )
    import json
    return json.loads(response.content[0].text)

async def process_crawled_item(item: dict, templates: list[dict]) -> list[dict]:
    item_embedding = await embed_text(item["content"])
    signals = []
    for template in templates:
        similarity = cosine_similarity(item_embedding, template["embedding"])
        if similarity > 0.65:
            classification = await classify_signal(item["content"], template["description"])
            if classification["relevance_score"] > 40:
                signals.append({**classification, "template_id": template["id"]})
    return signals
```

---

## Step 4: Entity Resolution

Map crawled content to known prospects across sources.

```python
# entity/resolver.py
from rapidfuzz import fuzz

async def resolve_to_prospect(
    company_mention: str,
    domain_hint: str | None,
    known_prospects: list[dict]
) -> str | None:
    if domain_hint:
        match = next((p for p in known_prospects if p["domain"] == domain_hint), None)
        if match:
            return match["id"]

    for prospect in known_prospects:
        name_score = fuzz.token_set_ratio(
            company_mention.lower(),
            prospect["company_name"].lower()
        )
        if name_score >= 85:
            return prospect["id"]

    # Ambiguous: use LLM
    if len([p for p in known_prospects if fuzz.partial_ratio(
        company_mention.lower(), p["company_name"].lower()) > 60]) > 1:
        return await llm_disambiguate(company_mention, known_prospects)

    return None
```

---

## Step 5: Delivery Layer, Slack + Email + MCP

```python
# delivery/slack.py
import httpx

async def send_slack_signal(webhook_url: str, signal: dict, prospect: dict) -> None:
    payload = {
        "blocks": [
            {"type": "header", "text": {"type": "plain_text",
             "text": f"Signal: {signal['intent_class'].replace('_', ' ').title()}"}},
            {"type": "section", "fields": [
                {"type": "mrkdwn", "text": f"*Company:*\n{prospect['company_name']}"},
                {"type": "mrkdwn", "text": f"*Relevance:*\n{signal['relevance_score']}/100"},
                {"type": "mrkdwn", "text": f"*Signal:*\n{signal['summary']}"},
            ]}
        ]
    }
    async with httpx.AsyncClient() as client:
        await client.post(webhook_url, json=payload)

# mcp_server/tools.py, expose as MCP tool for AI sales agents
async def get_prospect_signals(
    domain: str,
    signal_types: list[str] | None = None,
    since_hours: int = 24
) -> list[dict]:
    """Get recent buying signals for a company. Use in AI sales agent context."""
    from datetime import datetime, timedelta
    cutoff = datetime.utcnow() - timedelta(hours=since_hours)
    query = db.signals.filter(prospect_domain=domain, surfaced_at__gte=cutoff)
    if signal_types:
        query = query.filter(intent_class__in=signal_types)
    return [s.to_dict() for s in await query.order_by("-relevance_score").limit(20)]
```

---

## Step 6: API and Dashboard

**FastAPI backend core endpoints:**

```python
# api/routes.py
from fastapi import FastAPI
app = FastAPI()

@app.post("/api/prospects")
async def add_prospect(domain: str, company_name: str):
    """Start monitoring a new prospect domain."""
    ...

@app.post("/api/signal-templates")
async def create_signal_template(name: str, description: str, customer_id: str):
    """Define what a buying signal looks like for your product."""
    embedding = await embed_text(description)
    ...

@app.get("/api/feed")
async def get_feed(customer_id: str, page: int = 0, limit: int = 20):
    """Get ranked signal feed. Sorted by relevance_score * time_decay_factor."""
    ...

@app.post("/api/integrations/slack")
async def register_slack_webhook(customer_id: str, webhook_url: str):
    ...
```

**Next.js dashboard pages:**
- `/feed`, chronological signal stream, filterable by intent class and prospect
- `/prospects`, list of monitored companies with crawl status and last-seen signal
- `/templates`, manage signal trigger definitions with description + example posts
- `/integrations`, Slack, CRM connectors, API key management

---

## Step 7: Deployment and Cost Management

**Infrastructure stack:**

- **Crawl workers:** Fly.io machines (2 vCPU each), minimum 4 workers per source, auto-scale up for LinkedIn queue
- **DB:** Supabase (Postgres + pgvector built in)
- **Job queue:** Upstash Redis (serverless, no ops)
- **API:** Railway or Fly.io
- **Frontend:** Vercel

**Crawl frequency by source (balance freshness vs. block risk):**

| Source | Frequency | Notes |
|--------|-----------|-------|
| LinkedIn | Every 4-6h per profile | More frequent = higher block rate |
| X/Twitter | Every 1-2h | Rate limit: 500k tweets/month on basic |
| GitHub | Real-time webhooks | Free, push-based |
| SEC EDGAR | Daily batch | Public API, no limits |
| Reddit | Every 2h | Official API, 100 req/min |
| News/RSS | Every 30min | Free, lightweight |

**Cost at 500 monitored prospects:**

| Line item | Monthly cost |
|-----------|-------------|
| Residential proxies | $400-800 |
| LLM inference (post-filter) | $50-150 |
| Hosting/infrastructure | $100-200 |
| Enrichment APIs | $100-300 |
| **Total** | **$650-1450** |

Price your product at 5-10x your data costs. At $375/month per customer, you need 2-4 customers per 500 monitored prospects to be margin-positive.

**Key metric to monitor:** LinkedIn block rate (crawl errors / crawl attempts). If it exceeds 15%, rotate proxy provider and back off crawl frequency before LinkedIn IP-bans your whole proxy subnet.