""" Competitor Intelligence Engine ───────────────────────────────────────────────────────────────────────────── Problem: Strategy teams were making product decisions without knowing how their brand sentiment compared to competitors — or what competitor weaknesses they could exploit. Solution: Extract and analyze competitor mentions from the same corpus, building a comparative intelligence layer that surfaces switch signals, competitive advantage gaps, and opportunity areas. """ from __future__ import annotations import re import logging from typing import List, Dict, Optional from collections import defaultdict, Counter logger = logging.getLogger(__name__) # ─── Tracked entities ───────────────────────────────────────────────────── DEFAULT_COMPETITORS = { "RivalOne": ["rivalone", "rival one", "rival-one"], "CompeteX": ["competex", "compete x", "compete-x", "cx platform"], "AltStream": ["altstream", "alt stream", "alt-stream"], } SWITCH_SIGNALS = [ "switching from", "switched from", "migrating from", "moved from", "replaced", "replacing", "considering switching", "evaluating alternatives", "compared to", "better than", "worse than", "instead of", "vs ", "versus", ] ADVANTAGE_KEYWORDS = { "pricing": ["cheaper", "expensive", "pricing", "cost", "value", "affordable"], "features": ["feature", "capability", "function", "support", "integration"], "support": ["support", "customer service", "response", "help"], "ease_of_use": ["easier", "simpler", "intuitive", "complex", "confusing", "user-friendly"], "performance": ["faster", "slower", "reliable", "uptime", "performance", "stable"], "documentation": ["docs", "documentation", "guide", "tutorial", "onboarding"], } class CompetitorIntel: """ Competitor mention extraction and comparative intelligence. Scans a corpus for competitor mentions, extracts context, classifies switch direction, and identifies competitive gaps. """ def __init__(self, competitors: Optional[Dict[str, List[str]]] = None): self.competitors = competitors or DEFAULT_COMPETITORS # Pre-compile patterns for speed self._patterns = { name: re.compile( r"\b(" + "|".join(re.escape(alias) for alias in aliases) + r")\b", re.IGNORECASE, ) for name, aliases in self.competitors.items() } def extract_mentions(self, posts: List[Dict]) -> Dict[str, List[Dict]]: """Extract all competitor mentions from the corpus.""" mentions: Dict[str, List[Dict]] = defaultdict(list) for post in posts: text = post.get("text", "") for name, pattern in self._patterns.items(): if pattern.search(text): mentions[name].append({ "post_id": post.get("id", ""), "text": text, "timestamp": post.get("timestamp", ""), "source": post.get("source", ""), "sentiment": post.get("sentiment", post.get("true_label", "neutral")), "likes": post.get("likes", 0), }) return dict(mentions) def _detect_switch_direction(self, text: str, competitor: str) -> Optional[str]: """Detect if the post signals switching to or from the competitor.""" text_lower = text.lower() comp_lower = competitor.lower() for signal in SWITCH_SIGNALS: if signal in text_lower: signal_pos = text_lower.find(signal) comp_pos = text_lower.find(comp_lower) if comp_pos == -1: continue # If competitor comes after "switched FROM" → user left competitor if comp_pos > signal_pos and "from" in signal: return "switched_away_from_competitor" # If competitor mentioned in comparison context if "compared to" in signal or "vs" in signal: return "comparison" return "considering_switch" return None def _detect_advantage_gaps(self, text: str) -> List[str]: """Identify which dimensions are being compared.""" text_lower = text.lower() gaps = [] for dimension, keywords in ADVANTAGE_KEYWORDS.items(): if any(kw in text_lower for kw in keywords): gaps.append(dimension) return gaps def build_competitive_report( self, posts: List[Dict], brand_name: str = "TechFlow", brand_overall_sentiment: float = 0.72, ) -> Dict: """ Full competitive intelligence report. Returns per-competitor analysis plus brand positioning summary. """ mentions = self.extract_mentions(posts) competitor_profiles = {} for comp_name in self.competitors: comp_mentions = mentions.get(comp_name, []) # Sentiment breakdown of competitor mentions sent_dist = Counter(m["sentiment"] for m in comp_mentions) total_mentions = len(comp_mentions) # Switch signals switch_signals = [] advantage_gaps = Counter() for m in comp_mentions: direction = self._detect_switch_direction(m["text"], comp_name) if direction: switch_signals.append({"direction": direction, "text": m["text"][:150]}) gaps = self._detect_advantage_gaps(m["text"]) for gap in gaps: advantage_gaps[gap] += 1 switched_away = sum(1 for s in switch_signals if s["direction"] == "switched_away_from_competitor") # Rough sentiment score from mention context pos = sent_dist.get("positive", 0) neg = sent_dist.get("negative", 0) + sent_dist.get("crisis", 0) comp_sentiment = pos / max(total_mentions, 1) if total_mentions > 0 else 0.5 competitor_profiles[comp_name] = { "name": comp_name, "mention_count": total_mentions, "sentiment_score": round(comp_sentiment, 3), "sentiment_distribution": dict(sent_dist), "switch_signals": switch_signals[:5], "users_switched_away": switched_away, "top_comparison_dimensions": dict(advantage_gaps.most_common(4)), "top_mentions": sorted(comp_mentions, key=lambda x: x["likes"], reverse=True)[:3], } # Opportunity matrix: where competitors are weak, we can win opportunities = self._find_opportunities(competitor_profiles) return { "brand": brand_name, "brand_sentiment": brand_overall_sentiment, "competitors": competitor_profiles, "opportunities": opportunities, "total_competitive_mentions": sum(len(v) for v in mentions.values()), "market_share_of_voice": self._share_of_voice(mentions, len(posts)), } def _find_opportunities(self, profiles: Dict) -> List[Dict]: """Surface dimensions where competitors are underperforming.""" opportunities = [] for comp_name, profile in profiles.items(): if profile["sentiment_score"] < 0.55: opportunities.append({ "competitor": comp_name, "opportunity": f"{comp_name} shows weak sentiment ({profile['sentiment_score']:.0%}). " f"Users are looking for alternatives.", "action": "Create targeted comparison content highlighting your strengths.", "priority": "high" if profile["sentiment_score"] < 0.45 else "medium", }) for dim, count in profile.get("top_comparison_dimensions", {}).items(): if count >= 2: opportunities.append({ "competitor": comp_name, "opportunity": f"Users frequently compare {comp_name} on '{dim}' ({count} mentions).", "action": f"Strengthen your {dim} positioning in marketing and product.", "priority": "medium", }) return sorted(opportunities, key=lambda x: x["priority"] == "high", reverse=True)[:6] def _share_of_voice(self, mentions: Dict, total_posts: int) -> Dict: """Calculate share of voice for each competitor.""" if total_posts == 0: return {} return { name: round(100 * len(posts) / total_posts, 1) for name, posts in mentions.items() } # ─── Singleton ───────────────────────────────────────────────────────────── _intel: Optional[CompetitorIntel] = None def get_competitor_intel() -> CompetitorIntel: global _intel if _intel is None: _intel = CompetitorIntel() return _intel