feat: add Ethereum chain support

.env.example

@@ -3,6 +3,12 @@ export AWS_REGION=''
 export AWS_ACCESS_KEY_ID=''
 export AWS_SECRET_ACCESS_KEY=''
 export SOLANA_RPC_URL=''
+export ETH_RPC_URL='' # Ethereum RPC (Alchemy/Infura recommended)
+
+# The Graph subgraph URLs (optional — defaults are provided)
+# Production usage requires a Graph API key: https://thegraph.com/studio/
+export UNISWAP_V3_SUBGRAPH_URL=''
+export AAVE_V3_SUBGRAPH_URL=''
 export OPENROUTER_API_KEY=''
 export GEMINI_API_KEY=''
 export DD_API_KEY=''

agent/tools.py

@@ -44,7 +44,7 @@ async def retrieve_solana_pools(
 
  # Create a query to filter pools
  query = PoolQuery(
- chain=Chain.SOLANA, # Currently only supporting Solana
+ chain=Chain.SOLANA,
  tokens=tokens or [],
  user_tokens=user_tokens,
  )
@@ -56,9 +56,37 @@ async def retrieve_solana_pools(
  return pools
 
 
+@tool
+@track_tool_usage("retrieve_ethereum_pools")
+async def retrieve_ethereum_pools(
+ tokens: List[str] = None,
+ config: RunnableConfig = None,
+) -> List[Pool]:
+ """
+ Retrieves Ethereum pools matching the specified criteria that the user can invest in.
+ Includes Uniswap V3 AMM pools and Aave V3 lending pools.
+ """
+ configurable = config["configurable"]
+ user_tokens: List[WalletTokenHolding] = configurable["tokens"]
+ protocol_registry: ProtocolRegistry = configurable["protocol_registry"]
+
+ query = PoolQuery(
+ chain=Chain.ETHEREUM,
+ tokens=tokens or [],
+ user_tokens=user_tokens,
+ )
+
+ pools = await protocol_registry.get_pools(query)
+ if len(pools) == 0:
+ return "No Ethereum pools found."
+
+ return pools
+
+
 def create_investor_agent_toolkit() -> List[BaseTool]:
  return [
  retrieve_solana_pools,
+ retrieve_ethereum_pools,
  ]
 
 

main.py

@@ -19,12 +19,16 @@
 from onchain.pools.solana.orca_protocol import OrcaProtocol
 from onchain.pools.solana.save_protocol import SaveProtocol
 from onchain.pools.solana.kamino_protocol import KaminoProtocol
+from onchain.pools.ethereum.uniswap_v3_protocol import UniswapV3Protocol
+from onchain.pools.ethereum.aave_protocol import AaveProtocol
 
 # Define protocols enabled
 protocols = [
  OrcaProtocol.PROTOCOL_NAME,
  SaveProtocol.PROTOCOL_NAME,
  KaminoProtocol.PROTOCOL_NAME,
+ UniswapV3Protocol.PROTOCOL_NAME,
+ AaveProtocol.PROTOCOL_NAME,
 ]
 
 # Create the FastAPI app

onchain/pools/ethereum/aave_protocol.py

@@ -0,0 +1,162 @@
+"""Aave V3 protocol implementation for Ethereum lending pool discovery."""
+
+from typing import List, Optional, Dict, Any
+import logging
+import os
+
+import aiohttp
+
+from api.api_types import Pool, Token, Chain, PoolType
+from onchain.pools.protocol import Protocol
+from onchain.tokens.metadata import TokenMetadataRepo
+
+logger = logging.getLogger(__name__)
+
+# Aave V3 Ethereum subgraph — configurable via env var.
+AAVE_V3_SUBGRAPH_URL = os.environ.get(
+ "AAVE_V3_SUBGRAPH_URL",
+ "https://gateway.thegraph.com/api/subgraphs/id/Cd2gEDVeqnjBn1hSeqFMitw8Q1iiyV9FYUZkLNRcL87g",
+)
+
+# Stablecoins for classification
+STABLECOIN_SYMBOLS = {"USDC", "USDT", "DAI", "BUSD", "TUSD", "FRAX", "LUSD", "GUSD", "sUSD"}
+
+
+class AaveProtocol(Protocol):
+ """
+ Aave V3 protocol — fetches Ethereum lending reserves.
+
+ Returns each reserve as a lending Pool with supply APR.
+ Uses The Graph subgraph for on-chain data.
+ """
+
+ PROTOCOL_NAME = "aave-v3"
+
+ _session: Optional[aiohttp.ClientSession] = None
+
+ @property
+ def name(self) -> str:
+ return self.PROTOCOL_NAME
+
+ @property
+ async def session(self) -> aiohttp.ClientSession:
+ if self._session is None:
+ self._session = aiohttp.ClientSession(
+ timeout=aiohttp.ClientTimeout(total=30)
+ )
+ return self._session
+
+ async def close(self):
+ if self._session:
+ await self._session.close()
+ self._session = None
+
+ async def get_pools(self, token_metadata_repo: TokenMetadataRepo) -> List[Pool]:
+ """Fetch Aave V3 reserves from the subgraph."""
+ query = """
+ {
+ reserves(
+ first: 50,
+ where: { isActive: true }
+ ) {
+ id
+ symbol
+ name
+ decimals
+ underlyingAsset
+ liquidityRate
+ totalATokenSupply
+ totalCurrentVariableDebt
+ availableLiquidity
+ price {
+ priceInEth
+ }
+ }
+ }
+ """
+
+ try:
+ session = await self.session
+ async with session.post(
+ AAVE_V3_SUBGRAPH_URL,
+ json={"query": query},
+ ) as response:
+ if response.status in (401, 403):
+ logger.error(
+ f"Aave V3 subgraph auth failed ({response.status}). "
+ f"Set AAVE_V3_SUBGRAPH_URL with a valid Graph API key."
+ )
+ return []
+ if response.status != 200:
+ logger.error(f"Aave V3 subgraph returned {response.status}")
+ return []
+ data = await response.json()
+ except Exception as e:
+ logger.error(f"Error fetching Aave V3 reserves: {e}")
+ return []
+
+ reserves = data.get("data", {}).get("reserves", [])
+
+ # Resolve ETH/USD price for TVL conversion.
+ # Use the WETH token metadata from DexScreener (already cached).
+ eth_usd_price = 0.0
+ eth_meta = await token_metadata_repo.get_token_metadata(
+ "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2", "ethereum"
+ )
+ if eth_meta and eth_meta.price:
+ eth_usd_price = float(eth_meta.price)
+
+ return self._convert_to_pools(reserves, eth_usd_price)
+
+ def _convert_to_pools(
+ self, reserves: List[Dict[str, Any]], eth_usd_price: float
+ ) -> List[Pool]:
+ result: List[Pool] = []
+
+ for reserve in reserves:
+ symbol = reserve.get("symbol", "")
+ name = reserve.get("name", "")
+ address = reserve.get("underlyingAsset", "")
+ decimals = int(reserve.get("decimals", 18))
+
+ tokens = [
+ Token(address=address, name=name, symbol=symbol),
+ ]
+
+ # Aave liquidityRate is in RAY units (1e27), convert to APR percentage
+ liquidity_rate = int(reserve.get("liquidityRate", 0))
+ supply_apr = (liquidity_rate / 1e27) * 100
+
+ # Calculate TVL in USD:
+ # totalATokenSupply is in token-native units (needs /10^decimals)
+ # priceInEth is the token price denominated in ETH (wei-scaled, 1e18)
+ # Multiply by eth_usd_price to get USD
+ total_supply_raw = int(reserve.get("totalATokenSupply", 0))
+ total_supply = total_supply_raw / (10**decimals)
+
+ price_in_eth_raw = reserve.get("price", {}).get("priceInEth", "0")
+ price_in_eth = int(price_in_eth_raw) / 1e18
+
+ if eth_usd_price > 0 and price_in_eth > 0:
+ tvl_usd = str(round(total_supply * price_in_eth * eth_usd_price, 2))
+ else:
+ tvl_usd = str(round(total_supply, 2))
+
+ is_stablecoin = symbol.upper() in STABLECOIN_SYMBOLS
+
+ pool = Pool(
+ id=reserve.get("id", ""),
+ chain=Chain.ETHEREUM,
+ protocol="Aave V3",
+ tokens=tokens,
+ type=PoolType.LENDING,
+ TVL=tvl_usd,
+ APRLastDay=round(supply_apr, 2),
+ APRLastWeek=round(supply_apr, 2),
+ APRLastMonth=round(supply_apr, 2),
+ isStableCoin=is_stablecoin,
+ impermanentLossRisk=False, # Lending has no IL
+ )
+ result.append(pool)
+
+ return result

onchain/pools/ethereum/test_ethereum_protocols.py

@@ -0,0 +1,126 @@
+import unittest
+import os
+import boto3
+import asyncio
+
+from onchain.pools.ethereum.uniswap_v3_protocol import UniswapV3Protocol
+from onchain.pools.ethereum.aave_protocol import AaveProtocol
+from onchain.portfolio.ethereum_portfolio import EthereumPortfolioFetcher
+from onchain.tokens.metadata import TokenMetadataRepo
+from api.api_types import Chain, PoolType
+import dotenv
+
+dotenv.load_dotenv()
+
+
+class TestEthereumProtocols(unittest.TestCase):
+ def setUp(self):
+ dynamodb = boto3.resource(
+ "dynamodb",
+ region_name=os.environ.get("AWS_REGION"),
+ aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"),
+ aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"),
+ )
+ tokens_table = dynamodb.Table("token_metadata_v2")
+ self.token_metadata_repo = TokenMetadataRepo(tokens_table)
+
+ def test_uniswap_v3(self):
+ uniswap = UniswapV3Protocol()
+ pools = asyncio.run(uniswap.get_pools(self.token_metadata_repo))
+
+ self.assertGreater(len(pools), 5)
+ print(f"Uniswap V3 pools: {len(pools)}")
+
+ # Verify pool structure
+ for pool in pools[:3]:
+ self.assertEqual(pool.chain, Chain.ETHEREUM)
+ self.assertEqual(pool.type, PoolType.AMM)
+ self.assertIn("Uniswap V3", pool.protocol)
+ self.assertEqual(len(pool.tokens), 2)
+ self.assertIsNotNone(pool.TVL)
+ self.assertGreaterEqual(pool.APRLastDay, 0)
+
+ def test_uniswap_v3_stablecoin_detection(self):
+ uniswap = UniswapV3Protocol()
+ pools = asyncio.run(uniswap.get_pools(self.token_metadata_repo))
+
+ # Non-stablecoin pools should have IL risk
+ for pool in pools:
+ if not pool.isStableCoin:
+ self.assertTrue(pool.impermanentLossRisk)
+ else:
+ self.assertFalse(pool.impermanentLossRisk)
+
+ def test_aave_v3(self):
+ aave = AaveProtocol()
+ pools = asyncio.run(aave.get_pools(self.token_metadata_repo))
+
+ self.assertGreater(len(pools), 5)
+ print(f"Aave V3 reserves: {len(pools)}")
+
+ # Verify pool structure
+ for pool in pools[:3]:
+ self.assertEqual(pool.chain, Chain.ETHEREUM)
+ self.assertEqual(pool.type, PoolType.LENDING)
+ self.assertEqual(pool.protocol, "Aave V3")
+ self.assertEqual(len(pool.tokens), 1)
+ self.assertFalse(pool.impermanentLossRisk)
+ self.assertGreaterEqual(pool.APRLastDay, 0)
+
+ def test_aave_v3_tvl_is_usd(self):
+ """TVL should be in USD (not raw token units)."""
+ aave = AaveProtocol()
+ pools = asyncio.run(aave.get_pools(self.token_metadata_repo))
+
+ for pool in pools:
+ tvl = float(pool.TVL)
+ # Any active Aave reserve should have meaningful TVL in USD
+ if tvl > 0:
+ self.assertGreater(tvl, 100, f"{pool.tokens[0].symbol} TVL too low: {tvl}")
+
+
+class TestEthereumPortfolio(unittest.TestCase):
+ def setUp(self):
+ dynamodb = boto3.resource(
+ "dynamodb",
+ region_name=os.environ.get("AWS_REGION"),
+ aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"),
+ aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"),
+ )
+ tokens_table = dynamodb.Table("token_metadata_v2")
+ self.token_metadata_repo = TokenMetadataRepo(tokens_table)
+ self.portfolio_fetcher = EthereumPortfolioFetcher(self.token_metadata_repo)
+
+ def tearDown(self):
+ asyncio.run(self.portfolio_fetcher.close())
+
+ def test_get_portfolio(self):
+ # Vitalik's public wallet
+ portfolio = asyncio.run(
+ self.portfolio_fetcher.get_portfolio(
+ "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045"
+ )
+ )
+ print(portfolio)
+
+ self.assertGreater(len(portfolio.holdings), 0)
+ # Vitalik holds ETH
+ eth_holdings = [h for h in portfolio.holdings if h.symbol == "ETH"]
+ self.assertEqual(len(eth_holdings), 1)
+ self.assertGreater(eth_holdings[0].amount, 0)
+
+ def test_empty_wallet(self):
+ # Zero address should have no holdings
+ portfolio = asyncio.run(
+ self.portfolio_fetcher.get_portfolio(
+ "0x0000000000000000000000000000000000000001"
+ )
+ )
+ self.assertEqual(len(portfolio.holdings), 0)
+
+ def test_invalid_address(self):
+ portfolio = asyncio.run(
+ self.portfolio_fetcher.get_portfolio("not-a-valid-address")
+ )
+ self.assertEqual(len(portfolio.holdings), 0)
+ self.assertEqual(portfolio.total_value_usd, 0)

.env.example

@@ -3,6 +3,12 @@ export AWS_REGION=''
 export AWS_ACCESS_KEY_ID=''
 export AWS_SECRET_ACCESS_KEY=''
 export SOLANA_RPC_URL=''
+export ETH_RPC_URL='' # Ethereum RPC (Alchemy/Infura recommended)
+
+# The Graph subgraph URLs (optional — defaults are provided)
+# Production usage requires a Graph API key: https://thegraph.com/studio/
+export UNISWAP_V3_SUBGRAPH_URL=''
+export AAVE_V3_SUBGRAPH_URL=''
 export OPENROUTER_API_KEY=''
 export GEMINI_API_KEY=''
 export DD_API_KEY=''

agent/tools.py

@@ -44,7 +44,7 @@ async def retrieve_solana_pools(
 
  # Create a query to filter pools
  query = PoolQuery(
- chain=Chain.SOLANA, # Currently only supporting Solana
+ chain=Chain.SOLANA,
  tokens=tokens or [],
  user_tokens=user_tokens,
  )
@@ -56,9 +56,37 @@ async def retrieve_solana_pools(
  return pools
 
 
+@tool
+@track_tool_usage("retrieve_ethereum_pools")
+async def retrieve_ethereum_pools(
+ tokens: List[str] = None,
+ config: RunnableConfig = None,
+) -> List[Pool]:
+ """
+ Retrieves Ethereum pools matching the specified criteria that the user can invest in.
+ Includes Uniswap V3 AMM pools and Aave V3 lending pools.
+ """
+ configurable = config["configurable"]
+ user_tokens: List[WalletTokenHolding] = configurable["tokens"]
+ protocol_registry: ProtocolRegistry = configurable["protocol_registry"]
+
+ query = PoolQuery(
+ chain=Chain.ETHEREUM,
+ tokens=tokens or [],
+ user_tokens=user_tokens,
+ )
+
+ pools = await protocol_registry.get_pools(query)
+ if len(pools) == 0:
+ return "No Ethereum pools found."
+
+ return pools
+
+
 def create_investor_agent_toolkit() -> List[BaseTool]:
  return [
  retrieve_solana_pools,
+ retrieve_ethereum_pools,
  ]
 
 

main.py

@@ -19,12 +19,16 @@
 from onchain.pools.solana.orca_protocol import OrcaProtocol
 from onchain.pools.solana.save_protocol import SaveProtocol
 from onchain.pools.solana.kamino_protocol import KaminoProtocol
+from onchain.pools.ethereum.uniswap_v3_protocol import UniswapV3Protocol
+from onchain.pools.ethereum.aave_protocol import AaveProtocol
 
 # Define protocols enabled
 protocols = [
  OrcaProtocol.PROTOCOL_NAME,
  SaveProtocol.PROTOCOL_NAME,
  KaminoProtocol.PROTOCOL_NAME,
+ UniswapV3Protocol.PROTOCOL_NAME,
+ AaveProtocol.PROTOCOL_NAME,
 ]
 
 # Create the FastAPI app

onchain/pools/ethereum/aave_protocol.py

@@ -0,0 +1,162 @@
+"""Aave V3 protocol implementation for Ethereum lending pool discovery."""
+
+from typing import List, Optional, Dict, Any
+import logging
+import os
+
+import aiohttp
+
+from api.api_types import Pool, Token, Chain, PoolType
+from onchain.pools.protocol import Protocol
+from onchain.tokens.metadata import TokenMetadataRepo
+
+logger = logging.getLogger(__name__)
+
+# Aave V3 Ethereum subgraph — configurable via env var.
+AAVE_V3_SUBGRAPH_URL = os.environ.get(
+ "AAVE_V3_SUBGRAPH_URL",
+ "https://gateway.thegraph.com/api/subgraphs/id/Cd2gEDVeqnjBn1hSeqFMitw8Q1iiyV9FYUZkLNRcL87g",
+)
+
+# Stablecoins for classification
+STABLECOIN_SYMBOLS = {"USDC", "USDT", "DAI", "BUSD", "TUSD", "FRAX", "LUSD", "GUSD", "sUSD"}
+
+
+class AaveProtocol(Protocol):
+ """
+ Aave V3 protocol — fetches Ethereum lending reserves.
+
+ Returns each reserve as a lending Pool with supply APR.
+ Uses The Graph subgraph for on-chain data.
+ """
+
+ PROTOCOL_NAME = "aave-v3"
+
+ _session: Optional[aiohttp.ClientSession] = None
+
+ @property
+ def name(self) -> str:
+ return self.PROTOCOL_NAME
+
+ @property
+ async def session(self) -> aiohttp.ClientSession:
+ if self._session is None:
+ self._session = aiohttp.ClientSession(
+ timeout=aiohttp.ClientTimeout(total=30)
+ )
+ return self._session
+
+ async def close(self):
+ if self._session:
+ await self._session.close()
+ self._session = None
+
+ async def get_pools(self, token_metadata_repo: TokenMetadataRepo) -> List[Pool]:
+ """Fetch Aave V3 reserves from the subgraph."""
+ query = """
+ {
+ reserves(
+ first: 50,
+ where: { isActive: true }
+ ) {
+ id
+ symbol
+ name
+ decimals
+ underlyingAsset
+ liquidityRate
+ totalATokenSupply
+ totalCurrentVariableDebt
+ availableLiquidity
+ price {
+ priceInEth
+ }
+ }
+ }
+ """
+
+ try:
+ session = await self.session
+ async with session.post(
+ AAVE_V3_SUBGRAPH_URL,
+ json={"query": query},
+ ) as response:
+ if response.status in (401, 403):
+ logger.error(
+ f"Aave V3 subgraph auth failed ({response.status}). "
+ f"Set AAVE_V3_SUBGRAPH_URL with a valid Graph API key."
+ )
+ return []
+ if response.status != 200:
+ logger.error(f"Aave V3 subgraph returned {response.status}")
+ return []
+ data = await response.json()
+ except Exception as e:
+ logger.error(f"Error fetching Aave V3 reserves: {e}")
+ return []
+
+ reserves = data.get("data", {}).get("reserves", [])
+
+ # Resolve ETH/USD price for TVL conversion.
+ # Use the WETH token metadata from DexScreener (already cached).
+ eth_usd_price = 0.0
+ eth_meta = await token_metadata_repo.get_token_metadata(
+ "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2", "ethereum"
+ )
+ if eth_meta and eth_meta.price:
+ eth_usd_price = float(eth_meta.price)
+
+ return self._convert_to_pools(reserves, eth_usd_price)
+
+ def _convert_to_pools(
+ self, reserves: List[Dict[str, Any]], eth_usd_price: float
+ ) -> List[Pool]:
+ result: List[Pool] = []
+
+ for reserve in reserves:
+ symbol = reserve.get("symbol", "")
+ name = reserve.get("name", "")
+ address = reserve.get("underlyingAsset", "")
+ decimals = int(reserve.get("decimals", 18))
+
+ tokens = [
+ Token(address=address, name=name, symbol=symbol),
+ ]
+
+ # Aave liquidityRate is in RAY units (1e27), convert to APR percentage
+ liquidity_rate = int(reserve.get("liquidityRate", 0))
+ supply_apr = (liquidity_rate / 1e27) * 100
+
+ # Calculate TVL in USD:
+ # totalATokenSupply is in token-native units (needs /10^decimals)
+ # priceInEth is the token price denominated in ETH (wei-scaled, 1e18)
+ # Multiply by eth_usd_price to get USD
+ total_supply_raw = int(reserve.get("totalATokenSupply", 0))
+ total_supply = total_supply_raw / (10**decimals)
+
+ price_in_eth_raw = reserve.get("price", {}).get("priceInEth", "0")
+ price_in_eth = int(price_in_eth_raw) / 1e18
+
+ if eth_usd_price > 0 and price_in_eth > 0:
+ tvl_usd = str(round(total_supply * price_in_eth * eth_usd_price, 2))
+ else:
+ tvl_usd = str(round(total_supply, 2))
+
+ is_stablecoin = symbol.upper() in STABLECOIN_SYMBOLS
+
+ pool = Pool(
+ id=reserve.get("id", ""),
+ chain=Chain.ETHEREUM,
+ protocol="Aave V3",
+ tokens=tokens,
+ type=PoolType.LENDING,
+ TVL=tvl_usd,
+ APRLastDay=round(supply_apr, 2),
+ APRLastWeek=round(supply_apr, 2),
+ APRLastMonth=round(supply_apr, 2),
+ isStableCoin=is_stablecoin,
+ impermanentLossRisk=False, # Lending has no IL
+ )
+ result.append(pool)
+
+ return result

onchain/pools/ethereum/test_ethereum_protocols.py

@@ -0,0 +1,126 @@
+import unittest
+import os
+import boto3
+import asyncio
+
+from onchain.pools.ethereum.uniswap_v3_protocol import UniswapV3Protocol
+from onchain.pools.ethereum.aave_protocol import AaveProtocol
+from onchain.portfolio.ethereum_portfolio import EthereumPortfolioFetcher
+from onchain.tokens.metadata import TokenMetadataRepo
+from api.api_types import Chain, PoolType
+import dotenv
+
+dotenv.load_dotenv()
+
+
+class TestEthereumProtocols(unittest.TestCase):
+ def setUp(self):
+ dynamodb = boto3.resource(
+ "dynamodb",
+ region_name=os.environ.get("AWS_REGION"),
+ aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"),
+ aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"),
+ )
+ tokens_table = dynamodb.Table("token_metadata_v2")
+ self.token_metadata_repo = TokenMetadataRepo(tokens_table)
+
+ def test_uniswap_v3(self):
+ uniswap = UniswapV3Protocol()
+ pools = asyncio.run(uniswap.get_pools(self.token_metadata_repo))
+
+ self.assertGreater(len(pools), 5)
+ print(f"Uniswap V3 pools: {len(pools)}")
+
+ # Verify pool structure
+ for pool in pools[:3]:
+ self.assertEqual(pool.chain, Chain.ETHEREUM)
+ self.assertEqual(pool.type, PoolType.AMM)
+ self.assertIn("Uniswap V3", pool.protocol)
+ self.assertEqual(len(pool.tokens), 2)
+ self.assertIsNotNone(pool.TVL)
+ self.assertGreaterEqual(pool.APRLastDay, 0)
+
+ def test_uniswap_v3_stablecoin_detection(self):
+ uniswap = UniswapV3Protocol()
+ pools = asyncio.run(uniswap.get_pools(self.token_metadata_repo))
+
+ # Non-stablecoin pools should have IL risk
+ for pool in pools:
+ if not pool.isStableCoin:
+ self.assertTrue(pool.impermanentLossRisk)
+ else:
+ self.assertFalse(pool.impermanentLossRisk)
+
+ def test_aave_v3(self):
+ aave = AaveProtocol()
+ pools = asyncio.run(aave.get_pools(self.token_metadata_repo))
+
+ self.assertGreater(len(pools), 5)
+ print(f"Aave V3 reserves: {len(pools)}")
+
+ # Verify pool structure
+ for pool in pools[:3]:
+ self.assertEqual(pool.chain, Chain.ETHEREUM)
+ self.assertEqual(pool.type, PoolType.LENDING)
+ self.assertEqual(pool.protocol, "Aave V3")
+ self.assertEqual(len(pool.tokens), 1)
+ self.assertFalse(pool.impermanentLossRisk)
+ self.assertGreaterEqual(pool.APRLastDay, 0)
+
+ def test_aave_v3_tvl_is_usd(self):
+ """TVL should be in USD (not raw token units)."""
+ aave = AaveProtocol()
+ pools = asyncio.run(aave.get_pools(self.token_metadata_repo))
+
+ for pool in pools:
+ tvl = float(pool.TVL)
+ # Any active Aave reserve should have meaningful TVL in USD
+ if tvl > 0:
+ self.assertGreater(tvl, 100, f"{pool.tokens[0].symbol} TVL too low: {tvl}")
+
+
+class TestEthereumPortfolio(unittest.TestCase):
+ def setUp(self):
+ dynamodb = boto3.resource(
+ "dynamodb",
+ region_name=os.environ.get("AWS_REGION"),
+ aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"),
+ aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"),
+ )
+ tokens_table = dynamodb.Table("token_metadata_v2")
+ self.token_metadata_repo = TokenMetadataRepo(tokens_table)
+ self.portfolio_fetcher = EthereumPortfolioFetcher(self.token_metadata_repo)
+
+ def tearDown(self):
+ asyncio.run(self.portfolio_fetcher.close())
+
+ def test_get_portfolio(self):
+ # Vitalik's public wallet
+ portfolio = asyncio.run(
+ self.portfolio_fetcher.get_portfolio(
+ "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045"
+ )
+ )
+ print(portfolio)
+
+ self.assertGreater(len(portfolio.holdings), 0)
+ # Vitalik holds ETH
+ eth_holdings = [h for h in portfolio.holdings if h.symbol == "ETH"]
+ self.assertEqual(len(eth_holdings), 1)
+ self.assertGreater(eth_holdings[0].amount, 0)
+
+ def test_empty_wallet(self):
+ # Zero address should have no holdings
+ portfolio = asyncio.run(
+ self.portfolio_fetcher.get_portfolio(
+ "0x0000000000000000000000000000000000000001"
+ )
+ )
+ self.assertEqual(len(portfolio.holdings), 0)
+
+ def test_invalid_address(self):
+ portfolio = asyncio.run(
+ self.portfolio_fetcher.get_portfolio("not-a-valid-address")
+ )
+ self.assertEqual(len(portfolio.holdings), 0)
+ self.assertEqual(portfolio.total_value_usd, 0)

.env.example

@@ -3,6 +3,12 @@ export AWS_REGION=''
 export AWS_ACCESS_KEY_ID=''
 export AWS_SECRET_ACCESS_KEY=''
 export SOLANA_RPC_URL=''
+export ETH_RPC_URL='' # Ethereum RPC (Alchemy/Infura recommended)
+
+# The Graph subgraph URLs (optional — defaults are provided)
+# Production usage requires a Graph API key: https://thegraph.com/studio/
+export UNISWAP_V3_SUBGRAPH_URL=''
+export AAVE_V3_SUBGRAPH_URL=''
 export OPENROUTER_API_KEY=''
 export GEMINI_API_KEY=''
 export DD_API_KEY=''

agent/tools.py

@@ -44,7 +44,7 @@ async def retrieve_solana_pools(
 
  # Create a query to filter pools
  query = PoolQuery(
- chain=Chain.SOLANA, # Currently only supporting Solana
+ chain=Chain.SOLANA,
  tokens=tokens or [],
  user_tokens=user_tokens,
  )
@@ -56,9 +56,37 @@ async def retrieve_solana_pools(
  return pools
 
 
+@tool
+@track_tool_usage("retrieve_ethereum_pools")
+async def retrieve_ethereum_pools(
+ tokens: List[str] = None,
+ config: RunnableConfig = None,
+) -> List[Pool]:
+ """
+ Retrieves Ethereum pools matching the specified criteria that the user can invest in.
+ Includes Uniswap V3 AMM pools and Aave V3 lending pools.
+ """
+ configurable = config["configurable"]
+ user_tokens: List[WalletTokenHolding] = configurable["tokens"]
+ protocol_registry: ProtocolRegistry = configurable["protocol_registry"]
+
+ query = PoolQuery(
+ chain=Chain.ETHEREUM,
+ tokens=tokens or [],
+ user_tokens=user_tokens,
+ )
+
+ pools = await protocol_registry.get_pools(query)
+ if len(pools) == 0:
+ return "No Ethereum pools found."
+
+ return pools
+
+
 def create_investor_agent_toolkit() -> List[BaseTool]:
  return [
  retrieve_solana_pools,
+ retrieve_ethereum_pools,
  ]
 
 

main.py

@@ -19,12 +19,16 @@
 from onchain.pools.solana.orca_protocol import OrcaProtocol
 from onchain.pools.solana.save_protocol import SaveProtocol
 from onchain.pools.solana.kamino_protocol import KaminoProtocol
+from onchain.pools.ethereum.uniswap_v3_protocol import UniswapV3Protocol
+from onchain.pools.ethereum.aave_protocol import AaveProtocol
 
 # Define protocols enabled
 protocols = [
  OrcaProtocol.PROTOCOL_NAME,
  SaveProtocol.PROTOCOL_NAME,
  KaminoProtocol.PROTOCOL_NAME,
+ UniswapV3Protocol.PROTOCOL_NAME,
+ AaveProtocol.PROTOCOL_NAME,
 ]
 
 # Create the FastAPI app

onchain/pools/ethereum/aave_protocol.py

@@ -0,0 +1,162 @@
+"""Aave V3 protocol implementation for Ethereum lending pool discovery."""
+
+from typing import List, Optional, Dict, Any
+import logging
+import os
+
+import aiohttp
+
+from api.api_types import Pool, Token, Chain, PoolType
+from onchain.pools.protocol import Protocol
+from onchain.tokens.metadata import TokenMetadataRepo
+
+logger = logging.getLogger(__name__)
+
+# Aave V3 Ethereum subgraph — configurable via env var.
+AAVE_V3_SUBGRAPH_URL = os.environ.get(
+ "AAVE_V3_SUBGRAPH_URL",
+ "https://gateway.thegraph.com/api/subgraphs/id/Cd2gEDVeqnjBn1hSeqFMitw8Q1iiyV9FYUZkLNRcL87g",
+)
+
+# Stablecoins for classification
+STABLECOIN_SYMBOLS = {"USDC", "USDT", "DAI", "BUSD", "TUSD", "FRAX", "LUSD", "GUSD", "sUSD"}
+
+
+class AaveProtocol(Protocol):
+ """
+ Aave V3 protocol — fetches Ethereum lending reserves.
+
+ Returns each reserve as a lending Pool with supply APR.
+ Uses The Graph subgraph for on-chain data.
+ """
+
+ PROTOCOL_NAME = "aave-v3"
+
+ _session: Optional[aiohttp.ClientSession] = None
+
+ @property
+ def name(self) -> str:
+ return self.PROTOCOL_NAME
+
+ @property
+ async def session(self) -> aiohttp.ClientSession:
+ if self._session is None:
+ self._session = aiohttp.ClientSession(
+ timeout=aiohttp.ClientTimeout(total=30)
+ )
+ return self._session
+
+ async def close(self):
+ if self._session:
+ await self._session.close()
+ self._session = None
+
+ async def get_pools(self, token_metadata_repo: TokenMetadataRepo) -> List[Pool]:
+ """Fetch Aave V3 reserves from the subgraph."""
+ query = """
+ {
+ reserves(
+ first: 50,
+ where: { isActive: true }
+ ) {
+ id
+ symbol
+ name
+ decimals
+ underlyingAsset
+ liquidityRate
+ totalATokenSupply
+ totalCurrentVariableDebt
+ availableLiquidity
+ price {
+ priceInEth
+ }
+ }
+ }
+ """
+
+ try:
+ session = await self.session
+ async with session.post(
+ AAVE_V3_SUBGRAPH_URL,
+ json={"query": query},
+ ) as response:
+ if response.status in (401, 403):
+ logger.error(
+ f"Aave V3 subgraph auth failed ({response.status}). "
+ f"Set AAVE_V3_SUBGRAPH_URL with a valid Graph API key."
+ )
+ return []
+ if response.status != 200:
+ logger.error(f"Aave V3 subgraph returned {response.status}")
+ return []
+ data = await response.json()
+ except Exception as e:
+ logger.error(f"Error fetching Aave V3 reserves: {e}")
+ return []
+
+ reserves = data.get("data", {}).get("reserves", [])
+
+ # Resolve ETH/USD price for TVL conversion.
+ # Use the WETH token metadata from DexScreener (already cached).
+ eth_usd_price = 0.0
+ eth_meta = await token_metadata_repo.get_token_metadata(
+ "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2", "ethereum"
+ )
+ if eth_meta and eth_meta.price:
+ eth_usd_price = float(eth_meta.price)
+
+ return self._convert_to_pools(reserves, eth_usd_price)
+
+ def _convert_to_pools(
+ self, reserves: List[Dict[str, Any]], eth_usd_price: float
+ ) -> List[Pool]:
+ result: List[Pool] = []
+
+ for reserve in reserves:
+ symbol = reserve.get("symbol", "")
+ name = reserve.get("name", "")
+ address = reserve.get("underlyingAsset", "")
+ decimals = int(reserve.get("decimals", 18))
+
+ tokens = [
+ Token(address=address, name=name, symbol=symbol),
+ ]
+
+ # Aave liquidityRate is in RAY units (1e27), convert to APR percentage
+ liquidity_rate = int(reserve.get("liquidityRate", 0))
+ supply_apr = (liquidity_rate / 1e27) * 100
+
+ # Calculate TVL in USD:
+ # totalATokenSupply is in token-native units (needs /10^decimals)
+ # priceInEth is the token price denominated in ETH (wei-scaled, 1e18)
+ # Multiply by eth_usd_price to get USD
+ total_supply_raw = int(reserve.get("totalATokenSupply", 0))
+ total_supply = total_supply_raw / (10**decimals)
+
+ price_in_eth_raw = reserve.get("price", {}).get("priceInEth", "0")
+ price_in_eth = int(price_in_eth_raw) / 1e18
+
+ if eth_usd_price > 0 and price_in_eth > 0:
+ tvl_usd = str(round(total_supply * price_in_eth * eth_usd_price, 2))
+ else:
+ tvl_usd = str(round(total_supply, 2))
+
+ is_stablecoin = symbol.upper() in STABLECOIN_SYMBOLS
+
+ pool = Pool(
+ id=reserve.get("id", ""),
+ chain=Chain.ETHEREUM,
+ protocol="Aave V3",
+ tokens=tokens,
+ type=PoolType.LENDING,
+ TVL=tvl_usd,
+ APRLastDay=round(supply_apr, 2),
+ APRLastWeek=round(supply_apr, 2),
+ APRLastMonth=round(supply_apr, 2),
+ isStableCoin=is_stablecoin,
+ impermanentLossRisk=False, # Lending has no IL
+ )
+ result.append(pool)
+
+ return result

onchain/pools/ethereum/test_ethereum_protocols.py

@@ -0,0 +1,126 @@
+import unittest
+import os
+import boto3
+import asyncio
+
+from onchain.pools.ethereum.uniswap_v3_protocol import UniswapV3Protocol
+from onchain.pools.ethereum.aave_protocol import AaveProtocol
+from onchain.portfolio.ethereum_portfolio import EthereumPortfolioFetcher
+from onchain.tokens.metadata import TokenMetadataRepo
+from api.api_types import Chain, PoolType
+import dotenv
+
+dotenv.load_dotenv()
+
+
+class TestEthereumProtocols(unittest.TestCase):
+ def setUp(self):
+ dynamodb = boto3.resource(
+ "dynamodb",
+ region_name=os.environ.get("AWS_REGION"),
+ aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"),
+ aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"),
+ )
+ tokens_table = dynamodb.Table("token_metadata_v2")
+ self.token_metadata_repo = TokenMetadataRepo(tokens_table)
+
+ def test_uniswap_v3(self):
+ uniswap = UniswapV3Protocol()
+ pools = asyncio.run(uniswap.get_pools(self.token_metadata_repo))
+
+ self.assertGreater(len(pools), 5)
+ print(f"Uniswap V3 pools: {len(pools)}")
+
+ # Verify pool structure
+ for pool in pools[:3]:
+ self.assertEqual(pool.chain, Chain.ETHEREUM)
+ self.assertEqual(pool.type, PoolType.AMM)
+ self.assertIn("Uniswap V3", pool.protocol)
+ self.assertEqual(len(pool.tokens), 2)
+ self.assertIsNotNone(pool.TVL)
+ self.assertGreaterEqual(pool.APRLastDay, 0)
+
+ def test_uniswap_v3_stablecoin_detection(self):
+ uniswap = UniswapV3Protocol()
+ pools = asyncio.run(uniswap.get_pools(self.token_metadata_repo))
+
+ # Non-stablecoin pools should have IL risk
+ for pool in pools:
+ if not pool.isStableCoin:
+ self.assertTrue(pool.impermanentLossRisk)
+ else:
+ self.assertFalse(pool.impermanentLossRisk)
+
+ def test_aave_v3(self):
+ aave = AaveProtocol()
+ pools = asyncio.run(aave.get_pools(self.token_metadata_repo))
+
+ self.assertGreater(len(pools), 5)
+ print(f"Aave V3 reserves: {len(pools)}")
+
+ # Verify pool structure
+ for pool in pools[:3]:
+ self.assertEqual(pool.chain, Chain.ETHEREUM)
+ self.assertEqual(pool.type, PoolType.LENDING)
+ self.assertEqual(pool.protocol, "Aave V3")
+ self.assertEqual(len(pool.tokens), 1)
+ self.assertFalse(pool.impermanentLossRisk)
+ self.assertGreaterEqual(pool.APRLastDay, 0)
+
+ def test_aave_v3_tvl_is_usd(self):
+ """TVL should be in USD (not raw token units)."""
+ aave = AaveProtocol()
+ pools = asyncio.run(aave.get_pools(self.token_metadata_repo))
+
+ for pool in pools:
+ tvl = float(pool.TVL)
+ # Any active Aave reserve should have meaningful TVL in USD
+ if tvl > 0:
+ self.assertGreater(tvl, 100, f"{pool.tokens[0].symbol} TVL too low: {tvl}")
+
+
+class TestEthereumPortfolio(unittest.TestCase):
+ def setUp(self):
+ dynamodb = boto3.resource(
+ "dynamodb",
+ region_name=os.environ.get("AWS_REGION"),
+ aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"),
+ aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"),
+ )
+ tokens_table = dynamodb.Table("token_metadata_v2")
+ self.token_metadata_repo = TokenMetadataRepo(tokens_table)
+ self.portfolio_fetcher = EthereumPortfolioFetcher(self.token_metadata_repo)
+
+ def tearDown(self):
+ asyncio.run(self.portfolio_fetcher.close())
+
+ def test_get_portfolio(self):
+ # Vitalik's public wallet
+ portfolio = asyncio.run(
+ self.portfolio_fetcher.get_portfolio(
+ "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045"
+ )
+ )
+ print(portfolio)
+
+ self.assertGreater(len(portfolio.holdings), 0)
+ # Vitalik holds ETH
+ eth_holdings = [h for h in portfolio.holdings if h.symbol == "ETH"]
+ self.assertEqual(len(eth_holdings), 1)
+ self.assertGreater(eth_holdings[0].amount, 0)
+
+ def test_empty_wallet(self):
+ # Zero address should have no holdings
+ portfolio = asyncio.run(
+ self.portfolio_fetcher.get_portfolio(
+ "0x0000000000000000000000000000000000000001"
+ )
+ )
+ self.assertEqual(len(portfolio.holdings), 0)
+
+ def test_invalid_address(self):
+ portfolio = asyncio.run(
+ self.portfolio_fetcher.get_portfolio("not-a-valid-address")
+ )
+ self.assertEqual(len(portfolio.holdings), 0)
+ self.assertEqual(portfolio.total_value_usd, 0)

.env.example

@@ -3,6 +3,12 @@ export AWS_REGION=''
 export AWS_ACCESS_KEY_ID=''
 export AWS_SECRET_ACCESS_KEY=''
 export SOLANA_RPC_URL=''
+export ETH_RPC_URL='' # Ethereum RPC (Alchemy/Infura recommended)
+
+# The Graph subgraph URLs (optional — defaults are provided)
+# Production usage requires a Graph API key: https://thegraph.com/studio/
+export UNISWAP_V3_SUBGRAPH_URL=''
+export AAVE_V3_SUBGRAPH_URL=''
 export OPENROUTER_API_KEY=''
 export GEMINI_API_KEY=''
 export DD_API_KEY=''

agent/tools.py

@@ -44,7 +44,7 @@ async def retrieve_solana_pools(
 
  # Create a query to filter pools
  query = PoolQuery(
- chain=Chain.SOLANA, # Currently only supporting Solana
+ chain=Chain.SOLANA,
  tokens=tokens or [],
  user_tokens=user_tokens,
  )
@@ -56,9 +56,37 @@ async def retrieve_solana_pools(
  return pools
 
 
+@tool
+@track_tool_usage("retrieve_ethereum_pools")
+async def retrieve_ethereum_pools(
+ tokens: List[str] = None,
+ config: RunnableConfig = None,
+) -> List[Pool]:
+ """
+ Retrieves Ethereum pools matching the specified criteria that the user can invest in.
+ Includes Uniswap V3 AMM pools and Aave V3 lending pools.
+ """
+ configurable = config["configurable"]
+ user_tokens: List[WalletTokenHolding] = configurable["tokens"]
+ protocol_registry: ProtocolRegistry = configurable["protocol_registry"]
+
+ query = PoolQuery(
+ chain=Chain.ETHEREUM,
+ tokens=tokens or [],
+ user_tokens=user_tokens,
+ )
+
+ pools = await protocol_registry.get_pools(query)
+ if len(pools) == 0:
+ return "No Ethereum pools found."
+
+ return pools
+
+
 def create_investor_agent_toolkit() -> List[BaseTool]:
  return [
  retrieve_solana_pools,
+ retrieve_ethereum_pools,
  ]
 
 

main.py

@@ -19,12 +19,16 @@
 from onchain.pools.solana.orca_protocol import OrcaProtocol
 from onchain.pools.solana.save_protocol import SaveProtocol
 from onchain.pools.solana.kamino_protocol import KaminoProtocol
+from onchain.pools.ethereum.uniswap_v3_protocol import UniswapV3Protocol
+from onchain.pools.ethereum.aave_protocol import AaveProtocol
 
 # Define protocols enabled
 protocols = [
  OrcaProtocol.PROTOCOL_NAME,
  SaveProtocol.PROTOCOL_NAME,
  KaminoProtocol.PROTOCOL_NAME,
+ UniswapV3Protocol.PROTOCOL_NAME,
+ AaveProtocol.PROTOCOL_NAME,
 ]
 
 # Create the FastAPI app

onchain/pools/ethereum/aave_protocol.py

@@ -0,0 +1,162 @@
+"""Aave V3 protocol implementation for Ethereum lending pool discovery."""
+
+from typing import List, Optional, Dict, Any
+import logging
+import os
+
+import aiohttp
+
+from api.api_types import Pool, Token, Chain, PoolType
+from onchain.pools.protocol import Protocol
+from onchain.tokens.metadata import TokenMetadataRepo
+
+logger = logging.getLogger(__name__)
+
+# Aave V3 Ethereum subgraph — configurable via env var.
+AAVE_V3_SUBGRAPH_URL = os.environ.get(
+ "AAVE_V3_SUBGRAPH_URL",
+ "https://gateway.thegraph.com/api/subgraphs/id/Cd2gEDVeqnjBn1hSeqFMitw8Q1iiyV9FYUZkLNRcL87g",
+)
+
+# Stablecoins for classification
+STABLECOIN_SYMBOLS = {"USDC", "USDT", "DAI", "BUSD", "TUSD", "FRAX", "LUSD", "GUSD", "sUSD"}
+
+
+class AaveProtocol(Protocol):
+ """
+ Aave V3 protocol — fetches Ethereum lending reserves.
+
+ Returns each reserve as a lending Pool with supply APR.
+ Uses The Graph subgraph for on-chain data.
+ """
+
+ PROTOCOL_NAME = "aave-v3"
+
+ _session: Optional[aiohttp.ClientSession] = None
+
+ @property
+ def name(self) -> str:
+ return self.PROTOCOL_NAME
+
+ @property
+ async def session(self) -> aiohttp.ClientSession:
+ if self._session is None:
+ self._session = aiohttp.ClientSession(
+ timeout=aiohttp.ClientTimeout(total=30)
+ )
+ return self._session
+
+ async def close(self):
+ if self._session:
+ await self._session.close()
+ self._session = None
+
+ async def get_pools(self, token_metadata_repo: TokenMetadataRepo) -> List[Pool]:
+ """Fetch Aave V3 reserves from the subgraph."""
+ query = """
+ {
+ reserves(
+ first: 50,
+ where: { isActive: true }
+ ) {
+ id
+ symbol
+ name
+ decimals
+ underlyingAsset
+ liquidityRate
+ totalATokenSupply
+ totalCurrentVariableDebt
+ availableLiquidity
+ price {
+ priceInEth
+ }
+ }
+ }
+ """
+
+ try:
+ session = await self.session
+ async with session.post(
+ AAVE_V3_SUBGRAPH_URL,
+ json={"query": query},
+ ) as response:
+ if response.status in (401, 403):
+ logger.error(
+ f"Aave V3 subgraph auth failed ({response.status}). "
+ f"Set AAVE_V3_SUBGRAPH_URL with a valid Graph API key."
+ )
+ return []
+ if response.status != 200:
+ logger.error(f"Aave V3 subgraph returned {response.status}")
+ return []
+ data = await response.json()
+ except Exception as e:
+ logger.error(f"Error fetching Aave V3 reserves: {e}")
+ return []
+
+ reserves = data.get("data", {}).get("reserves", [])
+
+ # Resolve ETH/USD price for TVL conversion.
+ # Use the WETH token metadata from DexScreener (already cached).
+ eth_usd_price = 0.0
+ eth_meta = await token_metadata_repo.get_token_metadata(
+ "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2", "ethereum"
+ )
+ if eth_meta and eth_meta.price:
+ eth_usd_price = float(eth_meta.price)
+
+ return self._convert_to_pools(reserves, eth_usd_price)
+
+ def _convert_to_pools(
+ self, reserves: List[Dict[str, Any]], eth_usd_price: float
+ ) -> List[Pool]:
+ result: List[Pool] = []
+
+ for reserve in reserves:
+ symbol = reserve.get("symbol", "")
+ name = reserve.get("name", "")
+ address = reserve.get("underlyingAsset", "")
+ decimals = int(reserve.get("decimals", 18))
+
+ tokens = [
+ Token(address=address, name=name, symbol=symbol),
+ ]
+
+ # Aave liquidityRate is in RAY units (1e27), convert to APR percentage
+ liquidity_rate = int(reserve.get("liquidityRate", 0))
+ supply_apr = (liquidity_rate / 1e27) * 100
+
+ # Calculate TVL in USD:
+ # totalATokenSupply is in token-native units (needs /10^decimals)
+ # priceInEth is the token price denominated in ETH (wei-scaled, 1e18)
+ # Multiply by eth_usd_price to get USD
+ total_supply_raw = int(reserve.get("totalATokenSupply", 0))
+ total_supply = total_supply_raw / (10**decimals)
+
+ price_in_eth_raw = reserve.get("price", {}).get("priceInEth", "0")
+ price_in_eth = int(price_in_eth_raw) / 1e18
+
+ if eth_usd_price > 0 and price_in_eth > 0:
+ tvl_usd = str(round(total_supply * price_in_eth * eth_usd_price, 2))
+ else:
+ tvl_usd = str(round(total_supply, 2))
+
+ is_stablecoin = symbol.upper() in STABLECOIN_SYMBOLS
+
+ pool = Pool(
+ id=reserve.get("id", ""),
+ chain=Chain.ETHEREUM,
+ protocol="Aave V3",
+ tokens=tokens,
+ type=PoolType.LENDING,
+ TVL=tvl_usd,
+ APRLastDay=round(supply_apr, 2),
+ APRLastWeek=round(supply_apr, 2),
+ APRLastMonth=round(supply_apr, 2),
+ isStableCoin=is_stablecoin,
+ impermanentLossRisk=False, # Lending has no IL
+ )
+ result.append(pool)
+
+ return result

onchain/pools/ethereum/test_ethereum_protocols.py

@@ -0,0 +1,126 @@
+import unittest
+import os
+import boto3
+import asyncio
+
+from onchain.pools.ethereum.uniswap_v3_protocol import UniswapV3Protocol
+from onchain.pools.ethereum.aave_protocol import AaveProtocol
+from onchain.portfolio.ethereum_portfolio import EthereumPortfolioFetcher
+from onchain.tokens.metadata import TokenMetadataRepo
+from api.api_types import Chain, PoolType
+import dotenv
+
+dotenv.load_dotenv()
+
+
+class TestEthereumProtocols(unittest.TestCase):
+ def setUp(self):
+ dynamodb = boto3.resource(
+ "dynamodb",
+ region_name=os.environ.get("AWS_REGION"),
+ aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"),
+ aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"),
+ )
+ tokens_table = dynamodb.Table("token_metadata_v2")
+ self.token_metadata_repo = TokenMetadataRepo(tokens_table)
+
+ def test_uniswap_v3(self):
+ uniswap = UniswapV3Protocol()
+ pools = asyncio.run(uniswap.get_pools(self.token_metadata_repo))
+
+ self.assertGreater(len(pools), 5)
+ print(f"Uniswap V3 pools: {len(pools)}")
+
+ # Verify pool structure
+ for pool in pools[:3]:
+ self.assertEqual(pool.chain, Chain.ETHEREUM)
+ self.assertEqual(pool.type, PoolType.AMM)
+ self.assertIn("Uniswap V3", pool.protocol)
+ self.assertEqual(len(pool.tokens), 2)
+ self.assertIsNotNone(pool.TVL)
+ self.assertGreaterEqual(pool.APRLastDay, 0)
+
+ def test_uniswap_v3_stablecoin_detection(self):
+ uniswap = UniswapV3Protocol()
+ pools = asyncio.run(uniswap.get_pools(self.token_metadata_repo))
+
+ # Non-stablecoin pools should have IL risk
+ for pool in pools:
+ if not pool.isStableCoin:
+ self.assertTrue(pool.impermanentLossRisk)
+ else:
+ self.assertFalse(pool.impermanentLossRisk)
+
+ def test_aave_v3(self):
+ aave = AaveProtocol()
+ pools = asyncio.run(aave.get_pools(self.token_metadata_repo))
+
+ self.assertGreater(len(pools), 5)
+ print(f"Aave V3 reserves: {len(pools)}")
+
+ # Verify pool structure
+ for pool in pools[:3]:
+ self.assertEqual(pool.chain, Chain.ETHEREUM)
+ self.assertEqual(pool.type, PoolType.LENDING)
+ self.assertEqual(pool.protocol, "Aave V3")
+ self.assertEqual(len(pool.tokens), 1)
+ self.assertFalse(pool.impermanentLossRisk)
+ self.assertGreaterEqual(pool.APRLastDay, 0)
+
+ def test_aave_v3_tvl_is_usd(self):
+ """TVL should be in USD (not raw token units)."""
+ aave = AaveProtocol()
+ pools = asyncio.run(aave.get_pools(self.token_metadata_repo))
+
+ for pool in pools:
+ tvl = float(pool.TVL)
+ # Any active Aave reserve should have meaningful TVL in USD
+ if tvl > 0:
+ self.assertGreater(tvl, 100, f"{pool.tokens[0].symbol} TVL too low: {tvl}")
+
+
+class TestEthereumPortfolio(unittest.TestCase):
+ def setUp(self):
+ dynamodb = boto3.resource(
+ "dynamodb",
+ region_name=os.environ.get("AWS_REGION"),
+ aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"),
+ aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"),
+ )
+ tokens_table = dynamodb.Table("token_metadata_v2")
+ self.token_metadata_repo = TokenMetadataRepo(tokens_table)
+ self.portfolio_fetcher = EthereumPortfolioFetcher(self.token_metadata_repo)
+
+ def tearDown(self):
+ asyncio.run(self.portfolio_fetcher.close())
+
+ def test_get_portfolio(self):
+ # Vitalik's public wallet
+ portfolio = asyncio.run(
+ self.portfolio_fetcher.get_portfolio(
+ "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045"
+ )
+ )
+ print(portfolio)
+
+ self.assertGreater(len(portfolio.holdings), 0)
+ # Vitalik holds ETH
+ eth_holdings = [h for h in portfolio.holdings if h.symbol == "ETH"]
+ self.assertEqual(len(eth_holdings), 1)
+ self.assertGreater(eth_holdings[0].amount, 0)
+
+ def test_empty_wallet(self):
+ # Zero address should have no holdings
+ portfolio = asyncio.run(
+ self.portfolio_fetcher.get_portfolio(
+ "0x0000000000000000000000000000000000000001"
+ )
+ )
+ self.assertEqual(len(portfolio.holdings), 0)
+
+ def test_invalid_address(self):
+ portfolio = asyncio.run(
+ self.portfolio_fetcher.get_portfolio("not-a-valid-address")
+ )
+ self.assertEqual(len(portfolio.holdings), 0)
+ self.assertEqual(portfolio.total_value_usd, 0)

.env.example

@@ -3,6 +3,12 @@ export AWS_REGION=''
 export AWS_ACCESS_KEY_ID=''
 export AWS_SECRET_ACCESS_KEY=''
 export SOLANA_RPC_URL=''
+export ETH_RPC_URL='' # Ethereum RPC (Alchemy/Infura recommended)
+
+# The Graph subgraph URLs (optional — defaults are provided)
+# Production usage requires a Graph API key: https://thegraph.com/studio/
+export UNISWAP_V3_SUBGRAPH_URL=''
+export AAVE_V3_SUBGRAPH_URL=''
 export OPENROUTER_API_KEY=''
 export GEMINI_API_KEY=''
 export DD_API_KEY=''

agent/tools.py

@@ -44,7 +44,7 @@ async def retrieve_solana_pools(
 
  # Create a query to filter pools
  query = PoolQuery(
- chain=Chain.SOLANA, # Currently only supporting Solana
+ chain=Chain.SOLANA,
  tokens=tokens or [],
  user_tokens=user_tokens,
  )
@@ -56,9 +56,37 @@ async def retrieve_solana_pools(
  return pools
 
 
+@tool
+@track_tool_usage("retrieve_ethereum_pools")
+async def retrieve_ethereum_pools(
+ tokens: List[str] = None,
+ config: RunnableConfig = None,
+) -> List[Pool]:
+ """
+ Retrieves Ethereum pools matching the specified criteria that the user can invest in.
+ Includes Uniswap V3 AMM pools and Aave V3 lending pools.
+ """
+ configurable = config["configurable"]
+ user_tokens: List[WalletTokenHolding] = configurable["tokens"]
+ protocol_registry: ProtocolRegistry = configurable["protocol_registry"]
+
+ query = PoolQuery(
+ chain=Chain.ETHEREUM,
+ tokens=tokens or [],
+ user_tokens=user_tokens,
+ )
+
+ pools = await protocol_registry.get_pools(query)
+ if len(pools) == 0:
+ return "No Ethereum pools found."
+
+ return pools
+
+
 def create_investor_agent_toolkit() -> List[BaseTool]:
  return [
  retrieve_solana_pools,
+ retrieve_ethereum_pools,
  ]
 
 

main.py

@@ -19,12 +19,16 @@
 from onchain.pools.solana.orca_protocol import OrcaProtocol
 from onchain.pools.solana.save_protocol import SaveProtocol
 from onchain.pools.solana.kamino_protocol import KaminoProtocol
+from onchain.pools.ethereum.uniswap_v3_protocol import UniswapV3Protocol
+from onchain.pools.ethereum.aave_protocol import AaveProtocol
 
 # Define protocols enabled
 protocols = [
  OrcaProtocol.PROTOCOL_NAME,
  SaveProtocol.PROTOCOL_NAME,
  KaminoProtocol.PROTOCOL_NAME,
+ UniswapV3Protocol.PROTOCOL_NAME,
+ AaveProtocol.PROTOCOL_NAME,
 ]
 
 # Create the FastAPI app

onchain/pools/ethereum/aave_protocol.py

@@ -0,0 +1,162 @@
+"""Aave V3 protocol implementation for Ethereum lending pool discovery."""
+
+from typing import List, Optional, Dict, Any
+import logging
+import os
+
+import aiohttp
+
+from api.api_types import Pool, Token, Chain, PoolType
+from onchain.pools.protocol import Protocol
+from onchain.tokens.metadata import TokenMetadataRepo
+
+logger = logging.getLogger(__name__)
+
+# Aave V3 Ethereum subgraph — configurable via env var.
+AAVE_V3_SUBGRAPH_URL = os.environ.get(
+ "AAVE_V3_SUBGRAPH_URL",
+ "https://gateway.thegraph.com/api/subgraphs/id/Cd2gEDVeqnjBn1hSeqFMitw8Q1iiyV9FYUZkLNRcL87g",
+)
+
+# Stablecoins for classification
+STABLECOIN_SYMBOLS = {"USDC", "USDT", "DAI", "BUSD", "TUSD", "FRAX", "LUSD", "GUSD", "sUSD"}
+
+
+class AaveProtocol(Protocol):
+ """
+ Aave V3 protocol — fetches Ethereum lending reserves.
+
+ Returns each reserve as a lending Pool with supply APR.
+ Uses The Graph subgraph for on-chain data.
+ """
+
+ PROTOCOL_NAME = "aave-v3"
+
+ _session: Optional[aiohttp.ClientSession] = None
+
+ @property
+ def name(self) -> str:
+ return self.PROTOCOL_NAME
+
+ @property
+ async def session(self) -> aiohttp.ClientSession:
+ if self._session is None:
+ self._session = aiohttp.ClientSession(
+ timeout=aiohttp.ClientTimeout(total=30)
+ )
+ return self._session
+
+ async def close(self):
+ if self._session:
+ await self._session.close()
+ self._session = None
+
+ async def get_pools(self, token_metadata_repo: TokenMetadataRepo) -> List[Pool]:
+ """Fetch Aave V3 reserves from the subgraph."""
+ query = """
+ {
+ reserves(
+ first: 50,
+ where: { isActive: true }
+ ) {
+ id
+ symbol
+ name
+ decimals
+ underlyingAsset
+ liquidityRate
+ totalATokenSupply
+ totalCurrentVariableDebt
+ availableLiquidity
+ price {
+ priceInEth
+ }
+ }
+ }
+ """
+
+ try:
+ session = await self.session
+ async with session.post(
+ AAVE_V3_SUBGRAPH_URL,
+ json={"query": query},
+ ) as response:
+ if response.status in (401, 403):
+ logger.error(
+ f"Aave V3 subgraph auth failed ({response.status}). "
+ f"Set AAVE_V3_SUBGRAPH_URL with a valid Graph API key."
+ )
+ return []
+ if response.status != 200:
+ logger.error(f"Aave V3 subgraph returned {response.status}")
+ return []
+ data = await response.json()
+ except Exception as e:
+ logger.error(f"Error fetching Aave V3 reserves: {e}")
+ return []
+
+ reserves = data.get("data", {}).get("reserves", [])
+
+ # Resolve ETH/USD price for TVL conversion.
+ # Use the WETH token metadata from DexScreener (already cached).
+ eth_usd_price = 0.0
+ eth_meta = await token_metadata_repo.get_token_metadata(
+ "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2", "ethereum"
+ )
+ if eth_meta and eth_meta.price:
+ eth_usd_price = float(eth_meta.price)
+
+ return self._convert_to_pools(reserves, eth_usd_price)
+
+ def _convert_to_pools(
+ self, reserves: List[Dict[str, Any]], eth_usd_price: float
+ ) -> List[Pool]:
+ result: List[Pool] = []
+
+ for reserve in reserves:
+ symbol = reserve.get("symbol", "")
+ name = reserve.get("name", "")
+ address = reserve.get("underlyingAsset", "")
+ decimals = int(reserve.get("decimals", 18))
+
+ tokens = [
+ Token(address=address, name=name, symbol=symbol),
+ ]
+
+ # Aave liquidityRate is in RAY units (1e27), convert to APR percentage
+ liquidity_rate = int(reserve.get("liquidityRate", 0))
+ supply_apr = (liquidity_rate / 1e27) * 100
+
+ # Calculate TVL in USD:
+ # totalATokenSupply is in token-native units (needs /10^decimals)
+ # priceInEth is the token price denominated in ETH (wei-scaled, 1e18)
+ # Multiply by eth_usd_price to get USD
+ total_supply_raw = int(reserve.get("totalATokenSupply", 0))
+ total_supply = total_supply_raw / (10**decimals)
+
+ price_in_eth_raw = reserve.get("price", {}).get("priceInEth", "0")
+ price_in_eth = int(price_in_eth_raw) / 1e18
+
+ if eth_usd_price > 0 and price_in_eth > 0:
+ tvl_usd = str(round(total_supply * price_in_eth * eth_usd_price, 2))
+ else:
+ tvl_usd = str(round(total_supply, 2))
+
+ is_stablecoin = symbol.upper() in STABLECOIN_SYMBOLS
+
+ pool = Pool(
+ id=reserve.get("id", ""),
+ chain=Chain.ETHEREUM,
+ protocol="Aave V3",
+ tokens=tokens,
+ type=PoolType.LENDING,
+ TVL=tvl_usd,
+ APRLastDay=round(supply_apr, 2),
+ APRLastWeek=round(supply_apr, 2),
+ APRLastMonth=round(supply_apr, 2),
+ isStableCoin=is_stablecoin,
+ impermanentLossRisk=False, # Lending has no IL
+ )
+ result.append(pool)
+
+ return result

onchain/pools/ethereum/test_ethereum_protocols.py

@@ -0,0 +1,126 @@
+import unittest
+import os
+import boto3
+import asyncio
+
+from onchain.pools.ethereum.uniswap_v3_protocol import UniswapV3Protocol
+from onchain.pools.ethereum.aave_protocol import AaveProtocol
+from onchain.portfolio.ethereum_portfolio import EthereumPortfolioFetcher
+from onchain.tokens.metadata import TokenMetadataRepo
+from api.api_types import Chain, PoolType
+import dotenv
+
+dotenv.load_dotenv()
+
+
+class TestEthereumProtocols(unittest.TestCase):
+ def setUp(self):
+ dynamodb = boto3.resource(
+ "dynamodb",
+ region_name=os.environ.get("AWS_REGION"),
+ aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"),
+ aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"),
+ )
+ tokens_table = dynamodb.Table("token_metadata_v2")
+ self.token_metadata_repo = TokenMetadataRepo(tokens_table)
+
+ def test_uniswap_v3(self):
+ uniswap = UniswapV3Protocol()
+ pools = asyncio.run(uniswap.get_pools(self.token_metadata_repo))
+
+ self.assertGreater(len(pools), 5)
+ print(f"Uniswap V3 pools: {len(pools)}")
+
+ # Verify pool structure
+ for pool in pools[:3]:
+ self.assertEqual(pool.chain, Chain.ETHEREUM)
+ self.assertEqual(pool.type, PoolType.AMM)
+ self.assertIn("Uniswap V3", pool.protocol)
+ self.assertEqual(len(pool.tokens), 2)
+ self.assertIsNotNone(pool.TVL)
+ self.assertGreaterEqual(pool.APRLastDay, 0)
+
+ def test_uniswap_v3_stablecoin_detection(self):
+ uniswap = UniswapV3Protocol()
+ pools = asyncio.run(uniswap.get_pools(self.token_metadata_repo))
+
+ # Non-stablecoin pools should have IL risk
+ for pool in pools:
+ if not pool.isStableCoin:
+ self.assertTrue(pool.impermanentLossRisk)
+ else:
+ self.assertFalse(pool.impermanentLossRisk)
+
+ def test_aave_v3(self):
+ aave = AaveProtocol()
+ pools = asyncio.run(aave.get_pools(self.token_metadata_repo))
+
+ self.assertGreater(len(pools), 5)
+ print(f"Aave V3 reserves: {len(pools)}")
+
+ # Verify pool structure
+ for pool in pools[:3]:
+ self.assertEqual(pool.chain, Chain.ETHEREUM)
+ self.assertEqual(pool.type, PoolType.LENDING)
+ self.assertEqual(pool.protocol, "Aave V3")
+ self.assertEqual(len(pool.tokens), 1)
+ self.assertFalse(pool.impermanentLossRisk)
+ self.assertGreaterEqual(pool.APRLastDay, 0)
+
+ def test_aave_v3_tvl_is_usd(self):
+ """TVL should be in USD (not raw token units)."""
+ aave = AaveProtocol()
+ pools = asyncio.run(aave.get_pools(self.token_metadata_repo))
+
+ for pool in pools:
+ tvl = float(pool.TVL)
+ # Any active Aave reserve should have meaningful TVL in USD
+ if tvl > 0:
+ self.assertGreater(tvl, 100, f"{pool.tokens[0].symbol} TVL too low: {tvl}")
+
+
+class TestEthereumPortfolio(unittest.TestCase):
+ def setUp(self):
+ dynamodb = boto3.resource(
+ "dynamodb",
+ region_name=os.environ.get("AWS_REGION"),
+ aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"),
+ aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"),
+ )
+ tokens_table = dynamodb.Table("token_metadata_v2")
+ self.token_metadata_repo = TokenMetadataRepo(tokens_table)
+ self.portfolio_fetcher = EthereumPortfolioFetcher(self.token_metadata_repo)
+
+ def tearDown(self):
+ asyncio.run(self.portfolio_fetcher.close())
+
+ def test_get_portfolio(self):
+ # Vitalik's public wallet
+ portfolio = asyncio.run(
+ self.portfolio_fetcher.get_portfolio(
+ "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045"
+ )
+ )
+ print(portfolio)
+
+ self.assertGreater(len(portfolio.holdings), 0)
+ # Vitalik holds ETH
+ eth_holdings = [h for h in portfolio.holdings if h.symbol == "ETH"]
+ self.assertEqual(len(eth_holdings), 1)
+ self.assertGreater(eth_holdings[0].amount, 0)
+
+ def test_empty_wallet(self):
+ # Zero address should have no holdings
+ portfolio = asyncio.run(
+ self.portfolio_fetcher.get_portfolio(
+ "0x0000000000000000000000000000000000000001"
+ )
+ )
+ self.assertEqual(len(portfolio.holdings), 0)
+
+ def test_invalid_address(self):
+ portfolio = asyncio.run(
+ self.portfolio_fetcher.get_portfolio("not-a-valid-address")
+ )
+ self.assertEqual(len(portfolio.holdings), 0)
+ self.assertEqual(portfolio.total_value_usd, 0)

.env.example

@@ -3,6 +3,12 @@ export AWS_REGION=''
 export AWS_ACCESS_KEY_ID=''
 export AWS_SECRET_ACCESS_KEY=''
 export SOLANA_RPC_URL=''
+export ETH_RPC_URL='' # Ethereum RPC (Alchemy/Infura recommended)
+
+# The Graph subgraph URLs (optional — defaults are provided)
+# Production usage requires a Graph API key: https://thegraph.com/studio/
+export UNISWAP_V3_SUBGRAPH_URL=''
+export AAVE_V3_SUBGRAPH_URL=''
 export OPENROUTER_API_KEY=''
 export GEMINI_API_KEY=''
 export DD_API_KEY=''

agent/tools.py

@@ -44,7 +44,7 @@ async def retrieve_solana_pools(
 
  # Create a query to filter pools
  query = PoolQuery(
- chain=Chain.SOLANA, # Currently only supporting Solana
+ chain=Chain.SOLANA,
  tokens=tokens or [],
  user_tokens=user_tokens,
  )
@@ -56,9 +56,37 @@ async def retrieve_solana_pools(
  return pools
 
 
+@tool
+@track_tool_usage("retrieve_ethereum_pools")
+async def retrieve_ethereum_pools(
+ tokens: List[str] = None,
+ config: RunnableConfig = None,
+) -> List[Pool]:
+ """
+ Retrieves Ethereum pools matching the specified criteria that the user can invest in.
+ Includes Uniswap V3 AMM pools and Aave V3 lending pools.
+ """
+ configurable = config["configurable"]
+ user_tokens: List[WalletTokenHolding] = configurable["tokens"]
+ protocol_registry: ProtocolRegistry = configurable["protocol_registry"]
+
+ query = PoolQuery(
+ chain=Chain.ETHEREUM,
+ tokens=tokens or [],
+ user_tokens=user_tokens,
+ )
+
+ pools = await protocol_registry.get_pools(query)
+ if len(pools) == 0:
+ return "No Ethereum pools found."
+
+ return pools
+
+
 def create_investor_agent_toolkit() -> List[BaseTool]:
  return [
  retrieve_solana_pools,
+ retrieve_ethereum_pools,
  ]
 
 

main.py

@@ -19,12 +19,16 @@
 from onchain.pools.solana.orca_protocol import OrcaProtocol
 from onchain.pools.solana.save_protocol import SaveProtocol
 from onchain.pools.solana.kamino_protocol import KaminoProtocol
+from onchain.pools.ethereum.uniswap_v3_protocol import UniswapV3Protocol
+from onchain.pools.ethereum.aave_protocol import AaveProtocol
 
 # Define protocols enabled
 protocols = [
  OrcaProtocol.PROTOCOL_NAME,
  SaveProtocol.PROTOCOL_NAME,
  KaminoProtocol.PROTOCOL_NAME,
+ UniswapV3Protocol.PROTOCOL_NAME,
+ AaveProtocol.PROTOCOL_NAME,
 ]
 
 # Create the FastAPI app

onchain/pools/ethereum/aave_protocol.py

@@ -0,0 +1,162 @@
+"""Aave V3 protocol implementation for Ethereum lending pool discovery."""
+
+from typing import List, Optional, Dict, Any
+import logging
+import os
+
+import aiohttp
+
+from api.api_types import Pool, Token, Chain, PoolType
+from onchain.pools.protocol import Protocol
+from onchain.tokens.metadata import TokenMetadataRepo
+
+logger = logging.getLogger(__name__)
+
+# Aave V3 Ethereum subgraph — configurable via env var.
+AAVE_V3_SUBGRAPH_URL = os.environ.get(
+ "AAVE_V3_SUBGRAPH_URL",
+ "https://gateway.thegraph.com/api/subgraphs/id/Cd2gEDVeqnjBn1hSeqFMitw8Q1iiyV9FYUZkLNRcL87g",
+)
+
+# Stablecoins for classification
+STABLECOIN_SYMBOLS = {"USDC", "USDT", "DAI", "BUSD", "TUSD", "FRAX", "LUSD", "GUSD", "sUSD"}
+
+
+class AaveProtocol(Protocol):
+ """
+ Aave V3 protocol — fetches Ethereum lending reserves.
+
+ Returns each reserve as a lending Pool with supply APR.
+ Uses The Graph subgraph for on-chain data.
+ """
+
+ PROTOCOL_NAME = "aave-v3"
+
+ _session: Optional[aiohttp.ClientSession] = None
+
+ @property
+ def name(self) -> str:
+ return self.PROTOCOL_NAME
+
+ @property
+ async def session(self) -> aiohttp.ClientSession:
+ if self._session is None:
+ self._session = aiohttp.ClientSession(
+ timeout=aiohttp.ClientTimeout(total=30)
+ )
+ return self._session
+
+ async def close(self):
+ if self._session:
+ await self._session.close()
+ self._session = None
+
+ async def get_pools(self, token_metadata_repo: TokenMetadataRepo) -> List[Pool]:
+ """Fetch Aave V3 reserves from the subgraph."""
+ query = """
+ {
+ reserves(
+ first: 50,
+ where: { isActive: true }
+ ) {
+ id
+ symbol
+ name
+ decimals
+ underlyingAsset
+ liquidityRate
+ totalATokenSupply
+ totalCurrentVariableDebt
+ availableLiquidity
+ price {
+ priceInEth
+ }
+ }
+ }
+ """
+
+ try:
+ session = await self.session
+ async with session.post(
+ AAVE_V3_SUBGRAPH_URL,
+ json={"query": query},
+ ) as response:
+ if response.status in (401, 403):
+ logger.error(
+ f"Aave V3 subgraph auth failed ({response.status}). "
+ f"Set AAVE_V3_SUBGRAPH_URL with a valid Graph API key."
+ )
+ return []
+ if response.status != 200:
+ logger.error(f"Aave V3 subgraph returned {response.status}")
+ return []
+ data = await response.json()
+ except Exception as e:
+ logger.error(f"Error fetching Aave V3 reserves: {e}")
+ return []
+
+ reserves = data.get("data", {}).get("reserves", [])
+
+ # Resolve ETH/USD price for TVL conversion.
+ # Use the WETH token metadata from DexScreener (already cached).
+ eth_usd_price = 0.0
+ eth_meta = await token_metadata_repo.get_token_metadata(
+ "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2", "ethereum"
+ )
+ if eth_meta and eth_meta.price:
+ eth_usd_price = float(eth_meta.price)
+
+ return self._convert_to_pools(reserves, eth_usd_price)
+
+ def _convert_to_pools(
+ self, reserves: List[Dict[str, Any]], eth_usd_price: float
+ ) -> List[Pool]:
+ result: List[Pool] = []
+
+ for reserve in reserves:
+ symbol = reserve.get("symbol", "")
+ name = reserve.get("name", "")
+ address = reserve.get("underlyingAsset", "")
+ decimals = int(reserve.get("decimals", 18))
+
+ tokens = [
+ Token(address=address, name=name, symbol=symbol),
+ ]
+
+ # Aave liquidityRate is in RAY units (1e27), convert to APR percentage
+ liquidity_rate = int(reserve.get("liquidityRate", 0))
+ supply_apr = (liquidity_rate / 1e27) * 100
+
+ # Calculate TVL in USD:
+ # totalATokenSupply is in token-native units (needs /10^decimals)
+ # priceInEth is the token price denominated in ETH (wei-scaled, 1e18)
+ # Multiply by eth_usd_price to get USD
+ total_supply_raw = int(reserve.get("totalATokenSupply", 0))
+ total_supply = total_supply_raw / (10**decimals)
+
+ price_in_eth_raw = reserve.get("price", {}).get("priceInEth", "0")
+ price_in_eth = int(price_in_eth_raw) / 1e18
+
+ if eth_usd_price > 0 and price_in_eth > 0:
+ tvl_usd = str(round(total_supply * price_in_eth * eth_usd_price, 2))
+ else:
+ tvl_usd = str(round(total_supply, 2))
+
+ is_stablecoin = symbol.upper() in STABLECOIN_SYMBOLS
+
+ pool = Pool(
+ id=reserve.get("id", ""),
+ chain=Chain.ETHEREUM,
+ protocol="Aave V3",
+ tokens=tokens,
+ type=PoolType.LENDING,
+ TVL=tvl_usd,
+ APRLastDay=round(supply_apr, 2),
+ APRLastWeek=round(supply_apr, 2),
+ APRLastMonth=round(supply_apr, 2),
+ isStableCoin=is_stablecoin,
+ impermanentLossRisk=False, # Lending has no IL
+ )
+ result.append(pool)
+
+ return result

onchain/pools/ethereum/test_ethereum_protocols.py

@@ -0,0 +1,126 @@
+import unittest
+import os
+import boto3
+import asyncio
+
+from onchain.pools.ethereum.uniswap_v3_protocol import UniswapV3Protocol
+from onchain.pools.ethereum.aave_protocol import AaveProtocol
+from onchain.portfolio.ethereum_portfolio import EthereumPortfolioFetcher
+from onchain.tokens.metadata import TokenMetadataRepo
+from api.api_types import Chain, PoolType
+import dotenv
+
+dotenv.load_dotenv()
+
+
+class TestEthereumProtocols(unittest.TestCase):
+ def setUp(self):
+ dynamodb = boto3.resource(
+ "dynamodb",
+ region_name=os.environ.get("AWS_REGION"),
+ aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"),
+ aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"),
+ )
+ tokens_table = dynamodb.Table("token_metadata_v2")
+ self.token_metadata_repo = TokenMetadataRepo(tokens_table)
+
+ def test_uniswap_v3(self):
+ uniswap = UniswapV3Protocol()
+ pools = asyncio.run(uniswap.get_pools(self.token_metadata_repo))
+
+ self.assertGreater(len(pools), 5)
+ print(f"Uniswap V3 pools: {len(pools)}")
+
+ # Verify pool structure
+ for pool in pools[:3]:
+ self.assertEqual(pool.chain, Chain.ETHEREUM)
+ self.assertEqual(pool.type, PoolType.AMM)
+ self.assertIn("Uniswap V3", pool.protocol)
+ self.assertEqual(len(pool.tokens), 2)
+ self.assertIsNotNone(pool.TVL)
+ self.assertGreaterEqual(pool.APRLastDay, 0)
+
+ def test_uniswap_v3_stablecoin_detection(self):
+ uniswap = UniswapV3Protocol()
+ pools = asyncio.run(uniswap.get_pools(self.token_metadata_repo))
+
+ # Non-stablecoin pools should have IL risk
+ for pool in pools:
+ if not pool.isStableCoin:
+ self.assertTrue(pool.impermanentLossRisk)
+ else:
+ self.assertFalse(pool.impermanentLossRisk)
+
+ def test_aave_v3(self):
+ aave = AaveProtocol()
+ pools = asyncio.run(aave.get_pools(self.token_metadata_repo))
+
+ self.assertGreater(len(pools), 5)
+ print(f"Aave V3 reserves: {len(pools)}")
+
+ # Verify pool structure
+ for pool in pools[:3]:
+ self.assertEqual(pool.chain, Chain.ETHEREUM)
+ self.assertEqual(pool.type, PoolType.LENDING)
+ self.assertEqual(pool.protocol, "Aave V3")
+ self.assertEqual(len(pool.tokens), 1)
+ self.assertFalse(pool.impermanentLossRisk)
+ self.assertGreaterEqual(pool.APRLastDay, 0)
+
+ def test_aave_v3_tvl_is_usd(self):
+ """TVL should be in USD (not raw token units)."""
+ aave = AaveProtocol()
+ pools = asyncio.run(aave.get_pools(self.token_metadata_repo))
+
+ for pool in pools:
+ tvl = float(pool.TVL)
+ # Any active Aave reserve should have meaningful TVL in USD
+ if tvl > 0:
+ self.assertGreater(tvl, 100, f"{pool.tokens[0].symbol} TVL too low: {tvl}")
+
+
+class TestEthereumPortfolio(unittest.TestCase):
+ def setUp(self):
+ dynamodb = boto3.resource(
+ "dynamodb",
+ region_name=os.environ.get("AWS_REGION"),
+ aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"),
+ aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"),
+ )
+ tokens_table = dynamodb.Table("token_metadata_v2")
+ self.token_metadata_repo = TokenMetadataRepo(tokens_table)
+ self.portfolio_fetcher = EthereumPortfolioFetcher(self.token_metadata_repo)
+
+ def tearDown(self):
+ asyncio.run(self.portfolio_fetcher.close())
+
+ def test_get_portfolio(self):
+ # Vitalik's public wallet
+ portfolio = asyncio.run(
+ self.portfolio_fetcher.get_portfolio(
+ "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045"
+ )
+ )
+ print(portfolio)
+
+ self.assertGreater(len(portfolio.holdings), 0)
+ # Vitalik holds ETH
+ eth_holdings = [h for h in portfolio.holdings if h.symbol == "ETH"]
+ self.assertEqual(len(eth_holdings), 1)
+ self.assertGreater(eth_holdings[0].amount, 0)
+
+ def test_empty_wallet(self):
+ # Zero address should have no holdings
+ portfolio = asyncio.run(
+ self.portfolio_fetcher.get_portfolio(
+ "0x0000000000000000000000000000000000000001"
+ )
+ )
+ self.assertEqual(len(portfolio.holdings), 0)
+
+ def test_invalid_address(self):
+ portfolio = asyncio.run(
+ self.portfolio_fetcher.get_portfolio("not-a-valid-address")
+ )
+ self.assertEqual(len(portfolio.holdings), 0)
+ self.assertEqual(portfolio.total_value_usd, 0)

.env.example

@@ -3,6 +3,12 @@ export AWS_REGION=''
 export AWS_ACCESS_KEY_ID=''
 export AWS_SECRET_ACCESS_KEY=''
 export SOLANA_RPC_URL=''
+export ETH_RPC_URL='' # Ethereum RPC (Alchemy/Infura recommended)
+
+# The Graph subgraph URLs (optional — defaults are provided)
+# Production usage requires a Graph API key: https://thegraph.com/studio/
+export UNISWAP_V3_SUBGRAPH_URL=''
+export AAVE_V3_SUBGRAPH_URL=''
 export OPENROUTER_API_KEY=''
 export GEMINI_API_KEY=''
 export DD_API_KEY=''

agent/tools.py

@@ -44,7 +44,7 @@ async def retrieve_solana_pools(
 
  # Create a query to filter pools
  query = PoolQuery(
- chain=Chain.SOLANA, # Currently only supporting Solana
+ chain=Chain.SOLANA,
  tokens=tokens or [],
  user_tokens=user_tokens,
  )
@@ -56,9 +56,37 @@ async def retrieve_solana_pools(
  return pools
 
 
+@tool
+@track_tool_usage("retrieve_ethereum_pools")
+async def retrieve_ethereum_pools(
+ tokens: List[str] = None,
+ config: RunnableConfig = None,
+) -> List[Pool]:
+ """
+ Retrieves Ethereum pools matching the specified criteria that the user can invest in.
+ Includes Uniswap V3 AMM pools and Aave V3 lending pools.
+ """
+ configurable = config["configurable"]
+ user_tokens: List[WalletTokenHolding] = configurable["tokens"]
+ protocol_registry: ProtocolRegistry = configurable["protocol_registry"]
+
+ query = PoolQuery(
+ chain=Chain.ETHEREUM,
+ tokens=tokens or [],
+ user_tokens=user_tokens,
+ )
+
+ pools = await protocol_registry.get_pools(query)
+ if len(pools) == 0:
+ return "No Ethereum pools found."
+
+ return pools
+
+
 def create_investor_agent_toolkit() -> List[BaseTool]:
  return [
  retrieve_solana_pools,
+ retrieve_ethereum_pools,
  ]
 
 

main.py

@@ -19,12 +19,16 @@
 from onchain.pools.solana.orca_protocol import OrcaProtocol
 from onchain.pools.solana.save_protocol import SaveProtocol
 from onchain.pools.solana.kamino_protocol import KaminoProtocol
+from onchain.pools.ethereum.uniswap_v3_protocol import UniswapV3Protocol
+from onchain.pools.ethereum.aave_protocol import AaveProtocol
 
 # Define protocols enabled
 protocols = [
  OrcaProtocol.PROTOCOL_NAME,
  SaveProtocol.PROTOCOL_NAME,
  KaminoProtocol.PROTOCOL_NAME,
+ UniswapV3Protocol.PROTOCOL_NAME,
+ AaveProtocol.PROTOCOL_NAME,
 ]
 
 # Create the FastAPI app

onchain/pools/ethereum/aave_protocol.py

@@ -0,0 +1,162 @@
+"""Aave V3 protocol implementation for Ethereum lending pool discovery."""
+
+from typing import List, Optional, Dict, Any
+import logging
+import os
+
+import aiohttp
+
+from api.api_types import Pool, Token, Chain, PoolType
+from onchain.pools.protocol import Protocol
+from onchain.tokens.metadata import TokenMetadataRepo
+
+logger = logging.getLogger(__name__)
+
+# Aave V3 Ethereum subgraph — configurable via env var.
+AAVE_V3_SUBGRAPH_URL = os.environ.get(
+ "AAVE_V3_SUBGRAPH_URL",
+ "https://gateway.thegraph.com/api/subgraphs/id/Cd2gEDVeqnjBn1hSeqFMitw8Q1iiyV9FYUZkLNRcL87g",
+)
+
+# Stablecoins for classification
+STABLECOIN_SYMBOLS = {"USDC", "USDT", "DAI", "BUSD", "TUSD", "FRAX", "LUSD", "GUSD", "sUSD"}
+
+
+class AaveProtocol(Protocol):
+ """
+ Aave V3 protocol — fetches Ethereum lending reserves.
+
+ Returns each reserve as a lending Pool with supply APR.
+ Uses The Graph subgraph for on-chain data.
+ """
+
+ PROTOCOL_NAME = "aave-v3"
+
+ _session: Optional[aiohttp.ClientSession] = None
+
+ @property
+ def name(self) -> str:
+ return self.PROTOCOL_NAME
+
+ @property
+ async def session(self) -> aiohttp.ClientSession:
+ if self._session is None:
+ self._session = aiohttp.ClientSession(
+ timeout=aiohttp.ClientTimeout(total=30)
+ )
+ return self._session
+
+ async def close(self):
+ if self._session:
+ await self._session.close()
+ self._session = None
+
+ async def get_pools(self, token_metadata_repo: TokenMetadataRepo) -> List[Pool]:
+ """Fetch Aave V3 reserves from the subgraph."""
+ query = """
+ {
+ reserves(
+ first: 50,
+ where: { isActive: true }
+ ) {
+ id
+ symbol
+ name
+ decimals
+ underlyingAsset
+ liquidityRate
+ totalATokenSupply
+ totalCurrentVariableDebt
+ availableLiquidity
+ price {
+ priceInEth
+ }
+ }
+ }
+ """
+
+ try:
+ session = await self.session
+ async with session.post(
+ AAVE_V3_SUBGRAPH_URL,
+ json={"query": query},
+ ) as response:
+ if response.status in (401, 403):
+ logger.error(
+ f"Aave V3 subgraph auth failed ({response.status}). "
+ f"Set AAVE_V3_SUBGRAPH_URL with a valid Graph API key."
+ )
+ return []
+ if response.status != 200:
+ logger.error(f"Aave V3 subgraph returned {response.status}")
+ return []
+ data = await response.json()
+ except Exception as e:
+ logger.error(f"Error fetching Aave V3 reserves: {e}")
+ return []
+
+ reserves = data.get("data", {}).get("reserves", [])
+
+ # Resolve ETH/USD price for TVL conversion.
+ # Use the WETH token metadata from DexScreener (already cached).
+ eth_usd_price = 0.0
+ eth_meta = await token_metadata_repo.get_token_metadata(
+ "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2", "ethereum"
+ )
+ if eth_meta and eth_meta.price:
+ eth_usd_price = float(eth_meta.price)
+
+ return self._convert_to_pools(reserves, eth_usd_price)
+
+ def _convert_to_pools(
+ self, reserves: List[Dict[str, Any]], eth_usd_price: float
+ ) -> List[Pool]:
+ result: List[Pool] = []
+
+ for reserve in reserves:
+ symbol = reserve.get("symbol", "")
+ name = reserve.get("name", "")
+ address = reserve.get("underlyingAsset", "")
+ decimals = int(reserve.get("decimals", 18))
+
+ tokens = [
+ Token(address=address, name=name, symbol=symbol),
+ ]
+
+ # Aave liquidityRate is in RAY units (1e27), convert to APR percentage
+ liquidity_rate = int(reserve.get("liquidityRate", 0))
+ supply_apr = (liquidity_rate / 1e27) * 100
+
+ # Calculate TVL in USD:
+ # totalATokenSupply is in token-native units (needs /10^decimals)
+ # priceInEth is the token price denominated in ETH (wei-scaled, 1e18)
+ # Multiply by eth_usd_price to get USD
+ total_supply_raw = int(reserve.get("totalATokenSupply", 0))
+ total_supply = total_supply_raw / (10**decimals)
+
+ price_in_eth_raw = reserve.get("price", {}).get("priceInEth", "0")
+ price_in_eth = int(price_in_eth_raw) / 1e18
+
+ if eth_usd_price > 0 and price_in_eth > 0:
+ tvl_usd = str(round(total_supply * price_in_eth * eth_usd_price, 2))
+ else:
+ tvl_usd = str(round(total_supply, 2))
+
+ is_stablecoin = symbol.upper() in STABLECOIN_SYMBOLS
+
+ pool = Pool(
+ id=reserve.get("id", ""),
+ chain=Chain.ETHEREUM,
+ protocol="Aave V3",
+ tokens=tokens,
+ type=PoolType.LENDING,
+ TVL=tvl_usd,
+ APRLastDay=round(supply_apr, 2),
+ APRLastWeek=round(supply_apr, 2),
+ APRLastMonth=round(supply_apr, 2),
+ isStableCoin=is_stablecoin,
+ impermanentLossRisk=False, # Lending has no IL
+ )
+ result.append(pool)
+
+ return result

onchain/pools/ethereum/test_ethereum_protocols.py

@@ -0,0 +1,126 @@
+import unittest
+import os
+import boto3
+import asyncio
+
+from onchain.pools.ethereum.uniswap_v3_protocol import UniswapV3Protocol
+from onchain.pools.ethereum.aave_protocol import AaveProtocol
+from onchain.portfolio.ethereum_portfolio import EthereumPortfolioFetcher
+from onchain.tokens.metadata import TokenMetadataRepo
+from api.api_types import Chain, PoolType
+import dotenv
+
+dotenv.load_dotenv()
+
+
+class TestEthereumProtocols(unittest.TestCase):
+ def setUp(self):
+ dynamodb = boto3.resource(
+ "dynamodb",
+ region_name=os.environ.get("AWS_REGION"),
+ aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"),
+ aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"),
+ )
+ tokens_table = dynamodb.Table("token_metadata_v2")
+ self.token_metadata_repo = TokenMetadataRepo(tokens_table)
+
+ def test_uniswap_v3(self):
+ uniswap = UniswapV3Protocol()
+ pools = asyncio.run(uniswap.get_pools(self.token_metadata_repo))
+
+ self.assertGreater(len(pools), 5)
+ print(f"Uniswap V3 pools: {len(pools)}")
+
+ # Verify pool structure
+ for pool in pools[:3]:
+ self.assertEqual(pool.chain, Chain.ETHEREUM)
+ self.assertEqual(pool.type, PoolType.AMM)
+ self.assertIn("Uniswap V3", pool.protocol)
+ self.assertEqual(len(pool.tokens), 2)
+ self.assertIsNotNone(pool.TVL)
+ self.assertGreaterEqual(pool.APRLastDay, 0)
+
+ def test_uniswap_v3_stablecoin_detection(self):
+ uniswap = UniswapV3Protocol()
+ pools = asyncio.run(uniswap.get_pools(self.token_metadata_repo))
+
+ # Non-stablecoin pools should have IL risk
+ for pool in pools:
+ if not pool.isStableCoin:
+ self.assertTrue(pool.impermanentLossRisk)
+ else:
+ self.assertFalse(pool.impermanentLossRisk)
+
+ def test_aave_v3(self):
+ aave = AaveProtocol()
+ pools = asyncio.run(aave.get_pools(self.token_metadata_repo))
+
+ self.assertGreater(len(pools), 5)
+ print(f"Aave V3 reserves: {len(pools)}")
+
+ # Verify pool structure
+ for pool in pools[:3]:
+ self.assertEqual(pool.chain, Chain.ETHEREUM)
+ self.assertEqual(pool.type, PoolType.LENDING)
+ self.assertEqual(pool.protocol, "Aave V3")
+ self.assertEqual(len(pool.tokens), 1)
+ self.assertFalse(pool.impermanentLossRisk)
+ self.assertGreaterEqual(pool.APRLastDay, 0)
+
+ def test_aave_v3_tvl_is_usd(self):
+ """TVL should be in USD (not raw token units)."""
+ aave = AaveProtocol()
+ pools = asyncio.run(aave.get_pools(self.token_metadata_repo))
+
+ for pool in pools:
+ tvl = float(pool.TVL)
+ # Any active Aave reserve should have meaningful TVL in USD
+ if tvl > 0:
+ self.assertGreater(tvl, 100, f"{pool.tokens[0].symbol} TVL too low: {tvl}")
+
+
+class TestEthereumPortfolio(unittest.TestCase):
+ def setUp(self):
+ dynamodb = boto3.resource(
+ "dynamodb",
+ region_name=os.environ.get("AWS_REGION"),
+ aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"),
+ aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"),
+ )
+ tokens_table = dynamodb.Table("token_metadata_v2")
+ self.token_metadata_repo = TokenMetadataRepo(tokens_table)
+ self.portfolio_fetcher = EthereumPortfolioFetcher(self.token_metadata_repo)
+
+ def tearDown(self):
+ asyncio.run(self.portfolio_fetcher.close())
+
+ def test_get_portfolio(self):
+ # Vitalik's public wallet
+ portfolio = asyncio.run(
+ self.portfolio_fetcher.get_portfolio(
+ "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045"
+ )
+ )
+ print(portfolio)
+
+ self.assertGreater(len(portfolio.holdings), 0)
+ # Vitalik holds ETH
+ eth_holdings = [h for h in portfolio.holdings if h.symbol == "ETH"]
+ self.assertEqual(len(eth_holdings), 1)
+ self.assertGreater(eth_holdings[0].amount, 0)
+
+ def test_empty_wallet(self):
+ # Zero address should have no holdings
+ portfolio = asyncio.run(
+ self.portfolio_fetcher.get_portfolio(
+ "0x0000000000000000000000000000000000000001"
+ )
+ )
+ self.assertEqual(len(portfolio.holdings), 0)
+
+ def test_invalid_address(self):
+ portfolio = asyncio.run(
+ self.portfolio_fetcher.get_portfolio("not-a-valid-address")
+ )
+ self.assertEqual(len(portfolio.holdings), 0)
+ self.assertEqual(portfolio.total_value_usd, 0)

.env.example

@@ -3,6 +3,12 @@ export AWS_REGION=''
 export AWS_ACCESS_KEY_ID=''
 export AWS_SECRET_ACCESS_KEY=''
 export SOLANA_RPC_URL=''
+export ETH_RPC_URL='' # Ethereum RPC (Alchemy/Infura recommended)
+
+# The Graph subgraph URLs (optional — defaults are provided)
+# Production usage requires a Graph API key: https://thegraph.com/studio/
+export UNISWAP_V3_SUBGRAPH_URL=''
+export AAVE_V3_SUBGRAPH_URL=''
 export OPENROUTER_API_KEY=''
 export GEMINI_API_KEY=''
 export DD_API_KEY=''

agent/tools.py

@@ -44,7 +44,7 @@ async def retrieve_solana_pools(
 
  # Create a query to filter pools
  query = PoolQuery(
- chain=Chain.SOLANA, # Currently only supporting Solana
+ chain=Chain.SOLANA,
  tokens=tokens or [],
  user_tokens=user_tokens,
  )
@@ -56,9 +56,37 @@ async def retrieve_solana_pools(
  return pools
 
 
+@tool
+@track_tool_usage("retrieve_ethereum_pools")
+async def retrieve_ethereum_pools(
+ tokens: List[str] = None,
+ config: RunnableConfig = None,
+) -> List[Pool]:
+ """
+ Retrieves Ethereum pools matching the specified criteria that the user can invest in.
+ Includes Uniswap V3 AMM pools and Aave V3 lending pools.
+ """
+ configurable = config["configurable"]
+ user_tokens: List[WalletTokenHolding] = configurable["tokens"]
+ protocol_registry: ProtocolRegistry = configurable["protocol_registry"]
+
+ query = PoolQuery(
+ chain=Chain.ETHEREUM,
+ tokens=tokens or [],
+ user_tokens=user_tokens,
+ )
+
+ pools = await protocol_registry.get_pools(query)
+ if len(pools) == 0:
+ return "No Ethereum pools found."
+
+ return pools
+
+
 def create_investor_agent_toolkit() -> List[BaseTool]:
  return [
  retrieve_solana_pools,
+ retrieve_ethereum_pools,
  ]
 
 

main.py

@@ -19,12 +19,16 @@
 from onchain.pools.solana.orca_protocol import OrcaProtocol
 from onchain.pools.solana.save_protocol import SaveProtocol
 from onchain.pools.solana.kamino_protocol import KaminoProtocol
+from onchain.pools.ethereum.uniswap_v3_protocol import UniswapV3Protocol
+from onchain.pools.ethereum.aave_protocol import AaveProtocol
 
 # Define protocols enabled
 protocols = [
  OrcaProtocol.PROTOCOL_NAME,
  SaveProtocol.PROTOCOL_NAME,
  KaminoProtocol.PROTOCOL_NAME,
+ UniswapV3Protocol.PROTOCOL_NAME,
+ AaveProtocol.PROTOCOL_NAME,
 ]
 
 # Create the FastAPI app

onchain/pools/ethereum/aave_protocol.py

@@ -0,0 +1,162 @@
+"""Aave V3 protocol implementation for Ethereum lending pool discovery."""
+
+from typing import List, Optional, Dict, Any
+import logging
+import os
+
+import aiohttp
+
+from api.api_types import Pool, Token, Chain, PoolType
+from onchain.pools.protocol import Protocol
+from onchain.tokens.metadata import TokenMetadataRepo
+
+logger = logging.getLogger(__name__)
+
+# Aave V3 Ethereum subgraph — configurable via env var.
+AAVE_V3_SUBGRAPH_URL = os.environ.get(
+ "AAVE_V3_SUBGRAPH_URL",
+ "https://gateway.thegraph.com/api/subgraphs/id/Cd2gEDVeqnjBn1hSeqFMitw8Q1iiyV9FYUZkLNRcL87g",
+)
+
+# Stablecoins for classification
+STABLECOIN_SYMBOLS = {"USDC", "USDT", "DAI", "BUSD", "TUSD", "FRAX", "LUSD", "GUSD", "sUSD"}
+
+
+class AaveProtocol(Protocol):
+ """
+ Aave V3 protocol — fetches Ethereum lending reserves.
+
+ Returns each reserve as a lending Pool with supply APR.
+ Uses The Graph subgraph for on-chain data.
+ """
+
+ PROTOCOL_NAME = "aave-v3"
+
+ _session: Optional[aiohttp.ClientSession] = None
+
+ @property
+ def name(self) -> str:
+ return self.PROTOCOL_NAME
+
+ @property
+ async def session(self) -> aiohttp.ClientSession:
+ if self._session is None:
+ self._session = aiohttp.ClientSession(
+ timeout=aiohttp.ClientTimeout(total=30)
+ )
+ return self._session
+
+ async def close(self):
+ if self._session:
+ await self._session.close()
+ self._session = None
+
+ async def get_pools(self, token_metadata_repo: TokenMetadataRepo) -> List[Pool]:
+ """Fetch Aave V3 reserves from the subgraph."""
+ query = """
+ {
+ reserves(
+ first: 50,
+ where: { isActive: true }
+ ) {
+ id
+ symbol
+ name
+ decimals
+ underlyingAsset
+ liquidityRate
+ totalATokenSupply
+ totalCurrentVariableDebt
+ availableLiquidity
+ price {
+ priceInEth
+ }
+ }
+ }
+ """
+
+ try:
+ session = await self.session
+ async with session.post(
+ AAVE_V3_SUBGRAPH_URL,
+ json={"query": query},
+ ) as response:
+ if response.status in (401, 403):
+ logger.error(
+ f"Aave V3 subgraph auth failed ({response.status}). "
+ f"Set AAVE_V3_SUBGRAPH_URL with a valid Graph API key."
+ )
+ return []
+ if response.status != 200:
+ logger.error(f"Aave V3 subgraph returned {response.status}")
+ return []
+ data = await response.json()
+ except Exception as e:
+ logger.error(f"Error fetching Aave V3 reserves: {e}")
+ return []
+
+ reserves = data.get("data", {}).get("reserves", [])
+
+ # Resolve ETH/USD price for TVL conversion.
+ # Use the WETH token metadata from DexScreener (already cached).
+ eth_usd_price = 0.0
+ eth_meta = await token_metadata_repo.get_token_metadata(
+ "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2", "ethereum"
+ )
+ if eth_meta and eth_meta.price:
+ eth_usd_price = float(eth_meta.price)
+
+ return self._convert_to_pools(reserves, eth_usd_price)
+
+ def _convert_to_pools(
+ self, reserves: List[Dict[str, Any]], eth_usd_price: float
+ ) -> List[Pool]:
+ result: List[Pool] = []
+
+ for reserve in reserves:
+ symbol = reserve.get("symbol", "")
+ name = reserve.get("name", "")
+ address = reserve.get("underlyingAsset", "")
+ decimals = int(reserve.get("decimals", 18))
+
+ tokens = [
+ Token(address=address, name=name, symbol=symbol),
+ ]
+
+ # Aave liquidityRate is in RAY units (1e27), convert to APR percentage
+ liquidity_rate = int(reserve.get("liquidityRate", 0))
+ supply_apr = (liquidity_rate / 1e27) * 100
+
+ # Calculate TVL in USD:
+ # totalATokenSupply is in token-native units (needs /10^decimals)
+ # priceInEth is the token price denominated in ETH (wei-scaled, 1e18)
+ # Multiply by eth_usd_price to get USD
+ total_supply_raw = int(reserve.get("totalATokenSupply", 0))
+ total_supply = total_supply_raw / (10**decimals)
+
+ price_in_eth_raw = reserve.get("price", {}).get("priceInEth", "0")
+ price_in_eth = int(price_in_eth_raw) / 1e18
+
+ if eth_usd_price > 0 and price_in_eth > 0:
+ tvl_usd = str(round(total_supply * price_in_eth * eth_usd_price, 2))
+ else:
+ tvl_usd = str(round(total_supply, 2))
+
+ is_stablecoin = symbol.upper() in STABLECOIN_SYMBOLS
+
+ pool = Pool(
+ id=reserve.get("id", ""),
+ chain=Chain.ETHEREUM,
+ protocol="Aave V3",
+ tokens=tokens,
+ type=PoolType.LENDING,
+ TVL=tvl_usd,
+ APRLastDay=round(supply_apr, 2),
+ APRLastWeek=round(supply_apr, 2),
+ APRLastMonth=round(supply_apr, 2),
+ isStableCoin=is_stablecoin,
+ impermanentLossRisk=False, # Lending has no IL
+ )
+ result.append(pool)
+
+ return result

onchain/pools/ethereum/test_ethereum_protocols.py

@@ -0,0 +1,126 @@
+import unittest
+import os
+import boto3
+import asyncio
+
+from onchain.pools.ethereum.uniswap_v3_protocol import UniswapV3Protocol
+from onchain.pools.ethereum.aave_protocol import AaveProtocol
+from onchain.portfolio.ethereum_portfolio import EthereumPortfolioFetcher
+from onchain.tokens.metadata import TokenMetadataRepo
+from api.api_types import Chain, PoolType
+import dotenv
+
+dotenv.load_dotenv()
+
+
+class TestEthereumProtocols(unittest.TestCase):
+ def setUp(self):
+ dynamodb = boto3.resource(
+ "dynamodb",
+ region_name=os.environ.get("AWS_REGION"),
+ aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"),
+ aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"),
+ )
+ tokens_table = dynamodb.Table("token_metadata_v2")
+ self.token_metadata_repo = TokenMetadataRepo(tokens_table)
+
+ def test_uniswap_v3(self):
+ uniswap = UniswapV3Protocol()
+ pools = asyncio.run(uniswap.get_pools(self.token_metadata_repo))
+
+ self.assertGreater(len(pools), 5)
+ print(f"Uniswap V3 pools: {len(pools)}")
+
+ # Verify pool structure
+ for pool in pools[:3]:
+ self.assertEqual(pool.chain, Chain.ETHEREUM)
+ self.assertEqual(pool.type, PoolType.AMM)
+ self.assertIn("Uniswap V3", pool.protocol)
+ self.assertEqual(len(pool.tokens), 2)
+ self.assertIsNotNone(pool.TVL)
+ self.assertGreaterEqual(pool.APRLastDay, 0)
+
+ def test_uniswap_v3_stablecoin_detection(self):
+ uniswap = UniswapV3Protocol()
+ pools = asyncio.run(uniswap.get_pools(self.token_metadata_repo))
+
+ # Non-stablecoin pools should have IL risk
+ for pool in pools:
+ if not pool.isStableCoin:
+ self.assertTrue(pool.impermanentLossRisk)
+ else:
+ self.assertFalse(pool.impermanentLossRisk)
+
+ def test_aave_v3(self):
+ aave = AaveProtocol()
+ pools = asyncio.run(aave.get_pools(self.token_metadata_repo))
+
+ self.assertGreater(len(pools), 5)
+ print(f"Aave V3 reserves: {len(pools)}")
+
+ # Verify pool structure
+ for pool in pools[:3]:
+ self.assertEqual(pool.chain, Chain.ETHEREUM)
+ self.assertEqual(pool.type, PoolType.LENDING)
+ self.assertEqual(pool.protocol, "Aave V3")
+ self.assertEqual(len(pool.tokens), 1)
+ self.assertFalse(pool.impermanentLossRisk)
+ self.assertGreaterEqual(pool.APRLastDay, 0)
+
+ def test_aave_v3_tvl_is_usd(self):
+ """TVL should be in USD (not raw token units)."""
+ aave = AaveProtocol()
+ pools = asyncio.run(aave.get_pools(self.token_metadata_repo))
+
+ for pool in pools:
+ tvl = float(pool.TVL)
+ # Any active Aave reserve should have meaningful TVL in USD
+ if tvl > 0:
+ self.assertGreater(tvl, 100, f"{pool.tokens[0].symbol} TVL too low: {tvl}")
+
+
+class TestEthereumPortfolio(unittest.TestCase):
+ def setUp(self):
+ dynamodb = boto3.resource(
+ "dynamodb",
+ region_name=os.environ.get("AWS_REGION"),
+ aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"),
+ aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"),
+ )
+ tokens_table = dynamodb.Table("token_metadata_v2")
+ self.token_metadata_repo = TokenMetadataRepo(tokens_table)
+ self.portfolio_fetcher = EthereumPortfolioFetcher(self.token_metadata_repo)
+
+ def tearDown(self):
+ asyncio.run(self.portfolio_fetcher.close())
+
+ def test_get_portfolio(self):
+ # Vitalik's public wallet
+ portfolio = asyncio.run(
+ self.portfolio_fetcher.get_portfolio(
+ "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045"
+ )
+ )
+ print(portfolio)
+
+ self.assertGreater(len(portfolio.holdings), 0)
+ # Vitalik holds ETH
+ eth_holdings = [h for h in portfolio.holdings if h.symbol == "ETH"]
+ self.assertEqual(len(eth_holdings), 1)
+ self.assertGreater(eth_holdings[0].amount, 0)
+
+ def test_empty_wallet(self):
+ # Zero address should have no holdings
+ portfolio = asyncio.run(
+ self.portfolio_fetcher.get_portfolio(
+ "0x0000000000000000000000000000000000000001"
+ )
+ )
+ self.assertEqual(len(portfolio.holdings), 0)
+
+ def test_invalid_address(self):
+ portfolio = asyncio.run(
+ self.portfolio_fetcher.get_portfolio("not-a-valid-address")
+ )
+ self.assertEqual(len(portfolio.holdings), 0)
+ self.assertEqual(portfolio.total_value_usd, 0)