How to Build Memecoin Sentiment Monitor with Ollama: Viral Prediction System

# Install Ollama
curl -fsSL https://ollama.ai/install.sh | sh

# Pull the language model
ollama pull llama2

# Install Python dependencies
pip install ollama requests pandas numpy tweepy praw discord.py

# config.py
TWITTER_BEARER_TOKEN = "your_twitter_bearer_token"
REDDIT_CLIENT_ID = "your_reddit_client_id" 
REDDIT_CLIENT_SECRET = "your_reddit_client_secret"
DISCORD_BOT_TOKEN = "your_discord_bot_token"

# Target channels and keywords
MEMECOIN_KEYWORDS = ["$PEPE", "$SHIB", "$DOGE", "$WIF", "$BONK"]
TWITTER_ACCOUNTS = ["@elonmusk", "@VitalikButerin", "@CryptoCobain"]
REDDIT_SUBREDDITS = ["CryptoCurrency", "SatoshiStreetBets", "memecoins"]

# sentiment_analyzer.py
import ollama
from typing import Dict, List
import re

class MemesentimentAnalyzer:
    def __init__(self):
        self.model = "llama2"
        
    def analyze_text(self, text: str, token: str) -> Dict:
        """
        Analyze text sentiment for specific memecoin
        Returns sentiment score and viral indicators
        """
        prompt = f"""
        Analyze this social media post about {token}:
        "{text}"
        
        Rate 1-10:
        - Bullish sentiment
        - Viral potential  
        - Community excitement
        - FOMO level
        
        Response format: "sentiment:X viral:X excitement:X fomo:X"
        """
        
        response = ollama.chat(model=self.model, messages=[{
            'role': 'user',
            'content': prompt
        }])
        
        return self._parse_scores(response['message']['content'])
    
    def _parse_scores(self, response: str) -> Dict:
        """Extract numerical scores from Ollama response"""
        scores = {}
        patterns = {
            'sentiment': r'sentiment:(\d+)',
            'viral': r'viral:(\d+)', 
            'excitement': r'excitement:(\d+)',
            'fomo': r'fomo:(\d+)'
        }
        
        for key, pattern in patterns.items():
            match = re.search(pattern, response)
            scores[key] = int(match.group(1)) if match else 5
            
        return scores

# data_collector.py
import tweepy
import praw
import discord
from datetime import datetime, timedelta

class SocialMediaCollector:
    def __init__(self, config):
        self.twitter_api = self._setup_twitter(config.TWITTER_BEARER_TOKEN)
        self.reddit_api = self._setup_reddit(config)
        
    def _setup_twitter(self, bearer_token: str):
        """Initialize Twitter API client"""
        return tweepy.Client(bearer_token=bearer_token)
    
    def _setup_reddit(self, config):
        """Initialize Reddit API client"""
        return praw.Reddit(
            client_id=config.REDDIT_CLIENT_ID,
            client_secret=config.REDDIT_CLIENT_SECRET,
            user_agent="MemesentimentMonitor/1.0"
        )
    
    def collect_twitter_mentions(self, keywords: List[str], hours: int = 1) -> List[Dict]:
        """Collect recent Twitter mentions for memecoin keywords"""
        mentions = []
        end_time = datetime.utcnow()
        start_time = end_time - timedelta(hours=hours)
        
        for keyword in keywords:
            tweets = tweepy.Paginator(
                self.twitter_api.search_recent_tweets,
                query=f"{keyword} -is:retweet",
                tweet_fields=['created_at', 'public_metrics'],
                start_time=start_time,
                end_time=end_time,
                max_results=100
            ).flatten(limit=500)
            
            for tweet in tweets:
                mentions.append({
                    'platform': 'twitter',
                    'text': tweet.text,
                    'created_at': tweet.created_at,
                    'keyword': keyword,
                    'metrics': tweet.public_metrics
                })
                
        return mentions
    
    def collect_reddit_posts(self, subreddits: List[str], keywords: List[str]) -> List[Dict]:
        """Collect Reddit posts mentioning memecoin keywords"""
        posts = []
        
        for subreddit_name in subreddits:
            subreddit = self.reddit_api.subreddit(subreddit_name)
            
            for submission in subreddit.new(limit=100):
                text = f"{submission.title} {submission.selftext}"
                
                for keyword in keywords:
                    if keyword.lower() in text.lower():
                        posts.append({
                            'platform': 'reddit',
                            'text': text,
                            'created_at': datetime.fromtimestamp(submission.created_utc),
                            'keyword': keyword,
                            'score': submission.score,
                            'comments': submission.num_comments
                        })
                        break
                        
        return posts

