Advanced Usage

from decimal import Decimal
from typing import Dict, Any, Callable, List
from gmgnapi import TokenFilter, GmGnEnhancedClient
import asyncio

class AdvancedTokenFilter:
    """
    Advanced filtering system with custom rules and machine learning integration.
    """
    
    def __init__(self):
        self.custom_rules: List[Callable] = []
        self.scoring_weights = {
            'market_cap_score': 0.3,
            'volume_score': 0.25,
            'liquidity_score': 0.2,
            'community_score': 0.15,
            'risk_score': 0.1
        }
        
        # Historical data for pattern analysis
        self.token_history = {}
        
    def add_custom_rule(self, rule_func: Callable[[Dict[str, Any]], bool]):
        """Add a custom filtering rule."""
        self.custom_rules.append(rule_func)
    
    def calculate_token_score(self, token_data: Dict[str, Any]) -> float:
        """Calculate a comprehensive score for token quality."""
        
        scores = {
            'market_cap_score': self._score_market_cap(token_data),
            'volume_score': self._score_volume(token_data),
            'liquidity_score': self._score_liquidity(token_data),
            'community_score': self._score_community(token_data),
            'risk_score': self._score_risk(token_data)
        }
        
        # Weighted average
        total_score = sum(
            scores[key] * self.scoring_weights[key] 
            for key in scores
        )
        
        return min(max(total_score, 0), 100)  # Clamp to 0-100
    
    def _score_market_cap(self, token_data: Dict[str, Any]) -> float:
        """Score based on market cap (0-100)."""
        mc = token_data.get('market_cap', 0)
        
        if mc < 10000:
            return 0
        elif mc < 100000:
            return 20
        elif mc < 1000000:
            return 60
        elif mc < 10000000:
            return 90
        else:
            return 100
    
    def _score_volume(self, token_data: Dict[str, Any]) -> float:
        """Score based on 24h volume activity."""
        volume = token_data.get('volume_24h', 0)
        mc = token_data.get('market_cap', 1)
        
        volume_ratio = volume / mc if mc > 0 else 0
        
        if volume_ratio < 0.01:
            return 0
        elif volume_ratio < 0.05:
            return 30
        elif volume_ratio < 0.1:
            return 60
        elif volume_ratio < 0.3:
            return 90
        else:
            return 100
    
    def _score_liquidity(self, token_data: Dict[str, Any]) -> float:
        """Score based on liquidity depth."""
        liquidity = token_data.get('liquidity', 0)
        
        if liquidity < 1000:
            return 0
        elif liquidity < 10000:
            return 30
        elif liquidity < 50000:
            return 60
        elif liquidity < 200000:
            return 90
        else:
            return 100
    
    def _score_community(self, token_data: Dict[str, Any]) -> float:
        """Score based on community metrics."""
        holders = token_data.get('holder_count', 0)
        
        if holders < 10:
            return 0
        elif holders < 100:
            return 25
        elif holders < 1000:
            return 50
        elif holders < 5000:
            return 75
        else:
            return 100
    
    def _score_risk(self, token_data: Dict[str, Any]) -> float:
        """Score based on risk factors (higher score = lower risk)."""
        risk_factors = 0
        
        # Check for risk indicators
        if token_data.get('creator_balance_ratio', 0) > 0.5:
            risk_factors += 30  # High creator holding
        
        if token_data.get('top_10_ratio', 0) > 0.8:
            risk_factors += 25  # High concentration
        
        if token_data.get('burn_rate', 0) > 0.1:
            risk_factors += 20  # High burn rate
        
        # Suspicious name patterns
        symbol = token_data.get('symbol', '').upper()
        suspicious_keywords = ['MOON', 'SAFE', 'BABY', 'DOGE', 'PEPE', 'TEST']
        if any(keyword in symbol for keyword in suspicious_keywords):
            risk_factors += 15
        
        return max(0, 100 - risk_factors)
    
    async def apply_filters(self, token_data: Dict[str, Any]) -> Dict[str, Any]:
        """Apply all filters and return results."""
        
        # Calculate base score
        score = self.calculate_token_score(token_data)
        
        # Apply custom rules
        custom_results = []
        for rule in self.custom_rules:
            try:
                result = rule(token_data)
                custom_results.append(result)
            except Exception as e:
                print(f"Custom rule error: {e}")
                custom_results.append(False)
        
        # Determine if token passes filters
        passes_score = score >= 60  # Minimum score threshold
        passes_custom = all(custom_results) if custom_results else True
        
        return {
            'score': score,
            'passes_filter': passes_score and passes_custom,
            'custom_rule_results': custom_results,
            'breakdown': {
                'market_cap_score': self._score_market_cap(token_data),
                'volume_score': self._score_volume(token_data),
                'liquidity_score': self._score_liquidity(token_data),
                'community_score': self._score_community(token_data),
                'risk_score': self._score_risk(token_data)
            }
        }

