Refresh cW public price exports

This commit is contained in:
defiQUG
2026-04-29 23:53:14 -07:00
parent 55dacdac08
commit ba8f1b9b76
31 changed files with 2365 additions and 187 deletions

View File

@@ -6,6 +6,8 @@ import math
import re
import subprocess
import time
import urllib.parse
import urllib.request
from collections import deque
from dataclasses import dataclass
from decimal import Decimal, InvalidOperation, getcontext
@@ -39,6 +41,30 @@ CHAIN_CONFIG = {
STABLES = {"USDC": Decimal("1"), "USDT": Decimal("1")}
ORACLE_ISO_USD_REFERENCE = {
"USD": Decimal("1"),
"EUR": Decimal("1.08"),
"GBP": Decimal("1.27"),
"AUD": Decimal("0.66"),
"JPY": Decimal("0.0067"),
"CHF": Decimal("1.11"),
"CAD": Decimal("0.74"),
"XAU": Decimal("3200"),
}
COINGECKO_PLATFORM_IDS = {
"1": "ethereum",
"10": "optimistic-ethereum",
"25": "cronos",
"56": "binance-smart-chain",
"100": "xdai",
"137": "polygon-pos",
"8453": "base",
"42161": "arbitrum-one",
"42220": "celo",
"43114": "avalanche",
}
@dataclass(frozen=True)
class Edge:
@@ -185,6 +211,47 @@ def format_decimal(value: Decimal | None, places: int = 8) -> str:
return format(rounded, "f")
def format_metamask_price(row: dict) -> str:
price = safe_decimal(row.get("metamaskDisplayedPriceOracleQuote"))
if price is not None:
return format_decimal(price)
status = row.get("metamaskDisplayStatus") or "unknown"
if status == "api_error":
note = row.get("metamaskDisplayNote") or ""
if "429" in note:
return "unknown (API 429)"
return "unknown (API error)"
return {
"no_coingecko_platform": "unknown",
"not_listed": "not displayed",
"not_found": "not displayed",
}.get(status, status)
def symbol_oracle_iso4217(symbol: str) -> str:
"""Return the ISO-4217 unit embedded in a cW public wrapper symbol."""
if not symbol.startswith("cW"):
return ""
if len(symbol) >= 7 and symbol[2].isalpha() and symbol[3:6].isalpha():
return symbol[3:6]
if len(symbol) >= 6 and symbol[2:5].isalpha():
return symbol[2:5]
return ""
def oracle_iso_usd_reference(symbol: str) -> tuple[str, Decimal | None]:
oracle_iso = symbol_oracle_iso4217(symbol)
return oracle_iso, ORACLE_ISO_USD_REFERENCE.get(oracle_iso)
def oracle_quote_price(symbol: str, price_usd: Decimal | None) -> Decimal | None:
_oracle_iso, usd_reference = oracle_iso_usd_reference(symbol)
if price_usd is None or usd_reference is None or usd_reference <= 0:
return None
return price_usd / usd_reference
def normalize_18(raw: int) -> Decimal:
return Decimal(raw) / (Decimal(10) ** 18)
@@ -202,6 +269,92 @@ def rpc_for_chain(chain_id: str, env_values: dict[str, str]) -> str:
return ""
def fetch_metamask_display_prices(chain_id: str, chain: dict, env_values: dict[str, str]) -> tuple[dict[str, dict], str]:
"""Approximate MetaMask native display via CoinGecko token prices.
Repo docs record that MetaMask does not read the on-chain PMM/oracle price
directly; CoinGecko listing is the practical observable proxy.
"""
platform = COINGECKO_PLATFORM_IDS.get(chain_id)
if not platform:
return {}, "no_coingecko_platform"
tokens = {
symbol: address
for symbol, address in (chain.get("cwTokens") or {}).items()
if symbol.startswith("cW") and isinstance(address, str) and address.startswith("0x")
}
if not tokens:
return {}, "not_found"
vs_currencies = sorted(
{
iso.lower()
for symbol in tokens
for iso, _usd_reference in [oracle_iso_usd_reference(symbol)]
if iso
}
)
if not vs_currencies:
vs_currencies = ["usd"]
query = urllib.parse.urlencode(
{
"contract_addresses": ",".join(address.lower() for address in tokens.values()),
"vs_currencies": ",".join(vs_currencies),
},
safe=",",
)
url = f"https://api.coingecko.com/api/v3/simple/token_price/{platform}?{query}"
headers = {"User-Agent": "proxmox-cw-price-table/1.0"}
api_key = resolve_env_value("COINGECKO_API_KEY", env_values) or resolve_env_value(
"COINGECKO_DEMO_API_KEY", env_values
)
if api_key:
headers["x-cg-demo-api-key"] = api_key
try:
request = urllib.request.Request(url, headers=headers)
with urllib.request.urlopen(request, timeout=12) as response:
payload = json.loads(response.read().decode("utf-8"))
except Exception as exc:
return {
symbol: {
"priceOracleQuote": None,
"status": "api_error",
"source": "coingecko_token_price",
"note": f"CoinGecko token-price lookup failed: {exc}",
}
for symbol in tokens
}, "api_error"
prices: dict[str, dict] = {}
by_address = {address.lower(): symbol for symbol, address in tokens.items()}
for address, symbol in by_address.items():
oracle_iso, _usd_reference = oracle_iso_usd_reference(symbol)
vs_currency = (oracle_iso or "USD").lower()
oracle_quote_price_value = (payload.get(address) or {}).get(vs_currency)
if oracle_quote_price_value is None:
prices[symbol] = {
"priceOracleQuote": None,
"status": "not_listed",
"source": "coingecko_token_price",
"note": (
f"No CoinGecko token-price result for {oracle_iso or 'USD'}; "
"MetaMask native display is expected to be unavailable."
),
}
continue
prices[symbol] = {
"priceOracleQuote": str(oracle_quote_price_value),
"status": "priced",
"source": "coingecko_token_price",
"note": f"CoinGecko token-price result in {oracle_iso or 'USD'} used as MetaMask native-display proxy.",
}
return prices, "ok"
def build_uniswap_edges(entry: dict) -> list[Edge]:
edges: list[Edge] = []
for row in entry.get("pairsChecked") or []:
@@ -354,17 +507,30 @@ def best_prices_for_chain(chain: dict, edges: list[Edge]) -> dict[str, dict]:
out: dict[str, dict] = {}
for symbol in sorted((chain.get("cwTokens") or {}).keys()):
oracle_iso, usd_reference = oracle_iso_usd_reference(symbol)
resolution = best.get(symbol)
if resolution is None:
out[symbol] = {
"priceUsd": None,
"priceOracleQuote": "1" if oracle_iso else None,
"observedPriceOracleQuote": None,
"oracleIso4217": oracle_iso,
"oracleIsoUsdReference": str(usd_reference) if usd_reference is not None else None,
"shownValueOracleQuote": "1" if oracle_iso else None,
"derivedFrom": "not found",
"sourceType": "not_found",
"notes": ["No live direct or bridged price path was found from USDC/USDT anchors."],
}
continue
price_usd = resolution["price"]
price_oracle_quote = oracle_quote_price(symbol, price_usd)
out[symbol] = {
"priceUsd": str(resolution["price"]),
"priceUsd": str(price_usd),
"priceOracleQuote": "1" if oracle_iso else None,
"observedPriceOracleQuote": str(price_oracle_quote) if price_oracle_quote is not None else None,
"oracleIso4217": oracle_iso,
"oracleIsoUsdReference": str(usd_reference) if usd_reference is not None else None,
"shownValueOracleQuote": "1" if oracle_iso else None,
"derivedFrom": " -> ".join(resolution["steps"]) if resolution["steps"] else "stable anchor",
"sourceType": resolution["venues"][0] if resolution["venues"] else "stable_anchor",
"notes": resolution["notes"],
@@ -386,6 +552,15 @@ def build_report() -> dict:
uniswap_edges = build_uniswap_edges(discovery_by_chain.get(chain_id, {}))
pmm_edges, pmm_snapshots = build_pmm_edges(chain, rpc_url)
price_rows = best_prices_for_chain(chain, uniswap_edges + pmm_edges)
metamask_prices, metamask_status = fetch_metamask_display_prices(chain_id, chain, env_values)
for symbol, row in price_rows.items():
metamask_row = metamask_prices.get(symbol)
row["metamaskDisplayedPriceOracleQuote"] = (metamask_row or {}).get("priceOracleQuote")
row["metamaskDisplayStatus"] = (metamask_row or {}).get("status", metamask_status)
row["metamaskDisplaySource"] = (metamask_row or {}).get("source", "coingecko_token_price")
row["metamaskDisplayNote"] = (metamask_row or {}).get(
"note", "No supported CoinGecko platform mapping for MetaMask display-price lookup."
)
chains_out.append(
{
"chainId": int(chain_id),
@@ -414,11 +589,13 @@ def render_markdown(payload: dict) -> str:
f"- Generated: `{payload['generatedAt']}`",
f"- Deployment inventory: `{payload['inputs']['deploymentStatus']}`",
f"- Uniswap discovery snapshot: `{payload['inputs']['uniswapDiscovery']}`",
"- Price convention: USD per 1 token.",
"- Public Price (USD) is the observed USDC/USDT-anchored execution price per 1 token.",
"- Quote Price (Oracle ISO) and Shown Value (Oracle ISO) are the intended oracle peg display: 1 token = 1 ISO-4217 unit (or 1 troy oz for XAU).",
"- MetaMask Display (Oracle ISO) uses CoinGecko token-price lookup in the token's oracle ISO-4217 unit as the closest observable proxy for native MetaMask balance pricing; `not displayed` means no token-price listing was returned.",
"- `not found` means the generator could not reach the token from a live USDC/USDT anchor using the current public-pair snapshot plus live PMM mid-price reads.",
"",
"| Chain | Token | Price (USD) | Derived From | Source | Notes |",
"|---|---|---:|---|---|---|",
"| Chain | Token | Oracle ISO | Quote Price (Oracle ISO) | Shown Value (Oracle ISO) | Public Price (USD) | MetaMask Display (Oracle ISO) | Derived From | Source | Notes |",
"|---|---|---|---:|---:|---:|---:|---|---|---|",
]
for chain in payload["chains"]:
@@ -429,13 +606,19 @@ def render_markdown(payload: dict) -> str:
chain_cell = f"`{chain['chainId']}` {chain['network']}" if first_row else ""
first_row = False
notes = "; ".join(row["notes"][:2])
oracle_iso = row.get("oracleIso4217") or "unknown"
shown_value = safe_decimal(row.get("shownValueOracleQuote"))
lines.append(
f"| {chain_cell} | `{symbol}` | `{format_decimal(safe_decimal(row['priceUsd']))}` | "
f"| {chain_cell} | `{symbol}` | `{oracle_iso}` | "
f"`{format_decimal(safe_decimal(row.get('priceOracleQuote')))}` | "
f"`{format_decimal(shown_value) if shown_value is not None else 'not found'}` | "
f"`{format_decimal(safe_decimal(row['priceUsd']))}` | "
f"`{format_metamask_price(row)}` | "
f"`{row['derivedFrom']}` | `{row['sourceType']}` | {notes} |"
)
if prices:
lines.append(
f"| | | | | | Activation state: `{chain['activationState'] or 'active'}`; RPC configured: `{chain['rpcConfigured']}` |"
f"| | | | | | | | | | Activation state: `{chain['activationState'] or 'active'}`; RPC configured: `{chain['rpcConfigured']}` |"
)
return "\n".join(lines)