Crypto Whale Tracking: Ollama Blockchain Transaction Analysis Tutorial

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

# Pull the required model for analysis
ollama pull llama2:7b

# Verify installation
ollama list

pip install requests pandas numpy matplotlib seaborn
pip install web3 python-dotenv jupyter
pip install ollama-python blockchain-analytics

import requests
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import ollama
import json
from typing import Dict, List, Tuple

class WhaleTracker:
    def __init__(self, api_key: str, threshold: float = 1000000):
        """
        Initialize whale tracking system
        
        Args:
            api_key: Blockchain API key
            threshold: Minimum transaction value to classify as whale
        """
        self.api_key = api_key
        self.threshold = threshold
        self.ollama_client = ollama.Client()
        self.whale_addresses = set()
        self.alert_history = []
    
    def fetch_recent_transactions(self, hours: int = 24) -> pd.DataFrame:
        """Fetch recent large transactions from blockchain"""
        # Implementation depends on your blockchain API provider
        pass
    
    def analyze_transaction_patterns(self, df: pd.DataFrame) -> Dict:
        """Use Ollama to analyze transaction patterns"""
        pass
    
    def detect_whale_movements(self, df: pd.DataFrame) -> List[Dict]:
        """Identify potential whale transactions"""
        pass

class BlockchainDataConnector:
    def __init__(self, providers: Dict[str, str]):
        """
        Initialize connections to multiple blockchain data providers
        
        Args:
            providers: Dictionary of provider names and API keys
        """
        self.providers = providers
        self.session = requests.Session()
    
    def get_ethereum_transactions(self, limit: int = 1000) -> pd.DataFrame:
        """Fetch recent Ethereum transactions"""
        url = "https://api.etherscan.io/api"
        params = {
            'module': 'account',
            'action': 'txlist',
            'sort': 'desc',
            'apikey': self.providers['etherscan']
        }
        
        response = self.session.get(url, params=params)
        data = response.json()
        
        if data['status'] == '1':
            df = pd.DataFrame(data['result'])
            df['value_usd'] = df['value'].astype(float) / 1e18 * self.get_eth_price()
            return df[df['value_usd'] >= 100000]  # Filter for large transactions
        
        return pd.DataFrame()
    
    def get_eth_price(self) -> float:
        """Get current ETH price in USD"""
        url = "https://api.coingecko.com/api/v3/simple/price"
        params = {'ids': 'ethereum', 'vs_currencies': 'usd'}
        
        response = self.session.get(url, params=params)
        return response.json()['ethereum']['usd']

def analyze_whale_patterns(self, transactions: pd.DataFrame) -> Dict:
    """
    Use Ollama to analyze transaction patterns and identify whale behavior
    
    Args:
        transactions: DataFrame with transaction data
        
    Returns:
        Dictionary with analysis results and recommendations
    """
    # Prepare data summary for AI analysis
    summary = {
        'total_transactions': len(transactions),
        'total_value': transactions['value_usd'].sum(),
        'average_value': transactions['value_usd'].mean(),
        'top_addresses': transactions.groupby('from')['value_usd'].sum().head(10).to_dict(),
        'time_distribution': transactions['timestamp'].value_counts().head(10).to_dict()
    }
    
    # Create analysis prompt
    prompt = f"""
    Analyze the following cryptocurrency transaction data for whale activity patterns:
    
    Transaction Summary:
    - Total transactions: {summary['total_transactions']}
    - Total value: ${summary['total_value']:,.2f}
    - Average value: ${summary['average_value']:,.2f}
    
    Top sending addresses by volume:
    {json.dumps(summary['top_addresses'], indent=2)}
    
    Time distribution:
    {json.dumps(summary['time_distribution'], indent=2)}
    
    Please provide:
    1. Whale activity assessment (High/Medium/Low)
    2. Suspicious patterns identified
    3. Market impact prediction
    4. Recommended trading actions
    5. Risk level assessment
    
    Format your response as JSON with clear sections.
    """
    
    # Get AI analysis
    response = self.ollama_client.generate(
        model='llama2:7b',
        prompt=prompt,
        options={'temperature': 0.3}  # Lower temperature for more consistent analysis
    )
    
    try:
        analysis = json.loads(response['response'])
        return analysis
    except json.JSONDecodeError:
        # Fallback to text analysis if JSON parsing fails
        return {'raw_analysis': response['response']}

