proxmox/scripts/verify/check-cmc-provider-report-sanity.py

#!/usr/bin/env python3
"""Advisory sanity checks for the repo CMC-shaped Mainnet report."""

from __future__ import annotations

import json
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Any


ROOT = Path(__file__).resolve().parents[2]
CMC_REPORT = ROOT / "reports/status/token-aggregation-cmc-report-chain1-latest.json"
TRACKERS = ROOT / "reports/status/cwusdc-external-trackers-live-latest.json"
OUT_JSON = ROOT / "reports/status/cmc-provider-report-sanity-latest.json"
OUT_MD = ROOT / "reports/status/cmc-provider-report-sanity-latest.md"

OFFICIAL_QUOTES = {
    "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48": "USDC",
    "0xdac17f958d2ee523a2206206994597c13d831ec7": "USDT",
}
# CMC-shaped report intentionally binds DBIS compliant-face symbols to the official
# USDC/USDT contract addresses (see provider packets). Do not flag as accidental alias.
INTENTIONAL_DBIS_QUOTE_FACE = {
    ("0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48", "cusdc"),
    ("0xdac17f958d2ee523a2206206994597c13d831ec7", "cusdt"),
}
PROMOTED = {"cWUSDC", "cWUSDT", "cWXAUC", "cWXAUT", "cWBTC", "cWETH"}


def token_discards_listing_liquidity_claim(token: dict[str, Any]) -> bool:
    """True when the CMC-shaped row already tells reviewers not to use liquidity as listing evidence."""
    prov = token.get("supply_proof_provenance") or {}
    status = str(prov.get("status") or "")
    source = str(prov.get("source") or "")
    if status in {"proof_required", "non_reportable_until_erc20_deployed"}:
        return True
    if source == "deterministic-placeholder-address":
        return True
    return False


def load(path: Path) -> dict[str, Any]:
    return json.loads(path.read_text()) if path.exists() else {}


def dec(value: Any) -> Decimal:
    try:
        return Decimal(str(value or "0"))
    except (InvalidOperation, ValueError):
        return Decimal(0)


def table(headers: list[str], rows: list[list[Any]]) -> str:
    def cell(value: Any) -> str:
        if isinstance(value, list):
            value = "<br>".join(str(item) for item in value)
        return str(value).replace("|", "\\|").replace("\n", "<br>")

    return "\n".join(
        [
            f"| {' | '.join(cell(header) for header in headers)} |",
            f"| {' | '.join('---' for _ in headers)} |",
            *[f"| {' | '.join(cell(value) for value in row)} |" for row in rows],
        ]
    )


def main() -> int:
    cmc = load(CMC_REPORT)
    trackers = load(TRACKERS)
    tokens = cmc.get("tokens", [])
    warnings: list[dict[str, Any]] = []
    promoted_rows = []

    for token in tokens:
        symbol = token.get("symbol")
        address = str(token.get("contract_address", "")).lower()
        liquidity = dec(token.get("liquidity_usd"))
        volume = dec(token.get("volume_24h"))
        pairs = token.get("pairs", [])
        if address in OFFICIAL_QUOTES and symbol != OFFICIAL_QUOTES[address]:
            face = (address, str(symbol or "").lower())
            if face not in INTENTIONAL_DBIS_QUOTE_FACE:
                warnings.append(
                    {
                        "id": "official_quote_symbol_alias",
                        "symbol": symbol,
                        "address": address,
                        "severity": "warning",
                        "message": f"Official {OFFICIAL_QUOTES[address]} address is presented with symbol {symbol}; keep provider packets explicit about official quote vs DBIS wrapped/compliant symbols.",
                    }
                )
        if symbol in PROMOTED:
            promoted_rows.append(
                {
                    "symbol": symbol,
                    "address": address,
                    "liquidityUsd": str(liquidity),
                    "volume24hUsd": str(volume),
                    "pairCount": len(pairs),
                }
            )
            if liquidity <= 0 and not token_discards_listing_liquidity_claim(token):
                warnings.append(
                    {
                        "id": "zero_reported_liquidity",
                        "symbol": symbol,
                        "address": address,
                        "severity": "warning",
                        "message": "CMC-shaped report shows zero liquidity_usd; do not use it as listing-quality liquidity evidence.",
                    }
                )

    gecko_reserves = []
    for check in trackers.get("checks", []):
        if not str(check.get("id", "")).startswith("geckoterminal"):
            continue
        preview = check.get("jsonPreview")
        if not isinstance(preview, dict):
            continue
        data = preview.get("data")
        if not isinstance(data, dict):
            continue
        attrs_raw = data.get("attributes")
        attrs = attrs_raw if isinstance(attrs_raw, dict) else {}
        gecko_reserves.append(
            {
                "id": check.get("id"),
                "pool": attrs.get("address"),
                "name": attrs.get("name"),
                "reserveUsd": attrs.get("reserve_in_usd"),
                "volume24hUsd": (attrs.get("volume_usd") or {}).get("h24"),
            }
        )

    payload = {
        "schema": "cmc-provider-report-sanity/v1",
        "generatedAt": datetime.now(timezone.utc).isoformat(),
        "inputs": {
            "cmcReport": str(CMC_REPORT.relative_to(ROOT)),
            "trackerReport": str(TRACKERS.relative_to(ROOT)) if TRACKERS.exists() else None,
        },
        "summary": {
            "tokenCount": len(tokens),
            "promotedTokenCount": len(promoted_rows),
            "warningCount": len(warnings),
            "geckoReserveEvidenceCount": len(gecko_reserves),
        },
        "promotedTokens": promoted_rows,
        "geckoReserveEvidence": gecko_reserves,
        "warnings": warnings,
    }
    OUT_JSON.parent.mkdir(parents=True, exist_ok=True)
    OUT_JSON.write_text(json.dumps(payload, indent=2) + "\n")

    lines = [
        "# CMC Provider Report Sanity",
        "",
        f"- Generated: `{payload['generatedAt']}`",
        f"- Warnings: `{len(warnings)}`",
        "",
        "## Promoted Mainnet Rows",
        "",
        table(
            ["Symbol", "Address", "Liquidity USD", "24h volume USD", "Pairs"],
            [[row["symbol"], row["address"], row["liquidityUsd"], row["volume24hUsd"], row["pairCount"]] for row in promoted_rows],
        ),
        "",
        "## GeckoTerminal Reserve Cross-Check",
        "",
        table(
            ["Check", "Pool", "Name", "Reserve USD", "24h volume USD"],
            [[row["id"], row["pool"], row["name"], row["reserveUsd"], row["volume24hUsd"]] for row in gecko_reserves],
        ) if gecko_reserves else "No GeckoTerminal tracker reserve evidence found.",
        "",
        "## Advisory Warnings",
        "",
        table(
            ["ID", "Symbol", "Address", "Severity", "Message"],
            [[w["id"], w.get("symbol", "-"), w.get("address", "-"), w["severity"], w["message"]] for w in warnings],
        ) if warnings else "No warnings.",
    ]
    OUT_MD.write_text("\n".join(lines) + "\n")
    print(f"Wrote {OUT_JSON.relative_to(ROOT)}")
    print(f"Wrote {OUT_MD.relative_to(ROOT)}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())