Computer Vision NFT Farming: AI Image Analysis for Rare Traits

# requirements.txt
opencv-python==4.8.1.78
pillow==10.0.0
numpy==1.24.3
requests==2.31.0
scikit-learn==1.3.0
pandas==2.0.3
web3==6.9.0
python-dotenv==1.0.0

pip install -r requirements.txt

# nft_analyzer.py
import cv2
import numpy as np
from PIL import Image
import requests
from sklearn.cluster import KMeans
import pandas as pd
from collections import Counter
import json

class NFTImageAnalyzer:
    def __init__(self):
        self.trait_database = {}
        self.rarity_scores = {}
        
    def download_nft_image(self, image_url):
        """Download NFT image from IPFS or HTTP URL"""
        try:
            response = requests.get(image_url, timeout=10)
            response.raise_for_status()
            
            # Convert to PIL Image
            image = Image.open(io.BytesIO(response.content))
            # Convert to OpenCV format (BGR)
            opencv_image = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2BGR)
            return opencv_image
        except Exception as e:
            print(f"Error downloading image: {e}")
            return None
    
    def extract_color_palette(self, image, n_colors=5):
        """Extract dominant colors from NFT image"""
        # Reshape image to be a list of pixels
        data = image.reshape((-1, 3))
        data = np.float32(data)
        
        # Apply K-means clustering to find dominant colors
        criteria = (cv2.TERM_CRITERIA_EPS + cv2.TERM_CRITERIA_MAX_ITER, 20, 1.0)
        _, labels, centers = cv2.kmeans(data, n_colors, None, criteria, 10, cv2.KMEANS_RANDOM_CENTERS)
        
        # Convert back to uint8 and return color palette
        centers = np.uint8(centers)
        return centers.tolist()
    
    def detect_shapes_and_features(self, image):
        """Detect geometric shapes and visual features"""
        gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
        
        # Edge detection for shape analysis
        edges = cv2.Canny(gray, 50, 150)
        
        # Find contours for shape detection
        contours, _ = cv2.findContours(edges, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
        
        features = {
            'total_contours': len(contours),
            'large_contours': len([c for c in contours if cv2.contourArea(c) > 1000]),
            'edge_density': np.sum(edges > 0) / edges.size
        }
        
        return features
    
    def analyze_texture_patterns(self, image):
        """Analyze texture patterns using Local Binary Patterns"""
        gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
        
        # Calculate histogram of gradients for texture analysis
        hist = cv2.calcHist([gray], [0], None, [256], [0, 256])
        
        # Texture features
        texture_features = {
            'histogram_variance': np.var(hist),
            'brightness_mean': np.mean(gray),
            'contrast_std': np.std(gray)
        }
        
        return texture_features

# metadata_processor.py
import requests
import json
from web3 import Web3

class NFTMetadataProcessor:
    def __init__(self, provider_url):
        self.w3 = Web3(Web3.HTTPProvider(provider_url))
        
    def get_nft_metadata(self, contract_address, token_id):
        """Fetch NFT metadata from contract or API"""
        # Example using OpenSea API (replace with your preferred method)
        url = f"https://api.opensea.io/api/v1/asset/{contract_address}/{token_id}/"
        
        headers = {
            "X-API-KEY": "your-api-key"  # Replace with actual API key
        }
        
        try:
            response = requests.get(url, headers=headers)
            response.raise_for_status()
            return response.json()
        except Exception as e:
            print(f"Error fetching metadata: {e}")
            return None
    
    def extract_traits_from_metadata(self, metadata):
        """Extract trait information from NFT metadata"""
        if not metadata or 'traits' not in metadata:
            return {}
        
        traits = {}
        for trait in metadata['traits']:
            trait_type = trait.get('trait_type', 'unknown')
            trait_value = trait.get('value', 'unknown')
            traits[trait_type] = trait_value
        
        return traits
    
    def calculate_trait_rarity(self, collection_traits):
        """Calculate rarity scores for traits across collection"""
        trait_counts = {}
        total_nfts = len(collection_traits)
        
        # Count occurrences of each trait value
        for nft_traits in collection_traits:
            for trait_type, trait_value in nft_traits.items():
                key = f"{trait_type}:{trait_value}"
                trait_counts[key] = trait_counts.get(key, 0) + 1
        
        # Calculate rarity scores (lower count = higher rarity)
        rarity_scores = {}
        for trait_key, count in trait_counts.items():
            rarity_score = 1 / (count / total_nfts)
            rarity_scores[trait_key] = rarity_score
        
        return rarity_scores

# nft_farming_pipeline.py
import asyncio
import aiohttp
from concurrent.futures import ThreadPoolExecutor
import time

class NFTFarmingPipeline:
    def __init__(self, contract_address, provider_url):
        self.contract_address = contract_address
        self.image_analyzer = NFTImageAnalyzer()
        self.metadata_processor = NFTMetadataProcessor(provider_url)
        self.collection_data = []
        
    async def analyze_collection(self, token_range):
        """Analyze entire NFT collection for rare traits"""
        start_time = time.time()
        
        # Process tokens in parallel for speed
        tasks = []
        for token_id in token_range:
            task = self.analyze_single_nft(token_id)
            tasks.append(task)
        
        # Execute analysis in batches to avoid rate limits
        batch_size = 10
        for i in range(0, len(tasks), batch_size):
            batch = tasks[i:i + batch_size]
            results = await asyncio.gather(*batch, return_exceptions=True)
            
            # Process results
            for result in results:
                if isinstance(result, dict):
                    self.collection_data.append(result)
            
            # Rate limiting pause
            await asyncio.sleep(1)
        
        end_time = time.time()
        print(f"Analyzed {len(self.collection_data)} NFTs in {end_time - start_time:.2f} seconds")
        
        return self.calculate_collection_rarity()
    
    async def analyze_single_nft(self, token_id):
        """Analyze individual NFT for traits and rarity"""
        try:
            # Fetch metadata
            metadata = self.metadata_processor.get_nft_metadata(
                self.contract_address, token_id
            )
            
            if not metadata or not metadata.get('image_url'):
                return None
            
            # Download and analyze image
            image = self.image_analyzer.download_nft_image(metadata['image_url'])
            if image is None:
                return None
            
            # Extract visual features
            color_palette = self.image_analyzer.extract_color_palette(image)
            shape_features = self.image_analyzer.detect_shapes_and_features(image)
            texture_features = self.image_analyzer.analyze_texture_patterns(image)
            
            # Extract metadata traits
            metadata_traits = self.metadata_processor.extract_traits_from_metadata(metadata)
            
            # Combine all analysis results
            nft_analysis = {
                'token_id': token_id,
                'metadata_traits': metadata_traits,
                'visual_features': {
                    'dominant_colors': color_palette,
                    'shape_features': shape_features,
                    'texture_features': texture_features
                },
                'image_url': metadata['image_url'],
                'name': metadata.get('name', f'Token #{token_id}')
            }
            
            return nft_analysis
            
        except Exception as e:
            print(f"Error analyzing token {token_id}: {e}")
            return None
    
    def calculate_collection_rarity(self):
        """Calculate rarity scores for entire collection"""
        if not self.collection_data:
            return {}
        
        # Extract all traits for rarity calculation
        all_traits = [nft['metadata_traits'] for nft in self.collection_data]
        rarity_scores = self.metadata_processor.calculate_trait_rarity(all_traits)
        
        # Calculate visual feature rarity
        visual_rarity = self.calculate_visual_rarity()
        
        # Combine metadata and visual rarity scores
        final_scores = {}
        for nft in self.collection_data:
            token_id = nft['token_id']
            
            # Metadata rarity score
            metadata_score = 0
            for trait_type, trait_value in nft['metadata_traits'].items():
                trait_key = f"{trait_type}:{trait_value}"
                metadata_score += rarity_scores.get(trait_key, 1)
            
            # Visual rarity score
            visual_score = visual_rarity.get(token_id, 1)
            
            # Combined rarity score (weighted)
            final_score = (metadata_score * 0.7) + (visual_score * 0.3)
            final_scores[token_id] = final_score
        
        return final_scores
    
    def calculate_visual_rarity(self):
        """Calculate rarity based on visual features"""
        visual_scores = {}
        
        # Extract dominant colors for comparison
        all_colors = []
        for nft in self.collection_data:
            colors = nft['visual_features']['dominant_colors']
            all_colors.extend(colors)
        
        # Count unique color combinations
        color_counter = Counter(tuple(color) for color in all_colors)
        
        # Calculate visual rarity scores
        for nft in self.collection_data:
            token_id = nft['token_id']
            colors = nft['visual_features']['dominant_colors']
            
            # Calculate color rarity
            color_rarity = sum(1 / color_counter[tuple(color)] for color in colors)
            
            # Add shape and texture rarity factors
            shape_score = nft['visual_features']['shape_features']['total_contours']
            texture_score = nft['visual_features']['texture_features']['histogram_variance']
            
            visual_scores[token_id] = color_rarity + (shape_score * 0.1) + (texture_score * 0.001)
        
        return visual_scores
    
    def export_analysis_results(self, filename='nft_analysis_results.json'):
        """Export analysis results to JSON file"""
        export_data = {
            'collection_address': self.contract_address,
            'analysis_timestamp': time.time(),
            'total_analyzed': len(self.collection_data),
            'nft_data': self.collection_data,
            'rarity_rankings': self.calculate_collection_rarity()
        }
        
        with open(filename, 'w') as f:
            json.dump(export_data, f, indent=2)
        
        print(f"Analysis results exported to {filename}")

mkdir nft-computer-vision
cd nft-computer-vision
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate
pip install -r requirements.txt

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

class Config:
    # Blockchain settings
    ETHEREUM_PROVIDER_URL = os.getenv('ETHEREUM_PROVIDER_URL', 'https://mainnet.infura.io/v3/your-key')
    OPENSEA_API_KEY = os.getenv('OPENSEA_API_KEY', 'your-opensea-key')
    
    # Analysis parameters
    BATCH_SIZE = 10
    RATE_LIMIT_DELAY = 1.0
    MAX_RETRIES = 3
    
    # Image processing settings
    DOMINANT_COLORS_COUNT = 5
    MIN_CONTOUR_AREA = 1000
    
    # Rarity calculation weights
    METADATA_WEIGHT = 0.7
    VISUAL_WEIGHT = 0.3

# main.py
import asyncio
from nft_farming_pipeline import NFTFarmingPipeline
from config import Config

async def main():
    # Configure target collection
    contract_address = "0xBC4CA0EdA7647A8aB7C2061c2E118A18a936f13D"  # BAYC example
    token_range = range(1, 101)  # Analyze first 100 tokens
    
    # Initialize pipeline
    pipeline = NFTFarmingPipeline(contract_address, Config.ETHEREUM_PROVIDER_URL)
    
    print(f"Starting analysis of {len(token_range)} NFTs...")
    
    # Run complete analysis
    rarity_scores = await pipeline.analyze_collection(token_range)
    
    # Display top rare NFTs
    sorted_nfts = sorted(rarity_scores.items(), key=lambda x: x[1], reverse=True)
    
    print("\nTop 10 Rarest NFTs Found:")
    for i, (token_id, score) in enumerate(sorted_nfts[:10], 1):
        print(f"{i}. Token #{token_id}: Rarity Score {score:.2f}")
    
    # Export results
    pipeline.export_analysis_results('rare_nfts_analysis.json')
    
    print("\nAnalysis complete! Check rare_nfts_analysis.json for detailed results.")

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

def detect_sift_features(self, image):
    """Detect unique visual features using SIFT algorithm"""
    gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
    
    # Initialize SIFT detector
    sift = cv2.SIFT_create()
    
    # Detect keypoints and descriptors
    keypoints, descriptors = sift.detectAndCompute(gray, None)
    
    return {
        'keypoint_count': len(keypoints),
        'feature_descriptors': descriptors.tolist() if descriptors is not None else [],
        'feature_strength': np.mean([kp.response for kp in keypoints]) if keypoints else 0
    }

def detect_specific_traits(self, image, trait_templates):
    """Detect specific traits using template matching"""
    gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
    detected_traits = []
    
    for trait_name, template_path in trait_templates.items():
        template = cv2.imread(template_path, 0)
        
        # Perform template matching
        result = cv2.matchTemplate(gray, template, cv2.TM_CCOEFF_NORMED)
        min_val, max_val, min_loc, max_loc = cv2.minMaxLoc(result)
        
        # If match confidence is high enough
        if max_val > 0.8:
            detected_traits.append({
                'trait': trait_name,
                'confidence': max_val,
                'location': max_loc
            })
    
    return detected_traits

from multiprocessing import Pool, cpu_count
import functools

def optimize_batch_processing(self, token_ids, batch_size=None):
    """Process NFT analysis in parallel batches"""
    if batch_size is None:
        batch_size = cpu_count() * 2
    
    # Create worker function with bound parameters
    worker_func = functools.partial(self.analyze_single_nft_sync)
    
    # Process in parallel
    with Pool(processes=cpu_count()) as pool:
        results = pool.map(worker_func, token_ids)
    
    # Filter successful results
    valid_results = [r for r in results if r is not None]
    return valid_results

import pickle
import hashlib
from pathlib import Path

class AnalysisCache:
    def __init__(self, cache_dir='cache'):
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(exist_ok=True)
    
    def get_cache_key(self, contract_address, token_id):
        """Generate unique cache key for NFT"""
        key_string = f"{contract_address}:{token_id}"
        return hashlib.md5(key_string.encode()).hexdigest()
    
    def cache_analysis(self, contract_address, token_id, analysis_data):
        """Cache analysis results to disk"""
        cache_key = self.get_cache_key(contract_address, token_id)
        cache_file = self.cache_dir / f"{cache_key}.pkl"
        
        with open(cache_file, 'wb') as f:
            pickle.dump(analysis_data, f)
    
    def load_cached_analysis(self, contract_address, token_id):
        """Load cached analysis if available"""
        cache_key = self.get_cache_key(contract_address, token_id)
        cache_file = self.cache_dir / f"{cache_key}.pkl"
        
        if cache_file.exists():
            with open(cache_file, 'rb') as f:
                return pickle.load(f)
        return None