class RealTimeWhaleMonitor:
    def __init__(self, tracker: WhaleTracker, update_interval: int = 300):
        """
        Initialize real-time monitoring system
        
        Args:
            tracker: WhaleTracker instance
            update_interval: Update frequency in seconds
        """
        self.tracker = tracker
        self.update_interval = update_interval
        self.is_monitoring = False
        self.alerts = []
    
    def start_monitoring(self):
        """Start continuous whale monitoring"""
        self.is_monitoring = True
        print("🐋 Whale monitoring started...")
        
        while self.is_monitoring:
            try:
                # Fetch recent transactions
                transactions = self.tracker.fetch_recent_transactions(hours=1)
                
                if not transactions.empty:
                    # Analyze for whale activity
                    analysis = self.tracker.analyze_transaction_patterns(transactions)
                    
                    # Check for significant movements
                    whale_movements = self.tracker.detect_whale_movements(transactions)
                    
                    # Generate alerts
                    for movement in whale_movements:
                        self.generate_alert(movement, analysis)
                
                # Wait before next update
                time.sleep(self.update_interval)
                
            except Exception as e:
                print(f"❌ Monitoring error: {e}")
                time.sleep(60)  # Wait 1 minute before retrying
    
    def generate_alert(self, movement: Dict, analysis: Dict):
        """Generate and send whale movement alert"""
        alert = {
            'timestamp': datetime.now().isoformat(),
            'type': 'whale_movement',
            'address': movement['from'],
            'value_usd': movement['value_usd'],
            'risk_level': analysis.get('risk_level', 'medium'),
            'recommendation': analysis.get('recommended_actions', ['monitor'])
        }
        
        self.alerts.append(alert)
        self.send_alert(alert)
    
    def send_alert(self, alert: Dict):
        """Send alert notification (implement your preferred method)"""
        print(f"🚨 WHALE ALERT: ${alert['value_usd']:,.2f} moved by {alert['address'][:10]}...")
        print(f"📊 Risk Level: {alert['risk_level'].upper()}")
        print(f"💡 Recommendation: {', '.join(alert['recommendation'])}")

def detect_volume_anomalies(self, df: pd.DataFrame, window: int = 24) -> List[Dict]:
    """
    Detect unusual volume patterns that indicate whale activity
    
    Args:
        df: Transaction DataFrame
        window: Hours to look back for baseline calculation
        
    Returns:
        List of detected anomalies
    """
    anomalies = []
    
    # Calculate rolling volume statistics
    df['hour'] = pd.to_datetime(df['timestamp']).dt.hour
    hourly_volume = df.groupby('hour')['value_usd'].sum()
    
    # Detect volume spikes
    volume_mean = hourly_volume.rolling(window=window, min_periods=1).mean()
    volume_std = hourly_volume.rolling(window=window, min_periods=1).std()
    
    # Identify anomalies (volume > mean + 2*std)
    for hour, volume in hourly_volume.items():
        if volume > volume_mean[hour] + 2 * volume_std[hour]:
            anomalies.append({
                'hour': hour,
                'volume': volume,
                'threshold': volume_mean[hour] + 2 * volume_std[hour],
                'severity': 'high' if volume > volume_mean[hour] + 3 * volume_std[hour] else 'medium'
            })
    
    return anomalies

def cluster_whale_addresses(self, df: pd.DataFrame) -> Dict[str, List[str]]:
    """
    Cluster addresses based on transaction patterns to identify whale networks
    
    Args:
        df: Transaction DataFrame
        
    Returns:
        Dictionary mapping cluster IDs to address lists
    """
    from sklearn.cluster import DBSCAN
    from sklearn.preprocessing import StandardScaler
    
    # Create feature matrix for addresses
    address_features = df.groupby('from').agg({
        'value_usd': ['sum', 'mean', 'count'],
        'timestamp': ['min', 'max']
    }).reset_index()
    
    # Flatten column names
    address_features.columns = ['address', 'total_value', 'avg_value', 'tx_count', 'first_tx', 'last_tx']
    
    # Prepare features for clustering
    features = address_features[['total_value', 'avg_value', 'tx_count']].fillna(0)
    scaler = StandardScaler()
    features_scaled = scaler.fit_transform(features)
    
    # Perform clustering
    clusterer = DBSCAN(eps=0.5, min_samples=2)
    clusters = clusterer.fit_predict(features_scaled)
    
    # Group addresses by cluster
    cluster_groups = {}
    for i, cluster_id in enumerate(clusters):
        if cluster_id != -1:  # Ignore noise points
            if cluster_id not in cluster_groups:
                cluster_groups[cluster_id] = []
            cluster_groups[cluster_id].append(address_features.iloc[i]['address'])
    
    return cluster_groups

