Build an SMB Prospect Warehouse

Generate 55K synthetic SMBs, filter by ICP, and enrich top 25 with Claude Code.

Introduction

30 Second Summary

Finding the right clients is the hardest part of running a solo consulting business. What if you had a system that could sift through thousands of local businesses and hand you a ranked shortlist with AI-generated research on each one?

In this project, you will build a local prospect intelligence warehouse powered by Claude. You will generate 55,000 synthetic Austin-area small businesses, filter them down to 25 high-value prospects, and enrich each one with AI-generated research signals.

What You'll Build

A local prospect intelligence warehouse powered by Claude, built with Python and open-source tools.

By the end of this project, you'll have:

A DuckDB warehouse containing 55,000 synthetic Austin SMBs, narrowed to 25 high-value prospects by an Ideal Customer Profile filter.
A Pydantic-validated extraction loop that runs Claude Code headlessly to enrich prospects and retries automatically on validation errors.
A Streamlit dashboard showing funnel metrics, estimated pipeline earnings, and a sortable prospect table backed by a Great Expectations data-quality report.
A "Monday Refresh" script that diffs new extractions against existing records and flags stale prospects (Secret Mission).

Are there any prerequisites?

You need a Claude Max subscription with Claude Code installed and authenticated. You should also be comfortable using a terminal and writing basic Python.

Set Up Your Environment and Verify Claude

Every script in this project depends on five Python packages and a working connection to Claude Code. Without a clean environment, dependency conflicts and authentication errors will block you later.

In this step, you'll create an isolated project folder, install the packages, and confirm that Claude Code can run headless commands from your terminal.

In this step, get ready to:

Create the signals project folder and a Python virtual environment.
Install DuckDB, Great Expectations, Streamlit, Pydantic, and NumPy.
Verify Claude Code is installed and authenticated.

Create the project folder and virtual environment

A virtual environment gives this project its own isolated set of Python packages. Anything you install inside it stays separate from your system Python.

Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.

Move to your Desktop by running this command:

cd ~/Desktop

Create the signals folder and move into it by running these commands:

mkdir signals
cd signals

Confirm you're inside the project folder by running:

pwd

You should see a path ending in /signals (for example, /Users/yourname/Desktop/signals).

Seeing 'File exists' or 'cannot create directory'?

You already have a folder named signals on your Desktop from a previous attempt. Delete it first, or just run cd signals to move into the existing folder.

Help me fix this folder creation error.

Create and activate a Python virtual environment by running these commands:

macOS

python3 -m venv .venv
source .venv/bin/activate

Linux users can use the same commands.

Windows

python -m venv .venv
.venv\Scripts\activate

What is a virtual environment?

A virtual environment is a self-contained Python installation inside your project folder. Every package you install while the environment is active stays isolated here. Your system Python and other projects are unaffected.

Your terminal prompt now shows (.venv) at the beginning, confirming the virtual environment is active.

Confirm your Python version is 3.10 or higher by running:

python --version

I see Python 3.10+

Your Python version is good to go. Move on to installing the project dependencies.

I don't have Python 3.10+

This project requires Python 3.10 or higher. Download the latest version from the official Python website.

Go to python.org/downloads and download the latest Python installer for your operating system.
Run the installer.
On Windows, check the Add python.exe to PATH checkbox before clicking Install Now.
Close and reopen your terminal, then recreate the virtual environment using the platform-specific commands above.
Run python --version again to confirm you see 3.10 or higher.

Install all Python dependencies

The project uses five Python packages. Installing them all in one command keeps the setup fast.

Install the project dependencies by running this command:

pip install duckdb==1.5.1 great_expectations==1.16.1 streamlit==1.56.0 pydantic==2.13.3 numpy

What are these five packages?

duckdb is an in-process analytical database. It stores prospect data in columnar format and exports to Parquet with zero server setup.
great_expectations runs automated data-quality checks on your warehouse and generates an HTML proof report.
streamlit builds the interactive dashboard that displays your prospect funnel and pipeline earnings.
pydantic validates every AI-enriched record against a strict 12-field schema before it enters the warehouse.
numpy generates the right-skewed employee count distribution for the 55,000-row synthetic universe.

The install finishes when you see Successfully installed followed by a list of package names and version numbers.

Seeing an error during install?

If you see No module named pip, your Python installation is missing pip. Reinstall Python from python.org and make sure the pip option is checked during setup.
If you see a network timeout or ConnectionError, check your internet connection and try the command again.
Make sure your virtual environment is active. You should see (.venv) in your terminal prompt.

Help me fix this pip install error.

Verify Claude Code is installed and authenticated

Claude Code is the engine that powers the enrichment loop in this project. It runs headlessly from your terminal, taking a prompt and returning structured JSON with no interactive UI required.

Check if Claude Code is installed by running:

claude --version

I see a version number

Claude Code is installed. Skip ahead to the authentication check below.

Command not found

Claude Code is installed via npm, which requires Node.js 18 or higher.

Check if Node.js is installed by running:

node --version

If you see a version number below v18 or get another error, download the latest version from nodejs.org and install it.
Close and reopen your terminal after installing Node.js.
Install Claude Code globally by running:

npm install -g @anthropic-ai/claude-code

Verify the installation succeeded by running claude --version again. You should now see a version number.

Claude Code needs to be linked to your Claude Max account to run headless commands.

Authenticate Claude Code by running:

claude auth login

What happens when you run this?

This command opens a browser window where you sign in with your Claude Max account. Once you approve the connection, the terminal confirms authentication is complete.

If you've already authenticated in a previous session, the command confirms your existing login.

Confirm authentication worked by running a quick test command:

claude -p "Say hello" --output-format json

You should see a JSON response with a result field containing Claude's reply. This is the same headless pattern that the extraction script uses to enrich all 25 prospects in Step 5.

Seeing an authentication or permission error?

Make sure you have an active Claude Max subscription. Claude Code requires Claude Max to run headless commands.
If the browser window did not open during claude auth login, try running the command again. Some terminals block the browser launch on the first attempt.
If the test command times out, check your internet connection and try again.

Help me fix this Claude Code authentication issue.

Your environment is ready. Next up, you'll open Claude Desktop and design the 12-field schema that every script in this project depends on.

Design the Schema and Taxonomy with Claude Desktop

Your environment is ready with Python, DuckDB, and all the packages you need. Now it's time to design the data model that powers every script in this project.

Before generating or enriching any prospect data, you need a strict schema that defines exactly what each record should look like. You'll use Claude Desktop to design a 12-field Pydantic model and an Austin-industry taxonomy that feeds the entire pipeline.

In this step, get ready to:

Create a Claude Desktop Project for the Signals warehouse.
Design a 12-field Pydantic schema with enum validators.
Generate an Austin-industry taxonomy with real NAICS codes and zip codes.

Create a Claude Desktop Project for Signals

Claude Desktop's Projects feature lets you save context as project knowledge so every conversation in that Project starts with the same background. You'll use a Project to keep the schema design and taxonomy in one place.

Open Claude Desktop (the desktop app, not the browser at claude.ai).
Click Projects in the left sidebar.
Click Create project.
Name the project Signals - Austin SMB Warehouse.

Why use a Claude Desktop Project?

A Project acts as a persistent workspace. Any files you save as project knowledge are automatically included in every new conversation you start inside that Project. This means Claude will always have your schema decisions and taxonomy on hand without you re-explaining them.

Design the Pydantic schema with Claude

The schema defines the shape of every enriched prospect record. It uses Pydantic's BaseModel to enforce types, patterns, and allowed values at the field level.

Inside your Signals - Austin SMB Warehouse Project, start a new conversation.
Send the following prompt to Claude:

Design a Pydantic v2 model called ProspectSignal for Austin SMB prospect
enrichment data. It needs 12 fields:

1. company_name (str, required, min_length=1)
2. austin_zip (str, regex pattern for exactly 5 digits)
3. naics_code (str, exactly 6 characters)
4. employee_band (str, enum with field_validator: "1-10", "11-50", "51-200", "201-500")
5. revenue_band_est (str, enum with field_validator: "<500K", "500K-1M", "1M-5M", "5M-20M", "20M+")
6. ai_readiness_signals (optional list of strings, default None)
7. hiring_last_90d (bool, required)
8. compliance_flags (optional list of strings, default None)
9. decision_maker_title (optional str)
10. preferred_channel (str, enum with field_validator: "email", "linkedin", "phone", "referral")
11. last_signal_date (str, ISO date format)
12. confidence_score (float, between 0.0 and 1.0 inclusive)

Use Pydantic v2 syntax: BaseModel, Field, and @field_validator with @classmethod
for the three enum fields. Import from typing for Optional.

Pro tip: save as an Artifact

Claude Desktop can generate the code as an Artifact. Artifacts are interactive code blocks you can preview, copy, and iterate on without losing context. If Claude doesn't automatically create one, ask it to "save this as an Artifact."

Claude's output may differ slightly from the reference code below. That's normal. Use the following code as your source of truth for schema.py.

Create a file called schema.py inside your signals folder, and paste the following code into it:

from pydantic import BaseModel, Field, field_validator
from typing import Optional
import json


