Building a Real-Time Twitter Monitoring System

Real-time Twitter monitoring enables businesses to track brand mentions, competitors, industry trends, and crisis situations as they unfold. Unlike periodic scraping, real-time systems process tweets within seconds of posting.

In this guide, we'll walk through the architecture and implementation of a production-grade Twitter monitoring system.

Use Cases for Real-Time Monitoring

Before diving into architecture, let's understand what real-time monitoring enables:

Brand Monitoring

• Track all mentions of your brand name
• Monitor product names and common misspellings
• Detect customer complaints immediately
• Identify brand advocates and influencers

Crisis Detection

• Alert on sudden spikes in negative sentiment
• Detect viral complaints before they escalate
• Monitor for security incidents or data breaches
• Track PR crisis situations in real-time

Competitive Intelligence

• Monitor competitor announcements
• Track competitor customer complaints
• Identify market opportunities
• Benchmark share of voice

Lead Generation

• Find people asking for product recommendations
• Identify users expressing purchase intent
• Monitor industry-specific hashtags
• Track event-related conversations

System Architecture

A robust real-time monitoring system consists of several layers:

Key Components

1. Data Collection Layer - Fetches tweets from Twitter
2. Message Queue - Buffers tweets for processing
3. Processing Pipeline - Filters, analyzes, enriches data
4. Storage Layer - Persists tweets and analytics
5. Alert System - Sends notifications on triggers
6. Dashboard - Visualizes data and trends

Data Collection Layer

The collection layer continuously fetches tweets matching your criteria:

import asyncio
import aiohttp
from datetime import datetime

class TwitterCollector:
    def __init__(self, api_key, keywords):
        self.api_key = api_key
        self.keywords = keywords
        self.base_url = "https://api.x-scraper.com/v1"
        self.last_ids = {}  # Track last seen tweet per keyword

    async def fetch_tweets(self, session, keyword):
        """Fetch recent tweets for a keyword."""
        params = {
            "q": keyword,
            "count": 100,
            "since_id": self.last_ids.get(keyword)
        }

        headers = {"Authorization": f"Bearer {self.api_key}"}

        async with session.get(
            f"{self.base_url}/search",
            params=params,
            headers=headers
        ) as response:
            if response.status == 200:
                data = await response.json()
                tweets = data.get("tweets", [])

                if tweets:
                    # Update last seen ID
                    self.last_ids[keyword] = tweets[0]["id"]

                return tweets
            return []

    async def collect_loop(self, callback, interval=10):
        """Continuous collection loop."""
        async with aiohttp.ClientSession() as session:
            while True:
                tasks = [
                    self.fetch_tweets(session, kw)
                    for kw in self.keywords
                ]

                results = await asyncio.gather(*tasks)

                for tweets in results:
                    for tweet in tweets:
                        await callback(tweet)

                await asyncio.sleep(interval)

# Usage
async def process_tweet(tweet):
    print(f"New tweet: @{tweet['username']}: {tweet['text'][:50]}...")

collector = TwitterCollector(
    api_key="your-key",
    keywords=["your brand", "competitor", "#industry"]
)

asyncio.run(collector.collect_loop(process_tweet))

Polling vs Streaming

For most use cases, polling every 5-10 seconds provides sufficient real-time capability with simpler implementation.

Processing Pipeline

Once collected, tweets flow through processing stages:

import redis
import json
from transformers import pipeline

