Real-Time Reddit Monitoring: Building Streaming Data Systems for Live Insights

import praw
import json
import logging
from datetime import datetime
from typing import Generator, Dict, Any
from dataclasses import dataclass, asdict
import redis

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

@dataclass
class RedditEvent:
    """Standardized Reddit event structure"""
    event_id: str
    event_type: str  # 'post' or 'comment'
    subreddit: str
    author: str
    title: str
    content: str
    url: str
    score: int
    created_utc: float
    parent_id: str = None

    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)

    def to_json(self) -> str:
        return json.dumps(self.to_dict())


class RedditStreamCollector:
    """Real-time Reddit stream collector with error handling"""

    def __init__(self,
                 client_id: str,
                 client_secret: str,
                 user_agent: str,
                 redis_url: str = 'redis://localhost:6379'):

        self.reddit = praw.Reddit(
            client_id=client_id,
            client_secret=client_secret,
            user_agent=user_agent
        )

        # Redis for message queue
        self.redis_client = redis.from_url(redis_url)
        self.stream_key = 'reddit:stream:events'

        # Statistics
        self.stats = {
            'posts_processed': 0,
            'comments_processed': 0,
            'errors': 0,
            'start_time': datetime.now()
        }

    def _create_post_event(self, submission) -> RedditEvent:
        """Convert PRAW submission to event"""
        return RedditEvent(
            event_id=submission.id,
            event_type='post',
            subreddit=submission.subreddit.display_name,
            author=str(submission.author) if submission.author else '[deleted]',
            title=submission.title,
            content=submission.selftext[:5000],  # Truncate long posts
            url=submission.url,
            score=submission.score,
            created_utc=submission.created_utc
        )

    def _create_comment_event(self, comment) -> RedditEvent:
        """Convert PRAW comment to event"""
        return RedditEvent(
            event_id=comment.id,
            event_type='comment',
            subreddit=comment.subreddit.display_name,
            author=str(comment.author) if comment.author else '[deleted]',
            title='',
            content=comment.body[:5000],
            url=f'https://reddit.com{comment.permalink}',
            score=comment.score,
            created_utc=comment.created_utc,
            parent_id=comment.parent_id
        )

    def stream_subreddits(self,
                           subreddits: list[str],
                           include_comments: bool = True) -> Generator:
        """
        Stream posts and comments from multiple subreddits

        Args:
            subreddits: List of subreddit names to monitor
            include_comments: Whether to also stream comments
        """

        subreddit_str = '+'.join(subreddits)
        subreddit = self.reddit.subreddit(subreddit_str)

        logger.info(f"Starting stream for: {subreddit_str}")

        if include_comments:
            # Use stream.all() for both posts and comments
            for item in subreddit.stream.submissions(skip_existing=True):
                try:
                    event = self._create_post_event(item)
                    self.stats['posts_processed'] += 1
                    yield event
                except Exception as e:
                    self.stats['errors'] += 1
                    logger.error(f"Error processing post: {e}")
        else:
            for submission in subreddit.stream.submissions(skip_existing=True):
                try:
                    event = self._create_post_event(submission)
                    self.stats['posts_processed'] += 1
                    yield event
                except Exception as e:
                    self.stats['errors'] += 1
                    logger.error(f"Error processing post: {e}")

    def publish_to_redis(self, event: RedditEvent):
        """Publish event to Redis stream"""
        try:
            self.redis_client.xadd(
                self.stream_key,
                {'data': event.to_json()},
                maxlen=100000  # Keep last 100k events
            )
        except Exception as e:
            logger.error(f"Redis publish error: {e}")

    def get_stats(self) -> Dict:
        """Get current streaming statistics"""
        runtime = (datetime.now() - self.stats['start_time']).seconds
        total = self.stats['posts_processed'] + self.stats['comments_processed']

        return {
            **self.stats,
            'runtime_seconds': runtime,
            'events_per_second': total / max(runtime, 1)
        }

import threading
import queue
from concurrent.futures import ThreadPoolExecutor
import signal
import sys

