Analyzing On Chain Data

# On-Chain Analytics Configuration

# API endpoints
apis:
  defillama:
    base_url: https://api.llama.fi
    yields_url: https://yields.llama.fi
    rate_limit: null  # No rate limit

  coingecko:
    base_url: https://api.coingecko.com/api/v3
    api_key_env: COINGECKO_API_KEY
    rate_limit: 30  # requests per minute

# Caching
cache:
  enabled: true
  protocols_ttl: 300      # 5 minutes
  chains_ttl: 300         # 5 minutes
  fees_ttl: 900           # 15 minutes
  yields_ttl: 300         # 5 minutes
  history_ttl: 3600       # 1 hour
  stablecoins_ttl: 600    # 10 minutes

# Default query limits
defaults:
  protocol_limit: 50
  yield_limit: 20
  fee_limit: 30
  min_tvl: 0
  trend_threshold: 10.0   # % for trending

# Output settings
output:
  default_format: table
  show_percentages: true
  number_format: abbreviated  # 1.5B vs 1500000000

# Categories to track
categories:
  - Liquid Staking
  - Lending
  - DEX
  - CDP
  - Bridge
  - Yield
  - Derivatives
  - Insurance
  - Yield Aggregator
  - RWA

# Chains to prioritize
priority_chains:
  - Ethereum
  - Arbitrum
  - Polygon
  - Optimism
  - BSC
  - Avalanche
  - Base
  - Solana

#!/usr/bin/env python3
"""
On-Chain Data Fetcher

Fetch protocol metrics from DeFiLlama and other sources.

Author: Jeremy Longshore <jeremy@intentsolutions.io>
Version: 1.0.0
License: MIT
"""

import json
import time
from pathlib import Path
from typing import Dict, Any, List, Optional
from dataclasses import dataclass

try:
    import requests
except ImportError:
    requests = None


DEFILLAMA_BASE = "https://api.llama.fi"
COINGECKO_BASE = "https://api.coingecko.com/api/v3"


@dataclass
class ProtocolData:
    """Protocol metrics."""
    name: str
    slug: str
    tvl: float
    tvl_change_24h: float
    tvl_change_7d: float
    chains: List[str]
    category: str
    token: Optional[str] = None
    mcap: Optional[float] = None
    fdv: Optional[float] = None