class TweetProcessor:
    def __init__(self):
        self.redis = redis.Redis(host='localhost', port=6379)
        self.sentiment_analyzer = pipeline(
            "sentiment-analysis",
            model="cardiffnlp/twitter-roberta-base-sentiment"
        )

    def enqueue(self, tweet):
        """Add tweet to processing queue."""
        self.redis.lpush("tweets:raw", json.dumps(tweet))

    def process_batch(self, batch_size=100):
        """Process a batch of tweets."""
        tweets = []

        for _ in range(batch_size):
            data = self.redis.rpop("tweets:raw")
            if not data:
                break
            tweets.append(json.loads(data))

        if not tweets:
            return []

        # Analyze sentiment
        texts = [t["text"] for t in tweets]
        sentiments = self.sentiment_analyzer(texts)

        # Enrich tweets with sentiment
        for tweet, sentiment in zip(tweets, sentiments):
            tweet["sentiment"] = sentiment["label"]
            tweet["sentiment_score"] = sentiment["score"]
            tweet["processed_at"] = datetime.utcnow().isoformat()

        return tweets

    def filter_tweet(self, tweet, rules):
        """Apply filtering rules."""
        # Minimum engagement filter
        if rules.get("min_likes"):
            if tweet.get("likes", 0) < rules["min_likes"]:
                return False

        # Verified only filter
        if rules.get("verified_only"):
            if not tweet.get("user_verified"):
                return False

        # Language filter
        if rules.get("language"):
            if tweet.get("lang") != rules["language"]:
                return False

        return True

Sentiment Analysis

Sentiment analysis categorizes tweets as positive, negative, or neutral:

• Rule-based: Fast but limited accuracy
• ML-based: More accurate, requires more resources
• API-based: Easy to implement, per-call cost

# Simple rule-based sentiment
def quick_sentiment(text):
    positive = ["love", "great", "amazing", "awesome", "best"]
    negative = ["hate", "terrible", "awful", "worst", "bad"]

    text_lower = text.lower()
    pos_count = sum(1 for w in positive if w in text_lower)
    neg_count = sum(1 for w in negative if w in text_lower)

    if pos_count > neg_count:
        return "positive"
    elif neg_count > pos_count:
        return "negative"
    return "neutral"

Storage Solutions

Choose storage based on your query patterns:

PostgreSQL - Structured Queries

-- Schema for tweet storage
CREATE TABLE tweets (
    id BIGINT PRIMARY KEY,
    text TEXT NOT NULL,
    username VARCHAR(50),
    created_at TIMESTAMP,
    likes INTEGER DEFAULT 0,
    retweets INTEGER DEFAULT 0,
    sentiment VARCHAR(20),
    sentiment_score FLOAT,
    keyword VARCHAR(100),
    processed_at TIMESTAMP DEFAULT NOW()
);

-- Index for time-based queries
CREATE INDEX idx_tweets_created ON tweets(created_at DESC);

-- Index for keyword filtering
CREATE INDEX idx_tweets_keyword ON tweets(keyword);

Elasticsearch - Full-Text Search

# Elasticsearch mapping
mapping = {
    "mappings": {
        "properties": {
            "text": {"type": "text", "analyzer": "english"},
            "username": {"type": "keyword"},
            "created_at": {"type": "date"},
            "sentiment": {"type": "keyword"},
            "likes": {"type": "integer"},
            "location": {"type": "geo_point"}
        }
    }
}

Time-Series Database - Metrics

# InfluxDB for metrics
from influxdb_client import InfluxDBClient, Point

def store_metrics(bucket, tweet):
    point = Point("twitter_metrics") \
        .tag("keyword", tweet["keyword"]) \
        .tag("sentiment", tweet["sentiment"]) \
        .field("likes", tweet["likes"]) \
        .field("retweets", tweet["retweets"]) \
        .time(tweet["created_at"])

    write_api.write(bucket=bucket, record=point)

Alert System

Alerts notify you when specific conditions are met:

import smtplib
from slack_sdk import WebClient