import matplotlib.pyplot as plt
import seaborn as sns
from matplotlib.animation import FuncAnimation

class WhaleVisualization:
    def __init__(self, tracker: WhaleTracker):
        self.tracker = tracker
        self.fig, self.axes = plt.subplots(2, 2, figsize=(15, 10))
        self.fig.suptitle('Crypto Whale Activity Dashboard', fontsize=16)
    
    def create_volume_chart(self, df: pd.DataFrame):
        """Create volume over time chart"""
        ax = self.axes[0, 0]
        ax.clear()
        
        # Group by hour and sum volumes
        hourly_volume = df.groupby(df['timestamp'].dt.hour)['value_usd'].sum()
        
        ax.bar(hourly_volume.index, hourly_volume.values, color='steelblue', alpha=0.7)
        ax.set_title('Hourly Transaction Volume')
        ax.set_xlabel('Hour')
        ax.set_ylabel('Volume (USD)')
        ax.grid(True, alpha=0.3)
    
    def create_whale_distribution(self, df: pd.DataFrame):
        """Create whale size distribution"""
        ax = self.axes[0, 1]
        ax.clear()
        
        # Create volume bins
        bins = [100000, 500000, 1000000, 5000000, 10000000, float('inf')]
        labels = ['100K-500K', '500K-1M', '1M-5M', '5M-10M', '10M+']
        
        df['volume_category'] = pd.cut(df['value_usd'], bins=bins, labels=labels)
        category_counts = df['volume_category'].value_counts()
        
        ax.pie(category_counts.values, labels=category_counts.index, autopct='%1.1f%%')
        ax.set_title('Whale Transaction Distribution')
    
    def create_address_network(self, df: pd.DataFrame):
        """Create address interaction network"""
        ax = self.axes[1, 0]
        ax.clear()
        
        # Top 20 addresses by volume
        top_addresses = df.groupby('from')['value_usd'].sum().head(20)
        
        ax.barh(range(len(top_addresses)), top_addresses.values)
        ax.set_yticks(range(len(top_addresses)))
        ax.set_yticklabels([f"{addr[:8]}..." for addr in top_addresses.index])
        ax.set_title('Top Whale Addresses')
        ax.set_xlabel('Total Volume (USD)')
    
    def create_risk_heatmap(self, analysis_results: List[Dict]):
        """Create risk assessment heatmap"""
        ax = self.axes[1, 1]
        ax.clear()
        
        # Create risk matrix
        risk_data = np.random.rand(10, 10)  # Replace with actual risk calculations
        
        sns.heatmap(risk_data, cmap='Reds', ax=ax, cbar_kws={'label': 'Risk Level'})
        ax.set_title('Market Risk Heatmap')
        ax.set_xlabel('Time Period')
        ax.set_ylabel('Asset Category')

class WhaleReportGenerator:
    def __init__(self, tracker: WhaleTracker):
        self.tracker = tracker
        self.template = """
        # Whale Activity Report
        Generated: {timestamp}
        
        ## Executive Summary
        - Total whale transactions: {total_transactions}
        - Total volume: ${total_volume:,.2f}
        - Risk level: {risk_level}
        
        ## Key Findings
        {key_findings}
        
        ## Recommendations
        {recommendations}
        
        ## Detailed Analysis
        {detailed_analysis}
        """
    
    def generate_daily_report(self, df: pd.DataFrame, analysis: Dict) -> str:
        """Generate daily whale activity report"""
        return self.template.format(
            timestamp=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            total_transactions=len(df),
            total_volume=df['value_usd'].sum(),
            risk_level=analysis.get('risk_level', 'Unknown'),
            key_findings=self.format_findings(analysis.get('key_findings', [])),
            recommendations=self.format_recommendations(analysis.get('recommendations', [])),
            detailed_analysis=analysis.get('detailed_analysis', 'No detailed analysis available')
        )
    
    def format_findings(self, findings: List[str]) -> str:
        """Format key findings for report"""
        return '\n'.join([f"- {finding}" for finding in findings])
    
    def format_recommendations(self, recommendations: List[str]) -> str:
        """Format recommendations for report"""
        return '\n'.join([f"- {rec}" for rec in recommendations])

