sentiment-trend-tracking

Sentiment Trend Tracking

You are an expert in building sales bots that monitor whether overall campaign tone is improving or degrading. Your goal is to help developers create systems that track sentiment trends across conversations to identify systemic issues.

Why Trend Tracking Matters

The Slow Decline Problem

Individual conversations look fine.
Aggregate tells a different story:

Week 1: Average sentiment 0.65
Week 2: Average sentiment 0.60
Week 3: Average sentiment 0.52
Week 4: Average sentiment 0.45

Something's wrong that individual
conversation review wouldn't catch.

With Trend Tracking

Alert: "Campaign sentiment declined 25%
over 4 weeks. Top drivers:
- Message fatigue (same template)
- Increased competitor mentions
- Pricing complaints up 40%"

Early warning enables correction.

Sentiment Measurement

Per-Message Sentiment

def analyze_message_sentiment(message):
    # Use NLP model for sentiment
    sentiment_result = sentiment_model.analyze(message.text)

    return {
        "message_id": message.id,
        "timestamp": message.timestamp,
        "sentiment_score": sentiment_result.score,  # -1 to 1
        "sentiment_label": sentiment_result.label,  # negative/neutral/positive
        "confidence": sentiment_result.confidence,
        "aspects": extract_aspect_sentiments(message)
    }

def extract_aspect_sentiments(message):
    """Break down sentiment by topic"""
    aspects = {}

    topics = ["pricing", "product", "support", "timeline", "competitor"]
    for topic in topics:
        if mentions_topic(message.text, topic):
            aspects[topic] = analyze_topic_sentiment(message.text, topic)

    return aspects

Conversation Sentiment

def calculate_conversation_sentiment(conversation):
    message_sentiments = [
        analyze_message_sentiment(m)
        for m in conversation.messages
        if m.sender == "prospect"
    ]

    if not message_sentiments:
        return None

    return {
        "conversation_id": conversation.id,
        "average_sentiment": mean([m["sentiment_score"] for m in message_sentiments]),
        "sentiment_trend": calculate_trend(message_sentiments),
        "starting_sentiment": message_sentiments[0]["sentiment_score"],
        "ending_sentiment": message_sentiments[-1]["sentiment_score"],
        "delta": message_sentiments[-1]["sentiment_score"] - message_sentiments[0]["sentiment_score"],
        "message_count": len(message_sentiments),
        "aspect_sentiments": aggregate_aspects(message_sentiments)
    }

Trend Calculation

Time Series Analysis

def calculate_sentiment_trend(time_period, granularity="day"):
    """Calculate sentiment trend over time"""

    conversations = get_conversations(time_period)

    # Group by time bucket
    buckets = defaultdict(list)
    for conv in conversations:
        bucket = get_time_bucket(conv.timestamp, granularity)
        sentiment = calculate_conversation_sentiment(conv)
        if sentiment:
            buckets[bucket].append(sentiment["average_sentiment"])

    # Calculate average per bucket
    trend_data = []
    for bucket in sorted(buckets.keys()):
        trend_data.append({
            "period": bucket,
            "average_sentiment": mean(buckets[bucket]),
            "conversation_count": len(buckets[bucket]),
            "std_dev": stdev(buckets[bucket]) if len(buckets[bucket]) > 1 else 0
        })

    # Calculate trend direction
    if len(trend_data) >= 2:
        sentiments = [d["average_sentiment"] for d in trend_data]
        trend_direction = calculate_trend_direction(sentiments)
        trend_strength = calculate_trend_strength(sentiments)
    else:
        trend_direction = "insufficient_data"
        trend_strength = 0

    return {
        "data": trend_data,
        "direction": trend_direction,  # improving, declining, stable
        "strength": trend_strength,    # 0-1
        "period": time_period,
        "granularity": granularity
    }

def calculate_trend_direction(values):
    if len(values) < 2:
        return "insufficient_data"

    # Simple linear regression
    slope = calculate_slope(values)

    if slope > 0.02:
        return "improving"
    elif slope < -0.02:
        return "declining"
    else:
        return "stable"

Segment-Level Trends

def calculate_segment_trends(time_period):
    """Compare trends across segments"""

    segments = ["smb", "mid_market", "enterprise", "inbound", "outbound"]
    segment_trends = {}

    for segment in segments:
        conversations = get_conversations(time_period, segment=segment)
        trend = calculate_sentiment_trend_from_conversations(conversations)
        segment_trends[segment] = trend

    # Compare segments
    comparison = {
        "segment_trends": segment_trends,
        "best_performing": max(segment_trends, key=lambda s: segment_trends[s]["average"]),
        "worst_performing": min(segment_trends, key=lambda s: segment_trends[s]["average"]),
        "most_improved": max(segment_trends, key=lambda s: segment_trends[s].get("change", 0)),
        "most_declined": min(segment_trends, key=lambda s: segment_trends[s].get("change", 0))
    }

    return comparison

Anomaly Detection

Statistical Anomaly Detection

def detect_sentiment_anomalies(trend_data, sensitivity=2):
    """Detect unusual sentiment changes"""

    anomalies = []

    # Calculate rolling statistics
    window_size = 7  # days
    for i, point in enumerate(trend_data):
        if i < window_size:
            continue

        window = trend_data[i-window_size:i]
        window_values = [d["average_sentiment"] for d in window]

        rolling_mean = mean(window_values)
        rolling_std = stdev(window_values) if len(window_values) > 1 else 0.1

        # Check if current point is anomalous
        z_score = (point["average_sentiment"] - rolling_mean) / rolling_std if rolling_std > 0 else 0

        if abs(z_score) > sensitivity:
            anomalies.append({
                "period": point["period"],
                "value": point["average_sentiment"],
                "expected": rolling_mean,
                "z_score": z_score,
                "type": "positive_anomaly" if z_score > 0 else "negative_anomaly"
            })

    return anomalies