class MultiStreamManager:
    """Manage multiple Reddit streams concurrently"""

    def __init__(self, collector: RedditStreamCollector, max_workers: int = 4):
        self.collector = collector
        self.max_workers = max_workers
        self.event_queue = queue.Queue(maxsize=10000)
        self.running = True
        self.threads = []

        # Handle graceful shutdown
        signal.signal(signal.SIGINT, self._signal_handler)
        signal.signal(signal.SIGTERM, self._signal_handler)

    def _signal_handler(self, signum, frame):
        """Handle shutdown signals"""
        logger.info("Shutdown signal received...")
        self.running = False

    def _stream_worker(self, subreddits: list[str], stream_type: str):
        """Worker thread for streaming"""
        subreddit_str = '+'.join(subreddits)
        subreddit = self.collector.reddit.subreddit(subreddit_str)

        logger.info(f"Starting {stream_type} stream for: {subreddits}")

        try:
            if stream_type == 'posts':
                stream = subreddit.stream.submissions(skip_existing=True)
            else:
                stream = subreddit.stream.comments(skip_existing=True)

            for item in stream:
                if not self.running:
                    break

                try:
                    if stream_type == 'posts':
                        event = self.collector._create_post_event(item)
                    else:
                        event = self.collector._create_comment_event(item)

                    # Non-blocking put with timeout
                    self.event_queue.put(event, timeout=1.0)

                except queue.Full:
                    logger.warning("Event queue full, dropping event")
                except Exception as e:
                    logger.error(f"Error in {stream_type} stream: {e}")

        except Exception as e:
            logger.error(f"Stream worker failed: {e}")

    def _processor_worker(self, processor_fn):
        """Worker to process events from queue"""
        while self.running or not self.event_queue.empty():
            try:
                event = self.event_queue.get(timeout=1.0)
                processor_fn(event)
                self.event_queue.task_done()
            except queue.Empty:
                continue
            except Exception as e:
                logger.error(f"Processor error: {e}")

    def start(self,
              subreddit_groups: list[list[str]],
              processor_fn,
              include_comments: bool = True,
              num_processors: int = 2):
        """
        Start multi-threaded streaming

        Args:
            subreddit_groups: List of subreddit lists to stream
            processor_fn: Function to process each event
            include_comments: Whether to stream comments
            num_processors: Number of processor threads
        """

        # Start post streams
        for subreddits in subreddit_groups:
            t = threading.Thread(
                target=self._stream_worker,
                args=(subreddits, 'posts'),
                daemon=True
            )
            t.start()
            self.threads.append(t)

        # Start comment streams if enabled
        if include_comments:
            for subreddits in subreddit_groups:
                t = threading.Thread(
                    target=self._stream_worker,
                    args=(subreddits, 'comments'),
                    daemon=True
                )
                t.start()
                self.threads.append(t)

        # Start processor threads
        for _ in range(num_processors):
            t = threading.Thread(
                target=self._processor_worker,
                args=(processor_fn,),
                daemon=True
            )
            t.start()
            self.threads.append(t)

        logger.info(f"Started {len(self.threads)} threads")

        # Keep main thread alive
        try:
            while self.running:
                threading.Event().wait(1.0)
        except KeyboardInterrupt:
            self.running = False

        logger.info("Waiting for threads to finish...")
        for t in self.threads:
            t.join(timeout=5.0)

from abc import ABC, abstractmethod
from typing import Optional, List, Callable
import re
from transformers import pipeline
import numpy as np

class PipelineStage(ABC):
    """Base class for pipeline stages"""

    @abstractmethod
    def process(self, event: RedditEvent) -> Optional[RedditEvent]:
        """Process event, return None to filter out"""
        pass


class KeywordFilter(PipelineStage):
    """Filter events by keywords"""

    def __init__(self,
                 keywords: List[str],
                 exclude_keywords: List[str] = None,
                 case_sensitive: bool = False):
        self.keywords = keywords
        self.exclude_keywords = exclude_keywords or []
        self.case_sensitive = case_sensitive

        # Compile patterns for efficiency
        flags = 0 if case_sensitive else re.IGNORECASE
        self.include_pattern = re.compile(
            '|'.join(re.escape(kw) for kw in keywords),
            flags
        )
        if exclude_keywords:
            self.exclude_pattern = re.compile(
                '|'.join(re.escape(kw) for kw in exclude_keywords),
                flags
            )
        else:
            self.exclude_pattern = None

    def process(self, event: RedditEvent) -> Optional[RedditEvent]:
        text = f"{event.title} {event.content}"

        # Check exclusions first
        if self.exclude_pattern and self.exclude_pattern.search(text):
            return None

        # Check inclusions
        if self.include_pattern.search(text):
            return event

        return None


