""" Data Pipeline — Real-time yield, price, and macro data aggregation =================================================================== Fetches data from DeFiLlama, CoinGecko, Bybit, Mantle RPC, and FRED to construct the RL agent's observation state vector. """ import asyncio import logging import time from dataclasses import dataclass, field from datetime import datetime, timedelta from typing import Any, Dict, List, Optional, Tuple import aiohttp import numpy as np from config.constants import ( APIEndpoints, MANTLE_RPC_URL, Tokens, LendingProtocols, ERC20_ABI, AAVE_POOL_ABI, METH_STAKING_ABI, ) logger = logging.getLogger("data_pipeline") @dataclass class YieldSnapshot: """Point-in-time yield data for all portfolio assets.""" timestamp: float # Core yields (annualized %) usdy_apy: float = 0.0 # Ondo USDY — tokenized T-bills meth_apy: float = 0.0 # mETH staking yield mi4_apy: float = 0.0 # MI4 index fund yield # Lending yields (additional yield layer) aave_usdy_supply_apy: float = 0.0 aave_meth_supply_apy: float = 0.0 # Prices (USD) eth_price: float = 0.0 mnt_price: float = 0.0 btc_price: float = 0.0 # Peg health (1.0 = perfect peg) usdy_peg: float = 1.0 # USDY/USD meth_peg: float = 1.0 # mETH/ETH ratio vs fair value # Macro fed_funds_rate: float = 5.25 # Federal funds rate % btc_dominance: float = 50.0 # BTC dominance % # Volatility (30d rolling) eth_volatility_30d: float = 0.0 usdy_volatility_30d: float = 0.0 # On-chain gas_price_gwei: float = 0.0 def to_state_vector(self) -> np.ndarray: """Convert snapshot to normalized state vector for RL agent.""" return np.array([ self.usdy_apy / 10.0, # normalize to ~0-1 self.meth_apy / 10.0, self.mi4_apy / 10.0, self.eth_price / 10000.0, # normalize ETH price self.btc_price / 100000.0, # normalize BTC price self.btc_dominance / 100.0, self.fed_funds_rate / 10.0, self.usdy_peg, self.meth_peg, self.eth_volatility_30d, self.usdy_volatility_30d, self.aave_usdy_supply_apy / 10.0, self.aave_meth_supply_apy / 10.0, self.gas_price_gwei / 100.0, self.mnt_price / 10.0, ], dtype=np.float32) @dataclass class HistoricalData: """Historical price/yield series for backtesting and volatility calculation.""" timestamps: List[float] = field(default_factory=list) eth_prices: List[float] = field(default_factory=list) btc_prices: List[float] = field(default_factory=list) usdy_yields: List[float] = field(default_factory=list) meth_yields: List[float] = field(default_factory=list) mi4_navs: List[float] = field(default_factory=list) class DataPipeline: """ Aggregates real-time and historical data from multiple sources. Data Sources: - DeFiLlama: TVL, protocol yields, pool APYs - CoinGecko/Bybit: Asset prices, market data - Mantle RPC: On-chain state (exchange rates, gas, reserves) - FRED: Federal funds rate, macro indicators Design: All external calls are async with timeout + retry + fallback. """ def __init__( self, rpc_url: str = MANTLE_RPC_URL, cache_ttl_seconds: int = 60, request_timeout: int = 15, ): self.rpc_url = rpc_url self.cache_ttl = cache_ttl_seconds self.timeout = aiohttp.ClientTimeout(total=request_timeout) self._cache: Dict[str, Tuple[float, Any]] = {} self._session: Optional[aiohttp.ClientSession] = None async def _get_session(self) -> aiohttp.ClientSession: if self._session is None or self._session.closed: self._session = aiohttp.ClientSession(timeout=self.timeout) return self._session async def close(self): if self._session and not self._session.closed: await self._session.close() def _cache_get(self, key: str) -> Optional[Any]: if key in self._cache: ts, val = self._cache[key] if time.time() - ts < self.cache_ttl: return val return None def _cache_set(self, key: str, value: Any): self._cache[key] = (time.time(), value) # ─────────────────── Price Data ─────────────────── async def fetch_prices(self) -> Dict[str, float]: """Fetch current prices for ETH, BTC, MNT from CoinGecko.""" cached = self._cache_get("prices") if cached: return cached session = await self._get_session() try: params = { "ids": "ethereum,bitcoin,mantle", "vs_currencies": "usd", } async with session.get(APIEndpoints.COINGECKO_PRICE, params=params) as resp: if resp.status == 200: data = await resp.json() prices = { "ETH": data.get("ethereum", {}).get("usd", 0), "BTC": data.get("bitcoin", {}).get("usd", 0), "MNT": data.get("mantle", {}).get("usd", 0), } self._cache_set("prices", prices) return prices except Exception as e: logger.warning(f"CoinGecko price fetch failed: {e}, trying Bybit fallback") # Fallback: Bybit return await self._fetch_prices_bybit() async def _fetch_prices_bybit(self) -> Dict[str, float]: """Bybit fallback for price data.""" session = await self._get_session() prices = {} for symbol, key in [("ETHUSDT", "ETH"), ("BTCUSDT", "BTC"), ("MNTUSDT", "MNT")]: try: params = {"category": "spot", "symbol": symbol} async with session.get(APIEndpoints.BYBIT_TICKER, params=params) as resp: if resp.status == 200: data = await resp.json() result = data.get("result", {}).get("list", [{}])[0] prices[key] = float(result.get("lastPrice", 0)) except Exception as e: logger.warning(f"Bybit {symbol} fetch failed: {e}") prices[key] = 0.0 self._cache_set("prices", prices) return prices # ─────────────────── Yield Data ─────────────────── async def fetch_defillama_yields(self) -> Dict[str, float]: """Fetch yield data from DeFiLlama pools API.""" cached = self._cache_get("defi_yields") if cached: return cached session = await self._get_session() yields = {"usdy": 4.25, "meth": 3.5, "mi4": 5.0} # defaults try: async with session.get(APIEndpoints.DEFILLAMA_YIELDS) as resp: if resp.status == 200: data = await resp.json() pools = data.get("data", []) for pool in pools: chain = pool.get("chain", "").lower() symbol = pool.get("symbol", "").upper() project = pool.get("project", "").lower() apy = pool.get("apy", 0) or pool.get("apyBase", 0) or 0 # Match USDY on any chain (Ondo is cross-chain) if "usdy" in symbol.lower() and apy > 0: yields["usdy"] = apy # Match mETH on Mantle if ("meth" in symbol.lower() and chain == "mantle") and apy > 0: yields["meth"] = apy # MI4 index if "mi4" in symbol.lower() and apy > 0: yields["mi4"] = apy except Exception as e: logger.warning(f"DeFiLlama yields fetch failed: {e}, using defaults") self._cache_set("defi_yields", yields) return yields async def fetch_meth_exchange_rate(self) -> float: """Fetch mETH/ETH exchange rate from Mantle RPC.""" cached = self._cache_get("meth_rate") if cached: return cached session = await self._get_session() try: # Call mETHToETH() on mETH staking contract # mETH on Mantle: 0xcDA86A272531e8640cD7F1a92c01839911B90bb0 payload = { "jsonrpc": "2.0", "method": "eth_call", "params": [ { "to": Tokens.METH, "data": "0x1f1fcd51", # mETHToETH() selector }, "latest", ], "id": 1, } async with session.post(self.rpc_url, json=payload) as resp: if resp.status == 200: data = await resp.json() result = data.get("result", "0x0") rate = int(result, 16) / 1e18 if result != "0x0" else 1.0 self._cache_set("meth_rate", rate) return rate except Exception as e: logger.warning(f"mETH exchange rate fetch failed: {e}") return 1.0 # assume 1:1 as fallback async def fetch_aave_yields(self) -> Dict[str, float]: """Fetch Aave V3 supply APYs for USDY and mETH on Mantle.""" cached = self._cache_get("aave_yields") if cached: return cached session = await self._get_session() yields = {"usdy_supply": 0.0, "meth_supply": 0.0} for asset_name, asset_addr in [("usdy_supply", Tokens.USDY), ("meth_supply", Tokens.METH)]: try: # getReserveData(address) selector = "0x35ea6a75" padded_addr = asset_addr[2:].lower().zfill(64) payload = { "jsonrpc": "2.0", "method": "eth_call", "params": [ { "to": LendingProtocols.AAVE_V3_POOL, "data": f"0x35ea6a75{padded_addr}", }, "latest", ], "id": 1, } async with session.post(self.rpc_url, json=payload) as resp: if resp.status == 200: data = await resp.json() result = data.get("result", "0x") if len(result) > 130: # currentLiquidityRate is at offset 64 (2nd field), 128 bits # It's in RAY (1e27) per second, need to annualize rate_hex = result[66:130] rate_ray = int(rate_hex, 16) if rate_hex else 0 apy = (rate_ray / 1e27) * 100 # rough APY yields[asset_name] = apy except Exception as e: logger.warning(f"Aave yield fetch for {asset_name} failed: {e}") self._cache_set("aave_yields", yields) return yields # ─────────────────── Macro Data ─────────────────── async def fetch_fed_rate(self) -> float: """Fetch current federal funds rate from FRED.""" cached = self._cache_get("fed_rate") if cached: return cached fred_key = "DEMO_KEY" # Use env var in production session = await self._get_session() try: params = { "series_id": "FEDFUNDS", "api_key": fred_key, "file_type": "json", "sort_order": "desc", "limit": 1, } async with session.get(APIEndpoints.FRED_API, params=params) as resp: if resp.status == 200: data = await resp.json() obs = data.get("observations", []) if obs: rate = float(obs[0].get("value", 5.25)) self._cache_set("fed_rate", rate) return rate except Exception as e: logger.warning(f"FRED rate fetch failed: {e}") return 5.25 # fallback async def fetch_btc_dominance(self) -> float: """Fetch BTC dominance from CoinGecko global data.""" cached = self._cache_get("btc_dom") if cached: return cached session = await self._get_session() try: url = "https://api.coingecko.com/api/v3/global" async with session.get(url) as resp: if resp.status == 200: data = await resp.json() dom = data.get("data", {}).get("market_cap_percentage", {}).get("btc", 50.0) self._cache_set("btc_dom", dom) return dom except Exception as e: logger.warning(f"BTC dominance fetch failed: {e}") return 50.0 # ─────────────────── Volatility ─────────────────── async def fetch_price_history( self, coin_id: str = "ethereum", days: int = 30 ) -> List[float]: """Fetch historical daily prices from CoinGecko.""" session = await self._get_session() try: url = APIEndpoints.COINGECKO_HISTORY.format(coin_id=coin_id) params = {"vs_currency": "usd", "days": days, "interval": "daily"} async with session.get(url, params=params) as resp: if resp.status == 200: data = await resp.json() prices = [p[1] for p in data.get("prices", [])] return prices except Exception as e: logger.warning(f"Price history fetch failed for {coin_id}: {e}") return [] def compute_volatility(self, prices: List[float]) -> float: """Compute annualized volatility from daily prices.""" if len(prices) < 2: return 0.0 returns = np.diff(np.log(np.array(prices))) return float(np.std(returns) * np.sqrt(365)) # ─────────────────── Gas Price ─────────────────── async def fetch_gas_price(self) -> float: """Fetch current Mantle gas price in Gwei.""" session = await self._get_session() try: payload = { "jsonrpc": "2.0", "method": "eth_gasPrice", "params": [], "id": 1, } async with session.post(self.rpc_url, json=payload) as resp: if resp.status == 200: data = await resp.json() gas_wei = int(data.get("result", "0x0"), 16) return gas_wei / 1e9 # Convert to Gwei except Exception as e: logger.warning(f"Gas price fetch failed: {e}") return 0.02 # Mantle default ~0.02 Gwei # ─────────────────── Full Snapshot ─────────────────── async def get_snapshot(self) -> YieldSnapshot: """ Aggregate all data sources into a single YieldSnapshot. All fetches run concurrently for minimal latency. """ # Run all fetches concurrently prices_task = self.fetch_prices() yields_task = self.fetch_defillama_yields() meth_rate_task = self.fetch_meth_exchange_rate() aave_task = self.fetch_aave_yields() fed_task = self.fetch_fed_rate() btc_dom_task = self.fetch_btc_dominance() gas_task = self.fetch_gas_price() eth_hist_task = self.fetch_price_history("ethereum", 30) results = await asyncio.gather( prices_task, yields_task, meth_rate_task, aave_task, fed_task, btc_dom_task, gas_task, eth_hist_task, return_exceptions=True, ) prices = results[0] if not isinstance(results[0], Exception) else {"ETH": 0, "BTC": 0, "MNT": 0} yields = results[1] if not isinstance(results[1], Exception) else {"usdy": 4.25, "meth": 3.5, "mi4": 5.0} meth_rate = results[2] if not isinstance(results[2], Exception) else 1.0 aave_yields = results[3] if not isinstance(results[3], Exception) else {"usdy_supply": 0, "meth_supply": 0} fed_rate = results[4] if not isinstance(results[4], Exception) else 5.25 btc_dom = results[5] if not isinstance(results[5], Exception) else 50.0 gas_price = results[6] if not isinstance(results[6], Exception) else 0.02 eth_hist = results[7] if not isinstance(results[7], Exception) else [] eth_vol = self.compute_volatility(eth_hist) if eth_hist else 0.5 snapshot = YieldSnapshot( timestamp=time.time(), usdy_apy=yields.get("usdy", 4.25), meth_apy=yields.get("meth", 3.5), mi4_apy=yields.get("mi4", 5.0), aave_usdy_supply_apy=aave_yields.get("usdy_supply", 0.0), aave_meth_supply_apy=aave_yields.get("meth_supply", 0.0), eth_price=prices.get("ETH", 0), mnt_price=prices.get("MNT", 0), btc_price=prices.get("BTC", 0), usdy_peg=1.0, # refined in risk_manager via oracle check meth_peg=min(meth_rate / 1.0, 1.05), # cap at 1.05 for normalization fed_funds_rate=fed_rate, btc_dominance=btc_dom, eth_volatility_30d=eth_vol, usdy_volatility_30d=0.001, # USDY has near-zero vol gas_price_gwei=gas_price, ) logger.info( f"Snapshot: USDY={snapshot.usdy_apy:.2f}% mETH={snapshot.meth_apy:.2f}% " f"MI4={snapshot.mi4_apy:.2f}% ETH=${snapshot.eth_price:.0f} " f"Fed={snapshot.fed_funds_rate:.2f}% Gas={snapshot.gas_price_gwei:.4f}gwei" ) return snapshot # ─────────────────── Historical Data for Backtesting ─────────────── async def get_historical_data(self, days: int = 365) -> HistoricalData: """Fetch historical data for backtesting the RL environment.""" eth_prices = await self.fetch_price_history("ethereum", days) btc_prices = await self.fetch_price_history("bitcoin", days) # Generate synthetic yield histories based on known ranges # In production, these come from indexed on-chain data n = len(eth_prices) np.random.seed(42) hist = HistoricalData( timestamps=list(np.linspace(time.time() - days * 86400, time.time(), n)), eth_prices=eth_prices, btc_prices=btc_prices if btc_prices else [60000] * n, # USDY yield tracks Fed rate with small noise usdy_yields=list(np.clip(np.random.normal(4.25, 0.3, n), 3.5, 5.5)), # mETH yield varies with staking demand meth_yields=list(np.clip(np.random.normal(3.8, 0.8, n), 2.0, 8.0)), # MI4 NAV follows a diversified index mi4_navs=list(100 * np.cumprod(1 + np.random.normal(0.0002, 0.005, n))), ) return hist