Defillama Api

#!/usr/bin/env python3
"""Solana DeFi overview using DeFiLlama (free, no auth).

Fetches TVL, DEX volumes, fees, and top protocols for Solana.
Produces a comprehensive DeFi macro snapshot.

Usage:
    python scripts/defi_snapshot.py
    CHAIN="Ethereum" python scripts/defi_snapshot.py

Dependencies:
    uv pip install httpx
"""

import os
import sys
import time
from typing import Optional

import httpx

# ── Configuration ───────────────────────────────────────────────────

CHAIN = os.getenv("CHAIN", "Solana")
BASE = "https://api.llama.fi"
COINS = "https://coins.llama.fi"

# ── API Helper ──────────────────────────────────────────────────────


def llama_get(url: str, params: Optional[dict] = None) -> dict | list:
    """GET request to DeFiLlama with retry.

    Args:
        url: Full URL.
        params: Query parameters.

    Returns:
        Parsed JSON response.
    """
    for attempt in range(3):
        try:
            resp = httpx.get(url, params=params or {}, timeout=30.0)
            if resp.status_code == 429:
                time.sleep(30.0)
                continue
            if resp.status_code >= 500:
                time.sleep(5.0)
                continue
            resp.raise_for_status()
            return resp.json()
        except httpx.TimeoutException:
            if attempt < 2:
                time.sleep(5.0)
                continue
            raise
    return {}


# ── Data Fetching ───────────────────────────────────────────────────


def get_chain_tvl(chain: str) -> float:
    """Get current TVL for a chain.

    Args:
        chain: Chain name (e.g., 'Solana').

    Returns:
        TVL in USD.
    """
    chains = llama_get(f"{BASE}/v2/chains")
    if not isinstance(chains, list):
        return 0
    for c in chains:
        if c.get("name", "").lower() == chain.lower():
            return c.get("tvl", 0)
    return 0


def get_top_protocols(chain: str, limit: int = 15) -> list[dict]:
    """Get top protocols by TVL for a chain.

    Args:
        chain: Chain name.
        limit: Number of protocols to return.

    Returns:
        Sorted list of protocol dicts.
    """
    protocols = llama_get(f"{BASE}/protocols")
    if not isinstance(protocols, list):
        return []

    chain_protocols = []
    for p in protocols:
        chains = p.get("chains", [])
        if chain in chains:
            chain_tvl = p.get("chainTvls", {}).get(chain, 0)
            if chain_tvl > 0:
                chain_protocols.append({
                    "name": p.get("name", "?"),
                    "category": p.get("category", "?"),
                    "tvl": chain_tvl,
                    "change_1d": p.get("change_1d", 0),
                    "change_7d": p.get("change_7d", 0),
                })

    chain_protocols.sort(key=lambda x: x["tvl"], reverse=True)
    return chain_protocols[:limit]


def get_dex_volumes(chain: str) -> dict:
    """Get DEX volume overview for a chain.

    Args:
        chain: Chain name.

    Returns:
        Volume summary dict.
    """
    data = llama_get(f"{BASE}/overview/dexs/{chain}", {
        "excludeTotalDataChart": "true",
        "excludeTotalDataChartBreakdown": "true",
    })
    if not isinstance(data, dict):
        return {}
    return data


def get_fees(chain: str) -> dict:
    """Get fee overview for a chain.

    Args:
        chain: Chain name.

    Returns:
        Fee summary dict.
    """
    data = llama_get(f"{BASE}/overview/fees/{chain}", {
        "excludeTotalDataChart": "true",
        "excludeTotalDataChartBreakdown": "true",
    })
    if not isinstance(data, dict):
        return {}
    return data


def get_stablecoin_supply(chain: str) -> dict:
    """Get stablecoin supply for a chain.

    Args:
        chain: Chain name.

    Returns:
        Stablecoin summary dict.
    """
    stables = llama_get("https://stablecoins.llama.fi/stablecoins")
    if not isinstance(stables, list):
        return {}

    total = 0
    breakdown = {}
    for s in stables:
        chain_data = s.get("chainCirculating", {}).get(chain, {})
        current = chain_data.get("current", {}).get("peggedUSD", 0)
        if current > 0:
            total += current
            breakdown[s["symbol"]] = current

    return {"total": total, "breakdown": breakdown}


