proxmox/scripts/verify/build-live-multivenue-expansion-bundle.py

#!/usr/bin/env python3
from __future__ import annotations

import json
from collections import Counter, defaultdict
from datetime import datetime, timezone
from pathlib import Path


ROOT = Path(__file__).resolve().parents[2]
MASTER_JSON = ROOT / "reports" / "status" / "liquidity-pools-master-map-latest.json"
OUT_JSON = ROOT / "reports" / "status" / "live-multivenue-expansion-bundle-latest.json"
OUT_MD = ROOT / "reports" / "status" / "live-multivenue-expansion-bundle-latest.md"

NATIVE_CANDIDATE_STATUSES = {"documented_reference_surface"}
AGGREGATOR_STATUSES = {"documented_aggregator_surface"}
UNSUPPORTED_STATUSES = {"documented_unsupported_surface"}
EXECUTION_PROTOCOLS = {"uniswap_v3", "balancer", "curve"}


def now() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat()


def load_json(path: Path) -> dict:
    return json.loads(path.read_text())


def write_json(path: Path, payload: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(payload, indent=2) + "\n")


def write_text(path: Path, text: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(text.rstrip() + "\n")


def md_table(headers: list[str], rows: list[list[str]]) -> str:
    out = ["| " + " | ".join(headers) + " |", "| " + " | ".join(["---"] * len(headers)) + " |"]
    out.extend("| " + " | ".join(row) + " |" for row in rows)
    return "\n".join(out)


def build() -> dict:
    report = load_json(MASTER_JSON)
    generated_at = now()

    native_candidates = []
    aggregator_surfaces = []
    unsupported_surfaces = []
    by_protocol = Counter()
    by_chain = defaultdict(lambda: {"network": "", "nativeCandidates": [], "aggregatorOnly": [], "unsupported": []})

    for chain in report.get("chains", []):
        chain_id = chain["chainId"]
        network = chain["network"]
        for venue in chain.get("referenceVenues", []):
            item = {
                "chainId": chain_id,
                "network": network,
                "protocol": venue["protocol"],
                "pair": f"{venue['baseSymbol']}/{venue['quoteSymbol']}",
                "baseTokenAddress": venue.get("baseAddress"),
                "quoteTokenAddress": venue.get("quoteAddress"),
                "venueAddress": venue.get("venueAddress"),
                "status": venue["status"],
            }
            by_chain[chain_id]["network"] = network
            if venue["status"] in NATIVE_CANDIDATE_STATUSES and venue["protocol"] in EXECUTION_PROTOCOLS:
                native_candidates.append(item)
                by_protocol[venue["protocol"]] += 1
                by_chain[chain_id]["nativeCandidates"].append(item)
            elif venue["status"] in AGGREGATOR_STATUSES:
                aggregator_surfaces.append(item)
                by_chain[chain_id]["aggregatorOnly"].append(item)
            elif venue["status"] in UNSUPPORTED_STATUSES:
                unsupported_surfaces.append(item)
                by_chain[chain_id]["unsupported"].append(item)

    rollout_order = [
        {"protocol": "uniswap_v3", "reason": "best first native venue candidate set; appears on every target chain"},
        {"protocol": "balancer", "reason": "second-stage native venue activation where explicitly modeled as supported"},
        {"protocol": "curve", "reason": "third-stage activation after v3 / Balancer coverage where applicable"},
    ]

    chain_summaries = []
    for chain_id in sorted(by_chain):
        row = by_chain[chain_id]
        chain_summaries.append(
            {
                "chainId": chain_id,
                "network": row["network"],
                "nativeCandidateCount": len(row["nativeCandidates"]),
                "aggregatorOnlyCount": len(row["aggregatorOnly"]),
                "unsupportedCount": len(row["unsupported"]),
                "nativeCandidates": row["nativeCandidates"],
                "aggregatorOnly": row["aggregatorOnly"],
                "unsupported": row["unsupported"],
            }
        )

    return {
        "generatedAt": generated_at,
        "source": str(MASTER_JSON.relative_to(ROOT)),
        "summary": {
            "nativeProtocolCandidates": len(native_candidates),
            "aggregatorOnlySurfaces": len(aggregator_surfaces),
            "unsupportedSurfaces": len(unsupported_surfaces),
            "nativeProtocolCounts": dict(by_protocol),
        },
        "rolloutOrder": rollout_order,
        "executionBlockers": [
            "The repo does not currently include public-chain deployment/funding automation for Uniswap v3, Balancer, or Curve venues on the gas-family cW* surfaces.",
            "1inch rows are aggregator overlays, not standalone pool deployments, so they cannot be made live as independent venue contracts through this inventory flow.",
            "Unsupported venue rows are explicitly non-actionable unless protocol support strategy changes.",
        ],
        "chains": chain_summaries,
    }


def render_markdown(bundle: dict) -> str:
    lines = [
        "# Live Multivenue Expansion Bundle",
        "",
        f"- Generated: `{bundle['generatedAt']}`",
        f"- Source: `{bundle['source']}`",
        "",
        "## Summary",
        "",
        f"- Native protocol live-candidate venues: `{bundle['summary']['nativeProtocolCandidates']}`",
        f"- Aggregator-only surfaces: `{bundle['summary']['aggregatorOnlySurfaces']}`",
        f"- Unsupported surfaces: `{bundle['summary']['unsupportedSurfaces']}`",
        f"- Native candidate split: `{json.dumps(bundle['summary']['nativeProtocolCounts'], sort_keys=True)}`",
        "",
        "## Execution Blockers",
        "",
    ]
    lines.extend(f"- {item}" for item in bundle["executionBlockers"])
    lines += ["", "## Rollout Order", ""]
    lines.extend(f"- `{row['protocol']}`: {row['reason']}" for row in bundle["rolloutOrder"])
    lines += ["", "## By Chain", ""]

    summary_rows = []
    for chain in bundle["chains"]:
        summary_rows.append(
            [
                str(chain["chainId"]),
                chain["network"],
                str(chain["nativeCandidateCount"]),
                str(chain["aggregatorOnlyCount"]),
                str(chain["unsupportedCount"]),
            ]
        )
    lines += [md_table(["ChainID", "Network", "Native Candidates", "Aggregator Only", "Unsupported"], summary_rows), ""]

    for chain in bundle["chains"]:
        if chain["nativeCandidateCount"]:
            rows = [
                [row["protocol"], row["pair"], row["venueAddress"] or "—"]
                for row in chain["nativeCandidates"]
            ]
            lines += [f"## {chain['network']} ({chain['chainId']})", "", "### Native Candidates", "", md_table(["Protocol", "Pair", "Reference Address"], rows), ""]
        if chain["aggregatorOnlyCount"]:
            rows = [[row["protocol"], row["pair"]] for row in chain["aggregatorOnly"]]
            lines += ["### Aggregator Only", "", md_table(["Protocol", "Pair"], rows), ""]
        if chain["unsupportedCount"]:
            rows = [[row["protocol"], row["pair"]] for row in chain["unsupported"]]
            lines += ["### Unsupported", "", md_table(["Protocol", "Pair"], rows), ""]

    return "\n".join(lines)


def main() -> int:
    bundle = build()
    write_json(OUT_JSON, bundle)
    write_text(OUT_MD, render_markdown(bundle))
    print(f"Wrote {OUT_JSON.relative_to(ROOT)}")
    print(f"Wrote {OUT_MD.relative_to(ROOT)}")
    return 0


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