class SentimentAnalyzer(PipelineStage):
    """Add sentiment scores to events"""

    def __init__(self, model_name: str = 'cardiffnlp/twitter-roberta-base-sentiment'):
        self.analyzer = pipeline(
            'sentiment-analysis',
            model=model_name,
            device=0 if torch.cuda.is_available() else -1
        )
        self.label_map = {
            'LABEL_0': 'negative',
            'LABEL_1': 'neutral',
            'LABEL_2': 'positive'
        }

    def process(self, event: RedditEvent) -> RedditEvent:
        text = f"{event.title} {event.content}"[:512]  # Model limit

        try:
            result = self.analyzer(text)[0]
            event.sentiment = self.label_map.get(result['label'], result['label'])
            event.sentiment_score = result['score']
        except Exception as e:
            event.sentiment = 'unknown'
            event.sentiment_score = 0.0

        return event


class UrgencyClassifier(PipelineStage):
    """Classify event urgency for alerting"""

    URGENT_PATTERNS = [
        r'\b(urgent|emergency|asap|immediately|critical|crisis)\b',
        r'\b(lawsuit|legal action|sue|attorney)\b',
        r'\b(scam|fraud|hack|breach|vulnerability)\b',
        r'\b(recall|warning|danger|unsafe)\b',
    ]

    HIGH_PATTERNS = [
        r'\b(complaint|issue|problem|broken|failed)\b',
        r'\b(disappointed|frustrated|angry|upset)\b',
        r'\b(refund|return|cancel)\b',
    ]

    def __init__(self):
        self.urgent_regex = re.compile(
            '|'.join(self.URGENT_PATTERNS),
            re.IGNORECASE
        )
        self.high_regex = re.compile(
            '|'.join(self.HIGH_PATTERNS),
            re.IGNORECASE
        )

    def process(self, event: RedditEvent) -> RedditEvent:
        text = f"{event.title} {event.content}"

        if self.urgent_regex.search(text):
            event.urgency = 'urgent'
            event.urgency_score = 1.0
        elif self.high_regex.search(text):
            event.urgency = 'high'
            event.urgency_score = 0.7
        else:
            event.urgency = 'normal'
            event.urgency_score = 0.3

        return event


class EventPipeline:
    """Composable event processing pipeline"""

    def __init__(self):
        self.stages: List[PipelineStage] = []
        self.processed_count = 0
        self.filtered_count = 0

    def add_stage(self, stage: PipelineStage) -> 'EventPipeline':
        """Add processing stage (fluent API)"""
        self.stages.append(stage)
        return self

    def process(self, event: RedditEvent) -> Optional[RedditEvent]:
        """Run event through all pipeline stages"""

        for stage in self.stages:
            if event is None:
                self.filtered_count += 1
                return None

            event = stage.process(event)

        if event:
            self.processed_count += 1

        return event


# Example usage
def create_brand_monitoring_pipeline(brand_name: str) -> EventPipeline:
    """Create a pipeline for brand monitoring"""

    return (EventPipeline()
        .add_stage(KeywordFilter(
            keywords=[brand_name, f"@{brand_name}", f"#{brand_name}"],
            exclude_keywords=['spam', 'giveaway', 'follow for follow']
        ))
        .add_stage(SentimentAnalyzer())
        .add_stage(UrgencyClassifier())
    )

import asyncio
import aiohttp
from typing import Dict, List, Optional
from datetime import datetime, timedelta
from collections import defaultdict
from dataclasses import dataclass
from enum import Enum

class AlertPriority(Enum):
    LOW = 1
    MEDIUM = 2
    HIGH = 3
    URGENT = 4