class ProspectSignal(BaseModel):
    # Required company identifiers
    company_name: str = Field(..., min_length=1)
    austin_zip: str = Field(..., pattern=r"^\d{5}$")
    naics_code: str = Field(..., min_length=6, max_length=6)

    # Business size and revenue classification
    employee_band: str = Field(...)
    revenue_band_est: str = Field(...)

    # AI and hiring signals (nullable)
    ai_readiness_signals: Optional[list[str]] = None
    hiring_last_90d: bool = Field(...)
    compliance_flags: Optional[list[str]] = None

    # Contact and timing info
    decision_maker_title: Optional[str] = None
    preferred_channel: str = Field(...)
    last_signal_date: str = Field(...)
    confidence_score: float = Field(..., ge=0.0, le=1.0)

    # Validate employee_band against allowed enum values
    @field_validator("employee_band")
    @classmethod
    def validate_employee_band(cls, v):
        valid = {"1-10", "11-50", "51-200", "201-500"}
        if v not in valid:
            raise ValueError(f"Must be one of {valid}")
        return v

    # Validate revenue_band_est against allowed enum values
    @field_validator("revenue_band_est")
    @classmethod
    def validate_revenue_band(cls, v):
        valid = {"<500K", "500K-1M", "1M-5M", "5M-20M", "20M+"}
        if v not in valid:
            raise ValueError(f"Must be one of {valid}")
        return v

    # Validate preferred_channel against allowed enum values
    @field_validator("preferred_channel")
    @classmethod
    def validate_channel(cls, v):
        valid = {"email", "linkedin", "phone", "referral"}
        if v not in valid:
            raise ValueError(f"Must be one of {valid}")
        return v


# Print the JSON Schema so other scripts can reference it
if __name__ == "__main__":
    print(json.dumps(ProspectSignal.model_json_schema(), indent=2))

What does this code do?

ProspectSignal is a Pydantic BaseModel that acts as a strict contract for prospect data. Any record that doesn't match all 12 field rules is rejected.
austin_zip uses a regex pattern to enforce exactly 5 digits. naics_code uses min_length=6, max_length=6 to lock the length to exactly 6 characters.
The three @field_validator methods check that employee_band, revenue_band_est, and preferred_channel only accept values from a fixed set. Anything else raises a ValueError.
The __main__ block prints the JSON Schema for this model. The extraction script in Step 5 sends this schema to Claude Code so it knows exactly what shape to return.

Save schema.py.
Verify the schema by running:

python schema.py

You should see a formatted JSON Schema printed to the terminal. It includes a properties object listing all 12 fields, a required array, and pattern/enum constraints.

Seeing a ModuleNotFoundError?

Make sure your virtual environment is activated. You should see (.venv) at the start of your terminal prompt.
If you see No module named 'pydantic', re-run the pip install command from Step 1.
help me fix a ModuleNotFoundError when running schema.py

Awesome, I've got everything!

Double check you've saved schema.py and that the JSON Schema printed cleanly in your terminal.

I'd like to double check the full code

from pydantic import BaseModel, Field, field_validator
from typing import Optional
import json


class ProspectSignal(BaseModel):
    company_name: str = Field(..., min_length=1)
    austin_zip: str = Field(..., pattern=r"^\d{5}$")
    naics_code: str = Field(..., min_length=6, max_length=6)
    employee_band: str = Field(...)
    revenue_band_est: str = Field(...)
    ai_readiness_signals: Optional[list[str]] = None
    hiring_last_90d: bool = Field(...)
    compliance_flags: Optional[list[str]] = None
    decision_maker_title: Optional[str] = None
    preferred_channel: str = Field(...)
    last_signal_date: str = Field(...)
    confidence_score: float = Field(..., ge=0.0, le=1.0)

    @field_validator("employee_band")
    @classmethod
    def validate_employee_band(cls, v):
        valid = {"1-10", "11-50", "51-200", "201-500"}
        if v not in valid:
            raise ValueError(f"Must be one of {valid}")
        return v

    @field_validator("revenue_band_est")
    @classmethod
    def validate_revenue_band(cls, v):
        valid = {"<500K", "500K-1M", "1M-5M", "5M-20M", "20M+"}
        if v not in valid:
            raise ValueError(f"Must be one of {valid}")
        return v

    @field_validator("preferred_channel")
    @classmethod
    def validate_channel(cls, v):
        valid = {"email", "linkedin", "phone", "referral"}
        if v not in valid:
            raise ValueError(f"Must be one of {valid}")
        return v


if __name__ == "__main__":
    print(json.dumps(ProspectSignal.model_json_schema(), indent=2))

Generate the Austin-industry taxonomy

The universe generator in Step 3 needs two things: a list of real Austin-area zip codes, and a set of real 6-digit NAICS codes covering diverse industry verticals. You'll use Claude Desktop to research and compile this taxonomy.

Inside your Signals - Austin SMB Warehouse Project in Claude Desktop, start a new conversation.
Send the following prompt to Claude:

Generate a taxonomy of 25-30 Austin-metro SMB industry verticals.
For each vertical, provide:
- The real 6-digit NAICS code
- A short industry label (e.g. "Management Consulting", "Dental Practice")

Cover diverse sectors: healthcare (dental, medical, chiropractic),
construction (plumbing/HVAC, electrical, residential, commercial),
professional services (law, bookkeeping, IT consulting, management consulting,
marketing), food service (restaurant, catering), retail (florist, farm supply),
logistics (trucking, equipment rental), personal care (barber, salon, fitness),
and other services (staffing, property management, insurance, daycare,
auto repair, veterinary, financial advisory, printing).

Also provide a list of approximately 50 real Austin-metro zip codes
covering central Austin (787xx), plus surrounding areas like
Round Rock, Pflugerville, Georgetown, Cedar Park, Buda, Kyle,
Bastrop, Leander, and Dripping Springs.

Format both lists as Python dictionaries I can copy directly into a script.

What are NAICS codes?

NAICS (North American Industry Classification System) codes are standardized 6-digit codes that classify businesses by industry. For example, 541611 is Management Consulting and 621210 is Dental Practice. Using real NAICS codes makes the synthetic data statistically representative of the actual Austin market.

Review Claude's output. Confirm it includes at least 25 NAICS codes and approximately 50 zip codes.
Click the three-dot menu at the top right of the Claude Desktop conversation, then select Add to project knowledge. This saves the taxonomy so future conversations in this Project have it as context.

Why save the taxonomy as project knowledge?

Project knowledge persists across conversations. If you come back to this Project later to adjust the taxonomy or add new industries, Claude already knows what you built. The taxonomy you designed here feeds directly into generate_universe.py in the next step.

Pro tip: compare with the reference

You don't need to use Claude's exact output for the rest of the project. The scripts in Steps 3 and 4 already contain curated NAICS codes and zip codes that work with the pipeline. This exercise trains you to use Claude Desktop as a schema-first design tool.

Your schema and taxonomy are locked in. Next up, you'll use the taxonomy to generate a 55,000-row synthetic universe of Austin SMBs in DuckDB.

Generate the 55,000-Row Austin SMB Universe

Your Pydantic schema and Austin-industry taxonomy are locked in. You have a 12-field data model and a list of 30 real industry verticals with their NAICS codes and zip codes.

In this step, you'll write a Python script that creates 55,000 synthetic Austin SMBs and loads them into a DuckDB database. This is the haystack. The next step finds the needles.

In this step, get ready to:

Write generate_universe.py with real Austin zip codes, NAICS codes, and weighted randomization.
Run the generator and verify 55,000 rows exist in the DuckDB universe table.

Write generate_universe.py

Synthetic data is only useful if it mirrors real-world patterns. Your generator uses real Austin-metro zip codes, real 6-digit NAICS industry codes, and a skewed size distribution so that most companies are small and a few are larger.

Create a new file called generate_universe.py in your signals folder.
Add the imports, reference data constants, and name generator function by copying this code into the file:

import duckdb
import numpy as np
import random

# Real Austin-metro zip codes (~55 covering Austin and surrounding suburbs)
AUSTIN_ZIPS = [
    "78701", "78702", "78703", "78704", "78705", "78712", "78717",
    "78719", "78721", "78722", "78723", "78724", "78725", "78726",
    "78727", "78728", "78729", "78730", "78731", "78732", "78733",
    "78734", "78735", "78736", "78737", "78738", "78739", "78741",
    "78742", "78744", "78745", "78746", "78748", "78749", "78750",
    "78751", "78752", "78753", "78754", "78756", "78757", "78758",
    "78759", "78602", "78610", "78613", "78620", "78626", "78628",
    "78640", "78641", "78653", "78660", "78664", "78681",
]

# 30 real 6-digit NAICS codes mapped to Austin-area industry names
NAICS_INDUSTRIES = {
    "541611": "Management Consulting",
    "541512": "IT Consulting",
    "541810": "Marketing Agency",
    "541211": "Bookkeeping",
    "541110": "Law Firm",
    "541940": "Veterinary",
    "523930": "Financial Advisory",
    "621210": "Dental Practice",
    "621111": "Medical Office",
    "621310": "Chiropractic",
    "238220": "Plumbing HVAC",
    "238210": "Electrical",
    "236220": "Construction",
    "236118": "Residential Builder",
    "722511": "Restaurant",
    "722320": "Catering",
    "531311": "Property Mgmt",
    "561311": "Staffing",
    "561730": "Landscaping",
    "811121": "Auto Repair",
    "713940": "Fitness Center",
    "624410": "Child Daycare",
    "524210": "Insurance Agency",
    "484110": "Trucking",
    "323111": "Printing",
    "532490": "Equipment Rental",
    "444240": "Farm Supply",
    "812111": "Barber Shop",
    "812112": "Salon",
    "453110": "Florist",
}

