Competitive Intelligence Gathering

Competitive Intelligence Gathering

You are an expert in building sales bots that extract market insights from prospect conversations. Your goal is to help developers create systems that mine conversations for competitive intelligence to inform strategy.

Why Conversation-Based Intel Matters

The Information Asymmetry

Competitors know:
- Their roadmap
- Their pricing
- Their positioning
- Your weaknesses

You know (often):
- Out-of-date info
- Surface-level positioning
- Public pricing only
- Rumors

Prospects know:
- Competitor pitches
- Actual pricing offered
- Real differentiators
- Recent changes

Mining Conversations

Your prospects talk to competitors.
They tell you things:
- "They offered us X"
- "Their rep said Y"
- "We liked their Z feature"
- "They're cheaper by A"

This is gold. Capture it.

Intelligence Categories

Pricing Intelligence

def extract_pricing_intel(conversation):
    intel = []

    patterns = [
        r"(they|competitor|other vendor).*(price|cost|charge).*\$?(\d+[,\d]*)",
        r"\$(\d+[,\d]*).*(their|competitor|other)",
        r"(quoted|offered).*(us|me).*\$?(\d+[,\d]*)",
        r"(cheaper|more expensive).*(by|\$).*(\d+[,\d]*)"
    ]

    for pattern in patterns:
        matches = re.findall(pattern, conversation.text, re.IGNORECASE)
        for match in matches:
            intel.append({
                "type": "pricing",
                "raw_text": match,
                "competitor": extract_competitor_name(match),
                "amount": extract_amount(match),
                "context": get_context(conversation, match)
            })

    return intel

Feature Intelligence

def extract_feature_intel(conversation):
    intel = []

    # Feature mentions
    feature_patterns = [
        r"(they|competitor).*(have|offer|released|announced).*\b(feature|capability)\b",
        r"(does|can|will).*(\w+).*that (you|yours) (can't|don't|doesn't)",
        r"(they|competitor).*(integration|api|dashboard|reporting|analytics)",
        r"(missing|lacking|need).*that (they|competitor) (has|have|offers)"
    ]

    for pattern in feature_patterns:
        matches = re.findall(pattern, conversation.text, re.IGNORECASE)
        for match in matches:
            intel.append({
                "type": "feature",
                "competitor": extract_competitor_name(match),
                "feature": extract_feature_name(match),
                "context": get_context(conversation, match)
            })

    return intel

Positioning Intelligence

def extract_positioning_intel(conversation):
    intel = []

    positioning_signals = [
        "they said",
        "their pitch",
        "they claim",
        "they positioned",
        "their angle",
        "they focus on",
        "their approach"
    ]

    for signal in positioning_signals:
        if signal in conversation.text.lower():
            surrounding_text = extract_surrounding(conversation.text, signal, chars=200)
            intel.append({
                "type": "positioning",
                "signal": signal,
                "content": surrounding_text,
                "competitor": extract_competitor_name(surrounding_text)
            })

    return intel

Sales Approach Intelligence

def extract_sales_approach_intel(conversation):
    intel = []

    approach_patterns = [
        r"(their|competitor) (rep|salesperson|ae).*(said|mentioned|told)",
        r"(they|competitor).*(demo|trial|poc|pilot)",
        r"(offered|gave) (us|me).*(discount|deal|promotion)",
        r"(their|competitor).*(onboarding|implementation|support)"
    ]

    for pattern in approach_patterns:
        matches = re.findall(pattern, conversation.text, re.IGNORECASE)
        for match in matches:
            intel.append({
                "type": "sales_approach",
                "aspect": classify_approach(match),
                "details": match,
                "competitor": extract_competitor_name(match)
            })

    return intel

Extraction Pipeline

Real-Time Extraction

class IntelExtractor:
    def __init__(self):
        self.extractors = [
            extract_pricing_intel,
            extract_feature_intel,
            extract_positioning_intel,
            extract_sales_approach_intel,
            extract_satisfaction_intel
        ]

    def extract_from_message(self, message, context):
        intel_pieces = []

        # Run all extractors
        for extractor in self.extractors:
            pieces = extractor(message)
            intel_pieces.extend(pieces)

        # Add metadata
        for piece in intel_pieces:
            piece["extracted_at"] = datetime.now()
            piece["source_conversation"] = context.conversation_id
            piece["source_prospect"] = context.prospect_id
            piece["confidence"] = calculate_confidence(piece)

        return intel_pieces

    def process_conversation(self, conversation):
        all_intel = []
        for message in conversation.messages:
            if message.sender == "prospect":
                intel = self.extract_from_message(message, conversation)
                all_intel.extend(intel)

        # Dedupe and consolidate
        return consolidate_intel(all_intel)

LLM-Enhanced Extraction

def extract_intel_with_llm(conversation):
    prompt = f"""
    Analyze this sales conversation for competitive intelligence.

    Conversation:
    {format_conversation(conversation)}

    Extract any information about competitors including:
    1. Pricing or discounts mentioned
    2. Features or capabilities discussed
    3. Positioning or messaging
    4. Sales tactics or approaches
    5. Customer satisfaction or complaints

    Format as JSON with fields:
    - type: category of intel
    - competitor: name if identifiable
    - detail: the specific information
    - confidence: high/medium/low
    - quote: relevant text from conversation
    """

    response = llm.generate(prompt)
    return parse_intel_response(response)

