Birdeye

"""
Birdeye Extension — Token Intelligence and Wallet Analytics

Provides 3 working tools for token analysis and wallet tracking across Solana and EVM chains.

Token Intelligence Tools (2):
- birdeye_token_security: Security score and risk analysis
- birdeye_token_overview: Comprehensive token data

Wallet Analytics Tools (1):
- birdeye_wallet_networth: Current net worth snapshot

Environment Variables Required:
- BIRDEYE_API_KEY: Your Birdeye API key

Note: Wallet APIs have limited rate (5 req/s, 75 req/min).

Usage:
    This extension is auto-loaded by the ExtensionLoader.
"""

import logging
from typing import List

logger = logging.getLogger(__name__)


def register(api) -> List[str]:
    """
    Extension entry point — register all Birdeye tools.

    Args:
        api: ExtensionApi instance with registry and config

    Returns:
        List of registered tool names
    """
    registered = []

    # Token Intelligence Tools (2)
    try:
        from ._legacy_token import (
            BirdeyeTokenSecurityTool,
            BirdeyeTokenOverviewTool
        )

        api.register_tool(BirdeyeTokenSecurityTool())
        api.register_tool(BirdeyeTokenOverviewTool())

        registered.extend([
            "birdeye_token_security",
            "birdeye_token_overview"
        ])

        logger.info("Registered Birdeye token intelligence tools (2 tools)")
    except Exception as e:
        logger.warning(f"Failed to load Birdeye token intelligence tools: {e}")

    # Wallet Analytics Tools (1)
    try:
        from ._legacy_wallet import (
            BirdeyeWalletNetworthTool
        )

        api.register_tool(BirdeyeWalletNetworthTool())

        registered.append("birdeye_wallet_networth")

        logger.info("Registered Birdeye wallet analytics tools (1 tool)")
    except Exception as e:
        logger.warning(f"Failed to load Birdeye wallet analytics tools: {e}")

    logger.info(f"Registered Birdeye extension ({len(registered)} tools total)")
    return registered


# Extension metadata
EXTENSION_INFO = {
    "name": "birdeye",
    "version": "2.0.0",
    "description": "Birdeye token intelligence and wallet analytics for Solana and EVM chains",
    "tools": [
        # Token Intelligence (2)
        "birdeye_token_security",
        "birdeye_token_overview",
        # Wallet Analytics (1)
        "birdeye_wallet_networth"
    ],
    "env_vars": [
        "BIRDEYE_API_KEY",
    ],
}

"""
Birdeye Smart Money Tool Wrappers

BaseTool wrappers for smart money tracking and trader analysis.

NOTE: All smart money tools have been removed due to API errors.
"""
import logging

logger = logging.getLogger(__name__)

# All smart money tools removed - they were failing with API errors
BIRDEYE_SMART_MONEY_AVAILABLE = False
logger.warning("Birdeye smart money tools have been disabled due to API issues")

"""
Birdeye Token Intelligence Tool Wrappers

BaseTool wrappers for token security and overview analysis.
"""
import asyncio
import logging

from core.tool import BaseTool, ToolContext, ToolResult

logger = logging.getLogger(__name__)

try:
    from .tools.token import (
        get_token_security,
        get_token_overview
    )
    BIRDEYE_TOKEN_AVAILABLE = True
except ImportError as e:
    logger.warning(f"Birdeye token tools not available: {e}")
    BIRDEYE_TOKEN_AVAILABLE = False


class BirdeyeTokenSecurityTool(BaseTool):
    """Get token security score and analysis."""

    @property
    def name(self) -> str:
        return "birdeye_token_security"

    @property
    def description(self) -> str:
        return """Get token security score and risk analysis.

Identifies rug pull risks, contract issues, and liquidity concerns.

Parameters:
- address: Token contract address (required)
- chain: Blockchain (default: solana)

Returns: Security score, risk level, and list of detected issues"""

    @property
    def parameters(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "address": {"type": "string", "description": "Token address"},
                "chain": {"type": "string", "description": "Blockchain", "default": "solana"}
            },
            "required": ["address"]
        }

    async def execute(self, ctx: ToolContext, address: str, chain: str = "solana", **kwargs) -> ToolResult:
        if not BIRDEYE_TOKEN_AVAILABLE:
            return ToolResult(success=False, output=None, error="Birdeye token tools not available")
        try:
            result = await asyncio.to_thread(get_token_security, address=address, chain=chain)
            if result is None:
                return ToolResult(success=False, output=None, error="Failed to fetch token security.")
            return ToolResult(success=True, output=result)
        except Exception as e:
            return ToolResult(success=False, output=None, error=str(e))