# Location-based prefixes for composing realistic company names
PREFIXES = [
    "Austin", "Capital", "Lone Star", "Hill Country", "Central Texas",
    "South", "North", "East", "West", "Downtown", "Lakeside", "Cedar",
    "Oak", "Round Rock", "Mueller", "Domain", "Travis", "Barton",
    "Congress", "Lamar", "Guadalupe", "Manor", "Pflugerville", "Kyle",
    "Buda", "Leander", "Georgetown", "Bastrop", "Dripping Springs",
]

# Business entity suffixes
SUFFIXES = ["LLC", "Inc", "Co", "Group", "Services", "Solutions", "Partners"]


def generate_company_name(industry_name):
    # Build a realistic name: prefix + industry keyword + suffix
    prefix = random.choice(PREFIXES)
    words = industry_name.split()
    core = random.choice(words) if len(words) > 1 else industry_name
    suffix = random.choice(SUFFIXES)
    return f"{prefix} {core} {suffix}"

What does this code do?

AUSTIN_ZIPS holds ~55 real zip codes covering Austin proper and surrounding suburbs like Round Rock, Pflugerville, and Georgetown.
NAICS_INDUSTRIES maps 30 real 6-digit NAICS codes to human-readable industry names. These come from the taxonomy you designed in Step 2.
PREFIXES and SUFFIXES are word pools for building random company names that sound like real Austin businesses.
generate_company_name() picks one prefix, one keyword from the industry name, and one suffix to compose each name.

What is a NAICS code?

NAICS stands for North American Industry Classification System. Every business in the US, Canada, and Mexico is assigned a 6-digit code that identifies its industry. The first two digits are the broad sector (e.g. 54 = Professional Services). Each additional digit narrows the classification further.

Save generate_universe.py.

You won't see any output yet. The next chunk adds the main() function that uses all of this reference data to generate 55,000 rows.

Add the main() function and entry point below generate_company_name() in generate_universe.py:

def main():
    # Lock random seeds for reproducible output
    random.seed(42)
    np.random.seed(42)

    n = 55000
    naics_codes = list(NAICS_INDUSTRIES.keys())
    naics_names = list(NAICS_INDUSTRIES.values())

    # Generate 55,000 rows of synthetic SMB data
    rows = []
    for i in range(n):
        idx = random.randint(0, len(naics_codes) - 1)
        naics = naics_codes[idx]
        industry = naics_names[idx]
        name = generate_company_name(industry)
        name = f"{name} {i}"  # Append index to ensure unique names
        zip_code = random.choice(AUSTIN_ZIPS)
        # Lognormal distribution: most SMBs are small, a few are larger
        emp = int(np.random.lognormal(mean=2.5, sigma=1.2))
        emp = max(1, min(emp, 500))
        rows.append((name, zip_code, naics, emp))

    # Connect to DuckDB and create the universe table
    con = duckdb.connect("signals.duckdb")
    con.execute("""
        CREATE OR REPLACE TABLE universe (
            company_name VARCHAR,
            zip_code VARCHAR(5),
            naics_code VARCHAR(6),
            employee_count_est INTEGER
        )
    """)

    # Batch insert all 55,000 rows at once
    con.executemany(
        "INSERT INTO universe VALUES (?, ?, ?, ?)", rows
    )

    # Print summary statistics
    total = con.execute("SELECT COUNT(*) FROM universe").fetchone()[0]
    print(f"Universe created: {total:,} rows")

    print("\nZip code distribution (top 10):")
    zips = con.execute("""
        SELECT zip_code, COUNT(*) as cnt
        FROM universe GROUP BY zip_code
        ORDER BY cnt DESC LIMIT 10
    """).df()
    print(zips.to_string(index=False))

    print("\nNAICS distribution (top 10):")
    naics_dist = con.execute("""
        SELECT naics_code, COUNT(*) as cnt
        FROM universe GROUP BY naics_code
        ORDER BY cnt DESC LIMIT 10
    """).df()
    print(naics_dist.to_string(index=False))

    con.close()
    print("\nDone. signals.duckdb now contains the 'universe' table.")


if __name__ == "__main__":
    main()

What does this code do?

random.seed(42) and np.random.seed(42) lock the random number generators so every run produces identical data. This makes your output reproducible.
The loop generates 55,000 rows. Each row gets a random NAICS code, a composed company name, a random Austin zip code, and an employee count drawn from a lognormal distribution.
np.random.lognormal(mean=2.5, sigma=1.2) produces right-skewed numbers. Most values land between 1 and 50, with a few reaching up to 500. This mirrors real SMB size distributions where small companies vastly outnumber large ones.
duckdb.connect("signals.duckdb") creates the database file if it doesn't exist. CREATE OR REPLACE TABLE ensures a clean slate on every run.
executemany() batch-inserts all 55,000 rows in one call, which is much faster than inserting one at a time.
The summary queries use DuckDB's .df() method to convert results to a Pandas DataFrame for clean terminal output.

Save generate_universe.py.

The file is complete. You'll run it in the next substep to generate the data and verify the output.

Awesome, I've got everything!

Double check you've saved your file before running it.

I'd like to double check the full code

import duckdb
import numpy as np
import random

AUSTIN_ZIPS = [
    "78701", "78702", "78703", "78704", "78705", "78712", "78717",
    "78719", "78721", "78722", "78723", "78724", "78725", "78726",
    "78727", "78728", "78729", "78730", "78731", "78732", "78733",
    "78734", "78735", "78736", "78737", "78738", "78739", "78741",
    "78742", "78744", "78745", "78746", "78748", "78749", "78750",
    "78751", "78752", "78753", "78754", "78756", "78757", "78758",
    "78759", "78602", "78610", "78613", "78620", "78626", "78628",
    "78640", "78641", "78653", "78660", "78664", "78681",
]

NAICS_INDUSTRIES = {
    "541611": "Management Consulting",
    "541512": "IT Consulting",
    "541810": "Marketing Agency",
    "541211": "Bookkeeping",
    "541110": "Law Firm",
    "541940": "Veterinary",
    "523930": "Financial Advisory",
    "621210": "Dental Practice",
    "621111": "Medical Office",
    "621310": "Chiropractic",
    "238220": "Plumbing HVAC",
    "238210": "Electrical",
    "236220": "Construction",
    "236118": "Residential Builder",
    "722511": "Restaurant",
    "722320": "Catering",
    "531311": "Property Mgmt",
    "561311": "Staffing",
    "561730": "Landscaping",
    "811121": "Auto Repair",
    "713940": "Fitness Center",
    "624410": "Child Daycare",
    "524210": "Insurance Agency",
    "484110": "Trucking",
    "323111": "Printing",
    "532490": "Equipment Rental",
    "444240": "Farm Supply",
    "812111": "Barber Shop",
    "812112": "Salon",
    "453110": "Florist",
}

PREFIXES = [
    "Austin", "Capital", "Lone Star", "Hill Country", "Central Texas",
    "South", "North", "East", "West", "Downtown", "Lakeside", "Cedar",
    "Oak", "Round Rock", "Mueller", "Domain", "Travis", "Barton",
    "Congress", "Lamar", "Guadalupe", "Manor", "Pflugerville", "Kyle",
    "Buda", "Leander", "Georgetown", "Bastrop", "Dripping Springs",
]

SUFFIXES = ["LLC", "Inc", "Co", "Group", "Services", "Solutions", "Partners"]


def generate_company_name(industry_name):
    prefix = random.choice(PREFIXES)
    words = industry_name.split()
    core = random.choice(words) if len(words) > 1 else industry_name
    suffix = random.choice(SUFFIXES)
    return f"{prefix} {core} {suffix}"


def main():
    random.seed(42)
    np.random.seed(42)

    n = 55000
    naics_codes = list(NAICS_INDUSTRIES.keys())
    naics_names = list(NAICS_INDUSTRIES.values())

    rows = []
    for i in range(n):
        idx = random.randint(0, len(naics_codes) - 1)
        naics = naics_codes[idx]
        industry = naics_names[idx]
        name = generate_company_name(industry)
        name = f"{name} {i}"
        zip_code = random.choice(AUSTIN_ZIPS)
        emp = int(np.random.lognormal(mean=2.5, sigma=1.2))
        emp = max(1, min(emp, 500))
        rows.append((name, zip_code, naics, emp))

    con = duckdb.connect("signals.duckdb")
    con.execute("""
        CREATE OR REPLACE TABLE universe (
            company_name VARCHAR,
            zip_code VARCHAR(5),
            naics_code VARCHAR(6),
            employee_count_est INTEGER
        )
    """)

    con.executemany(
        "INSERT INTO universe VALUES (?, ?, ?, ?)", rows
    )

    total = con.execute("SELECT COUNT(*) FROM universe").fetchone()[0]
    print(f"Universe created: {total:,} rows")

    print("\nZip code distribution (top 10):")
    zips = con.execute("""
        SELECT zip_code, COUNT(*) as cnt
        FROM universe GROUP BY zip_code
        ORDER BY cnt DESC LIMIT 10
    """).df()
    print(zips.to_string(index=False))

    print("\nNAICS distribution (top 10):")
    naics_dist = con.execute("""
        SELECT naics_code, COUNT(*) as cnt
        FROM universe GROUP BY naics_code
        ORDER BY cnt DESC LIMIT 10
    """).df()
    print(naics_dist.to_string(index=False))

    con.close()
    print("\nDone. signals.duckdb now contains the 'universe' table.")


if __name__ == "__main__":
    main()