@dataclass
class Alert:
    event: RedditEvent
    priority: AlertPriority
    rule_name: str
    created_at: datetime
    message: str

class AlertRule:
    """Define conditions for generating alerts"""

    def __init__(self,
                 name: str,
                 condition: Callable[[RedditEvent], bool],
                 priority: AlertPriority,
                 message_template: str,
                 cooldown_minutes: int = 5):
        self.name = name
        self.condition = condition
        self.priority = priority
        self.message_template = message_template
        self.cooldown = timedelta(minutes=cooldown_minutes)
        self.last_triggered: Dict[str, datetime] = {}

    def check(self, event: RedditEvent) -> Optional[Alert]:
        """Check if event triggers this rule"""

        # Check cooldown per subreddit
        cooldown_key = f"{self.name}:{event.subreddit}"
        if cooldown_key in self.last_triggered:
            if datetime.now() - self.last_triggered[cooldown_key] < self.cooldown:
                return None

        if self.condition(event):
            self.last_triggered[cooldown_key] = datetime.now()

            message = self.message_template.format(
                subreddit=event.subreddit,
                author=event.author,
                title=event.title,
                url=event.url,
                sentiment=getattr(event, 'sentiment', 'unknown'),
                urgency=getattr(event, 'urgency', 'normal')
            )

            return Alert(
                event=event,
                priority=self.priority,
                rule_name=self.name,
                created_at=datetime.now(),
                message=message
            )

        return None


class AlertChannel(ABC):
    """Base class for alert delivery channels"""

    @abstractmethod
    async def send(self, alert: Alert):
        pass


class SlackChannel(AlertChannel):
    """Send alerts to Slack"""

    PRIORITY_COLORS = {
        AlertPriority.LOW: '#36a64f',
        AlertPriority.MEDIUM: '#ffcc00',
        AlertPriority.HIGH: '#ff6600',
        AlertPriority.URGENT: '#ff0000',
    }

    def __init__(self, webhook_url: str, channel: str = None):
        self.webhook_url = webhook_url
        self.channel = channel

    async def send(self, alert: Alert):
        payload = {
            'attachments': [{
                'color': self.PRIORITY_COLORS[alert.priority],
                'title': f"🚨 {alert.rule_name}",
                'text': alert.message,
                'fields': [
                    {'title': 'Subreddit', 'value': f"r/{alert.event.subreddit}", 'short': True},
                    {'title': 'Priority', 'value': alert.priority.name, 'short': True},
                ],
                'actions': [{
                    'type': 'button',
                    'text': 'View on Reddit',
                    'url': alert.event.url
                }],
                'footer': f"Reddit Monitor | {alert.created_at.isoformat()}"
            }]
        }

        if self.channel:
            payload['channel'] = self.channel

        async with aiohttp.ClientSession() as session:
            async with session.post(self.webhook_url, json=payload) as response:
                if response.status != 200:
                    logger.error(f"Slack alert failed: {await response.text()}")


class WebhookChannel(AlertChannel):
    """Send alerts to custom webhook"""

    def __init__(self, url: str, headers: Dict = None):
        self.url = url
        self.headers = headers or {}

    async def send(self, alert: Alert):
        payload = {
            'rule': alert.rule_name,
            'priority': alert.priority.name,
            'message': alert.message,
            'event': alert.event.to_dict(),
            'timestamp': alert.created_at.isoformat()
        }

        async with aiohttp.ClientSession() as session:
            async with session.post(
                self.url,
                json=payload,
                headers=self.headers
            ) as response:
                if response.status >= 400:
                    logger.error(f"Webhook alert failed: {response.status}")