class DataFetcher:
    """Fetch on-chain analytics data."""

    def __init__(self, verbose: bool = False):
        """Initialize fetcher."""
        self.verbose = verbose
        self.cache_file = Path.home() / ".onchain_analytics_cache.json"
        self.cache_ttl = 300  # 5 minutes
        self._cache = self._load_cache()

    def _load_cache(self) -> Dict[str, Any]:
        """Load cache from file."""
        try:
            if self.cache_file.exists():
                with open(self.cache_file) as f:
                    return json.load(f)
        except (json.JSONDecodeError, IOError):
            pass
        return {}

    def _save_cache(self) -> None:
        """Save cache to file."""
        try:
            with open(self.cache_file, "w") as f:
                json.dump(self._cache, f)
        except IOError:
            pass

    def _is_cache_valid(self, key: str, ttl: int = None) -> bool:
        """Check if cached data is still valid."""
        if key not in self._cache:
            return False
        cached_time = self._cache.get(f"{key}_time", 0)
        return time.time() - cached_time < (ttl or self.cache_ttl)

    def _api_get(self, url: str, params: Dict = None) -> Any:
        """Make API request."""
        if not requests:
            raise ImportError("requests library required")

        if self.verbose:
            print(f"API: {url}")

        response = requests.get(url, params=params, timeout=30)
        response.raise_for_status()
        return response.json()

    def fetch_protocols(self, limit: int = 100) -> List[ProtocolData]:
        """Fetch all protocols with TVL.

        Args:
            limit: Max protocols to return

        Returns:
            List of protocol data
        """
        cache_key = "protocols"
        if self._is_cache_valid(cache_key):
            data = self._cache[cache_key]
        else:
            data = self._api_get(f"{DEFILLAMA_BASE}/protocols")
            self._cache[cache_key] = data
            self._cache[f"{cache_key}_time"] = time.time()
            self._save_cache()

        protocols = []
        for p in data[:limit]:
            protocols.append(ProtocolData(
                name=p.get("name", "Unknown"),
                slug=p.get("slug", ""),
                tvl=p.get("tvl", 0),
                tvl_change_24h=p.get("change_1d", 0) or 0,
                tvl_change_7d=p.get("change_7d", 0) or 0,
                chains=p.get("chains", []),
                category=p.get("category", "Unknown"),
                token=p.get("symbol"),
                mcap=p.get("mcap"),
                fdv=p.get("fdv"),
            ))

        return protocols

    def fetch_chains(self) -> List[Dict[str, Any]]:
        """Fetch TVL by chain.

        Returns:
            List of chain data
        """
        cache_key = "chains"
        if self._is_cache_valid(cache_key):
            return self._cache[cache_key]

        data = self._api_get(f"{DEFILLAMA_BASE}/v2/chains")

        self._cache[cache_key] = data
        self._cache[f"{cache_key}_time"] = time.time()
        self._save_cache()

        return data

    def fetch_protocol_tvl_history(self, slug: str) -> List[Dict[str, Any]]:
        """Fetch historical TVL for protocol.

        Args:
            slug: Protocol slug

        Returns:
            List of {date, tvl} entries
        """
        cache_key = f"tvl_history_{slug}"
        if self._is_cache_valid(cache_key, ttl=3600):
            return self._cache[cache_key]

        data = self._api_get(f"{DEFILLAMA_BASE}/protocol/{slug}")
        tvl_data = data.get("tvl", [])

        self._cache[cache_key] = tvl_data
        self._cache[f"{cache_key}_time"] = time.time()
        self._save_cache()

        return tvl_data

    def fetch_fees_revenue(self, protocol: str = None) -> List[Dict[str, Any]]:
        """Fetch protocol fees and revenue.

        Args:
            protocol: Optional specific protocol

        Returns:
            List of fee/revenue data
        """
        cache_key = f"fees_{protocol or 'all'}"
        if self._is_cache_valid(cache_key):
            return self._cache[cache_key]

        if protocol:
            url = f"{DEFILLAMA_BASE}/summary/fees/{protocol}"
        else:
            url = f"{DEFILLAMA_BASE}/overview/fees"

        try:
            data = self._api_get(url)
            result = data.get("protocols", [data]) if protocol else data.get("protocols", [])
        except Exception:
            result = []

        self._cache[cache_key] = result
        self._cache[f"{cache_key}_time"] = time.time()
        self._save_cache()

        return result

    def fetch_dex_volumes(self, chain: str = None) -> List[Dict[str, Any]]:
        """Fetch DEX trading volumes.

        Args:
            chain: Optional chain filter

        Returns:
            List of DEX volume data
        """
        cache_key = f"dex_volumes_{chain or 'all'}"
        if self._is_cache_valid(cache_key):
            return self._cache[cache_key]

        url = f"{DEFILLAMA_BASE}/overview/dexs"
        if chain:
            url += f"/{chain}"

        data = self._api_get(url)
        result = data.get("protocols", [])

        self._cache[cache_key] = result
        self._cache[f"{cache_key}_time"] = time.time()
        self._save_cache()

        return result

    def fetch_yields(self, chain: str = None) -> List[Dict[str, Any]]:
        """Fetch yield/APY data.

        Args:
            chain: Optional chain filter

        Returns:
            List of yield pools
        """
        cache_key = f"yields_{chain or 'all'}"
        if self._is_cache_valid(cache_key):
            return self._cache[cache_key]

        url = "https://yields.llama.fi/pools"
        data = self._api_get(url)
        pools = data.get("data", [])

        if chain:
            pools = [p for p in pools if p.get("chain", "").lower() == chain.lower()]

        self._cache[cache_key] = pools
        self._cache[f"{cache_key}_time"] = time.time()
        self._save_cache()

        return pools

    def fetch_stablecoin_data(self) -> List[Dict[str, Any]]:
        """Fetch stablecoin metrics.

        Returns:
            List of stablecoin data
        """
        cache_key = "stablecoins"
        if self._is_cache_valid(cache_key):
            return self._cache[cache_key]

        data = self._api_get(f"{DEFILLAMA_BASE}/stablecoins")
        result = data.get("peggedAssets", [])

        self._cache[cache_key] = result
        self._cache[f"{cache_key}_time"] = time.time()
        self._save_cache()

        return result