class BirdeyeTokenOverviewTool(BaseTool):
    """Get comprehensive token data."""

    @property
    def name(self) -> str:
        return "birdeye_token_overview"

    @property
    def description(self) -> str:
        return """Get comprehensive token overview.

Includes price, volume, market cap, liquidity, and price changes.

Parameters:
- address: Token contract address (required)
- chain: Blockchain (default: solana)

Returns: Symbol, price, volume, market cap, liquidity, and price change data"""

    @property
    def parameters(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "address": {"type": "string", "description": "Token address"},
                "chain": {"type": "string", "description": "Blockchain", "default": "solana"}
            },
            "required": ["address"]
        }

    async def execute(self, ctx: ToolContext, address: str, chain: str = "solana", **kwargs) -> ToolResult:
        if not BIRDEYE_TOKEN_AVAILABLE:
            return ToolResult(success=False, output=None, error="Birdeye token tools not available")
        try:
            result = await asyncio.to_thread(get_token_overview, address=address, chain=chain)
            if result is None:
                return ToolResult(success=False, output=None, error="Failed to fetch token overview.")
            return ToolResult(success=True, output=result)
        except Exception as e:
            return ToolResult(success=False, output=None, error=str(e))

"""
Birdeye Wallet Analytics Tool Wrappers

BaseTool wrappers for wallet net worth analysis.

Note: Wallet APIs have limited rate (5 req/s, 75 req/min).
"""
import asyncio
import logging

from core.tool import BaseTool, ToolContext, ToolResult

logger = logging.getLogger(__name__)

try:
    from .tools.wallet import (
        get_wallet_networth,
    )
    BIRDEYE_WALLET_AVAILABLE = True
except ImportError as e:
    logger.warning(f"Birdeye wallet tools not available: {e}")
    BIRDEYE_WALLET_AVAILABLE = False


class BirdeyeWalletNetworthTool(BaseTool):
    """Get current wallet net worth and portfolio breakdown."""

    @property
    def name(self) -> str:
        return "birdeye_wallet_networth"

    @property
    def description(self) -> str:
        return """Get current net worth and portfolio snapshot for a wallet.

Returns total value, token breakdown, and asset allocation.

Parameters:
- wallet: Wallet address (required)
- chain: Blockchain (default: solana)

Returns: Total USD value and detailed token holdings with balances and values"""

    @property
    def parameters(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "wallet": {"type": "string", "description": "Wallet address"},
                "chain": {"type": "string", "description": "Blockchain", "default": "solana"}
            },
            "required": ["wallet"]
        }

    async def execute(self, ctx: ToolContext, wallet: str, chain: str = "solana", **kwargs) -> ToolResult:
        if not BIRDEYE_WALLET_AVAILABLE:
            return ToolResult(success=False, output=None, error="Birdeye wallet tools not available")

        try:
            result = await asyncio.to_thread(get_wallet_networth, wallet=wallet, chain=chain)
            if result is None:
                return ToolResult(success=False, output=None, error="Failed to fetch wallet net worth. Check BIRDEYE_API_KEY.")
            return ToolResult(success=True, output=result)
        except Exception as e:
            return ToolResult(success=False, output=None, error=str(e))

"""
Birdeye skill exports — tool names match SKILL.md frontmatter.

Usage in task scripts:
    from core.skill_tools import birdeye
    data = birdeye.birdeye_token_overview(address="So11...", chain="solana")
"""
import importlib.util
import os

_base = os.path.join(os.path.dirname(__file__), "tools")


def _load_func(subpath, func_name):
    full_path = os.path.join(_base, subpath)
    spec = importlib.util.spec_from_file_location(f"_birdeye_{func_name}", full_path)
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return getattr(mod, func_name)