Run the generator and verify

The script is ready. Running it creates the signals.duckdb database file, inserts 55,000 rows, and prints a summary of the data distribution.

Run the generator by executing this command in your terminal:

python generate_universe.py

The script should complete in under 10 seconds. Your terminal prints a row count, the top 10 zip codes by frequency, and the top 10 NAICS codes by frequency.

What should I see?

The first line reads Universe created: 55,000 rows. Below that, you'll see two tables. The zip code table shows each zip with its count, spread roughly evenly across the ~55 codes. The NAICS table shows each industry code with its count, also spread roughly evenly across the 30 industries.

Confirm your terminal output starts with Universe created: 55,000 rows.
Confirm you see a signals.duckdb file in your signals folder by running:

ls

You should see signals.duckdb, generate_universe.py, and schema.py in the output.

Seeing an error when you run the script?

ModuleNotFoundError: No module named 'duckdb' means your virtual environment is not activated or the packages were not installed. Run pip install duckdb numpy and try again.
If you see ModuleNotFoundError: No module named 'numpy', the numpy package is missing. Run pip install numpy to install it.
If the row count is wrong or the script errors partway through, make sure your NAICS_INDUSTRIES dictionary and AUSTIN_ZIPS list match the code exactly. A missing comma or unclosed bracket is the most common cause.

Help me debug my generate_universe.py script

Why 55,000 rows?

The SBA estimates approximately 55,000-60,000 small businesses in the Austin metro area. Your synthetic universe matches that real-world scale. Starting with a realistic volume means the ICP funnel in the next step produces a realistic hit rate too.

Your database now holds 55,000 Austin SMBs spread across 30 industries and 55 zip codes. Next up, you'll write an ICP filter that narrows this universe down to ~150 matches and scores the top 25 for enrichment.

Apply the ICP Filter and Select the Top 25

Your 55,000-row universe is sitting in DuckDB. Most of those businesses are not a fit for a solo AI consultant. A florist or barber shop probably does not need an AI readiness assessment.

In this step, you will apply an Ideal Customer Profile filter to narrow the universe down to roughly 150 matches. Then a scoring function ranks them and selects the top 25 prospects for Claude Code to enrich in the next step.

In this step, get ready to:

Write a funnel script that filters the universe by industry, employee count, and geography.
Run the funnel and inspect the 25 seed prospects it selects.

Write the funnel script

Not every business in the universe is worth enriching. An Ideal Customer Profile (ICP) defines the characteristics of your ideal buyer so you can focus your resources on the prospects most likely to close.

What is an Ideal Customer Profile?

An ICP is a set of filters that describe the companies most likely to buy from you. For a solo AI consultant targeting Austin SMBs, the ICP in this project is:

Industry: consulting, healthcare, construction, professional services, financial, staffing, and property management.
Employee count: between 5 and 200 (big enough to have budget, small enough to need outside help).
Geography: Austin metro zip codes (already baked into the universe from Step 3).

Create a new file called funnel.py in your signals folder.
Add the imports and ICP configuration by pasting this code:

import duckdb
import csv
import json

# NAICS codes for industries a solo AI consultant would target
TARGET_NAICS = [
    "541611", "541512", "541810", "541211", "541110",  # Professional services
    "621210", "621111", "621310",                       # Healthcare
    "238220", "238210", "236220",                       # Construction trades
    "531311", "561311",                                 # Property mgmt, staffing
    "523930", "524210",                                 # Financial, insurance
]

# Human-readable industry label for each NAICS code
NAICS_TO_HINT = {
    "541611": "management consulting",
    "541512": "IT consulting",
    "541810": "digital marketing agency",
    "541211": "bookkeeping services",
    "541110": "law firm",
    "621210": "dental practice",
    "621111": "medical office",
    "621310": "chiropractic practice",
    "238220": "plumbing and HVAC",
    "238210": "electrical contractor",
    "236220": "commercial construction",
    "531311": "property management",
    "561311": "staffing agency",
    "523930": "financial advisory",
    "524210": "insurance agency",
}

# Desirability score per industry (higher = better fit for AI consulting)
INDUSTRY_WEIGHT = {
    "541611": 10, "541512": 10, "541810": 9, "541211": 7,
    "541110": 8, "621210": 8, "621111": 7, "621310": 6,
    "238220": 6, "238210": 6, "236220": 7, "531311": 7,
    "561311": 8, "523930": 8, "524210": 6,
}

What does this code do?

TARGET_NAICS lists the 15 six-digit NAICS codes that represent industries a solo AI consultant would target. These come from the taxonomy you designed in Step 2.
NAICS_TO_HINT maps each code to a plain-English industry label. Claude Code uses these hints during enrichment to understand what kind of business it is researching.
INDUSTRY_WEIGHT assigns a desirability score (1-10) to each industry. Management consulting and IT consulting score 10 because they are the highest-value targets for AI services.

You will not see a change yet. The next chunk uses these constants to run the actual filter.

In funnel.py, add the main function below the INDUSTRY_WEIGHT dictionary by pasting this code:


def main():
    # Connect in read-only mode since we only query the universe table
    con = duckdb.connect("signals.duckdb", read_only=True)

    # Get total universe size for funnel metrics
    universe_count = con.execute(
        "SELECT COUNT(*) FROM universe"
    ).fetchone()[0]

    # Build the ICP filter: target industries + employee count 5-200
    naics_list = ", ".join(f"'{n}'" for n in TARGET_NAICS)
    icp_query = f"""
        SELECT company_name, zip_code, naics_code, employee_count_est
        FROM universe
        WHERE employee_count_est BETWEEN 5 AND 200
        AND naics_code IN ({naics_list})
    """
    icp_df = con.execute(icp_query).df()
    icp_count = len(icp_df)

    # Score each ICP match by industry desirability
    icp_df["industry_score"] = icp_df["naics_code"].map(INDUSTRY_WEIGHT)

    # Score by employee band (sweet spot is 10-50 employees)
    def emp_score(emp):
        if 10 <= emp <= 50:
            return 10
        elif 51 <= emp <= 200:
            return 8
        elif 5 <= emp <= 9:
            return 5
        return 3

    icp_df["emp_score"] = icp_df["employee_count_est"].apply(emp_score)
    icp_df["total_score"] = icp_df["industry_score"] + icp_df["emp_score"]

    # Take the top 25 by combined score
    top25 = icp_df.nlargest(25, "total_score")

    # Write the 25 seeds to CSV for the extraction step
    with open("seeds.csv", "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(
            f, fieldnames=["company_name", "zip_code", "naics_code", "industry_hint"]
        )
        writer.writeheader()
        for _, row in top25.iterrows():
            hint = NAICS_TO_HINT.get(row["naics_code"], "general business")
            writer.writerow({
                "company_name": row["company_name"],
                "zip_code": row["zip_code"],
                "naics_code": row["naics_code"],
                "industry_hint": hint,
            })

    # Save funnel counts for the dashboard
    metrics = {
        "universe_count": int(universe_count),
        "icp_count": int(icp_count),
        "enriched_count": 25,
    }
    with open("funnel_metrics.json", "w", encoding="utf-8") as f:
        json.dump(metrics, f, indent=2)

    con.close()

    print(f"Universe: {universe_count:,}")
    print(f"ICP matches: {icp_count:,}")
    print(f"Top 25 selected and written to seeds.csv")
    print(f"Funnel metrics saved to funnel_metrics.json")


if __name__ == "__main__":
    main()

What does this code do?

The main function runs a two-stage funnel:

Stage 1 (ICP filter): A SQL query selects only rows where employee_count_est is between 5 and 200 AND the naics_code is in your target list. This drops the universe from ~55,000 to roughly 150 matches.
Stage 2 (scoring): Each match gets an industry desirability score plus an employee band score. Companies with 10-50 employees in high-value industries like IT consulting score highest. nlargest(25) picks the top 25.
The script writes the 25 seeds to seeds.csv with an industry_hint column that gives Claude Code context about each business. It also saves the funnel counts to funnel_metrics.json for the dashboard in Step 6.

Save funnel.py.

Seeing a syntax error?

Make sure the main() function starts at the top level (no indentation), not nested inside a dictionary.
Check that the f-string triple quotes in icp_query use three double quotes on each side: f"""...""".
help me debug a syntax error in funnel.py

Awesome, I've got everything!

Double check you have saved funnel.py before moving on.

I'd like to double check the full code

import duckdb
import csv
import json

# NAICS codes for industries a solo AI consultant would target
TARGET_NAICS = [
    "541611", "541512", "541810", "541211", "541110",
    "621210", "621111", "621310",
    "238220", "238210", "236220",
    "531311", "561311",
    "523930", "524210",
]

# Human-readable industry label for each NAICS code
NAICS_TO_HINT = {
    "541611": "management consulting",
    "541512": "IT consulting",
    "541810": "digital marketing agency",
    "541211": "bookkeeping services",
    "541110": "law firm",
    "621210": "dental practice",
    "621111": "medical office",
    "621310": "chiropractic practice",
    "238220": "plumbing and HVAC",
    "238210": "electrical contractor",
    "236220": "commercial construction",
    "531311": "property management",
    "561311": "staffing agency",
    "523930": "financial advisory",
    "524210": "insurance agency",
}

# Desirability score per industry (higher = better fit for AI consulting)
INDUSTRY_WEIGHT = {
    "541611": 10, "541512": 10, "541810": 9, "541211": 7,
    "541110": 8, "621210": 8, "621111": 7, "621310": 6,
    "238220": 6, "238210": 6, "236220": 7, "531311": 7,
    "561311": 8, "523930": 8, "524210": 6,
}


