app.py

import requests
from decimal import Decimal
from datetime import datetime, timedelta
from typing import List, Dict, Any
import logging
import asyncio
from flask import Flask, render_template_string, request

# Flask setup
app = Flask(__name__)

# Logging setup
logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)

# 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)

class ValidationError(Exception):
    pass

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

    @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

    @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

    @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

    @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)

        # Detect large token transfers
        fraud_report["large_token_transfers"] = FraudDetection.detect_large_token_transfers(transactions)

        # Detect suspicious addresses
        fraud_report["suspicious_addresses"] = FraudDetection.detect_suspicious_addresses(transactions)

        # Detect unusual NFT sales
        fraud_report["unusual_nft_sales"] = FraudDetection.detect_unusual_nft_sales(nft_transactions)

        return fraud_report


class WalletAnalyzer:
    """Analyzes Ethereum wallet contents using Etherscan API."""

    API_KEY = "Your_Etherscan_API_Key"
    ETHERSCAN_URL = "https://api.etherscan.io/api"

    @staticmethod
    def _validate_address(address: str) -> bool:
        """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"
        }
        return await self._fetch_data(params)["result"]

    async def _get_token_holdings(self, address: str) -> List[Dict[str, Any]]:
        """Fetch token holdings of the given address."""
        params = {
            "module": "account",
            "action": "tokentx",
            "address": address,
            "startblock": 0,
            "endblock": 99999999,
            "page": 1,
            "offset": 10,
            "sort": "desc"
        }
        return await self._fetch_data(params)["result"]

    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"
        }
        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)

        # Get token data
        token_holdings = await self._get_token_holdings(address)

        # Get NFT data
        nft_collections = await self._get_nft_holdings(address)

        # Get transactions data for fraud detection
        params = {
            "module": "account",
            "action": "txlist",
            "address": address,
            "startblock": 0,
            "endblock": 99999999,
            "page": 1,
            "offset": 10,
            "sort": "desc",
        }
        transaction_data = await self._fetch_data(params)

        # 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", [])
        }

        # Run fraud detection
        fraud_report = FraudDetection.run_fraud_detection(wallet_data)
        wallet_data["fraud_report"] = fraud_report

        return wallet_data


# Flask Routes to Handle Requests
@app.route('/')
def index():
    return render_template_string('''
        <!DOCTYPE html>
        <html lang="en">
        <head>
            <meta charset="UTF-8">
            <meta name="viewport" content="width=device-width, initial-scale=1.0">
            <title>Ethereum Wallet Analyzer</title>
            <link href="https://cdn.jsdelivr.net/npm/bootstrap@5.1.3/dist/css/bootstrap.min.css" rel="stylesheet">
            <style>
                .result-section {
                    margin-top: 30px;
                }
                .alert-danger {
                    background-color: #f8d7da;
                    color: #721c24;
                }
                .alert-success {
                    background-color: #d4edda;
                    color: #155724;
                }
                .loading {
                    display: none;
                }
            </style>
        </head>
        <body>
            <div class="container">
                <h2 class="mt-5">Ethereum Wallet Fraud Detection</h2>
                <form action="/analyze_wallet" method="POST" class="mt-4">
                    <div class="mb-3">
                        <label for="address" class="form-label">Enter Ethereum Address</label>
                        <input type="text" name="address" id="address" class="form-control" required>
                    </div>
                    <button type="submit" class="btn btn-primary">Analyze Wallet</button>
                </form>

                <div id="loading" class="loading text-center mt-4">
                    <div class="spinner-border text-primary" role="status"></div>
                    <p>Loading...</p>
                </div>
            </div>

            <script>
                const form = document.querySelector('form');
                const loading = document.getElementById('loading');

                form.addEventListener('submit', function() {
                    loading.style.display = 'block';
                });
            </script>
        </body>
        </html>
    ''')

@app.route('/analyze_wallet', methods=['POST'])
def analyze_wallet():
    address = request.form['address']
    
    wallet_analyzer = WalletAnalyzer()
    loop = asyncio.new_event_loop()  # Create a new event loop to run async code
    asyncio.set_event_loop(loop)
    
    # Get wallet data
    portfolio_data = loop.run_until_complete(wallet_analyzer.get_portfolio_data(address))
    
    return render_template_string('''
        <!DOCTYPE html>
        <html lang="en">
        <head>
            <meta charset="UTF-8">
            <meta name="viewport" content="width=device-width, initial-scale=1.0">
            <title>Ethereum Wallet Analyzer - Results</title>
            <link href="https://cdn.jsdelivr.net/npm/bootstrap@5.1.3/dist/css/bootstrap.min.css" rel="stylesheet">
            <style>
                .result-section {
                    margin-top: 30px;
                }
                .alert-danger {
                    background-color: #f8d7da;
                    color: #721c24;
                }
                .alert-success {
                    background-color: #d4edda;
                    color: #155724;
                }
            </style>
        </head>
        <body>
            <div class="container">
                <h2 class="mt-5">Ethereum Wallet Fraud Detection Results</h2>
                <p><strong>Address:</strong> {{ portfolio_data['address'] }}</p>
                <p><strong>ETH Balance:</strong> {{ portfolio_data['eth_balance'] }} ETH</p>

                <div class="result-section">
                    <h4>Fraud Detection Results:</h4>

                    <!-- High Frequency Transactions -->
                    {% if portfolio_data['fraud_report']['high_frequency_transactions'] %}
                        <div class="alert alert-danger">
                            High Frequency Transactions Detected!
                        </div>
                    {% else %}
                        <div class="alert alert-success">
                            No High Frequency Transactions Detected.
                        </div>
                    {% endif %}

                    <!-- Large Token Transfers -->
                    {% if portfolio_data['fraud_report']['large_token_transfers'] %}
                        <div class="alert alert-danger">
                            Large Token Transfers Detected!
                        </div>
                    {% else %}
                        <div class="alert alert-success">
                            No Large Token Transfers Detected.
                        </div>
                    {% endif %}

                    <!-- Suspicious Addresses -->
                    {% if portfolio_data['fraud_report']['suspicious_addresses'] %}
                        <div class="alert alert-danger">
                            Suspicious Addresses Detected!
                        </div>
                    {% else %}
                        <div class="alert alert-success">
                            No Suspicious Addresses Detected.
                        </div>
                    {% endif %}

                    <!-- Unusual NFT Sales -->
                    {% if portfolio_data['fraud_report']['unusual_nft_sales'] %}
                        <div class="alert alert-danger">
                            Unusual NFT Sales Detected!
                        </div>
                    {% else %}
                        <div class="alert alert-success">
                            No Unusual NFT Sales Detected.
                        </div>
                    {% endif %}
                </div>

                <a href="/" class="btn btn-primary mt-3">Analyze Another Wallet</a>
            </div>
        </body>
        </html>
    ''', portfolio_data=portfolio_data)

if __name__ == "__main__":
    app.run(debug=True)