Market Manipulation Detection: Ollama Whale Activity and Pump Schemes

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

# Download financial analysis models
ollama pull deepseek-r1:7b
ollama pull llama3.1:8b
ollama pull arcee-ai/llama3-sec

pip install ollama requests pandas numpy websocket-client ccxt

# whale_detector.py
import ollama
import requests
import pandas as pd
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import asyncio

class WhaleActivityDetector:
    def __init__(self, model_name="deepseek-r1:7b"):
        self.model = model_name
        self.client = ollama.Client()
        self.whale_threshold = 1000000  # $1M USD threshold
        self.alert_history = []
    
    def fetch_whale_transactions(self, symbol: str, timeframe: str = "1h") -> List[Dict]:
        """Fetch large transactions from blockchain explorers"""
        # Example integration with Whale Alert API
        url = f"https://api.whale-alert.io/v1/transactions"
        headers = {"X-WA-API-KEY": "your_api_key"}
        
        params = {
            "symbol": symbol.lower(),
            "min_value": self.whale_threshold,
            "start": int((datetime.now() - timedelta(hours=24)).timestamp()),
            "limit": 100
        }
        
        try:
            response = requests.get(url, headers=headers, params=params)
            return response.json().get("transactions", [])
        except Exception as e:
            print(f"Error fetching whale data: {e}")
            return []
    
    def analyze_whale_patterns(self, transactions: List[Dict]) -> Dict:
        """Use Ollama to analyze whale transaction patterns"""
        if not transactions:
            return {"risk_level": "low", "pattern_detected": False}
        
        # Prepare data summary for AI analysis
        total_volume = sum(tx.get("amount_usd", 0) for tx in transactions)
        exchange_inflows = sum(tx.get("amount_usd", 0) for tx in transactions 
                              if tx.get("to_owner_type") == "exchange")
        exchange_outflows = sum(tx.get("amount_usd", 0) for tx in transactions 
                               if tx.get("from_owner_type") == "exchange")
        
        analysis_prompt = f"""
        Analyze these whale transaction patterns for market manipulation:
        
        Total Volume (24h): ${total_volume:,.0f}
        Exchange Inflows: ${exchange_inflows:,.0f}
        Exchange Outflows: ${exchange_outflows:,.0f}
        Transaction Count: {len(transactions)}
        
        Recent Transactions:
        {self.format_transactions(transactions[:5])}
        
        Identify:
        1. Manipulation risk level (low/medium/high)
        2. Pattern type (accumulation/distribution/pump-prep/dump-signal)
        3. Confidence score (0-100)
        4. Time urgency (immediate/hours/days)
        5. Recommended action (monitor/caution/alert)
        
        Provide analysis in JSON format.
        """
        
        try:
            response = self.client.chat(
                model=self.model,
                messages=[{"role": "user", "content": analysis_prompt}]
            )
            return self.parse_ai_response(response['message']['content'])
        except Exception as e:
            print(f"AI analysis error: {e}")
            return {"risk_level": "unknown", "error": str(e)}
    
    def format_transactions(self, transactions: List[Dict]) -> str:
        """Format transaction data for AI analysis"""
        formatted = []
        for tx in transactions:
            formatted.append(
                f"- ${tx.get('amount_usd', 0):,.0f} | "
                f"{tx.get('from_owner_type', 'unknown')} → "
                f"{tx.get('to_owner_type', 'unknown')} | "
                f"{tx.get('timestamp', 'N/A')}"
            )
        return "\n".join(formatted)
    
    def parse_ai_response(self, response: str) -> Dict:
        """Parse AI response into structured data"""
        try:
            import json
            # Extract JSON from response (handle potential markdown formatting)
            start = response.find('{')
            end = response.rfind('}') + 1
            if start != -1 and end != 0:
                return json.loads(response[start:end])
        except:
            pass
        
        # Fallback parsing for non-JSON responses
        risk_level = "medium"
        if "high risk" in response.lower() or "manipulation" in response.lower():
            risk_level = "high"
        elif "low risk" in response.lower() or "normal" in response.lower():
            risk_level = "low"
            
        return {
            "risk_level": risk_level,
            "analysis": response,
            "confidence": 50
        }

# pump_detector.py
import ollama
import pandas as pd
import numpy as np
from typing import Dict, List, Tuple