# Example usage with custom rules
async def advanced_filtering_example():
    """Example of advanced filtering with custom rules."""
    
    filter_system = AdvancedTokenFilter()
    
    # Add custom rules
    def no_meme_coins(token_data):
        """Filter out obvious meme coins."""
        symbol = token_data.get('symbol', '').upper()
        meme_keywords = ['DOGE', 'SHIB', 'PEPE', 'FLOKI', 'BABY']
        return not any(keyword in symbol for keyword in meme_keywords)
    
    def minimum_age_rule(token_data):
        """Require tokens to be at least 1 hour old."""
        created_time = token_data.get('created_timestamp', 0)
        current_time = asyncio.get_event_loop().time()
        age_hours = (current_time - created_time) / 3600
        return age_hours >= 1
    
    def whale_activity_rule(token_data):
        """Check for healthy whale activity."""
        top_10_ratio = token_data.get('top_10_ratio', 1)
        return 0.3 <= top_10_ratio <= 0.7  # Balanced distribution
    
    # Register custom rules
    filter_system.add_custom_rule(no_meme_coins)
    filter_system.add_custom_rule(minimum_age_rule)
    filter_system.add_custom_rule(whale_activity_rule)
    
    # Use with enhanced client
    async with GmGnEnhancedClient() as client:
        await client.subscribe_new_pools()
        
        @client.on_new_pool
        async def analyze_with_advanced_filter(pool_data):
            for pool in pool_data.pools:
                if pool.bti:
                    token_info = {
                        'symbol': pool.bti.s,
                        'market_cap': pool.bti.mc,
                        'volume_24h': pool.bti.v24h,
                        'holder_count': pool.bti.hc,
                        'liquidity': pool.il,
                        'created_timestamp': asyncio.get_event_loop().time(),
                        'top_10_ratio': pool.bti.t10hr or 0.5,
                        'creator_balance_ratio': pool.bti.cbr or 0.1
                    }
                    
                    result = await filter_system.apply_filters(token_info)
                    
                    if result['passes_filter']:
                        print(f"✅ HIGH QUALITY TOKEN: {token_info['symbol']}")
                        print(f"   Score: {result['score']:.1f}/100")
                        print(f"   Breakdown: {result['breakdown']}")
                    else:
                        print(f"❌ Filtered out: {token_info['symbol']} (Score: {result['score']:.1f})")
        
        await client.listen()

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

import os
import ssl
import hashlib
from cryptography.fernet import Fernet
from typing import Optional

class SecureConfigManager:
    """Secure configuration and credential management."""
    
    def __init__(self, encryption_key: Optional[bytes] = None):
        if encryption_key:
            self.cipher = Fernet(encryption_key)
        else:
            # Generate new key (store securely!)
            self.cipher = Fernet(Fernet.generate_key())
    
    def encrypt_credential(self, credential: str) -> str:
        """Encrypt a credential for secure storage."""
        return self.cipher.encrypt(credential.encode()).decode()
    
    def decrypt_credential(self, encrypted_credential: str) -> str:
        """Decrypt a stored credential."""
        return self.cipher.decrypt(encrypted_credential.encode()).decode()
    
    def get_secure_token(self) -> str:
        """Get API token with fallback hierarchy."""
        
        # Try environment variable first
        token = os.getenv('GMGN_ACCESS_TOKEN')
        if token:
            return token
        
        # Try encrypted config file
        config_file = os.getenv('GMGN_CONFIG_FILE')
        if config_file and os.path.exists(config_file):
            with open(config_file, 'r') as f:
                encrypted_token = f.read().strip()
                return self.decrypt_credential(encrypted_token)
        
        raise ValueError("No valid API token found")
    
    @staticmethod
    def create_ssl_context() -> ssl.SSLContext:
        """Create secure SSL context for connections."""
        context = ssl.create_default_context()
        context.minimum_version = ssl.TLSVersion.TLSv1_3
        context.check_hostname = True
        context.verify_mode = ssl.CERT_REQUIRED
        return context