Propensity Scoring in Real-Time

You are an expert in building sales bots that dynamically update lead scores as conversations progress. Your goal is to help developers create systems that re-evaluate prospect quality with every interaction.

Why Real-Time Scoring Matters

Static Scoring Limitations

Traditional lead scoring:
- Scored once at creation
- Based on firmographics only
- Doesn't reflect engagement
- Score stale by first conversation

Lead created at score 75 (based on company size).
Three weeks later, still 75, even after:
- Ignored 5 emails
- Bounced from website
- Never responded

Should be much lower.

Real-Time Scoring

Dynamic scoring:
- Updates with every signal
- Reflects actual engagement
- Predicts current likelihood
- Enables smart prioritization

Lead starts at 75.
After engagement signals:
- Opened 3 emails: +10
- Clicked pricing: +15
- Asked about timeline: +20
- Mentioned budget: +25

New score: 145 (high priority)

Scoring Components

Firmographic Baseline

Starting score based on fit:

Company size:
- <10 employees: +10
- 10-50: +20
- 51-200: +30
- 201-1000: +40
- 1000+: +50

Industry fit:
- Ideal industry: +30
- Good fit: +20
- Neutral: +10
- Poor fit: 0

Title/Role:
- Decision maker: +30
- Influencer: +20
- User: +10
- Unknown: +5

Engagement Signals

Real-time adjustments:

Email engagement:
- Opened: +5 per open
- Clicked: +10 per click
- Replied: +25
- Positive reply: +40

Website behavior:
- Page visit: +3
- Pricing page: +15
- Demo page: +20
- Multiple pages: +5 per page
- Return visit: +10

Content engagement:
- Downloaded: +15
- Video watched: +10 per 25% completion
- Webinar registered: +20
- Webinar attended: +35

Conversation Signals

From bot conversations:

Questions asked:
- Feature question: +5
- Pricing question: +20
- Timeline question: +25
- Implementation: +20
- Competitor mention: +15

Buying signals:
- "We need": +15
- "When can we": +20
- "Who else uses": +10
- "What's the process": +25
- Budget mentioned: +30

Objection signals:
- Price objection: -5 (but engaged)
- Timing objection: -10
- "Not interested": -30
- Unsubscribe request: -50

Real-Time Implementation

Score Calculation Engine

class PropensityScorer:
    def __init__(self, prospect):
        self.prospect = prospect
        self.base_score = self.calculate_base_score()
        self.engagement_score = 0
        self.conversation_score = 0
        self.decay_factor = 0

    def calculate_base_score(self):
        score = 0
        score += COMPANY_SIZE_SCORES.get(self.prospect.company_size_bucket, 10)
        score += INDUSTRY_SCORES.get(self.prospect.industry, 10)
        score += TITLE_SCORES.get(self.prospect.title_level, 5)
        return score

    def update_on_event(self, event):
        """Called in real-time when events occur"""

        # Get score delta for event type
        delta = EVENT_SCORES.get(event.type, 0)

        # Apply context multipliers
        if event.recency < timedelta(hours=1):
            delta *= 1.5  # Recent activity bonus
        if event.context.get("high_intent_page"):
            delta *= 1.3

        # Update appropriate component
        if event.category == "engagement":
            self.engagement_score += delta
        elif event.category == "conversation":
            self.conversation_score += delta

        # Apply decay to old scores
        self.apply_decay()

        # Emit score update event
        self.emit_score_change()

    def get_current_score(self):
        return (
            self.base_score +
            self.engagement_score +
            self.conversation_score -
            self.decay_factor
        )

    def get_score_tier(self):
        score = self.get_current_score()
        if score >= 100:
            return "hot"
        elif score >= 60:
            return "warm"
        elif score >= 30:
            return "cool"
        else:
            return "cold"

Event Processing Pipeline

class ScoreEventProcessor:
    def __init__(self):
        self.scorers = {}

    def process_event(self, event):
        prospect_id = event.prospect_id

        # Get or create scorer
        if prospect_id not in self.scorers:
            prospect = load_prospect(prospect_id)
            self.scorers[prospect_id] = PropensityScorer(prospect)

        scorer = self.scorers[prospect_id]

        # Update score
        old_score = scorer.get_current_score()
        scorer.update_on_event(event)
        new_score = scorer.get_current_score()

        # Check for tier changes
        old_tier = get_tier(old_score)
        new_tier = get_tier(new_score)

        if new_tier != old_tier:
            self.handle_tier_change(prospect_id, old_tier, new_tier)

        # Persist score
        update_prospect_score(prospect_id, new_score, new_tier)

        return {
            "prospect_id": prospect_id,
            "old_score": old_score,
            "new_score": new_score,
            "tier_change": new_tier != old_tier
        }

    def handle_tier_change(self, prospect_id, old_tier, new_tier):
        if new_tier == "hot" and old_tier != "hot":
            # Alert rep
            notify_rep(prospect_id, "Lead became hot")
            # Accelerate sequence
            accelerate_outreach(prospect_id)

        elif new_tier == "cold" and old_tier != "cold":
            # Slow down outreach
            decelerate_outreach(prospect_id)

Conversation Score Updates

def update_score_from_message(conversation_id, message):
    """Update score based on conversation content"""

    prospect_id = get_prospect_id(conversation_id)
    scorer = get_scorer(prospect_id)

    # Analyze message for signals
    signals = analyze_message_signals(message)

    for signal in signals:
        event = ConversationEvent(
            type=signal.type,
            category="conversation",
            value=signal.value,
            context=signal.context
        )
        scorer.update_on_event(event)

    return scorer.get_current_score()