def main():
    """CLI entry point for testing."""
    fetcher = DataFetcher(verbose=True)

    print("=== Top 10 Protocols by TVL ===")
    protocols = fetcher.fetch_protocols(limit=10)

    for p in protocols:
        print(f"{p.name}: ${p.tvl / 1e9:.2f}B ({p.tvl_change_24h:+.1f}% 24h)")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Output Formatters

Format on-chain analytics data for various outputs.

Author: Jeremy Longshore <jeremy@intentsolutions.io>
Version: 1.0.0
License: MIT
"""

import json
import csv
import io
from typing import Dict, Any, List


def format_usd(value: float) -> str:
    """Format USD value."""
    if value >= 1e12:
        return f"${value / 1e12:.2f}T"
    elif value >= 1e9:
        return f"${value / 1e9:.2f}B"
    elif value >= 1e6:
        return f"${value / 1e6:.2f}M"
    elif value >= 1e3:
        return f"${value / 1e3:.2f}K"
    else:
        return f"${value:.2f}"


def format_percent(value: float) -> str:
    """Format percentage with sign."""
    if value > 0:
        return f"+{value:.1f}%"
    else:
        return f"{value:.1f}%"


def format_protocols_table(protocols: List[Dict[str, Any]], title: str = "Protocol Rankings") -> str:
    """Format protocols as ASCII table.

    Args:
        protocols: List of protocol data
        title: Table title

    Returns:
        Formatted table string
    """
    lines = [
        f"\n{title}",
        "=" * 95,
        f"{'Rank':<6} {'Protocol':<20} {'TVL':<14} {'24h':<10} {'7d':<10} {'Share':<10} {'Category':<15}",
        "-" * 95,
    ]

    for p in protocols[:50]:
        rank = p.get("rank", "-")
        name = p.get("name", "Unknown")[:18]
        tvl = format_usd(p.get("tvl", 0))
        change_24h = format_percent(p.get("change_1d", 0) or p.get("tvl_change_24h", 0) or 0)
        change_7d = format_percent(p.get("change_7d", 0) or p.get("tvl_change_7d", 0) or 0)
        share = f"{p.get('market_share', 0):.1f}%"
        category = p.get("category", "Unknown")[:13]

        lines.append(f"{rank:<6} {name:<20} {tvl:<14} {change_24h:<10} {change_7d:<10} {share:<10} {category:<15}")

    lines.append("-" * 95)
    lines.append(f"Total: {len(protocols)} protocols")

    return "\n".join(lines)


def format_chains_table(chains: List[Dict[str, Any]]) -> str:
    """Format chains as ASCII table.

    Args:
        chains: List of chain data

    Returns:
        Formatted table string
    """
    lines = [
        "\nChain TVL Rankings",
        "=" * 70,
        f"{'Rank':<6} {'Chain':<20} {'TVL':<14} {'Dominance':<12} {'Protocols':<10}",
        "-" * 70,
    ]

    for i, c in enumerate(chains[:30], 1):
        name = c.get("name", "Unknown")[:18]
        tvl = format_usd(c.get("tvl", 0))
        dominance = f"{c.get('dominance', 0):.2f}%"
        protocols = c.get("protocols", "-")

        lines.append(f"{i:<6} {name:<20} {tvl:<14} {dominance:<12} {protocols:<10}")

    lines.append("-" * 70)

    return "\n".join(lines)


def format_fees_table(fees_data: List[Dict[str, Any]]) -> str:
    """Format fees/revenue as ASCII table.

    Args:
        fees_data: List of fee data

    Returns:
        Formatted table string
    """
    lines = [
        "\nProtocol Fees & Revenue",
        "=" * 85,
        f"{'Protocol':<20} {'Fees 24h':<14} {'Fees 30d':<14} {'Rev 24h':<14} {'Rev 30d':<14}",
        "-" * 85,
    ]

    for p in fees_data[:30]:
        name = p.get("name", p.get("displayName", "Unknown"))[:18]
        fees_24h = format_usd(p.get("total24h", 0) or 0)
        fees_30d = format_usd(p.get("total30d", 0) or 0)
        rev_24h = format_usd(p.get("revenue24h", 0) or 0)
        rev_30d = format_usd(p.get("revenue30d", 0) or 0)

        lines.append(f"{name:<20} {fees_24h:<14} {fees_30d:<14} {rev_24h:<14} {rev_30d:<14}")

    lines.append("-" * 85)

    return "\n".join(lines)


def format_trends_table(trends: Dict[str, List[Dict]]) -> str:
    """Format trending protocols.

    Args:
        trends: Dict with trending_up and trending_down lists

    Returns:
        Formatted table string
    """
    lines = ["\nTrending Protocols (7d)", "=" * 60]

    lines.append("\nTrending Up:")
    lines.append("-" * 40)
    for p in trends.get("trending_up", [])[:5]:
        lines.append(f"  {p['name']}: {format_percent(p['growth'])} ({format_usd(p['tvl'])})")

    lines.append("\nTrending Down:")
    lines.append("-" * 40)
    for p in trends.get("trending_down", [])[:5]:
        lines.append(f"  {p['name']}: {format_percent(p['growth'])} ({format_usd(p['tvl'])})")

    return "\n".join(lines)


def format_category_summary(categories: Dict[str, Dict[str, Any]]) -> str:
    """Format category summary.

    Args:
        categories: Dict of category metrics

    Returns:
        Formatted table string
    """
    sorted_cats = sorted(categories.values(), key=lambda x: x["total_tvl"], reverse=True)

    lines = [
        "\nCategory Summary",
        "=" * 70,
        f"{'Category':<25} {'TVL':<14} {'Share':<10} {'Protocols':<10}",
        "-" * 70,
    ]

    for c in sorted_cats[:20]:
        name = c.get("name", "Unknown")[:23]
        tvl = format_usd(c.get("total_tvl", 0))
        share = f"{c.get('market_share', 0):.1f}%"
        count = c.get("protocol_count", 0)

        lines.append(f"{name:<25} {tvl:<14} {share:<10} {count:<10}")

    lines.append("-" * 70)

    return "\n".join(lines)


def format_json(data: Any) -> str:
    """Format data as JSON."""
    return json.dumps(data, indent=2, default=str)


def format_csv(data: List[Dict[str, Any]], fields: List[str] = None) -> str:
    """Format data as CSV."""
    if not data:
        return ""

    if not fields:
        fields = list(data[0].keys())

    output = io.StringIO()
    writer = csv.DictWriter(output, fieldnames=fields, extrasaction='ignore')
    writer.writeheader()
    writer.writerows(data)

    return output.getvalue()


def main():
    """CLI entry point for testing."""
    # Test with sample data
    protocols = [
        {"rank": 1, "name": "Lido", "tvl": 15e9, "change_1d": 2.5, "change_7d": 5.0, "market_share": 30.0, "category": "Liquid Staking"},
        {"rank": 2, "name": "Aave", "tvl": 8e9, "change_1d": -1.2, "change_7d": 3.0, "market_share": 16.0, "category": "Lending"},
    ]

    print(format_protocols_table(protocols))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Metrics Calculator

Calculate derived on-chain metrics and analytics.

Author: Jeremy Longshore <jeremy@intentsolutions.io>
Version: 1.0.0
License: MIT
"""

