177 lines
6.9 KiB
Python
Executable File
177 lines
6.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""DefiLlama API connector — TVL, token prices, yield/APY data.
|
|
|
|
No authentication required. All endpoints are free.
|
|
API base: https://api.llama.fi | Prices: https://coins.llama.fi | Yields: https://yields.llama.fi
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from typing import Any
|
|
|
|
import requests
|
|
|
|
BASE = "https://api.llama.fi"
|
|
COINS_BASE = "https://coins.llama.fi"
|
|
YIELDS_BASE = "https://yields.llama.fi"
|
|
|
|
TIMEOUT = 30
|
|
|
|
|
|
def _get(url: str, params: dict | None = None) -> Any:
|
|
r = requests.get(url, params=params, timeout=TIMEOUT)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
# ── Protocol / TVL ──────────────────────────────────────────────────────────
|
|
|
|
def get_protocols(limit: int = 20) -> list[dict]:
|
|
"""Top protocols by TVL."""
|
|
data = _get(f"{BASE}/protocols")
|
|
# Sort by tvl descending, filter out CEXes
|
|
protos = [p for p in data if p.get("category") != "CEX" and p.get("tvl")]
|
|
protos.sort(key=lambda p: p.get("tvl", 0), reverse=True)
|
|
return protos[:limit]
|
|
|
|
|
|
def get_tvl(protocol: str) -> dict:
|
|
"""Get current TVL for a specific protocol (slug name)."""
|
|
val = _get(f"{BASE}/tvl/{protocol}")
|
|
return {"protocol": protocol, "tvl": val}
|
|
|
|
|
|
def get_protocol_detail(protocol: str) -> dict:
|
|
"""Full protocol details including chain breakdowns."""
|
|
return _get(f"{BASE}/protocol/{protocol}")
|
|
|
|
|
|
# ── Token Prices ────────────────────────────────────────────────────────────
|
|
|
|
def get_prices(coins: list[str]) -> dict:
|
|
"""Get current prices. Coins format: 'coingecko:ethereum', 'ethereum:0x...', etc."""
|
|
joined = ",".join(coins)
|
|
data = _get(f"{COINS_BASE}/prices/current/{joined}")
|
|
return data.get("coins", {})
|
|
|
|
|
|
# ── Yields / APY ────────────────────────────────────────────────────────────
|
|
|
|
def get_yield_pools(limit: int = 30, min_tvl: float = 1_000_000, stablecoin_only: bool = False) -> list[dict]:
|
|
"""Top yield pools sorted by APY."""
|
|
data = _get(f"{YIELDS_BASE}/pools")
|
|
pools = data.get("data", [])
|
|
# Filter
|
|
pools = [p for p in pools if (p.get("tvlUsd") or 0) >= min_tvl and (p.get("apy") or 0) > 0]
|
|
if stablecoin_only:
|
|
pools = [p for p in pools if p.get("stablecoin")]
|
|
pools.sort(key=lambda p: p.get("apy", 0), reverse=True)
|
|
return pools[:limit]
|
|
|
|
|
|
# ── Summary helpers ─────────────────────────────────────────────────────────
|
|
|
|
def _fmt_usd(v: float) -> str:
|
|
if v >= 1e9:
|
|
return f"${v/1e9:.2f}B"
|
|
if v >= 1e6:
|
|
return f"${v/1e6:.1f}M"
|
|
return f"${v:,.0f}"
|
|
|
|
|
|
def summary_protocols(protos: list[dict]) -> str:
|
|
lines = ["═══ Top Protocols by TVL ═══", ""]
|
|
for i, p in enumerate(protos, 1):
|
|
lines.append(f" {i:>2}. {p['name']:<25} TVL: {_fmt_usd(p.get('tvl', 0)):>12} chain: {p.get('chain', '?')}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def summary_prices(prices: dict) -> str:
|
|
lines = ["═══ Token Prices ═══", ""]
|
|
for coin, info in prices.items():
|
|
lines.append(f" {info.get('symbol', coin):<10} ${info['price']:>12,.2f} (confidence: {info.get('confidence', '?')})")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def summary_yields(pools: list[dict]) -> str:
|
|
lines = ["═══ Top Yield Pools ═══", ""]
|
|
for i, p in enumerate(pools, 1):
|
|
lines.append(
|
|
f" {i:>2}. {p.get('symbol','?'):<25} APY: {p.get('apy',0):>8.2f}% "
|
|
f"TVL: {_fmt_usd(p.get('tvlUsd',0)):>10} {p.get('chain','?')}/{p.get('project','?')}"
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
|
|
# ── CLI ─────────────────────────────────────────────────────────────────────
|
|
|
|
def main():
|
|
common = argparse.ArgumentParser(add_help=False)
|
|
common.add_argument("--pretty", action="store_true", help="Pretty-print JSON output")
|
|
common.add_argument("--summary", action="store_true", help="Human-readable summary")
|
|
parser = argparse.ArgumentParser(description="DefiLlama data connector", parents=[common])
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
|
|
# protocols
|
|
p_proto = sub.add_parser("protocols", help="Top protocols by TVL", parents=[common])
|
|
p_proto.add_argument("--limit", type=int, default=20)
|
|
|
|
# tvl
|
|
p_tvl = sub.add_parser("tvl", help="TVL for a specific protocol", parents=[common])
|
|
p_tvl.add_argument("protocol", help="Protocol slug (e.g. aave, lido)")
|
|
|
|
# prices
|
|
p_price = sub.add_parser("prices", help="Token prices", parents=[common])
|
|
p_price.add_argument("coins", nargs="+", help="Coin IDs: coingecko:ethereum, ethereum:0x...")
|
|
|
|
# yields
|
|
p_yield = sub.add_parser("yields", help="Top yield pools", parents=[common])
|
|
p_yield.add_argument("--limit", type=int, default=30)
|
|
p_yield.add_argument("--min-tvl", type=float, default=1_000_000)
|
|
p_yield.add_argument("--stablecoins", action="store_true")
|
|
|
|
args = parser.parse_args()
|
|
|
|
try:
|
|
if args.command == "protocols":
|
|
data = get_protocols(args.limit)
|
|
if args.summary:
|
|
print(summary_protocols(data))
|
|
return
|
|
result = [{"name": p["name"], "tvl": p.get("tvl"), "chain": p.get("chain"), "category": p.get("category"), "symbol": p.get("symbol")} for p in data]
|
|
elif args.command == "tvl":
|
|
result = get_tvl(args.protocol)
|
|
if args.summary:
|
|
print(f"{args.protocol}: {_fmt_usd(result['tvl'])}")
|
|
return
|
|
elif args.command == "prices":
|
|
result = get_prices(args.coins)
|
|
if args.summary:
|
|
print(summary_prices(result))
|
|
return
|
|
elif args.command == "yields":
|
|
data = get_yield_pools(args.limit, args.min_tvl, args.stablecoins)
|
|
if args.summary:
|
|
print(summary_yields(data))
|
|
return
|
|
result = [{"symbol": p.get("symbol"), "apy": p.get("apy"), "tvlUsd": p.get("tvlUsd"), "chain": p.get("chain"), "project": p.get("project"), "pool": p.get("pool")} for p in data]
|
|
else:
|
|
parser.print_help()
|
|
return
|
|
|
|
indent = 2 if args.pretty else None
|
|
json.dump(result, sys.stdout, indent=indent)
|
|
print()
|
|
|
|
except requests.HTTPError as e:
|
|
print(json.dumps({"error": str(e)}), file=sys.stderr)
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(json.dumps({"error": f"Unexpected: {e}"}), file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|