class OptimizedWhaleTracker:
    def __init__(self, cache_size: int = 10000):
        self.cache = {}
        self.cache_size = cache_size
        self.processed_transactions = set()
    
    def process_transactions_batch(self, transactions: List[Dict]) -> pd.DataFrame:
        """Process transactions in batches for better performance"""
        batch_size = 1000
        processed_data = []
        
        for i in range(0, len(transactions), batch_size):
            batch = transactions[i:i + batch_size]
            
            # Filter out already processed transactions
            new_batch = [tx for tx in batch if tx['hash'] not in self.processed_transactions]
            
            if new_batch:
                # Process batch
                df_batch = pd.DataFrame(new_batch)
                df_batch = self.enrich_transaction_data(df_batch)
                processed_data.append(df_batch)
                
                # Update processed set
                self.processed_transactions.update([tx['hash'] for tx in new_batch])
        
        return pd.concat(processed_data, ignore_index=True) if processed_data else pd.DataFrame()
    
    def enrich_transaction_data(self, df: pd.DataFrame) -> pd.DataFrame:
        """Add computed fields and classifications"""
        # Add USD values
        df['value_usd'] = df['value'] * self.get_cached_price('ethereum')
        
        # Add whale classification
        df['is_whale'] = df['value_usd'] >= 1000000
        
        # Add time-based features
        df['timestamp'] = pd.to_datetime(df['timestamp'])
        df['hour'] = df['timestamp'].dt.hour
        df['day_of_week'] = df['timestamp'].dt.dayofweek
        
        return df

import gc
from functools import lru_cache

class MemoryOptimizedTracker:
    def __init__(self, max_history_hours: int = 168):  # 1 week
        self.max_history_hours = max_history_hours
        self.transaction_buffer = []
        self.analysis_cache = {}
    
    @lru_cache(maxsize=128)
    def get_cached_analysis(self, data_hash: str) -> Dict:
        """Cache analysis results to avoid recomputation"""
        pass
    
    def cleanup_old_data(self):
        """Remove old data to free memory"""
        cutoff_time = datetime.now() - timedelta(hours=self.max_history_hours)
        
        # Clean transaction buffer
        self.transaction_buffer = [
            tx for tx in self.transaction_buffer 
            if datetime.fromisoformat(tx['timestamp']) > cutoff_time
        ]
        
        # Clean analysis cache
        old_keys = [
            key for key, value in self.analysis_cache.items()
            if value['timestamp'] < cutoff_time
        ]
        
        for key in old_keys:
            del self.analysis_cache[key]
        
        # Force garbage collection
        gc.collect()

import os
from cryptography.fernet import Fernet

class SecureConfig:
    def __init__(self, config_file: str = '.env.encrypted'):
        self.config_file = config_file
        self.cipher = Fernet(self.generate_key())
        
    def generate_key(self) -> bytes:
        """Generate or load encryption key"""
        key_file = '.key'
        if os.path.exists(key_file):
            with open(key_file, 'rb') as f:
                return f.read()
        else:
            key = Fernet.generate_key()
            with open(key_file, 'wb') as f:
                f.write(key)
            return key
    
    def encrypt_config(self, config: Dict[str, str]):
        """Encrypt configuration data"""
        config_json = json.dumps(config)
        encrypted_data = self.cipher.encrypt(config_json.encode())
        
        with open(self.config_file, 'wb') as f:
            f.write(encrypted_data)
    
    def decrypt_config(self) -> Dict[str, str]:
        """Decrypt configuration data"""
        with open(self.config_file, 'rb') as f:
            encrypted_data = f.read()
        
        decrypted_data = self.cipher.decrypt(encrypted_data)
        return json.loads(decrypted_data.decode())

import time
from functools import wraps

def rate_limit(calls_per_minute: int = 60):
    """Decorator to implement rate limiting"""
    def decorator(func):
        last_calls = []
        
        @wraps(func)
        def wrapper(*args, **kwargs):
            now = time.time()
            
            # Remove calls older than 1 minute
            last_calls[:] = [call_time for call_time in last_calls if now - call_time < 60]
            
            # Check if we've exceeded the limit
            if len(last_calls) >= calls_per_minute:
                sleep_time = 60 - (now - last_calls[0])
                if sleep_time > 0:
                    time.sleep(sleep_time)
            
            # Record this call
            last_calls.append(now)
            
            return func(*args, **kwargs)
        
        return wrapper
    return decorator