birdeye_token_security = _load_func("token/security.py", "get_token_security")
birdeye_token_overview = _load_func("token/overview.py", "get_token_overview")
birdeye_wallet_networth = _load_func("wallet/networth.py", "get_wallet_networth")

"""
Birdeye Smart Money Tools

Track smart money flows, top traders, and high-performance wallet activity.
"""

from .smart_money_tokens import get_smart_money_tokens
from .top_traders import get_top_traders
from .trader_analysis import get_trader_gainers_losers, get_trader_trades

__all__ = [
    "get_smart_money_tokens",
    "get_top_traders",
    "get_trader_gainers_losers",
    "get_trader_trades",
]

"""
Birdeye Token Intelligence Tools

Token security analysis and comprehensive overview data.
"""

from .security import get_token_security
from .overview import get_token_overview

__all__ = [
    "get_token_security",
    "get_token_overview",
]

#!/usr/bin/env python3
"""
Birdeye Token Overview Module

Get comprehensive token data including price, volume, market cap, liquidity.
"""

import os
import sys
import json
import argparse
from typing import Dict, Any, Optional

try:
    from dotenv import load_dotenv
    project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../../..'))
    load_dotenv(os.path.join(project_root, '.env'))
except ImportError:
    pass

import requests
from core.http_client import proxied_get

BASE_URL = "https://public-api.birdeye.so"
HEADER_KEY = "X-API-KEY"
CHAINS = ["solana", "ethereum", "arbitrum", "avalanche", "bsc", "optimism", "polygon", "base", "zksync", "sui"]


def _get_api_key() -> Optional[str]:
    return os.getenv("BIRDEYE_API_KEY")


def get_token_overview(address: str, chain: str = "solana") -> Optional[Dict[str, Any]]:
    """Get comprehensive token overview with price, volume, market data."""
    api_key = _get_api_key()
    if not api_key:
        print("Error: BIRDEYE_API_KEY environment variable is required", file=sys.stderr)
        return None

    if chain not in CHAINS:
        print(f"Error: Unsupported chain '{chain}'", file=sys.stderr)
        return None

    url = f"{BASE_URL}/defi/token_overview"
    headers = {"accept": "application/json", HEADER_KEY: api_key, "x-chain": chain}
    params = {"address": address}

    try:
        response = proxied_get(url, headers=headers, params=params, timeout=30)
        response.raise_for_status()
        data = response.json()
        if not data.get("success"):
            print(f"API Error: {data.get('message', 'Unknown error')}", file=sys.stderr)
            return None
        return data
    except requests.exceptions.RequestException as e:
        print(f"Request failed: {e}", file=sys.stderr)
        return None
    except json.JSONDecodeError as e:
        print(f"Failed to parse response: {e}", file=sys.stderr)
        return None


def main():
    parser = argparse.ArgumentParser(description="Birdeye token overview")
    parser.add_argument("--address", "-a", required=True, help="Token address")
    parser.add_argument("--chain", "-c", default="solana", choices=CHAINS)
    parser.add_argument("--json", "-j", action="store_true")
    args = parser.parse_args()

    result = get_token_overview(args.address, args.chain)
    if result and result.get("success"):
        if args.json:
            print(json.dumps(result, indent=2))
        else:
            data = result.get("data", {})
            print(f"\nToken Overview: {data.get('symbol', 'N/A')}")
            print("=" * 60)
            print(f"Price:        ${data.get('price', 0):.6f}")
            print(f"24h Change:   {data.get('price_change_24h', 0):.2f}%")
            print(f"Volume 24h:   ${data.get('volume_24h', 0):,.2f}")
            print(f"Market Cap:   ${data.get('market_cap', 0):,.2f}")
            print(f"Liquidity:    ${data.get('liquidity', 0):,.2f}")
    else:
        print("Failed to fetch token overview")
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Birdeye Token Security Module

Get token security score and analysis. Identifies rug pull risks, contract issues,
and liquidity concerns.

Usage Example:
    from tools.token.security import get_token_security

    # Check token security on Solana
    security = get_token_security(address="token_address", chain="solana")