# viral_predictor.py
import pandas as pd
import numpy as np
from collections import defaultdict
from datetime import datetime, timedelta

class ViralPredictor:
    def __init__(self, sentiment_analyzer):
        self.analyzer = sentiment_analyzer
        self.historical_data = defaultdict(list)
        
    def calculate_viral_score(self, mentions: List[Dict]) -> Dict[str, float]:
        """
        Calculate viral prediction score for each memecoin
        Combines sentiment, volume, and momentum indicators
        """
        token_scores = defaultdict(lambda: {
            'volume': 0,
            'sentiment_avg': 0,
            'viral_avg': 0,
            'momentum': 0,
            'final_score': 0
        })
        
        # Group mentions by token
        token_mentions = defaultdict(list)
        for mention in mentions:
            token = mention['keyword']
            token_mentions[token].append(mention)
        
        for token, token_data in token_mentions.items():
            scores = self._analyze_token_mentions(token, token_data)
            token_scores[token] = scores
            
        return dict(token_scores)
    
    def _analyze_token_mentions(self, token: str, mentions: List[Dict]) -> Dict:
        """Analyze sentiment and viral indicators for specific token"""
        if not mentions:
            return {'volume': 0, 'sentiment_avg': 0, 'viral_avg': 0, 'momentum': 0, 'final_score': 0}
        
        # Calculate volume metrics
        volume_score = min(len(mentions) / 10, 10)  # Cap at 10
        
        # Analyze sentiment for each mention
        sentiment_scores = []
        viral_scores = []
        
        for mention in mentions:
            analysis = self.analyzer.analyze_text(mention['text'], token)
            sentiment_scores.append(analysis['sentiment'])
            viral_scores.append(analysis['viral'])
        
        sentiment_avg = np.mean(sentiment_scores)
        viral_avg = np.mean(viral_scores)
        
        # Calculate momentum (recent vs older mentions)
        momentum = self._calculate_momentum(mentions)
        
        # Final viral score formula
        final_score = (
            volume_score * 0.3 +
            sentiment_avg * 0.25 +
            viral_avg * 0.25 +
            momentum * 0.2
        )
        
        return {
            'volume': volume_score,
            'sentiment_avg': sentiment_avg,
            'viral_avg': viral_avg,
            'momentum': momentum,
            'final_score': final_score
        }
    
    def _calculate_momentum(self, mentions: List[Dict]) -> float:
        """Calculate momentum based on mention timing"""
        now = datetime.utcnow()
        recent_cutoff = now - timedelta(minutes=30)
        
        recent_count = sum(1 for m in mentions if m['created_at'] > recent_cutoff)
        total_count = len(mentions)
        
        return (recent_count / total_count) * 10 if total_count > 0 else 0

# alert_system.py
import smtplib
from email.mime.text import MIMEText
from typing import Dict

