NFT Floor Price Prediction with Ollama: Collection Trend Analysis Guide

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

# Pull the required model for Data Analysis
ollama pull llama2:13b
ollama pull codellama:7b

# requirements.txt
import requests
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime, timedelta
import ollama
import json
import time
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split

class NFTDataCollector:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.opensea.io/api/v1"
        self.headers = {
            "X-API-KEY": api_key,
            "User-Agent": "NFT-Analysis-Tool"
        }
    
    def get_collection_stats(self, collection_slug):
        """Fetch current collection statistics"""
        url = f"{self.base_url}/collection/{collection_slug}/stats"
        response = requests.get(url, headers=self.headers)
        
        if response.status_code == 200:
            return response.json()['stats']
        else:
            print(f"Error fetching data: {response.status_code}")
            return None
    
    def get_recent_sales(self, collection_slug, limit=50):
        """Get recent sales data for trend analysis"""
        url = f"{self.base_url}/events"
        params = {
            "collection_slug": collection_slug,
            "event_type": "successful",
            "limit": limit
        }
        
        response = requests.get(url, headers=self.headers, params=params)
        return response.json()['asset_events'] if response.status_code == 200 else []

# Initialize collector
collector = NFTDataCollector("your_opensea_api_key")

def track_floor_prices(collections, interval_minutes=5):
    """Track floor prices across multiple collections"""
    price_history = {}
    
    for collection in collections:
        print(f"Tracking {collection}...")
        stats = collector.get_collection_stats(collection)
        
        if stats:
            current_price = stats['floor_price']
            timestamp = datetime.now()
            
            if collection not in price_history:
                price_history[collection] = []
            
            price_history[collection].append({
                'timestamp': timestamp,
                'floor_price': current_price,
                'volume_24h': stats['one_day_volume'],
                'sales_24h': stats['one_day_sales']
            })
    
    return price_history

# Track popular collections
collections = ['cryptopunks', 'bored-ape-yacht-club', 'azuki']
price_data = track_floor_prices(collections)

class OllamaNFTAnalyzer:
    def __init__(self, model_name="llama2:13b"):
        self.model = model_name
        self.client = ollama.Client()
    
    def analyze_market_sentiment(self, collection_data):
        """Use Ollama to analyze market sentiment patterns"""
        prompt = f"""
        Analyze this NFT collection data and identify key trends:
        
        Collection Stats:
        - Floor Price: {collection_data['floor_price']} ETH
        - 24h Volume: {collection_data['volume_24h']} ETH  
        - 24h Sales: {collection_data['sales_24h']}
        - Price Change: {collection_data.get('price_change_24h', 'N/A')}%
        
        Identify:
        1. Current trend direction (bullish/bearish/neutral)
        2. Volume-price correlation strength
        3. Potential resistance/support levels
        4. Risk factors for next 24-48 hours
        
        Provide structured analysis with confidence scores.
        """
        
        response = self.client.generate(
            model=self.model,
            prompt=prompt,
            options={'temperature': 0.3}  # Lower temperature for consistent analysis
        )
        
        return response['response']
    
    def predict_price_movement(self, historical_data):
        """Generate price movement predictions"""
        # Prepare data summary for Ollama
        recent_prices = [point['floor_price'] for point in historical_data[-10:]]
        price_trend = "increasing" if recent_prices[-1] > recent_prices[0] else "decreasing"
        volatility = np.std(recent_prices) / np.mean(recent_prices)
        
        prediction_prompt = f"""
        Based on this NFT collection's recent price history:
        
        Recent prices (last 10 data points): {recent_prices}
        Current trend: {price_trend}
        Volatility coefficient: {volatility:.3f}
        
        Predict the likely floor price movement in the next:
        - 6 hours
        - 24 hours  
        - 7 days
        
        Include:
        - Specific price targets with confidence intervals
        - Key risk factors
        - Market conditions that could invalidate predictions
        
        Format as structured data for automated processing.
        """
        
        response = self.client.generate(
            model=self.model,
            prompt=prediction_prompt,
            options={'temperature': 0.2}
        )
        
        return self.parse_prediction_response(response['response'])
    
    def parse_prediction_response(self, response_text):
        """Parse Ollama's prediction into structured data"""
        # This would parse the response into a structured format
        # Implementation depends on your specific response format needs
        return {
            'prediction_text': response_text,
            'timestamp': datetime.now(),
            'confidence': 'medium'  # Would be extracted from response
        }