from typing import Dict, Any, List, Optional
from dataclasses import dataclass


@dataclass
class ProtocolMetrics:
    """Calculated protocol metrics."""
    name: str
    tvl: float
    market_share: float
    tvl_to_mcap: Optional[float]
    revenue_24h: Optional[float]
    revenue_30d: Optional[float]
    fees_24h: Optional[float]
    pe_ratio: Optional[float]
    growth_7d: float
    growth_30d: float
    rank: int


class MetricsCalculator:
    """Calculate on-chain analytics metrics."""

    def calculate_market_share(
        self,
        protocols: List[Dict[str, Any]],
        category: str = None
    ) -> List[Dict[str, Any]]:
        """Calculate market share for protocols.

        Args:
            protocols: List of protocol data
            category: Optional category filter

        Returns:
            Protocols with market_share added
        """
        if category:
            protocols = [p for p in protocols if p.get("category", "").lower() == category.lower()]

        total_tvl = sum(p.get("tvl", 0) for p in protocols)

        for p in protocols:
            tvl = p.get("tvl", 0)
            p["market_share"] = (tvl / total_tvl * 100) if total_tvl > 0 else 0

        return protocols

    def calculate_tvl_to_mcap(
        self,
        protocols: List[Dict[str, Any]]
    ) -> List[Dict[str, Any]]:
        """Calculate TVL/Market Cap ratio.

        Args:
            protocols: List of protocol data

        Returns:
            Protocols with tvl_to_mcap ratio
        """
        for p in protocols:
            tvl = p.get("tvl", 0)
            mcap = p.get("mcap")

            if mcap and mcap > 0:
                p["tvl_to_mcap"] = tvl / mcap
            else:
                p["tvl_to_mcap"] = None

        return protocols

    def calculate_pe_ratio(
        self,
        protocols: List[Dict[str, Any]]
    ) -> List[Dict[str, Any]]:
        """Calculate P/E ratios for protocols.

        Args:
            protocols: List of protocol data with fees

        Returns:
            Protocols with pe_ratio
        """
        for p in protocols:
            mcap = p.get("mcap")
            annual_fees = p.get("total30d", 0) * 12  # Annualized from 30d

            if mcap and annual_fees > 0:
                p["pe_ratio"] = mcap / annual_fees
            else:
                p["pe_ratio"] = None

        return protocols

    def calculate_growth_rates(
        self,
        tvl_history: List[Dict[str, Any]]
    ) -> Dict[str, float]:
        """Calculate growth rates from TVL history.

        Args:
            tvl_history: List of {date, tvl} entries

        Returns:
            Growth rates for various periods
        """
        if not tvl_history or len(tvl_history) < 2:
            return {"growth_24h": 0, "growth_7d": 0, "growth_30d": 0}

        current = tvl_history[-1].get("totalLiquidityUSD", 0)

        # 24h ago (index -2 if daily data)
        if len(tvl_history) >= 2:
            prev_24h = tvl_history[-2].get("totalLiquidityUSD", current)
            growth_24h = ((current - prev_24h) / prev_24h * 100) if prev_24h > 0 else 0
        else:
            growth_24h = 0

        # 7d ago
        if len(tvl_history) >= 8:
            prev_7d = tvl_history[-8].get("totalLiquidityUSD", current)
            growth_7d = ((current - prev_7d) / prev_7d * 100) if prev_7d > 0 else 0
        else:
            growth_7d = 0

        # 30d ago
        if len(tvl_history) >= 31:
            prev_30d = tvl_history[-31].get("totalLiquidityUSD", current)
            growth_30d = ((current - prev_30d) / prev_30d * 100) if prev_30d > 0 else 0
        else:
            growth_30d = 0

        return {
            "growth_24h": growth_24h,
            "growth_7d": growth_7d,
            "growth_30d": growth_30d,
        }

    def rank_protocols(
        self,
        protocols: List[Dict[str, Any]],
        metric: str = "tvl"
    ) -> List[Dict[str, Any]]:
        """Rank protocols by metric.

        Args:
            protocols: List of protocol data
            metric: Metric to rank by

        Returns:
            Ranked protocols
        """
        # Sort by metric (descending for most metrics)
        reverse = metric not in ["pe_ratio"]  # Lower PE is better
        protocols.sort(key=lambda x: x.get(metric, 0) or 0, reverse=reverse)

        # Add rank
        for i, p in enumerate(protocols):
            p["rank"] = i + 1

        return protocols

    def calculate_chain_dominance(
        self,
        chains: List[Dict[str, Any]]
    ) -> List[Dict[str, Any]]:
        """Calculate chain TVL dominance.

        Args:
            chains: List of chain TVL data

        Returns:
            Chains with dominance percentage
        """
        total_tvl = sum(c.get("tvl", 0) for c in chains)

        for c in chains:
            tvl = c.get("tvl", 0)
            c["dominance"] = (tvl / total_tvl * 100) if total_tvl > 0 else 0

        chains.sort(key=lambda x: x.get("tvl", 0), reverse=True)

        return chains

    def calculate_category_metrics(
        self,
        protocols: List[Dict[str, Any]]
    ) -> Dict[str, Dict[str, Any]]:
        """Calculate metrics grouped by category.

        Args:
            protocols: List of protocol data

        Returns:
            Category-level metrics
        """
        categories: Dict[str, Dict[str, Any]] = {}

        for p in protocols:
            cat = p.get("category", "Other")
            if cat not in categories:
                categories[cat] = {
                    "name": cat,
                    "protocol_count": 0,
                    "total_tvl": 0,
                    "protocols": []
                }

            categories[cat]["protocol_count"] += 1
            categories[cat]["total_tvl"] += p.get("tvl", 0)
            categories[cat]["protocols"].append(p.get("name"))

        # Calculate market share per category
        total_tvl = sum(c["total_tvl"] for c in categories.values())
        for cat in categories.values():
            cat["market_share"] = (cat["total_tvl"] / total_tvl * 100) if total_tvl > 0 else 0

        return categories

    def identify_trends(
        self,
        protocols: List[Dict[str, Any]],
        min_growth: float = 10.0
    ) -> Dict[str, List[Dict[str, Any]]]:
        """Identify trending protocols.

        Args:
            protocols: List of protocol data
            min_growth: Minimum growth % to be considered trending

        Returns:
            Trending up and down protocols
        """
        trending_up = []
        trending_down = []

        for p in protocols:
            growth = p.get("change_7d", 0) or 0

            if growth >= min_growth:
                trending_up.append({"name": p.get("name"), "growth": growth, "tvl": p.get("tvl", 0)})
            elif growth <= -min_growth:
                trending_down.append({"name": p.get("name"), "growth": growth, "tvl": p.get("tvl", 0)})

        trending_up.sort(key=lambda x: x["growth"], reverse=True)
        trending_down.sort(key=lambda x: x["growth"])

        return {
            "trending_up": trending_up[:10],
            "trending_down": trending_down[:10]
        }


