Score HubSpot leads based on firmographic and technographic fit using code

Prerequisites

Python 3.9+ or Node.js 18+
HubSpot private app token with crm.objects.contacts.read and crm.objects.contacts.write scopes
A custom HubSpot contact property for the fit score (e.g., icp_fit_score, number type, 0-100)
A cron scheduler (crontab, GitHub Actions, or a serverless function)

Step 1: Define your scoring model

Create a config that maps your ICP criteria to point values. This is the core of the recipe -- everything else is just plumbing.

SCORING_MODEL = {
    "company_size": {
        "ranges": [
            {"min": 200, "max": 2000, "points": 30},   # sweet spot
            {"min": 50,  "max": 199,  "points": 20},   # good fit
            {"min": 2000, "max": 10000, "points": 15},  # enterprise
        ],
        "default": 5,  # any known value
        "unknown": 0,
    },
    "industry": {
        "ideal": {"saas": 25, "technology": 25, "software": 25, "computer software": 25},
        "good": {"financial services": 15, "consulting": 15, "marketing": 15},
        "default": 5,
        "unknown": 0,
    },
    "seniority": {
        "c_suite": {"keywords": ["ceo", "cto", "cfo", "coo", "cmo", "cro", "chief"], "points": 30},
        "vp": {"keywords": ["vp", "vice president", "head of"], "points": 25},
        "director": {"keywords": ["director"], "points": 20},
        "manager": {"keywords": ["manager", "lead"], "points": 10},
    },
    "source": {
        "organic_search": 15,
        "direct_traffic": 12,
        "referrals": 10,
        "paid_search": 8,
        "social_media": 5,
        "email_marketing": 5,
    },
}

Step 2: Build the scoring function

def score_contact(contact):
    props = contact.get("properties", {})
    score = 0
 
    # Company size
    employees = int(props.get("numberofemployees") or 0)
    size_scored = False
    for r in SCORING_MODEL["company_size"]["ranges"]:
        if r["min"] <= employees <= r["max"]:
            score += r["points"]
            size_scored = True
            break
    if not size_scored and employees > 0:
        score += SCORING_MODEL["company_size"]["default"]
 
    # Industry
    industry = (props.get("industry") or "").lower()
    if industry in SCORING_MODEL["industry"]["ideal"]:
        score += SCORING_MODEL["industry"]["ideal"][industry]
    elif industry in SCORING_MODEL["industry"]["good"]:
        score += SCORING_MODEL["industry"]["good"][industry]
    elif industry:
        score += SCORING_MODEL["industry"]["default"]
 
    # Seniority
    title = (props.get("jobtitle") or "").lower()
    for level in SCORING_MODEL["seniority"].values():
        if any(kw in title for kw in level["keywords"]):
            score += level["points"]
            break
 
    # Lead source
    source = (props.get("hs_analytics_source") or "").lower()
    score += SCORING_MODEL["source"].get(source, 0)
 
    return min(score, 100)

Step 3: Fetch contacts and write scores

import requests
import os
 
HUBSPOT_TOKEN = os.environ["HUBSPOT_TOKEN"]
HEADERS = {"Authorization": f"Bearer {HUBSPOT_TOKEN}", "Content-Type": "application/json"}
 
def fetch_unscored_contacts():
    """Fetch contacts that don't have a fit score yet, or were recently updated."""
    all_contacts = []
    after = None
 
    while True:
        body = {
            "filterGroups": [{"filters": [
                {"propertyName": "icp_fit_score", "operator": "NOT_HAS_PROPERTY"}
            ]}],
            "properties": ["jobtitle", "company", "numberofemployees",
                           "industry", "hs_analytics_source"],
            "limit": 100,
        }
        if after:
            body["after"] = after
 
        resp = requests.post(
            "https://api.hubapi.com/crm/v3/objects/contacts/search",
            headers=HEADERS, json=body
        )
        resp.raise_for_status()
        data = resp.json()
        all_contacts.extend(data.get("results", []))
 
        after = data.get("paging", {}).get("next", {}).get("after")
        if not after:
            break
 
    return all_contacts
 
def update_score(contact_id, score):
    resp = requests.patch(
        f"https://api.hubapi.com/crm/v3/objects/contacts/{contact_id}",
        headers=HEADERS,
        json={"properties": {"icp_fit_score": str(score)}}
    )
    resp.raise_for_status()
 
if __name__ == "__main__":
    contacts = fetch_unscored_contacts()
    print(f"Scoring {len(contacts)} contacts...")
 
    for contact in contacts:
        score = score_contact(contact)
        update_score(contact["id"], score)
        name = f"{contact['properties'].get('firstname', '')} {contact['properties'].get('lastname', '')}".strip()
        print(f"  {name}: {score}/100")
 
    print("Done.")

HubSpot API rate limits

HubSpot allows 100 requests per 10 seconds for private apps. The batch update endpoint (/crm/v3/objects/contacts/batch/update) lets you update up to 100 contacts per request -- use it if you're scoring more than a few hundred contacts at a time.

Step 4: Schedule the scoring run

Run the script on a schedule to catch new and updated contacts:

# Score unscored contacts every hour
0 * * * * cd /path/to/project && python score_leads.py >> /var/log/lead-scoring.log 2>&1

For a batch re-score of all contacts (e.g., when you change your scoring model), modify the filter to remove the NOT_HAS_PROPERTY condition and process everyone.

Cost

Hosting: Free via cron on any server, or use GitHub Actions (2,000 free minutes/month)
No per-execution cost beyond hosting