# ── Display ─────────────────────────────────────────────────────────


def format_usd(value: float) -> str:
    """Format USD value for display."""
    if not value:
        return "$0"
    if abs(value) >= 1e9:
        return f"${value / 1e9:.2f}B"
    elif abs(value) >= 1e6:
        return f"${value / 1e6:.2f}M"
    elif abs(value) >= 1e3:
        return f"${value / 1e3:.1f}K"
    return f"${value:.2f}"


def print_report(
    chain: str,
    tvl: float,
    protocols: list[dict],
    volumes: dict,
    fees: dict,
    stables: dict,
) -> None:
    """Print DeFi snapshot report."""
    print(f"\n{'='*60}")
    print(f"DeFi SNAPSHOT — {chain}")
    print(f"{'='*60}")

    # Overview
    print(f"\n--- Overview ---")
    print(f"  Total TVL:        {format_usd(tvl)}")
    vol_24h = volumes.get("total24h", 0)
    print(f"  DEX Volume 24h:   {format_usd(vol_24h)}")
    fees_24h = fees.get("total24h", 0)
    print(f"  Fees 24h:         {format_usd(fees_24h)}")
    print(f"  Stablecoin Supply:{format_usd(stables.get('total', 0))}")

    # Top protocols
    if protocols:
        print(f"\n--- Top Protocols by TVL ---")
        print(f"  {'Protocol':<25} {'Category':<15} {'TVL':>12} {'1d%':>7} {'7d%':>7}")
        print(f"  {'─'*25} {'─'*15} {'─'*12} {'─'*7} {'─'*7}")
        for p in protocols:
            d1 = p.get("change_1d", 0) or 0
            d7 = p.get("change_7d", 0) or 0
            print(f"  {p['name']:<25} {p['category']:<15} "
                  f"{format_usd(p['tvl']):>12} {d1:>+6.1f}% {d7:>+6.1f}%")

    # Top DEXes by volume
    dex_protocols = volumes.get("protocols", [])
    if dex_protocols:
        top_dexes = sorted(dex_protocols, key=lambda x: x.get("total24h", 0) or 0, reverse=True)[:10]
        print(f"\n--- Top DEXes by Volume ---")
        print(f"  {'DEX':<20} {'Vol 24h':>14} {'Vol 7d':>14}")
        print(f"  {'─'*20} {'─'*14} {'─'*14}")
        for d in top_dexes:
            v24 = d.get("total24h", 0) or 0
            v7d = d.get("total7d", 0) or 0
            if v24 > 0:
                print(f"  {d.get('name', '?'):<20} {format_usd(v24):>14} {format_usd(v7d):>14}")

    # Stablecoin breakdown
    breakdown = stables.get("breakdown", {})
    if breakdown:
        print(f"\n--- Stablecoin Supply ---")
        sorted_stables = sorted(breakdown.items(), key=lambda x: x[1], reverse=True)
        for symbol, supply in sorted_stables[:5]:
            pct = supply / stables["total"] * 100 if stables["total"] > 0 else 0
            print(f"  {symbol:<8} {format_usd(supply):>14} ({pct:.1f}%)")

    print()


# ── Main ────────────────────────────────────────────────────────────


def main() -> None:
    """Run DeFi snapshot."""
    print(f"Generating DeFi snapshot for {CHAIN}...")

    print("Fetching TVL...")
    tvl = get_chain_tvl(CHAIN)
    time.sleep(0.5)

    print("Fetching top protocols...")
    protocols = get_top_protocols(CHAIN)
    time.sleep(0.5)

    print("Fetching DEX volumes...")
    volumes = get_dex_volumes(CHAIN)
    time.sleep(0.5)

    print("Fetching fees...")
    fees = get_fees(CHAIN)
    time.sleep(0.5)

    print("Fetching stablecoin data...")
    stables = get_stablecoin_supply(CHAIN)

    print_report(CHAIN, tvl, protocols, volumes, fees, stables)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Multi-token price lookup with historical comparison via DeFiLlama.