class AlertManager:
    """Manage alert rules and delivery"""

    def __init__(self):
        self.rules: List[AlertRule] = []
        self.channels: Dict[AlertPriority, List[AlertChannel]] = defaultdict(list)
        self.alert_history: List[Alert] = []
        self.max_history = 1000

    def add_rule(self, rule: AlertRule):
        """Add an alert rule"""
        self.rules.append(rule)

    def add_channel(self, channel: AlertChannel, min_priority: AlertPriority):
        """Add delivery channel for priority level and above"""
        for priority in AlertPriority:
            if priority.value >= min_priority.value:
                self.channels[priority].append(channel)

    async def process_event(self, event: RedditEvent):
        """Check event against rules and send alerts"""

        for rule in self.rules:
            alert = rule.check(event)

            if alert:
                # Store in history
                self.alert_history.append(alert)
                if len(self.alert_history) > self.max_history:
                    self.alert_history.pop(0)

                # Send to appropriate channels
                channels = self.channels[alert.priority]

                if channels:
                    tasks = [channel.send(alert) for channel in channels]
                    await asyncio.gather(*tasks, return_exceptions=True)

                logger.info(f"Alert triggered: {alert.rule_name} ({alert.priority.name})")

def create_brand_alert_rules(brand_name: str) -> List[AlertRule]:
    """Create standard brand monitoring alert rules"""

    brand_pattern = re.compile(rf'\b{re.escape(brand_name)}\b', re.IGNORECASE)

    return [
        # Crisis Detection
        AlertRule(
            name="Crisis Detection",
            condition=lambda e: (
                brand_pattern.search(f"{e.title} {e.content}") and
                getattr(e, 'urgency', '') == 'urgent' and
                getattr(e, 'sentiment', '') == 'negative'
            ),
            priority=AlertPriority.URGENT,
            message_template="""
🚨 **URGENT: Potential Crisis Detected**

**Post:** {title}
**Subreddit:** r/{subreddit}
**Author:** u/{author}
**Sentiment:** {sentiment}

This post matches crisis keywords and has negative sentiment.
Immediate review recommended.

[View Post]({url})
""",
            cooldown_minutes=1
        ),

        # Negative Sentiment Spike
        AlertRule(
            name="Negative Mention",
            condition=lambda e: (
                brand_pattern.search(f"{e.title} {e.content}") and
                getattr(e, 'sentiment', '') == 'negative' and
                getattr(e, 'sentiment_score', 0) > 0.8
            ),
            priority=AlertPriority.HIGH,
            message_template="""
⚠️ **Negative Brand Mention**

**Post:** {title}
**Subreddit:** r/{subreddit}
**Sentiment Score:** High negative

Consider engaging with this post.

[View Post]({url})
""",
            cooldown_minutes=10
        ),

        # Viral Potential
        AlertRule(
            name="Viral Post",
            condition=lambda e: (
                brand_pattern.search(f"{e.title} {e.content}") and
                e.score >= 100 and
                e.event_type == 'post'
            ),
            priority=AlertPriority.MEDIUM,
            message_template="""
📈 **High-Engagement Brand Mention**

**Post:** {title}
**Subreddit:** r/{subreddit}
**Score:** {score}+

This post is gaining traction.

[View Post]({url})
""",
            cooldown_minutes=30
        ),

        # Any Brand Mention (for logging)
        AlertRule(
            name="Brand Mention",
            condition=lambda e: brand_pattern.search(f"{e.title} {e.content}"),
            priority=AlertPriority.LOW,
            message_template="New mention in r/{subreddit}: {title}",
            cooldown_minutes=0  # No cooldown for logging
        ),
    ]

import redis
import json
import asyncio
from typing import Callable, List
import uuid