# Initialize analyzer
analyzer = OllamaNFTAnalyzer()

def analyze_collection_correlations(price_histories):
    """Find correlation patterns between different NFT collections"""
    correlation_matrix = {}
    collections = list(price_histories.keys())
    
    for i, collection1 in enumerate(collections):
        for j, collection2 in enumerate(collections[i+1:], i+1):
            # Extract price arrays for correlation calculation
            prices1 = [point['floor_price'] for point in price_histories[collection1]]
            prices2 = [point['floor_price'] for point in price_histories[collection2]]
            
            # Ensure equal length arrays
            min_length = min(len(prices1), len(prices2))
            if min_length > 5:  # Need minimum data points
                correlation = np.corrcoef(
                    prices1[-min_length:], 
                    prices2[-min_length:]
                )[0, 1]
                
                correlation_matrix[f"{collection1}_{collection2}"] = correlation
    
    return correlation_matrix

# Analyze correlations
correlations = analyze_collection_correlations(price_data)
print("Collection Correlations:")
for pair, correlation in correlations.items():
    print(f"{pair}: {correlation:.3f}")

def calculate_risk_metrics(price_history):
    """Calculate comprehensive risk metrics for NFT collections"""
    prices = [point['floor_price'] for point in price_history]
    returns = np.diff(prices) / prices[:-1]  # Calculate returns
    
    metrics = {
        'volatility_daily': np.std(returns) * np.sqrt(24),  # Assuming hourly data
        'max_drawdown': calculate_max_drawdown(prices),
        'sharpe_ratio': np.mean(returns) / np.std(returns) if np.std(returns) > 0 else 0,
        'var_95': np.percentile(returns, 5),  # Value at Risk (95% confidence)
        'price_momentum': (prices[-1] - prices[-5]) / prices[-5] if len(prices) >= 5 else 0
    }
    
    return metrics

def calculate_max_drawdown(prices):
    """Calculate maximum drawdown from peak"""
    peak = prices[0]
    max_dd = 0
    
    for price in prices:
        if price > peak:
            peak = price
        else:
            drawdown = (peak - price) / peak
            max_dd = max(max_dd, drawdown)
    
    return max_dd

# Calculate risk for each collection
risk_profiles = {}
for collection, history in price_data.items():
    if len(history) > 10:  # Need sufficient data
        risk_profiles[collection] = calculate_risk_metrics(history)

def engineer_prediction_features(collection_history):
    """Create features for machine learning prediction"""
    df = pd.DataFrame(collection_history)
    
    # Technical indicators
    df['price_ma_5'] = df['floor_price'].rolling(window=5).mean()
    df['price_ma_20'] = df['floor_price'].rolling(window=20).mean()
    df['rsi'] = calculate_rsi(df['floor_price'])
    df['volume_trend'] = df['volume_24h'].pct_change()
    
    # Time-based features
    df['hour'] = df['timestamp'].dt.hour
    df['day_of_week'] = df['timestamp'].dt.dayofweek
    df['is_weekend'] = df['day_of_week'].isin([5, 6])
    
    # Volatility features
    df['price_volatility'] = df['floor_price'].rolling(window=10).std()
    df['volume_volatility'] = df['volume_24h'].rolling(window=10).std()
    
    return df

def calculate_rsi(prices, period=14):
    """Calculate Relative Strength Index"""
    delta = prices.diff()
    gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
    rs = gain / loss
    rsi = 100 - (100 / (1 + rs))
    return rsi