class PumpSchemeDetector:
    def __init__(self, model_name="llama3.1:8b"):
        self.model = model_name
        self.client = ollama.Client()
        self.pump_indicators = {
            "volume_spike": 5.0,     # 5x normal volume
            "price_spike": 0.3,      # 30% price increase
            "social_buzz": 0.7,      # 70% positive sentiment
            "new_holders": 0.5       # 50% increase in holders
        }
    
    def collect_market_data(self, symbol: str) -> Dict:
        """Collect comprehensive market data for analysis"""
        return {
            "price_data": self.get_price_metrics(symbol),
            "volume_data": self.get_volume_metrics(symbol),
            "social_data": self.get_social_metrics(symbol),
            "blockchain_data": self.get_blockchain_metrics(symbol)
        }
    
    def get_price_metrics(self, symbol: str) -> Dict:
        """Calculate price-based manipulation indicators"""
        # This would integrate with your preferred exchange API
        # Example using a generic structure
        return {
            "current_price": 0.00,
            "price_change_1h": 0.00,
            "price_change_24h": 0.00,
            "volatility": 0.00,
            "support_levels": [],
            "resistance_levels": []
        }
    
    def get_volume_metrics(self, symbol: str) -> Dict:
        """Analyze trading volume patterns"""
        return {
            "current_volume": 0,
            "avg_volume_7d": 0,
            "volume_ratio": 0.00,
            "buy_sell_ratio": 0.00,
            "large_orders": []
        }
    
    def get_social_metrics(self, symbol: str) -> Dict:
        """Gather social sentiment and buzz metrics"""
        return {
            "sentiment_score": 0.00,
            "mention_count": 0,
            "influencer_activity": 0,
            "telegram_activity": 0,
            "reddit_activity": 0
        }
    
    def get_blockchain_metrics(self, symbol: str) -> Dict:
        """Analyze on-chain activity"""
        return {
            "holder_count": 0,
            "new_holders": 0,
            "token_distribution": {},
            "contract_interactions": 0,
            "liquidity_changes": 0.00
        }
    
    def detect_pump_scheme(self, symbol: str) -> Dict:
        """Main detection algorithm using AI analysis"""
        market_data = self.collect_market_data(symbol)
        
        # Calculate pump probability scores
        pump_signals = self.calculate_pump_signals(market_data)
        
        # Generate AI analysis
        ai_analysis = self.ai_pump_analysis(symbol, market_data, pump_signals)
        
        return {
            "symbol": symbol,
            "pump_probability": pump_signals["overall_score"],
            "risk_factors": pump_signals["risk_factors"],
            "ai_analysis": ai_analysis,
            "timestamp": datetime.now().isoformat(),
            "recommended_action": self.get_recommendation(pump_signals["overall_score"])
        }
    
    def calculate_pump_signals(self, data: Dict) -> Dict:
        """Calculate quantitative pump scheme indicators"""
        signals = {}
        risk_factors = []
        
        # Volume Analysis
        volume_ratio = data["volume_data"]["volume_ratio"]
        if volume_ratio > self.pump_indicators["volume_spike"]:
            signals["volume_signal"] = min(volume_ratio / 10, 1.0)
            risk_factors.append(f"Volume spike: {volume_ratio:.1f}x normal")
        else:
            signals["volume_signal"] = 0.0
        
        # Price Movement Analysis
        price_change = abs(data["price_data"]["price_change_1h"])
        if price_change > self.pump_indicators["price_spike"]:
            signals["price_signal"] = min(price_change / 0.5, 1.0)
            risk_factors.append(f"Rapid price movement: {price_change:.1%}")
        else:
            signals["price_signal"] = 0.0
        
        # Social Sentiment Analysis
        sentiment = data["social_data"]["sentiment_score"]
        if sentiment > self.pump_indicators["social_buzz"]:
            signals["social_signal"] = sentiment
            risk_factors.append(f"High social buzz: {sentiment:.1%} positive")
        else:
            signals["social_signal"] = 0.0
        
        # Calculate overall pump probability
        overall_score = np.mean(list(signals.values()))
        
        return {
            "individual_signals": signals,
            "overall_score": overall_score,
            "risk_factors": risk_factors
        }
    
    def ai_pump_analysis(self, symbol: str, market_data: Dict, pump_signals: Dict) -> str:
        """Use AI to provide detailed pump scheme analysis"""
        analysis_prompt = f"""
        Analyze this cryptocurrency for potential pump scheme activity:
        
        Token: {symbol}
        
        Market Metrics:
        - Current Price: ${market_data['price_data']['current_price']}
        - 1h Change: {market_data['price_data']['price_change_1h']:.2%}
        - 24h Change: {market_data['price_data']['price_change_24h']:.2%}
        - Volume Ratio: {market_data['volume_data']['volume_ratio']:.1f}x
        - Social Sentiment: {market_data['social_data']['sentiment_score']:.1%}
        
        Pump Signals Detected:
        {pump_signals['risk_factors']}
        
        Overall Pump Probability: {pump_signals['overall_score']:.1%}
        
        Provide analysis covering:
        1. Likelihood this is a coordinated pump (0-100%)
        2. Key evidence supporting/contradicting pump theory
        3. Estimated timeline (if pump detected)
        4. Risk level for potential investors
        5. Signs to watch for confirming/denying pump activity
        
        Focus on actionable insights for risk assessment.
        """
        
        try:
            response = self.client.chat(
                model=self.model,
                messages=[{"role": "user", "content": analysis_prompt}]
            )
            return response['message']['content']
        except Exception as e:
            return f"Analysis error: {e}"
    
    def get_recommendation(self, pump_score: float) -> str:
        """Provide trading recommendations based on pump probability"""
        if pump_score > 0.8:
            return "HIGH RISK: Avoid trading. Likely pump scheme in progress."
        elif pump_score > 0.6:
            return "MEDIUM RISK: Exercise extreme caution. Monitor closely."
        elif pump_score > 0.4:
            return "LOW RISK: Some suspicious activity. Use small position sizes."
        else:
            return "NORMAL: Standard market activity detected."

