feat: add Ethereum chain support

- ETH portfolio fetcher with Alchemy API + web3.py fallback
- Uniswap V3 protocol (top pools by TVL, fee-based APR)
- Aave V3 protocol (lending reserves, USD-denominated TVL)
- Token metadata for ETH ecosystem (WETH, USDC, USDT, DAI)
- Configurable subgraph URLs via env vars
- Integration tests for ETH protocols and portfolio

Author: Ridwannurudeen

.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/__init__.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/aave_protocol.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)