From 7dfc5dbd2cada73d61d9fdd86b8a6c36d8c7286d Mon Sep 17 00:00:00 2001 From: Muninn Date: Sun, 23 Feb 2025 23:27:22 +0800 Subject: [PATCH 1/3] improve: better system prompt --- app/core/prompt.py | 37 ++++++++++++++----------------------- 1 file changed, 14 insertions(+), 23 deletions(-) diff --git a/app/core/prompt.py b/app/core/prompt.py index dd89151..da55bc8 100644 --- a/app/core/prompt.py +++ b/app/core/prompt.py @@ -8,21 +8,28 @@ def agent_prompt(agent: Agent, agent_data: AgentData) -> str: prompt = "# SYSTEM PROMPT\n\n" if config.system_prompt: prompt += config.system_prompt + "\n\n" - prompt += "You are an autonomous AI agent in an opensource platform 'IntentKit'.\n" + prompt += "You are an ai agent built using IntentKit.\n" prompt += "Your tools are called 'skills'.\n" + prompt += "If your skill fails to execute due to a technical error ask the user to try again later. If someone asks you to do something you can't do with your currently available skills, you must say so, recommend them to submit their feedback to the IntentKit team at https://github.com/crestalnetwork/intentkit. Be concise and helpful with your responses. Refrain from restating your skills' descriptions unless it is explicitly requested.\n" if agent.name: prompt += f"Your name is {agent.name}.\n" if agent.ticker: prompt += f"Your ticker symbol is {agent.ticker}.\n" - if agent_data and agent.cdp_enabled: - if agent_data.cdp_wallet_data: + if agent_data: + if agent_data.cdp_wallet_data and agent.cdp_enabled: wallet_data = json.loads(agent_data.cdp_wallet_data) prompt += f"Your CDP wallet address in {agent.cdp_network_id} is {wallet_data['default_address_id']} .\n" + prompt += ( + "## Crypto Wallet\n\n" + "Wallet addresses are public information. If someone asks for your default wallet, current wallet, " + "personal wallet, crypto wallet, or wallet public address, don't use any address in message history, " + "you must use the 'get_wallet_details' tool to retrieve your wallet address every time." + ) + prompt += "When you get balance from tools, or pass amount to tools, don't forget they have decimals.\n" + prompt += "USDC and USDT has 6 decimals, you must divide the amount you get by 10^6, multiply 10^6 when passing to tools.\n" + prompt += "Other currencies include native ETH usually has 18 decimals, you need divide or multiply 10^18.\n" if agent.cdp_network_id == "base-mainnet": prompt += "The USDC contract address in base-mainnet is 0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913\n" - prompt += "When you get balance from tools, or pass amount to tools, don't forget they have decimals.\n" - prompt += "USDC and USDT has 6 decimals, you must divide the amount you get by 10^6, multiply 10^6 when passing to tools.\n" - prompt += "Other currencies include native ETH usually has 18 decimals, you need divide or multiply 10^18.\n" if agent_data.twitter_id: prompt += f"Your twitter id is {agent_data.twitter_id}, never reply or retweet yourself.\n" if agent_data.twitter_username: @@ -44,23 +51,7 @@ def agent_prompt(agent: Agent, agent_data: AgentData) -> str: prompt += f"## Principles\n\n{agent.principles}\n\n" if agent.prompt: prompt += f"## Initial Rules\n\n{agent.prompt}\n\n" - elif agent.cdp_enabled: - prompt += ( - "## Coinbase AgentKit Skills Guide\n\n" - "You are a helpful agent that can interact onchain using the Coinbase Developer Platform AgentKit. " - "You are empowered to interact onchain using your tools. If you ever need funds, you can request " - "them from the faucet if you are on network ID 'base-mainnet'. If not, you can provide your wallet " - "details and request funds from the user. Before executing your first action, get the wallet details " - "to see what network you're on. If there is a 5XX (internal) HTTP error code, ask the user to try " - "again later. If someone asks you to do something you can't do with your currently available tools, " - "you must say so, and encourage them to implement it themselves using the CDP SDK + Agentkit, " - "recommend they go to docs.cdp.coinbase.com for more information. Be concise and helpful with your " - "responses. Refrain from restating your tools' descriptions unless it is explicitly requested." - "\n\nWallet addresses are public information. If someone asks for your default wallet, current wallet, " - "personal wallet, crypto wallet, or wallet public address, don't use any address in message history, " - "you must use the 'get_wallet_details' tool to retrieve your wallet address every time." - ) - if agent.enso_enabled: + if agent.enso_skills and len(agent.enso_skills) > 0: prompt += """## ENSO Skills Guide\n\nYou are integrated with the Enso API. You can use enso_get_tokens to retrieve token information, including APY, Protocol Slug, Symbol, Address, Decimals, and underlying tokens. When interacting with token amounts, ensure to multiply input amounts by the token's decimal places and divide output amounts by the token's decimals. From 15adcc1071451c9aef32e176104045f224edcd68 Mon Sep 17 00:00:00 2001 From: Muninn Date: Sun, 23 Feb 2025 23:27:34 +0800 Subject: [PATCH 2/3] doc: cdp skills update --- docs/skills/cdp.md | 116 ++++++++++++--------------------------------- 1 file changed, 29 insertions(+), 87 deletions(-) diff --git a/docs/skills/cdp.md b/docs/skills/cdp.md index dc8e98d..ff8a631 100644 --- a/docs/skills/cdp.md +++ b/docs/skills/cdp.md @@ -2,93 +2,35 @@ ## CDP AgentKit -All CDP Skills are supported by [AgentKit](https://github.com/coinbase/cdp-agentkit/). +All CDP Skills are supported by [AgentKit](https://github.com/coinbase/agentkit/). AgentKit supports the following tools: -### Wallet Management -- `get_wallet_details` - Get detailed information about your MPC Wallet, including the wallet's address and current network. - -- `get_balance` - Query token balances for specific assets in your wallet. Supports querying balances for any token using its asset ID (e.g., "eth", "usdc") or contract address. - -- `request_faucet_funds` - Request test tokens from the [Base Sepolia faucet](https://portal.cdp.coinbase.com/products/faucet). Useful for testing and development on testnet. - -- `address_reputation` - Check the reputation and risk assessment of any Ethereum address. Helps evaluate the trustworthiness of addresses you interact with. - -### Asset Operations -- `transfer` - Transfer tokens between addresses. Features: - - Supports any ERC20 token or native asset (ETH) - - Accepts amount in standard units (e.g., "15", "0.000001") - - Destination can be an address, ENS name (example.eth), or Basename (example.base.eth) - - Supports gasless transfers for USDC on Base networks (base-sepolia and base-mainnet) - -- `trade` - Execute token trades on supported DEXs (mainnet networks only). Allows trading between any supported assets by specifying: - - Amount to trade - - Source asset ID (e.g., "eth", "usdc", or contract address) - - Target asset ID - -- `wrap_eth` - Convert ETH to WETH (Wrapped Ether) for DeFi compatibility. WETH is the ERC20 token version of ETH required by many DeFi protocols. - -### Token Management -- `deploy_token` - Deploy custom [ERC-20](https://www.coinbase.com/learn/crypto-glossary/what-is-erc-20) token contracts. Specify: - - Token name - - Symbol - - Total supply - The deploying wallet becomes the owner and initial token holder. - -- `deploy_contract` - Deploy any smart contract with custom bytecode and constructor arguments. Provides full flexibility for deploying any type of contract. - -### NFT Operations -- `get_balance_nft` - Query NFT balances for specific collections. Check ownership of NFTs in any ERC721 contract. - -- `mint_nft` - Mint NFTs from existing contracts. Useful for participating in NFT drops or creating new tokens in your own collections. - -- `deploy_nft` - Deploy new NFT (ERC-721) contracts with customizable parameters: - - Collection name - - Symbol - - Base URI for token metadata - -- `transfer_nft` - Transfer NFTs between addresses. Supports: - - Any ERC721 contract - - Transfer by token ID - - Custom source address - - Destination can be address, ENS, or Basename - -### Base Name Service -- `register_basename` - Register a [Basename](https://www.base.org/names) for your wallet address. Basenames are human-readable identifiers for your wallet on Base network (e.g., "example.base.eth"). - -### Zora Wow Integration -- `wow_create_token` - Deploy a token using [Zora's Wow Launcher](https://wow.xyz/mechanics) with bonding curve (Base only). Specify: - - Token name - - Symbol - - Optional IPFS metadata URI - -- `wow_buy_token` - Buy [Zora Wow](https://wow.xyz/) ERC-20 memecoins with ETH (Base only). Purchase tokens from existing Wow contracts. - -- `wow_sell_token` - Sell [Zora Wow](https://wow.xyz/) ERC-20 memecoins for ETH (Base only). Specify: - - Contract address - - Amount in wei (1 wei = 0.000000000000000001 tokens) - -### Superfluid Integration -- `superfluid_create_flow` - Create a continuous token streaming flow. Set up recurring payments by specifying: - - Recipient address - - Token address - - Flow rate (tokens per second in wei) - -- `superfluid_update_flow` - Modify an existing token streaming flow. Adjust the flow rate of ongoing streams. - -- `superfluid_delete_flow` - Stop and delete an existing token streaming flow. Terminate ongoing payment streams. - -### Morpho Integration -- `morpho_deposit` - Deposit assets into Morpho's yield-generating vaults. Earn yield on your deposited tokens. - -- `morpho_withdraw` - Withdraw assets from Morpho's yield-generating vaults. Access your deposited funds and earned yield. - -### Pyth Network Integration -- `pyth_fetch_price` - Get real-time price data from Pyth Network oracles. Access accurate price feeds for various assets. - -- `pyth_fetch_price_feed_id` - Retrieve price feed IDs for Pyth Network price feeds. Required for accessing specific price data streams. - -Any action not supported by default by AgentKit can be added by [adding agent capabilities](https://docs.cdp.coinbase.com/agentkit/docs/add-agent-capabilities). - -AgentKit supports every network that the [CDP SDK supports](https://docs.cdp.coinbase.com/cdp-apis/docs/networks). +``` +WalletActionProvider_get_balance +WalletActionProvider_get_wallet_details +WalletActionProvider_native_transfer +CdpApiActionProvider_address_reputation +CdpApiActionProvider_request_faucet_funds +CdpWalletActionProvider_deploy_contract +CdpWalletActionProvider_deploy_nft +CdpWalletActionProvider_deploy_token +CdpWalletActionProvider_trade +PythActionProvider_fetch_price +PythActionProvider_fetch_price_feed_id +BasenameActionProvider_register_basename +ERC20ActionProvider_get_balance +ERC20ActionProvider_transfer +Erc721ActionProvider_get_balance +Erc721ActionProvider_mint +Erc721ActionProvider_transfer +WethActionProvider_wrap_eth +MorphoActionProvider_deposit +MorphoActionProvider_withdraw +SuperfluidActionProvider_create_flow +SuperfluidActionProvider_delete_flow +SuperfluidActionProvider_update_flow +WowActionProvider_buy_token +WowActionProvider_create_token +WowActionProvider_sell_token +``` \ No newline at end of file From 0f86132fdb49ab6cfacec1324742934f5cf41ea2 Mon Sep 17 00:00:00 2001 From: Muninn Date: Sun, 23 Feb 2025 23:46:29 +0800 Subject: [PATCH 3/3] fix: defillama lint errors --- skills/defillama/__init__.py | 87 +++--- skills/defillama/api.py | 143 +++++----- skills/defillama/base.py | 21 +- .../coins/fetch_batch_historical_prices.py | 40 +-- skills/defillama/coins/fetch_block.py | 73 ++--- .../defillama/coins/fetch_current_prices.py | 44 +-- skills/defillama/coins/fetch_first_price.py | 41 +-- .../coins/fetch_historical_prices.py | 50 +--- skills/defillama/coins/fetch_price_chart.py | 56 +--- .../defillama/coins/fetch_price_percentage.py | 25 +- skills/defillama/config/chains.py | 12 +- skills/defillama/fees/fetch_fees_overview.py | 38 ++- .../stablecoins/fetch_stablecoin_chains.py | 83 ++---- .../stablecoins/fetch_stablecoin_charts.py | 58 ++-- .../stablecoins/fetch_stablecoin_prices.py | 30 +- .../stablecoins/fetch_stablecoins.py | 96 ++----- .../defillama/tests/api_integration.test.py | 92 +++--- skills/defillama/tests/api_unit.test.py | 266 ++++++++---------- .../tvl/fetch_chain_historical_tvl.py | 59 ++-- skills/defillama/tvl/fetch_chains.py | 73 ++--- skills/defillama/tvl/fetch_historical_tvl.py | 34 +-- skills/defillama/tvl/fetch_protocol.py | 55 +++- .../tvl/fetch_protocol_current_tvl.py | 44 +-- skills/defillama/tvl/fetch_protocols.py | 74 +++-- .../defillama/volumes/fetch_dex_overview.py | 121 +++----- skills/defillama/volumes/fetch_dex_summary.py | 24 +- .../volumes/fetch_options_overview.py | 38 ++- skills/defillama/yields/fetch_pool_chart.py | 71 ++--- skills/defillama/yields/fetch_pools.py | 169 +++-------- 29 files changed, 804 insertions(+), 1213 deletions(-) diff --git a/skills/defillama/__init__.py b/skills/defillama/__init__.py index 3f06735..a077b57 100644 --- a/skills/defillama/__init__.py +++ b/skills/defillama/__init__.py @@ -3,41 +3,58 @@ from abstracts.agent import AgentStoreABC from abstracts.skill import SkillStoreABC from skills.defillama.base import DefiLlamaBaseTool - -# TVL Tools -from skills.defillama.tvl.fetch_protocols import DefiLlamaFetchProtocols -from skills.defillama.tvl.fetch_protocol import DefiLlamaFetchProtocol -from skills.defillama.tvl.fetch_historical_tvl import DefiLlamaFetchHistoricalTvl -from skills.defillama.tvl.fetch_chain_historical_tvl import DefiLlamaFetchChainHistoricalTvl -from skills.defillama.tvl.fetch_protocol_current_tvl import DefiLlamaFetchProtocolCurrentTvl -from skills.defillama.tvl.fetch_chains import DefiLlamaFetchChains +from skills.defillama.coins.fetch_batch_historical_prices import ( + DefiLlamaFetchBatchHistoricalPrices, +) +from skills.defillama.coins.fetch_block import DefiLlamaFetchBlock # Coins Tools from skills.defillama.coins.fetch_current_prices import DefiLlamaFetchCurrentPrices -from skills.defillama.coins.fetch_historical_prices import DefiLlamaFetchHistoricalPrices -from skills.defillama.coins.fetch_batch_historical_prices import DefiLlamaFetchBatchHistoricalPrices +from skills.defillama.coins.fetch_first_price import DefiLlamaFetchFirstPrice +from skills.defillama.coins.fetch_historical_prices import ( + DefiLlamaFetchHistoricalPrices, +) from skills.defillama.coins.fetch_price_chart import DefiLlamaFetchPriceChart from skills.defillama.coins.fetch_price_percentage import DefiLlamaFetchPricePercentage -from skills.defillama.coins.fetch_first_price import DefiLlamaFetchFirstPrice -from skills.defillama.coins.fetch_block import DefiLlamaFetchBlock + +# Fees Tools +from skills.defillama.fees.fetch_fees_overview import DefiLlamaFetchFeesOverview +from skills.defillama.stablecoins.fetch_stablecoin_chains import ( + DefiLlamaFetchStablecoinChains, +) +from skills.defillama.stablecoins.fetch_stablecoin_charts import ( + DefiLlamaFetchStablecoinCharts, +) +from skills.defillama.stablecoins.fetch_stablecoin_prices import ( + DefiLlamaFetchStablecoinPrices, +) # Stablecoins Tools from skills.defillama.stablecoins.fetch_stablecoins import DefiLlamaFetchStablecoins -from skills.defillama.stablecoins.fetch_stablecoin_charts import DefiLlamaFetchStablecoinCharts -from skills.defillama.stablecoins.fetch_stablecoin_chains import DefiLlamaFetchStablecoinChains -from skills.defillama.stablecoins.fetch_stablecoin_prices import DefiLlamaFetchStablecoinPrices +from skills.defillama.tvl.fetch_chain_historical_tvl import ( + DefiLlamaFetchChainHistoricalTvl, +) +from skills.defillama.tvl.fetch_chains import DefiLlamaFetchChains +from skills.defillama.tvl.fetch_historical_tvl import DefiLlamaFetchHistoricalTvl +from skills.defillama.tvl.fetch_protocol import DefiLlamaFetchProtocol +from skills.defillama.tvl.fetch_protocol_current_tvl import ( + DefiLlamaFetchProtocolCurrentTvl, +) -# Yields Tools -from skills.defillama.yields.fetch_pools import DefiLlamaFetchPools -from skills.defillama.yields.fetch_pool_chart import DefiLlamaFetchPoolChart +# TVL Tools +from skills.defillama.tvl.fetch_protocols import DefiLlamaFetchProtocols # Volumes Tools from skills.defillama.volumes.fetch_dex_overview import DefiLlamaFetchDexOverview from skills.defillama.volumes.fetch_dex_summary import DefiLlamaFetchDexSummary -from skills.defillama.volumes.fetch_options_overview import DefiLlamaFetchOptionsOverview +from skills.defillama.volumes.fetch_options_overview import ( + DefiLlamaFetchOptionsOverview, +) +from skills.defillama.yields.fetch_pool_chart import DefiLlamaFetchPoolChart + +# Yields Tools +from skills.defillama.yields.fetch_pools import DefiLlamaFetchPools -# Fees Tools -from skills.defillama.fees.fetch_fees_overview import DefiLlamaFetchFeesOverview def get_defillama_skill( name: str, @@ -77,19 +94,19 @@ def get_defillama_skill( agent_id=agent_id, agent_store=agent_store, ) - elif name == "fetch_historical_tvl": + elif name == "fetch_historical_tvl": return DefiLlamaFetchHistoricalTvl( skill_store=store, agent_id=agent_id, agent_store=agent_store, ) - elif name == "fetch_chain_historical_tvl": + elif name == "fetch_chain_historical_tvl": return DefiLlamaFetchChainHistoricalTvl( skill_store=store, agent_id=agent_id, agent_store=agent_store, ) - elif name == "fetch_protocol_current_tvl": + elif name == "fetch_protocol_current_tvl": return DefiLlamaFetchProtocolCurrentTvl( skill_store=store, agent_id=agent_id, @@ -101,7 +118,7 @@ def get_defillama_skill( agent_id=agent_id, agent_store=agent_store, ) - + # Coins Skills elif name == "fetch_current_prices": return DefiLlamaFetchCurrentPrices( @@ -145,7 +162,7 @@ def get_defillama_skill( agent_id=agent_id, agent_store=agent_store, ) - + # Stablecoins Skills elif name == "fetch_stablecoins": return DefiLlamaFetchStablecoins( @@ -153,7 +170,9 @@ def get_defillama_skill( agent_id=agent_id, agent_store=agent_store, ) - elif name == "fetch_stablecoin_charts": # Handles both all and chain-specific charts + elif ( + name == "fetch_stablecoin_charts" + ): # Handles both all and chain-specific charts return DefiLlamaFetchStablecoinCharts( skill_store=store, agent_id=agent_id, @@ -171,7 +190,7 @@ def get_defillama_skill( agent_id=agent_id, agent_store=agent_store, ) - + # Yields Skills elif name == "fetch_pools": return DefiLlamaFetchPools( @@ -185,7 +204,7 @@ def get_defillama_skill( agent_id=agent_id, agent_store=agent_store, ) - + # Volumes Skills elif name == "fetch_dex_overview": # Handles both base and chain-specific overviews return DefiLlamaFetchDexOverview( @@ -199,15 +218,19 @@ def get_defillama_skill( agent_id=agent_id, agent_store=agent_store, ) - elif name == "fetch_options_overview": # Handles both base and chain-specific overviews + elif ( + name == "fetch_options_overview" + ): # Handles both base and chain-specific overviews return DefiLlamaFetchOptionsOverview( skill_store=store, agent_id=agent_id, agent_store=agent_store, ) - + # Fees Skills - elif name == "fetch_fees_overview": # Handles both base and chain-specific overviews + elif ( + name == "fetch_fees_overview" + ): # Handles both base and chain-specific overviews return DefiLlamaFetchFeesOverview( skill_store=store, agent_id=agent_id, diff --git a/skills/defillama/api.py b/skills/defillama/api.py index 891e060..45528ff 100644 --- a/skills/defillama/api.py +++ b/skills/defillama/api.py @@ -1,10 +1,9 @@ """DeFi Llama API implementation and shared schemas.""" -from typing import List, Optional, Union from datetime import datetime +from typing import List, Optional import httpx -from pydantic import BaseModel, Field DEFILLAMA_TVL_BASE_URL = "https://api.llama.fi" DEFILLAMA_COINS_BASE_URL = "https://coins.llama.fi" @@ -13,6 +12,7 @@ DEFILLAMA_VOLUMES_BASE_URL = "https://api.llama.fi" DEFILLAMA_FEES_BASE_URL = "https://api.llama.fi" + # TVL API Functions async def fetch_protocols() -> dict: """List all protocols on defillama along with their TVL.""" @@ -23,6 +23,7 @@ async def fetch_protocols() -> dict: return {"error": f"API returned status code {response.status_code}"} return response.json() + async def fetch_protocol(protocol: str) -> dict: """Get historical TVL of a protocol and breakdowns by token and chain.""" url = f"{DEFILLAMA_TVL_BASE_URL}/protocol/{protocol}" @@ -32,6 +33,7 @@ async def fetch_protocol(protocol: str) -> dict: return {"error": f"API returned status code {response.status_code}"} return response.json() + async def fetch_historical_tvl() -> dict: """Get historical TVL of DeFi on all chains.""" url = f"{DEFILLAMA_TVL_BASE_URL}/v2/historicalChainTvl" @@ -41,6 +43,7 @@ async def fetch_historical_tvl() -> dict: return {"error": f"API returned status code {response.status_code}"} return response.json() + async def fetch_chain_historical_tvl(chain: str) -> dict: """Get historical TVL of a specific chain.""" url = f"{DEFILLAMA_TVL_BASE_URL}/v2/historicalChainTvl/{chain}" @@ -50,6 +53,7 @@ async def fetch_chain_historical_tvl(chain: str) -> dict: return {"error": f"API returned status code {response.status_code}"} return response.json() + async def fetch_protocol_current_tvl(protocol: str) -> dict: """Get current TVL of a protocol.""" url = f"{DEFILLAMA_TVL_BASE_URL}/tvl/{protocol}" @@ -59,6 +63,7 @@ async def fetch_protocol_current_tvl(protocol: str) -> dict: return {"error": f"API returned status code {response.status_code}"} return response.json() + async def fetch_chains() -> dict: """Get current TVL of all chains.""" url = f"{DEFILLAMA_TVL_BASE_URL}/v2/chains" @@ -68,223 +73,227 @@ async def fetch_chains() -> dict: return {"error": f"API returned status code {response.status_code}"} return response.json() -# Coins API Functions + +# Coins API Functions async def fetch_current_prices(coins: List[str]) -> dict: - """Get current prices of tokens by contract address using a 4-hour search window. """ - coins_str = ','.join(coins) + """Get current prices of tokens by contract address using a 4-hour search window.""" + coins_str = ",".join(coins) url = f"{DEFILLAMA_COINS_BASE_URL}/prices/current/{coins_str}?searchWidth=4h" - + async with httpx.AsyncClient() as client: response = await client.get(url) if response.status_code != 200: return {"error": f"API returned status code {response.status_code}"} return response.json() + async def fetch_historical_prices(timestamp: int, coins: List[str]) -> dict: - """Get historical prices of tokens by contract address using a 4-hour search window. """ - coins_str = ','.join(coins) + """Get historical prices of tokens by contract address using a 4-hour search window.""" + coins_str = ",".join(coins) url = f"{DEFILLAMA_COINS_BASE_URL}/prices/historical/{timestamp}/{coins_str}?searchWidth=4h" - + async with httpx.AsyncClient() as client: response = await client.get(url) if response.status_code != 200: return {"error": f"API returned status code {response.status_code}"} return response.json() + async def fetch_batch_historical_prices(coins_timestamps: dict) -> dict: - """Get historical prices for multiple tokens at multiple timestamps. """ + """Get historical prices for multiple tokens at multiple timestamps.""" url = f"{DEFILLAMA_COINS_BASE_URL}/batchHistorical" - + async with httpx.AsyncClient() as client: response = await client.get( - url, - params={ - "coins": coins_timestamps, - "searchWidth": "600" - } + url, params={"coins": coins_timestamps, "searchWidth": "600"} ) if response.status_code != 200: return {"error": f"API returned status code {response.status_code}"} return response.json() + async def fetch_price_chart(coins: List[str]) -> dict: """Get historical price chart data from the past day for multiple tokens.""" - coins_str = ','.join(coins) + coins_str = ",".join(coins) start_time = int(datetime.now().timestamp()) - 86400 # now - 1 day - + url = f"{DEFILLAMA_COINS_BASE_URL}/chart/{coins_str}" - params = { - "start": start_time, - "span": 10, - "period": "2d", - "searchWidth": "600" - } - + params = {"start": start_time, "span": 10, "period": "2d", "searchWidth": "600"} + async with httpx.AsyncClient() as client: response = await client.get(url, params=params) if response.status_code != 200: return {"error": f"API returned status code {response.status_code}"} return response.json() + async def fetch_price_percentage(coins: List[str]) -> dict: - """Get price percentage changes for multiple tokens over a 24h period. """ - coins_str = ','.join(coins) + """Get price percentage changes for multiple tokens over a 24h period.""" + coins_str = ",".join(coins) current_timestamp = int(datetime.now().timestamp()) - + url = f"{DEFILLAMA_COINS_BASE_URL}/percentage/{coins_str}" - params = { - "timestamp": current_timestamp, - "lookForward": "false", - "period": "24h" - } - + params = {"timestamp": current_timestamp, "lookForward": "false", "period": "24h"} + async with httpx.AsyncClient() as client: response = await client.get(url, params=params) if response.status_code != 200: return {"error": f"API returned status code {response.status_code}"} return response.json() + async def fetch_first_price(coins: List[str]) -> dict: - """Get first recorded price data for multiple tokens. """ - coins_str = ','.join(coins) + """Get first recorded price data for multiple tokens.""" + coins_str = ",".join(coins) url = f"{DEFILLAMA_COINS_BASE_URL}/prices/first/{coins_str}" - + async with httpx.AsyncClient() as client: response = await client.get(url) if response.status_code != 200: return {"error": f"API returned status code {response.status_code}"} return response.json() + async def fetch_block(chain: str) -> dict: - """Get current block data for a specific chain. """ + """Get current block data for a specific chain.""" current_timestamp = int(datetime.now().timestamp()) url = f"{DEFILLAMA_COINS_BASE_URL}/block/{chain}/{current_timestamp}" - + async with httpx.AsyncClient() as client: response = await client.get(url) if response.status_code != 200: return {"error": f"API returned status code {response.status_code}"} return response.json() + # Stablecoins API Functions async def fetch_stablecoins() -> dict: - """Get comprehensive stablecoin data from DeFi Llama. """ + """Get comprehensive stablecoin data from DeFi Llama.""" url = f"{DEFILLAMA_STABLECOINS_BASE_URL}/stablecoins" - params = { - "includePrices": "true" - } - + params = {"includePrices": "true"} + async with httpx.AsyncClient() as client: response = await client.get(url, params=params) if response.status_code != 200: return {"error": f"API returned status code {response.status_code}"} return response.json() -async def fetch_stablecoin_charts(stablecoin_id: str, chain: Optional[str] = None) -> dict: - """Get historical circulating supply data for a stablecoin. """ + +async def fetch_stablecoin_charts( + stablecoin_id: str, chain: Optional[str] = None +) -> dict: + """Get historical circulating supply data for a stablecoin.""" base_url = f"{DEFILLAMA_STABLECOINS_BASE_URL}/stablecoincharts" - + # If chain is specified, fetch chain-specific data, otherwise fetch all chains endpoint = f"/{chain}" if chain else "/all" url = f"{base_url}{endpoint}?stablecoin={stablecoin_id}" - + async with httpx.AsyncClient() as client: response = await client.get(url) if response.status_code != 200: return {"error": f"API returned status code {response.status_code}"} return response.json() + async def fetch_stablecoin_chains() -> dict: - """Get stablecoin distribution data across all chains. """ + """Get stablecoin distribution data across all chains.""" url = f"{DEFILLAMA_STABLECOINS_BASE_URL}/stablecoinchains" - + async with httpx.AsyncClient() as client: response = await client.get(url) if response.status_code != 200: return {"error": f"API returned status code {response.status_code}"} return response.json() + async def fetch_stablecoin_prices() -> dict: """Get current stablecoin price data. - + Returns: Dictionary containing stablecoin prices with their dates """ url = f"{DEFILLAMA_STABLECOINS_BASE_URL}/stablecoinprices" - + async with httpx.AsyncClient() as client: response = await client.get(url) if response.status_code != 200: return {"error": f"API returned status code {response.status_code}"} return response.json() + # Yields API Functions async def fetch_pools() -> dict: - """Get comprehensive data for all yield-generating pools. """ + """Get comprehensive data for all yield-generating pools.""" url = f"{DEFILLAMA_YIELDS_BASE_URL}/pools" - + async with httpx.AsyncClient() as client: response = await client.get(url) if response.status_code != 200: return {"error": f"API returned status code {response.status_code}"} return response.json() + async def fetch_pool_chart(pool_id: str) -> dict: - """Get historical chart data for a specific pool. """ + """Get historical chart data for a specific pool.""" url = f"{DEFILLAMA_YIELDS_BASE_URL}/chart/{pool_id}" - + async with httpx.AsyncClient() as client: response = await client.get(url) if response.status_code != 200: return {"error": f"API returned status code {response.status_code}"} return response.json() + # Volumes API Functions async def fetch_dex_overview() -> dict: - """Get overview data for DEX protocols. """ + """Get overview data for DEX protocols.""" url = f"{DEFILLAMA_VOLUMES_BASE_URL}/overview/dexs" params = { "excludeTotalDataChart": "true", "excludeTotalDataChartBreakdown": "true", - "dataType": "dailyVolume" + "dataType": "dailyVolume", } - + async with httpx.AsyncClient() as client: response = await client.get(url, params=params) if response.status_code != 200: return {"error": f"API returned status code {response.status_code}"} return response.json() + async def fetch_dex_summary(protocol: str) -> dict: - """Get summary data for a specific DEX protocol. """ + """Get summary data for a specific DEX protocol.""" url = f"{DEFILLAMA_VOLUMES_BASE_URL}/summary/dexs/{protocol}" params = { "excludeTotalDataChart": "true", "excludeTotalDataChartBreakdown": "true", - "dataType": "dailyVolume" + "dataType": "dailyVolume", } - + async with httpx.AsyncClient() as client: response = await client.get(url, params=params) if response.status_code != 200: return {"error": f"API returned status code {response.status_code}"} return response.json() + async def fetch_options_overview() -> dict: - """Get overview data for options protocols from DeFi Llama. """ + """Get overview data for options protocols from DeFi Llama.""" url = f"{DEFILLAMA_VOLUMES_BASE_URL}/overview/options" params = { "excludeTotalDataChart": "true", "excludeTotalDataChartBreakdown": "true", - "dataType": "dailyPremiumVolume" + "dataType": "dailyPremiumVolume", } - + async with httpx.AsyncClient() as client: response = await client.get(url, params=params) if response.status_code != 200: return {"error": f"API returned status code {response.status_code}"} return response.json() + # Fees and Revenue API Functions async def fetch_fees_overview() -> dict: """Get overview data for fees from DeFi Llama. @@ -296,9 +305,9 @@ async def fetch_fees_overview() -> dict: params = { "excludeTotalDataChart": "true", "excludeTotalDataChartBreakdown": "true", - "dataType": "dailyFees" + "dataType": "dailyFees", } - + async with httpx.AsyncClient() as client: response = await client.get(url, params=params) if response.status_code != 200: diff --git a/skills/defillama/base.py b/skills/defillama/base.py index 1682f20..d1de059 100644 --- a/skills/defillama/base.py +++ b/skills/defillama/base.py @@ -2,19 +2,18 @@ from datetime import datetime, timedelta, timezone from typing import Type + from pydantic import BaseModel, Field from abstracts.agent import AgentStoreABC from abstracts.skill import IntentKitSkill, SkillStoreABC from skills.defillama.config.chains import ( get_chain_from_alias, - is_valid_chain, - get_all_chains, - get_chain_aliases, ) DEFILLAMA_BASE_URL = "https://api.llama.fi" + class DefiLlamaBaseTool(IntentKitSkill): """Base class for DeFi Llama tools. @@ -29,9 +28,15 @@ class DefiLlamaBaseTool(IntentKitSkill): description: str = Field(description="A description of what the tool does") args_schema: Type[BaseModel] agent_id: str = Field(description="The ID of the agent") - agent_store: AgentStoreABC = Field(description="The agent store for persisting data") - skill_store: SkillStoreABC = Field(description="The skill store for persisting data") - base_url: str = Field(default=DEFILLAMA_BASE_URL, description="Base URL for DeFi Llama API") + agent_store: AgentStoreABC = Field( + description="The agent store for persisting data" + ) + skill_store: SkillStoreABC = Field( + description="The skill store for persisting data" + ) + base_url: str = Field( + default=DEFILLAMA_BASE_URL, description="Base URL for DeFi Llama API" + ) async def check_rate_limit( self, max_requests: int = 30, interval: int = 1 @@ -58,7 +63,7 @@ async def check_rate_limit( ): if rate_limit["count"] >= max_requests: return True, "Rate limit exceeded" - + rate_limit["count"] += 1 await self.skill_store.save_agent_skill_data( self.agent_id, self.name, "rate_limit", rate_limit @@ -117,7 +122,7 @@ def format_error_response(self, status_code: int, message: str) -> dict: "error": True, "status_code": status_code, "message": message, - "timestamp": datetime.now(tz=timezone.utc).isoformat() + "timestamp": datetime.now(tz=timezone.utc).isoformat(), } def get_current_timestamp(self) -> int: diff --git a/skills/defillama/coins/fetch_batch_historical_prices.py b/skills/defillama/coins/fetch_batch_historical_prices.py index eb964bc..3adf469 100644 --- a/skills/defillama/coins/fetch_batch_historical_prices.py +++ b/skills/defillama/coins/fetch_batch_historical_prices.py @@ -1,10 +1,11 @@ """Tool for fetching batch historical token prices via DeFi Llama API.""" from typing import Dict, List, Optional, Type + from pydantic import BaseModel, Field -from skills.defillama.base import DefiLlamaBaseTool from skills.defillama.api import fetch_batch_historical_prices +from skills.defillama.base import DefiLlamaBaseTool FETCH_BATCH_HISTORICAL_PRICES_PROMPT = """ This tool fetches historical token prices from DeFi Llama for multiple tokens at multiple timestamps. @@ -23,30 +24,17 @@ class HistoricalPricePoint(BaseModel): """Model representing a single historical price point.""" - timestamp: int = Field( - ..., - description="Unix timestamp of the price data" - ) - price: float = Field( - ..., - description="Token price in USD at the timestamp" - ) - confidence: float = Field( - ..., - description="Confidence score for the price data" - ) + timestamp: int = Field(..., description="Unix timestamp of the price data") + price: float = Field(..., description="Token price in USD at the timestamp") + confidence: float = Field(..., description="Confidence score for the price data") class TokenPriceHistory(BaseModel): """Model representing historical price data for a single token.""" - symbol: str = Field( - ..., - description="Token symbol" - ) + symbol: str = Field(..., description="Token symbol") prices: List[HistoricalPricePoint] = Field( - ..., - description="List of historical price points" + ..., description="List of historical price points" ) @@ -54,8 +42,7 @@ class FetchBatchHistoricalPricesInput(BaseModel): """Input schema for fetching batch historical token prices.""" coins_timestamps: Dict[str, List[int]] = Field( - ..., - description="Dictionary mapping token identifiers to lists of timestamps" + ..., description="Dictionary mapping token identifiers to lists of timestamps" ) @@ -64,17 +51,14 @@ class FetchBatchHistoricalPricesResponse(BaseModel): coins: Dict[str, TokenPriceHistory] = Field( default_factory=dict, - description="Historical token prices keyed by token identifier" - ) - error: Optional[str] = Field( - None, - description="Error message if any" + description="Historical token prices keyed by token identifier", ) + error: Optional[str] = Field(None, description="Error message if any") class DefiLlamaFetchBatchHistoricalPrices(DefiLlamaBaseTool): """Tool for fetching batch historical token prices from DeFi Llama. - + This tool retrieves historical prices for multiple tokens at multiple timestamps, using a 4-hour search window around each requested time. @@ -123,7 +107,7 @@ async def _arun( result = await fetch_batch_historical_prices( coins_timestamps=coins_timestamps ) - + # Check for API errors if isinstance(result, dict) and "error" in result: return FetchBatchHistoricalPricesResponse(error=result["error"]) diff --git a/skills/defillama/coins/fetch_block.py b/skills/defillama/coins/fetch_block.py index bd01107..517b10c 100644 --- a/skills/defillama/coins/fetch_block.py +++ b/skills/defillama/coins/fetch_block.py @@ -1,10 +1,11 @@ """Tool for fetching current block data via DeFi Llama API.""" from typing import Optional, Type + from pydantic import BaseModel, Field -from skills.defillama.base import DefiLlamaBaseTool from skills.defillama.api import fetch_block +from skills.defillama.base import DefiLlamaBaseTool FETCH_BLOCK_PROMPT = """ This tool fetches current block data from DeFi Llama for a specific chain. @@ -19,49 +20,28 @@ class BlockData(BaseModel): """Model representing block data.""" - height: int = Field( - ..., - description="Block height number" - ) - timestamp: int = Field( - ..., - description="Unix timestamp of the block" - ) + height: int = Field(..., description="Block height number") + timestamp: int = Field(..., description="Unix timestamp of the block") class FetchBlockInput(BaseModel): """Input schema for fetching block data.""" - chain: str = Field( - ..., - description="Chain name to fetch block data for" - ) + chain: str = Field(..., description="Chain name to fetch block data for") class FetchBlockResponse(BaseModel): """Response schema for block data.""" - chain: str = Field( - ..., - description="Normalized chain name" - ) - height: Optional[int] = Field( - None, - description="Block height number" - ) - timestamp: Optional[int] = Field( - None, - description="Unix timestamp of the block" - ) - error: Optional[str] = Field( - None, - description="Error message if any" - ) + chain: str = Field(..., description="Normalized chain name") + height: Optional[int] = Field(None, description="Block height number") + timestamp: Optional[int] = Field(None, description="Unix timestamp of the block") + error: Optional[str] = Field(None, description="Error message if any") class DefiLlamaFetchBlock(DefiLlamaBaseTool): """Tool for fetching current block data from DeFi Llama. - + This tool retrieves current block data for a specific chain. Example: @@ -77,15 +57,11 @@ class DefiLlamaFetchBlock(DefiLlamaBaseTool): description: str = FETCH_BLOCK_PROMPT args_schema: Type[BaseModel] = FetchBlockInput - def _run( - self, chain: str - ) -> FetchBlockResponse: + def _run(self, chain: str) -> FetchBlockResponse: """Synchronous implementation - not supported.""" raise NotImplementedError("Use _arun instead") - async def _arun( - self, chain: str - ) -> FetchBlockResponse: + async def _arun(self, chain: str) -> FetchBlockResponse: """Fetch current block data for the given chain. Args: @@ -98,37 +74,26 @@ async def _arun( # Validate chain parameter is_valid, normalized_chain = await self.validate_chain(chain) if not is_valid or normalized_chain is None: - return FetchBlockResponse( - chain=chain, - error=f"Invalid chain: {chain}" - ) - + return FetchBlockResponse(chain=chain, error=f"Invalid chain: {chain}") + # Check rate limiting is_rate_limited, error_msg = await self.check_rate_limit() if is_rate_limited: - return FetchBlockResponse( - chain=normalized_chain, - error=error_msg - ) + return FetchBlockResponse(chain=normalized_chain, error=error_msg) # Fetch block data from API result = await fetch_block(chain=normalized_chain) - + # Check for API errors if isinstance(result, dict) and "error" in result: - return FetchBlockResponse( - chain=normalized_chain, - error=result["error"] - ) + return FetchBlockResponse(chain=normalized_chain, error=result["error"]) # Return the response matching the API structure return FetchBlockResponse( chain=normalized_chain, height=result["height"], - timestamp=result["timestamp"] + timestamp=result["timestamp"], ) except Exception as e: - return FetchBlockResponse( - chain=chain, - error=str(e)) + return FetchBlockResponse(chain=chain, error=str(e)) diff --git a/skills/defillama/coins/fetch_current_prices.py b/skills/defillama/coins/fetch_current_prices.py index a31f305..9900644 100644 --- a/skills/defillama/coins/fetch_current_prices.py +++ b/skills/defillama/coins/fetch_current_prices.py @@ -1,10 +1,11 @@ """Tool for fetching token prices via DeFi Llama API.""" from typing import Dict, List, Optional, Type + from pydantic import BaseModel, Field -from skills.defillama.base import DefiLlamaBaseTool from skills.defillama.api import fetch_current_prices +from skills.defillama.base import DefiLlamaBaseTool FETCH_PRICES_PROMPT = """ This tool fetches current token prices from DeFi Llama with a 4-hour search window. @@ -24,34 +25,19 @@ class TokenPrice(BaseModel): """Model representing token price data.""" - price: float = Field( - ..., - description="Current token price in USD" - ) - symbol: str = Field( - ..., - description="Token symbol" - ) - timestamp: int = Field( - ..., - description="Unix timestamp of last price update" - ) - confidence: float = Field( - ..., - description="Confidence score for the price data" - ) - decimals: Optional[int] = Field( - None, - description="Token decimals, if available" - ) + price: float = Field(..., description="Current token price in USD") + symbol: str = Field(..., description="Token symbol") + timestamp: int = Field(..., description="Unix timestamp of last price update") + confidence: float = Field(..., description="Confidence score for the price data") + decimals: Optional[int] = Field(None, description="Token decimals, if available") class FetchCurrentPricesInput(BaseModel): """Input schema for fetching current token prices with a 4-hour search window.""" coins: List[str] = Field( - ..., - description="List of token identifiers (e.g. 'ethereum:0x...', 'coingecko:ethereum')" + ..., + description="List of token identifiers (e.g. 'ethereum:0x...', 'coingecko:ethereum')", ) @@ -59,18 +45,14 @@ class FetchCurrentPricesResponse(BaseModel): """Response schema for current token prices.""" coins: Dict[str, TokenPrice] = Field( - default_factory=dict, - description="Token prices keyed by token identifier" - ) - error: Optional[str] = Field( - None, - description="Error message if any" + default_factory=dict, description="Token prices keyed by token identifier" ) + error: Optional[str] = Field(None, description="Error message if any") class DefiLlamaFetchCurrentPrices(DefiLlamaBaseTool): """Tool for fetching current token prices from DeFi Llama. - + This tool retrieves current prices for multiple tokens in a single request, using a 4-hour search window to ensure fresh data. @@ -110,7 +92,7 @@ async def _arun(self, coins: List[str]) -> FetchCurrentPricesResponse: # Fetch prices from API result = await fetch_current_prices(coins=coins) - + # Check for API errors if isinstance(result, dict) and "error" in result: return FetchCurrentPricesResponse(error=result["error"]) diff --git a/skills/defillama/coins/fetch_first_price.py b/skills/defillama/coins/fetch_first_price.py index 79d5868..5376f3b 100644 --- a/skills/defillama/coins/fetch_first_price.py +++ b/skills/defillama/coins/fetch_first_price.py @@ -1,10 +1,11 @@ """Tool for fetching first recorded token prices via DeFi Llama API.""" from typing import Dict, List, Optional, Type + from pydantic import BaseModel, Field -from skills.defillama.base import DefiLlamaBaseTool from skills.defillama.api import fetch_first_price +from skills.defillama.base import DefiLlamaBaseTool FETCH_FIRST_PRICE_PROMPT = """ This tool fetches the first recorded price data from DeFi Llama for multiple tokens. @@ -22,26 +23,16 @@ class FirstPriceData(BaseModel): """Model representing first price data for a single token.""" - symbol: str = Field( - ..., - description="Token symbol" - ) - price: float = Field( - ..., - description="First recorded price in USD" - ) - timestamp: int = Field( - ..., - description="Unix timestamp of first recorded price" - ) + symbol: str = Field(..., description="Token symbol") + price: float = Field(..., description="First recorded price in USD") + timestamp: int = Field(..., description="Unix timestamp of first recorded price") class FetchFirstPriceInput(BaseModel): """Input schema for fetching first token prices.""" coins: List[str] = Field( - ..., - description="List of token identifiers to fetch first prices for" + ..., description="List of token identifiers to fetch first prices for" ) @@ -49,18 +40,14 @@ class FetchFirstPriceResponse(BaseModel): """Response schema for first token prices.""" coins: Dict[str, FirstPriceData] = Field( - default_factory=dict, - description="First price data keyed by token identifier" - ) - error: Optional[str] = Field( - None, - description="Error message if any" + default_factory=dict, description="First price data keyed by token identifier" ) + error: Optional[str] = Field(None, description="Error message if any") class DefiLlamaFetchFirstPrice(DefiLlamaBaseTool): """Tool for fetching first recorded token prices from DeFi Llama. - + This tool retrieves the first price data recorded for multiple tokens, including the initial price, symbol, and timestamp. @@ -79,15 +66,11 @@ class DefiLlamaFetchFirstPrice(DefiLlamaBaseTool): description: str = FETCH_FIRST_PRICE_PROMPT args_schema: Type[BaseModel] = FetchFirstPriceInput - def _run( - self, coins: List[str] - ) -> FetchFirstPriceResponse: + def _run(self, coins: List[str]) -> FetchFirstPriceResponse: """Synchronous implementation - not supported.""" raise NotImplementedError("Use _arun instead") - async def _arun( - self, coins: List[str] - ) -> FetchFirstPriceResponse: + async def _arun(self, coins: List[str]) -> FetchFirstPriceResponse: """Fetch first recorded prices for the given tokens. Args: @@ -104,7 +87,7 @@ async def _arun( # Fetch first price data from API result = await fetch_first_price(coins=coins) - + # Check for API errors if isinstance(result, dict) and "error" in result: return FetchFirstPriceResponse(error=result["error"]) diff --git a/skills/defillama/coins/fetch_historical_prices.py b/skills/defillama/coins/fetch_historical_prices.py index 4b2d503..2a071ec 100644 --- a/skills/defillama/coins/fetch_historical_prices.py +++ b/skills/defillama/coins/fetch_historical_prices.py @@ -1,10 +1,11 @@ """Tool for fetching historical token prices via DeFi Llama API.""" from typing import Dict, List, Optional, Type + from pydantic import BaseModel, Field -from skills.defillama.base import DefiLlamaBaseTool from skills.defillama.api import fetch_historical_prices +from skills.defillama.base import DefiLlamaBaseTool FETCH_HISTORICAL_PRICES_PROMPT = """ This tool fetches historical token prices from DeFi Llama for a specific timestamp. @@ -24,34 +25,21 @@ class HistoricalTokenPrice(BaseModel): """Model representing historical token price data.""" - price: float = Field( - ..., - description="Token price in USD at the specified time" - ) - symbol: Optional[str] = Field( - None, - description="Token symbol" - ) - timestamp: int = Field( - ..., - description="Unix timestamp of the price data" - ) - decimals: Optional[int] = Field( - None, - description="Token decimals, if available" - ) + price: float = Field(..., description="Token price in USD at the specified time") + symbol: Optional[str] = Field(None, description="Token symbol") + timestamp: int = Field(..., description="Unix timestamp of the price data") + decimals: Optional[int] = Field(None, description="Token decimals, if available") class FetchHistoricalPricesInput(BaseModel): """Input schema for fetching historical token prices.""" timestamp: int = Field( - ..., - description="Unix timestamp for historical price lookup" + ..., description="Unix timestamp for historical price lookup" ) coins: List[str] = Field( - ..., - description="List of token identifiers (e.g. 'ethereum:0x...', 'coingecko:ethereum')" + ..., + description="List of token identifiers (e.g. 'ethereum:0x...', 'coingecko:ethereum')", ) @@ -60,17 +48,14 @@ class FetchHistoricalPricesResponse(BaseModel): coins: Dict[str, HistoricalTokenPrice] = Field( default_factory=dict, - description="Historical token prices keyed by token identifier" - ) - error: Optional[str] = Field( - None, - description="Error message if any" + description="Historical token prices keyed by token identifier", ) + error: Optional[str] = Field(None, description="Error message if any") class DefiLlamaFetchHistoricalPrices(DefiLlamaBaseTool): """Tool for fetching historical token prices from DeFi Llama. - + This tool retrieves historical prices for multiple tokens at a specific timestamp, using a 4-hour search window around the requested time. @@ -90,9 +75,7 @@ class DefiLlamaFetchHistoricalPrices(DefiLlamaBaseTool): description: str = FETCH_HISTORICAL_PRICES_PROMPT args_schema: Type[BaseModel] = FetchHistoricalPricesInput - def _run( - self, timestamp: int, coins: List[str] - ) -> FetchHistoricalPricesResponse: + def _run(self, timestamp: int, coins: List[str]) -> FetchHistoricalPricesResponse: """Synchronous implementation - not supported.""" raise NotImplementedError("Use _arun instead") @@ -115,11 +98,8 @@ async def _arun( return FetchHistoricalPricesResponse(error=error_msg) # Fetch historical prices from API - result = await fetch_historical_prices( - timestamp=timestamp, - coins=coins - ) - + result = await fetch_historical_prices(timestamp=timestamp, coins=coins) + # Check for API errors if isinstance(result, dict) and "error" in result: return FetchHistoricalPricesResponse(error=result["error"]) diff --git a/skills/defillama/coins/fetch_price_chart.py b/skills/defillama/coins/fetch_price_chart.py index 542210b..ba51543 100644 --- a/skills/defillama/coins/fetch_price_chart.py +++ b/skills/defillama/coins/fetch_price_chart.py @@ -1,10 +1,11 @@ """Tool for fetching token price charts via DeFi Llama API.""" from typing import Dict, List, Optional, Type + from pydantic import BaseModel, Field -from skills.defillama.base import DefiLlamaBaseTool from skills.defillama.api import fetch_price_chart +from skills.defillama.base import DefiLlamaBaseTool FETCH_PRICE_CHART_PROMPT = """ This tool fetches price chart data from DeFi Llama for multiple tokens. @@ -23,43 +24,24 @@ class PricePoint(BaseModel): """Model representing a single price point in the chart.""" - timestamp: int = Field( - ..., - description="Unix timestamp of the price data" - ) - price: float = Field( - ..., - description="Token price in USD at the timestamp" - ) + timestamp: int = Field(..., description="Unix timestamp of the price data") + price: float = Field(..., description="Token price in USD at the timestamp") class TokenPriceChart(BaseModel): """Model representing price chart data for a single token.""" - symbol: str = Field( - ..., - description="Token symbol" - ) - confidence: float = Field( - ..., - description="Confidence score for the price data" - ) - decimals: Optional[int] = Field( - None, - description="Token decimals" - ) - prices: List[PricePoint] = Field( - ..., - description="List of historical price points" - ) + symbol: str = Field(..., description="Token symbol") + confidence: float = Field(..., description="Confidence score for the price data") + decimals: Optional[int] = Field(None, description="Token decimals") + prices: List[PricePoint] = Field(..., description="List of historical price points") class FetchPriceChartInput(BaseModel): """Input schema for fetching token price charts.""" coins: List[str] = Field( - ..., - description="List of token identifiers to fetch price charts for" + ..., description="List of token identifiers to fetch price charts for" ) @@ -67,18 +49,14 @@ class FetchPriceChartResponse(BaseModel): """Response schema for token price charts.""" coins: Dict[str, TokenPriceChart] = Field( - default_factory=dict, - description="Price chart data keyed by token identifier" - ) - error: Optional[str] = Field( - None, - description="Error message if any" + default_factory=dict, description="Price chart data keyed by token identifier" ) + error: Optional[str] = Field(None, description="Error message if any") class DefiLlamaFetchPriceChart(DefiLlamaBaseTool): """Tool for fetching token price charts from DeFi Llama. - + This tool retrieves price chart data for multiple tokens over the last 24 hours, including historical price points and token metadata. @@ -97,15 +75,11 @@ class DefiLlamaFetchPriceChart(DefiLlamaBaseTool): description: str = FETCH_PRICE_CHART_PROMPT args_schema: Type[BaseModel] = FetchPriceChartInput - def _run( - self, coins: List[str] - ) -> FetchPriceChartResponse: + def _run(self, coins: List[str]) -> FetchPriceChartResponse: """Synchronous implementation - not supported.""" raise NotImplementedError("Use _arun instead") - async def _arun( - self, coins: List[str] - ) -> FetchPriceChartResponse: + async def _arun(self, coins: List[str]) -> FetchPriceChartResponse: """Fetch price charts for the given tokens. Args: @@ -122,7 +96,7 @@ async def _arun( # Fetch price chart data from API result = await fetch_price_chart(coins=coins) - + # Check for API errors if isinstance(result, dict) and "error" in result: return FetchPriceChartResponse(error=result["error"]) diff --git a/skills/defillama/coins/fetch_price_percentage.py b/skills/defillama/coins/fetch_price_percentage.py index df5667f..2ee5f90 100644 --- a/skills/defillama/coins/fetch_price_percentage.py +++ b/skills/defillama/coins/fetch_price_percentage.py @@ -1,10 +1,11 @@ """Tool for fetching token price percentage changes via DeFi Llama API.""" from typing import Dict, List, Optional, Type + from pydantic import BaseModel, Field -from skills.defillama.base import DefiLlamaBaseTool from skills.defillama.api import fetch_price_percentage +from skills.defillama.base import DefiLlamaBaseTool FETCH_PRICE_PERCENTAGE_PROMPT = """ This tool fetches 24-hour price percentage changes from DeFi Llama for multiple tokens. @@ -23,8 +24,7 @@ class FetchPricePercentageInput(BaseModel): """Input schema for fetching token price percentage changes.""" coins: List[str] = Field( - ..., - description="List of token identifiers to fetch price changes for" + ..., description="List of token identifiers to fetch price changes for" ) @@ -33,17 +33,14 @@ class FetchPricePercentageResponse(BaseModel): coins: Dict[str, float] = Field( default_factory=dict, - description="Price percentage changes keyed by token identifier" - ) - error: Optional[str] = Field( - None, - description="Error message if any" + description="Price percentage changes keyed by token identifier", ) + error: Optional[str] = Field(None, description="Error message if any") class DefiLlamaFetchPricePercentage(DefiLlamaBaseTool): """Tool for fetching token price percentage changes from DeFi Llama. - + This tool retrieves 24-hour price percentage changes for multiple tokens, calculated from the current time. @@ -62,15 +59,11 @@ class DefiLlamaFetchPricePercentage(DefiLlamaBaseTool): description: str = FETCH_PRICE_PERCENTAGE_PROMPT args_schema: Type[BaseModel] = FetchPricePercentageInput - def _run( - self, coins: List[str] - ) -> FetchPricePercentageResponse: + def _run(self, coins: List[str]) -> FetchPricePercentageResponse: """Synchronous implementation - not supported.""" raise NotImplementedError("Use _arun instead") - async def _arun( - self, coins: List[str] - ) -> FetchPricePercentageResponse: + async def _arun(self, coins: List[str]) -> FetchPricePercentageResponse: """Fetch price percentage changes for the given tokens. Args: @@ -87,7 +80,7 @@ async def _arun( # Fetch price percentage data from API result = await fetch_price_percentage(coins=coins) - + # Check for API errors if isinstance(result, dict) and "error" in result: return FetchPricePercentageResponse(error=result["error"]) diff --git a/skills/defillama/config/chains.py b/skills/defillama/config/chains.py index 7043f7b..c63b002 100644 --- a/skills/defillama/config/chains.py +++ b/skills/defillama/config/chains.py @@ -372,9 +372,10 @@ "lung": ["lung"], "bone": ["bone"], "lukso": ["lukso"], - "joltify": ["joltify"] + "joltify": ["joltify"], } + def get_chain_from_alias(alias: str) -> str | None: """Get the main chain identifier from an alias. @@ -385,18 +386,19 @@ def get_chain_from_alias(alias: str) -> str | None: The main chain identifier if found, None otherwise """ normalized_alias = alias.lower().strip() - + # Check if it's a main chain name if normalized_alias in VALID_CHAINS: return normalized_alias - + # Check aliases for chain, aliases in VALID_CHAINS.items(): if normalized_alias in [a.lower() for a in aliases]: return chain - + return None + def is_valid_chain(chain: str) -> bool: """Check if a chain identifier is valid. @@ -408,6 +410,7 @@ def is_valid_chain(chain: str) -> bool: """ return get_chain_from_alias(chain) is not None + def get_all_chains() -> list[str]: """Get a list of all valid main chain identifiers. @@ -416,6 +419,7 @@ def get_all_chains() -> list[str]: """ return list(VALID_CHAINS.keys()) + def get_chain_aliases(chain: str) -> list[str]: """Get all aliases for a given chain. diff --git a/skills/defillama/fees/fetch_fees_overview.py b/skills/defillama/fees/fetch_fees_overview.py index 3cd7f74..8bb1662 100644 --- a/skills/defillama/fees/fetch_fees_overview.py +++ b/skills/defillama/fees/fetch_fees_overview.py @@ -1,10 +1,11 @@ """Tool for fetching fees overview data via DeFi Llama API.""" from typing import Dict, List, Optional, Type + from pydantic import BaseModel, Field -from skills.defillama.base import DefiLlamaBaseTool from skills.defillama.api import fetch_fees_overview +from skills.defillama.base import DefiLlamaBaseTool FETCH_FEES_OVERVIEW_PROMPT = """ This tool fetches comprehensive overview data for protocol fees from DeFi Llama. @@ -15,17 +16,27 @@ - Chain breakdowns """ + class ProtocolMethodology(BaseModel): """Model representing protocol methodology data.""" + UserFees: Optional[str] = Field(None, description="Description of user fees") Fees: Optional[str] = Field(None, description="Description of fees") Revenue: Optional[str] = Field(None, description="Description of revenue") - ProtocolRevenue: Optional[str] = Field(None, description="Description of protocol revenue") - HoldersRevenue: Optional[str] = Field(None, description="Description of holders revenue") - SupplySideRevenue: Optional[str] = Field(None, description="Description of supply side revenue") + ProtocolRevenue: Optional[str] = Field( + None, description="Description of protocol revenue" + ) + HoldersRevenue: Optional[str] = Field( + None, description="Description of holders revenue" + ) + SupplySideRevenue: Optional[str] = Field( + None, description="Description of supply side revenue" + ) + class Protocol(BaseModel): """Model representing protocol data.""" + name: str = Field(..., description="Protocol name") displayName: str = Field(..., description="Display name of protocol") category: str = Field(..., description="Protocol category") @@ -40,12 +51,20 @@ class Protocol(BaseModel): change_1d: Optional[float] = Field(None, description="24-hour change percentage") change_7d: Optional[float] = Field(None, description="7-day change percentage") change_1m: Optional[float] = Field(None, description="30-day change percentage") - methodology: Optional[ProtocolMethodology] = Field(None, description="Protocol methodology") - breakdown24h: Optional[Dict[str, Dict[str, float]]] = Field(None, description="24-hour breakdown by chain") - breakdown30d: Optional[Dict[str, Dict[str, float]]] = Field(None, description="30-day breakdown by chain") + methodology: Optional[ProtocolMethodology] = Field( + None, description="Protocol methodology" + ) + breakdown24h: Optional[Dict[str, Dict[str, float]]] = Field( + None, description="24-hour breakdown by chain" + ) + breakdown30d: Optional[Dict[str, Dict[str, float]]] = Field( + None, description="30-day breakdown by chain" + ) + class FetchFeesOverviewResponse(BaseModel): """Response schema for fees overview data.""" + total24h: float = Field(..., description="Total fees in last 24 hours") total7d: float = Field(..., description="Total fees in last 7 days") total30d: float = Field(..., description="Total fees in last 30 days") @@ -57,9 +76,10 @@ class FetchFeesOverviewResponse(BaseModel): protocols: List[Protocol] = Field(..., description="List of protocols") error: Optional[str] = Field(None, description="Error message if any") + class DefiLlamaFetchFeesOverview(DefiLlamaBaseTool): """Tool for fetching fees overview data from DeFi Llama. - + This tool retrieves comprehensive data about protocol fees, including fee metrics, change percentages, and detailed protocol information. @@ -94,7 +114,7 @@ async def _arun(self) -> FetchFeesOverviewResponse: # Fetch fees data from API result = await fetch_fees_overview() - + # Check for API errors if isinstance(result, dict) and "error" in result: return FetchFeesOverviewResponse(error=result["error"]) diff --git a/skills/defillama/stablecoins/fetch_stablecoin_chains.py b/skills/defillama/stablecoins/fetch_stablecoin_chains.py index 589df46..de4fd41 100644 --- a/skills/defillama/stablecoins/fetch_stablecoin_chains.py +++ b/skills/defillama/stablecoins/fetch_stablecoin_chains.py @@ -1,10 +1,11 @@ """Tool for fetching stablecoin chains data via DeFi Llama API.""" -from typing import Dict, List, Optional, Type +from typing import List, Optional + from pydantic import BaseModel, Field -from skills.defillama.base import DefiLlamaBaseTool from skills.defillama.api import fetch_stablecoin_chains +from skills.defillama.base import DefiLlamaBaseTool FETCH_STABLECOIN_CHAINS_PROMPT = """ This tool fetches stablecoin distribution data across all chains from DeFi Llama. @@ -18,85 +19,43 @@ class CirculatingUSD(BaseModel): """Model representing circulating amounts in different pegs.""" - peggedUSD: Optional[float] = Field( - None, - description="Amount pegged to USD" - ) - peggedEUR: Optional[float] = Field( - None, - description="Amount pegged to EUR" - ) - peggedVAR: Optional[float] = Field( - None, - description="Amount in variable pegs" - ) - peggedJPY: Optional[float] = Field( - None, - description="Amount pegged to JPY" - ) - peggedCHF: Optional[float] = Field( - None, - description="Amount pegged to CHF" - ) - peggedCAD: Optional[float] = Field( - None, - description="Amount pegged to CAD" - ) - peggedGBP: Optional[float] = Field( - None, - description="Amount pegged to GBP" - ) - peggedAUD: Optional[float] = Field( - None, - description="Amount pegged to AUD" - ) - peggedCNY: Optional[float] = Field( - None, - description="Amount pegged to CNY" - ) + peggedUSD: Optional[float] = Field(None, description="Amount pegged to USD") + peggedEUR: Optional[float] = Field(None, description="Amount pegged to EUR") + peggedVAR: Optional[float] = Field(None, description="Amount in variable pegs") + peggedJPY: Optional[float] = Field(None, description="Amount pegged to JPY") + peggedCHF: Optional[float] = Field(None, description="Amount pegged to CHF") + peggedCAD: Optional[float] = Field(None, description="Amount pegged to CAD") + peggedGBP: Optional[float] = Field(None, description="Amount pegged to GBP") + peggedAUD: Optional[float] = Field(None, description="Amount pegged to AUD") + peggedCNY: Optional[float] = Field(None, description="Amount pegged to CNY") peggedREAL: Optional[float] = Field( - None, - description="Amount pegged to Brazilian Real" + None, description="Amount pegged to Brazilian Real" ) class ChainData(BaseModel): """Model representing stablecoin data for a single chain.""" - gecko_id: Optional[str] = Field( - None, - description="CoinGecko ID of the chain" - ) + gecko_id: Optional[str] = Field(None, description="CoinGecko ID of the chain") totalCirculatingUSD: CirculatingUSD = Field( - ..., - description="Total circulating amounts in different pegs" - ) - tokenSymbol: Optional[str] = Field( - None, - description="Native token symbol" - ) - name: str = Field( - ..., - description="Chain name" + ..., description="Total circulating amounts in different pegs" ) + tokenSymbol: Optional[str] = Field(None, description="Native token symbol") + name: str = Field(..., description="Chain name") class FetchStablecoinChainsResponse(BaseModel): """Response schema for stablecoin chains data.""" chains: List[ChainData] = Field( - default_factory=list, - description="List of chains with their stablecoin data" - ) - error: Optional[str] = Field( - None, - description="Error message if any" + default_factory=list, description="List of chains with their stablecoin data" ) + error: Optional[str] = Field(None, description="Error message if any") class DefiLlamaFetchStablecoinChains(DefiLlamaBaseTool): """Tool for fetching stablecoin distribution across chains from DeFi Llama. - + This tool retrieves data about how stablecoins are distributed across different blockchain networks, including circulation amounts and token information. @@ -131,7 +90,7 @@ async def _arun(self) -> FetchStablecoinChainsResponse: # Fetch chains data from API result = await fetch_stablecoin_chains() - + # Check for API errors if isinstance(result, dict) and "error" in result: return FetchStablecoinChainsResponse(error=result["error"]) diff --git a/skills/defillama/stablecoins/fetch_stablecoin_charts.py b/skills/defillama/stablecoins/fetch_stablecoin_charts.py index e306617..434cef1 100644 --- a/skills/defillama/stablecoins/fetch_stablecoin_charts.py +++ b/skills/defillama/stablecoins/fetch_stablecoin_charts.py @@ -1,10 +1,11 @@ """Tool for fetching stablecoin charts via DeFi Llama API.""" -from typing import Dict, List, Optional, Type +from typing import List, Optional, Type + from pydantic import BaseModel, Field -from skills.defillama.base import DefiLlamaBaseTool from skills.defillama.api import fetch_stablecoin_charts +from skills.defillama.base import DefiLlamaBaseTool FETCH_STABLECOIN_CHARTS_PROMPT = """ This tool fetches historical circulating supply data from DeFi Llama for a specific stablecoin. @@ -22,26 +23,18 @@ class CirculatingSupply(BaseModel): """Model representing circulating supply amounts.""" - peggedUSD: float = Field( - ..., - description="Amount pegged to USD" - ) + peggedUSD: float = Field(..., description="Amount pegged to USD") class StablecoinDataPoint(BaseModel): """Model representing a single historical data point.""" - date: str = Field( - ..., - description="Unix timestamp of the data point" - ) + date: str = Field(..., description="Unix timestamp of the data point") totalCirculating: CirculatingSupply = Field( - ..., - description="Total circulating supply" + ..., description="Total circulating supply" ) totalCirculatingUSD: CirculatingSupply = Field( - ..., - description="Total circulating supply in USD" + ..., description="Total circulating supply in USD" ) @@ -49,12 +42,10 @@ class FetchStablecoinChartsInput(BaseModel): """Input schema for fetching stablecoin chart data.""" stablecoin_id: str = Field( - ..., - description="ID of the stablecoin to fetch data for" + ..., description="ID of the stablecoin to fetch data for" ) chain: Optional[str] = Field( - None, - description="Optional chain name for chain-specific data" + None, description="Optional chain name for chain-specific data" ) @@ -62,22 +53,17 @@ class FetchStablecoinChartsResponse(BaseModel): """Response schema for stablecoin chart data.""" data: List[StablecoinDataPoint] = Field( - default_factory=list, - description="List of historical data points" + default_factory=list, description="List of historical data points" ) chain: Optional[str] = Field( - None, - description="Chain name if chain-specific data was requested" - ) - error: Optional[str] = Field( - None, - description="Error message if any" + None, description="Chain name if chain-specific data was requested" ) + error: Optional[str] = Field(None, description="Error message if any") class DefiLlamaFetchStablecoinCharts(DefiLlamaBaseTool): """Tool for fetching stablecoin chart data from DeFi Llama. - + This tool retrieves historical circulating supply data for a specific stablecoin, optionally filtered by chain. @@ -97,14 +83,12 @@ class DefiLlamaFetchStablecoinCharts(DefiLlamaBaseTool): description: str = FETCH_STABLECOIN_CHARTS_PROMPT args_schema: Type[BaseModel] = FetchStablecoinChartsInput - def _run( - self, stablecoin_id: str - ) -> FetchStablecoinChartsResponse: + def _run(self, stablecoin_id: str) -> FetchStablecoinChartsResponse: """Synchronous implementation - not supported.""" raise NotImplementedError("Use _arun instead") async def _arun( - self, stablecoin_id: str + self, stablecoin_id: str, chain: Optional[str] = None ) -> FetchStablecoinChartsResponse: """Fetch historical chart data for the given stablecoin. @@ -124,7 +108,7 @@ async def _arun( error=f"Invalid chain: {chain}" ) chain = normalized_chain - + # Check rate limiting is_rate_limited, error_msg = await self.check_rate_limit() if is_rate_limited: @@ -132,19 +116,15 @@ async def _arun( # Fetch chart data from API result = await fetch_stablecoin_charts( - stablecoin_id=stablecoin_id, - chain=chain + stablecoin_id=stablecoin_id, chain=chain ) - + # Check for API errors if isinstance(result, dict) and "error" in result: return FetchStablecoinChartsResponse(error=result["error"]) # Parse response data - return FetchStablecoinChartsResponse( - data=result, - chain=chain - ) + return FetchStablecoinChartsResponse(data=result, chain=chain) except Exception as e: return FetchStablecoinChartsResponse(error=str(e)) diff --git a/skills/defillama/stablecoins/fetch_stablecoin_prices.py b/skills/defillama/stablecoins/fetch_stablecoin_prices.py index f506099..c1b922b 100644 --- a/skills/defillama/stablecoins/fetch_stablecoin_prices.py +++ b/skills/defillama/stablecoins/fetch_stablecoin_prices.py @@ -1,10 +1,11 @@ """Tool for fetching stablecoin prices via DeFi Llama API.""" -from typing import Dict, List, Optional, Type +from typing import Dict, List, Optional + from pydantic import BaseModel, Field -from skills.defillama.base import DefiLlamaBaseTool from skills.defillama.api import fetch_stablecoin_prices +from skills.defillama.base import DefiLlamaBaseTool FETCH_STABLECOIN_PRICES_PROMPT = """ This tool fetches current price data for stablecoins from DeFi Llama. @@ -18,13 +19,9 @@ class PriceDataPoint(BaseModel): """Model representing a price data point.""" - date: str = Field( - ..., - description="Unix timestamp for the price data" - ) + date: str = Field(..., description="Unix timestamp for the price data") prices: Dict[str, float] = Field( - ..., - description="Dictionary of stablecoin prices indexed by identifier" + ..., description="Dictionary of stablecoin prices indexed by identifier" ) @@ -32,18 +29,14 @@ class FetchStablecoinPricesResponse(BaseModel): """Response schema for stablecoin prices data.""" data: List[PriceDataPoint] = Field( - default_factory=list, - description="List of price data points" - ) - error: Optional[str] = Field( - None, - description="Error message if any" + default_factory=list, description="List of price data points" ) + error: Optional[str] = Field(None, description="Error message if any") class DefiLlamaFetchStablecoinPrices(DefiLlamaBaseTool): """Tool for fetching stablecoin prices from DeFi Llama. - + This tool retrieves current price data for stablecoins, including historical price points and their timestamps. @@ -78,16 +71,13 @@ async def _arun(self) -> FetchStablecoinPricesResponse: # Fetch price data from API result = await fetch_stablecoin_prices() - + # Check for API errors if isinstance(result, dict) and "error" in result: return FetchStablecoinPricesResponse(error=result["error"]) # Parse results into models - data_points = [ - PriceDataPoint(**point) - for point in result - ] + data_points = [PriceDataPoint(**point) for point in result] return FetchStablecoinPricesResponse(data=data_points) diff --git a/skills/defillama/stablecoins/fetch_stablecoins.py b/skills/defillama/stablecoins/fetch_stablecoins.py index 364474d..d0dd279 100644 --- a/skills/defillama/stablecoins/fetch_stablecoins.py +++ b/skills/defillama/stablecoins/fetch_stablecoins.py @@ -1,10 +1,11 @@ """Tool for fetching stablecoin data via DeFi Llama API.""" -from typing import Dict, List, Optional, Type +from typing import Dict, List, Optional + from pydantic import BaseModel, Field -from skills.defillama.base import DefiLlamaBaseTool from skills.defillama.api import fetch_stablecoins +from skills.defillama.base import DefiLlamaBaseTool FETCH_STABLECOINS_PROMPT = """ This tool fetches comprehensive stablecoin data from DeFi Llama. @@ -19,111 +20,68 @@ class CirculatingAmount(BaseModel): """Model representing circulating amounts for a specific peg type.""" - - peggedUSD: float = Field( - ..., - description="Amount pegged to USD" - ) + + peggedUSD: float = Field(..., description="Amount pegged to USD") class ChainCirculating(BaseModel): """Model representing circulating amounts on a specific chain.""" - current: CirculatingAmount = Field( - ..., - description="Current circulating amount" - ) + current: CirculatingAmount = Field(..., description="Current circulating amount") circulatingPrevDay: CirculatingAmount = Field( - ..., - description="Circulating amount from previous day" + ..., description="Circulating amount from previous day" ) circulatingPrevWeek: CirculatingAmount = Field( - ..., - description="Circulating amount from previous week" + ..., description="Circulating amount from previous week" ) circulatingPrevMonth: CirculatingAmount = Field( - ..., - description="Circulating amount from previous month" + ..., description="Circulating amount from previous month" ) class Stablecoin(BaseModel): """Model representing a single stablecoin's data.""" - id: str = Field( - ..., - description="Unique identifier" - ) - name: str = Field( - ..., - description="Stablecoin name" - ) - symbol: str = Field( - ..., - description="Token symbol" - ) - gecko_id: Optional[str] = Field( - None, - description="CoinGecko ID if available" - ) - pegType: str = Field( - ..., - description="Type of peg (e.g. peggedUSD)" - ) - priceSource: str = Field( - ..., - description="Source of price data" - ) - pegMechanism: str = Field( - ..., - description="Mechanism maintaining the peg" - ) + id: str = Field(..., description="Unique identifier") + name: str = Field(..., description="Stablecoin name") + symbol: str = Field(..., description="Token symbol") + gecko_id: Optional[str] = Field(None, description="CoinGecko ID if available") + pegType: str = Field(..., description="Type of peg (e.g. peggedUSD)") + priceSource: str = Field(..., description="Source of price data") + pegMechanism: str = Field(..., description="Mechanism maintaining the peg") circulating: CirculatingAmount = Field( - ..., - description="Current total circulating amount" + ..., description="Current total circulating amount" ) circulatingPrevDay: CirculatingAmount = Field( - ..., - description="Total circulating amount from previous day" + ..., description="Total circulating amount from previous day" ) circulatingPrevWeek: CirculatingAmount = Field( - ..., - description="Total circulating amount from previous week" + ..., description="Total circulating amount from previous week" ) circulatingPrevMonth: CirculatingAmount = Field( - ..., - description="Total circulating amount from previous month" + ..., description="Total circulating amount from previous month" ) chainCirculating: Dict[str, ChainCirculating] = Field( - ..., - description="Circulating amounts per chain" + ..., description="Circulating amounts per chain" ) chains: List[str] = Field( - ..., - description="List of chains where the stablecoin is present" - ) - price: float = Field( - ..., - description="Current price in USD" + ..., description="List of chains where the stablecoin is present" ) + price: float = Field(..., description="Current price in USD") class FetchStablecoinsResponse(BaseModel): """Response schema for stablecoin data.""" peggedAssets: List[Stablecoin] = Field( - default_factory=list, - description="List of stablecoins with their data" - ) - error: Optional[str] = Field( - None, - description="Error message if any" + default_factory=list, description="List of stablecoins with their data" ) + error: Optional[str] = Field(None, description="Error message if any") class DefiLlamaFetchStablecoins(DefiLlamaBaseTool): """Tool for fetching stablecoin data from DeFi Llama. - + This tool retrieves comprehensive data about stablecoins, including their circulating supply across different chains, price information, and peg details. @@ -158,7 +116,7 @@ async def _arun(self) -> FetchStablecoinsResponse: # Fetch stablecoin data from API result = await fetch_stablecoins() - + # Check for API errors if isinstance(result, dict) and "error" in result: return FetchStablecoinsResponse(error=result["error"]) diff --git a/skills/defillama/tests/api_integration.test.py b/skills/defillama/tests/api_integration.test.py index 1a098c1..4f4295a 100644 --- a/skills/defillama/tests/api_integration.test.py +++ b/skills/defillama/tests/api_integration.test.py @@ -1,29 +1,33 @@ -import unittest import asyncio -from datetime import datetime, timedelta import logging +import unittest +from datetime import datetime, timedelta from unittest.runner import TextTestResult -from unittest.signals import registerResult -import sys # Import all functions from your API module from skills.defillama.api import ( - fetch_protocols, fetch_protocol, fetch_historical_tvl, - fetch_chain_historical_tvl, fetch_protocol_current_tvl, - fetch_chains, fetch_current_prices, fetch_historical_prices, - fetch_batch_historical_prices, fetch_price_chart, - fetch_price_percentage, fetch_first_price, fetch_block, - fetch_stablecoins, fetch_stablecoin_charts, fetch_stablecoin_chains, - fetch_stablecoin_prices, fetch_pools, fetch_pool_chart, - fetch_dex_overview, fetch_dex_summary, fetch_options_overview, - fetch_fees_overview + fetch_chains, + fetch_current_prices, + fetch_dex_overview, + fetch_dex_summary, + fetch_fees_overview, + fetch_historical_prices, + fetch_historical_tvl, + fetch_price_chart, + fetch_protocol, + fetch_protocols, + fetch_stablecoin_chains, + fetch_stablecoin_prices, + fetch_stablecoins, ) # Configure logging to only show warnings and errors logging.basicConfig(level=logging.WARNING) + class QuietTestResult(TextTestResult): """Custom TestResult class that minimizes output unless there's a failure""" + def startTest(self, test): self._started_at = datetime.now() super().startTest(test) @@ -31,36 +35,39 @@ def startTest(self, test): def addSuccess(self, test): super().addSuccess(test) if self.showAll: - self.stream.write('.') + self.stream.write(".") self.stream.flush() def addError(self, test, err): super().addError(test, err) - self.stream.write('\n') - self.stream.write(self.separator1 + '\n') - self.stream.write(f'ERROR: {self.getDescription(test)}\n') - self.stream.write(self.separator2 + '\n') + self.stream.write("\n") + self.stream.write(self.separator1 + "\n") + self.stream.write(f"ERROR: {self.getDescription(test)}\n") + self.stream.write(self.separator2 + "\n") self.stream.write(self._exc_info_to_string(err, test)) - self.stream.write('\n') + self.stream.write("\n") self.stream.flush() def addFailure(self, test, err): super().addFailure(test, err) - self.stream.write('\n') - self.stream.write(self.separator1 + '\n') - self.stream.write(f'FAIL: {self.getDescription(test)}\n') - self.stream.write(self.separator2 + '\n') + self.stream.write("\n") + self.stream.write(self.separator1 + "\n") + self.stream.write(f"FAIL: {self.getDescription(test)}\n") + self.stream.write(self.separator2 + "\n") self.stream.write(self._exc_info_to_string(err, test)) - self.stream.write('\n') + self.stream.write("\n") self.stream.flush() + class QuietTestRunner(unittest.TextTestRunner): """Custom TestRunner that uses QuietTestResult""" + resultclass = QuietTestResult + class TestDefiLlamaAPI(unittest.TestCase): """Integration tests for DeFi Llama API client""" - + def setUp(self): """Set up the async event loop""" self.loop = asyncio.new_event_loop() @@ -95,21 +102,21 @@ def test_tvl_endpoints(self): self.assertIsInstance(protocols, list) if len(protocols) > 0: self.assertIn("tvl", protocols[0]) - + # Test fetch_protocol using Aave as an example protocol_data = self.run_async(fetch_protocol("aave")) self.assert_successful_response(protocol_data) self.assertIsInstance(protocol_data, dict) - + # Test fetch_historical_tvl historical_tvl = self.run_async(fetch_historical_tvl()) self.assert_successful_response(historical_tvl) self.assertIsInstance(historical_tvl, list) # Verify the structure of historical TVL data points if len(historical_tvl) > 0: - self.assertIn('date', historical_tvl[0]) - self.assertIn('tvl', historical_tvl[0]) - + self.assertIn("date", historical_tvl[0]) + self.assertIn("tvl", historical_tvl[0]) + # Test fetch_chains chains = self.run_async(fetch_chains()) self.assert_successful_response(chains) @@ -118,28 +125,30 @@ def test_tvl_endpoints(self): def test_coins_endpoints(self): """Test coin price-related endpoints""" test_coins = ["ethereum:0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2"] - + # Test fetch_current_prices current_prices = self.run_async(fetch_current_prices(test_coins)) self.assert_successful_response(current_prices) self.assertIsInstance(current_prices, dict) - + # Test fetch_historical_prices timestamp = int((datetime.now() - timedelta(days=1)).timestamp()) - historical_prices = self.run_async(fetch_historical_prices(timestamp, test_coins)) + historical_prices = self.run_async( + fetch_historical_prices(timestamp, test_coins) + ) self.assert_successful_response(historical_prices) self.assertIsInstance(historical_prices, dict) - + # Test fetch_price_chart price_chart = self.run_async(fetch_price_chart(test_coins)) self.assert_successful_response(price_chart) self.assertIsInstance(price_chart, dict) - self.assertIn('coins', price_chart) + self.assertIn("coins", price_chart) # Verify the structure of the response - coin_data = price_chart['coins'].get(test_coins[0]) + coin_data = price_chart["coins"].get(test_coins[0]) self.assertIsNotNone(coin_data) - self.assertIn('prices', coin_data) - self.assertIsInstance(coin_data['prices'], list) + self.assertIn("prices", coin_data) + self.assertIsInstance(coin_data["prices"], list) def test_stablecoin_endpoints(self): """Test stablecoin-related endpoints""" @@ -147,12 +156,12 @@ def test_stablecoin_endpoints(self): stablecoins = self.run_async(fetch_stablecoins()) self.assert_successful_response(stablecoins) self.assertIsInstance(stablecoins, dict) - + # Test fetch_stablecoin_chains chains = self.run_async(fetch_stablecoin_chains()) self.assert_successful_response(chains) self.assertIsInstance(chains, list) - + # Test fetch_stablecoin_prices prices = self.run_async(fetch_stablecoin_prices()) self.assert_successful_response(prices) @@ -164,7 +173,7 @@ def test_volume_endpoints(self): dex_overview = self.run_async(fetch_dex_overview()) self.assert_successful_response(dex_overview) self.assertIsInstance(dex_overview, dict) - + # Test fetch_dex_summary using Uniswap as example dex_summary = self.run_async(fetch_dex_summary("uniswap")) self.assert_successful_response(dex_summary) @@ -176,6 +185,7 @@ def test_fees_endpoint(self): self.assert_successful_response(fees_overview) self.assertIsInstance(fees_overview, dict) + if __name__ == "__main__": # Use the quiet test runner runner = QuietTestRunner(verbosity=1) diff --git a/skills/defillama/tests/api_unit.test.py b/skills/defillama/tests/api_unit.test.py index 5faa73c..5684d33 100644 --- a/skills/defillama/tests/api_unit.test.py +++ b/skills/defillama/tests/api_unit.test.py @@ -1,40 +1,41 @@ import unittest -from unittest.mock import patch, AsyncMock -import asyncio +from unittest.mock import AsyncMock, patch # Import the endpoints from your module. # Adjust the import path if your module has a different name or location. from skills.defillama.api import ( - # Original functions - fetch_protocols, - fetch_protocol, - fetch_historical_tvl, + fetch_batch_historical_prices, + fetch_block, fetch_chain_historical_tvl, - fetch_protocol_current_tvl, fetch_chains, fetch_current_prices, + # Volume related functions + fetch_dex_overview, + fetch_dex_summary, + # Fees related functions + fetch_fees_overview, + fetch_first_price, fetch_historical_prices, - fetch_batch_historical_prices, + fetch_historical_tvl, + fetch_options_overview, + fetch_pool_chart, + # Yields related functions + fetch_pools, # Price related functions fetch_price_chart, fetch_price_percentage, - fetch_first_price, - fetch_block, - # Stablecoin related functions - fetch_stablecoins, - fetch_stablecoin_charts, + fetch_protocol, + fetch_protocol_current_tvl, + # Original functions + fetch_protocols, fetch_stablecoin_chains, + fetch_stablecoin_charts, fetch_stablecoin_prices, - # Yields related functions - fetch_pools, - fetch_pool_chart, - # Volume related functions - fetch_dex_overview, - fetch_dex_summary, - fetch_options_overview, - # Fees related functions - fetch_fees_overview, + # Stablecoin related functions + fetch_stablecoins, ) + + # Dummy response to simulate httpx responses. class DummyResponse: def __init__(self, status_code, json_data): @@ -44,8 +45,8 @@ def __init__(self, status_code, json_data): def json(self): return self._json_data -class TestDefiLlamaAPI(unittest.IsolatedAsyncioTestCase): +class TestDefiLlamaAPI(unittest.IsolatedAsyncioTestCase): @classmethod def setUpClass(cls): # Set up a fixed timestamp that all tests will use @@ -53,7 +54,7 @@ def setUpClass(cls): async def asyncSetUp(self): # Start the patcher before each test - self.datetime_patcher = patch('skills.defillama.api.datetime') + self.datetime_patcher = patch("skills.defillama.api.datetime") self.mock_datetime = self.datetime_patcher.start() # Configure the mock to return our fixed timestamp self.mock_datetime.now.return_value.timestamp.return_value = self.mock_timestamp @@ -63,7 +64,9 @@ async def asyncTearDown(self): self.datetime_patcher.stop() # Helper method to patch httpx.AsyncClient and set up the dummy client. - async def _run_with_dummy(self, func, expected_url, dummy_response, *args, expected_kwargs=None): + async def _run_with_dummy( + self, func, expected_url, dummy_response, *args, expected_kwargs=None + ): if expected_kwargs is None: expected_kwargs = {} with patch("httpx.AsyncClient") as MockClient: @@ -101,10 +104,7 @@ async def test_fetch_protocol_success(self): dummy = DummyResponse(200, {"protocol": protocol}) expected_url = f"https://api.llama.fi/protocol/{protocol}" result = await self._run_with_dummy( - fetch_protocol, - expected_url, - dummy, - protocol + fetch_protocol, expected_url, dummy, protocol ) self.assertEqual(result, {"protocol": protocol}) @@ -113,10 +113,7 @@ async def test_fetch_protocol_error(self): dummy = DummyResponse(500, None) expected_url = f"https://api.llama.fi/protocol/{protocol}" result = await self._run_with_dummy( - fetch_protocol, - expected_url, - dummy, - protocol + fetch_protocol, expected_url, dummy, protocol ) self.assertEqual(result, {"error": "API returned status code 500"}) @@ -147,10 +144,7 @@ async def test_fetch_chain_historical_tvl_success(self): dummy = DummyResponse(200, {"chain": chain}) expected_url = f"https://api.llama.fi/v2/historicalChainTvl/{chain}" result = await self._run_with_dummy( - fetch_chain_historical_tvl, - expected_url, - dummy, - chain + fetch_chain_historical_tvl, expected_url, dummy, chain ) self.assertEqual(result, {"chain": chain}) @@ -159,10 +153,7 @@ async def test_fetch_chain_historical_tvl_error(self): dummy = DummyResponse(503, None) expected_url = f"https://api.llama.fi/v2/historicalChainTvl/{chain}" result = await self._run_with_dummy( - fetch_chain_historical_tvl, - expected_url, - dummy, - chain + fetch_chain_historical_tvl, expected_url, dummy, chain ) self.assertEqual(result, {"error": "API returned status code 503"}) @@ -172,10 +163,7 @@ async def test_fetch_protocol_current_tvl_success(self): dummy = DummyResponse(200, {"current_tvl": 12345}) expected_url = f"https://api.llama.fi/tvl/{protocol}" result = await self._run_with_dummy( - fetch_protocol_current_tvl, - expected_url, - dummy, - protocol + fetch_protocol_current_tvl, expected_url, dummy, protocol ) self.assertEqual(result, {"current_tvl": 12345}) @@ -184,10 +172,7 @@ async def test_fetch_protocol_current_tvl_error(self): dummy = DummyResponse(418, None) expected_url = f"https://api.llama.fi/tvl/{protocol}" result = await self._run_with_dummy( - fetch_protocol_current_tvl, - expected_url, - dummy, - protocol + fetch_protocol_current_tvl, expected_url, dummy, protocol ) self.assertEqual(result, {"error": "API returned status code 418"}) @@ -219,10 +204,7 @@ async def test_fetch_current_prices_success(self): dummy = DummyResponse(200, {"prices": "data"}) expected_url = f"https://api.llama.fi/prices/current/{coins_str}?searchWidth=4h" result = await self._run_with_dummy( - fetch_current_prices, - expected_url, - dummy, - coins + fetch_current_prices, expected_url, dummy, coins ) self.assertEqual(result, {"prices": "data"}) @@ -232,10 +214,7 @@ async def test_fetch_current_prices_error(self): dummy = DummyResponse(500, None) expected_url = f"https://api.llama.fi/prices/current/{coins_str}?searchWidth=4h" result = await self._run_with_dummy( - fetch_current_prices, - expected_url, - dummy, - coins + fetch_current_prices, expected_url, dummy, coins ) self.assertEqual(result, {"error": "API returned status code 500"}) @@ -247,11 +226,7 @@ async def test_fetch_historical_prices_success(self): dummy = DummyResponse(200, {"historical_prices": "data"}) expected_url = f"https://api.llama.fi/prices/historical/{timestamp}/{coins_str}?searchWidth=4h" result = await self._run_with_dummy( - fetch_historical_prices, - expected_url, - dummy, - timestamp, - coins + fetch_historical_prices, expected_url, dummy, timestamp, coins ) self.assertEqual(result, {"historical_prices": "data"}) @@ -262,11 +237,7 @@ async def test_fetch_historical_prices_error(self): dummy = DummyResponse(400, None) expected_url = f"https://api.llama.fi/prices/historical/{timestamp}/{coins_str}?searchWidth=4h" result = await self._run_with_dummy( - fetch_historical_prices, - expected_url, - dummy, - timestamp, - coins + fetch_historical_prices, expected_url, dummy, timestamp, coins ) self.assertEqual(result, {"error": "API returned status code 400"}) @@ -282,7 +253,9 @@ async def test_fetch_batch_historical_prices_success(self): client_instance.get.return_value = dummy MockClient.return_value.__aenter__.return_value = client_instance result = await fetch_batch_historical_prices(coins_timestamps) - client_instance.get.assert_called_once_with(expected_url, params=expected_params) + client_instance.get.assert_called_once_with( + expected_url, params=expected_params + ) self.assertEqual(result, {"batch": "data"}) async def test_fetch_batch_historical_prices_error(self): @@ -295,7 +268,9 @@ async def test_fetch_batch_historical_prices_error(self): client_instance.get.return_value = dummy MockClient.return_value.__aenter__.return_value = client_instance result = await fetch_batch_historical_prices(coins_timestamps) - client_instance.get.assert_called_once_with(expected_url, params=expected_params) + client_instance.get.assert_called_once_with( + expected_url, params=expected_params + ) self.assertEqual(result, {"error": "API returned status code 503"}) async def test_fetch_price_chart_success(self): @@ -303,22 +278,24 @@ async def test_fetch_price_chart_success(self): coins_str = ",".join(coins) dummy = DummyResponse(200, {"chart": "data"}) expected_url = f"https://api.llama.fi/chart/{coins_str}" - + # Calculate start time based on mock timestamp start_time = self.mock_timestamp - 86400 # mock timestamp - 1 day expected_params = { "start": start_time, "span": 10, "period": "2d", - "searchWidth": "600" + "searchWidth": "600", } - + with patch("httpx.AsyncClient") as MockClient: client_instance = AsyncMock() client_instance.get.return_value = dummy MockClient.return_value.__aenter__.return_value = client_instance result = await fetch_price_chart(coins) - client_instance.get.assert_called_once_with(expected_url, params=expected_params) + client_instance.get.assert_called_once_with( + expected_url, params=expected_params + ) self.assertEqual(result, {"chart": "data"}) async def test_fetch_price_chart_error(self): @@ -326,47 +303,50 @@ async def test_fetch_price_chart_error(self): coins_str = ",".join(coins) dummy = DummyResponse(500, None) expected_url = f"https://api.llama.fi/chart/{coins_str}" - + # Calculate start time based on mock timestamp start_time = self.mock_timestamp - 86400 # mock timestamp - 1 day expected_params = { "start": start_time, "span": 10, "period": "2d", - "searchWidth": "600" + "searchWidth": "600", } - + with patch("httpx.AsyncClient") as MockClient: client_instance = AsyncMock() client_instance.get.return_value = dummy MockClient.return_value.__aenter__.return_value = client_instance result = await fetch_price_chart(coins) - client_instance.get.assert_called_once_with(expected_url, params=expected_params) + client_instance.get.assert_called_once_with( + expected_url, params=expected_params + ) self.assertEqual(result, {"error": "API returned status code 500"}) - # --- Tests for fetch_price_percentage --- async def test_fetch_price_percentage_success(self): coins = ["bitcoin", "ethereum"] coins_str = ",".join(coins) dummy = DummyResponse(200, {"percentage": "data"}) expected_url = f"https://api.llama.fi/percentage/{coins_str}" - + mock_timestamp = 1677648000 # Fixed timestamp with patch("skills.defillama.api.datetime") as mock_datetime: mock_datetime.now.return_value.timestamp.return_value = mock_timestamp expected_params = { "timestamp": mock_timestamp, "lookForward": "false", - "period": "24h" + "period": "24h", } - + with patch("httpx.AsyncClient") as MockClient: client_instance = AsyncMock() client_instance.get.return_value = dummy MockClient.return_value.__aenter__.return_value = client_instance result = await fetch_price_percentage(coins) - client_instance.get.assert_called_once_with(expected_url, params=expected_params) + client_instance.get.assert_called_once_with( + expected_url, params=expected_params + ) self.assertEqual(result, {"percentage": "data"}) async def test_fetch_price_percentage_error(self): @@ -374,42 +354,45 @@ async def test_fetch_price_percentage_error(self): coins_str = ",".join(coins) dummy = DummyResponse(404, None) expected_url = f"https://api.llama.fi/percentage/{coins_str}" - + expected_params = { "timestamp": self.mock_timestamp, "lookForward": "false", - "period": "24h" + "period": "24h", } - + with patch("httpx.AsyncClient") as MockClient: client_instance = AsyncMock() client_instance.get.return_value = dummy MockClient.return_value.__aenter__.return_value = client_instance result = await fetch_price_percentage(coins) - client_instance.get.assert_called_once_with(expected_url, params=expected_params) + client_instance.get.assert_called_once_with( + expected_url, params=expected_params + ) self.assertEqual(result, {"error": "API returned status code 404"}) - - async def test_fetch_price_percentage_error(self): + async def test_fetch_price_percentage_error2(self): coins = ["bitcoin", "ethereum"] coins_str = ",".join(coins) dummy = DummyResponse(404, None) expected_url = f"https://api.llama.fi/percentage/{coins_str}" - + with patch("datetime.datetime") as mock_datetime: mock_datetime.now.return_value.timestamp.return_value = 1677648000 expected_params = { "timestamp": 1677648000, "lookForward": "false", - "period": "24h" + "period": "24h", } - + with patch("httpx.AsyncClient") as MockClient: client_instance = AsyncMock() client_instance.get.return_value = dummy MockClient.return_value.__aenter__.return_value = client_instance result = await fetch_price_percentage(coins) - client_instance.get.assert_called_once_with(expected_url, params=expected_params) + client_instance.get.assert_called_once_with( + expected_url, params=expected_params + ) self.assertEqual(result, {"error": "API returned status code 404"}) # --- Tests for fetch_first_price --- @@ -419,10 +402,7 @@ async def test_fetch_first_price_success(self): dummy = DummyResponse(200, {"first_prices": "data"}) expected_url = f"https://api.llama.fi/prices/first/{coins_str}" result = await self._run_with_dummy( - fetch_first_price, - expected_url, - dummy, - coins + fetch_first_price, expected_url, dummy, coins ) self.assertEqual(result, {"first_prices": "data"}) @@ -432,10 +412,7 @@ async def test_fetch_first_price_error(self): dummy = DummyResponse(500, None) expected_url = f"https://api.llama.fi/prices/first/{coins_str}" result = await self._run_with_dummy( - fetch_first_price, - expected_url, - dummy, - coins + fetch_first_price, expected_url, dummy, coins ) self.assertEqual(result, {"error": "API returned status code 500"}) @@ -444,32 +421,22 @@ async def test_fetch_block_success(self): chain = "ethereum" dummy = DummyResponse(200, {"block": 123456}) mock_timestamp = 1677648000 # Fixed timestamp - + with patch("skills.defillama.api.datetime") as mock_datetime: mock_datetime.now.return_value.timestamp.return_value = mock_timestamp expected_url = f"https://api.llama.fi/block/{chain}/{mock_timestamp}" - result = await self._run_with_dummy( - fetch_block, - expected_url, - dummy, - chain - ) + result = await self._run_with_dummy(fetch_block, expected_url, dummy, chain) self.assertEqual(result, {"block": 123456}) async def test_fetch_block_error(self): chain = "ethereum" dummy = DummyResponse(404, None) mock_timestamp = 1677648000 # Fixed timestamp - + with patch("skills.defillama.api.datetime") as mock_datetime: mock_datetime.now.return_value.timestamp.return_value = mock_timestamp expected_url = f"https://api.llama.fi/block/{chain}/{mock_timestamp}" - result = await self._run_with_dummy( - fetch_block, - expected_url, - dummy, - chain - ) + result = await self._run_with_dummy(fetch_block, expected_url, dummy, chain) self.assertEqual(result, {"error": "API returned status code 404"}) # --- Tests for Stablecoins API --- @@ -477,26 +444,26 @@ async def test_fetch_stablecoins_success(self): dummy = DummyResponse(200, {"stablecoins": "data"}) expected_url = "https://api.llama.fi/stablecoins" expected_params = {"includePrices": "true"} - + with patch("httpx.AsyncClient") as MockClient: client_instance = AsyncMock() client_instance.get.return_value = dummy MockClient.return_value.__aenter__.return_value = client_instance result = await fetch_stablecoins() - client_instance.get.assert_called_once_with(expected_url, params=expected_params) + client_instance.get.assert_called_once_with( + expected_url, params=expected_params + ) self.assertEqual(result, {"stablecoins": "data"}) async def test_fetch_stablecoin_charts_success(self): stablecoin_id = "USDT" chain = "ethereum" dummy = DummyResponse(200, {"charts": "data"}) - expected_url = f"https://api.llama.fi/stablecoincharts/{chain}?stablecoin={stablecoin_id}" + expected_url = ( + f"https://api.llama.fi/stablecoincharts/{chain}?stablecoin={stablecoin_id}" + ) result = await self._run_with_dummy( - fetch_stablecoin_charts, - expected_url, - dummy, - stablecoin_id, - chain + fetch_stablecoin_charts, expected_url, dummy, stablecoin_id, chain ) self.assertEqual(result, {"charts": "data"}) @@ -504,9 +471,7 @@ async def test_fetch_stablecoin_chains_success(self): dummy = DummyResponse(200, {"chains": "data"}) expected_url = "https://api.llama.fi/stablecoinchains" result = await self._run_with_dummy( - fetch_stablecoin_chains, - expected_url, - dummy + fetch_stablecoin_chains, expected_url, dummy ) self.assertEqual(result, {"chains": "data"}) @@ -514,9 +479,7 @@ async def test_fetch_stablecoin_prices_success(self): dummy = DummyResponse(200, {"prices": "data"}) expected_url = "https://api.llama.fi/stablecoinprices" result = await self._run_with_dummy( - fetch_stablecoin_prices, - expected_url, - dummy + fetch_stablecoin_prices, expected_url, dummy ) self.assertEqual(result, {"prices": "data"}) @@ -524,11 +487,7 @@ async def test_fetch_stablecoin_prices_success(self): async def test_fetch_pools_success(self): dummy = DummyResponse(200, {"pools": "data"}) expected_url = "https://api.llama.fi/pools" - result = await self._run_with_dummy( - fetch_pools, - expected_url, - dummy - ) + result = await self._run_with_dummy(fetch_pools, expected_url, dummy) self.assertEqual(result, {"pools": "data"}) async def test_fetch_pool_chart_success(self): @@ -536,10 +495,7 @@ async def test_fetch_pool_chart_success(self): dummy = DummyResponse(200, {"chart": "data"}) expected_url = f"https://api.llama.fi/chart/{pool_id}" result = await self._run_with_dummy( - fetch_pool_chart, - expected_url, - dummy, - pool_id + fetch_pool_chart, expected_url, dummy, pool_id ) self.assertEqual(result, {"chart": "data"}) @@ -550,15 +506,17 @@ async def test_fetch_dex_overview_success(self): expected_params = { "excludeTotalDataChart": "true", "excludeTotalDataChartBreakdown": "true", - "dataType": "dailyVolume" + "dataType": "dailyVolume", } - + with patch("httpx.AsyncClient") as MockClient: client_instance = AsyncMock() client_instance.get.return_value = dummy MockClient.return_value.__aenter__.return_value = client_instance result = await fetch_dex_overview() - client_instance.get.assert_called_once_with(expected_url, params=expected_params) + client_instance.get.assert_called_once_with( + expected_url, params=expected_params + ) self.assertEqual(result, {"overview": "data"}) async def test_fetch_dex_summary_success(self): @@ -568,15 +526,17 @@ async def test_fetch_dex_summary_success(self): expected_params = { "excludeTotalDataChart": "true", "excludeTotalDataChartBreakdown": "true", - "dataType": "dailyVolume" + "dataType": "dailyVolume", } - + with patch("httpx.AsyncClient") as MockClient: client_instance = AsyncMock() client_instance.get.return_value = dummy MockClient.return_value.__aenter__.return_value = client_instance result = await fetch_dex_summary(protocol) - client_instance.get.assert_called_once_with(expected_url, params=expected_params) + client_instance.get.assert_called_once_with( + expected_url, params=expected_params + ) self.assertEqual(result, {"summary": "data"}) async def test_fetch_options_overview_success(self): @@ -585,15 +545,17 @@ async def test_fetch_options_overview_success(self): expected_params = { "excludeTotalDataChart": "true", "excludeTotalDataChartBreakdown": "true", - "dataType": "dailyPremiumVolume" + "dataType": "dailyPremiumVolume", } - + with patch("httpx.AsyncClient") as MockClient: client_instance = AsyncMock() client_instance.get.return_value = dummy MockClient.return_value.__aenter__.return_value = client_instance result = await fetch_options_overview() - client_instance.get.assert_called_once_with(expected_url, params=expected_params) + client_instance.get.assert_called_once_with( + expected_url, params=expected_params + ) self.assertEqual(result, {"options": "data"}) # --- Tests for Fees API --- @@ -603,17 +565,19 @@ async def test_fetch_fees_overview_success(self): expected_params = { "excludeTotalDataChart": "true", "excludeTotalDataChartBreakdown": "true", - "dataType": "dailyFees" + "dataType": "dailyFees", } - + with patch("httpx.AsyncClient") as MockClient: client_instance = AsyncMock() client_instance.get.return_value = dummy MockClient.return_value.__aenter__.return_value = client_instance result = await fetch_fees_overview() - client_instance.get.assert_called_once_with(expected_url, params=expected_params) + client_instance.get.assert_called_once_with( + expected_url, params=expected_params + ) self.assertEqual(result, {"fees": "data"}) -if __name__ == '__main__': - unittest.main() +if __name__ == "__main__": + unittest.main() diff --git a/skills/defillama/tvl/fetch_chain_historical_tvl.py b/skills/defillama/tvl/fetch_chain_historical_tvl.py index c75fadb..3f41d1b 100644 --- a/skills/defillama/tvl/fetch_chain_historical_tvl.py +++ b/skills/defillama/tvl/fetch_chain_historical_tvl.py @@ -1,10 +1,11 @@ """Tool for fetching chain historical TVL via DeFiLlama API.""" from typing import List, Type + from pydantic import BaseModel, Field -from skills.defillama.base import DefiLlamaBaseTool from skills.defillama.api import fetch_chain_historical_tvl +from skills.defillama.base import DefiLlamaBaseTool FETCH_HISTORICAL_TVL_PROMPT = """ This tool fetches historical Total Value Locked (TVL) data for a specific blockchain. @@ -16,49 +17,35 @@ class HistoricalTVLDataPoint(BaseModel): """Model representing a single TVL data point.""" - date: int = Field( - ..., - description="Unix timestamp of the TVL measurement" - ) - tvl: float = Field( - ..., - description="Total Value Locked in USD at this timestamp" - ) + date: int = Field(..., description="Unix timestamp of the TVL measurement") + tvl: float = Field(..., description="Total Value Locked in USD at this timestamp") class FetchChainHistoricalTVLInput(BaseModel): """Input schema for fetching chain-specific historical TVL data.""" chain: str = Field( - ..., - description="Chain name to fetch TVL for (e.g., 'ethereum', 'solana')" + ..., description="Chain name to fetch TVL for (e.g., 'ethereum', 'solana')" ) class FetchChainHistoricalTVLResponse(BaseModel): """Response schema for chain-specific historical TVL data.""" - chain: str = Field( - ..., - description="Normalized chain name" - ) + chain: str = Field(..., description="Normalized chain name") data: List[HistoricalTVLDataPoint] = Field( - default_factory=list, - description="List of historical TVL data points" - ) - error: str | None = Field( - default=None, - description="Error message if any" + default_factory=list, description="List of historical TVL data points" ) + error: str | None = Field(default=None, description="Error message if any") class DefiLlamaFetchChainHistoricalTvl(DefiLlamaBaseTool): """Tool for fetching historical TVL data for a specific blockchain. - + This tool fetches the complete Total Value Locked (TVL) history for a given blockchain using the DeFiLlama API. It includes rate limiting and chain validation to ensure reliable data retrieval. - + Example: tvl_tool = DefiLlamaFetchChainHistoricalTvl( skill_store=store, @@ -89,42 +76,30 @@ async def _arun(self, chain: str) -> FetchChainHistoricalTVLResponse: # Check rate limiting is_rate_limited, error_msg = await self.check_rate_limit() if is_rate_limited: - return FetchChainHistoricalTVLResponse( - chain=chain, - error=error_msg - ) + return FetchChainHistoricalTVLResponse(chain=chain, error=error_msg) # Validate chain parameter is_valid, normalized_chain = await self.validate_chain(chain) if not is_valid or normalized_chain is None: return FetchChainHistoricalTVLResponse( - chain=chain, - error=f"Invalid chain: {chain}" + chain=chain, error=f"Invalid chain: {chain}" ) # Fetch TVL history from API result = await fetch_chain_historical_tvl(normalized_chain) - + # Check for API errors if isinstance(result, dict) and "error" in result: return FetchChainHistoricalTVLResponse( - chain=normalized_chain, - error=result["error"] + chain=normalized_chain, error=result["error"] ) # Parse response into our schema - data_points = [ - HistoricalTVLDataPoint(**point) - for point in result - ] + data_points = [HistoricalTVLDataPoint(**point) for point in result] return FetchChainHistoricalTVLResponse( - chain=normalized_chain, - data=data_points + chain=normalized_chain, data=data_points ) except Exception as e: - return FetchChainHistoricalTVLResponse( - chain=chain, - error=str(e) - ) + return FetchChainHistoricalTVLResponse(chain=chain, error=str(e)) diff --git a/skills/defillama/tvl/fetch_chains.py b/skills/defillama/tvl/fetch_chains.py index d4c0694..74471f7 100644 --- a/skills/defillama/tvl/fetch_chains.py +++ b/skills/defillama/tvl/fetch_chains.py @@ -1,10 +1,11 @@ """Tool for fetching chain TVL data via DeFi Llama API.""" from typing import List, Optional, Type + from pydantic import BaseModel, Field -from skills.defillama.base import DefiLlamaBaseTool from skills.defillama.api import fetch_chains +from skills.defillama.base import DefiLlamaBaseTool FETCH_CHAINS_PROMPT = """ This tool fetches current Total Value Locked (TVL) data for all blockchains tracked by DeFi Llama. @@ -20,41 +21,27 @@ class ChainTVLData(BaseModel): """Model representing TVL data for a single chain.""" - name: str = Field( - ..., - description="Chain name" - ) - tvl: float = Field( - ..., - description="Total Value Locked in USD" - ) - gecko_id: Optional[str] = Field( - None, - description="CoinGecko identifier" - ) + name: str = Field(..., description="Chain name") + tvl: float = Field(..., description="Total Value Locked in USD") + gecko_id: Optional[str] = Field(None, description="CoinGecko identifier") token_symbol: Optional[str] = Field( - None, - alias="tokenSymbol", - description="Native token symbol" + None, alias="tokenSymbol", description="Native token symbol" ) cmc_id: Optional[str] = Field( - None, - alias="cmcId", - description="CoinMarketCap identifier" + None, alias="cmcId", description="CoinMarketCap identifier" ) chain_id: Optional[int | str] = Field( - None, - alias="chainId", - description="Chain identifier" + None, alias="chainId", description="Chain identifier" ) class FetchChainsInput(BaseModel): """Input schema for fetching all chains' TVL data. - + This endpoint doesn't require any parameters as it returns TVL data for all chains. """ + pass @@ -62,22 +49,15 @@ class FetchChainsResponse(BaseModel): """Response schema for all chains' TVL data.""" chains: List[ChainTVLData] = Field( - default_factory=list, - description="List of chains with their TVL data" - ) - total_tvl: float = Field( - ..., - description="Total TVL across all chains in USD" - ) - error: Optional[str] = Field( - None, - description="Error message if any" + default_factory=list, description="List of chains with their TVL data" ) + total_tvl: float = Field(..., description="Total TVL across all chains in USD") + error: Optional[str] = Field(None, description="Error message if any") class DefiLlamaFetchChains(DefiLlamaBaseTool): """Tool for fetching current TVL data for all blockchains. - + This tool retrieves the current Total Value Locked (TVL) for all chains tracked by DeFi Llama, including chain identifiers and metadata. @@ -108,35 +88,22 @@ async def _arun(self) -> FetchChainsResponse: # Check rate limiting is_rate_limited, error_msg = await self.check_rate_limit() if is_rate_limited: - return FetchChainsResponse( - chains=[], - total_tvl=0, - error=error_msg - ) + return FetchChainsResponse(chains=[], total_tvl=0, error=error_msg) # Fetch chains data from API result = await fetch_chains() - + # Check for API errors if isinstance(result, dict) and "error" in result: return FetchChainsResponse( - chains=[], - total_tvl=0, - error=result["error"] + chains=[], total_tvl=0, error=result["error"] ) # Parse chains data and calculate total TVL chains = [ChainTVLData(**chain_data) for chain_data in result] total_tvl = sum(chain.tvl for chain in chains) - - return FetchChainsResponse( - chains=chains, - total_tvl=total_tvl - ) + + return FetchChainsResponse(chains=chains, total_tvl=total_tvl) except Exception as e: - return FetchChainsResponse( - chains=[], - total_tvl=0, - error=str(e) - ) + return FetchChainsResponse(chains=[], total_tvl=0, error=str(e)) diff --git a/skills/defillama/tvl/fetch_historical_tvl.py b/skills/defillama/tvl/fetch_historical_tvl.py index 1279546..981a026 100644 --- a/skills/defillama/tvl/fetch_historical_tvl.py +++ b/skills/defillama/tvl/fetch_historical_tvl.py @@ -1,10 +1,11 @@ """Tool for fetching total historical TVL via DeFiLlama API.""" from typing import List, Type + from pydantic import BaseModel, Field -from skills.defillama.base import DefiLlamaBaseTool from skills.defillama.api import fetch_historical_tvl +from skills.defillama.base import DefiLlamaBaseTool FETCH_TOTAL_HISTORICAL_TVL_PROMPT = """ This tool fetches historical Total Value Locked (TVL) data across all blockchains. @@ -16,22 +17,17 @@ class HistoricalTVLDataPoint(BaseModel): """Model representing a single TVL data point.""" - date: int = Field( - ..., - description="Unix timestamp of the TVL measurement" - ) - tvl: float = Field( - ..., - description="Total Value Locked in USD at this timestamp" - ) + date: int = Field(..., description="Unix timestamp of the TVL measurement") + tvl: float = Field(..., description="Total Value Locked in USD at this timestamp") class FetchHistoricalTVLInput(BaseModel): """Input schema for fetching historical TVL data. - + This endpoint doesn't require any parameters as it returns global TVL data across all chains. """ + pass @@ -40,21 +36,18 @@ class FetchHistoricalTVLResponse(BaseModel): data: List[HistoricalTVLDataPoint] = Field( default_factory=list, - description="List of historical TVL data points across all chains" - ) - error: str | None = Field( - default=None, - description="Error message if any" + description="List of historical TVL data points across all chains", ) + error: str | None = Field(default=None, description="Error message if any") class DefiLlamaFetchHistoricalTvl(DefiLlamaBaseTool): """Tool for fetching historical TVL data across all blockchains. - + This tool fetches the complete Total Value Locked (TVL) history aggregated across all chains using the DeFiLlama API. It includes rate limiting to ensure reliable data retrieval. - + Example: tvl_tool = DefiLlamaFetchHistoricalTvl( skill_store=store, @@ -86,16 +79,13 @@ async def _arun(self) -> FetchHistoricalTVLResponse: # Fetch TVL history from API result = await fetch_historical_tvl() - + # Check for API errors if isinstance(result, dict) and "error" in result: return FetchHistoricalTVLResponse(error=result["error"]) # Parse response into our schema - data_points = [ - HistoricalTVLDataPoint(**point) - for point in result - ] + data_points = [HistoricalTVLDataPoint(**point) for point in result] return FetchHistoricalTVLResponse(data=data_points) diff --git a/skills/defillama/tvl/fetch_protocol.py b/skills/defillama/tvl/fetch_protocol.py index f545ad9..998e724 100644 --- a/skills/defillama/tvl/fetch_protocol.py +++ b/skills/defillama/tvl/fetch_protocol.py @@ -1,11 +1,11 @@ """Tool for fetching specific protocol details via DeFi Llama API.""" -from typing import Dict, List, Optional, Union, Type -from datetime import datetime +from typing import Dict, List, Optional, Type + from pydantic import BaseModel, Field -from skills.defillama.base import DefiLlamaBaseTool from skills.defillama.api import fetch_protocol +from skills.defillama.base import DefiLlamaBaseTool FETCH_PROTOCOL_PROMPT = """ This tool fetches comprehensive details about a specific DeFi protocol. @@ -19,24 +19,36 @@ Returns complete protocol details or an error if the protocol is not found. """ + class TokenAmount(BaseModel): """Model representing token amounts at a specific date.""" + date: int = Field(..., description="Unix timestamp") tokens: Dict[str, float] = Field(..., description="Token amounts keyed by symbol") + class ChainTVLData(BaseModel): """Model representing TVL data for a specific chain.""" + tvl: List[Dict[str, float]] = Field(..., description="Historical TVL data points") - tokens: Optional[Dict[str, float]] = Field(None, description="Current token amounts") - tokensInUsd: Optional[Dict[str, float]] = Field(None, description="Current token amounts in USD") + tokens: Optional[Dict[str, float]] = Field( + None, description="Current token amounts" + ) + tokensInUsd: Optional[Dict[str, float]] = Field( + None, description="Current token amounts in USD" + ) + class HistoricalTVL(BaseModel): """Model representing a historical TVL data point.""" + date: int = Field(..., description="Unix timestamp") totalLiquidityUSD: float = Field(..., description="Total TVL in USD") + class Raise(BaseModel): """Model representing a funding round.""" + date: int = Field(..., description="Funding date") name: str = Field(..., description="Protocol name") round: str = Field(..., description="Funding round type") @@ -47,17 +59,23 @@ class Raise(BaseModel): categoryGroup: str = Field(..., description="Category group") source: str = Field(..., description="Information source") leadInvestors: List[str] = Field(default_factory=list, description="Lead investors") - otherInvestors: List[str] = Field(default_factory=list, description="Other investors") + otherInvestors: List[str] = Field( + default_factory=list, description="Other investors" + ) valuation: Optional[float] = Field(None, description="Valuation at time of raise") defillamaId: Optional[str] = Field(None, description="DefiLlama ID") + class Hallmark(BaseModel): """Model representing a significant protocol event.""" + timestamp: int description: str + class ProtocolDetail(BaseModel): """Model representing detailed protocol information.""" + # Basic Info id: str = Field(..., description="Protocol unique identifier") name: str = Field(..., description="Protocol name") @@ -70,24 +88,32 @@ class ProtocolDetail(BaseModel): # Chain Info chains: List[str] = Field(default_factory=list, description="Supported chains") currentChainTvls: Dict[str, float] = Field(..., description="Current TVL by chain") - chainTvls: Dict[str, ChainTVLData] = Field(..., description="Historical TVL data by chain") + chainTvls: Dict[str, ChainTVLData] = Field( + ..., description="Historical TVL data by chain" + ) # Identifiers gecko_id: Optional[str] = Field(None, description="CoinGecko ID") cmcId: Optional[str] = Field(None, description="CoinMarketCap ID") - + # Social & Development twitter: Optional[str] = Field(None, description="Twitter handle") treasury: Optional[str] = Field(None, description="Treasury information") - governanceID: Optional[List[str]] = Field(None, description="Governance identifiers") + governanceID: Optional[List[str]] = Field( + None, description="Governance identifiers" + ) github: Optional[List[str]] = Field(None, description="GitHub repositories") # Protocol Relationships - isParentProtocol: Optional[bool] = Field(None, description="Whether this is a parent protocol") + isParentProtocol: Optional[bool] = Field( + None, description="Whether this is a parent protocol" + ) otherProtocols: Optional[List[str]] = Field(None, description="Related protocols") # Historical Data - tokens: List[TokenAmount] = Field(default_factory=list, description="Historical token amounts") + tokens: List[TokenAmount] = Field( + default_factory=list, description="Historical token amounts" + ) tvl: List[HistoricalTVL] = Field(..., description="Historical TVL data points") raises: Optional[List[Raise]] = Field(None, description="Funding rounds") hallmarks: Optional[List[Hallmark]] = Field(None, description="Significant events") @@ -96,15 +122,20 @@ class ProtocolDetail(BaseModel): mcap: Optional[float] = Field(None, description="Market capitalization") metrics: Dict = Field(default_factory=dict, description="Additional metrics") + class DefiLlamaProtocolInput(BaseModel): """Input model for fetching protocol details.""" + protocol: str = Field(..., description="Protocol identifier to fetch") + class DefiLlamaProtocolOutput(BaseModel): """Output model for the protocol fetching tool.""" + protocol: Optional[ProtocolDetail] = Field(None, description="Protocol details") error: Optional[str] = Field(None, description="Error message if any") + class DefiLlamaFetchProtocol(DefiLlamaBaseTool): """Tool for fetching detailed protocol information from DeFi Llama. @@ -166,7 +197,7 @@ async def _arun(self, protocol: str) -> DefiLlamaProtocolOutput: protocol_detail = ProtocolDetail( **{k: v for k, v in result.items() if k not in ["hallmarks", "raises"]}, hallmarks=hallmarks, - raises=raises + raises=raises, ) return DefiLlamaProtocolOutput(protocol=protocol_detail) diff --git a/skills/defillama/tvl/fetch_protocol_current_tvl.py b/skills/defillama/tvl/fetch_protocol_current_tvl.py index 15683e1..7af582e 100644 --- a/skills/defillama/tvl/fetch_protocol_current_tvl.py +++ b/skills/defillama/tvl/fetch_protocol_current_tvl.py @@ -1,10 +1,11 @@ """Tool for fetching protocol TVL via DeFiLlama API.""" from typing import Type + from pydantic import BaseModel, Field -from skills.defillama.base import DefiLlamaBaseTool from skills.defillama.api import fetch_protocol_current_tvl +from skills.defillama.base import DefiLlamaBaseTool FETCH_TVL_PROMPT = """ This tool fetches the current Total Value Locked (TVL) for a specific DeFi protocol. @@ -15,36 +16,26 @@ class FetchProtocolCurrentTVLInput(BaseModel): """Input schema for fetching current protocol TVL.""" - + protocol: str = Field( - ..., - description="Protocol slug to fetch TVL for (e.g., 'aave', 'curve')" + ..., description="Protocol slug to fetch TVL for (e.g., 'aave', 'curve')" ) class FetchProtocolCurrentTVLResponse(BaseModel): """Response schema for current protocol TVL.""" - - protocol: str = Field( - ..., - description="Normalized protocol slug" - ) - tvl: float = Field( - ..., - description="Current Total Value Locked in USD" - ) - error: str | None = Field( - default=None, - description="Error message if any" - ) + + protocol: str = Field(..., description="Normalized protocol slug") + tvl: float = Field(..., description="Current Total Value Locked in USD") + error: str | None = Field(default=None, description="Error message if any") class DefiLlamaFetchProtocolCurrentTvl(DefiLlamaBaseTool): """Tool for fetching current TVL of a specific DeFi protocol. - + This tool fetches the current Total Value Locked (TVL) for a given protocol using the DeFiLlama API. It includes rate limiting to avoid API abuse. - + Example: tvl_tool = DefiLlamaFetchProtocolCurrentTvl( skill_store=store, @@ -76,9 +67,7 @@ async def _arun(self, protocol: str) -> FetchProtocolCurrentTVLResponse: is_rate_limited, error_msg = await self.check_rate_limit() if is_rate_limited: return FetchProtocolCurrentTVLResponse( - protocol=protocol, - tvl=0, - error=error_msg + protocol=protocol, tvl=0, error=error_msg ) # Normalize protocol slug @@ -90,19 +79,14 @@ async def _arun(self, protocol: str) -> FetchProtocolCurrentTVLResponse: # Check for API errors if isinstance(result, dict) and "error" in result: return FetchProtocolCurrentTVLResponse( - protocol=normalized_protocol, - tvl=0, - error=result["error"] + protocol=normalized_protocol, tvl=0, error=result["error"] ) return FetchProtocolCurrentTVLResponse( - protocol=normalized_protocol, - tvl=float(result) + protocol=normalized_protocol, tvl=float(result) ) except Exception as e: return FetchProtocolCurrentTVLResponse( - protocol=protocol, - tvl=0, - error=str(e) + protocol=protocol, tvl=0, error=str(e) ) diff --git a/skills/defillama/tvl/fetch_protocols.py b/skills/defillama/tvl/fetch_protocols.py index 34c6ba8..42b6480 100644 --- a/skills/defillama/tvl/fetch_protocols.py +++ b/skills/defillama/tvl/fetch_protocols.py @@ -1,10 +1,11 @@ """Tool for fetching all protocols via DeFi Llama API.""" -from typing import Dict, List, Optional, Union, Type +from typing import Dict, List, Optional, Type, Union + from pydantic import BaseModel, Field -from skills.defillama.base import DefiLlamaBaseTool from skills.defillama.api import fetch_protocols +from skills.defillama.base import DefiLlamaBaseTool FETCH_PROTOCOLS_PROMPT = """ This tool fetches information about all protocols tracked by DeFi Llama. @@ -20,13 +21,17 @@ Returns the complete list of protocols or an error if the request fails. """ + class Hallmark(BaseModel): """Model representing a protocol hallmark (significant event).""" + timestamp: int description: str + class Protocol(BaseModel): """Model representing a DeFi protocol.""" + # Basic Info id: str = Field(..., description="Protocol unique identifier") name: str = Field(..., description="Protocol name") @@ -36,7 +41,7 @@ class Protocol(BaseModel): description: Optional[str] = Field(None, description="Protocol description") chain: Optional[str] = Field(None, description="Main chain of the protocol") logo: Optional[str] = Field(None, description="URL to protocol logo") - + # Audit Information audits: Union[str, int] = Field("0", description="Number of audits") audit_note: Optional[str] = Field(None, description="Additional audit information") @@ -45,61 +50,76 @@ class Protocol(BaseModel): # External IDs gecko_id: Optional[str] = Field(None, description="CoinGecko ID") cmcId: Optional[Union[str, int]] = Field(None, description="CoinMarketCap ID") - + # Classification category: str = Field(..., description="Protocol category") - chains: List[str] = Field(default_factory=list, description="Chains the protocol operates on") - + chains: List[str] = Field( + default_factory=list, description="Chains the protocol operates on" + ) + # Module and Related Info module: str = Field(..., description="Module name in DefiLlama") - parentProtocol: Optional[str] = Field(None, description="Parent protocol identifier") - + parentProtocol: Optional[str] = Field( + None, description="Parent protocol identifier" + ) + # Social and Development twitter: Optional[str] = Field(None, description="Twitter handle") github: Optional[List[str]] = Field(None, description="GitHub organization names") - + # Protocol Relationships oracles: List[str] = Field(default_factory=list, description="Oracle services used") - forkedFrom: List[str] = Field(default_factory=list, description="Protocols this one was forked from") - + forkedFrom: List[str] = Field( + default_factory=list, description="Protocols this one was forked from" + ) + # Additional Metadata methodology: Optional[str] = Field(None, description="TVL calculation methodology") - listedAt: Optional[int] = Field(None, description="Timestamp when protocol was listed") - openSource: Optional[bool] = Field(None, description="Whether protocol is open source") + listedAt: Optional[int] = Field( + None, description="Timestamp when protocol was listed" + ) + openSource: Optional[bool] = Field( + None, description="Whether protocol is open source" + ) treasury: Optional[str] = Field(None, description="Treasury information") - misrepresentedTokens: Optional[bool] = Field(None, description="Whether tokens are misrepresented") - hallmarks: Optional[List[Hallmark]] = Field(None, description="Significant protocol events") - + misrepresentedTokens: Optional[bool] = Field( + None, description="Whether tokens are misrepresented" + ) + hallmarks: Optional[List[Hallmark]] = Field( + None, description="Significant protocol events" + ) + # TVL Related Data tvl: Optional[float] = Field(None, description="Total Value Locked in USD") chainTvls: Dict[str, float] = Field( - default_factory=dict, - description="TVL breakdown by chain including special types (staking, borrowed, etc.)" + default_factory=dict, + description="TVL breakdown by chain including special types (staking, borrowed, etc.)", ) change_1h: Optional[float] = Field(None, description="1 hour TVL change percentage") change_1d: Optional[float] = Field(None, description="1 day TVL change percentage") change_7d: Optional[float] = Field(None, description="7 day TVL change percentage") - + # Additional TVL Components staking: Optional[float] = Field(None, description="Value in staking") pool2: Optional[float] = Field(None, description="Value in pool2") borrowed: Optional[float] = Field(None, description="Value borrowed") - + # Token Information tokenBreakdowns: Dict[str, float] = Field( - default_factory=dict, - description="TVL breakdown by token" + default_factory=dict, description="TVL breakdown by token" ) mcap: Optional[float] = Field(None, description="Market capitalization") + class DefiLlamaProtocolsOutput(BaseModel): """Output model for the protocols fetching tool.""" + protocols: List[Protocol] = Field( - default_factory=list, - description="List of fetched protocols" + default_factory=list, description="List of fetched protocols" ) error: Optional[str] = Field(None, description="Error message if any") + class DefiLlamaFetchProtocols(DefiLlamaBaseTool): """Tool for fetching all protocols from DeFi Llama. @@ -156,12 +176,14 @@ async def _arun(self) -> DefiLlamaProtocolsOutput: # Create protocol model protocol = Protocol( **{k: v for k, v in protocol_data.items() if k != "hallmarks"}, - hallmarks=hallmarks + hallmarks=hallmarks, ) protocols.append(protocol) except Exception as e: # Log error for individual protocol processing but continue with others - print(f"Error processing protocol {protocol_data.get('name', 'unknown')}: {str(e)}") + print( + f"Error processing protocol {protocol_data.get('name', 'unknown')}: {str(e)}" + ) continue return DefiLlamaProtocolsOutput(protocols=protocols) diff --git a/skills/defillama/volumes/fetch_dex_overview.py b/skills/defillama/volumes/fetch_dex_overview.py index 3605a51..0456ea4 100644 --- a/skills/defillama/volumes/fetch_dex_overview.py +++ b/skills/defillama/volumes/fetch_dex_overview.py @@ -1,10 +1,11 @@ """Tool for fetching DEX overview data via DeFi Llama API.""" -from typing import Dict, List, Optional, Type +from typing import Dict, List, Optional + from pydantic import BaseModel, Field -from skills.defillama.base import DefiLlamaBaseTool from skills.defillama.api import fetch_dex_overview +from skills.defillama.base import DefiLlamaBaseTool FETCH_DEX_OVERVIEW_PROMPT = """ This tool fetches comprehensive overview data for DEX protocols from DeFi Llama. @@ -24,7 +25,9 @@ class MethodologyInfo(BaseModel): Revenue: Optional[str] = Field(None, description="Revenue model") ProtocolRevenue: Optional[str] = Field(None, description="Protocol revenue info") HoldersRevenue: Optional[str] = Field(None, description="Holder revenue info") - SupplySideRevenue: Optional[str] = Field(None, description="Supply side revenue info") + SupplySideRevenue: Optional[str] = Field( + None, description="Supply side revenue info" + ) class ProtocolInfo(BaseModel): @@ -61,7 +64,9 @@ class ProtocolInfo(BaseModel): chains: List[str] = Field(..., description="Supported chains") protocolType: str = Field(..., description="Protocol type") methodologyURL: Optional[str] = Field(None, description="Methodology URL") - methodology: Optional[MethodologyInfo] = Field(None, description="Methodology details") + methodology: Optional[MethodologyInfo] = Field( + None, description="Methodology details" + ) latestFetchIsOk: bool = Field(..., description="Latest fetch status") disabled: Optional[bool] = Field(None, description="Whether protocol is disabled") parentProtocol: Optional[str] = Field(None, description="Parent protocol") @@ -74,98 +79,40 @@ class FetchDexOverviewResponse(BaseModel): """Response schema for DEX overview data.""" totalDataChart: List = Field( - default_factory=list, - description="Total data chart points" + default_factory=list, description="Total data chart points" ) totalDataChartBreakdown: List = Field( - default_factory=list, - description="Total data chart breakdown" + default_factory=list, description="Total data chart breakdown" ) breakdown24h: Optional[Dict[str, Dict[str, float]]] = Field( - None, - description="24h breakdown by chain" + None, description="24h breakdown by chain" ) breakdown30d: Optional[Dict[str, Dict[str, float]]] = Field( - None, - description="30d breakdown by chain" - ) - chain: Optional[str] = Field( - None, - description="Specific chain" - ) - allChains: List[str] = Field( - ..., - description="List of all chains" - ) - total24h: float = Field( - ..., - description="24h total" - ) - total48hto24h: float = Field( - ..., - description="48h to 24h total" - ) - total7d: float = Field( - ..., - description="7d total" - ) - total14dto7d: float = Field( - ..., - description="14d to 7d total" - ) - total60dto30d: float = Field( - ..., - description="60d to 30d total" - ) - total30d: float = Field( - ..., - description="30d total" - ) - total1y: float = Field( - ..., - description="1y total" - ) - change_1d: float = Field( - ..., - description="1d change" - ) - change_7d: float = Field( - ..., - description="7d change" - ) - change_1m: float = Field( - ..., - description="1m change" - ) - change_7dover7d: float = Field( - ..., - description="7d over 7d change" - ) - change_30dover30d: float = Field( - ..., - description="30d over 30d change" - ) - total7DaysAgo: float = Field( - ..., - description="Total 7 days ago" - ) - total30DaysAgo: float = Field( - ..., - description="Total 30 days ago" - ) - protocols: List[ProtocolInfo] = Field( - ..., - description="List of protocol data" - ) - error: Optional[str] = Field( - None, - description="Error message if any" + None, description="30d breakdown by chain" ) + chain: Optional[str] = Field(None, description="Specific chain") + allChains: List[str] = Field(..., description="List of all chains") + total24h: float = Field(..., description="24h total") + total48hto24h: float = Field(..., description="48h to 24h total") + total7d: float = Field(..., description="7d total") + total14dto7d: float = Field(..., description="14d to 7d total") + total60dto30d: float = Field(..., description="60d to 30d total") + total30d: float = Field(..., description="30d total") + total1y: float = Field(..., description="1y total") + change_1d: float = Field(..., description="1d change") + change_7d: float = Field(..., description="7d change") + change_1m: float = Field(..., description="1m change") + change_7dover7d: float = Field(..., description="7d over 7d change") + change_30dover30d: float = Field(..., description="30d over 30d change") + total7DaysAgo: float = Field(..., description="Total 7 days ago") + total30DaysAgo: float = Field(..., description="Total 30 days ago") + protocols: List[ProtocolInfo] = Field(..., description="List of protocol data") + error: Optional[str] = Field(None, description="Error message if any") class DefiLlamaFetchDexOverview(DefiLlamaBaseTool): """Tool for fetching DEX overview data from DeFi Llama. - + This tool retrieves comprehensive data about DEX protocols, including volumes, metrics, and chain breakdowns. @@ -193,14 +140,14 @@ async def _arun(self) -> FetchDexOverviewResponse: FetchDexOverviewResponse containing overview data or error """ try: - # Check rate limiting + # Check rate limiting is_rate_limited, error_msg = await self.check_rate_limit() if is_rate_limited: return FetchDexOverviewResponse(error=error_msg) # Fetch overview data from API result = await fetch_dex_overview() - + # Check for API errors if isinstance(result, dict) and "error" in result: return FetchDexOverviewResponse(error=result["error"]) diff --git a/skills/defillama/volumes/fetch_dex_summary.py b/skills/defillama/volumes/fetch_dex_summary.py index 7f62d94..cf4a083 100644 --- a/skills/defillama/volumes/fetch_dex_summary.py +++ b/skills/defillama/volumes/fetch_dex_summary.py @@ -1,10 +1,11 @@ """Tool for fetching DEX protocol summary data via DeFi Llama API.""" from typing import Dict, List, Optional, Type + from pydantic import BaseModel, Field -from skills.defillama.base import DefiLlamaBaseTool from skills.defillama.api import fetch_dex_summary +from skills.defillama.base import DefiLlamaBaseTool FETCH_DEX_SUMMARY_PROMPT = """ This tool fetches summary data for a specific DEX protocol from DeFi Llama. @@ -21,10 +22,7 @@ class FetchDexSummaryInput(BaseModel): """Input schema for fetching DEX protocol summary.""" - protocol: str = Field( - ..., - description="Protocol identifier (e.g. 'uniswap')" - ) + protocol: str = Field(..., description="Protocol identifier (e.g. 'uniswap')") class FetchDexSummaryResponse(BaseModel): @@ -65,14 +63,16 @@ class FetchDexSummaryResponse(BaseModel): total7d: Optional[float] = Field(None, description="7d total volume") totalAllTime: Optional[float] = Field(None, description="All time total volume") totalDataChart: List = Field(default_factory=list, description="Total data chart") - totalDataChartBreakdown: List = Field(default_factory=list, description="Chart breakdown") + totalDataChartBreakdown: List = Field( + default_factory=list, description="Chart breakdown" + ) change_1d: Optional[float] = Field(None, description="1d change percentage") error: Optional[str] = Field(None, description="Error message if any") class DefiLlamaFetchDexSummary(DefiLlamaBaseTool): """Tool for fetching DEX protocol summary data from DeFi Llama. - + This tool retrieves detailed information about a specific DEX protocol, including metadata, metrics, and related protocols. @@ -89,15 +89,11 @@ class DefiLlamaFetchDexSummary(DefiLlamaBaseTool): description: str = FETCH_DEX_SUMMARY_PROMPT args_schema: Type[BaseModel] = FetchDexSummaryInput - def _run( - self, protocol: str - ) -> FetchDexSummaryResponse: + def _run(self, protocol: str) -> FetchDexSummaryResponse: """Synchronous implementation - not supported.""" raise NotImplementedError("Use _arun instead") - async def _arun( - self, protocol: str - ) -> FetchDexSummaryResponse: + async def _arun(self, protocol: str) -> FetchDexSummaryResponse: """Fetch summary data for the given DEX protocol. Args: @@ -114,7 +110,7 @@ async def _arun( # Fetch protocol data from API result = await fetch_dex_summary(protocol=protocol) - + # Check for API errors if isinstance(result, dict) and "error" in result: return FetchDexSummaryResponse(error=result["error"]) diff --git a/skills/defillama/volumes/fetch_options_overview.py b/skills/defillama/volumes/fetch_options_overview.py index 57f27a8..41674dd 100644 --- a/skills/defillama/volumes/fetch_options_overview.py +++ b/skills/defillama/volumes/fetch_options_overview.py @@ -1,10 +1,11 @@ """Tool for fetching options overview data via DeFi Llama API.""" from typing import Dict, List, Optional, Type + from pydantic import BaseModel, Field -from skills.defillama.base import DefiLlamaBaseTool from skills.defillama.api import fetch_options_overview +from skills.defillama.base import DefiLlamaBaseTool FETCH_OPTIONS_OVERVIEW_PROMPT = """ This tool fetches comprehensive overview data for all options protocols from DeFi Llama. @@ -15,17 +16,27 @@ - Chain breakdowns """ + class ProtocolMethodology(BaseModel): """Model representing protocol methodology data.""" + UserFees: Optional[str] = Field(None, description="User fees description") Fees: Optional[str] = Field(None, description="Fees description") Revenue: Optional[str] = Field(None, description="Revenue description") - ProtocolRevenue: Optional[str] = Field(None, description="Protocol revenue description") - HoldersRevenue: Optional[str] = Field(None, description="Holders revenue description") - SupplySideRevenue: Optional[str] = Field(None, description="Supply side revenue description") + ProtocolRevenue: Optional[str] = Field( + None, description="Protocol revenue description" + ) + HoldersRevenue: Optional[str] = Field( + None, description="Holders revenue description" + ) + SupplySideRevenue: Optional[str] = Field( + None, description="Supply side revenue description" + ) + class Protocol(BaseModel): """Model representing protocol data.""" + name: str = Field(..., description="Protocol name") displayName: str = Field(..., description="Display name of protocol") defillamaId: str = Field(..., description="DeFi Llama ID") @@ -41,12 +52,20 @@ class Protocol(BaseModel): change_1d: Optional[float] = Field(None, description="24-hour change percentage") change_7d: Optional[float] = Field(None, description="7-day change percentage") change_1m: Optional[float] = Field(None, description="30-day change percentage") - methodology: Optional[ProtocolMethodology] = Field(None, description="Protocol methodology") - breakdown24h: Optional[Dict[str, Dict[str, float]]] = Field(None, description="24-hour breakdown by chain") - breakdown30d: Optional[Dict[str, Dict[str, float]]] = Field(None, description="30-day breakdown by chain") + methodology: Optional[ProtocolMethodology] = Field( + None, description="Protocol methodology" + ) + breakdown24h: Optional[Dict[str, Dict[str, float]]] = Field( + None, description="24-hour breakdown by chain" + ) + breakdown30d: Optional[Dict[str, Dict[str, float]]] = Field( + None, description="30-day breakdown by chain" + ) + class FetchOptionsOverviewResponse(BaseModel): """Response schema for options overview data.""" + total24h: float = Field(..., description="Total volume in last 24 hours") total7d: float = Field(..., description="Total volume in last 7 days") total30d: float = Field(..., description="Total volume in last 30 days") @@ -58,9 +77,10 @@ class FetchOptionsOverviewResponse(BaseModel): protocols: List[Protocol] = Field(..., description="List of protocols") error: Optional[str] = Field(None, description="Error message if any") + class DefiLlamaFetchOptionsOverview(DefiLlamaBaseTool): """Tool for fetching options overview data from DeFi Llama. - + This tool retrieves comprehensive data about all options protocols, including volume metrics, change percentages, and detailed protocol information. @@ -95,7 +115,7 @@ async def _arun(self) -> FetchOptionsOverviewResponse: # Fetch overview data from API result = await fetch_options_overview() - + # Check for API errors if isinstance(result, dict) and "error" in result: return FetchOptionsOverviewResponse(error=result["error"]) diff --git a/skills/defillama/yields/fetch_pool_chart.py b/skills/defillama/yields/fetch_pool_chart.py index 3f0b719..acb244a 100644 --- a/skills/defillama/yields/fetch_pool_chart.py +++ b/skills/defillama/yields/fetch_pool_chart.py @@ -1,11 +1,11 @@ """Tool for fetching pool chart data via DeFi Llama API.""" -from typing import Dict, List, Optional, Type +from typing import List, Optional, Type + from pydantic import BaseModel, Field -from datetime import datetime -from skills.defillama.base import DefiLlamaBaseTool from skills.defillama.api import fetch_pool_chart +from skills.defillama.base import DefiLlamaBaseTool FETCH_POOL_CHART_PROMPT = """ This tool fetches historical chart data from DeFi Llama for a specific pool. @@ -21,65 +21,34 @@ class PoolDataPoint(BaseModel): """Model representing a single historical data point.""" - timestamp: str = Field( - ..., - description="ISO formatted timestamp of the data point" - ) - tvlUsd: float = Field( - ..., - description="Total Value Locked in USD" - ) - apy: Optional[float] = Field( - None, - description="Total APY including rewards" - ) - apyBase: Optional[float] = Field( - None, - description="Base APY without rewards" - ) - apyReward: Optional[float] = Field( - None, - description="Additional APY from rewards" - ) - il7d: Optional[float] = Field( - None, - description="7-day impermanent loss" - ) - apyBase7d: Optional[float] = Field( - None, - description="7-day base APY" - ) + timestamp: str = Field(..., description="ISO formatted timestamp of the data point") + tvlUsd: float = Field(..., description="Total Value Locked in USD") + apy: Optional[float] = Field(None, description="Total APY including rewards") + apyBase: Optional[float] = Field(None, description="Base APY without rewards") + apyReward: Optional[float] = Field(None, description="Additional APY from rewards") + il7d: Optional[float] = Field(None, description="7-day impermanent loss") + apyBase7d: Optional[float] = Field(None, description="7-day base APY") class FetchPoolChartInput(BaseModel): """Input schema for fetching pool chart data.""" - pool_id: str = Field( - ..., - description="ID of the pool to fetch chart data for" - ) + pool_id: str = Field(..., description="ID of the pool to fetch chart data for") class FetchPoolChartResponse(BaseModel): """Response schema for pool chart data.""" - status: str = Field( - "success", - description="Response status" - ) + status: str = Field("success", description="Response status") data: List[PoolDataPoint] = Field( - default_factory=list, - description="List of historical data points" - ) - error: Optional[str] = Field( - None, - description="Error message if any" + default_factory=list, description="List of historical data points" ) + error: Optional[str] = Field(None, description="Error message if any") class DefiLlamaFetchPoolChart(DefiLlamaBaseTool): """Tool for fetching pool chart data from DeFi Llama. - + This tool retrieves historical data for a specific pool, including TVL and APY metrics over time. @@ -98,15 +67,11 @@ class DefiLlamaFetchPoolChart(DefiLlamaBaseTool): description: str = FETCH_POOL_CHART_PROMPT args_schema: Type[BaseModel] = FetchPoolChartInput - def _run( - self, pool_id: str - ) -> FetchPoolChartResponse: + def _run(self, pool_id: str) -> FetchPoolChartResponse: """Synchronous implementation - not supported.""" raise NotImplementedError("Use _arun instead") - async def _arun( - self, pool_id: str - ) -> FetchPoolChartResponse: + async def _arun(self, pool_id: str) -> FetchPoolChartResponse: """Fetch historical chart data for the given pool. Args: @@ -123,7 +88,7 @@ async def _arun( # Fetch chart data from API result = await fetch_pool_chart(pool_id=pool_id) - + # Check for API errors if isinstance(result, dict) and "error" in result: return FetchPoolChartResponse(error=result["error"]) diff --git a/skills/defillama/yields/fetch_pools.py b/skills/defillama/yields/fetch_pools.py index a2748a7..5a30b09 100644 --- a/skills/defillama/yields/fetch_pools.py +++ b/skills/defillama/yields/fetch_pools.py @@ -1,10 +1,11 @@ """Tool for fetching pool data via DeFi Llama API.""" -from typing import Dict, List, Optional, Type +from typing import List, Optional + from pydantic import BaseModel, Field -from skills.defillama.base import DefiLlamaBaseTool from skills.defillama.api import fetch_pools +from skills.defillama.base import DefiLlamaBaseTool FETCH_POOLS_PROMPT = """ This tool fetches comprehensive data about yield-generating pools from DeFi Llama. @@ -21,156 +22,66 @@ class PredictionData(BaseModel): """Model representing prediction data for a pool.""" predictedClass: Optional[str] = Field( - None, - description="Predicted direction of APY movement" + None, description="Predicted direction of APY movement" ) predictedProbability: Optional[float] = Field( - None, - description="Probability of the prediction" - ) - binnedConfidence: Optional[int] = Field( - None, - description="Confidence level bucket" + None, description="Probability of the prediction" ) + binnedConfidence: Optional[int] = Field(None, description="Confidence level bucket") class PoolData(BaseModel): """Model representing a single pool's data.""" - chain: str = Field( - ..., - description="Blockchain network" - ) - project: str = Field( - ..., - description="Protocol or project name" - ) - symbol: str = Field( - ..., - description="Token or pool symbol" - ) - tvlUsd: float = Field( - ..., - description="Total Value Locked in USD" - ) - apyBase: Optional[float] = Field( - None, - description="Base APY without rewards" - ) - apyReward: Optional[float] = Field( - None, - description="Additional APY from rewards" - ) - apy: Optional[float] = Field( - None, - description="Total APY including rewards" - ) + chain: str = Field(..., description="Blockchain network") + project: str = Field(..., description="Protocol or project name") + symbol: str = Field(..., description="Token or pool symbol") + tvlUsd: float = Field(..., description="Total Value Locked in USD") + apyBase: Optional[float] = Field(None, description="Base APY without rewards") + apyReward: Optional[float] = Field(None, description="Additional APY from rewards") + apy: Optional[float] = Field(None, description="Total APY including rewards") rewardTokens: Optional[List[str]] = Field( - None, - description="List of reward token addresses" - ) - pool: Optional[str] = Field( - None, - description="Pool identifier" - ) - apyPct1D: Optional[float] = Field( - None, - description="1-day APY percentage change" - ) - apyPct7D: Optional[float] = Field( - None, - description="7-day APY percentage change" - ) - apyPct30D: Optional[float] = Field( - None, - description="30-day APY percentage change" - ) - stablecoin: bool = Field( - False, - description="Whether pool involves stablecoins" - ) - ilRisk: str = Field( - "no", - description="Impermanent loss risk assessment" - ) - exposure: str = Field( - "single", - description="Asset exposure type" - ) + None, description="List of reward token addresses" + ) + pool: Optional[str] = Field(None, description="Pool identifier") + apyPct1D: Optional[float] = Field(None, description="1-day APY percentage change") + apyPct7D: Optional[float] = Field(None, description="7-day APY percentage change") + apyPct30D: Optional[float] = Field(None, description="30-day APY percentage change") + stablecoin: bool = Field(False, description="Whether pool involves stablecoins") + ilRisk: str = Field("no", description="Impermanent loss risk assessment") + exposure: str = Field("single", description="Asset exposure type") predictions: Optional[PredictionData] = Field( - None, - description="APY movement predictions" - ) - poolMeta: Optional[str] = Field( - None, - description="Additional pool metadata" - ) - mu: Optional[float] = Field( - None, - description="Mean APY value" - ) - sigma: Optional[float] = Field( - None, - description="APY standard deviation" - ) - count: Optional[int] = Field( - None, - description="Number of data points" - ) - outlier: bool = Field( - False, - description="Whether pool is an outlier" + None, description="APY movement predictions" ) + poolMeta: Optional[str] = Field(None, description="Additional pool metadata") + mu: Optional[float] = Field(None, description="Mean APY value") + sigma: Optional[float] = Field(None, description="APY standard deviation") + count: Optional[int] = Field(None, description="Number of data points") + outlier: bool = Field(False, description="Whether pool is an outlier") underlyingTokens: Optional[List[str]] = Field( - None, - description="List of underlying token addresses" - ) - il7d: Optional[float] = Field( - None, - description="7-day impermanent loss" - ) - apyBase7d: Optional[float] = Field( - None, - description="7-day base APY" - ) - apyMean30d: Optional[float] = Field( - None, - description="30-day mean APY" - ) - volumeUsd1d: Optional[float] = Field( - None, - description="24h volume in USD" - ) - volumeUsd7d: Optional[float] = Field( - None, - description="7-day volume in USD" + None, description="List of underlying token addresses" ) + il7d: Optional[float] = Field(None, description="7-day impermanent loss") + apyBase7d: Optional[float] = Field(None, description="7-day base APY") + apyMean30d: Optional[float] = Field(None, description="30-day mean APY") + volumeUsd1d: Optional[float] = Field(None, description="24h volume in USD") + volumeUsd7d: Optional[float] = Field(None, description="7-day volume in USD") apyBaseInception: Optional[float] = Field( - None, - description="Base APY since inception" + None, description="Base APY since inception" ) class FetchPoolsResponse(BaseModel): """Response schema for pool data.""" - status: str = Field( - "success", - description="Response status" - ) - data: List[PoolData] = Field( - default_factory=list, - description="List of pool data" - ) - error: Optional[str] = Field( - None, - description="Error message if any" - ) + status: str = Field("success", description="Response status") + data: List[PoolData] = Field(default_factory=list, description="List of pool data") + error: Optional[str] = Field(None, description="Error message if any") class DefiLlamaFetchPools(DefiLlamaBaseTool): """Tool for fetching pool data from DeFi Llama. - + This tool retrieves comprehensive data about yield-generating pools, including TVL, APYs, risk metrics, and predictions. @@ -205,7 +116,7 @@ async def _arun(self) -> FetchPoolsResponse: # Fetch pool data from API result = await fetch_pools() - + # Check for API errors if isinstance(result, dict) and "error" in result: return FetchPoolsResponse(error=result["error"])