"""

import os
import sys
import json
import argparse
from typing import Dict, Any, Optional

try:
    from dotenv import load_dotenv
    project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../../..'))
    load_dotenv(os.path.join(project_root, '.env'))
except ImportError:
    pass

import requests
from core.http_client import proxied_get

BASE_URL = "https://public-api.birdeye.so"
HEADER_KEY = "X-API-KEY"
CHAINS = ["solana", "ethereum", "arbitrum", "avalanche", "bsc", "optimism", "polygon", "base", "zksync", "sui"]


def _get_api_key() -> Optional[str]:
    return os.getenv("BIRDEYE_API_KEY")


def get_token_security(address: str, chain: str = "solana") -> Optional[Dict[str, Any]]:
    """Get token security score and analysis."""
    api_key = _get_api_key()
    if not api_key:
        print("Error: BIRDEYE_API_KEY environment variable is required", file=sys.stderr)
        return None

    if chain not in CHAINS:
        print(f"Error: Unsupported chain '{chain}'", file=sys.stderr)
        return None

    url = f"{BASE_URL}/defi/token_security"
    headers = {"accept": "application/json", HEADER_KEY: api_key, "x-chain": chain}
    params = {"address": address}

    try:
        response = proxied_get(url, headers=headers, params=params, timeout=30)
        response.raise_for_status()
        data = response.json()
        if not data.get("success"):
            print(f"API Error: {data.get('message', 'Unknown error')}", file=sys.stderr)
            return None
        return data
    except requests.exceptions.RequestException as e:
        print(f"Request failed: {e}", file=sys.stderr)
        return None
    except json.JSONDecodeError as e:
        print(f"Failed to parse response: {e}", file=sys.stderr)
        return None


def main():
    parser = argparse.ArgumentParser(description="Birdeye token security check")
    parser.add_argument("--address", "-a", required=True, help="Token address")
    parser.add_argument("--chain", "-c", default="solana", choices=CHAINS)
    parser.add_argument("--json", "-j", action="store_true")
    args = parser.parse_args()

    result = get_token_security(args.address, args.chain)
    if result and result.get("success"):
        if args.json:
            print(json.dumps(result, indent=2))
        else:
            data = result.get("data", {})
            print(f"\nSecurity Analysis for {args.address[:8]}... on {args.chain.upper()}")
            print("=" * 60)
            print(f"Score: {data.get('security_score', 'N/A')}/100")
            print(f"Risk Level: {data.get('risk_level', 'N/A')}")
            if data.get("issues"):
                print("\nIssues Found:")
                for issue in data["issues"]:
                    print(f"  - {issue}")
    else:
        print("Failed to fetch token security")
        sys.exit(1)


if __name__ == "__main__":
    main()

"""
Birdeye Wallet Analytics Tools

Track wallet net worth and portfolio breakdown.
Wallet APIs have limited rate (5 req/s, 75 req/min).
"""

from .networth import get_wallet_networth, get_wallet_networth_chart

__all__ = [
    "get_wallet_networth",
    "get_wallet_networth_chart",
]

#!/usr/bin/env python3
"""
Birdeye Wallet Net Worth Module

Track wallet net worth (current snapshot and historical chart).
Provides portfolio value tracking across chains.

Note: Wallet APIs have limited rate (5 req/s, 75 req/min).

Dependencies:
- requests: For HTTP API calls
- python-dotenv: For environment variable management

Environment Variables Required:
- BIRDEYE_API_KEY: Your Birdeye API key

Usage Example:
    from tools.wallet.networth import get_wallet_networth, get_wallet_networth_chart

    # Get current net worth
    networth = get_wallet_networth(wallet="wallet_address", chain="solana")

    # Get net worth chart over time
    chart = get_wallet_networth_chart(wallet="wallet_address", chain="solana", interval="1d")

CLI Usage:
    python networth.py current --wallet <address> --chain solana
    python networth.py chart --wallet <address> --chain solana --interval 1d