# monitor.py
import asyncio
import time
from datetime import datetime
from whale_detector import WhaleActivityDetector
from pump_detector import PumpSchemeDetector

class MarketManipulationMonitor:
    def __init__(self, symbols: List[str]):
        self.symbols = symbols
        self.whale_detector = WhaleActivityDetector()
        self.pump_detector = PumpSchemeDetector()
        self.alerts = []
        self.running = False
    
    async def start_monitoring(self, interval: int = 300):  # 5-minute intervals
        """Start continuous monitoring of specified symbols"""
        self.running = True
        print(f"🔍 Starting market manipulation monitoring for {len(self.symbols)} symbols")
        
        while self.running:
            try:
                tasks = []
                for symbol in self.symbols:
                    tasks.append(self.analyze_symbol(symbol))
                
                results = await asyncio.gather(*tasks, return_exceptions=True)
                
                # Process results and generate alerts
                for symbol, result in zip(self.symbols, results):
                    if isinstance(result, Exception):
                        print(f"❌ Error analyzing {symbol}: {result}")
                        continue
                    
                    self.process_analysis_result(symbol, result)
                
                # Wait before next analysis cycle
                await asyncio.sleep(interval)
                
            except KeyboardInterrupt:
                print("\n🛑 Monitoring stopped by user")
                break
            except Exception as e:
                print(f"❌ Monitoring error: {e}")
                await asyncio.sleep(30)  # Brief pause before retry
    
    async def analyze_symbol(self, symbol: str) -> Dict:
        """Perform comprehensive analysis on a single symbol"""
        print(f"📊 Analyzing {symbol}...")
        
        # Run whale and pump detection in parallel
        whale_task = asyncio.create_task(self.run_whale_analysis(symbol))
        pump_task = asyncio.create_task(self.run_pump_analysis(symbol))
        
        whale_result, pump_result = await asyncio.gather(whale_task, pump_task)
        
        return {
            "whale_analysis": whale_result,
            "pump_analysis": pump_result,
            "timestamp": datetime.now().isoformat()
        }
    
    async def run_whale_analysis(self, symbol: str) -> Dict:
        """Run whale activity detection"""
        transactions = self.whale_detector.fetch_whale_transactions(symbol)
        return self.whale_detector.analyze_whale_patterns(transactions)
    
    async def run_pump_analysis(self, symbol: str) -> Dict:
        """Run pump scheme detection"""
        return self.pump_detector.detect_pump_scheme(symbol)
    
    def process_analysis_result(self, symbol: str, result: Dict):
        """Process analysis results and generate alerts"""
        whale_risk = result["whale_analysis"].get("risk_level", "unknown")
        pump_prob = result["pump_analysis"].get("pump_probability", 0.0)
        
        # Generate alerts based on risk levels
        if whale_risk == "high" or pump_prob > 0.7:
            alert = {
                "symbol": symbol,
                "alert_type": "HIGH_RISK",
                "whale_risk": whale_risk,
                "pump_probability": pump_prob,
                "timestamp": datetime.now().isoformat(),
                "message": f"🚨 HIGH RISK: {symbol} shows strong manipulation signals"
            }
            self.alerts.append(alert)
            print(alert["message"])
        elif whale_risk == "medium" or pump_prob > 0.5:
            print(f"⚠️  MEDIUM RISK: {symbol} - Monitor closely (Pump: {pump_prob:.1%})")
        else:
            print(f"✅ {symbol} - Normal activity detected")
    
    def stop_monitoring(self):
        """Stop the monitoring process"""
        self.running = False
    
    def get_recent_alerts(self, hours: int = 24) -> List[Dict]:
        """Get alerts from the last N hours"""
        cutoff_time = datetime.now() - timedelta(hours=hours)
        return [alert for alert in self.alerts 
                if datetime.fromisoformat(alert["timestamp"]) > cutoff_time]