Fetches current prices for multiple tokens and compares against
historical prices. Free, no authentication required.

Usage:
    python scripts/price_lookup.py
    TOKENS="SOL,JUP,BONK" python scripts/price_lookup.py

Dependencies:
    uv pip install httpx
"""

import os
import sys
import time
from datetime import datetime, timezone, timedelta
from typing import Optional

import httpx

# ── Configuration ───────────────────────────────────────────────────

COINS_BASE = "https://coins.llama.fi"

# Well-known Solana tokens
TOKEN_MAP = {
    "SOL":  "solana:So11111111111111111111111111111111111111112",
    "USDC": "solana:EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v",
    "USDT": "solana:Es9vMFrzaCERmJfrF4H2FYD4KCoNkY11McCe8BenwNYB",
    "JUP":  "solana:JUPyiwrYJFskUPiHa7hkeR8VUtAeFoSYbKedZNsDvCN",
    "BONK": "solana:DezXAZ8z7PnrnRJjz3wXBoRgixCa6xjnB7YaB1pPB263",
    "WIF":  "solana:EKpQGSJtjMFqKZ9KQanSqYXRcF8fBopzLHYxdM65zcjm",
    "RAY":  "solana:4k3Dyjzvzp8eMZWUXbBCjEvwSkkk59S5iCNLY3QrkX6R",
    "ORCA": "solana:orcaEKTdK7LKz57vaAYr9QeNsVEPfiu6QeMU1kektZE",
    "BTC":  "coingecko:bitcoin",
    "ETH":  "coingecko:ethereum",
}

TOKENS_INPUT = os.getenv("TOKENS", "SOL,JUP,BONK,WIF,RAY")
SELECTED_TOKENS = [t.strip().upper() for t in TOKENS_INPUT.split(",")]

# ── API Helper ──────────────────────────────────────────────────────


def llama_get(url: str, max_retries: int = 3) -> dict:
    """GET request to DeFiLlama with retry."""
    for attempt in range(max_retries):
        try:
            resp = httpx.get(url, timeout=30.0)
            if resp.status_code == 200:
                return resp.json()
            if resp.status_code == 429:
                time.sleep(30.0)
                continue
            if resp.status_code >= 500:
                time.sleep(5.0)
                continue
            resp.raise_for_status()
        except httpx.TimeoutException:
            if attempt < max_retries - 1:
                time.sleep(5.0)
                continue
            raise
    return {}


# ── Price Functions ─────────────────────────────────────────────────


def get_current_prices(coin_ids: list[str]) -> dict[str, dict]:
    """Get current prices for multiple coins.

    Args:
        coin_ids: List of DeFiLlama coin identifiers.

    Returns:
        Dict of coin_id -> {price, symbol, confidence, timestamp}.
    """
    joined = ",".join(coin_ids)
    data = llama_get(f"{COINS_BASE}/prices/current/{joined}")
    return data.get("coins", {})


def get_historical_prices(coin_ids: list[str], timestamp: int) -> dict[str, dict]:
    """Get prices at a historical timestamp.

    Args:
        coin_ids: List of DeFiLlama coin identifiers.
        timestamp: Unix timestamp (seconds).

    Returns:
        Dict of coin_id -> {price, symbol, confidence, timestamp}.
    """
    joined = ",".join(coin_ids)
    data = llama_get(f"{COINS_BASE}/prices/historical/{timestamp}/{joined}?searchWidth=12h")
    return data.get("coins", {})


def get_price_changes(coin_ids: list[str]) -> dict[str, float]:
    """Get percentage price changes.

    Args:
        coin_ids: List of DeFiLlama coin identifiers.

    Returns:
        Dict of coin_id -> percentage change.
    """
    joined = ",".join(coin_ids)
    data = llama_get(f"{COINS_BASE}/percentage/{joined}")
    return data.get("coins", {})


# ── Display ─────────────────────────────────────────────────────────


def print_report(
    selected: list[str],
    current: dict[str, dict],
    prices_7d: dict[str, dict],
    prices_30d: dict[str, dict],
) -> None:
    """Print price comparison report.

    Args:
        selected: List of token symbols.
        current: Current price data.
        prices_7d: 7-day ago prices.
        prices_30d: 30-day ago prices.
    """
    print(f"\n{'='*75}")
    print(f"TOKEN PRICE COMPARISON")
    print(f"{'='*75}")
    print(f"  {'Token':<8} {'Price':>14} {'7d Change':>12} {'30d Change':>12} {'Confidence':>12}")
    print(f"  {'─'*8} {'─'*14} {'─'*12} {'─'*12} {'─'*12}")

    for symbol in selected:
        coin_id = TOKEN_MAP.get(symbol)
        if not coin_id or coin_id not in current:
            print(f"  {symbol:<8} {'N/A':>14}")
            continue

        price = current[coin_id].get("price", 0)
        confidence = current[coin_id].get("confidence", 0)

        # 7d change
        price_7d = prices_7d.get(coin_id, {}).get("price", 0)
        change_7d = ((price / price_7d) - 1) * 100 if price_7d > 0 else 0

        # 30d change
        price_30d = prices_30d.get(coin_id, {}).get("price", 0)
        change_30d = ((price / price_30d) - 1) * 100 if price_30d > 0 else 0

        # Format price
        if price >= 100:
            price_str = f"${price:,.2f}"
        elif price >= 1:
            price_str = f"${price:.4f}"
        elif price >= 0.001:
            price_str = f"${price:.6f}"
        else:
            price_str = f"${price:.10f}"

        change_7d_str = f"{change_7d:+.1f}%" if price_7d > 0 else "N/A"
        change_30d_str = f"{change_30d:+.1f}%" if price_30d > 0 else "N/A"
        conf_str = f"{confidence:.2f}"

        print(f"  {symbol:<8} {price_str:>14} {change_7d_str:>12} {change_30d_str:>12} {conf_str:>12}")

    # Best/worst performers
    changes = {}
    for symbol in selected:
        coin_id = TOKEN_MAP.get(symbol)
        if not coin_id:
            continue
        price = current.get(coin_id, {}).get("price", 0)
        price_7d = prices_7d.get(coin_id, {}).get("price", 0)
        if price > 0 and price_7d > 0:
            changes[symbol] = ((price / price_7d) - 1) * 100

    if changes:
        best = max(changes, key=changes.get)
        worst = min(changes, key=changes.get)
        print(f"\n  7d Best:  {best} ({changes[best]:+.1f}%)")
        print(f"  7d Worst: {worst} ({changes[worst]:+.1f}%)")

    print()


# ── Main ────────────────────────────────────────────────────────────


def main() -> None:
    """Run price lookup and comparison."""
    # Resolve coin IDs
    coin_ids = []
    valid_tokens = []
    for symbol in SELECTED_TOKENS:
        if symbol in TOKEN_MAP:
            coin_ids.append(TOKEN_MAP[symbol])
            valid_tokens.append(symbol)
        elif symbol.startswith("solana:") or symbol.startswith("coingecko:"):
            coin_ids.append(symbol)
            valid_tokens.append(symbol)
        else:
            print(f"  Unknown token: {symbol} (use symbol name or full coin ID)")

    if not coin_ids:
        print("No valid tokens specified.")
        sys.exit(1)

    print(f"Looking up {len(coin_ids)} tokens...")

    # Timestamps
    now = datetime.now(tz=timezone.utc)
    ts_7d = int((now - timedelta(days=7)).timestamp())
    ts_30d = int((now - timedelta(days=30)).timestamp())

    # Fetch current and historical prices
    print("Fetching current prices...")
    current = get_current_prices(coin_ids)
    time.sleep(0.5)

    print("Fetching 7-day ago prices...")
    prices_7d = get_historical_prices(coin_ids, ts_7d)
    time.sleep(0.5)

    print("Fetching 30-day ago prices...")
    prices_30d = get_historical_prices(coin_ids, ts_30d)

    print_report(valid_tokens, current, prices_7d, prices_30d)


if __name__ == "__main__":
    main()