AI-Powered Upsell Detection: Finding Hidden Revenue with Claude Code [2026]
· 10 min read
Your best sales opportunity isn't that cold lead you're chasing. It's the customer you already have.
The data is clear:
- Acquiring a new customer costs 5-25x more than retaining one
- Existing customers are 50% more likely to try new products
- They spend 31% more than new customers
- Upsell/cross-sell drives 70-95% of revenue for mature SaaS companies
Yet most sales teams treat expansion like an afterthought. CSMs occasionally notice opportunities. AEs occasionally ask. Most revenue sits on the table.
With Claude Code, you can systematically scan every customer signal—usage data, support tickets, call transcripts, email threads—and surface upsell opportunities before customers even know they need more.
This guide shows you exactly how to build that system.
The Signals You're Missing
Every customer generates expansion signals. Most companies capture maybe 10% of them:
Usage Signals
- Hitting plan limits: Storage at 90%, seats maxed out, API calls capped
- Feature adoption velocity: Using new features faster than average
- Power user emergence: Individual users far exceeding normal usage
- Cross-team adoption: New departments/teams starting to use product
- Usage growth trajectory: 20%+ MoM growth = expansion candidate
Support Signals
- Feature requests: Asking for capabilities in higher tiers
- Integration questions: "Can we connect this to X?" = deeper investment
- Workaround requests: Hacking around limitations = need upgrade
- Multi-team tickets: Multiple departments engaging = spreading use
- "Is there a way to..." questions: Implicit feature discovery
Conversation Signals
- Budget mentions: "We're planning next year's budget..."
- Team growth: "We're hiring 5 more SDRs..."
- Initiative mentions: "We're launching a new product line..."
- Pain escalation: Problems getting worse = urgency to solve
- Competitive mentions: "We're also looking at X for this use case"
Behavioral Signals
- Login frequency increases: More engagement = more value extracted
- New user invites: Account expansion happening organically
- Report/export usage: Executives looking at data = strategic value
- API usage growth: Technical integration deepening
Building the Detection System
Here's how to build comprehensive upsell detection with Claude Code:
Architecture Overview
┌─────────────────────────────────────────────────────────────┐
│ Data Sources │
│ Usage metrics, Support tickets, Calls, Emails, CRM │
└─────────────────┬───────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ Signal Extraction │
│ Claude analyzes each source for expansion indicators │
└─────────────────┬───────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ Signal Aggregation │
│ Combine signals into customer-level expansion score │
└─────────────────┬───────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ Opportunity Identification │
│ Match signals to specific products/tiers/add-ons │
└─────────────────┬───────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ Alert & Action │
│ Notify CSM/AE with context, talking points, timing │
└─────────────────────────────────────────────────────────────┘
Step 1: Usage Signal Detection
# usage_signal_detector.py
from anthropic import Anthropic
import json
from datetime import datetime, timedelta
class UsageSignalDetector:
def __init__(self):
self.client = Anthropic()
def analyze_usage_patterns(self, customer_id: str) -> dict:
"""Analyze customer usage data for expansion signals."""
# Pull usage data from your analytics system
usage_data = self.get_usage_data(customer_id)
plan_limits = self.get_plan_limits(customer_id)
analysis_prompt = f"""
Analyze this customer's usage data for expansion/upsell signals.
## Current Plan
{json.dumps(plan_limits, indent=2)}
## Usage Data (Last 90 Days)
{json.dumps(usage_data, indent=2)}
## Identify
1. Features approaching or exceeding limits
2. Usage growth trends (week-over-week, month-over-month)
3. Power users who might need advanced features
4. New use cases emerging (based on feature adoption)
5. Comparison to similar customers who upgraded
## Output
Return JSON:
{{
"expansion_score": 0-100,
"primary_signal": "most significant expansion indicator",
"signals": [
{{
"type": "limit_approaching|growth_trend|power_user|new_use_case",
"description": "what we observed",
"urgency": "high|medium|low",
"relevant_upsell": "which product/tier addresses this"
}}
],
"recommended_timing": "immediate|this_month|next_quarter",
"talking_points": ["point 1", "point 2", "point 3"]
}}
"""
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1500,
messages=[{"role": "user", "content": analysis_prompt}]
)
return json.loads(response.content[0].text)
def get_usage_data(self, customer_id: str) -> dict:
"""Pull usage metrics from your analytics system."""
# Implementation depends on your stack (Mixpanel, Amplitude, custom)
pass
def detect_limit_approach(self, customer_id: str, threshold: float = 0.8) -> list:
"""Find customers approaching plan limits."""
usage = self.get_usage_data(customer_id)
limits = self.get_plan_limits(customer_id)
approaching_limits = []
for metric, current_value in usage.items():
if metric in limits:
limit = limits[metric]
utilization = current_value / limit if limit > 0 else 0
if utilization >= threshold:
approaching_limits.append({
"metric": metric,
"current": current_value,
"limit": limit,
"utilization": f"{utilization:.1%}",
"days_to_limit": self.estimate_days_to_limit(
customer_id, metric, current_value, limit
)
})
return approaching_limits
Step 2: Support Ticket Analysis
# support_signal_detector.py
class SupportSignalDetector:
def __init__(self):
self.client = Anthropic()
def analyze_support_tickets(self, customer_id: str, days: int = 90) -> dict:
"""Analyze support tickets for expansion signals."""
tickets = self.get_recent_tickets(customer_id, days)
if not tickets:
return {"expansion_score": 0, "signals": []}
analysis_prompt = f"""
Analyze these support tickets for upsell/expansion signals.
## Tickets
{json.dumps(tickets, indent=2)}
## Signal Types to Look For
1. Feature requests that exist in higher tiers
2. Questions about integrations/APIs
3. Workaround requests (working around limitations)
4. Multi-department/team involvement
5. Scaling concerns or performance issues
6. Questions about enterprise features (SSO, audit logs, etc.)
## Output
Return JSON:
{{
"expansion_score": 0-100,
"signals": [
{{
"ticket_id": "reference",
"signal_type": "feature_request|integration|workaround|scaling|enterprise",
"quote": "relevant excerpt from ticket",
"implied_need": "what this suggests they need",
"relevant_upsell": "product/tier that addresses this"
}}
],
"pattern_summary": "overall pattern observed",
"recommended_action": "specific next step"
}}
"""
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1500,
messages=[{"role": "user", "content": analysis_prompt}]
)
return json.loads(response.content[0].text)
def get_recent_tickets(self, customer_id: str, days: int) -> list:
"""Pull tickets from Zendesk, Intercom, Freshdesk, etc."""
# Implementation depends on your support platform
pass
Step 3: Conversation Analysis
# conversation_signal_detector.py
class ConversationSignalDetector:
def __init__(self):
self.client = Anthropic()
def analyze_call_transcripts(self, customer_id: str) -> dict:
"""Analyze recent call transcripts for expansion signals."""
transcripts = self.get_recent_calls(customer_id)
analysis_prompt = f"""
Analyze these customer call transcripts for expansion/upsell signals.
## Transcripts
{json.dumps(transcripts, indent=2)}
## Signals to Identify
1. Budget discussions or planning mentions
2. Team growth or hiring plans
3. New initiatives or product launches
4. Pain points getting worse
5. Competitive mentions or evaluations
6. Questions about capabilities they don't have
7. Success stories they're sharing (ready to expand)
8. Executive involvement (strategic importance)
## Output
Return JSON:
{{
"expansion_score": 0-100,
"signals": [
{{
"call_date": "date",
"signal_type": "budget|growth|initiative|pain|competitive|capability|success|executive",
"quote": "exact words that indicate the signal",
"context": "surrounding context",
"urgency": "high|medium|low",
"recommended_response": "how to follow up"
}}
],
"key_stakeholders": ["names and roles identified"],
"best_timing": "when to approach based on what was said",
"risk_factors": ["any concerns or objections mentioned"]
}}
"""
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=2000,
messages=[{"role": "user", "content": analysis_prompt}]
)
return json.loads(response.content[0].text)
def analyze_email_threads(self, customer_id: str) -> dict:
"""Analyze email communications for expansion signals."""
emails = self.get_customer_emails(customer_id)
analysis_prompt = f"""
Analyze these email threads with the customer for expansion signals.
## Emails
{json.dumps(emails, indent=2)}
## Look For
1. Requests that indicate growing needs
2. Forward-looking statements about plans
3. Questions about pricing or contracts
4. Mentions of other teams or use cases
5. Positive feedback (good time to expand)
6. Frustrations that upgrades would solve
## Output
Return JSON with signals found, quotes, and recommendations.
"""
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1500,
messages=[{"role": "user", "content": analysis_prompt}]
)
return json.loads(response.content[0].text)
Step 4: Signal Aggregation
# expansion_scorer.py
class ExpansionScorer:
def __init__(self):
self.client = Anthropic()
self.usage_detector = UsageSignalDetector()
self.support_detector = SupportSignalDetector()
self.conversation_detector = ConversationSignalDetector()
def calculate_expansion_score(self, customer_id: str) -> dict:
"""Aggregate all signals into a comprehensive expansion score."""
# Gather signals from all sources
usage_signals = self.usage_detector.analyze_usage_patterns(customer_id)
support_signals = self.support_detector.analyze_support_tickets(customer_id)
conversation_signals = self.conversation_detector.analyze_call_transcripts(customer_id)
# Get customer context
customer_data = self.get_customer_data(customer_id)
aggregation_prompt = f"""
Calculate a comprehensive expansion opportunity score for this customer.
## Customer Profile
{json.dumps(customer_data, indent=2)}
## Usage Signals
{json.dumps(usage_signals, indent=2)}
## Support Signals
{json.dumps(support_signals, indent=2)}
## Conversation Signals
{json.dumps(conversation_signals, indent=2)}
## Scoring Framework
- Base score from usage patterns (0-30 points)
- Support signal strength (0-25 points)
- Conversation buying signals (0-25 points)
- Account health/NPS bonus (0-10 points)
- Timing factors (0-10 points)
## Output
Return JSON:
{{
"total_score": 0-100,
"score_breakdown": {{
"usage": 0-30,
"support": 0-25,
"conversation": 0-25,
"health": 0-10,
"timing": 0-10
}},
"confidence": "high|medium|low",
"primary_opportunity": {{
"product": "specific upsell opportunity",
"estimated_value": "ARR increase",
"win_probability": "percentage",
"key_signal": "strongest indicator"
}},
"secondary_opportunities": [...],
"recommended_approach": "how to start the conversation",
"ideal_timing": "when to reach out",
"key_stakeholder": "who to engage",
"talking_points": ["personalized points based on their signals"],
"risk_factors": ["potential objections to prepare for"]
}}
"""
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=2000,
messages=[{"role": "user", "content": aggregation_prompt}]
)
return json.loads(response.content[0].text)
def generate_expansion_playbook(self, customer_id: str) -> dict:
"""Generate a complete playbook for approaching this expansion opportunity."""
score_data = self.calculate_expansion_score(customer_id)
customer_data = self.get_customer_data(customer_id)
playbook_prompt = f"""
Create a detailed expansion playbook for this opportunity.
## Customer
{json.dumps(customer_data, indent=2)}
## Expansion Analysis
{json.dumps(score_data, indent=2)}
## Generate
1. Email template to open the conversation
2. Call script with discovery questions
3. Objection handling for likely concerns
4. ROI calculation framework
5. Timeline and next steps
Make it specific to this customer's signals, not generic.
"""
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=3000,
messages=[{"role": "user", "content": playbook_prompt}]
)
return {
"score_data": score_data,
"playbook": response.content[0].text
}
Step 5: Automated Alerting
# expansion_alerts.py
class ExpansionAlertSystem:
def __init__(self):
self.scorer = ExpansionScorer()
self.alert_threshold = 70 # Score threshold for alerts
def run_daily_scan(self):
"""Scan all customers for expansion opportunities."""
customers = self.get_all_active_customers()
opportunities = []
for customer in customers:
try:
score_data = self.scorer.calculate_expansion_score(customer['id'])
if score_data['total_score'] >= self.alert_threshold:
opportunities.append({
"customer": customer,
"score_data": score_data,
"playbook": self.scorer.generate_expansion_playbook(customer['id'])
})
except Exception as e:
self.log_error(f"Failed to score {customer['id']}: {e}")
# Sort by score and notify
opportunities.sort(key=lambda x: x['score_data']['total_score'], reverse=True)
self.send_alerts(opportunities)
return opportunities
def send_alerts(self, opportunities: list):
"""Send alerts to CSMs/AEs for each opportunity."""
for opp in opportunities:
customer = opp['customer']
score = opp['score_data']
# Format alert message
alert = f"""
🎯 *EXPANSION OPPORTUNITY*
🏢 {customer['name']}
📊 Score: {score['total_score']}/100
💰 Potential: {score['primary_opportunity']['estimated_value']}
🎲 Win Probability: {score['primary_opportunity']['win_probability']}
*Top Signal:* {score['primary_opportunity']['key_signal']}
*Recommended Approach:*
{score['recommended_approach']}
*Talking Points:*
{chr(10).join('• ' + p for p in score['talking_points'][:3])}
*Best Timing:* {score['ideal_timing']}
[View Full Playbook]
"""
# Route to appropriate owner
owner = self.get_account_owner(customer['id'])
self.send_notification(owner, alert, opp['playbook'])
def send_notification(self, owner: dict, alert: str, playbook: dict):
"""Send via appropriate channel (Slack, email, WhatsApp)."""
# Implementation depends on your notification setup
pass
Running It with OpenClaw
Set up automated scanning with OpenClaw cron:
# openclaw.yaml
cron:
jobs:
# Daily expansion opportunity scan
- name: "expansion-scan"
schedule: "0 7 * * 1-5" # 7am weekdays
payload:
kind: systemEvent
text: |
Run the expansion opportunity scan:
1. Score all active customers
2. Identify those above threshold
3. Generate playbooks for top 10
4. Alert account owners via Slack
5. Log results to dashboard
sessionTarget: main
# Weekly expansion summary
- name: "expansion-summary"
schedule: "0 9 * * 1" # Monday 9am
payload:
kind: systemEvent
text: |
Generate weekly expansion pipeline report:
- Total expansion opportunities identified
- Opportunities by segment
- This week's priority accounts
- Progress on last week's opportunities
Send to #sales-leadership channel.
sessionTarget: main
What This Looks Like in Practice
Without AI:
- CSM notices customer asked about SSO (3 months ago)
- CSM mentions to AE at quarterly review
- AE reaches out... customer already evaluated competitors
With AI:
- Day 1: Customer asks about SSO in support ticket
- Day 1: AI flags enterprise feature interest, scores opportunity
- Day 2: AE receives alert with context and talking points
- Day 3: AE reaches out, references their specific need
- Day 14: Deal closed—customer upgraded to Enterprise
The difference:
- Opportunity identified 90 days faster
- Context preserved (no telephone game)
- Personalized outreach (not generic "time for a review?")
- Higher conversion (right message, right time)
Measuring Success
Track these metrics:
def calculate_expansion_metrics(period_days: int = 30) -> dict:
return {
# Detection metrics
"opportunities_identified": 47,
"avg_score_of_opportunities": 78,
"false_positive_rate": 0.12, # Flagged but no real opportunity
# Conversion metrics
"opportunities_acted_on": 38,
"opportunities_converted": 24,
"conversion_rate": 0.63,
# Revenue metrics
"expansion_arr_identified": 890000,
"expansion_arr_closed": 540000,
"avg_deal_size": 22500,
# Efficiency metrics
"time_from_signal_to_outreach_days": 1.8,
"time_from_outreach_to_close_days": 18,
# Comparison (AI vs manual)
"ai_identified_conversion_rate": 0.63,
"manually_identified_conversion_rate": 0.34
}
Getting Started
Week 1: Connect Data Sources
- Connect usage analytics (Mixpanel, Amplitude, custom)
- Connect support system (Zendesk, Intercom)
- Connect CRM (HubSpot, Salesforce)
- Set up Claude Code API access
Week 2: Build Core Detection
- Implement usage signal detection
- Implement support ticket analysis
- Test on 10 known expansion accounts (validate accuracy)
Week 3: Add Conversation Analysis
- Connect call recording system (Gong, Chorus)
- Implement email analysis
- Build aggregation scoring
Week 4: Deploy & Iterate
- Set up daily scans via OpenClaw cron
- Route alerts to CSMs/AEs
- Track outcomes, tune thresholds
- Add conversation analysis as transcripts become available
Ready to find the revenue hiding in your customer base? See how MarketBetter identifies expansion opportunities automatically →