Memecoin Analysis with Ollama: Early Detection and Risk Assessment

# Install Ollama (Linux/macOS)
curl -fsSL https://ollama.ai/install.sh | sh

# Pull the recommended model for analysis
ollama pull llama2:13b

# Verify installation
ollama list

# requirements.txt
requests==2.31.0
pandas==2.0.3
web3==6.9.0
python-dotenv==1.0.0
ollama==0.1.7
ccxt==4.0.77

pip install -r requirements.txt

# .env file
ETHERSCAN_API_KEY=your_etherscan_key
COINMARKETCAP_API_KEY=your_cmc_key
TWITTER_BEARER_TOKEN=your_twitter_token
TELEGRAM_BOT_TOKEN=your_telegram_token

# data_collector.py
import requests
import pandas as pd
from web3 import Web3
import ccxt
from datetime import datetime, timedelta

class MememcoinDataCollector:
    def __init__(self, config):
        self.etherscan_key = config['ETHERSCAN_API_KEY']
        self.cmc_key = config['COINMARKETCAP_API_KEY']
        self.w3 = Web3(Web3.HTTPProvider('https://mainnet.infura.io/v3/your-key'))
        
    def get_token_metrics(self, contract_address):
        """Collect essential token metrics for analysis"""
        
        # Get basic token information
        token_info = self._get_token_info(contract_address)
        
        # Collect holder distribution
        holder_data = self._get_holder_distribution(contract_address)
        
        # Analyze transaction patterns
        tx_patterns = self._analyze_transactions(contract_address)
        
        # Social media metrics
        social_metrics = self._get_social_metrics(contract_address)
        
        return {
            'token_info': token_info,
            'holders': holder_data,
            'transactions': tx_patterns,
            'social': social_metrics,
            'timestamp': datetime.now()
        }
    
    def _get_holder_distribution(self, contract_address):
        """Analyze token holder concentration"""
        url = f"https://api.etherscan.io/api"
        params = {
            'module': 'token',
            'action': 'tokenholderlist',
            'contractaddress': contract_address,
            'page': 1,
            'offset': 100,
            'apikey': self.etherscan_key
        }
        
        response = requests.get(url, params=params)
        data = response.json()
        
        if data['status'] == '1':
            holders = data['result']
            total_supply = sum(int(h['TokenHolderQuantity']) for h in holders)
            
            # Calculate concentration metrics
            top_10_percentage = sum(int(h['TokenHolderQuantity']) 
                                  for h in holders[:10]) / total_supply * 100
            
            return {
                'total_holders': len(holders),
                'top_10_concentration': top_10_percentage,
                'holders_data': holders
            }
        
        return None

# risk_analyzer.py
import ollama
import json
from typing import Dict, List