class RedisStreamConsumer:
    """Scalable Redis Streams consumer with consumer groups"""

    def __init__(self,
                 redis_url: str,
                 stream_key: str,
                 group_name: str,
                 consumer_name: str = None):

        self.redis_client = redis.from_url(redis_url)
        self.stream_key = stream_key
        self.group_name = group_name
        self.consumer_name = consumer_name or f"consumer-{uuid.uuid4().hex[:8]}"

        # Create consumer group if it doesn't exist
        try:
            self.redis_client.xgroup_create(
                stream_key,
                group_name,
                id='0',
                mkstream=True
            )
        except redis.exceptions.ResponseError as e:
            if 'BUSYGROUP' not in str(e):
                raise

    def consume(self,
                 processor: Callable[[RedditEvent], None],
                 batch_size: int = 10,
                 block_ms: int = 5000):
        """
        Consume events from Redis Stream

        Args:
            processor: Function to process each event
            batch_size: Number of events to read at once
            block_ms: How long to block waiting for new events
        """

        logger.info(f"Starting consumer {self.consumer_name} in group {self.group_name}")

        while True:
            try:
                # Read new events
                events = self.redis_client.xreadgroup(
                    groupname=self.group_name,
                    consumername=self.consumer_name,
                    streams={self.stream_key: '>'},
                    count=batch_size,
                    block=block_ms
                )

                if not events:
                    continue

                for stream_name, messages in events:
                    for message_id, data in messages:
                        try:
                            # Parse event
                            event_data = json.loads(data[b'data'])
                            event = RedditEvent(**event_data)

                            # Process
                            processor(event)

                            # Acknowledge
                            self.redis_client.xack(
                                self.stream_key,
                                self.group_name,
                                message_id
                            )

                        except Exception as e:
                            logger.error(f"Error processing message {message_id}: {e}")
                            # Don't ack - will be reprocessed

            except KeyboardInterrupt:
                logger.info("Consumer shutting down...")
                break
            except Exception as e:
                logger.error(f"Consumer error: {e}")
                asyncio.sleep(1)

    def get_pending_count(self) -> int:
        """Get count of unprocessed messages"""
        info = self.redis_client.xpending(self.stream_key, self.group_name)
        return info['pending'] if info else 0

    def claim_stale_messages(self, min_idle_ms: int = 60000) -> int:
        """Claim messages from dead consumers"""
        claimed = self.redis_client.xautoclaim(
            self.stream_key,
            self.group_name,
            self.consumer_name,
            min_idle_time=min_idle_ms,
            count=100
        )
        return len(claimed[1]) if claimed else 0

from prometheus_client import Counter, Gauge, Histogram, start_http_server
from datetime import datetime
import time

class StreamingMetrics:
    """Prometheus metrics for streaming system"""

    def __init__(self, port: int = 8000):
        # Event counters
        self.events_total = Counter(
            'reddit_events_total',
            'Total Reddit events processed',
            ['event_type', 'subreddit']
        )

        self.events_filtered = Counter(
            'reddit_events_filtered_total',
            'Events filtered out by pipeline',
            ['stage']
        )

        # Alert counters
        self.alerts_total = Counter(
            'reddit_alerts_total',
            'Total alerts generated',
            ['rule', 'priority']
        )

        self.alerts_sent = Counter(
            'reddit_alerts_sent_total',
            'Alerts sent to channels',
            ['channel', 'status']
        )

        # Latency metrics
        self.processing_latency = Histogram(
            'reddit_processing_latency_seconds',
            'Event processing latency',
            buckets=[.01, .05, .1, .25, .5, 1.0, 2.5, 5.0]
        )

        self.event_lag = Histogram(
            'reddit_event_lag_seconds',
            'Time between event creation and processing',
            buckets=[1, 5, 10, 30, 60, 120, 300]
        )

        # Queue metrics
        self.queue_size = Gauge(
            'reddit_queue_size',
            'Current event queue size'
        )

        self.pending_messages = Gauge(
            'reddit_pending_messages',
            'Unacknowledged messages in Redis'
        )

        # Sentiment distribution
        self.sentiment_counts = Counter(
            'reddit_sentiment_total',
            'Events by sentiment',
            ['sentiment', 'subreddit']
        )

        # Start metrics server
        start_http_server(port)
        logger.info(f"Metrics server started on port {port}")

    def record_event(self, event: RedditEvent):
        """Record event metrics"""
        self.events_total.labels(
            event_type=event.event_type,
            subreddit=event.subreddit
        ).inc()

        # Calculate lag
        lag = time.time() - event.created_utc
        self.event_lag.observe(lag)

        # Record sentiment if available
        if hasattr(event, 'sentiment'):
            self.sentiment_counts.labels(
                sentiment=event.sentiment,
                subreddit=event.subreddit
            ).inc()

    def record_alert(self, alert: Alert):
        """Record alert metrics"""
        self.alerts_total.labels(
            rule=alert.rule_name,
            priority=alert.priority.name
        ).inc()

    @contextmanager
    def measure_processing_time(self):
        """Context manager to measure processing time"""
        start = time.time()
        yield
        self.processing_latency.observe(time.time() - start)