def main():
    # Connect in read-only mode since we only query the universe table
    con = duckdb.connect("signals.duckdb", read_only=True)

    # Get total universe size for funnel metrics
    universe_count = con.execute(
        "SELECT COUNT(*) FROM universe"
    ).fetchone()[0]

    # Build the ICP filter: target industries + employee count 5-200
    naics_list = ", ".join(f"'{n}'" for n in TARGET_NAICS)
    icp_query = f"""
        SELECT company_name, zip_code, naics_code, employee_count_est
        FROM universe
        WHERE employee_count_est BETWEEN 5 AND 200
        AND naics_code IN ({naics_list})
    """
    icp_df = con.execute(icp_query).df()
    icp_count = len(icp_df)

    # Score each ICP match by industry desirability
    icp_df["industry_score"] = icp_df["naics_code"].map(INDUSTRY_WEIGHT)

    # Score by employee band (sweet spot is 10-50 employees)
    def emp_score(emp):
        if 10 <= emp <= 50:
            return 10
        elif 51 <= emp <= 200:
            return 8
        elif 5 <= emp <= 9:
            return 5
        return 3

    icp_df["emp_score"] = icp_df["employee_count_est"].apply(emp_score)
    icp_df["total_score"] = icp_df["industry_score"] + icp_df["emp_score"]

    # Take the top 25 by combined score
    top25 = icp_df.nlargest(25, "total_score")

    # Write the 25 seeds to CSV for the extraction step
    with open("seeds.csv", "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(
            f, fieldnames=["company_name", "zip_code", "naics_code", "industry_hint"]
        )
        writer.writeheader()
        for _, row in top25.iterrows():
            hint = NAICS_TO_HINT.get(row["naics_code"], "general business")
            writer.writerow({
                "company_name": row["company_name"],
                "zip_code": row["zip_code"],
                "naics_code": row["naics_code"],
                "industry_hint": hint,
            })

    # Save funnel counts for the dashboard
    metrics = {
        "universe_count": int(universe_count),
        "icp_count": int(icp_count),
        "enriched_count": 25,
    }
    with open("funnel_metrics.json", "w", encoding="utf-8") as f:
        json.dump(metrics, f, indent=2)

    con.close()

    print(f"Universe: {universe_count:,}")
    print(f"ICP matches: {icp_count:,}")
    print(f"Top 25 selected and written to seeds.csv")
    print(f"Funnel metrics saved to funnel_metrics.json")


if __name__ == "__main__":
    main()

Run the funnel and inspect the seeds

Run the funnel script by executing this command:

python funnel.py

The script prints four lines showing the funnel in action. Your ICP match count will be roughly 150, depending on how the random distribution landed in Step 3.

What should I see?

Your terminal output should look similar to this:

Universe: 55,000
ICP matches: ~150
Top 25 selected and written to seeds.csv
Funnel metrics saved to funnel_metrics.json

The exact ICP match count varies, but it should be in the range of 100 to 200. The important thing is that the funnel narrowed 55,000 rows down to a manageable set and selected the top 25.

Open seeds.csv in your editor.
Confirm it has exactly 25 rows (plus the header) with columns company_name, zip_code, naics_code, and industry_hint.
Verify that the zip codes are valid 5-digit Austin-area codes and that each naics_code is exactly 6 digits.

Open funnel_metrics.json in your editor.
Confirm it contains three keys: universe_count (55000), icp_count (roughly 150), and enriched_count (25).

Seeing zero ICP matches?

Make sure you ran generate_universe.py first and that signals.duckdb exists in your signals folder.
Check that the NAICS codes in TARGET_NAICS match the codes used in generate_universe.py. A typo in one code means that entire industry gets filtered out.
If you see a Table 'universe' does not exist error, re-run python generate_universe.py to recreate the table.
help me debug why funnel.py returns zero ICP matches

You now have 25 high-quality seed prospects in seeds.csv. These are the needles you pulled from a 55,000-row haystack. Next up, Claude Code will enrich each one with detailed prospect signals, validated against your Pydantic schema.

Enrich the Top 25 with Claude Code

Your funnel narrowed 55,000 Austin SMBs down to 25 high-scoring prospects sitting in seeds.csv. Names, zip codes, and NAICS codes are useful, but a consultant needs richer signals before picking up the phone.

In this step, you will write a script that sends each seed to Claude Code running in headless mode. Claude enriches each prospect with 12 validated fields, including revenue band, AI readiness signals, and a confidence score. If a response fails Pydantic validation, the script retries with the error message appended so Claude can self-correct.

In this step, get ready to:

Write an extraction script that calls Claude Code headlessly for each of the 25 seeds.
Run the extraction loop and watch the validation-retry cycle in action.
Inspect the enriched records and confirm they match the 12-field Pydantic schema.

Write the extraction script

The key command powering this script is claude -p "prompt" --output-format json. Claude Code processes the prompt, returns structured JSON, and exits. No interactive UI, no manual approval.

Create a new file called extract.py in your signals folder.
Start with the imports and module constants by copying in the following code:

import csv
import json
import subprocess
from schema import ProspectSignal
from pydantic import ValidationError

# Serialize the Pydantic schema to JSON so it can be embedded in each prompt
SCHEMA_JSON = json.dumps(ProspectSignal.model_json_schema(), indent=2)
# Cap the number of retries per seed company
MAX_RETRIES = 3

What does this code do?

ProspectSignal is the Pydantic model you built in Step 2. Every enrichment record must satisfy its 12-field contract.
SCHEMA_JSON converts that model into a JSON Schema string. This gets embedded in every prompt so Claude knows the exact shape of data to return.
MAX_RETRIES limits how many times the script re-asks Claude if a response fails validation.

You will not see a change yet. The next chunk builds the function that calls Claude.

Below the constants you just added, copy in the extract_one function:

def extract_one(seed_row, attempt=1, prev_error=""):
    # Pull company details from the seed row
    company = seed_row["company_name"]
    zip_code = seed_row["zip_code"]
    naics = seed_row["naics_code"]
    hint = seed_row["industry_hint"]

    # If a previous attempt failed, include the error so Claude can self-correct
    error_context = ""
    if prev_error:
        error_context = (
            "\n\nYour previous response failed validation with this error:\n"
            f"{prev_error}\nFix the issue and try again."
        )

    # Build the extraction prompt with seed info and the full JSON Schema
    prompt = (
        "You are a business research assistant. Generate realistic prospect "
        "enrichment data for this synthetic Austin-area SMB.\n\n"
        f"Company: {company}\n"
        f"ZIP: {zip_code}\n"
        f"NAICS: {naics}\n"
        f"Industry: {hint}\n\n"
        "Return ONLY valid JSON matching this schema. "
        "No markdown fences, no explanation, just the JSON object.\n\n"
        f"{SCHEMA_JSON}"
        f"{error_context}"
    )

    try:
        # Call Claude Code headlessly and capture the output
        result = subprocess.run(
            ["claude", "-p", prompt, "--output-format", "json"],
            capture_output=True,
            text=True,
            timeout=120,
        )

        if result.returncode != 0:
            print(f"  Claude error: {result.stderr[:200]}")
            if attempt < MAX_RETRIES:
                return extract_one(seed_row, attempt + 1, "Non-zero exit code")
            return None

        # Parse the outer JSON wrapper that Claude Code returns
        response = json.loads(result.stdout)
        content = response.get("result", "")

        # Strip any markdown fences Claude might include
        content = content.strip()
        if content.startswith("```"):
            content = content.split("\n", 1)[1]
        if content.endswith("```"):
            content = content.rsplit("```", 1)[0]
        content = content.strip()

        # Validate the parsed record against the Pydantic model
        record = json.loads(content)
        validated = ProspectSignal.model_validate(record)
        return validated.model_dump()

    except json.JSONDecodeError as e:
        error_msg = f"JSON parse error: {e}"
    except ValidationError as e:
        error_msg = str(e)
    except subprocess.TimeoutExpired:
        error_msg = "Timed out after 120s"
    except Exception as e:
        error_msg = f"Unexpected error: {e}"

    print(f"  Attempt {attempt} failed: {error_msg[:150]}")

    # Retry with the error message fed back into the next prompt
    if attempt < MAX_RETRIES:
        return extract_one(seed_row, attempt + 1, error_msg)

    print(f"  SKIPPED after {MAX_RETRIES} attempts")
    return None

What does this function do?

It builds a prompt containing the seed company's name, zip, NAICS code, and industry hint, plus the full JSON Schema.
It runs claude -p as a subprocess. The --output-format json flag tells Claude Code to return a JSON object with a result field containing the text response.
The response is parsed and validated against ProspectSignal. If validation fails, the Pydantic error message is appended to the next prompt so Claude sees exactly what went wrong.
After MAX_RETRIES failed attempts, the seed is skipped and the loop moves on.

Why feed errors back into the prompt?

Pydantic error messages are very specific. They say exactly which field failed and why. By appending the error to the retry prompt, Claude can fix the exact issue instead of guessing. This self-correction pattern is how production AI pipelines handle structured output at scale.

You will not see a change yet. The next chunk adds the main function that ties everything together.

Below extract_one, add the main function and the entry point:

