Update app.py

app.py

@@ -1,527 +1,202 @@
-"""
-Blockchain Wallet Analyzer - A tool for analyzing Ethereum wallet contents and NFT holdings.
-
-This module provides a complete implementation of a blockchain wallet analysis tool
-with a Gradio web interface. It includes wallet analysis, NFT tracking, and 
-interactive chat capabilities using the Groq API.
+import requests
+from decimal import Decimal
+from datetime import datetime, timedelta
+from typing import List, Dict, Any
+import logging
 
-Author: Mane
-Date: March 2025
-"""
+logger = logging.getLogger(__name__)
 
-from __future__ import annotations
+# Fraud Detection Constants
+BLACKLISTED_ADDRESSES = ["0x0000000000000000000000000000000000000000", "0xdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef"]  # Example blacklisted addresses
+FRAUDULENT_THRESHOLD = 10  # Flag transactions where the value is higher than this threshold in ETH
+TRANSACTION_FREQUENCY_LIMIT = 50  # Maximum number of transactions per day to be considered "normal"
+TIME_WINDOW = timedelta(hours=24)  # Time window for detecting high-frequency transactions (24 hours)
 
-import os
-import re
-import json
-import time
-import logging
-import asyncio
-from typing import List, Dict, Tuple, Any, Optional, TypeVar, cast
-from datetime import datetime
-from decimal import Decimal
-from dataclasses import dataclass
-from enum import Enum
-from pathlib import Path
+class ValidationError(Exception):
+    pass
 
-import aiohttp
-import gradio as gr
-from tenacity import retry, stop_after_attempt, wait_exponential
-import groq
+class FraudDetection:
+    """Handles fraud detection by analyzing wallet transactions."""
+    
+    @staticmethod
+    def detect_high_frequency_transactions(transactions: List[Dict[str, Any]]) -> bool:
+        """Detect high-frequency transactions within a short time period (e.g., 24 hours)."""
+        recent_transactions = []
+        current_time = datetime.now()
+        for tx in transactions:
+            timestamp = datetime.utcfromtimestamp(int(tx['timeStamp']))
+            if current_time - timestamp < TIME_WINDOW:
+                recent_transactions.append(tx)
+
+        if len(recent_transactions) > TRANSACTION_FREQUENCY_LIMIT:
+            return True  # Flag as high-frequency fraud
+        return False
 
-# Type variables
-T = TypeVar('T')
-WalletData = Dict[str, Any]
-ChatHistory = List[Tuple[str, str]]
+    @staticmethod
+    def detect_large_token_transfers(transactions: List[Dict[str, Any]]) -> List[str]:
+        """Detect large transfers (fraudulent tokens above a threshold)."""
+        flagged_transactions = []
+        for tx in transactions:
+            amount = Decimal(tx["value"]) / Decimal(10 ** int(tx["tokenDecimal"]))
+            if amount > FRAUDULENT_THRESHOLD:
+                flagged_transactions.append(tx)
+        return flagged_transactions
 
-# API Keys - Define as constants in the script
-GROQ_API_KEY = "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"
-ETHERSCAN_API_KEY = "V562U3P76KKEUJ2DV9Q3U2EPDK89T5CKNH"
+    @staticmethod
+    def detect_suspicious_addresses(transactions: List[Dict[str, Any]]) -> List[str]:
+        """Detect transactions involving blacklisted or suspicious addresses."""
+        flagged_addresses = []
+        for tx in transactions:
+            if tx["to"].lower() in BLACKLISTED_ADDRESSES or tx["from"].lower() in BLACKLISTED_ADDRESSES:
+                flagged_addresses.append(tx)
+        return flagged_addresses
 
-# Configure logging
-logging.basicConfig(
-    level=logging.INFO,
-    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
-    handlers=[
-        logging.FileHandler('blockchain_analyzer.log'),
-        logging.StreamHandler()
-    ]
-)
-logger = logging.getLogger(__name__)
+    @staticmethod
+    def detect_unusual_nft_sales(nft_transactions: List[Dict[str, Any]]) -> List[str]:
+        """Detect unusual NFT transfers that may indicate fraudulent activity (high-value sales)."""
+        flagged_nfts = []
+        for tx in nft_transactions:
+            if int(tx['value']) > FRAUDULENT_THRESHOLD:  # Assuming `value` here is the token value
+                flagged_nfts.append(tx)
+        return flagged_nfts
 