"""

import os
import sys
import json
import argparse
from typing import Dict, Any, Optional

try:
    from dotenv import load_dotenv
    project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../../..'))
    load_dotenv(os.path.join(project_root, '.env'))
except ImportError:
    pass

import requests
from core.http_client import proxied_get

# Birdeye Configuration
BASE_URL = "https://public-api.birdeye.so"
HEADER_KEY = "X-API-KEY"

# Supported chains
CHAINS = ["solana", "ethereum", "arbitrum", "avalanche", "bsc", "optimism", "polygon", "base", "zksync", "sui"]


def _get_api_key() -> Optional[str]:
    """Get Birdeye API key from environment."""
    return os.getenv("BIRDEYE_API_KEY")


def get_wallet_networth(
    wallet: str,
    chain: str = "solana"
) -> Optional[Dict[str, Any]]:
    """
    Get current net worth and portfolio snapshot for a wallet.

    Returns total portfolio value, token breakdown, and asset allocation.

    Args:
        wallet: Wallet address
        chain: Blockchain (solana, ethereum, arbitrum, etc.)

    Returns:
        Dictionary with net worth data:
        {
            "success": true,
            "data": {
                "wallet": "wallet_address",
                "total_usd": 12345.67,
                "items": [
                    {
                        "address": "token_address",
                        "symbol": "TOKEN",
                        "name": "Token Name",
                        "balance": 1000.0,
                        "price_usd": 1.23,
                        "value_usd": 1230.0,
                        "percentage": 10.0
                    },
                    ...
                ]
            }
        }
        Returns None if request fails.

    Example:
        networth = get_wallet_networth("wallet_address", "solana")
        print(f"Total: ${networth['data']['total_usd']:,.2f}")
        for token in networth["data"]["items"][:5]:
            print(f"{token['symbol']}: ${token['value_usd']:,.2f} ({token['percentage']:.1f}%)")
    """
    api_key = _get_api_key()
    if not api_key:
        print("Error: BIRDEYE_API_KEY environment variable is required", file=sys.stderr)
        return None

    if chain not in CHAINS:
        print(f"Error: Unsupported chain '{chain}'. Supported: {', '.join(CHAINS)}", file=sys.stderr)
        return None

    url = f"{BASE_URL}/wallet/v2/net-worth"

    headers = {
        "accept": "application/json",
        HEADER_KEY: api_key,
        "x-chain": chain
    }

    params = {"wallet": wallet}

    try:
        response = proxied_get(url, headers=headers, params=params, timeout=30)
        response.raise_for_status()
        data = response.json()

        if not data.get("success"):
            print(f"API Error: {data.get('message', 'Unknown error')}", file=sys.stderr)
            return None

        return data

    except requests.exceptions.RequestException as e:
        print(f"Request failed: {e}", file=sys.stderr)
        return None
    except json.JSONDecodeError as e:
        print(f"Failed to parse response: {e}", file=sys.stderr)
        return None


def get_wallet_networth_chart(
    wallet: str,
    chain: str = "solana",
    interval: str = "1d",
    time_from: Optional[int] = None,
    time_to: Optional[int] = None
) -> Optional[Dict[str, Any]]:
    """
    Get net worth chart data over time.

    Provides historical net worth timeseries for portfolio tracking.

    Args:
        wallet: Wallet address
        chain: Blockchain (solana, ethereum, arbitrum, etc.)
        interval: Time interval (1h, 4h, 1d, 1w)
        time_from: Start timestamp (Unix seconds) (optional)
        time_to: End timestamp (Unix seconds) (optional)

    Returns:
        Dictionary with chart data:
        {
            "success": true,
            "data": {
                "wallet": "wallet_address",
                "items": [
                    {
                        "timestamp": 1234567890,
                        "value_usd": 12345.67
                    },
                    ...
                ]
            }
        }
        Returns None if request fails.

    Example:
        chart = get_wallet_networth_chart("wallet_address", "solana", interval="1d")
        for point in chart["data"]["items"]:
            print(f"{point['timestamp']}: ${point['value_usd']:,.2f}")
    """
    api_key = _get_api_key()
    if not api_key:
        print("Error: BIRDEYE_API_KEY environment variable is required", file=sys.stderr)
        return None

    if chain not in CHAINS:
        print(f"Error: Unsupported chain '{chain}'. Supported: {', '.join(CHAINS)}", file=sys.stderr)
        return None

    url = f"{BASE_URL}/wallet/v2/net-worth"

    headers = {
        "accept": "application/json",
        HEADER_KEY: api_key,
        "x-chain": chain
    }

    params = {
        "wallet": wallet,
        "time_type": interval
    }

    if time_from:
        params["time_from"] = time_from
    if time_to:
        params["time_to"] = time_to

    try:
        response = proxied_get(url, headers=headers, params=params, timeout=30)
        response.raise_for_status()
        data = response.json()

        if not data.get("success"):
            print(f"API Error: {data.get('message', 'Unknown error')}", file=sys.stderr)
            return None

        return data

    except requests.exceptions.RequestException as e:
        print(f"Request failed: {e}", file=sys.stderr)
        return None
    except json.JSONDecodeError as e:
        print(f"Failed to parse response: {e}", file=sys.stderr)
        return None


def main():
    """CLI entry point."""
    parser = argparse.ArgumentParser(description="Birdeye wallet net worth tools")
    subparsers = parser.add_subparsers(dest="command", help="Command to execute")

    # Current net worth command
    current_parser = subparsers.add_parser("current", help="Get current net worth")
    current_parser.add_argument("--wallet", "-w", required=True, help="Wallet address")
    current_parser.add_argument("--chain", "-c", default="solana", choices=CHAINS, help="Blockchain")
    current_parser.add_argument("--json", "-j", action="store_true", help="Output as JSON")

    # Chart command
    chart_parser = subparsers.add_parser("chart", help="Get net worth chart")
    chart_parser.add_argument("--wallet", "-w", required=True, help="Wallet address")
    chart_parser.add_argument("--chain", "-c", default="solana", choices=CHAINS, help="Blockchain")
    chart_parser.add_argument("--interval", "-i", default="1d", choices=["1h", "4h", "1d", "1w"], help="Time interval")
    chart_parser.add_argument("--json", "-j", action="store_true", help="Output as JSON")

    args = parser.parse_args()

    if args.command == "current":
        result = get_wallet_networth(wallet=args.wallet, chain=args.chain)

        if result and result.get("success"):
            if args.json:
                print(json.dumps(result, indent=2))
            else:
                data = result.get("data", {})
                print(f"\nNet Worth for {args.wallet[:8]}... on {args.chain.upper()}")
                print("=" * 80)
                print(f"Total: ${data.get('total_usd', 0):,.2f}")
                print("\nTop Holdings:")
                print(f"{'Symbol':<12} {'Balance':>15} {'Price':>12} {'Value USD':>15} {'%':>8}")
                print("-" * 80)
                for token in data.get("items", [])[:10]:
                    print(
                        f"{token.get('symbol', 'N/A'):<12} "
                        f"{token.get('balance', 0):>15,.4f} "
                        f"${token.get('price_usd', 0):>11.6f} "
                        f"${token.get('value_usd', 0):>14,.2f} "
                        f"{token.get('percentage', 0):>7.1f}%"
                    )
        else:
            print("Failed to fetch wallet net worth")
            sys.exit(1)

    elif args.command == "chart":
        result = get_wallet_networth_chart(
            wallet=args.wallet,
            chain=args.chain,
            interval=args.interval
        )

        if result and result.get("success"):
            if args.json:
                print(json.dumps(result, indent=2))
            else:
                items = result.get("data", {}).get("items", [])
                print(f"\nNet Worth Chart for {args.wallet[:8]}... on {args.chain.upper()}")
                print(f"Interval: {args.interval}")
                print("=" * 60)
                print(f"{'Timestamp':<20} {'Value USD':>20}")
                print("-" * 60)
                for point in items[-20:]:  # Last 20 data points
                    from datetime import datetime
                    ts = datetime.fromtimestamp(point.get('timestamp', 0)).strftime('%Y-%m-%d %H:%M')
                    print(f"{ts:<20} ${point.get('value_usd', 0):>19,.2f}")
        else:
            print("Failed to fetch wallet net worth chart")
            sys.exit(1)

    else:
        parser.print_help()


if __name__ == "__main__":
    main()