def main():
    # Read the 25 seed companies from the CSV
    with open("seeds.csv", newline="", encoding="utf-8") as f:
        seeds = list(csv.DictReader(f))

    print(f"Processing {len(seeds)} seeds with Claude Code...\n")
    records = []

    # Loop through each seed and call extract_one
    for i, seed in enumerate(seeds, 1):
        print(f"[{i}/{len(seeds)}] {seed['company_name']}")
        record = extract_one(seed)
        if record:
            records.append(record)
            print(f"  OK (confidence: {record['confidence_score']})")
        print()

    # Write all valid records to a JSON file
    with open("extracted.json", "w", encoding="utf-8") as f:
        json.dump(records, f, indent=2)

    print(f"Done. {len(records)}/{len(seeds)} records saved to extracted.json")


if __name__ == "__main__":
    main()

What does this code do?

It opens seeds.csv and loops through all 25 rows, calling extract_one for each seed.
For each successful extraction, the record is appended to a list and the confidence score is printed.
At the end, all valid records are written to extracted.json and a summary count is printed.

Save extract.py.

Awesome, I've got everything!

Double check you have saved extract.py with all three sections: imports and constants, the extract_one function, and the main function.

I'd like to double check the full code

import csv
import json
import subprocess
from schema import ProspectSignal
from pydantic import ValidationError


SCHEMA_JSON = json.dumps(ProspectSignal.model_json_schema(), indent=2)
MAX_RETRIES = 3


def extract_one(seed_row, attempt=1, prev_error=""):
    company = seed_row["company_name"]
    zip_code = seed_row["zip_code"]
    naics = seed_row["naics_code"]
    hint = seed_row["industry_hint"]

    error_context = ""
    if prev_error:
        error_context = (
            "\n\nYour previous response failed validation with this error:\n"
            f"{prev_error}\nFix the issue and try again."
        )

    prompt = (
        "You are a business research assistant. Generate realistic prospect "
        "enrichment data for this synthetic Austin-area SMB.\n\n"
        f"Company: {company}\n"
        f"ZIP: {zip_code}\n"
        f"NAICS: {naics}\n"
        f"Industry: {hint}\n\n"
        "Return ONLY valid JSON matching this schema. "
        "No markdown fences, no explanation, just the JSON object.\n\n"
        f"{SCHEMA_JSON}"
        f"{error_context}"
    )

    try:
        result = subprocess.run(
            ["claude", "-p", prompt, "--output-format", "json"],
            capture_output=True,
            text=True,
            timeout=120,
        )

        if result.returncode != 0:
            print(f"  Claude error: {result.stderr[:200]}")
            if attempt < MAX_RETRIES:
                return extract_one(seed_row, attempt + 1, "Non-zero exit code")
            return None

        response = json.loads(result.stdout)
        content = response.get("result", "")

        content = content.strip()
        if content.startswith("```"):
            content = content.split("\n", 1)[1]
        if content.endswith("```"):
            content = content.rsplit("```", 1)[0]
        content = content.strip()

        record = json.loads(content)
        validated = ProspectSignal.model_validate(record)
        return validated.model_dump()

    except json.JSONDecodeError as e:
        error_msg = f"JSON parse error: {e}"
    except ValidationError as e:
        error_msg = str(e)
    except subprocess.TimeoutExpired:
        error_msg = "Timed out after 120s"
    except Exception as e:
        error_msg = f"Unexpected error: {e}"

    print(f"  Attempt {attempt} failed: {error_msg[:150]}")

    if attempt < MAX_RETRIES:
        return extract_one(seed_row, attempt + 1, error_msg)

    print(f"  SKIPPED after {MAX_RETRIES} attempts")
    return None


def main():
    with open("seeds.csv", newline="", encoding="utf-8") as f:
        seeds = list(csv.DictReader(f))

    print(f"Processing {len(seeds)} seeds with Claude Code...\n")
    records = []

    for i, seed in enumerate(seeds, 1):
        print(f"[{i}/{len(seeds)}] {seed['company_name']}")
        record = extract_one(seed)
        if record:
            records.append(record)
            print(f"  OK (confidence: {record['confidence_score']})")
        print()

    with open("extracted.json", "w", encoding="utf-8") as f:
        json.dump(records, f, indent=2)

    print(f"Done. {len(records)}/{len(seeds)} records saved to extracted.json")


if __name__ == "__main__":
    main()

Run the extraction loop

Each of the 25 seeds gets sent to Claude Code one at a time. Claude researches, synthesizes, and structures the enrichment data that would take a human analyst hours per company.

Start the extraction by running this command:

python extract.py

The script prints progress for each seed: a counter like [1/25], the company name, and either OK (confidence: 0.85) on success or a retry message on failure.

How long does this take?

The full run takes roughly 5-10 minutes depending on Claude's response times. Each seed requires a round-trip to Claude Code, so the total time scales linearly with the number of seeds.

Watch the console for any retry messages. These show the validation-retry loop working as designed. A retry means Pydantic caught an invalid field and the script is feeding the error back to Claude for self-correction.

Seeing "Claude error" or timeouts?

Make sure Claude Code is authenticated. Run claude --version in a separate terminal to confirm it responds.
If you see Timed out after 120s on multiple seeds, your network connection may be slow. The script retries automatically, so occasional timeouts are fine.
If you see FileNotFoundError or command not found, claude is not on your PATH. Reinstall with npm install -g @anthropic-ai/claude-code and open a fresh terminal.
Why is my extraction script failing?

When the script finishes, you should see a final line like Done. 25/25 records saved to extracted.json. Some seeds may be skipped after 3 retries. That is expected behavior.

Inspect the enriched records

The extraction output lives in extracted.json. Each object in the array is a prospect that passed Pydantic validation. Time to spot-check that the data looks right.

Open extracted.json in your code editor.
Pick 2-3 records and check the following fields:

confidence_score is a float between 0.0 and 1.0.
employee_band uses one of the valid enum values: "1-10", "11-50", "51-200", or "201-500".
ai_readiness_signals is either null or a list of strings.
austin_zip is a 5-digit string.

What if some seeds were skipped?

The console output from the previous step tells you which seeds were skipped after 3 retries. The final count (e.g. 23/25) shows how many passed. A few skips are normal. The pipeline continues to work with however many records passed validation.

Your 25 prospects now have 12 validated fields each. Next up, you will load these records into DuckDB, run automated quality checks with Great Expectations, and launch a Streamlit dashboard that shows the full funnel and pipeline earnings.

Load into DuckDB, Validate, and Launch the Dashboard

In the previous step, Claude Code enriched your top 25 prospects with detailed signals and validated each one against the Pydantic schema. You now have an extracted.json file full of clean, structured prospect data.

Now it is time to load those records into your DuckDB warehouse, prove the data is trustworthy with automated quality checks, and present the results in a live dashboard. By the end of this step, you will have a Parquet export, an HTML data-quality report from Great Expectations, and a Streamlit dashboard showing your full prospect funnel.

In this step, get ready to:

Load validated prospect records into DuckDB and export a Parquet snapshot.
Run 6 automated data-quality checks with Great Expectations and generate an HTML proof report.
Build and launch a Streamlit dashboard with funnel metrics, pipeline earnings, and an interactive prospect table.

Write the warehouse loader script

The loader script reads your enriched prospects from extracted.json, creates a prospects table in DuckDB with columns matching your 12-field schema, inserts every record, and exports the table to Parquet for downstream tools.

Create a new file called load_warehouse.py in your signals folder.
Copy the following code into load_warehouse.py:

import duckdb
import json


def main():
    # Load the validated records from the extraction step
    with open("extracted.json", "r", encoding="utf-8") as f:
        records = json.load(f)

    print(f"Loading {len(records)} records into DuckDB...")

    con = duckdb.connect("signals.duckdb")

    # Create the prospects table with columns matching the 12-field schema
    con.execute("""
        CREATE OR REPLACE TABLE prospects (
            company_name VARCHAR NOT NULL,
            austin_zip VARCHAR(5) NOT NULL,
            naics_code VARCHAR(6) NOT NULL,
            employee_band VARCHAR NOT NULL,
            revenue_band_est VARCHAR NOT NULL,
            ai_readiness_signals VARCHAR[],
            hiring_last_90d BOOLEAN NOT NULL,
            compliance_flags VARCHAR[],
            decision_maker_title VARCHAR,
            preferred_channel VARCHAR NOT NULL,
            last_signal_date DATE NOT NULL,
            confidence_score DOUBLE NOT NULL
        )
    """)

    # Insert each validated record using parameterized queries
    for r in records:
        con.execute(
            """
            INSERT INTO prospects VALUES (
                ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?
            )
            """,
            [
                r["company_name"],
                r["austin_zip"],
                r["naics_code"],
                r["employee_band"],
                r["revenue_band_est"],
                r.get("ai_readiness_signals"),
                r["hiring_last_90d"],
                r.get("compliance_flags"),
                r.get("decision_maker_title"),
                r["preferred_channel"],
                r["last_signal_date"],
                r["confidence_score"],
            ],
        )

    # Verify the row count
    row_count = con.execute("SELECT COUNT(*) FROM prospects").fetchone()[0]
    print(f"Loaded {row_count} rows into prospects table")

    # Export the full table to Parquet for downstream tools
    con.execute("COPY prospects TO 'prospects.parquet' (FORMAT PARQUET)")
    print("Exported to prospects.parquet")

    # Preview the top 5 prospects by confidence score
    print("\nTop 5 by confidence:")
    top5 = con.execute("""
        SELECT company_name, revenue_band_est, confidence_score
        FROM prospects
        ORDER BY confidence_score DESC
        LIMIT 5
    """).df()
    print(top5.to_string(index=False))

    con.close()