-class ConfigError(Exception):
-    """Raised when there's an error in configuration."""
-    pass
+    @staticmethod
+    def run_fraud_detection(wallet_data: Dict[str, Any]) -> Dict[str, Any]:
+        """Run all fraud detection checks and return flagged issues."""
+        transactions = wallet_data.get("transactions", [])
+        nft_transactions = wallet_data.get("nft_collections", [])
+        
+        fraud_report = {
+            "high_frequency_transactions": False,
+            "large_token_transfers": [],
+            "suspicious_addresses": [],
+            "unusual_nft_sales": []
+        }
+        
+        # Detect high-frequency transactions
+        fraud_report["high_frequency_transactions"] = FraudDetection.detect_high_frequency_transactions(transactions)
 
-class APIError(Exception):
-    """Raised when there's an error in API calls."""
-    pass
+        # Detect large token transfers
+        fraud_report["large_token_transfers"] = FraudDetection.detect_large_token_transfers(transactions)
 
-class ValidationError(Exception):
-    """Raised when there's an error in input validation."""
-    pass
+        # Detect suspicious addresses
+        fraud_report["suspicious_addresses"] = FraudDetection.detect_suspicious_addresses(transactions)
 
-@dataclass
-class Config:
-    """Application configuration settings."""
-    SYSTEM_PROMPT: str = """
-    You are MANE 👑 (Learning & Observing Smart Systems Digital Output Generator), 
-    an adorable blockchain-sniffing puppy!
-    Your personality:
-    - Friendly and enthusiastic
-    - Explain findings in fun, simple ways
+        # Detect unusual NFT sales
+        fraud_report["unusual_nft_sales"] = FraudDetection.detect_unusual_nft_sales(nft_transactions)
 
-    Instructions:
-    - You have access to detailed wallet data in your context
-    - Use this data to provide specific answers about holdings
-    - Reference exact numbers and collections when discussing NFTs
-    - Compare wallets if multiple are available
-    """
-    ETHERSCAN_BASE_URL: str = "https://api.etherscan.io/api"
-    ETHEREUM_ADDRESS_REGEX: str = r"0x[a-fA-F0-9]{40}"
-    RATE_LIMIT_DELAY: float = 0.2  # 5 requests per second max for free tier
-    MAX_RETRIES: int = 3
-    GROQ_MODEL: str = "llama3-70b-8192"  # Groq's high-performance model
-    MAX_TOKENS: int = 4000
-    TEMPERATURE: float = 0.7
-    HISTORY_LIMIT: int = 5
+        return fraud_report
 
-    @classmethod
-    def load(cls, config_path: str | Path) -> Config:
-        """Load configuration from a JSON file."""
-        try:
-            with open(config_path) as f:
-                config_data = json.load(f)
-                return cls(**config_data)
-        except Exception as e:
-            logger.error(f"Error loading config: {e}")
-            return cls()
 
 class WalletAnalyzer:
     """Analyzes Ethereum wallet contents using Etherscan API."""
     
