# { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } """Price source fetchers for 3 on-chain + 2 web-api feeds. Each fetcher returns a raw observation dict compatible with oracle_sentinel.validate_observation(). On failure, returns an observation with status=UNAVAILABLE — never guesses data. Sources: 1. Chainlink — eth_call latestRoundData() via public RPC 2. Pyth — eth_call getPriceUnsafe() via public RPC 3. Uniswap V3 — eth_call slot0() via public RPC 4. CoinGecko — GET /api/v3/simple/price (free, rate-limited) 5. CoinMarketCap — GET /public-api/v1/simple/price (keyless) """ from __future__ import annotations import json import time import urllib.request from typing import Any, Dict, List, Optional try: from . import asset_registry except ImportError: try: import asset_registry except ImportError: # GenVM single-file: sibling unavailable; contract methods don't need it asset_registry = None try: from genlayer import * except ImportError: # local pytest/demo: stub so helpers stay importable off-chain u256 = int Address = str class _GlPublic: @staticmethod def view(fn): return fn @staticmethod def write(fn): return fn class _Gl: Contract = object public = _GlPublic() gl = _Gl() # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- DEFAULT_RPC_URL = "https://ethereum-rpc.publicnode.com" COINGECKO_BASE = "https://api.coingecko.com/api/v3" CMC_BASE = "https://pro-api.coinmarketcap.com/public-api" # Fallback RPC if primary is unavailable RPC_FALLBACKS = [ "https://1rpc.io/eth", ] RPC_TIMEOUT = 15 # seconds per RPC call HTTP_TIMEOUT = 15 # seconds per HTTP call MAX_RETRIES = 2 RETRY_DELAY = 1.0 # seconds between retries # --------------------------------------------------------------------------- # Low-level helpers # --------------------------------------------------------------------------- def _rpc_call( method: str, params: list, rpc_url: str = DEFAULT_RPC_URL, timeout: int = RPC_TIMEOUT, ) -> Any: """Make a JSON-RPC call with bounded retries (AGENTS.md #408-#409). Retries transient failures (HTTP 429/5xx, connection errors) up to MAX_RETRIES with linear backoff. Never retries indefinitely. Returns result or raises. """ import urllib.error payload = json.dumps({ "jsonrpc": "2.0", "id": 1, "method": method, "params": params, }).encode("utf-8") headers = { "Content-Type": "application/json", "User-Agent": "OracleSentinel/1.0", } last_error: Optional[Exception] = None for attempt in range(MAX_RETRIES + 1): try: req = urllib.request.Request(rpc_url, data=payload, headers=headers) resp = urllib.request.urlopen(req, timeout=timeout) data = json.loads(resp.read()) if "error" in data: raise RuntimeError(f"RPC error: {data['error']}") return data["result"] except urllib.error.HTTPError as e: last_error = e if e.code not in (429, 500, 502, 503, 504): raise except Exception as e: # URLError, timeouts: transient, retry last_error = e if attempt < MAX_RETRIES: time.sleep(RETRY_DELAY * (attempt + 1)) assert last_error is not None raise last_error def _eth_call(to: str, data_hex: str, rpc_url: str = DEFAULT_RPC_URL) -> bytes: """eth_call returning raw bytes.""" result = _rpc_call("eth_call", [{"to": to, "data": data_hex}, "latest"], rpc_url) return bytes.fromhex(result[2:]) if result.startswith("0x") else bytes.fromhex(result) def _http_get(url: str, timeout: int = HTTP_TIMEOUT) -> Dict[str, Any]: """HTTP GET with bounded retries, returning parsed JSON.""" import urllib.error last_error: Optional[Exception] = None for attempt in range(MAX_RETRIES + 1): try: req = urllib.request.Request(url, headers={ "Accept": "application/json", "User-Agent": "OracleSentinel/1.0", }) resp = urllib.request.urlopen(req, timeout=timeout) return json.loads(resp.read()) except urllib.error.HTTPError as e: last_error = e if e.code not in (429, 500, 502, 503, 504): raise except Exception as e: last_error = e if attempt < MAX_RETRIES: time.sleep(RETRY_DELAY * (attempt + 1)) assert last_error is not None raise last_error def _unavailable( source_id: str, source_group: str, asset_pair: str, note: str ) -> Dict[str, Any]: """Build an unavailable observation (never guesses data).""" return { "source": source_id, "sourceGroup": source_group, "price": None, "timestamp": None, "asset": asset_pair, "status": "UNAVAILABLE", "note": note, } # --------------------------------------------------------------------------- # Source 1: Chainlink (on-chain) # --------------------------------------------------------------------------- # latestRoundData() selector: 0xfeaf968c CHAINLINK_SELECTOR = "0xfeaf968c" def fetch_chainlink( asset_cfg: asset_registry.AssetConfig, rpc_url: str = DEFAULT_RPC_URL, current_time: int = 0, ) -> Dict[str, Any]: """Read Chainlink AggregatorV3 latestRoundData() via RPC. Returns a raw observation dict with price in USD (int, scaled to match other sources) and timestamp. The contract returns answer * 10^decimals; we normalize to integer USD price (e.g. 2426 for $2,426). Returns status=UNAVAILABLE on any failure (RPC error, feed revert, etc). """ source_id = "chainlink" source_group = "chainlink" asset_pair = asset_cfg.pair if not asset_cfg.chainlink_feed: return _unavailable(source_id, source_group, asset_pair, "no chainlink feed configured") try: raw = _eth_call(asset_cfg.chainlink_feed, CHAINLINK_SELECTOR, rpc_url) if len(raw) < 160: return _unavailable(source_id, source_group, asset_pair, "short response") # latestRoundData returns (roundId, answer, startedAt, updatedAt, answeredInRound) answer = int.from_bytes(raw[32:64], "big", signed=True) updated_at = int.from_bytes(raw[96:128], "big") if answer <= 0: return _unavailable(source_id, source_group, asset_pair, "non-positive answer") # Normalize: Chainlink USD feeds use 8 decimals (answer / 1e8 = USD price) decimals = asset_cfg.chainlink_decimals price = answer // (10 ** decimals) if price <= 0: return _unavailable(source_id, source_group, asset_pair, "price rounds to zero") now = current_time if current_time > 0 else int(time.time()) if updated_at <= 0 or updated_at > now + 60: return _unavailable(source_id, source_group, asset_pair, "invalid timestamp") return { "source": source_id, "sourceGroup": source_group, "price": price, "timestamp": updated_at, "asset": asset_pair, } except Exception as e: return _unavailable(source_id, source_group, asset_pair, f"chainlink error: {e}") # --------------------------------------------------------------------------- # Source 2: Pyth (on-chain) # --------------------------------------------------------------------------- # getPriceUnsafe(bytes32) selector: compute from keccak256 # We precompute it here to avoid import-time side effects. def _pyth_selector() -> str: """Compute 4-byte selector for getPriceUnsafe(bytes32).""" from Crypto.Hash import keccak as _keccak h = _keccak.new(digest_bits=256) h.update(b"getPriceUnsafe(bytes32)") return "0x" + h.hexdigest()[:8] PYTH_SELECTOR = "0x00000000" try: PYTH_SELECTOR = _pyth_selector() except Exception: # GenVM: no Crypto — Pyth source reports UNAVAILABLE at runtime PYTH_SELECTOR = "0x00000000" def fetch_pyth( asset_cfg: asset_registry.AssetConfig, rpc_url: str = DEFAULT_RPC_URL, current_time: int = 0, ) -> Dict[str, Any]: """Read Pyth getPriceUnsafe(bytes32) via RPC. Returns a raw observation dict with integer USD price derived from the Pyth Price struct: price * 10^expo. The Pyth contract on Ethereum mainnet is at 0x4305FB66699C3B2702D4d05CF36551390A4c69C6. Price struct ABI encoding (each field padded to 32 bytes): - price: int64 at offset 0-31 - conf: uint64 at offset 32-63 - expo: int32 at offset 64-95 - publishTime: uint64 at offset 96-127 Returns status=UNAVAILABLE if no Pyth feed is configured or call fails. """ source_id = "pyth" source_group = "pyth" asset_pair = asset_cfg.pair if not asset_cfg.pyth_price_id: return _unavailable(source_id, source_group, asset_pair, "no pyth feed configured") if PYTH_SELECTOR == "0x00000000": return _unavailable(source_id, source_group, asset_pair, "pyth selector unavailable") try: # Encode call: getPriceUnsafe(bytes32 priceId) feed_id = asset_cfg.pyth_price_id.lower().replace("0x", "") call_data = PYTH_SELECTOR + feed_id.zfill(64) raw = _eth_call(asset_registry.PYTH_CONTRACT, call_data, rpc_url) if len(raw) < 128: return _unavailable(source_id, source_group, asset_pair, "short pyth response") # Decode Price struct price_val = int.from_bytes(raw[0:32], "big", signed=True) expo_val = int.from_bytes(raw[64:96], "big", signed=True) publish_time = int.from_bytes(raw[96:128], "big") if price_val <= 0: return _unavailable(source_id, source_group, asset_pair, "non-positive price") # Actual price = price * 10^expo. Convert to integer USD. # e.g. price=240700000000, expo=-8 → 2407 USD actual_price = price_val * (10 ** expo_val) int_price = int(actual_price) if int_price <= 0: return _unavailable(source_id, source_group, asset_pair, "price rounds to zero after expo") now = current_time if current_time > 0 else int(time.time()) if publish_time <= 0 or publish_time > now + 60: return _unavailable(source_id, source_group, asset_pair, "invalid publish time") return { "source": source_id, "sourceGroup": source_group, "price": int_price, "timestamp": publish_time, "asset": asset_pair, } except Exception as e: return _unavailable(source_id, source_group, asset_pair, f"pyth error: {e}") # --------------------------------------------------------------------------- # Source 3: Uniswap V3 (on-chain) # --------------------------------------------------------------------------- # slot0() selector: 0x3850c7bd UNISWAP_V3_SLOT0_SELECTOR = "0x3850c7bd" def fetch_uniswap_v3( asset_cfg: asset_registry.AssetConfig, rpc_url: str = DEFAULT_RPC_URL, current_time: int = 0, ) -> Dict[str, Any]: """Read Uniswap V3 pool slot0() via RPC. Returns a raw observation dict with integer USD price derived from sqrtPriceX96. Float math is acceptable here: this fetcher runs off-chain (scripts/GenLayer IC host) and consensus operates on the resulting ints; no on-chain fixed-point constraints apply. Token ordering matters: price_raw = token1/token0 in raw units, so the decimal scaling is applied in the correct direction via asset_cfg.v3_token0_is_asset. Wrong-direction scaling would silently produce garbage prices, hence the explicit flag (never guess ordering). Returns status=UNAVAILABLE if pool address is missing or call fails. """ source_id = "uniswap_v3" source_group = "uniswap_v3" asset_pair = asset_cfg.pair if not asset_cfg.v3_pool: return _unavailable(source_id, source_group, asset_pair, "no v3 pool configured") try: raw = _eth_call(asset_cfg.v3_pool, UNISWAP_V3_SLOT0_SELECTOR, rpc_url) if len(raw) < 64: return _unavailable(source_id, source_group, asset_pair, "short slot0 response") # slot0 returns (sqrtPriceX96, tick, observationIndex, ...) sqrt_price_x96 = int.from_bytes(raw[0:32], "big", signed=False) if sqrt_price_x96 <= 0: return _unavailable(source_id, source_group, asset_pair, "zero sqrtPriceX96") # price_raw = (sqrtPriceX96 / 2^96)^2 = token1/token0 in raw units. ratio = (sqrt_price_x96 / (2 ** 96)) ** 2 if not (ratio > 0) or ratio in (float("inf"),): return _unavailable(source_id, source_group, asset_pair, "non-finite ratio") # Scale raw ratio to USD: multiply by 10^(asset_dec - quote_dec) in # the direction matching token ordering. scale = 10 ** (asset_cfg.token_decimals - asset_cfg.quote_decimals) if asset_cfg.v3_token0_is_asset: # token0 = asset, token1 = quote: usd = raw * 10^(tok-q) usd = ratio * scale else: # token0 = quote, token1 = asset: usd = 10^(tok-q) / raw usd = scale / ratio price = int(usd) if price <= 0: return _unavailable(source_id, source_group, asset_pair, "price rounds to zero") # Use current time as timestamp (slot0 has no timestamp field). now = current_time if current_time > 0 else int(time.time()) return { "source": source_id, "sourceGroup": source_group, "price": int(price), "timestamp": now, "asset": asset_pair, } except Exception as e: return _unavailable(source_id, source_group, asset_pair, f"v3 error: {e}") # --------------------------------------------------------------------------- # Source 4: CoinGecko (web API) # --------------------------------------------------------------------------- def fetch_coingecko( asset_cfg: asset_registry.AssetConfig, current_time: int = 0, ) -> Dict[str, Any]: """Fetch price from CoinGecko free API. Returns a raw observation dict with integer USD price. Returns status=UNAVAILABLE on failure or rate limiting. """ source_id = "coingecko" source_group = "coingecko" asset_pair = asset_cfg.pair try: url = ( f"{COINGECKO_BASE}/simple/price" f"?ids={asset_cfg.coingecko_id}" f"&vs_currencies=usd" f"&include_last_updated_at=true" ) data = _http_get(url) coin_data = data.get(asset_cfg.coingecko_id) if not coin_data or "usd" not in coin_data: return _unavailable(source_id, source_group, asset_pair, "missing price data") price_raw = coin_data["usd"] updated_at = coin_data.get("last_updated_at", 0) # Normalize to integer price (e.g. 2426 for $2,426) price = int(price_raw) if price <= 0: return _unavailable(source_id, source_group, asset_pair, "non-positive price") now = current_time if current_time > 0 else int(time.time()) if updated_at <= 0: updated_at = now return { "source": source_id, "sourceGroup": source_group, "price": price, "timestamp": int(updated_at), "asset": asset_pair, } except Exception as e: return _unavailable(source_id, source_group, asset_pair, f"coingecko error: {e}") # --------------------------------------------------------------------------- # Source 5: CoinMarketCap (web API, keyless) # --------------------------------------------------------------------------- def fetch_coinmarketcap( asset_cfg: asset_registry.AssetConfig, current_time: int = 0, ) -> Dict[str, Any]: """Fetch price from CoinMarketCap keyless public API. Uses https://pro-api.coinmarketcap.com/public-api/v1/simple/price which requires NO API key (keyless tier). Returns a raw observation dict with integer USD price. Returns status=UNAVAILABLE on failure or rate limiting. """ source_id = "coinmarketcap" source_group = "coinmarketcap" asset_pair = asset_cfg.pair try: url = ( f"{CMC_BASE}/v1/simple/price" f"?ids={asset_cfg.cmc_id}" f"&convert=USD" ) data = _http_get(url) # CMC keyless response: {"data":[{"id":,"price":}], "status":{...}} data_list = data.get("data", []) if isinstance(data_list, list): coin_data = next((d for d in data_list if d.get("id") == asset_cfg.cmc_id), {}) else: coin_data = data_list.get(str(asset_cfg.cmc_id), {}) price_raw = coin_data.get("price") if price_raw is None: return _unavailable(source_id, source_group, asset_pair, "missing price data") price = int(float(price_raw)) if price <= 0: return _unavailable(source_id, source_group, asset_pair, "non-positive price") now = current_time if current_time > 0 else int(time.time()) # CMC keyless response doesn't include last_updated; use current time updated_at = now return { "source": source_id, "sourceGroup": source_group, "price": price, "timestamp": updated_at, "asset": asset_pair, } except Exception as e: return _unavailable(source_id, source_group, asset_pair, f"coinmarketcap error: {e}") # --------------------------------------------------------------------------- # Unified collection # --------------------------------------------------------------------------- def collect_all_sources( asset_cfg: asset_registry.AssetConfig, rpc_url: str = DEFAULT_RPC_URL, current_time: int = 0, ) -> List[Dict[str, Any]]: """Collect observations from all 5 sources for one asset. Returns list of 0-5 raw observation dicts. """ now = current_time if current_time > 0 else int(time.time()) observations = [] # On-chain sources (3). Small pacing between calls: free RPC tiers # rate-limit bursts (HTTP 429); retries in _rpc_call cover the rest. observations.append(fetch_chainlink(asset_cfg, rpc_url, now)) time.sleep(1.0) observations.append(fetch_pyth(asset_cfg, rpc_url, now)) time.sleep(1.0) observations.append(fetch_uniswap_v3(asset_cfg, rpc_url, now)) # Web-api sources (2) observations.append(fetch_coingecko(asset_cfg, now)) observations.append(fetch_coinmarketcap(asset_cfg, now)) return observations # --------------------------------------------------------------------------- # On-chain contract (GenLayer Studio entrypoint for THIS file) # --------------------------------------------------------------------------- # Observations arrive via guardian-submitted transactions (fail-closed # validation: price/timestamp must be > 0). Live network fetching belongs in a # gl.nondet.web variant; urllib has no sockets in GenVM, so the fetchers above # stay off-chain. class PriceSources(gl.Contract): admin: Address last_source: str last_price: u256 last_timestamp: u256 def __init__(self, admin: Address): self.admin = admin self.last_source = "none" self.last_price = u256(0) self.last_timestamp = u256(0) @gl.public.view def get_admin(self) -> str: return str(self.admin) @gl.public.view def get_last(self) -> str: if int(self.last_price) <= 0: return "none" return ( self.last_source + "|" + str(int(self.last_price)) + "|" + str(int(self.last_timestamp)) ) @gl.public.write def submit(self, source: str, price: u256, timestamp: u256) -> str: if int(price) <= 0 or int(timestamp) <= 0: return "INVALID" self.last_source = source self.last_price = price self.last_timestamp = timestamp return "VALID"

Hash	Type	Status		From	To	Method	GenVM Result	Consensus Result	Time
0x35f049a5...59abb5cc	Call	FINALIZED	IN	0x04e035...25DB1b	0x5a160C...285178	submit	SUCCESS	Accepted	4 days ago
0x2b545e35...6aae3a54	Deploy	FINALIZED	IN	0x04e035...25DB1b	0x5a160C...285178	(constructor)	SUCCESS	Accepted	4 days ago