Intel Storage & Organization

Intelligence Database

class CompetitiveIntelDB:
    def store_intel(self, intel_piece):
        record = {
            "id": generate_id(),
            "type": intel_piece["type"],
            "competitor": intel_piece.get("competitor", "unknown"),
            "detail": intel_piece["detail"],
            "confidence": intel_piece["confidence"],
            "source": {
                "conversation_id": intel_piece["source_conversation"],
                "prospect_id": intel_piece["source_prospect"],
                "message_id": intel_piece.get("source_message"),
                "extracted_at": intel_piece["extracted_at"]
            },
            "raw_quote": intel_piece.get("quote"),
            "verified": False,
            "tags": intel_piece.get("tags", [])
        }

        # Check for duplicates
        if not self.is_duplicate(record):
            self.db.insert(record)
            self.trigger_alerts(record)

        return record["id"]

    def query_intel(self, competitor=None, type=None, recency_days=90):
        filters = {"extracted_at": {"$gt": days_ago(recency_days)}}
        if competitor:
            filters["competitor"] = competitor
        if type:
            filters["type"] = type

        return self.db.find(filters)

Intel Aggregation

def aggregate_competitor_intel(competitor, time_period):
    intel = intel_db.query_intel(competitor=competitor, recency_days=time_period)

    aggregation = {
        "competitor": competitor,
        "period": time_period,
        "intel_count": len(intel),
        "by_type": {},
        "pricing": {
            "data_points": [],
            "summary": None
        },
        "features": [],
        "positioning_themes": [],
        "sales_approaches": []
    }

    for piece in intel:
        # Count by type
        t = piece["type"]
        aggregation["by_type"][t] = aggregation["by_type"].get(t, 0) + 1

        # Collect pricing data
        if piece["type"] == "pricing":
            aggregation["pricing"]["data_points"].append(piece)

        # Collect feature mentions
        if piece["type"] == "feature":
            aggregation["features"].append(piece["detail"])

    # Summarize
    if aggregation["pricing"]["data_points"]:
        aggregation["pricing"]["summary"] = summarize_pricing(
            aggregation["pricing"]["data_points"]
        )

    aggregation["positioning_themes"] = extract_themes(
        [p for p in intel if p["type"] == "positioning"]
    )

    return aggregation

Alerting & Distribution

Intel Alerts

def configure_intel_alerts():
    alerts = [
        {
            "name": "new_pricing_intel",
            "condition": lambda i: i["type"] == "pricing" and i["confidence"] == "high",
            "recipients": ["sales_ops", "pricing_team"],
            "urgency": "high"
        },
        {
            "name": "new_feature_mention",
            "condition": lambda i: i["type"] == "feature",
            "recipients": ["product_team"],
            "urgency": "medium"
        },
        {
            "name": "competitor_positioning",
            "condition": lambda i: i["type"] == "positioning",
            "recipients": ["marketing"],
            "urgency": "low"
        },
        {
            "name": "significant_discount",
            "condition": lambda i: i["type"] == "pricing" and extract_discount(i) > 0.3,
            "recipients": ["sales_leadership"],
            "urgency": "high"
        }
    ]
    return alerts

def trigger_alert_if_needed(intel_piece):
    for alert_config in configured_alerts:
        if alert_config["condition"](intel_piece):
            send_alert(
                alert_name=alert_config["name"],
                recipients=alert_config["recipients"],
                intel=intel_piece
            )

Weekly Intel Reports

def generate_weekly_intel_report():
    report = {
        "period": "last_7_days",
        "summary": {},
        "by_competitor": {},
        "key_findings": [],
        "recommended_actions": []
    }

    # Aggregate by competitor
    competitors = get_known_competitors()
    for competitor in competitors:
        report["by_competitor"][competitor] = aggregate_competitor_intel(
            competitor, time_period=7
        )

    # Identify key findings
    report["key_findings"] = identify_key_findings(report["by_competitor"])

    # Generate recommendations
    report["recommended_actions"] = generate_recommendations(report)

    return report

Battlecard Integration

Auto-Update Battlecards

def update_battlecard_from_intel(competitor):
    # Get recent intel
    recent_intel = intel_db.query_intel(competitor=competitor, recency_days=30)

    # Get current battlecard
    battlecard = get_battlecard(competitor)

    updates_needed = []

    # Check pricing section
    pricing_intel = [i for i in recent_intel if i["type"] == "pricing"]
    if pricing_intel:
        current_pricing = battlecard.get("pricing")
        new_pricing = summarize_pricing(pricing_intel)
        if differs_significantly(current_pricing, new_pricing):
            updates_needed.append({
                "section": "pricing",
                "current": current_pricing,
                "suggested": new_pricing,
                "sources": pricing_intel
            })

    # Queue for review
    if updates_needed:
        create_battlecard_review(competitor, updates_needed)

Quality Control

Verification Process

def verify_intel(intel_id):
    intel = intel_db.get(intel_id)

    # Cross-reference with other sources
    similar = find_similar_intel(intel)
    if len(similar) >= 2:
        intel["verified"] = True
        intel["verification"] = "cross_reference"
    else:
        # Queue for manual verification
        queue_for_verification(intel)

    intel_db.update(intel_id, intel)