class RobustWhaleTracker:
    @rate_limit(calls_per_minute=30)
    def fetch_blockchain_data(self, endpoint: str) -> Dict:
        """Fetch data with rate limiting and error handling"""
        max_retries = 3
        retry_delay = 1
        
        for attempt in range(max_retries):
            try:
                response = requests.get(endpoint, timeout=10)
                response.raise_for_status()
                return response.json()
                
            except requests.exceptions.Timeout:
                print(f"⏱️ Timeout on attempt {attempt + 1}")
                if attempt < max_retries - 1:
                    time.sleep(retry_delay * (2 ** attempt))
                    
            except requests.exceptions.HTTPError as e:
                print(f"❌ HTTP error: {e}")
                if e.response.status_code == 429:  # Rate limited
                    time.sleep(retry_delay * (2 ** attempt))
                else:
                    raise
                    
            except Exception as e:
                print(f"❌ Unexpected error: {e}")
                if attempt < max_retries - 1:
                    time.sleep(retry_delay * (2 ** attempt))
                else:
                    raise
        
        raise Exception("Max retries exceeded")

FROM python:3.11-slim

WORKDIR /app

# Install system dependencies
RUN apt-get update && apt-get install -y \
    curl \
    && rm -rf /var/lib/apt/lists/*

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

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

# Copy application code
COPY . .

# Expose port
EXPOSE 8000

# Start command
CMD ["python", "main.py"]

import logging
from prometheus_client import Counter, Histogram, Gauge, start_http_server

# Metrics
transaction_counter = Counter('whale_transactions_total', 'Total whale transactions processed')
processing_time = Histogram('whale_processing_seconds', 'Time spent processing transactions')
active_whales = Gauge('active_whales_count', 'Number of active whale addresses')

class ProductionWhaleTracker:
    def __init__(self):
        self.setup_logging()
        self.setup_metrics()
        
    def setup_logging(self):
        """Configure production logging"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('whale_tracker.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    
    def setup_metrics(self):
        """Start Prometheus metrics server"""
        start_http_server(8000)
        self.logger.info("Metrics server started on port 8000")
    
    @processing_time.time()
    def process_whale_transaction(self, transaction: Dict):
        """Process transaction with metrics"""
        try:
            # Process transaction
            result = self.analyze_transaction(transaction)
            
            # Update metrics
            transaction_counter.inc()
            if result['is_whale']:
                active_whales.inc()
            
            self.logger.info(f"Processed transaction: {transaction['hash']}")
            return result
            
        except Exception as e:
            self.logger.error(f"Error processing transaction: {e}")
            raise

def diagnose_performance_issues(self):
    """Diagnose and fix common performance problems"""
    import psutil
    import gc
    
    # Check memory usage
    memory_percent = psutil.virtual_memory().percent
    if memory_percent > 80:
        print(f"⚠️ High memory usage: {memory_percent}%")
        
        # Force garbage collection
        gc.collect()
        
        # Clear caches
        self.clear_caches()
        
        # Reduce data retention
        self.cleanup_old_data()
    
    # Check CPU usage
    cpu_percent = psutil.cpu_percent(interval=1)
    if cpu_percent > 80:
        print(f"⚠️ High CPU usage: {cpu_percent}%")
        
        # Reduce processing frequency
        self.update_interval = max(self.update_interval * 1.5, 60)
        
    # Check disk space
    disk_usage = psutil.disk_usage('/').percent
    if disk_usage > 90:
        print(f"⚠️ Low disk space: {disk_usage}% used")
        
        # Clean log files
        self.rotate_logs()

def test_api_connections(self):
    """Test all API connections and report status"""
    results = {}
    
    # Test blockchain API
    try:
        response = requests.get("https://api.etherscan.io/api?module=stats&action=ethsupply", timeout=5)
        results['etherscan'] = response.status_code == 200
    except:
        results['etherscan'] = False
    
    # Test price API
    try:
        response = requests.get("https://api.coingecko.com/api/v3/ping", timeout=5)
        results['coingecko'] = response.status_code == 200
    except:
        results['coingecko'] = False
    
    # Test Ollama
    try:
        response = self.ollama_client.list()
        results['ollama'] = True
    except:
        results['ollama'] = False
    
    # Report results
    for service, status in results.items():
        status_emoji = "✅" if status else "❌"
        print(f"{status_emoji} {service}: {'Connected' if status else 'Failed'}")
    
    return results