# Usage example
async def main():
    # Define symbols to monitor
    watch_list = ["BTC", "ETH", "SOL", "DOGE", "SHIB"]
    
    # Create and start monitor
    monitor = MarketManipulationMonitor(watch_list)
    await monitor.start_monitoring(interval=300)  # 5-minute intervals

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

def multi_timeframe_analysis(self, symbol: str) -> Dict:
    """Analyze manipulation patterns across multiple timeframes"""
    timeframes = ["5m", "1h", "4h", "1d"]
    analysis_results = {}
    
    for tf in timeframes:
        data = self.collect_timeframe_data(symbol, tf)
        analysis_results[tf] = self.analyze_timeframe(data, tf)
    
    # Use AI to synthesize multi-timeframe insights
    synthesis_prompt = f"""
    Analyze these multi-timeframe manipulation signals for {symbol}:
    
    {self.format_timeframe_results(analysis_results)}
    
    Identify:
    1. Consistent patterns across timeframes
    2. Divergences that indicate manipulation
    3. Overall manipulation confidence (0-100%)
    4. Primary timeframe showing strongest signals
    5. Recommended monitoring approach
    """
    
    return self.get_ai_synthesis(synthesis_prompt)

def detect_behavioral_patterns(self, market_data: Dict) -> Dict:
    """Detect specific manipulation behavioral patterns"""
    patterns = {
        "wash_trading": self.detect_wash_trading(market_data),
        "spoofing": self.detect_spoofing(market_data),
        "layering": self.detect_layering(market_data),
        "pump_dump": self.detect_pump_dump_pattern(market_data)
    }
    
    return {
        "detected_patterns": [k for k, v in patterns.items() if v["detected"]],
        "confidence_scores": {k: v["confidence"] for k, v in patterns.items()},
        "evidence": {k: v["evidence"] for k, v in patterns.items() if v["detected"]}
    }

# Dockerfile
FROM python:3.11-slim

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

# Set working directory
WORKDIR /app

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

# Copy application code
COPY . .

# Download required models
RUN ollama pull deepseek-r1:7b
RUN ollama pull llama3.1:8b

# Expose ports
EXPOSE 8000

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

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="Market Manipulation Detection API")

class AnalysisRequest(BaseModel):
    symbol: str
    analysis_type: str = "full"  # "whale", "pump", or "full"

@app.post("/analyze")
async def analyze_symbol(request: AnalysisRequest):
    """Analyze a symbol for market manipulation"""
    if request.analysis_type == "whale":
        result = await run_whale_analysis(request.symbol)
    elif request.analysis_type == "pump":
        result = await run_pump_analysis(request.symbol)
    else:
        result = await analyze_symbol_comprehensive(request.symbol)
    
    return {"symbol": request.symbol, "analysis": result}

class ResourceOptimizer:
    def __init__(self):
        self.model_cache = {}
        self.analysis_cache = {}
    
    def get_model(self, model_name: str):
        """Implement model caching to reduce memory usage"""
        if model_name not in self.model_cache:
            self.model_cache[model_name] = ollama.Client(model=model_name)
        return self.model_cache[model_name]
    
    def cache_analysis(self, symbol: str, result: Dict, ttl: int = 300):
        """Cache analysis results to reduce computation"""
        cache_key = f"{symbol}_{int(time.time() // ttl)}"
        self.analysis_cache[cache_key] = result

# security.py
import hashlib
import hmac
from cryptography.fernet import Fernet

class SecurityManager:
    def __init__(self, encryption_key: bytes):
        self.cipher = Fernet(encryption_key)
    
    def encrypt_analysis_data(self, data: Dict) -> str:
        """Encrypt sensitive analysis data"""
        json_data = json.dumps(data).encode()
        return self.cipher.encrypt(json_data).decode()
    
    def validate_api_request(self, request_data: str, signature: str, secret: str) -> bool:
        """Validate API request signatures"""
        expected_signature = hmac.new(
            secret.encode(),
            request_data.encode(),
            hashlib.sha256
        ).hexdigest()
        return hmac.compare_digest(signature, expected_signature)

# docker-compose.yml
version: '3.8'
services:
  manipulation-detector:
    build: .
    ports:
      - "127.0.0.1:8000:8000"  # Bind to localhost only
    environment:
      - OLLAMA_HOST=0.0.0.0:11434
      - API_SECRET_KEY=${API_SECRET_KEY}
    volumes:
      - ./data:/app/data:ro  # Read-only data mount
    restart: unless-stopped