class NFTPredictionEnsemble:
    def __init__(self, ollama_analyzer):
        self.ollama = ollama_analyzer
        self.scaler = StandardScaler()
    
    def prepare_training_data(self, collection_histories):
        """Prepare data for model training"""
        all_features = []
        all_targets = []
        
        for collection, history in collection_histories.items():
            if len(history) < 50:  # Need sufficient history
                continue
            
            df = engineer_prediction_features(history)
            df = df.dropna()
            
            # Create target variable (next period price change)
            df['target'] = df['floor_price'].shift(-1) / df['floor_price'] - 1
            df = df.dropna()
            
            # Select features
            feature_cols = ['price_ma_5', 'price_ma_20', 'rsi', 'volume_trend', 
                          'hour', 'day_of_week', 'price_volatility']
            
            features = df[feature_cols].values
            targets = df['target'].values
            
            all_features.append(features)
            all_targets.append(targets)
        
        return np.vstack(all_features), np.concatenate(all_targets)
    
    def generate_ensemble_prediction(self, current_data):
        """Combine multiple prediction methods"""
        # Get Ollama's qualitative analysis
        ollama_analysis = self.ollama.analyze_market_sentiment(current_data)
        ollama_prediction = self.ollama.predict_price_movement([current_data])
        
        # Get quantitative predictions (would implement ML models here)
        technical_prediction = self.technical_analysis_prediction(current_data)
        
        # Combine predictions with weights
        ensemble_prediction = {
            'ollama_analysis': ollama_analysis,
            'ollama_prediction': ollama_prediction,
            'technical_prediction': technical_prediction,
            'confidence_score': self.calculate_ensemble_confidence(),
            'recommendation': self.generate_recommendation()
        }
        
        return ensemble_prediction
    
    def technical_analysis_prediction(self, data):
        """Simple technical analysis prediction"""
        # Placeholder for technical analysis
        return {
            'direction': 'neutral',
            'magnitude': 0.05,
            'timeframe': '24h'
        }
    
    def calculate_ensemble_confidence(self):
        """Calculate confidence based on prediction agreement"""
        # Implement confidence calculation logic
        return 0.75  # Placeholder
    
    def generate_recommendation(self):
        """Generate trading recommendation"""
        return "HOLD - Wait for clearer trend signals"

# Initialize ensemble predictor
ensemble = NFTPredictionEnsemble(analyzer)

class NFTAlertSystem:
    def __init__(self, prediction_ensemble, alert_thresholds):
        self.ensemble = prediction_ensemble
        self.thresholds = alert_thresholds
        self.alert_history = []
    
    def monitor_collections(self, collections):
        """Continuously monitor collections for alert conditions"""
        while True:
            for collection in collections:
                current_data = collector.get_collection_stats(collection)
                
                if current_data:
                    prediction = self.ensemble.generate_ensemble_prediction(current_data)
                    alerts = self.check_alert_conditions(collection, current_data, prediction)
                    
                    for alert in alerts:
                        self.send_alert(alert)
            
            time.sleep(300)  # Check every 5 minutes
    
    def check_alert_conditions(self, collection, data, prediction):
        """Check if any alert conditions are met"""
        alerts = []
        
        # Price movement alerts
        if 'technical_prediction' in prediction:
            predicted_change = prediction['technical_prediction']['magnitude']
            if predicted_change > self.thresholds['price_increase']:
                alerts.append({
                    'type': 'PRICE_PUMP_PREDICTED',
                    'collection': collection,
                    'predicted_change': predicted_change,
                    'confidence': prediction['confidence_score']
                })
        
        # Volume spike alerts
        if data['volume_24h'] > self.thresholds['volume_spike']:
            alerts.append({
                'type': 'VOLUME_SPIKE',
                'collection': collection,
                'volume': data['volume_24h']
            })
        
        return alerts
    
    def send_alert(self, alert):
        """Send alert notification"""
        message = f"🚨 NFT Alert: {alert['type']} for {alert['collection']}"
        print(f"{datetime.now()}: {message}")
        self.alert_history.append({
            'timestamp': datetime.now(),
            'alert': alert
        })

# Set up alert system
alert_thresholds = {
    'price_increase': 0.15,  # 15% predicted increase
    'price_decrease': -0.20,  # 20% predicted decrease
    'volume_spike': 100.0     # 100 ETH volume spike
}

alert_system = NFTAlertSystem(ensemble, alert_thresholds)

def backtest_predictions(historical_data, prediction_model, lookback_days=30):
    """Backtest prediction accuracy over historical data"""
    results = []
    
    for i in range(lookback_days, len(historical_data)):
        # Use data up to point i for prediction
        training_data = historical_data[:i]
        actual_next_price = historical_data[i]['floor_price']
        
        # Generate prediction
        prediction = prediction_model.predict_price_movement(training_data)
        
        # Calculate accuracy metrics
        if i > 0:
            actual_change = (actual_next_price - historical_data[i-1]['floor_price']) / historical_data[i-1]['floor_price']
            
            results.append({
                'timestamp': historical_data[i]['timestamp'],
                'predicted_direction': prediction.get('direction', 'neutral'),
                'actual_change': actual_change,
                'prediction_confidence': prediction.get('confidence', 0.5)
            })
    
    return analyze_backtest_results(results)