class AlertSystem:
    def __init__(self, email_config: Dict):
        self.smtp_server = email_config['smtp_server']
        self.smtp_port = email_config['smtp_port']
        self.email = email_config['email']
        self.password = email_config['password']
        self.recipients = email_config['recipients']
        
    def check_alert_conditions(self, viral_scores: Dict[str, Dict]) -> List[Dict]:
        """Check if any tokens meet alert conditions"""
        alerts = []
        
        for token, scores in viral_scores.items():
            final_score = scores['final_score']
            
            # Alert conditions
            if final_score > 8.0:  # High viral potential
                alerts.append({
                    'token': token,
                    'score': final_score,
                    'level': 'HIGH',
                    'message': f'{token} showing extreme viral potential (Score: {final_score:.1f})'
                })
            elif final_score > 6.5:  # Medium viral potential
                alerts.append({
                    'token': token,
                    'score': final_score,
                    'level': 'MEDIUM', 
                    'message': f'{token} gaining momentum (Score: {final_score:.1f})'
                })
                
        return alerts
    
    def send_alerts(self, alerts: List[Dict]):
        """Send email alerts for high-scoring tokens"""
        if not alerts:
            return
            
        subject = f"🚀 Memecoin Alert: {len(alerts)} Tokens Detected"
        body = self._format_alert_email(alerts)
        
        try:
            with smtplib.SMTP(self.smtp_server, self.smtp_port) as server:
                server.starttls()
                server.login(self.email, self.password)
                
                for recipient in self.recipients:
                    msg = MIMEText(body)
                    msg['Subject'] = subject
                    msg['From'] = self.email
                    msg['To'] = recipient
                    
                    server.send_message(msg)
                    
            print(f"Sent alerts to {len(self.recipients)} recipients")
            
        except Exception as e:
            print(f"Failed to send alerts: {e}")
    
    def _format_alert_email(self, alerts: List[Dict]) -> str:
        """Format alerts into readable email"""
        body = "Memecoin Sentiment Monitor Alert\n\n"
        
        for alert in sorted(alerts, key=lambda x: x['score'], reverse=True):
            body += f"🎯 {alert['level']}: {alert['message']}\n"
            
        body += f"\n\nGenerated at: {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} UTC"
        return body

# main_monitor.py
import time
import schedule
from datetime import datetime
import config
from sentiment_analyzer import MemesentimentAnalyzer
from data_collector import SocialMediaCollector  
from viral_predictor import ViralPredictor
from alert_system import AlertSystem

class MemesentimentMonitor:
    def __init__(self):
        self.analyzer = MemesentimentAnalyzer()
        self.collector = SocialMediaCollector(config)
        self.predictor = ViralPredictor(self.analyzer)
        self.alert_system = AlertSystem(config.EMAIL_CONFIG)
        
    def run_analysis_cycle(self):
        """Run complete sentiment analysis cycle"""
        print(f"Starting analysis cycle at {datetime.utcnow()}")
        
        try:
            # Collect social media mentions
            twitter_mentions = self.collector.collect_twitter_mentions(
                config.MEMECOIN_KEYWORDS, hours=1
            )
            reddit_posts = self.collector.collect_reddit_posts(
                config.REDDIT_SUBREDDITS, config.MEMECOIN_KEYWORDS
            )
            
            all_mentions = twitter_mentions + reddit_posts
            print(f"Collected {len(all_mentions)} mentions")
            
            # Calculate viral scores
            viral_scores = self.predictor.calculate_viral_score(all_mentions)
            
            # Display results
            self._display_results(viral_scores)
            
            # Check for alerts
            alerts = self.alert_system.check_alert_conditions(viral_scores)
            if alerts:
                self.alert_system.send_alerts(alerts)
                
        except Exception as e:
            print(f"Analysis cycle failed: {e}")
    
    def _display_results(self, viral_scores: Dict):
        """Display current viral scores"""
        print("\n" + "="*50)
        print("MEMECOIN VIRAL SCORES")
        print("="*50)
        
        sorted_tokens = sorted(
            viral_scores.items(), 
            key=lambda x: x[1]['final_score'], 
            reverse=True
        )
        
        for token, scores in sorted_tokens:
            print(f"{token:8} | Score: {scores['final_score']:.1f} | "
                  f"Sentiment: {scores['sentiment_avg']:.1f} | "
                  f"Viral: {scores['viral_avg']:.1f} | "
                  f"Volume: {scores['volume']:.1f}")
    
    def start_monitoring(self):
        """Start continuous monitoring"""
        print("🚀 Starting Memecoin Sentiment Monitor")
        
        # Schedule analysis every 15 minutes
        schedule.every(15).minutes.do(self.run_analysis_cycle)
        
        # Run initial analysis
        self.run_analysis_cycle()
        
        # Keep running
        while True:
            schedule.run_pending()
            time.sleep(60)

if __name__ == "__main__":
    monitor = MemesentimentMonitor()
    monitor.start_monitoring()

# database.py
import sqlite3
from datetime import datetime
from typing import Dict, List