-    def __init__(self):
-        """Initialize the analyzer with API key."""
-        self.api_key = ETHERSCAN_API_KEY
-        self.base_url = Config.ETHERSCAN_BASE_URL
-        self.last_request_time = 0
-        self.session: Optional[aiohttp.ClientSession] = None
-    
-    async def __aenter__(self) -> WalletAnalyzer:
-        """Create aiohttp session on context manager enter."""
-        self.session = aiohttp.ClientSession()
-        return self
-    
-    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
-        """Close aiohttp session on context manager exit."""
-        if self.session:
-            await self.session.close()
-            self.session = None
-
-    @retry(
-        stop=stop_after_attempt(Config.MAX_RETRIES),
-        wait=wait_exponential(multiplier=1, min=4, max=10)
-    )
-    async def _fetch_data(self, params: Dict[str, str]) -> Dict[str, Any]:
-        """Fetch data from Etherscan API with retry logic."""
-        if not self.session:
-            raise APIError("No active session. Use context manager.")
-            
-        await self._rate_limit()
-        params["apikey"] = self.api_key
-        
-        try:
-            async with self.session.get(self.base_url, params=params) as response:
-                if response.status != 200:
-                    raise APIError(f"API request failed: {response.status}")
-                
-                data = await response.json()
-                if data["status"] == "0":
-                    error_msg = data.get('message', 'Unknown error')
-                    if "Max rate limit reached" in error_msg:
-                        raise APIError("Rate limit exceeded")
-                    raise APIError(f"API error: {error_msg}")
-                    
-                return data
-                
-        except aiohttp.ClientError as e:
-            raise APIError(f"Network error: {str(e)}")
-        except Exception as e:
-            raise APIError(f"Unexpected error: {str(e)}")
-
-    async def _rate_limit(self) -> None:
-        """Implement rate limiting for Etherscan API."""
-        current_time = time.time()
-        time_passed = current_time - self.last_request_time
-        if time_passed < Config.RATE_LIMIT_DELAY:
-            await asyncio.sleep(Config.RATE_LIMIT_DELAY - time_passed)
-        self.last_request_time = time.time()
+    API_KEY = "Your_Etherscan_API_Key"
+    ETHERSCAN_URL = "https://api.etherscan.io/api"
 
     @staticmethod
     def _validate_address(address: str) -> bool:
-        """Validate Ethereum address format."""
-        return bool(re.match(Config.ETHEREUM_ADDRESS_REGEX, address))
-
-    async def get_portfolio_data(self, address: str) -> WalletData:
-        """Get complete portfolio including ETH, tokens, and NFTs."""
-        if not self._validate_address(address):
-            raise ValidationError(f"Invalid Ethereum address: {address}")
-
-        logger.info(f"Fetching portfolio data for {address}")
-        
-        # Get ETH balance
-        eth_balance = await self._get_eth_balance(address)
-        
-        # Get token data
-        token_holdings = await self._get_token_holdings(address)
-        
-        # Get NFT data
-        nft_collections = await self._get_nft_holdings(address)
-        
-        return {
-            "address": address,
-            "last_updated": datetime.now().isoformat(),
-            "eth_balance": float(eth_balance),
-            "tokens": token_holdings,
-            "nft_collections": nft_collections
-        }
-
-    async def _get_eth_balance(self, address: str) -> Decimal:
-        """Get ETH balance for address."""
+        """Validate Ethereum address."""
+        if len(address) != 42 or not address.startswith("0x"):
+            return False
+        return True
+
+    async def _fetch_data(self, params: dict) -> dict:
+        """Fetch data from Etherscan API."""
+        params["apikey"] = self.API_KEY
+        response = requests.get(self.ETHERSCAN_URL, params=params)
+        return response.json()
+
+    async def _get_eth_balance(self, address: str) -> str:
+        """Fetch ETH balance of the given address."""
         params = {
             "module": "account",
             "action": "balance",
             "address": address,
             "tag": "latest"
         }
-        data = await self._fetch_data(params)
-        return Decimal(data["result"]) / Decimal("1000000000000000000")
+        return await self._fetch_data(params)["result"]
 
     async def _get_token_holdings(self, address: str) -> List[Dict[str, Any]]:
-        """Get token holdings for address."""
+        """Fetch token holdings of the given address."""
         params = {
             "module": "account",
             "action": "tokentx",
             "address": address,
+            "startblock": 0,
+            "endblock": 99999999,
+            "page": 1,
+            "offset": 10,
             "sort": "desc"
         }