Pattern-Based Detection

def detect_sentiment_patterns(trend_data):
    """Detect concerning patterns"""

    patterns = []

    # Consecutive decline
    consecutive_decline = 0
    for i in range(1, len(trend_data)):
        if trend_data[i]["average_sentiment"] < trend_data[i-1]["average_sentiment"]:
            consecutive_decline += 1
        else:
            consecutive_decline = 0

        if consecutive_decline >= 3:
            patterns.append({
                "type": "consecutive_decline",
                "duration": consecutive_decline,
                "severity": "high" if consecutive_decline >= 5 else "medium"
            })

    # Sudden drop
    for i in range(1, len(trend_data)):
        change = trend_data[i]["average_sentiment"] - trend_data[i-1]["average_sentiment"]
        if change < -0.15:
            patterns.append({
                "type": "sudden_drop",
                "period": trend_data[i]["period"],
                "magnitude": abs(change),
                "severity": "high"
            })

    # Volatility spike
    recent_volatility = calculate_volatility(trend_data[-7:])
    historical_volatility = calculate_volatility(trend_data[:-7])
    if recent_volatility > historical_volatility * 2:
        patterns.append({
            "type": "volatility_spike",
            "recent": recent_volatility,
            "historical": historical_volatility,
            "severity": "medium"
        })

    return patterns

Root Cause Analysis

Correlation Analysis

def analyze_sentiment_drivers(time_period):
    """Identify what's driving sentiment changes"""

    conversations = get_conversations(time_period)

    drivers = {
        "template_correlation": {},
        "segment_correlation": {},
        "topic_correlation": {},
        "rep_correlation": {}
    }

    # Analyze by template
    by_template = defaultdict(list)
    for conv in conversations:
        template_id = conv.template_id
        sentiment = calculate_conversation_sentiment(conv)
        if sentiment:
            by_template[template_id].append(sentiment["average_sentiment"])

    for template_id, sentiments in by_template.items():
        drivers["template_correlation"][template_id] = {
            "avg_sentiment": mean(sentiments),
            "count": len(sentiments)
        }

    # Identify lowest performing templates
    drivers["worst_templates"] = sorted(
        drivers["template_correlation"].items(),
        key=lambda x: x[1]["avg_sentiment"]
    )[:5]

    return drivers

Topic Sentiment Analysis

def analyze_topic_sentiments(time_period):
    """Track sentiment by topic over time"""

    topics = ["pricing", "features", "support", "implementation", "competitor"]
    topic_trends = {}

    for topic in topics:
        conversations = get_conversations_mentioning(topic, time_period)
        sentiments = []

        for conv in conversations:
            topic_sentiment = extract_topic_sentiment(conv, topic)
            if topic_sentiment:
                sentiments.append({
                    "date": conv.timestamp,
                    "sentiment": topic_sentiment
                })

        if sentiments:
            topic_trends[topic] = {
                "average": mean([s["sentiment"] for s in sentiments]),
                "trend": calculate_trend_direction([s["sentiment"] for s in sentiments]),
                "count": len(sentiments)
            }

    return topic_trends

Alerting

Trend Alerts

def configure_sentiment_alerts():
    return [
        {
            "name": "declining_trend",
            "condition": lambda t: t["direction"] == "declining" and t["strength"] > 0.3,
            "message": "Sentiment showing significant decline",
            "severity": "high"
        },
        {
            "name": "sudden_drop",
            "condition": lambda t: any(a["type"] == "sudden_drop" for a in t.get("anomalies", [])),
            "message": "Sudden sentiment drop detected",
            "severity": "high"
        },
        {
            "name": "segment_divergence",
            "condition": lambda t: segment_divergence(t) > 0.2,
            "message": "Significant sentiment divergence between segments",
            "severity": "medium"
        },
        {
            "name": "template_outlier",
            "condition": lambda t: has_template_outlier(t),
            "message": "Template with significantly lower sentiment",
            "severity": "medium"
        }
    ]

def check_and_alert(trend_analysis):
    for alert_config in configured_alerts:
        if alert_config["condition"](trend_analysis):
            send_alert(
                name=alert_config["name"],
                message=alert_config["message"],
                severity=alert_config["severity"],
                data=trend_analysis
            )

Visualization

Trend Dashboard

def generate_sentiment_dashboard(time_period):
    return {
        "overall_trend": calculate_sentiment_trend(time_period),
        "segment_comparison": calculate_segment_trends(time_period),
        "anomalies": detect_sentiment_anomalies(trend_data),
        "patterns": detect_sentiment_patterns(trend_data),
        "drivers": analyze_sentiment_drivers(time_period),
        "topic_breakdown": analyze_topic_sentiments(time_period),
        "alerts": get_active_alerts(),
        "recommendations": generate_recommendations()
    }

Metrics

Trend Health Metrics

Track:
- 7-day sentiment average
- 30-day sentiment trend
- Segment sentiment variance
- Template sentiment variance
- Topic sentiment breakdown

Alert thresholds:
- >10% decline over 7 days
- >20% decline over 30 days
- >0.3 variance between segments