def analyze_backtest_results(results):
    """Analyze backtesting performance"""
    df = pd.DataFrame(results)
    
    # Calculate accuracy metrics
    correct_direction = 0
    total_predictions = len(df)
    
    for _, row in df.iterrows():
        predicted_up = row['predicted_direction'] == 'bullish'
        actual_up = row['actual_change'] > 0
        
        if (predicted_up and actual_up) or (not predicted_up and not actual_up):
            correct_direction += 1
    
    accuracy = correct_direction / total_predictions if total_predictions > 0 else 0
    
    return {
        'directional_accuracy': accuracy,
        'total_predictions': total_predictions,
        'avg_confidence': df['prediction_confidence'].mean(),
        'performance_summary': f"Accuracy: {accuracy:.2%} over {total_predictions} predictions"
    }

# Run backtesting
if len(price_data['cryptopunks']) > 50:
    backtest_results = backtest_predictions(
        price_data['cryptopunks'], 
        analyzer, 
        lookback_days=20
    )
    print(f"Backtesting Results: {backtest_results['performance_summary']}")

def create_prediction_dashboard(collections_data, predictions):
    """Create comprehensive prediction dashboard"""
    fig, axes = plt.subplots(2, 2, figsize=(15, 10))
    fig.suptitle('NFT Floor Price Prediction Dashboard', fontsize=16)
    
    # Plot 1: Price trends
    ax1 = axes[0, 0]
    for collection, data in collections_data.items():
        timestamps = [point['timestamp'] for point in data]
        prices = [point['floor_price'] for point in data]
        ax1.plot(timestamps, prices, label=collection, marker='o', markersize=3)
    
    ax1.set_title('Floor Price Trends')
    ax1.set_xlabel('Time')
    ax1.set_ylabel('Floor Price (ETH)')
    ax1.legend()
    ax1.grid(True, alpha=0.3)
    
    # Plot 2: Volume analysis
    ax2 = axes[0, 1]
    collection_names = list(collections_data.keys())
    latest_volumes = [data[-1]['volume_24h'] for data in collections_data.values()]
    
    bars = ax2.bar(collection_names, latest_volumes, color=['blue', 'orange', 'green'])
    ax2.set_title('24h Volume Comparison')
    ax2.set_ylabel('Volume (ETH)')
    
    # Add value labels on bars
    for bar, volume in zip(bars, latest_volumes):
        height = bar.get_height()
        ax2.text(bar.get_x() + bar.get_width()/2., height + height*0.01,
                f'{volume:.1f}', ha='center', va='bottom')
    
    # Plot 3: Prediction confidence
    ax3 = axes[1, 0]
    if predictions:
        confidence_scores = [pred.get('confidence_score', 0.5) for pred in predictions.values()]
        ax3.bar(predictions.keys(), confidence_scores, color='lightcoral')
        ax3.set_title('Prediction Confidence Scores')
        ax3.set_ylabel('Confidence')
        ax3.set_ylim(0, 1)
    
    # Plot 4: Risk heatmap
    ax4 = axes[1, 1]
    if len(collections_data) > 1:
        risk_matrix = np.random.rand(len(collections_data), 3)  # Placeholder
        im = ax4.imshow(risk_matrix, cmap='RdYlGn_r', aspect='auto')
        ax4.set_title('Risk Assessment Heatmap')
        ax4.set_xticks(range(3))
        ax4.set_xticklabels(['Volatility', 'Volume Risk', 'Trend Risk'])
        ax4.set_yticks(range(len(collections_data)))
        ax4.set_yticklabels(collections_data.keys())
    
    plt.tight_layout()
    plt.savefig('nft_prediction_dashboard.png', dpi=300, bbox_inches='tight')
    plt.show()

# Generate dashboard
predictions_summary = {}
for collection in collections:
    if collection in price_data and len(price_data[collection]) > 5:
        latest_data = price_data[collection][-1]
        predictions_summary[collection] = ensemble.generate_ensemble_prediction(latest_data)

create_prediction_dashboard(price_data, predictions_summary)

