Files
proxmox/scripts/verify/generate-cwusdc-supply-circulating-attestation.py
defiQUG 4ebf2d7902
Some checks failed
Deploy to Phoenix / validate (push) Failing after 1s
Deploy to Phoenix / deploy (push) Has been skipped
Deploy to Phoenix / deploy-atomic-swap-dapp (push) Has been skipped
Deploy to Phoenix / cloudflare (push) Has been skipped
chore(repo): sync operator workspace (config, scripts, docs, multi-chain)
Add optional Cosmos/Engine-X/act-runner templates, CWUSDC/EI-matrix tooling,
non-EVM route planner in multi-chain-execution (tests passing), token list and
extraction updates, and documentation (MetaMask matrix, GRU/CWUSDC packets).

Ignore institutional evidence tarballs/sha256 under reports/status.

Validated with: bash scripts/verify/run-all-validation.sh --skip-genesis

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-11 16:25:08 -07:00

270 lines
11 KiB
Python

#!/usr/bin/env python3
"""Generate a current cWUSDC supply and circulating-supply attestation.
The output is intentionally tracker-facing: it separates on-chain total supply
from proposed circulating-supply methodology and does not silently exclude
operator or protocol balances unless explicitly requested.
"""
from __future__ import annotations
import argparse
import datetime as dt
import json
import os
import re
import time
import urllib.parse
import urllib.request
from decimal import Decimal, getcontext
from pathlib import Path
from typing import Any
getcontext().prec = 80
ROOT = Path(__file__).resolve().parents[2]
REPORT_JSON = ROOT / "reports" / "status" / "cwusdc-supply-circulating-attestation-latest.json"
REPORT_MD = ROOT / "reports" / "status" / "cwusdc-supply-circulating-attestation-latest.md"
ETHERSCAN_API = "https://api.etherscan.io/v2/api"
ETHERSCAN_PAGE = "https://etherscan.io/token/{address}"
CWUSDC = "0x2de5F116bFcE3d0f922d9C8351e0c5Fc24b9284a"
DECIMALS = 6
KNOWN_BALANCES = {
"operator": "0x4A666F96fC8764181194447A7dFdb7d471b301C8",
"engineXVirtualBatchVault": "0xf108586d1FC330EA1D4EA4ff8fd983cde94279B1",
"uniswapV3CwusdcUsdcPool": "0x1Cf2e685682C7F7beF508F0Af15Dfb5CDda01ee3",
"uniswapV2CwusdcUsdcPair": "0xC28706F899266b36BC43cc072b3a921BDf2C48D9",
}
def load_dotenv(path: Path) -> None:
if not path.exists():
return
for line in path.read_text().splitlines():
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
key = key.strip()
value = value.strip().strip('"').strip("'")
if key and key not in os.environ:
os.environ[key] = value
def fetch_json(url: str, timeout: int = 30) -> Any:
req = urllib.request.Request(url, headers={"User-Agent": "dbis-cwusdc-supply-attestation/1.0"})
with urllib.request.urlopen(req, timeout=timeout) as response:
return json.loads(response.read().decode("utf-8"))
def etherscan_call(params: dict[str, str], api_key: str) -> Any:
query = {"chainid": "1", **params, "apikey": api_key}
url = f"{ETHERSCAN_API}?{urllib.parse.urlencode(query)}"
last_payload: Any = None
for attempt in range(6):
payload = fetch_json(url)
last_payload = payload
message = str(payload.get("message", ""))
result = payload.get("result")
if str(payload.get("status")) != "0" or message.lower() == "no transactions found":
time.sleep(0.35)
return result
if isinstance(result, str) and result.startswith("0x"):
time.sleep(0.35)
return result
if "rate limit" in str(result).lower() or "rate limit" in message.lower():
time.sleep(1.25 + attempt * 0.5)
continue
raise RuntimeError(f"Etherscan API error: {payload.get('message')} {payload.get('result')}")
raise RuntimeError(f"Etherscan API error after retries: {last_payload}")
def human(raw: int, decimals: int = DECIMALS) -> str:
scaled = Decimal(raw) / (Decimal(10) ** decimals)
return f"{scaled:f}"
def parse_int(value: Any) -> int:
if isinstance(value, str) and value.startswith("0x"):
return int(value, 16)
return int(str(value))
def fetch_etherscan_page_stats(address: str) -> dict[str, Any]:
url = ETHERSCAN_PAGE.format(address=address)
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0 DBIS-cwusdc-attestation/1.0"})
with urllib.request.urlopen(req, timeout=30) as response:
html = response.read().decode("utf-8", errors="replace")
holders_match = re.search(r"<h4[^>]*>\s*Holders\s*</h4>\s*<div[^>]*>\s*<div>\s*([0-9,]+)", html, re.I)
total_supply_match = re.search(r'id="ContentPlaceHolder1_hdnTotalSupply" value="([^"]+)"', html)
onchain_marketcap_missing = "id=\"ContentPlaceHolder1_tr_marketcap\"" in html and re.search(
r"id=\"ContentPlaceHolder1_tr_marketcap\".*?<div>\s*-\s*</div>", html, re.I | re.S
)
circulating_marketcap_missing = "id=\"ContentPlaceHolder1_tr_circulatingmarketcap\"" in html and re.search(
r"id=\"ContentPlaceHolder1_tr_circulatingmarketcap\".*?<div>\s*-\s*</div>", html, re.I | re.S
)
return {
"url": url,
"holdersText": holders_match.group(1) if holders_match else None,
"totalSupplyText": total_supply_match.group(1) if total_supply_match else None,
"onchainMarketCapMissing": bool(onchain_marketcap_missing),
"circulatingMarketCapMissing": bool(circulating_marketcap_missing),
}
def build(args: argparse.Namespace) -> dict[str, Any]:
load_dotenv(ROOT / ".env")
api_key = args.etherscan_api_key or os.environ.get("ETHERSCAN_API_KEY", "")
if not api_key:
raise SystemExit("ETHERSCAN_API_KEY is required")
latest_block_raw = etherscan_call({"module": "proxy", "action": "eth_blockNumber"}, api_key)
latest_block = parse_int(latest_block_raw)
total_supply_raw = parse_int(
etherscan_call(
{"module": "stats", "action": "tokensupply", "contractaddress": args.token},
api_key,
)
)
known: dict[str, Any] = {}
excluded_raw = 0
exclude_set = set(args.exclude_known or [])
for label, address in KNOWN_BALANCES.items():
raw = parse_int(
etherscan_call(
{
"module": "account",
"action": "tokenbalance",
"contractaddress": args.token,
"address": address,
"tag": "latest",
},
api_key,
)
)
excluded = label in exclude_set
if excluded:
excluded_raw += raw
known[label] = {
"address": address,
"balanceRaw": str(raw),
"balanceUnits": human(raw),
"excludedFromCirculatingSupply": excluded,
}
circulating_raw = total_supply_raw - excluded_raw
page_stats = fetch_etherscan_page_stats(args.token)
return {
"schema": "cwusdc-supply-circulating-attestation/v1",
"generatedAt": dt.datetime.now(dt.UTC).isoformat().replace("+00:00", "Z"),
"purpose": "Tracker-facing supply and circulating-supply attestation for Etherscan Value propagation.",
"network": {"chainId": 1, "name": "Ethereum Mainnet", "referenceBlock": latest_block},
"token": {
"address": args.token,
"caip19": f"eip155:1/erc20:{args.token}",
"name": "Wrapped cUSDC",
"symbol": "cWUSDC",
"decimals": DECIMALS,
"etherscan": ETHERSCAN_PAGE.format(address=args.token),
},
"supply": {
"totalSupplyRaw": str(total_supply_raw),
"totalSupplyUnits": human(total_supply_raw),
"excludedProtocolControlledRaw": str(excluded_raw),
"excludedProtocolControlledUnits": human(excluded_raw),
"circulatingSupplyRaw": str(circulating_raw),
"circulatingSupplyUnits": human(circulating_raw),
"formula": "circulatingSupply = totalSupply - explicitlyExcludedProtocolControlledNonCirculatingBalances",
"defaultPolicy": "No known balance is excluded unless the operator passes --exclude-known for that label or a tracker requests a specific exclusion methodology.",
},
"knownBalances": known,
"etherscanPageObservation": page_stats,
"submissionPosition": {
"readyForTrackerReview": True,
"requestedProviderAction": "Accept total/circulating supply for the exact Mainnet cWUSDC contract and use it with accepted USD price data to populate market cap/value surfaces.",
"caveats": [
"This is an on-chain supply attestation, not third-party listing approval.",
"Chain 138 cUSDC source-asset activity must not be counted as Ethereum Mainnet cWUSDC transfer activity.",
"If a tracker requires treasury, bridge, operator, or pool exclusions, regenerate with explicit --exclude-known labels and attach the requested signed inventory.",
],
},
}
def write_md(payload: dict[str, Any], path: Path) -> None:
supply = payload["supply"]
lines = [
"# cWUSDC Supply and Circulating-Supply Attestation",
"",
f"- Generated: `{payload['generatedAt']}`",
f"- Reference block: `{payload['network']['referenceBlock']}`",
f"- Token: `{payload['token']['address']}`",
f"- CAIP-19: `{payload['token']['caip19']}`",
"",
"## Supply",
"",
"| Field | Value |",
"|---|---:|",
f"| Total supply | `{supply['totalSupplyUnits']}` |",
f"| Explicitly excluded protocol-controlled balances | `{supply['excludedProtocolControlledUnits']}` |",
f"| Circulating supply | `{supply['circulatingSupplyUnits']}` |",
"",
f"Formula: `{supply['formula']}`",
"",
"## Known Balances",
"",
"| Label | Address | Balance | Excluded |",
"|---|---|---:|---:|",
]
for label, item in payload["knownBalances"].items():
lines.append(
f"| `{label}` | `{item['address']}` | `{item['balanceUnits']}` | `{item['excludedFromCirculatingSupply']}` |"
)
page = payload["etherscanPageObservation"]
lines.extend(
[
"",
"## Etherscan Observation",
"",
f"- URL: `{page['url']}`",
f"- Holders text: `{page['holdersText']}`",
f"- Total supply text: `{page['totalSupplyText']}`",
f"- Onchain market cap missing: `{page['onchainMarketCapMissing']}`",
f"- Circulating market cap missing: `{page['circulatingMarketCapMissing']}`",
"",
"## Caveats",
"",
]
)
for caveat in payload["submissionPosition"]["caveats"]:
lines.append(f"- {caveat}")
path.write_text("\n".join(lines) + "\n")
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--token", default=CWUSDC)
parser.add_argument("--etherscan-api-key", default="")
parser.add_argument("--exclude-known", action="append", choices=sorted(KNOWN_BALANCES), help="Known balance label to exclude from circulating supply. Repeatable.")
parser.add_argument("--json-out", type=Path, default=REPORT_JSON)
parser.add_argument("--md-out", type=Path, default=REPORT_MD)
args = parser.parse_args()
payload = build(args)
args.json_out.parent.mkdir(parents=True, exist_ok=True)
args.json_out.write_text(json.dumps(payload, indent=2) + "\n")
write_md(payload, args.md_out)
print(f"Wrote {args.json_out.relative_to(ROOT)}")
print(f"Wrote {args.md_out.relative_to(ROOT)}")
print(f"circulatingSupply={payload['supply']['circulatingSupplyUnits']}")
return 0
if __name__ == "__main__":
raise SystemExit(main())