class HistoricalDatabase:
    def __init__(self, db_path: str = "memecoin_sentiment.db"):
        self.db_path = db_path
        self.init_database()
    
    def init_database(self):
        """Create database tables"""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS sentiment_scores (
                    id INTEGER PRIMARY KEY,
                    timestamp DATETIME,
                    token TEXT,
                    final_score REAL,
                    sentiment_avg REAL,
                    viral_avg REAL,
                    volume REAL,
                    momentum REAL
                )
            """)
            
            conn.execute("""
                CREATE INDEX IF NOT EXISTS idx_token_timestamp 
                ON sentiment_scores(token, timestamp)
            """)
    
    def save_scores(self, viral_scores: Dict):
        """Save current scores to database"""
        timestamp = datetime.utcnow()
        
        with sqlite3.connect(self.db_path) as conn:
            for token, scores in viral_scores.items():
                conn.execute("""
                    INSERT INTO sentiment_scores 
                    (timestamp, token, final_score, sentiment_avg, viral_avg, volume, momentum)
                    VALUES (?, ?, ?, ?, ?, ?, ?)
                """, (
                    timestamp, token, scores['final_score'],
                    scores['sentiment_avg'], scores['viral_avg'],
                    scores['volume'], scores['momentum']
                ))
    
    def get_historical_trend(self, token: str, hours: int = 24) -> List[Dict]:
        """Get historical trend for specific token"""
        cutoff_time = datetime.utcnow() - timedelta(hours=hours)
        
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute("""
                SELECT timestamp, final_score, sentiment_avg, viral_avg
                FROM sentiment_scores 
                WHERE token = ? AND timestamp > ?
                ORDER BY timestamp
            """, (token, cutoff_time))
            
            return [
                {
                    'timestamp': row[0],
                    'final_score': row[1], 
                    'sentiment_avg': row[2],
                    'viral_avg': row[3]
                }
                for row in cursor.fetchall()
            ]

# dashboard.py
from flask import Flask, render_template, jsonify
import json
from datetime import datetime, timedelta

app = Flask(__name__)

@app.route('/')
def dashboard():
    """Main dashboard page"""
    return render_template('dashboard.html')

@app.route('/api/current-scores')
def current_scores():
    """API endpoint for current viral scores"""
    # Get latest scores from database
    db = HistoricalDatabase()
    latest_scores = db.get_latest_scores()
    
    return jsonify(latest_scores)

@app.route('/api/historical/<token>')
def historical_data(token):
    """API endpoint for historical data"""
    db = HistoricalDatabase()
    trend_data = db.get_historical_trend(token, hours=48)
    
    return jsonify(trend_data)

if __name__ == '__main__':
    app.run(debug=True, host='0.0.0.0', port=5000)

FROM python:3.9-slim

WORKDIR /app

# Install Ollama
RUN curl -fsSL https://ollama.ai/install.sh | sh

# Copy requirements and install dependencies
COPY requirements.txt .
RUN pip install -r requirements.txt

# Copy application code
COPY . .

# Pull language model
RUN ollama serve & sleep 10 && ollama pull llama2

# Expose dashboard port
EXPOSE 5000

# Start the monitor
CMD ["python", "main_monitor.py"]

# production_config.py
import os

# Enhanced rate limiting
API_RATE_LIMITS = {
    'twitter': 100,  # requests per hour
    'reddit': 60,
    'discord': 50
}

# Advanced alert conditions
ALERT_CONDITIONS = {
    'viral_threshold': 8.0,
    'volume_spike': 5.0,  # 5x normal volume
    'sentiment_change': 3.0,  # 3+ point increase
    'momentum_acceleration': 7.0
}

# Email configuration
EMAIL_CONFIG = {
    'smtp_server': os.getenv('SMTP_SERVER', 'smtp.gmail.com'),
    'smtp_port': int(os.getenv('SMTP_PORT', 587)),
    'email': os.getenv('EMAIL_ADDRESS'),
    'password': os.getenv('EMAIL_PASSWORD'),
    'recipients': os.getenv('ALERT_RECIPIENTS', '').split(',')
}

# Database configuration
DATABASE_CONFIG = {
    'host': os.getenv('DB_HOST', 'localhost'),
    'port': int(os.getenv('DB_PORT', 5432)),
    'database': os.getenv('DB_NAME', 'memecoin_sentiment'),
    'user': os.getenv('DB_USER'),
    'password': os.getenv('DB_PASSWORD')
}

# backtesting.py
import pandas as pd
from datetime import datetime, timedelta
from typing import Dict, List

class Backtester:
    def __init__(self, db: HistoricalDatabase):
        self.db = db
        
    def run_backtest(self, start_date: datetime, end_date: datetime) -> Dict:
        """
        Run backtest to validate prediction accuracy
        Compare sentiment alerts with actual price movements
        """
        results = {
            'total_alerts': 0,
            'successful_predictions': 0,
            'false_positives': 0,
            'accuracy': 0.0,
            'avg_prediction_time': 0.0
        }
        
        # Get historical alerts
        alerts = self.db.get_alerts_in_period(start_date, end_date)
        
        for alert in alerts:
            # Check if price increased within 24 hours
            price_increase = self._check_price_movement(
                alert['token'], 
                alert['timestamp'],
                hours=24
            )
            
            if price_increase > 0.1:  # 10% increase threshold
                results['successful_predictions'] += 1
            else:
                results['false_positives'] += 1
                
        results['total_alerts'] = len(alerts)
        results['accuracy'] = (
            results['successful_predictions'] / results['total_alerts'] 
            if results['total_alerts'] > 0 else 0
        )
        
        return results
    
    def _check_price_movement(self, token: str, timestamp: datetime, hours: int) -> float:
        """Check actual price movement after alert"""
        # This would integrate with price data API
        # Placeholder implementation
        return 0.15  # 15% increase

# optimizations.py
import asyncio
import aiohttp
from concurrent.futures import ThreadPoolExecutor
import multiprocessing as mp

class OptimizedCollector:
    def __init__(self):
        self.session = None
        self.executor = ThreadPoolExecutor(max_workers=mp.cpu_count())
    
    async def collect_parallel(self, keywords: List[str]) -> List[Dict]:
        """Collect data from multiple sources in parallel"""
        async with aiohttp.ClientSession() as session:
            self.session = session
            
            tasks = [
                self._collect_twitter_async(keywords),
                self._collect_reddit_async(keywords),
                self._collect_discord_async(keywords)
            ]
            
            results = await asyncio.gather(*tasks)
            
        # Flatten results
        all_mentions = []
        for result_list in results:
            all_mentions.extend(result_list)
            
        return all_mentions
    
    async def _collect_twitter_async(self, keywords: List[str]) -> List[Dict]:
        """Async Twitter data collection"""
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(
            self.executor, 
            self._collect_twitter_sync, 
            keywords
        )

# memory_management.py
import gc
import psutil
from typing import Generator

class MemoryManager:
    @staticmethod
    def batch_process(items: List, batch_size: int = 100) -> Generator:
        """Process items in batches to control memory usage"""
        for i in range(0, len(items), batch_size):
            yield items[i:i + batch_size]
            gc.collect()  # Force garbage collection
    
    @staticmethod
    def monitor_memory():
        """Monitor current memory usage"""
        process = psutil.Process()
        memory_mb = process.memory_info().rss / 1024 / 1024
        print(f"Memory usage: {memory_mb:.1f} MB")
        
        if memory_mb > 1000:  # Alert if over 1GB
            print("⚠️ High memory usage detected")
            gc.collect()

# rate_limiter.py
import time
from collections import defaultdict
from datetime import datetime, timedelta

class RateLimiter:
    def __init__(self):
        self.requests = defaultdict(list)
        self.limits = {
            'twitter': {'requests': 100, 'window': 3600},  # 100/hour
            'reddit': {'requests': 60, 'window': 3600},    # 60/hour
            'discord': {'requests': 50, 'window': 3600}    # 50/hour
        }
    
    def can_make_request(self, service: str) -> bool:
        """Check if request is allowed within rate limits"""
        now = datetime.utcnow()
        window_start = now - timedelta(seconds=self.limits[service]['window'])
        
        # Remove old requests
        self.requests[service] = [
            req_time for req_time in self.requests[service]
            if req_time > window_start
        ]
        
        return len(self.requests[service]) < self.limits[service]['requests']
    
    def record_request(self, service: str):
        """Record a new request"""
        self.requests[service].append(datetime.utcnow())
    
    def wait_if_needed(self, service: str):
        """Wait if rate limit exceeded"""
        if not self.can_make_request(service):
            sleep_time = 60  # Wait 1 minute
            print(f"Rate limit hit for {service}, waiting {sleep_time}s")
            time.sleep(sleep_time)