# deployment_config.py
DEPLOYMENT_CONFIG = {
    'data_collection': {
        'update_interval': 300,  # 5 minutes
        'batch_size': 10,
        'api_rate_limit': 4,  # requests per second
        'retry_attempts': 3
    },
    'prediction_engine': {
        'model_update_frequency': 3600,  # 1 hour
        'ollama_timeout': 30,
        'ensemble_weights': {
            'ollama': 0.4,
            'technical': 0.3,
            'sentiment': 0.3
        }
    },
    'storage': {
        'database_url': 'postgresql://localhost:5432/nft_predictions',
        'redis_url': 'redis://localhost:6379',
        'backup_frequency': 86400  # 24 hours
    },
    'monitoring': {
        'log_level': 'INFO',
        'metrics_endpoint': '/metrics',
        'health_check_interval': 60
    }
}

class ProductionNFTPredictor:
    def __init__(self, config):
        self.config = config
        self.setup_logging()
        self.setup_database()
        self.setup_monitoring()
    
    def setup_logging(self):
        """Configure production logging"""
        import logging
        logging.basicConfig(
            level=getattr(logging, self.config['monitoring']['log_level']),
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('nft_predictor.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    
    def setup_database(self):
        """Initialize database connections"""
        # Database setup code would go here
        self.logger.info("Database connections established")
    
    def setup_monitoring(self):
        """Initialize monitoring and metrics"""
        # Monitoring setup code would go here
        self.logger.info("Monitoring systems initialized")
    
    def run_production_pipeline(self):
        """Main production pipeline"""
        self.logger.info("Starting NFT prediction pipeline")
        
        try:
            # Data collection
            self.collect_market_data()
            
            # Generate predictions
            self.generate_predictions()
            
            # Update models
            self.update_models()
            
            # Send alerts
            self.process_alerts()
            
        except Exception as e:
            self.logger.error(f"Pipeline error: {str(e)}")
            self.handle_error(e)
    
    def collect_market_data(self):
        """Collect data with error handling and rate limiting"""
        # Implementation with robust error handling
        pass
    
    def generate_predictions(self):
        """Generate predictions with fallback mechanisms"""
        # Implementation with fallback models
        pass
    
    def handle_error(self, error):
        """Handle production errors gracefully"""
        # Error handling and recovery logic
        pass

# Initialize production system
if __name__ == "__main__":
    predictor = ProductionNFTPredictor(DEPLOYMENT_CONFIG)
    predictor.run_production_pipeline()

def robust_ollama_query(analyzer, prompt, max_retries=3):
    """Implement retry logic for consistent Ollama responses"""
    for attempt in range(max_retries):
        try:
            response = analyzer.client.generate(
                model=analyzer.model,
                prompt=prompt,
                options={
                    'temperature': 0.2,  # Lower for consistency
                    'top_p': 0.9,
                    'repeat_penalty': 1.1
                }
            )
            
            # Validate response quality
            if len(response['response']) > 50:  # Minimum response length
                return response['response']
            
        except Exception as e:
            print(f"Ollama query attempt {attempt + 1} failed: {e}")
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)  # Exponential backoff
    
    return "Analysis temporarily unavailable"

import time
from functools import wraps

def rate_limit(calls_per_second=1):
    """Decorator to enforce API rate limiting"""
    min_interval = 1.0 / calls_per_second
    last_called = [0.0]
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            left_to_wait = min_interval - elapsed
            if left_to_wait > 0:
                time.sleep(left_to_wait)
            ret = func(*args, **kwargs)
            last_called[0] = time.time()
            return ret
        return wrapper
    return decorator

# Apply to data collection methods
@rate_limit(calls_per_second=2)
def rate_limited_api_call(url, headers):
    return requests.get(url, headers=headers)

def validate_nft_data(data):
    """Validate NFT data quality before analysis"""
    required_fields = ['floor_price', 'volume_24h', 'timestamp']
    
    # Check required fields
    for field in required_fields:
        if field not in data or data[field] is None:
            return False, f"Missing required field: {field}"
    
    # Validate price ranges
    if data['floor_price'] <= 0 or data['floor_price'] > 10000:
        return False, f"Invalid floor price: {data['floor_price']}"
    
    # Validate volume
    if data['volume_24h'] < 0:
        return False, f"Invalid volume: {data['volume_24h']}"
    
    return True, "Data validation passed"

# Use in data collection pipeline
def safe_data_collection(collection_slug):
    """Collect data with validation"""
    raw_data = collector.get_collection_stats(collection_slug)
    
    if raw_data:
        is_valid, message = validate_nft_data(raw_data)
        if is_valid:
            return raw_data
        else:
            print(f"Data validation failed for {collection_slug}: {message}")
    
    return None