Spaces:
Sleeping
Sleeping
File size: 6,930 Bytes
f104fee 9b006e9 f104fee 9b006e9 f104fee 9b006e9 f104fee |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 |
from typing import Dict, Any, Optional
from pydantic import BaseModel, PrivateAttr
from src.tools.base_tool import BaseWeb3Tool, Web3ToolInput
from src.utils.logger import get_logger
logger = get_logger(__name__)
class DeFiLlamaTool(BaseWeb3Tool):
name: str = "defillama_data"
description: str = """Get DeFi protocol data, TVL, and yields from DeFiLlama.
Useful for: DeFi analysis, protocol rankings, TVL trends, yield farming data.
Input: protocol name or general DeFi query."""
args_schema: type[BaseModel] = Web3ToolInput
_base_url: str = PrivateAttr(default="https://api.llama.fi")
def __init__(self):
super().__init__()
async def _arun(self, query: str, filters: Optional[Dict[str, Any]] = None) -> str:
try:
filters = filters or {}
if filters.get("type") == "tvl_overview":
return await self._get_tvl_overview()
elif filters.get("type") == "protocol_data":
return await self._get_protocol_data(query)
elif query:
return await self._search_protocols(query)
else:
return await self._get_top_protocols()
except Exception as e:
logger.error(f"DeFiLlama error: {e}")
return f"⚠️ DeFiLlama service temporarily unavailable: {str(e)}"
async def _get_top_protocols(self) -> str:
try:
data = await self.make_request(f"{self._base_url}/protocols")
if not data or not isinstance(data, list):
return "⚠️ DeFi protocol data temporarily unavailable"
if len(data) == 0:
return "❌ No DeFi protocols found"
# Filter and validate protocols
valid_protocols = []
for protocol in data:
try:
tvl = protocol.get("tvl", 0)
if tvl is not None and tvl > 0:
valid_protocols.append(protocol)
except (TypeError, ValueError):
continue
if not valid_protocols:
return "⚠️ No valid protocol data available"
# Sort by TVL and take top 10
top_protocols = sorted(valid_protocols, key=lambda x: x.get("tvl", 0), reverse=True)[:10]
result = "🏦 **Top DeFi Protocols by TVL:**\n\n"
for i, protocol in enumerate(top_protocols, 1):
try:
name = protocol.get("name", "Unknown")
tvl = protocol.get("tvl", 0)
change = protocol.get("change_1d", 0)
chain = protocol.get("chain", "Multi-chain")
# Handle edge cases
if tvl <= 0:
continue
emoji = "📈" if change >= 0 else "📉"
tvl_formatted = f"${tvl/1e9:.2f}B" if tvl >= 1e9 else f"${tvl/1e6:.1f}M"
change_formatted = f"({change:+.2f}%)" if change is not None else "(N/A)"
result += f"{i}. **{name}** ({chain}): {tvl_formatted} TVL {emoji} {change_formatted}\n"
except (TypeError, KeyError, ValueError) as e:
logger.warning(f"Skipping invalid protocol data: {e}")
continue
return result if len(result.split('\n')) > 3 else "⚠️ Unable to format protocol data properly"
except Exception as e:
logger.error(f"Top protocols error: {e}")
return "⚠️ DeFi protocol data temporarily unavailable"
async def _get_tvl_overview(self) -> str:
try:
protocols_data = await self.make_request(f"{self.base_url}/protocols")
chains_data = await self.make_request(f"{self.base_url}/chains")
if not protocols_data or not chains_data:
return "TVL overview data unavailable"
total_tvl = sum(p.get("tvl", 0) for p in protocols_data)
top_chains = sorted(chains_data, key=lambda x: x.get("tvl", 0), reverse=True)[:5]
result = "🌐 **DeFi TVL Overview:**\n\n"
result += f"💰 **Total TVL**: ${total_tvl/1e9:.2f}B\n\n"
result += "**Top Chains by TVL:**\n"
for i, chain in enumerate(top_chains, 1):
name = chain.get("name", "Unknown")
tvl = chain.get("tvl", 0)
result += f"{i}. **{name}**: ${tvl/1e9:.2f}B\n"
return result
except Exception:
return await self._get_top_protocols()
async def _get_protocol_data(self, protocol: str) -> str:
protocols = await self.make_request(f"{self.base_url}/protocols")
if not protocols:
return f"No data available for {protocol}"
matching_protocol = None
for p in protocols:
if protocol.lower() in p.get("name", "").lower():
matching_protocol = p
break
if not matching_protocol:
return f"Protocol '{protocol}' not found"
name = matching_protocol.get("name", "Unknown")
tvl = matching_protocol.get("tvl", 0)
change_1d = matching_protocol.get("change_1d", 0)
change_7d = matching_protocol.get("change_7d", 0)
chain = matching_protocol.get("chain", "Multi-chain")
category = matching_protocol.get("category", "Unknown")
result = f"🏛️ **{name} Protocol Analysis:**\n\n"
result += f"💰 **TVL**: ${tvl/1e9:.2f}B\n"
result += f"📊 **24h Change**: {change_1d:+.2f}%\n"
result += f"📈 **7d Change**: {change_7d:+.2f}%\n"
result += f"⛓️ **Chain**: {chain}\n"
result += f"🏷️ **Category**: {category}\n"
return result
async def _search_protocols(self, query: str) -> str:
protocols = await self.make_request(f"{self.base_url}/protocols")
if not protocols:
return "No protocol data available"
matching = [p for p in protocols if query.lower() in p.get("name", "").lower()][:5]
if not matching:
return f"No protocols found matching '{query}'"
result = f"🔍 **Protocols matching '{query}':**\n\n"
for protocol in matching:
name = protocol.get("name", "Unknown")
tvl = protocol.get("tvl", 0)
chain = protocol.get("chain", "Multi-chain")
result += f"• **{name}** ({chain}): ${tvl/1e9:.2f}B TVL\n"
return result
|