class MememcoinRiskAnalyzer:
    def __init__(self):
        self.model_name = "llama2:13b"
        self.risk_thresholds = {
            'holder_concentration': 50,  # Top 10 holders > 50% is high risk
            'liquidity_ratio': 0.1,      # < 10% liquidity is risky
            'whale_activity': 5,         # > 5 large transactions in 24h
            'social_sentiment': -0.3     # Negative sentiment threshold
        }
    
    def analyze_risk_factors(self, token_data: Dict) -> Dict:
        """Comprehensive risk assessment using multiple factors"""
        
        # Technical risk analysis
        technical_risks = self._analyze_technical_risks(token_data)
        
        # Social sentiment analysis
        sentiment_risks = self._analyze_sentiment_risks(token_data)
        
        # Liquidity and market risks
        liquidity_risks = self._analyze_liquidity_risks(token_data)
        
        # AI-powered pattern analysis
        ai_analysis = self._get_ai_insights(token_data)
        
        # Combine all risk factors
        overall_risk = self._calculate_overall_risk(
            technical_risks, sentiment_risks, liquidity_risks, ai_analysis
        )
        
        return {
            'overall_risk_score': overall_risk,
            'technical_risks': technical_risks,
            'sentiment_risks': sentiment_risks,
            'liquidity_risks': liquidity_risks,
            'ai_insights': ai_analysis,
            'recommendations': self._generate_recommendations(overall_risk)
        }
    
    def _analyze_technical_risks(self, token_data: Dict) -> Dict:
        """Analyze technical indicators and contract risks"""
        
        risks = {}
        
        # Holder concentration risk
        concentration = token_data['holders']['top_10_concentration']
        risks['holder_concentration'] = {
            'value': concentration,
            'risk_level': 'high' if concentration > 50 else 'medium' if concentration > 30 else 'low',
            'description': f"Top 10 holders control {concentration:.1f}% of supply"
        }
        
        # Contract verification
        contract_verified = token_data['token_info'].get('is_verified', False)
        risks['contract_verification'] = {
            'verified': contract_verified,
            'risk_level': 'low' if contract_verified else 'high',
            'description': 'Contract source code verified' if contract_verified else 'Unverified contract'
        }
        
        return risks
    
    def _get_ai_insights(self, token_data: Dict) -> Dict:
        """Use Ollama to analyze patterns and generate insights"""
        
        # Prepare data summary for AI analysis
        data_summary = f"""
        Token Analysis Data:
        - Total Holders: {token_data['holders']['total_holders']}
        - Top 10 Holder Concentration: {token_data['holders']['top_10_concentration']:.1f}%
        - Recent Transaction Volume: {token_data['transactions'].get('volume_24h', 'N/A')}
        - Social Mentions: {token_data['social'].get('mentions_24h', 'N/A')}
        - Price Change 24h: {token_data['token_info'].get('price_change_24h', 'N/A')}%
        
        Analyze this memecoin data and identify:
        1. Key risk factors
        2. Potential opportunities
        3. Red flags for rugpulls
        4. Market manipulation signs
        5. Overall investment recommendation
        """
        
        try:
            response = ollama.chat(model=self.model_name, messages=[
                {
                    'role': 'system',
                    'content': 'You are an expert cryptocurrency analyst specializing in memecoin risk assessment. Provide detailed, actionable insights based on the data provided.'
                },
                {
                    'role': 'user',
                    'content': data_summary
                }
            ])
            
            return {
                'ai_analysis': response['message']['content'],
                'confidence': 'high',
                'timestamp': datetime.now()
            }
            
        except Exception as e:
            return {
                'ai_analysis': f"AI analysis failed: {str(e)}",
                'confidence': 'low',
                'timestamp': datetime.now()
            }

# early_detection.py
import asyncio
import time
from datetime import datetime, timedelta

class EarlyDetectionSystem:
    def __init__(self, config):
        self.data_collector = MememcoinDataCollector(config)
        self.risk_analyzer = MememcoinRiskAnalyzer()
        self.monitoring_active = False
        self.scan_interval = 300  # 5 minutes
        
    async def start_monitoring(self):
        """Begin continuous memecoin monitoring"""
        self.monitoring_active = True
        print("🚀 Early detection system activated")
        
        while self.monitoring_active:
            try:
                # Scan for new tokens
                new_tokens = await self._scan_new_tokens()
                
                # Analyze promising candidates
                for token in new_tokens:
                    analysis = await self._analyze_token(token)
                    
                    if analysis['overall_risk_score'] < 0.3:  # Low risk threshold
                        await self._send_alert(token, analysis)
                
                # Wait before next scan
                await asyncio.sleep(self.scan_interval)
                
            except Exception as e:
                print(f"❌ Monitoring error: {e}")
                await asyncio.sleep(60)  # Wait 1 minute before retry
    
    async def _scan_new_tokens(self):
        """Identify newly launched tokens worth analyzing"""
        
        # Scan DEX listings for new tokens
        new_listings = self._get_new_dex_listings()
        
        # Filter by volume and holder count
        filtered_tokens = []
        
        for token in new_listings:
            # Quick preliminary check
            if (token.get('volume_24h', 0) > 10000 and 
                token.get('holder_count', 0) > 50):
                filtered_tokens.append(token)
        
        return filtered_tokens
    
    async def _analyze_token(self, token):
        """Perform comprehensive analysis on a token"""
        
        # Collect data
        token_data = self.data_collector.get_token_metrics(
            token['contract_address']
        )
        
        # Analyze risks
        risk_analysis = self.risk_analyzer.analyze_risk_factors(token_data)
        
        return risk_analysis
    
    async def _send_alert(self, token, analysis):
        """Send alert for promising opportunities"""
        
        alert_message = f"""
        🎯 MEMECOIN OPPORTUNITY DETECTED
        
        Token: {token['name']} ({token['symbol']})
        Contract: {token['contract_address']}
        Risk Score: {analysis['overall_risk_score']:.2f}/1.0
        
        Key Metrics:
        - Holders: {token.get('holder_count', 'N/A')}
        - 24h Volume: ${token.get('volume_24h', 'N/A'):,}
        - Risk Level: {analysis.get('risk_level', 'Unknown')}
        
        AI Insights:
        {analysis['ai_insights']['ai_analysis'][:200]}...
        
        ⚠️ Always DYOR before investing!
        """
        
        print(alert_message)
        # Here you could send to Telegram, Discord, etc.