if __name__ == "__main__":
    main()

What does this code do?

CREATE OR REPLACE TABLE prospects creates a fresh table every time you run the script. The VARCHAR[] type stores Python lists (like ai_readiness_signals) as array columns in DuckDB.
The for loop inserts each record using parameterized queries. DuckDB automatically converts Python lists to VARCHAR[] arrays.
COPY prospects TO 'prospects.parquet' (FORMAT PARQUET) exports the entire table to a columnar Parquet file. Great Expectations reads this file in the next substep.
The top-5 preview query confirms the data loaded correctly and shows which prospects have the highest confidence scores.

Save load_warehouse.py.
Run the loader by executing this command:

python load_warehouse.py

You should see output showing the number of rows loaded, a confirmation of the Parquet export, and a top-5 table sorted by confidence_score.

Seeing a "file not found" error?

Make sure extracted.json exists in your signals folder. This file is created by extract.py in the previous step.
Make sure you are running the command from inside the signals folder, not a parent directory.
Why is my load_warehouse.py failing?

Write the validation script and run the quality checks

Pydantic validated each record at insertion time. Now Great Expectations validates the warehouse as a whole.

Why two layers of validation?

This two-layer approach mirrors production data engineering. Pydantic catches bad records one at a time during ingest. Great Expectations checks the entire dataset after loading, looking for patterns like missing values across columns or unexpected distributions. If a record somehow slipped past Pydantic, GX catches it here.

Create a new file called validate.py in your signals folder.
Copy the following code into validate.py:

import great_expectations as gx
import pandas as pd


def main():
    # Read the Parquet snapshot into a Pandas DataFrame
    df = pd.read_parquet("prospects.parquet")
    print(f"Validating {len(df)} records...\n")

    # Create a File Data Context so GX can persist Data Docs to disk
    context = gx.get_context(mode="file", project_root_dir="gx_project")

    # Register a Pandas data source and dataframe asset
    data_source = context.data_sources.add_pandas("pandas")
    data_asset = data_source.add_dataframe_asset(name="prospects")
    batch_def = data_asset.add_batch_definition_whole_dataframe(
        "prospect_batch"
    )

    # Build an Expectation Suite with 6 data quality checks
    suite = context.suites.add(
        gx.core.expectation_suite.ExpectationSuite(
            name="prospect_quality"
        )
    )

    # Check 1: company_name is never null
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToNotBeNull(
            column="company_name"
        )
    )
    # Check 2: austin_zip matches 5-digit pattern
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToMatchRegex(
            column="austin_zip", regex=r"^\d{5}$"
        )
    )
    # Check 3: confidence_score is between 0.0 and 1.0
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="confidence_score", min_value=0.0, max_value=1.0
        )
    )
    # Check 4: employee_band uses valid enum values
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeInSet(
            column="employee_band",
            value_set=["1-10", "11-50", "51-200", "201-500"],
        )
    )
    # Check 5: revenue_band_est uses valid enum values
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeInSet(
            column="revenue_band_est",
            value_set=["<500K", "500K-1M", "1M-5M", "5M-20M", "20M+"],
        )
    )
    # Check 6: naics_code length is exactly 6
    suite.add_expectation(
        gx.expectations.ExpectColumnValueLengthsToBeBetween(
            column="naics_code", min_value=6, max_value=6
        )
    )

    # Wire together a Validation Definition and Checkpoint
    validation_def = context.validation_definitions.add(
        gx.core.validation_definition.ValidationDefinition(
            name="prospect_validation",
            data=batch_def,
            suite=suite,
        )
    )

    checkpoint = context.checkpoints.add(
        gx.checkpoint.checkpoint.Checkpoint(
            name="prospect_checkpoint",
            validation_definitions=[validation_def],
        )
    )

    # Run the checkpoint with the DataFrame as a batch parameter
    result = checkpoint.run(batch_parameters={"dataframe": df})
    print(result.describe())

    # Generate and open the HTML Data Docs report
    context.build_data_docs()
    context.open_data_docs()
    print("\nData Docs opened in browser.")


if __name__ == "__main__":
    main()

What does this code do?

gx.get_context(mode="file", project_root_dir="gx_project") creates a persistent File Data Context inside a gx_project/ directory. This is required for Data Docs to generate HTML files on disk.
The 6 expectations mirror the Pydantic schema constraints: not-null checks, regex patterns, numeric ranges, enum sets, and string lengths. If any record fails, GX flags it in the report.
The Validation Definition links the batch definition (your DataFrame) to the Expectation Suite. The Checkpoint runs all the checks in one call.
checkpoint.run(batch_parameters={"dataframe": df}) passes the in-memory DataFrame to GX at runtime.
context.build_data_docs() generates the HTML report. context.open_data_docs() opens it in your default browser.

Save validate.py.
Run the validation script by executing this command:

python validate.py

You should see a summary showing all 6 checks passed. Your browser opens an HTML Data Docs report with green checkmarks for each expectation.

Seeing a "prospects.parquet not found" error?

Make sure you ran python load_warehouse.py first. That script creates the Parquet file.
If you see a GX configuration error, try deleting the gx_project/ folder and running validate.py again. The script recreates it from scratch.
Why is my validate.py failing?

Write the dashboard and launch it

The dashboard brings everything together in one view. It shows the funnel that narrowed 55,000 companies down to 25 enriched prospects, estimates pipeline earnings, and lets you filter and sort the prospect table interactively.

Create a new file called dashboard.py in your signals folder.
Copy the following code into dashboard.py:

import streamlit as st
import duckdb
import json

# Configure the page layout and title
st.set_page_config(page_title="Signals: Austin SMB Prospects", layout="wide")
st.title("Signals: Austin SMB Prospect Intelligence")

# Load funnel metrics from the JSON file created by funnel.py
with open("funnel_metrics.json", "r") as f:
    funnel = json.load(f)

# Display the 3-stage funnel as metric cards
col1, col2, col3 = st.columns(3)
col1.metric("Universe", f"{funnel['universe_count']:,}")
col2.metric("ICP Match", f"{funnel['icp_count']:,}")
col3.metric("Enriched", f"{funnel['enriched_count']}")

st.divider()

# Query the prospects table from DuckDB in read-only mode
con = duckdb.connect("signals.duckdb", read_only=True)
df = con.execute("""
    SELECT company_name, austin_zip, naics_code,
           employee_band, revenue_band_est,
           ai_readiness_signals, confidence_score,
           decision_maker_title, preferred_channel,
           last_signal_date
    FROM prospects ORDER BY confidence_score DESC
""").df()
con.close()

# Sidebar filters for confidence score and revenue band
min_score = st.sidebar.slider("Min confidence", 0.0, 1.0, 0.5)
bands = st.sidebar.multiselect(
    "Revenue band",
    df["revenue_band_est"].unique().tolist(),
    default=df["revenue_band_est"].unique().tolist(),
)

# Apply the sidebar filters to the DataFrame
filtered = df[
    (df["confidence_score"] >= min_score)
    & (df["revenue_band_est"].isin(bands))
]

# Pipeline earnings estimate using solo AI consulting benchmarks
AVG_DEAL = 7500
CLOSE_RATE = 0.08
pipeline_value = len(filtered) * AVG_DEAL
projected_revenue = len(filtered) * AVG_DEAL * CLOSE_RATE

st.subheader("Pipeline Estimate")
p1, p2, p3 = st.columns(3)
p1.metric("Qualified Leads", len(filtered))
p2.metric("Pipeline Value", f"${pipeline_value:,.0f}")
p3.metric("Projected Revenue (8% close)", f"${projected_revenue:,.0f}")
st.caption(
    "Based on $7,500 avg deal size for solo AI consulting engagements "
    "and an 8% close rate on qualified, signal-enriched leads. "
    "Source: 2026 AI consulting market benchmarks."
)

st.divider()

# Display the interactive prospect table
st.subheader(f"Top Prospects ({len(filtered)})")
st.dataframe(filtered, use_container_width=True, hide_index=True)

What does this code do?

The funnel metrics section reads funnel_metrics.json (created by funnel.py in Step 4) and displays three metric cards showing how the universe narrowed from 55,000 to 25.
The sidebar slider and multiselect let you filter prospects by minimum confidence score and revenue band. Adjusting these updates the table and pipeline estimate in real time.
The pipeline estimate uses $7,500 average deal size and 8% close rate. These are conservative benchmarks for solo AI consulting engagements targeting SMBs.
st.dataframe() renders an interactive table with built-in sorting, searching, and column resizing. No extra configuration needed.

Save dashboard.py.
Launch the dashboard by running this command:

streamlit run dashboard.py

Your browser opens the Streamlit dashboard. You should see the funnel metrics at the top, the pipeline estimate in the middle, and a sortable prospect table at the bottom.

Dashboard not loading?

Make sure both signals.duckdb and funnel_metrics.json exist in your signals folder. The dashboard reads from both files.
If you see a port conflict, Streamlit defaults to port 8501. Close any other Streamlit apps running on that port.
Why is my Streamlit dashboard failing?

Awesome, I've got everything!

Great. Double check you have saved all three files: load_warehouse.py, validate.py, and dashboard.py.

I'd like to double check the full code

Here is the full code for all three files in this step.

import duckdb
import json