def main():
    """CLI entry point for testing."""
    calc = MetricsCalculator()

    # Test with sample data
    protocols = [
        {"name": "Lido", "tvl": 15000000000, "category": "Liquid Staking", "mcap": 2000000000},
        {"name": "Aave", "tvl": 8000000000, "category": "Lending", "mcap": 1500000000},
        {"name": "Uniswap", "tvl": 5000000000, "category": "DEX", "mcap": 4000000000},
    ]

    protocols = calc.calculate_market_share(protocols)
    protocols = calc.calculate_tvl_to_mcap(protocols)
    protocols = calc.rank_protocols(protocols, "tvl")

    for p in protocols:
        print(f"#{p['rank']} {p['name']}: ${p['tvl']/1e9:.2f}B ({p['market_share']:.1f}%)")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
On-Chain Analytics CLI

Analyze DeFi protocol metrics and blockchain data.

Author: Jeremy Longshore <jeremy@intentsolutions.io>
Version: 1.0.0
License: MIT
"""

import argparse
import sys
from pathlib import Path

# Add scripts directory to path
sys.path.insert(0, str(Path(__file__).parent))

from data_fetcher import DataFetcher
from metrics_calculator import MetricsCalculator
from formatters import (
    format_protocols_table,
    format_chains_table,
    format_fees_table,
    format_trends_table,
    format_category_summary,
    format_json,
    format_csv,
)


def cmd_protocols(args) -> int:
    """List protocols by TVL.

    Args:
        args: CLI arguments

    Returns:
        Exit code
    """
    fetcher = DataFetcher(verbose=args.verbose)
    calc = MetricsCalculator()

    protocols = fetcher.fetch_protocols(limit=args.limit)

    # Convert to dict format
    data = [vars(p) for p in protocols]

    # Filter by category
    if args.category:
        data = [p for p in data if p.get("category", "").lower() == args.category.lower()]

    # Filter by chain
    if args.chain:
        data = [p for p in data if args.chain.lower() in [c.lower() for c in p.get("chains", [])]]

    # Calculate metrics
    data = calc.calculate_market_share(data)
    data = calc.calculate_tvl_to_mcap(data)
    data = calc.rank_protocols(data, args.sort)

    # Output
    if args.format == "json":
        print(format_json(data))
    elif args.format == "csv":
        print(format_csv(data, ["rank", "name", "tvl", "market_share", "category"]))
    else:
        print(format_protocols_table(data))

    return 0


def cmd_chains(args) -> int:
    """List chains by TVL.

    Args:
        args: CLI arguments

    Returns:
        Exit code
    """
    fetcher = DataFetcher(verbose=args.verbose)
    calc = MetricsCalculator()

    chains = fetcher.fetch_chains()
    chains = calc.calculate_chain_dominance(chains)

    if args.format == "json":
        print(format_json(chains))
    elif args.format == "csv":
        print(format_csv(chains, ["name", "tvl", "dominance"]))
    else:
        print(format_chains_table(chains))

    return 0


def cmd_fees(args) -> int:
    """Show protocol fees and revenue.

    Args:
        args: CLI arguments

    Returns:
        Exit code
    """
    fetcher = DataFetcher(verbose=args.verbose)

    fees = fetcher.fetch_fees_revenue(args.protocol)

    if not fees:
        print("No fee data available.")
        return 0

    # Sort by 30d fees
    fees.sort(key=lambda x: x.get("total30d", 0) or 0, reverse=True)

    if args.format == "json":
        print(format_json(fees[:args.limit]))
    elif args.format == "csv":
        print(format_csv(fees[:args.limit], ["name", "total24h", "total30d", "revenue24h", "revenue30d"]))
    else:
        print(format_fees_table(fees[:args.limit]))

    return 0


def cmd_dex(args) -> int:
    """Show DEX volumes.

    Args:
        args: CLI arguments

    Returns:
        Exit code
    """
    fetcher = DataFetcher(verbose=args.verbose)

    volumes = fetcher.fetch_dex_volumes(args.chain)

    if not volumes:
        print("No DEX volume data available.")
        return 0

    # Sort by volume
    volumes.sort(key=lambda x: x.get("total24h", 0) or 0, reverse=True)

    if args.format == "json":
        print(format_json(volumes[:args.limit]))
    else:
        from formatters import format_usd
        print("\nDEX 24h Volumes")
        print("=" * 60)
        for i, d in enumerate(volumes[:args.limit], 1):
            name = d.get("displayName", d.get("name", "Unknown"))
            vol = d.get("total24h", 0) or 0
            print(f"{i}. {name}: {format_usd(vol)}")

    return 0


def cmd_categories(args) -> int:
    """Show category breakdown.

    Args:
        args: CLI arguments

    Returns:
        Exit code
    """
    fetcher = DataFetcher(verbose=args.verbose)
    calc = MetricsCalculator()

    protocols = fetcher.fetch_protocols(limit=500)
    data = [vars(p) for p in protocols]

    categories = calc.calculate_category_metrics(data)

    if args.format == "json":
        print(format_json(categories))
    else:
        print(format_category_summary(categories))

    return 0


def cmd_trends(args) -> int:
    """Show trending protocols.

    Args:
        args: CLI arguments

    Returns:
        Exit code
    """
    fetcher = DataFetcher(verbose=args.verbose)
    calc = MetricsCalculator()

    protocols = fetcher.fetch_protocols(limit=200)
    data = [vars(p) for p in protocols]

    trends = calc.identify_trends(data, min_growth=args.threshold)

    if args.format == "json":
        print(format_json(trends))
    else:
        print(format_trends_table(trends))

    return 0


def cmd_yields(args) -> int:
    """Show top yields.

    Args:
        args: CLI arguments

    Returns:
        Exit code
    """
    fetcher = DataFetcher(verbose=args.verbose)

    pools = fetcher.fetch_yields(args.chain)

    # Sort by APY
    pools.sort(key=lambda x: x.get("apy", 0) or 0, reverse=True)

    # Filter by minimum TVL
    if args.min_tvl:
        pools = [p for p in pools if (p.get("tvlUsd", 0) or 0) >= args.min_tvl]

    if args.format == "json":
        print(format_json(pools[:args.limit]))
    else:
        from formatters import format_usd
        print(f"\nTop Yields{' on ' + args.chain if args.chain else ''}")
        print("=" * 80)
        print(f"{'Pool':<30} {'APY':<12} {'TVL':<14} {'Chain':<12}")
        print("-" * 80)
        for p in pools[:args.limit]:
            name = f"{p.get('project', '?')}: {p.get('symbol', '?')}"[:28]
            apy = f"{p.get('apy', 0):.2f}%"
            tvl = format_usd(p.get('tvlUsd', 0) or 0)
            chain = p.get('chain', '?')[:10]
            print(f"{name:<30} {apy:<12} {tvl:<14} {chain:<12}")

    return 0


def cmd_stables(args) -> int:
    """Show stablecoin metrics.

    Args:
        args: CLI arguments

    Returns:
        Exit code
    """
    fetcher = DataFetcher(verbose=args.verbose)

    stables = fetcher.fetch_stablecoin_data()

    # Sort by market cap
    stables.sort(key=lambda x: x.get("circulating", {}).get("peggedUSD", 0) or 0, reverse=True)

    if args.format == "json":
        print(format_json(stables[:args.limit]))
    else:
        from formatters import format_usd
        print("\nStablecoin Market Caps")
        print("=" * 70)
        for i, s in enumerate(stables[:args.limit], 1):
            name = s.get("name", "Unknown")
            mcap = s.get("circulating", {}).get("peggedUSD", 0) or 0
            price = s.get("price", 1.0) or 1.0
            print(f"{i}. {name}: {format_usd(mcap)} (${price:.4f})")

    return 0


def main():
    """Main entry point."""
    parser = argparse.ArgumentParser(
        description="On-chain analytics for DeFi protocols",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s protocols                       # Top protocols by TVL
  %(prog)s protocols --category lending    # Lending protocols only
  %(prog)s chains                          # Chain TVL rankings
  %(prog)s fees                            # Protocol fees/revenue
  %(prog)s dex                             # DEX volumes
  %(prog)s trends                          # Trending protocols
  %(prog)s yields --min-tvl 1000000        # Top yields with >$1M TVL
  %(prog)s stables                         # Stablecoin market caps
        """
    )

    parser.add_argument("--format", "-f", choices=["table", "json", "csv"], default="table")
    parser.add_argument("--verbose", "-v", action="store_true")
    parser.add_argument("--version", action="version", version="%(prog)s 1.0.0")

    subparsers = parser.add_subparsers(dest="command", help="Commands")

    # Protocols command
    proto_parser = subparsers.add_parser("protocols", help="Protocol TVL rankings")
    proto_parser.add_argument("--limit", "-l", type=int, default=50)
    proto_parser.add_argument("--category", "-c", help="Filter by category")
    proto_parser.add_argument("--chain", help="Filter by chain")
    proto_parser.add_argument("--sort", default="tvl", choices=["tvl", "market_share", "tvl_to_mcap"])
    proto_parser.set_defaults(func=cmd_protocols)

    # Chains command
    chain_parser = subparsers.add_parser("chains", help="Chain TVL rankings")
    chain_parser.set_defaults(func=cmd_chains)

    # Fees command
    fees_parser = subparsers.add_parser("fees", help="Protocol fees and revenue")
    fees_parser.add_argument("--protocol", "-p", help="Specific protocol")
    fees_parser.add_argument("--limit", "-l", type=int, default=30)
    fees_parser.set_defaults(func=cmd_fees)

    # DEX command
    dex_parser = subparsers.add_parser("dex", help="DEX volumes")
    dex_parser.add_argument("--chain", "-c", help="Filter by chain")
    dex_parser.add_argument("--limit", "-l", type=int, default=20)
    dex_parser.set_defaults(func=cmd_dex)

    # Categories command
    cat_parser = subparsers.add_parser("categories", help="Category breakdown")
    cat_parser.set_defaults(func=cmd_categories)

    # Trends command
    trends_parser = subparsers.add_parser("trends", help="Trending protocols")
    trends_parser.add_argument("--threshold", "-t", type=float, default=10.0, help="Min growth %")
    trends_parser.set_defaults(func=cmd_trends)

    # Yields command
    yields_parser = subparsers.add_parser("yields", help="Top yields")
    yields_parser.add_argument("--chain", "-c", help="Filter by chain")
    yields_parser.add_argument("--min-tvl", type=float, help="Minimum TVL")
    yields_parser.add_argument("--limit", "-l", type=int, default=20)
    yields_parser.set_defaults(func=cmd_yields)

    # Stablecoins command
    stables_parser = subparsers.add_parser("stables", help="Stablecoin metrics")
    stables_parser.add_argument("--limit", "-l", type=int, default=20)
    stables_parser.set_defaults(func=cmd_stables)

    args = parser.parse_args()

    if not args.command:
        parser.print_help()
        return 1

    try:
        return args.func(args)
    except KeyboardInterrupt:
        print("\nInterrupted.")
        return 130
    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        if args.verbose:
            import traceback
            traceback.print_exc()
        return 1


if __name__ == "__main__":
    sys.exit(main())