-        data = await self._fetch_data(params)
-        
-        token_holdings: Dict[str, Dict[str, Any]] = {}
-        for tx in data.get("result", []):
-            contract = tx["contractAddress"]
-            if contract not in token_holdings:
-                token_holdings[contract] = {
-                    "name": tx["tokenName"],
-                    "symbol": tx["tokenSymbol"],
-                    "decimals": int(tx["tokenDecimal"]),
-                    "balance": Decimal(0)
-                }
-            
-            amount = Decimal(tx["value"]) / Decimal(10 ** int(tx["tokenDecimal"]))
-            if tx["to"].lower() == address.lower():
-                token_holdings[contract]["balance"] += amount
-            elif tx["from"].lower() == address.lower():
-                token_holdings[contract]["balance"] -= amount
-
-        return [
-            {
-                "name": data["name"],
-                "symbol": data["symbol"],
-                "balance": float(data["balance"])
-            }
-            for data in token_holdings.values()
-            if data["balance"] > 0
-        ]
+        return await self._fetch_data(params)["result"]
 
-    async def _get_nft_holdings(self, address: str) -> Dict[str, Dict[str, Any]]:
-        """Get NFT holdings for address."""
+    async def _get_nft_holdings(self, address: str) -> List[Dict[str, Any]]:
+        """Fetch NFT holdings of the given address."""
         params = {
             "module": "account",
             "action": "tokennfttx",
             "address": address,
+            "startblock": 0,
+            "endblock": 99999999,
+            "page": 1,
+            "offset": 10,
             "sort": "desc"
         }
-        data = await self._fetch_data(params)
+        return await self._fetch_data(params)["result"]
+
+    async def get_portfolio_data(self, address: str) -> Dict[str, Any]:
+        """Get complete portfolio including ETH, tokens, and NFTs.""" 
+        if not self._validate_address(address):
+            raise ValidationError(f"Invalid Ethereum address: {address}")
+
+        logger.info(f"Fetching portfolio data for {address}")
+        
+        # Get ETH balance
+        eth_balance = await self._get_eth_balance(address)
         
-        nft_holdings: Dict[str, Dict[str, Any]] = {}
-        collections: Dict[str, List[str]] = {}
+        # Get token data
+        token_holdings = await self._get_token_holdings(address)
         
-        for tx in data.get("result", []):
-            collection_name = tx.get("tokenName", "Unknown Collection")
-            token_id = tx["tokenID"]
-            key = f"{tx['contractAddress']}_{token_id}"
-            
-            if tx["to"].lower() == address.lower():
-                nft_holdings[key] = {
-                    "collection": collection_name,
-                    "token_id": token_id,
-                    "contract": tx["contractAddress"],
-                    "acquired_time": tx["timeStamp"]
-                }
-                
-                if collection_name not in collections:
-                    collections[collection_name] = []
-                collections[collection_name].append(token_id)
-                
-            elif tx["from"].lower() == address.lower():
-                nft_holdings.pop(key, None)
-                if collection_name in collections and token_id in collections[collection_name]:
-                    collections[collection_name].remove(token_id)
+        # Get NFT data
+        nft_collections = await self._get_nft_holdings(address)
 