class AlertManager:
    def __init__(self, config):
        self.slack = WebClient(token=config["slack_token"])
        self.email_config = config["email"]
        self.rules = config["rules"]

    def check_alerts(self, tweet):
        """Check if tweet triggers any alerts."""
        alerts = []

        for rule in self.rules:
            if self._matches_rule(tweet, rule):
                alerts.append({
                    "rule": rule["name"],
                    "tweet": tweet,
                    "priority": rule.get("priority", "normal")
                })

        return alerts

    def _matches_rule(self, tweet, rule):
        """Check if tweet matches alert rule."""
        # High engagement alert
        if rule["type"] == "high_engagement":
            threshold = rule.get("threshold", 100)
            return tweet.get("likes", 0) >= threshold

        # Negative sentiment spike
        if rule["type"] == "negative_sentiment":
            return tweet.get("sentiment") == "negative"

        # Keyword match
        if rule["type"] == "keyword":
            keywords = rule.get("keywords", [])
            text = tweet.get("text", "").lower()
            return any(kw.lower() in text for kw in keywords)

        # Influencer mention
        if rule["type"] == "influencer":
            min_followers = rule.get("min_followers", 10000)
            return tweet.get("user_followers", 0) >= min_followers

        return False

    def send_alert(self, alert):
        """Send alert via configured channels."""
        if alert["priority"] == "high":
            self._send_slack(alert)
            self._send_email(alert)
        else:
            self._send_slack(alert)

    def _send_slack(self, alert):
        """Send Slack notification."""
        tweet = alert["tweet"]
        self.slack.chat_postMessage(
            channel="#twitter-alerts",
            text=f"*{alert['rule']}*\n"
                 f"@{tweet['username']}: {tweet['text']}\n"
                 f"Likes: {tweet.get('likes', 0)} | "
                 f"Sentiment: {tweet.get('sentiment', 'unknown')}"
        )

    def _send_email(self, alert):
        """Send email notification."""
        # Email implementation
        pass

Common Alert Types

• Volume spike: Mentions increase 3x above baseline
• Sentiment shift: Negative sentiment exceeds threshold
• Viral tweet: High engagement on brand mention
• Influencer mention: Large account mentions you
• Crisis keywords: Words like "outage", "hack", "lawsuit"

Building a Dashboard

Visualize your monitoring data in real-time:

Key Metrics to Display

• Volume over time: Tweet count by hour/day
• Sentiment distribution: Positive/negative/neutral split
• Top influencers: Most impactful accounts
• Trending topics: Rising keywords and hashtags
• Engagement metrics: Total likes, retweets, replies

Simple Flask Dashboard

from flask import Flask, render_template, jsonify
from datetime import datetime, timedelta

app = Flask(__name__)

@app.route('/api/metrics')
def get_metrics():
    """Return dashboard metrics."""
    # Get data from your storage
    now = datetime.utcnow()
    hour_ago = now - timedelta(hours=1)

    return jsonify({
        "tweet_count": db.count_tweets(since=hour_ago),
        "sentiment": {
            "positive": db.count_sentiment("positive", since=hour_ago),
            "negative": db.count_sentiment("negative", since=hour_ago),
            "neutral": db.count_sentiment("neutral", since=hour_ago)
        },
        "top_tweets": db.get_top_tweets(limit=10),
        "volume_history": db.get_hourly_volume(hours=24)
    })

@app.route('/api/stream')
def stream_tweets():
    """Server-sent events for real-time updates."""
    def generate():
        pubsub = redis.pubsub()
        pubsub.subscribe("tweets:processed")

        for message in pubsub.listen():
            if message["type"] == "message":
                yield f"data: {message['data']}\n\n"

    return Response(generate(), mimetype='text/event-stream')

Scaling Considerations

Horizontal Scaling

• Collection: Run multiple collectors for different keywords
• Processing: Add worker processes as volume grows
• Storage: Shard by time or keyword

Performance Optimization

• Batch processing: Process tweets in batches, not individually
• Connection pooling: Reuse database connections
• Caching: Cache frequent queries
• Async I/O: Use async libraries for network calls

Cost Management

• Prioritize keywords: Monitor what matters most
• Adjust polling frequency: Slower for low-priority terms
• Data retention: Archive old data to cheaper storage
• Sample high-volume: Don't store every tweet for viral topics

Conclusion

A real-time Twitter monitoring system enables you to:

• Respond to customer issues within minutes
• Detect PR crises before they escalate
• Identify sales opportunities in real-time
• Track competitor activities as they happen

The architecture we covered—collection, queueing, processing, storage, and alerting—provides a solid foundation that can scale with your needs.

For teams that want real-time Twitter data without building and maintaining infrastructure, X (Twitter) Scraper API provides an easy-to-use solution with built-in real-time capabilities.