#!/usr/bin/env python3 """ Machine-readable dump of cW* tokens vs USD-like PMM quotes from deployment-status.json. Reads cross-chain-pmm-lps/config/deployment-status.json, finds pools where base == symbol and quote is USDC/USDT/cWUSDC/cWUSDT, then calls getMidPrice and getVaultReserve on-chain. Usage: python3 scripts/lib/dump_cw_usd_quotes.py [--output PATH] Requires: cast (foundry), RPC URLs in smom-dbis-138/.env (or env already exported). """ from __future__ import annotations import argparse import json import os import subprocess import sys from datetime import datetime, timezone from decimal import Decimal from pathlib import Path ROOT = Path(__file__).resolve().parents[2] DEPLOYMENT_STATUS = ROOT / "cross-chain-pmm-lps" / "config" / "deployment-status.json" DEFAULT_ENV = ROOT / "smom-dbis-138" / ".env" DEFAULT_OUT = ROOT / "output" / "cw-assets-usd-quote-dump.json" CHAIN_RPC_ENV = { "1": "ETHEREUM_MAINNET_RPC", "10": "OPTIMISM_MAINNET_RPC", "25": "CRONOS_RPC", "56": "BSC_MAINNET_RPC", "100": "GNOSIS_MAINNET_RPC", "137": "POLYGON_MAINNET_RPC", "42161": "ARBITRUM_MAINNET_RPC", "42220": "CELO_MAINNET_RPC", "43114": "AVALANCHE_MAINNET_RPC", "8453": "BASE_MAINNET_RPC", "138": "RPC_URL_138", } USD_LIKE = frozenset({"USDC", "USDT", "cWUSDC", "cWUSDT"}) def load_dotenv(path: Path) -> dict[str, str]: out: dict[str, str] = {} if not path.is_file(): return out for line in path.read_text().splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue k, v = line.split("=", 1) out[k] = v return out def cast(env: dict[str, str], rpc: str, args: list[str], timeout: float = 12.0) -> tuple[str, int]: try: r = subprocess.run( ["cast", *args, "--rpc-url", rpc], capture_output=True, text=True, timeout=timeout, env={**os.environ, **env}, ) return (r.stdout or "").strip(), r.returncode except Exception as e: return str(e), -1 def parse_u256(s: str) -> int | None: if not s: return None t = s.split()[0].strip() if "[" in t: t = t.split("[")[0] if t.startswith("0x"): return int(t, 16) return int(float(t)) if "e" in t.lower() else int(t) def find_usd_pool(ch: dict, sym: str) -> tuple[str | None, str | None]: for src in ("pmmPools", "pmmPoolsVolatile"): for p in ch.get(src) or []: if p.get("base") != sym: continue q = p.get("quote") or "" if q in USD_LIKE: return p.get("poolAddress"), q return None, None def main() -> int: ap = argparse.ArgumentParser() ap.add_argument( "--output", "-o", type=Path, default=DEFAULT_OUT, help=f"JSON output path (default: {DEFAULT_OUT})", ) ap.add_argument( "--env-file", type=Path, default=DEFAULT_ENV, help="Dotenv with RPC URLs", ) ap.add_argument( "--deployment-status", type=Path, default=DEPLOYMENT_STATUS, ) args = ap.parse_args() env = {**os.environ, **load_dotenv(args.env_file)} ds = json.loads(args.deployment_status.read_text()) dec_cache: dict[tuple[str, str], int] = {} def decimals(addr: str, rpc: str) -> int: key = (rpc[:32], addr.lower()) if key in dec_cache: return dec_cache[key] out, code = cast(env, rpc, ["call", addr, "decimals()(uint8)"]) d = int(parse_u256(out)) if code == 0 and out else 18 dec_cache[key] = d return d entries: list[dict] = [] gas_mirrors: list[dict] = [] for cid, ch in sorted(ds.get("chains", {}).items(), key=lambda x: int(x[0])): rpc_key = CHAIN_RPC_ENV.get(cid) rpc = (env.get(rpc_key) or "").strip() if rpc_key else "" net = ch.get("name", cid) gm = ch.get("gasMirrors") or {} for sym, addr in gm.items(): if sym.startswith("cW"): gas_mirrors.append( { "chain_id": int(cid), "network": net, "symbol": sym, "token_address": addr, } ) if not rpc_key or not rpc: for sym, addr in (ch.get("cwTokens") or {}).items(): if not sym.startswith("cW"): continue entries.append( { "chain_id": int(cid), "network": net, "symbol": sym, "token_address": addr, "rpc_env": rpc_key, "error": "rpc_env_missing_or_empty", } ) continue for sym, addr in (ch.get("cwTokens") or {}).items(): if not sym.startswith("cW"): continue pool, qlab = find_usd_pool(ch, sym) row: dict = { "chain_id": int(cid), "network": net, "symbol": sym, "token_address": addr, "rpc_env": rpc_key, "quote_leg": qlab, "pool_address": pool, } if not pool: row["error"] = "no_usd_quoted_pool_in_deployment_status" entries.append(row) continue code, cok = cast(env, rpc, ["code", pool]) if cok != 0 or not code or code == "0x": row["error"] = "pool_no_bytecode" entries.append(row) continue bout, bok = cast(env, rpc, ["call", pool, "_BASE_TOKEN_()(address)"]) qout, qok = cast(env, rpc, ["call", pool, "_QUOTE_TOKEN_()(address)"]) if bok != 0 or qok != 0: row["error"] = "not_dvm_abi" entries.append(row) continue base_a = bout.split()[0] quote_a = qout.split()[0] row["base_token_address"] = base_a row["quote_token_address"] = quote_a bd = decimals(base_a, rpc) qd = decimals(quote_a, rpc) row["base_decimals"] = bd row["quote_decimals"] = qd rv, rcode = cast(env, rpc, ["call", pool, "getVaultReserve()(uint256,uint256)"]) vault_ratio: str | None = None if rcode == 0 and rv: nums: list[int] = [] for p in rv.replace(",", " ").split(): p = p.strip() if p and (p[0].isdigit() or p.startswith("0x")): try: nums.append(parse_u256(p)) except Exception: pass if len(nums) >= 2: br, qr = nums[0], nums[1] bh = Decimal(br) / Decimal(10**bd) qh = Decimal(qr) / Decimal(10**qd) if bh > 0: vault_ratio = str((qh / bh).quantize(Decimal("1." + "0" * 18))) mp, mok = cast(env, rpc, ["call", pool, "getMidPrice()(uint256)"]) mid_raw: str | None = None mid_over_1e18: str | None = None if mok == 0 and mp: try: m = parse_u256(mp) mid_raw = str(m) mid_over_1e18 = str((Decimal(m) / Decimal(10**18)).quantize(Decimal("1." + "0" * 18))) except Exception as e: row["mid_price_error"] = str(e) if vault_ratio is not None: row["vault_implied_quote_per_base"] = vault_ratio if mid_raw is not None: row["mid_price_raw_uint"] = mid_raw if mid_over_1e18 is not None: row["mid_price_over_1e18"] = mid_over_1e18 if vault_ratio is None and mid_over_1e18 is None: row["error"] = "mid_and_vault_unavailable" entries.append(row) payload = { "schema_version": 1, "generated_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "source_file": str(args.deployment_status.relative_to(ROOT)), "description": ( "cW* cwTokens: PMM mid (getMidPrice) and vault-implied ratio vs USDC/USDT/cWUSDC/cWUSDT " "from deployment-status pools. mid_price_over_1e18 is quote per base in human scale when " "DODO uses 18-decimal mid; vault_implied_quote_per_base is quote_reserve/base_reserve." ), "gas_mirrors": gas_mirrors, "entries": sorted(entries, key=lambda r: (r["symbol"], r["chain_id"])), } args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") print(str(args.output), file=sys.stderr) return 0 if __name__ == "__main__": raise SystemExit(main())