def main():
    with open("extracted.json", "r", encoding="utf-8") as f:
        records = json.load(f)

    print(f"Loading {len(records)} records into DuckDB...")

    con = duckdb.connect("signals.duckdb")

    con.execute("""
        CREATE OR REPLACE TABLE prospects (
            company_name VARCHAR NOT NULL,
            austin_zip VARCHAR(5) NOT NULL,
            naics_code VARCHAR(6) NOT NULL,
            employee_band VARCHAR NOT NULL,
            revenue_band_est VARCHAR NOT NULL,
            ai_readiness_signals VARCHAR[],
            hiring_last_90d BOOLEAN NOT NULL,
            compliance_flags VARCHAR[],
            decision_maker_title VARCHAR,
            preferred_channel VARCHAR NOT NULL,
            last_signal_date DATE NOT NULL,
            confidence_score DOUBLE NOT NULL
        )
    """)

    for r in records:
        con.execute(
            """
            INSERT INTO prospects VALUES (
                ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?
            )
            """,
            [
                r["company_name"],
                r["austin_zip"],
                r["naics_code"],
                r["employee_band"],
                r["revenue_band_est"],
                r.get("ai_readiness_signals"),
                r["hiring_last_90d"],
                r.get("compliance_flags"),
                r.get("decision_maker_title"),
                r["preferred_channel"],
                r["last_signal_date"],
                r["confidence_score"],
            ],
        )

    row_count = con.execute("SELECT COUNT(*) FROM prospects").fetchone()[0]
    print(f"Loaded {row_count} rows into prospects table")

    con.execute("COPY prospects TO 'prospects.parquet' (FORMAT PARQUET)")
    print("Exported to prospects.parquet")

    print("\nTop 5 by confidence:")
    top5 = con.execute("""
        SELECT company_name, revenue_band_est, confidence_score
        FROM prospects
        ORDER BY confidence_score DESC
        LIMIT 5
    """).df()
    print(top5.to_string(index=False))

    con.close()


if __name__ == "__main__":
    main()

import great_expectations as gx
import pandas as pd


def main():
    df = pd.read_parquet("prospects.parquet")
    print(f"Validating {len(df)} records...\n")

    context = gx.get_context(mode="file", project_root_dir="gx_project")

    data_source = context.data_sources.add_pandas("pandas")
    data_asset = data_source.add_dataframe_asset(name="prospects")
    batch_def = data_asset.add_batch_definition_whole_dataframe(
        "prospect_batch"
    )

    suite = context.suites.add(
        gx.core.expectation_suite.ExpectationSuite(
            name="prospect_quality"
        )
    )

    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToNotBeNull(
            column="company_name"
        )
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToMatchRegex(
            column="austin_zip", regex=r"^\d{5}$"
        )
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="confidence_score", min_value=0.0, max_value=1.0
        )
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeInSet(
            column="employee_band",
            value_set=["1-10", "11-50", "51-200", "201-500"],
        )
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeInSet(
            column="revenue_band_est",
            value_set=["<500K", "500K-1M", "1M-5M", "5M-20M", "20M+"],
        )
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValueLengthsToBeBetween(
            column="naics_code", min_value=6, max_value=6
        )
    )

    validation_def = context.validation_definitions.add(
        gx.core.validation_definition.ValidationDefinition(
            name="prospect_validation",
            data=batch_def,
            suite=suite,
        )
    )

    checkpoint = context.checkpoints.add(
        gx.checkpoint.checkpoint.Checkpoint(
            name="prospect_checkpoint",
            validation_definitions=[validation_def],
        )
    )

    result = checkpoint.run(batch_parameters={"dataframe": df})
    print(result.describe())

    context.build_data_docs()
    context.open_data_docs()
    print("\nData Docs opened in browser.")


if __name__ == "__main__":
    main()

import streamlit as st
import duckdb
import json

st.set_page_config(page_title="Signals: Austin SMB Prospects", layout="wide")
st.title("Signals: Austin SMB Prospect Intelligence")

with open("funnel_metrics.json", "r") as f:
    funnel = json.load(f)

col1, col2, col3 = st.columns(3)
col1.metric("Universe", f"{funnel['universe_count']:,}")
col2.metric("ICP Match", f"{funnel['icp_count']:,}")
col3.metric("Enriched", f"{funnel['enriched_count']}")

st.divider()

con = duckdb.connect("signals.duckdb", read_only=True)
df = con.execute("""
    SELECT company_name, austin_zip, naics_code,
           employee_band, revenue_band_est,
           ai_readiness_signals, confidence_score,
           decision_maker_title, preferred_channel,
           last_signal_date
    FROM prospects ORDER BY confidence_score DESC
""").df()
con.close()

min_score = st.sidebar.slider("Min confidence", 0.0, 1.0, 0.5)
bands = st.sidebar.multiselect(
    "Revenue band",
    df["revenue_band_est"].unique().tolist(),
    default=df["revenue_band_est"].unique().tolist(),
)

filtered = df[
    (df["confidence_score"] >= min_score)
    & (df["revenue_band_est"].isin(bands))
]

AVG_DEAL = 7500
CLOSE_RATE = 0.08
pipeline_value = len(filtered) * AVG_DEAL
projected_revenue = len(filtered) * AVG_DEAL * CLOSE_RATE

st.subheader("Pipeline Estimate")
p1, p2, p3 = st.columns(3)
p1.metric("Qualified Leads", len(filtered))
p2.metric("Pipeline Value", f"${pipeline_value:,.0f}")
p3.metric("Projected Revenue (8% close)", f"${projected_revenue:,.0f}")
st.caption(
    "Based on $7,500 avg deal size for solo AI consulting engagements "
    "and an 8% close rate on qualified, signal-enriched leads. "
    "Source: 2026 AI consulting market benchmarks."
)

st.divider()
st.subheader(f"Top Prospects ({len(filtered)})")
st.dataframe(filtered, use_container_width=True, hide_index=True)

Your prospect warehouse is fully loaded, validated, and visualized. You have a DuckDB database, a Parquet export, an HTML quality report, and a live dashboard. Next up, the Secret Mission challenges you to build a Monday Refresh script that keeps your warehouse fresh.

Secret mission

Add a Monday Refresh with Staleness Detection

Your warehouse is a snapshot of one extraction run. Real prospect intelligence needs to stay fresh. Build a refresh.py script that re-runs the extraction loop, diffs new records against existing warehouse rows, appends only changed data, and flags any prospect whose last signal is older than 14 days as stale.

Before you go

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally, so there are no ongoing cloud costs. The only external dependency is your Claude Max subscription, which you manage separately.

Resources you used:

signals project folder with all Python scripts (schema.py, generate_universe.py, funnel.py, extract.py, load_warehouse.py, validate.py, dashboard.py).
Python virtual environment (.venv folder inside the project directory).
signals.duckdb database file containing the universe and prospects tables.
Generated data files: seeds.csv, extracted.json, funnel_metrics.json, prospects.parquet.
gx_project/ directory containing the Great Expectations Data Context and Data Docs HTML report.
Claude Desktop Project named "Signals - Austin SMB Warehouse" with saved schema and taxonomy conversations.
Globally installed Claude Code (@anthropic-ai/claude-code npm package).

Keep everything running

No action needed. Choose this if you're still actively building or want to extend the project with the Secret Mission or your own prospect data.

Your DuckDB database, Parquet exports, and all scripts stay in the signals folder ready to use.
Your Claude Desktop Project retains all schema and taxonomy conversations as project knowledge.
To relaunch the dashboard later, activate your virtual environment and run streamlit run dashboard.py.

Pause - I'll come back to this later

Shut down any running processes to free up memory. Your files and database stay on disk so you can pick up where you left off.

Stop the Streamlit dashboard if it is still running. Press Ctrl+C in the terminal where it is active.
Deactivate the virtual environment by running deactivate in the terminal.
Keep the entire signals folder on disk. All your scripts, database, and exports will be there when you return.
To resume later, navigate to the signals folder, reactivate the virtual environment, and pick up from the step you left off.

Delete - I don't want to use this again

Remove all project resources and start fresh. This cannot be undone.

Stop the Streamlit dashboard if it is still running. Press Ctrl+C in the terminal.
Deactivate the virtual environment by running deactivate.
Delete the entire signals folder. This removes all scripts, the DuckDB database, Parquet exports, CSV files, JSON files, and the Great Expectations project directory.
In Claude Desktop, open the "Signals - Austin SMB Warehouse" Project and delete it if you no longer need the schema and taxonomy conversations.
Claude Code remains globally installed. To uninstall it, run:

npm uninstall -g @anthropic-ai/claude-code

Only uninstall Claude Code if you do not plan to use it for other projects.

Mission accomplished

Nice Work!

You just built a local prospect intelligence warehouse powered by Claude. From schema design to a live dashboard, you now have a working pipeline that turns 55,000 raw records into 25 enriched, validated, business-ready prospects.

Here's what you accomplished:

Generated a 55,000-row synthetic SMB universe in DuckDB with real Austin zip codes and NAICS codes, then applied an ICP funnel to score and select the top 25 prospects from ~150 matches.
Automated AI-powered enrichment by running Claude Code in headless mode with a Pydantic validation retry loop that feeds errors back to Claude for self-correction.
Launched a Streamlit dashboard showing the full prospect funnel and estimated pipeline earnings, backed by 6 automated Great Expectations data-quality checks with an HTML proof report.
Built a Monday Refresh script that diffs new extractions against existing warehouse rows, updates changed records, and flags any prospect with signals older than 14 days as stale.

Ready to quiz yourself?