# manipulation_detector.py
import numpy as np
import pandas as pd
from scipy import stats

class ManipulationDetector:
    def __init__(self):
        self.manipulation_patterns = {
            'pump_and_dump': self._detect_pump_dump,
            'wash_trading': self._detect_wash_trading,
            'fake_volume': self._detect_fake_volume,
            'coordinated_buying': self._detect_coordinated_buying
        }
    
    def analyze_manipulation_risk(self, token_data):
        """Detect various manipulation patterns"""
        
        results = {}
        
        for pattern_name, detector_func in self.manipulation_patterns.items():
            try:
                result = detector_func(token_data)
                results[pattern_name] = result
            except Exception as e:
                results[pattern_name] = {
                    'detected': False,
                    'confidence': 0,
                    'error': str(e)
                }
        
        # Calculate overall manipulation risk
        manipulation_scores = [r.get('confidence', 0) for r in results.values() 
                             if r.get('detected', False)]
        
        overall_risk = np.mean(manipulation_scores) if manipulation_scores else 0
        
        return {
            'overall_manipulation_risk': overall_risk,
            'detected_patterns': results,
            'recommendation': self._get_manipulation_recommendation(overall_risk)
        }
    
    def _detect_pump_dump(self, token_data):
        """Identify pump and dump patterns"""
        
        price_data = token_data['transactions'].get('price_history', [])
        
        if len(price_data) < 24:  # Need at least 24 data points
            return {'detected': False, 'confidence': 0}
        
        # Convert to pandas for easier analysis
        df = pd.DataFrame(price_data)
        df['price_change'] = df['price'].pct_change()
        
        # Look for rapid price increase followed by decline
        recent_changes = df['price_change'].tail(12)  # Last 12 periods
        
        # Pump pattern: >50% increase in short time
        max_increase = recent_changes.max()
        
        # Dump pattern: >30% decrease after pump
        if max_increase > 0.5:
            post_pump_idx = recent_changes.idxmax()
            post_pump_changes = recent_changes.loc[post_pump_idx:]
            max_decrease = post_pump_changes.min()
            
            if max_decrease < -0.3:
                return {
                    'detected': True,
                    'confidence': 0.8,
                    'details': f"Pump: +{max_increase*100:.1f}%, Dump: {max_decrease*100:.1f}%"
                }
        
        return {'detected': False, 'confidence': 0}
    
    def _detect_wash_trading(self, token_data):
        """Identify artificial volume through wash trading"""
        
        transactions = token_data['transactions'].get('recent_txs', [])
        
        if len(transactions) < 10:
            return {'detected': False, 'confidence': 0}
        
        # Analyze transaction patterns
        df = pd.DataFrame(transactions)
        
        # Look for repetitive patterns
        # Same amounts, regular intervals, back-and-forth between addresses
        amount_frequency = df['amount'].value_counts()
        repeated_amounts = (amount_frequency > 3).sum()
        
        # Check for address ping-ponging
        address_pairs = []
        for i in range(len(df) - 1):
            from_addr = df.iloc[i]['from']
            to_addr = df.iloc[i]['to']
            next_from = df.iloc[i + 1]['from']
            next_to = df.iloc[i + 1]['to']
            
            if from_addr == next_to and to_addr == next_from:
                address_pairs.append((from_addr, to_addr))
        
        wash_indicators = len(address_pairs) + (repeated_amounts / len(df))
        confidence = min(wash_indicators / 5, 1.0)  # Normalize to 0-1
        
        return {
            'detected': confidence > 0.3,
            'confidence': confidence,
            'details': f"Repeated amounts: {repeated_amounts}, Address pairs: {len(address_pairs)}"
        }