import asyncio
import os
from dotenv import load_dotenv

load_dotenv()

async def main():
    """Main entry point for Reddit monitoring system"""

    # Configuration
    config = {
        'reddit': {
            'client_id': os.getenv('REDDIT_CLIENT_ID'),
            'client_secret': os.getenv('REDDIT_CLIENT_SECRET'),
            'user_agent': os.getenv('REDDIT_USER_AGENT', 'RedditMonitor/1.0')
        },
        'redis_url': os.getenv('REDIS_URL', 'redis://localhost:6379'),
        'slack_webhook': os.getenv('SLACK_WEBHOOK_URL'),
        'brand_name': os.getenv('BRAND_NAME', 'MyBrand'),
        'subreddits': os.getenv('SUBREDDITS', 'technology+programming').split('+')
    }

    # Initialize components
    metrics = StreamingMetrics(port=8000)

    collector = RedditStreamCollector(
        client_id=config['reddit']['client_id'],
        client_secret=config['reddit']['client_secret'],
        user_agent=config['reddit']['user_agent'],
        redis_url=config['redis_url']
    )

    # Build processing pipeline
    pipeline = create_brand_monitoring_pipeline(config['brand_name'])

    # Setup alert manager
    alert_manager = AlertManager()

    # Add alert rules
    for rule in create_brand_alert_rules(config['brand_name']):
        alert_manager.add_rule(rule)

    # Add alert channels
    if config['slack_webhook']:
        slack_channel = SlackChannel(config['slack_webhook'])
        alert_manager.add_channel(slack_channel, AlertPriority.MEDIUM)

    # Event processor function
    async def process_event(event: RedditEvent):
        with metrics.measure_processing_time():
            # Run through pipeline
            processed_event = pipeline.process(event)

            if processed_event:
                metrics.record_event(processed_event)
                await alert_manager.process_event(processed_event)

    # Start multi-threaded streaming
    stream_manager = MultiStreamManager(collector)

    # Wrap async processor for threading
    def sync_processor(event):
        asyncio.run(process_event(event))

    logger.info("Starting Reddit monitoring system...")
    logger.info(f"Monitoring brand: {config['brand_name']}")
    logger.info(f"Watching subreddits: {config['subreddits']}")

    # Start streaming
    stream_manager.start(
        subreddit_groups=[config['subreddits']],
        processor_fn=sync_processor,
        include_comments=True,
        num_processors=4
    )


if __name__ == '__main__':
    asyncio.run(main())

import numpy as np
from collections import deque
from scipy import stats

class RealTimeAnomalyDetector:
    """Detect anomalies in streaming metrics"""

    def __init__(self,
                 window_size: int = 100,
                 z_threshold: float = 3.0,
                 min_samples: int = 20):
        self.window_size = window_size
        self.z_threshold = z_threshold
        self.min_samples = min_samples
        self.metrics: Dict[str, deque] = defaultdict(
            lambda: deque(maxlen=window_size)
        )

    def check_volume_anomaly(self,
                              metric_name: str,
                              value: float) -> Optional[Dict]:
        """Check if current value is anomalous"""

        history = self.metrics[metric_name]
        history.append(value)

        if len(history) < self.min_samples:
            return None

        values = np.array(history)
        mean = np.mean(values[:-1])  # Exclude current
        std = np.std(values[:-1])

        if std == 0:
            return None

        z_score = (value - mean) / std

        if abs(z_score) > self.z_threshold:
            return {
                'metric': metric_name,
                'value': value,
                'z_score': z_score,
                'mean': mean,
                'std': std,
                'direction': 'spike' if z_score > 0 else 'drop'
            }

        return None

    def check_sentiment_shift(self,
                               subreddit: str,
                               sentiment_scores: List[float]) -> Optional[Dict]:
        """Detect sudden sentiment shifts"""

        metric_name = f"sentiment:{subreddit}"
        avg_sentiment = np.mean(sentiment_scores)

        return self.check_volume_anomaly(metric_name, avg_sentiment)