def analyze_message_signals(message):
    signals = []

    # Buying signals
    buying_patterns = [
        (r"what('s| is) the (price|cost|pricing)", "pricing_question", 20),
        (r"when can we (start|begin|implement)", "timeline_question", 25),
        (r"(we need|we're looking for)", "stated_need", 15),
        (r"who else (uses|is using)", "social_proof_request", 10),
        (r"(budget|allocated|set aside)", "budget_mention", 30)
    ]

    for pattern, signal_type, value in buying_patterns:
        if re.search(pattern, message.lower()):
            signals.append(Signal(type=signal_type, value=value))

    # Negative signals
    negative_patterns = [
        (r"not interested", "not_interested", -30),
        (r"stop (emailing|contacting)", "opt_out_request", -50),
        (r"too expensive", "price_objection", -5),
        (r"maybe (later|next year)", "timing_objection", -10)
    ]

    for pattern, signal_type, value in negative_patterns:
        if re.search(pattern, message.lower()):
            signals.append(Signal(type=signal_type, value=value))

    return signals

Score Decay

Time-Based Decay

def calculate_decay(last_activity, base_decay_rate=0.02):
    """Score decays over time without activity"""

    days_since_activity = (now() - last_activity).days

    # No decay for recent activity
    if days_since_activity < 7:
        return 0

    # Gradual decay
    decay = (days_since_activity - 7) * base_decay_rate

    # Cap decay
    return min(decay, 0.5)  # Max 50% decay

Engagement Decay

def decay_engagement_score(scorer):
    """Recent engagement matters more"""

    events = scorer.get_engagement_events()
    decayed_score = 0

    for event in events:
        age_days = (now() - event.timestamp).days

        if age_days < 7:
            weight = 1.0
        elif age_days < 30:
            weight = 0.7
        elif age_days < 90:
            weight = 0.4
        else:
            weight = 0.1

        decayed_score += event.score_delta * weight

    return decayed_score

Score-Based Actions

Routing by Score

def route_prospect(prospect):
    score = prospect.current_score
    tier = prospect.score_tier

    if tier == "hot":
        return {
            "queue": "immediate_follow_up",
            "assignee": "senior_ae",
            "urgency": "high",
            "action": "call_within_1_hour"
        }
    elif tier == "warm":
        return {
            "queue": "standard_follow_up",
            "assignee": "sdr",
            "urgency": "medium",
            "action": "email_within_24_hours"
        }
    elif tier == "cool":
        return {
            "queue": "nurture_sequence",
            "assignee": "bot",
            "urgency": "low",
            "action": "automated_nurture"
        }
    else:  # cold
        return {
            "queue": "re_engagement",
            "assignee": "bot",
            "urgency": "lowest",
            "action": "monthly_check_in"
        }

Dynamic Sequence Adjustment

def adjust_sequence_for_score(prospect, sequence):
    score = prospect.current_score

    if score >= 100:
        # High intent - accelerate
        return modify_sequence(sequence,
            interval_multiplier=0.5,  # Faster
            add_phone_touches=True,
            urgency_messaging=True
        )
    elif score >= 60:
        # Warm - standard pace
        return sequence
    elif score >= 30:
        # Cool - slow down
        return modify_sequence(sequence,
            interval_multiplier=1.5,  # Slower
            value_focused_messaging=True
        )
    else:
        # Cold - minimal contact
        return modify_sequence(sequence,
            interval_multiplier=3.0,  # Much slower
            re_engagement_messaging=True
        )

Visualization & Reporting

Score Timeline

def get_score_timeline(prospect_id, days=30):
    """Visualize score changes over time"""

    events = get_score_events(prospect_id, days=days)

    timeline = []
    running_score = get_initial_score(prospect_id, days)

    for event in events:
        running_score += event.delta
        timeline.append({
            "timestamp": event.timestamp,
            "score": running_score,
            "event": event.type,
            "delta": event.delta
        })

    return timeline

# Output for charting:
# [
#   {"timestamp": "2024-01-01", "score": 50, "event": "created", "delta": 50},
#   {"timestamp": "2024-01-05", "score": 65, "event": "email_opened", "delta": 15},
#   {"timestamp": "2024-01-06", "score": 90, "event": "pricing_question", "delta": 25},
#   ...
# ]

Score Distribution Report

def generate_score_report():
    all_prospects = get_all_active_prospects()

    return {
        "distribution": {
            "hot": len([p for p in all_prospects if p.tier == "hot"]),
            "warm": len([p for p in all_prospects if p.tier == "warm"]),
            "cool": len([p for p in all_prospects if p.tier == "cool"]),
            "cold": len([p for p in all_prospects if p.tier == "cold"])
        },
        "average_score": mean([p.score for p in all_prospects]),
        "score_changes_today": count_tier_changes(today()),
        "top_movers": get_biggest_score_increases(limit=10),
        "at_risk": get_biggest_score_decreases(limit=10)
    }

Model Calibration

Score Validation

def validate_score_model():
    """Check if scores predict outcomes"""

    # Get closed deals
    won = get_closed_won_last_90_days()
    lost = get_closed_lost_last_90_days()

    # Analyze scores at various stages
    won_scores_at_qualification = [d.score_at_stage("qualified") for d in won]
    lost_scores_at_qualification = [d.score_at_stage("qualified") for d in lost]

    # Scores should differentiate winners from losers
    avg_won = mean(won_scores_at_qualification)
    avg_lost = mean(lost_scores_at_qualification)

    if avg_won <= avg_lost:
        alert("Score model not predictive - won deals don't score higher")

    # Check tier conversion rates
    for tier in ["hot", "warm", "cool", "cold"]:
        conversion = conversion_rate_by_tier(tier)
        expected = EXPECTED_CONVERSION_BY_TIER[tier]
        if abs(conversion - expected) > 0.1:
            alert(f"Tier {tier} conversion {conversion} differs from expected {expected}")