# performance_tracker.py
import json
import sqlite3
from datetime import datetime, timedelta

class PerformanceTracker:
    def __init__(self, db_path="memecoin_analysis.db"):
        self.db_path = db_path
        self._init_database()
    
    def _init_database(self):
        """Initialize SQLite database for tracking"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS analyses (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                token_address TEXT,
                token_symbol TEXT,
                analysis_date DATETIME,
                risk_score REAL,
                predicted_outcome TEXT,
                actual_outcome TEXT,
                price_at_analysis REAL,
                price_after_7d REAL,
                price_after_30d REAL,
                accuracy_score REAL
            )
        ''')
        
        conn.commit()
        conn.close()
    
    def log_analysis(self, token_address, analysis_data, prediction):
        """Record a new analysis for future tracking"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT INTO analyses 
            (token_address, token_symbol, analysis_date, risk_score, 
             predicted_outcome, price_at_analysis)
            VALUES (?, ?, ?, ?, ?, ?)
        ''', (
            token_address,
            analysis_data.get('symbol', 'UNKNOWN'),
            datetime.now(),
            analysis_data['overall_risk_score'],
            prediction,
            analysis_data.get('current_price', 0)
        ))
        
        conn.commit()
        conn.close()
    
    def update_outcomes(self):
        """Update actual outcomes for past analyses"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # Get analyses from 7 and 30 days ago that need updates
        seven_days_ago = datetime.now() - timedelta(days=7)
        thirty_days_ago = datetime.now() - timedelta(days=30)
        
        cursor.execute('''
            SELECT id, token_address, analysis_date, price_at_analysis
            FROM analyses 
            WHERE analysis_date <= ? AND price_after_7d IS NULL
        ''', (seven_days_ago,))
        
        for row in cursor.fetchall():
            analysis_id, token_address, analysis_date, original_price = row
            
            # Get current price (implement your price fetching logic)
            current_price = self._get_current_price(token_address)
            
            if current_price:
                # Calculate price change
                price_change = (current_price - original_price) / original_price
                
                # Determine actual outcome
                if price_change > 0.5:
                    actual_outcome = "moon"
                elif price_change < -0.5:
                    actual_outcome = "dump"
                else:
                    actual_outcome = "stable"
                
                # Update record
                cursor.execute('''
                    UPDATE analyses 
                    SET price_after_7d = ?, actual_outcome = ?
                    WHERE id = ?
                ''', (current_price, actual_outcome, analysis_id))
        
        conn.commit()
        conn.close()
    
    def generate_performance_report(self):
        """Generate accuracy report for the analysis system"""
        conn = sqlite3.connect(self.db_path)
        
        # Get completed analyses
        df = pd.read_sql_query('''
            SELECT * FROM analyses 
            WHERE actual_outcome IS NOT NULL
        ''', conn)
        
        if len(df) == 0:
            return "No completed analyses found."
        
        # Calculate accuracy metrics
        correct_predictions = (df['predicted_outcome'] == df['actual_outcome']).sum()
        total_predictions = len(df)
        accuracy = correct_predictions / total_predictions * 100
        
        # Risk score effectiveness
        high_risk_dumps = df[(df['risk_score'] > 0.7) & (df['actual_outcome'] == 'dump')]
        low_risk_moons = df[(df['risk_score'] < 0.3) & (df['actual_outcome'] == 'moon')]
        
        report = f"""
        📊 MEMECOIN ANALYSIS PERFORMANCE REPORT
        
        Overall Accuracy: {accuracy:.1f}% ({correct_predictions}/{total_predictions})
        
        Risk Assessment Effectiveness:
        - High-risk tokens that dumped: {len(high_risk_dumps)}
        - Low-risk tokens that mooned: {len(low_risk_moons)}
        
        Average Risk Scores by Outcome:
        - Dumps: {df[df['actual_outcome'] == 'dump']['risk_score'].mean():.2f}
        - Moons: {df[df['actual_outcome'] == 'moon']['risk_score'].mean():.2f}
        - Stable: {df[df['actual_outcome'] == 'stable']['risk_score'].mean():.2f}
        
        Recent Performance (Last 30 days):
        {self._get_recent_performance_summary(df)}
        """
        
        conn.close()
        return report

# main.py
import asyncio
import schedule
import time
from threading import Thread
from config import load_config

class MememcoinAnalysisBot:
    def __init__(self):
        self.config = load_config()
        self.detection_system = EarlyDetectionSystem(self.config)
        self.performance_tracker = PerformanceTracker()
        self.running = False
    
    def start(self):
        """Start the complete analysis system"""
        print("🤖 Starting Memecoin Analysis Bot with Ollama")
        
        # Schedule regular tasks
        schedule.every(5).minutes.do(self._quick_scan)
        schedule.every(1).hours.do(self._deep_analysis)
        schedule.every(1).days.do(self._update_performance)
        
        # Start background tasks
        self.running = True
        
        # Start the detection system
        detection_thread = Thread(
            target=asyncio.run, 
            args=(self.detection_system.start_monitoring(),)
        )
        detection_thread.daemon = True
        detection_thread.start()
        
        # Start scheduler
        while self.running:
            schedule.run_pending()
            time.sleep(1)
    
    def _quick_scan(self):
        """Quick scan for immediate opportunities"""
        print("🔍 Running quick market scan...")
        # Implement quick scanning logic
    
    def _deep_analysis(self):
        """Comprehensive analysis of watchlist tokens"""
        print("🧠 Running deep analysis...")
        # Implement deep analysis logic
    
    def _update_performance(self):
        """Update performance tracking"""
        print("📈 Updating performance metrics...")
        self.performance_tracker.update_outcomes()
        report = self.performance_tracker.generate_performance_report()
        print(report)

if __name__ == "__main__":
    bot = MememcoinAnalysisBot()
    try:
        bot.start()
    except KeyboardInterrupt:
        print("🛑 Shutting down analysis bot...")
        bot.running = False

# portfolio_manager.py
class PortfolioManager:
    def __init__(self, config):
        self.config = config
        self.max_position_size = 0.05  # 5% max per memecoin
        self.max_memecoin_allocation = 0.20  # 20% total portfolio
    
    def should_invest(self, analysis_result, current_portfolio):
        """Determine if investment meets risk criteria"""
        
        # Check risk score
        if analysis_result['overall_risk_score'] > 0.4:
            return False, "Risk score too high"
        
        # Check portfolio allocation
        current_memecoin_allocation = self._calculate_memecoin_allocation(current_portfolio)
        if current_memecoin_allocation >= self.max_memecoin_allocation:
            return False, "Maximum memecoin allocation reached"
        
        # Check for manipulation signs
        manipulation_risk = analysis_result.get('manipulation_risk', 0)
        if manipulation_risk > 0.3:
            return False, "Manipulation risk detected"
        
        return True, "Investment approved"
    
    def calculate_position_size(self, portfolio_value, risk_score):
        """Calculate appropriate position size based on risk"""
        
        base_size = self.max_position_size
        risk_adjustment = 1 - risk_score  # Lower risk = larger position
        
        position_size = base_size * risk_adjustment
        return min(position_size, self.max_position_size)

# alert_system.py
import requests
import smtplib
from email.mime.text import MimeText

class AlertSystem:
    def __init__(self, config):
        self.telegram_token = config.get('TELEGRAM_BOT_TOKEN')
        self.telegram_chat_id = config.get('TELEGRAM_CHAT_ID')
        self.email_config = config.get('email', {})
    
    def send_opportunity_alert(self, token_data, analysis):
        """Send alert for investment opportunities"""
        
        message = self._format_opportunity_message(token_data, analysis)
        
        # Send via multiple channels
        self._send_telegram(message)
        self._send_email("🚀 Memecoin Opportunity", message)
    
    def send_risk_alert(self, token_address, risk_type, details):
        """Send alert for risk events"""
        
        message = f"""
        ⚠️ RISK ALERT
        
        Token: {token_address}
        Risk Type: {risk_type}
        Details: {details}
        
        Consider exiting position or reducing exposure.
        """
        
        self._send_telegram(message)
    
    def _send_telegram(self, message):
        """Send message via Telegram"""
        if not self.telegram_token:
            return
        
        url = f"https://api.telegram.org/bot{self.telegram_token}/sendMessage"
        data = {
            'chat_id': self.telegram_chat_id,
            'text': message,
            'parse_mode': 'Markdown'
        }
        
        try:
            requests.post(url, data=data)
        except Exception as e:
            print(f"Telegram alert failed: {e}")