-        return {
-            name: {
-                "count": len(tokens),
-                "token_ids": tokens
-            }
-            for name, tokens in collections.items()
-            if tokens  # Only include collections with tokens
+        # Get transactions data for fraud detection
+        params = {
+            "module": "account",
+            "action": "txlist",
+            "address": address,
+            "startblock": 0,
+            "endblock": 99999999,
+            "page": 1,
+            "offset": 10,
+            "sort": "desc",
         }
-
-
-class ChatInterface:
-    """Handles chat interaction using Groq API."""
-    
-    def __init__(self):
-        """Initialize chat interface with Groq client."""
-        self.groq_client = groq.Client(api_key=GROQ_API_KEY)
-        self.context: Dict[str, Any] = {}
-
-    def _format_context_message(self) -> str:
-        """Format wallet data as context message."""
-        if not self.context:
-            return ""
-            
-        context_msg = ["Current Wallet Data:\n"]
-        for addr, data in self.context.items():
-            context_msg.extend([
-                f"Wallet {addr[:8]}...{addr[-6:]}:",
-                f"- ETH Balance: {data['eth_balance']:.4f} ETH",
-                f"- Tokens: {len(data['tokens'])}"
-            ])
-            
-            if data['tokens']:
-                context_msg.append("  Token Holdings:")
-                for token in data['tokens']:
-                    context_msg.append(
-                        f"  * {token['name']} ({token['symbol']}): {token['balance']}"
-                    )
-            
-            if data['nft_collections']:
-                context_msg.append("  NFT Collections:")
-                for name, info in data['nft_collections'].items():
-                    context_msg.append(f"  * {name}: {info['count']} NFTs")
-                    if info['count'] <= 5:
-                        context_msg.append(
-                            f"    Token IDs: {', '.join(map(str, info['token_ids']))}"
-                        )
-                        
-        return "\n".join(context_msg)
-
-    async def process_message(
-        self,
-        message: str,
-        history: Optional[ChatHistory] = None
-    ) -> Tuple[ChatHistory, Dict[str, Any], str]:
-        """Process user message and generate response."""
-        if not message.strip():
-            return history or [], self.context, ""
-
-        history = history or []
+        transaction_data = await self._fetch_data(params)
         
-        # Check for Ethereum address
-        match = re.search(Config.ETHEREUM_ADDRESS_REGEX, message)
-        if match:
-            try:
-                address = match.group(0)
-                async with WalletAnalyzer() as analyzer:
-                    wallet_data = await analyzer.get_portfolio_data(address)
-                    self.context[address] = wallet_data
-                    
-                    summary = [
-                        f"📊 Portfolio Summary for {address[:8]}...{address[-6:]}",
-                        f"💎 ETH Balance: {wallet_data['eth_balance']:.4f} ETH",
-                        f"🪙 Tokens: {len(wallet_data['tokens'])} different tokens"
-                    ]
-                    
-                    total_nfts = sum(
-                        coll['count'] for coll in wallet_data['nft_collections'].values())
-                    
-                    summary.append(
-                        f"🎨 NFTs: {total_nfts} NFTs in {len(wallet_data['nft_collections'])} collections"
-                    )
-                    
-                    bot_message = "\n".join(summary)
-                    history.append((message, bot_message))
-                    return history, self.context, ""
-                    
-            except Exception as e:
-                logger.error(f"Error analyzing wallet: {e}")
-                error_message = f"Error analyzing wallet: {str(e)}"
-                history.append((message, error_message))
-                return history, self.context, ""
-
-        # Generate response using Groq
-        try:
-            # Format context message
-            context_msg = self._format_context_message()
-            
-            # Convert history to Groq format
-            chat_history = []
-            for user_msg, assistant_msg in history[-Config.HISTORY_LIMIT:]:
-                chat_history.extend([
-                    {"role": "user", "content": user_msg},
-                    {"role": "assistant", "content": assistant_msg}
-                ])
-
-            # Generate response using Groq
-            response = self.groq_client.chat.completions.create(
-                model=Config.GROQ_MODEL,
-                messages=[
-                    {"role": "system", "content": Config.SYSTEM_PROMPT},
-                    {"role": "system", "content": context_msg},
-                    *chat_history,
-                    {"role": "user", "content": message}
-                ],
-                temperature=Config.TEMPERATURE,
-                max_tokens=Config.MAX_TOKENS
-            )
-            
-            bot_message = response.choices[0].message.content
-            history.append((message, bot_message))
-            return history, self.context, ""
-            
-        except Exception as e:
-            logger.error(f"Error generating response: {e}")
-            error_message = f"Error generating response: {str(e)}"
-            history.append((message, error_message))
-            return history, self.context, ""
-
-    def clear_context(self) -> Tuple[Dict[str, Any], List[Tuple[str, str]]]:
-        """Clear the wallet context and chat history."""
-        self.context = {}
-        return {}, []
-
-
-class GradioInterface:
-    """Handles Gradio web interface setup and interactions."""
-    
-    def __init__(self):
-        """Initialize Gradio interface."""
-        self.chat_interface = ChatInterface()
-        self.demo = self._create_interface()
-
-    def _create_interface(self) -> gr.Blocks:
-        """Create and configure Gradio interface."""
-        with gr.Blocks(theme=gr.themes.Soft()) as demo:
-            gr.Markdown("""
-            # MANE👑: Blockchain Wallet Analyzer
-            
-            Welcome to 👑 MANE WAR - Your friendly blockchain analysis companion!
-            - Input an Ethereum wallet address to analyze
-            - Chat about your wallet contents with context!
-            """)
-
-            # Main Interface
-            with gr.Row():
-                # Chat Area (Left Side)
-                with gr.Column(scale=2):
-                    chatbot = gr.Chatbot(
-                        label="Chat History",
-                        height=500,
-                        value=[]
-                    )
-                    with gr.Row():
-                        msg_input = gr.Textbox(
-                            label="Message",
-                            placeholder="Enter wallet address or ask about holdings...",
-                            show_label=True
-                        )
-                        send_btn = gr.Button("Send", variant="primary")
-
-                # Context Sidebar (Right Side)
-                with gr.Column(scale=1):
-                    wallet_context = gr.JSON(
-                        label="Active Wallet Context",
-                        show_label=True,
-                        value={}
-                    )
-                    clear_btn = gr.Button("Clear Context", variant="secondary")
-
-            async def handle_message(
-                message: str,
-                chat_history: List[Tuple[str, str]],
-                context: Dict[str, Any]
-            ) -> Tuple[List[Tuple[str, str]], Dict[str, Any]]:
-                """Handle incoming messages."""
-                try:
-                    history, new_context, _ = await self.chat_interface.process_message(
-                        message,
-                        chat_history
-                    )
-                    return history, new_context
-                    
-                except Exception as e:
-                    logger.error(f"Error handling message: {e}")
-                    if chat_history is None:
-                        chat_history = []
-                    chat_history.append((message, f"Error: {str(e)}"))
-                    return chat_history, context
-
-            # Connect Event Handlers
-            clear_btn.click(
-                fn=self.chat_interface.clear_context,
-                inputs=[],
-                outputs=[wallet_context, chatbot]
-            )
-
-            # Message Handling
-            msg_input.submit(
-                fn=handle_message,
-                inputs=[msg_input, chatbot, wallet_context],
-                outputs=[chatbot, wallet_context]
-            ).then(
-                lambda: gr.update(value=""),
-                None,
-                [msg_input]
-            )
-
-            send_btn.click(
-                fn=handle_message,
-                inputs=[msg_input, chatbot, wallet_context],
-                outputs=[chatbot, wallet_context]
-            ).then(
-                lambda: gr.update(value=""),
-                None,
-                [msg_input]
-            )
-
-        return demo
+        # Prepare wallet data
+        wallet_data = {
+            "address": address,
+            "last_updated": datetime.now().isoformat(),
+            "eth_balance": float(eth_balance),
+            "tokens": token_holdings,
+            "nft_collections": nft_collections,
+            "transactions": transaction_data.get("result", [])
+        }
 
-    def launch(self, **kwargs):
-        """Launch the Gradio interface."""
-        self.demo.queue()
-        self.demo.launch(**kwargs)
+        # Run fraud detection
+        fraud_report = FraudDetection.run_fraud_detection(wallet_data)
+        wallet_data["fraud_report"] = fraud_report
 
+        return wallet_data
 
-def main():
-    """Main entry point for the application."""
-    try:
-        # Load configuration
-        config = Config.load("config.json")
-        
-        # Initialize and launch interface
-        interface = GradioInterface()
-        interface.launch(
-            server_name="0.0.0.0",
-            server_port=7860,
-            share=True
-        )
-        
-    except Exception as e:
-        logger.error(f"Application startup failed: {e}")
-        raise
 
+# Example usage:
+async def main():
+    address = "0xYourEthereumAddressHere"
+    wallet_analyzer = WalletAnalyzer()
+    portfolio_data = await wallet_analyzer.get_portfolio_data(address)
+    print(portfolio_data)
 
-if __name__ == "__main__":
-    main()
+# Run the main function
+# asyncio.run(main())  # Uncomment this line if using in an async environment.