analyze_tezos.py

#!/usr/bin/env python3
"""NFT spend analysis for Tezos wallets via TzKT.

Sums XTZ paid to known Tezos NFT marketplaces (mint + secondary), converting
each tx to USD at the historical XTZ/USD rate embedded in the TzKT response.
Outputs a CSV of every classified tx plus aggregate breakdowns.
"""
from __future__ import annotations

import csv
import sys
from collections import defaultdict
from pathlib import Path

sys.stdout.reconfigure(encoding="utf-8")
sys.stderr.reconfigure(encoding="utf-8")

try:
    import truststore
    truststore.inject_into_ssl()
except ImportError:
    pass

import requests

from _config import get_tezos_wallets, get_start_date_iso

TZKT_API = "https://api.tzkt.io/v1"
WALLETS = get_tezos_wallets()
START_DATE = get_start_date_iso()

MARKETPLACES: dict[str, str] = {
    # fxhash
    "KT1AEVuykWeuuFX7QkEAMNtffzwhe1Z98hJS": "fxhash (issuer v1)",
    "KT1Xo5B7PNBAeynZPmca4bRh6LQow4og1Zb3": "fxhash (mkt v1)",
    "KT1GbyoDi7H1sfXmimXpptZJuCdHMh66WS9u": "fxhash (mkt v2)",
    # objkt
    "KT1WvzYHCNBvDSdwafTHv7nJ1dWmZ8GCYuuC": "objkt.com (mkt v2)",
    "KT1Xjap1TwmDR1d8yEd8ErkraAj2mbdMrPZY": "objkt.com (mkt)",
    "KT1FvqJwEDWb1Gwc55Jd1jjTHRVWbYKUUpyq": "objkt.com (mkt old)",
    "KT1XXu88HkNzQRHNgAf7Mnq68LyS9MZJNoHP": "objkt.com (english auctions)",
    "KT1Vu1Ja6tBjJzwFtAuMnYqfHKAouWafkfGr": "objkt.com (mkt v4)",
    "KT1XJ1jrnDCWNXmqM4cVqB2HwHwUyJN1g6Vp": "objkt.com (mkt v4-2)",
    "KT1SwbTqhSKF6Pdokiu1K4Fpi17ahPPzmt1X": "objkt.com (mkt v3?)",
    # HEN / Teia
    "KT1HbQepzV1nVGg8QVznG7z4RcHseD5kwqBn": "hicetnunc (mkt)",
    "KT1Hkg5qeNhfwpKW4fXvq7HGZB9z2EnmCCA9": "hicetnunc (minter)",
    "KT1PHubm9HtyQEJ4BBpMTVomq6mhbfNZ9z5w": "teia",
    # versum / 8bidou
    "KT1GyRAJNdizF1nojQz62uGYkx8WFRUJm9X5": "versum (mkt)",
    "KT1MxDwChiDwd6WBVs24g1NjERUoK622ZEFp": "8bidou",
    # akaSwap (Taiwanese NFT platform)
    "KT1Dn3sambs7KZGW88hH2obZeSzfmCmGvpFo": "akaSwap (mkt v2.1)",
    "KT1HGL8vx7DP4xETVikL4LUYvFxSV19DxdFN": "akaSwap (mkt v1)",
    "KT1B2BN4qgmtsaoLRhhnFVDNoTyybwTfdrE4": "akaSwap (auction)",
    "KT1J2C7BsYNnSjQsGoyrSXShhYGkrDDLVGDd": "akaSwap (offer)",
    "KT1JRVrBzSyEX1xnidEgMkujeuc2Q6j5FJzB": "akaSwap (gacha)",
    # Other specific NFT contracts identified from this dataset
    "KT1QKCVV36RXtsaarVYTda4spdH7c8R6j3am": "USD-pegged NFT pay (custom)",
    "KT1Q2jUJnrvrrhi4gBpZVLm37nyCqaFNtK7X": "Marina Abramović NFT Sales",
    "KT1EpGgjQs73QfFJs9z7m1Mxm5MTnpC2tqse": "Kalamint",
    "KT1VoZeuBMJF6vxtLqEFMoc4no5VDG789D7z": "Typed marketplace",
    "KT19vw7kh7dzTRxFUZNWu39773baauzNWtzj": "8scribo",
    "KT191reDVKrLxU9rjTSxg53wRqj6zh8pnHgr": "Tezos Domains (TLD buy)",
}

# Known DeFi / non-NFT contracts to exclude from NFT-spend totals.
EXCLUDED_NON_NFT: dict[str, str] = {
    "KT1EtjRRCBC2exyCRXz8UfV7jz7svnkqi7di": "QuipuSwap uUSD (DEX)",
    "KT1Qej1k8WxPvBLUjGVtFXStgzQtcx3itSk5": "QuipuSwap akaDAO (DEX)",
    "KT1CAYNQGvYSF5UvHK21grMrKpe2563w9UcX": "Plenty XTZ/CTez (DEX)",
    "KT1Tuta6vbpHhZ15ixsYD3qJdhnpEAuogLQ9": "3Route v2 (DEX aggregator)",
    "KT1EVYBj3f1rZHNeUtq4ZvVxPTs77wuHwARU": "Tezos Domains registrar (renew)",
}

FXHASH_MINT_TARGETS = {"KT1AEVuykWeuuFX7QkEAMNtffzwhe1Z98hJS"}

# fxhash V4 uses per-project deployed contracts. They share a parameter shape
# with these distinctive keys — if a tx parameter contains >=2 of these, we
# infer fxhash V4 mint/buy/claim regardless of the specific contract address.
FXHASH_V4_PARAM_KEYS = {
    "entropy", "user_entropy", "proxy_for", "generator_id",
    "condition_extra", "sale_extra", "referrals", "burn_tokens",
}

# Pattern-match unknown contracts: any KT1 + amount > 0 + one of these
# entrypoint substrings is treated as an NFT mint/claim from a custom contract.
# 'claim' is included because fxhash V4 / drops use it heavily; false-positives
# would mostly be staking-reward claims which typically have amount == 0.
NFT_MINT_ENTRYPOINT_PATTERNS = ("mint", "claim", "redeem")


def alias_marketplace(alias: str | None) -> str | None:
    if not alias:
        return None
    a = alias.lower()
    if "fxhash" in a:
        return "fxhash (by alias)"
    if "objkt" in a:
        return "objkt.com (by alias)"
    if "teia" in a:
        return "teia (by alias)"
    if "hic et nunc" in a or a.startswith("hen "):
        return "hicetnunc (by alias)"
    if "versum" in a:
        return "versum (by alias)"
    if "8bidou" in a or "8bid" in a:
        return "8bidou (by alias)"
    if "rarible" in a:
        return "rarible (by alias)"
    # akaSwap family: akaMetaverse, akaAuction, akaOffer, akaGacha, akaDAO, akaSwap, ...
    if a.startswith("aka") or "akaswap" in a:
        return "akaSwap (by alias)"
    if "emprops" in a:
        return "Emprops (by alias)"
    if "kalamint" in a:
        return "Kalamint (by alias)"
    if "tezos domains" in a:
        return "Tezos Domains (by alias)"
    if "marketplace" in a or "nft sales" in a:
        return f"other marketplace ({alias})"
    return None


def classify_kind(target_addr: str, entrypoint: str | None) -> str:
    ep = (entrypoint or "").lower()
    if "mint" in ep or "claim" in ep or "redeem" in ep:
        return "mint"
    if target_addr in FXHASH_MINT_TARGETS and ep == "collect":
        return "mint"
    if ep == "buy" or ep == "pay":  # fxhash V4 buy = mint
        return "mint"
    return "purchase"


def detect_fxhash_v4(parameter: dict | None) -> bool:
    """fxhash V4 per-project contracts share a distinctive parameter shape."""
    if not parameter:
        return False
    value = parameter.get("value")
    if not isinstance(value, dict):
        return False
    return len(set(value.keys()) & FXHASH_V4_PARAM_KEYS) >= 2


def classify_target(target_addr: str, target_alias: str | None,
                    parameter: dict | None, entrypoint: str | None) -> str | None:
    if target_addr in EXCLUDED_NON_NFT:
        return None
    if target_addr in MARKETPLACES:
        return MARKETPLACES[target_addr]
    m = alias_marketplace(target_alias)
    if m:
        return m
    if detect_fxhash_v4(parameter):
        return "fxhash V4 (heuristic by param shape)"
    if entrypoint:
        ep = entrypoint.lower()
        if any(p in ep for p in NFT_MINT_ENTRYPOINT_PATTERNS):
            return "other NFT mint/claim (custom contract)"
    return None


def fetch_outgoing(wallet: str) -> list[dict]:
    out: list[dict] = []
    last_id = 0
    while True:
        url = (
            f"{TZKT_API}/operations/transactions"
            f"?sender={wallet}&status=applied"
            f"&timestamp.ge={START_DATE}"
            f"&amount.gt=0"
            f"&quote=usd"
            f"&id.gt={last_id}"
            f"&sort.asc=id"
            f"&limit=10000"
        )
        r = requests.get(url, timeout=60)
        r.raise_for_status()
        batch = r.json()
        if not batch:
            break
        out.extend(batch)
        last_id = batch[-1]["id"]
        print(f"  fetched {len(out)} so far (last id {last_id})", file=sys.stderr)
        if len(batch) < 10000:
            break
    return out


def main() -> None:
    rows: list[dict] = []
    unknown_contracts: dict[str, dict] = defaultdict(
        lambda: {"count": 0, "xtz": 0.0, "alias": None, "entrypoints": defaultdict(int)}
    )

    for wallet in WALLETS:
        print(f"Fetching outgoing tx for {wallet} since {START_DATE}...", file=sys.stderr)
        ops = fetch_outgoing(wallet)
        print(f"  total {len(ops)} outgoing tx with amount > 0\n", file=sys.stderr)

        for op in ops:
            target = op.get("target") or {}
            target_addr = target.get("address", "")
            target_alias = target.get("alias")
            if not target_addr.startswith("KT1"):
                continue

            entrypoint = (op.get("parameter") or {}).get("entrypoint")
            xtz = op["amount"] / 1_000_000
            usd_rate = (op.get("quote") or {}).get("usd")
            usd_val = xtz * usd_rate if usd_rate is not None else None

            if target_addr in EXCLUDED_NON_NFT:
                continue

            parameter = op.get("parameter")
            mkt = classify_target(target_addr, target_alias, parameter, entrypoint)
            if mkt is None:
                u = unknown_contracts[target_addr]
                u["count"] += 1
                u["xtz"] += xtz
                u["alias"] = target_alias
                u["entrypoints"][entrypoint or "(none)"] += 1
                continue

            rows.append({
                "wallet": wallet,
                "timestamp": op["timestamp"],
                "marketplace": mkt,
                "kind": classify_kind(target_addr, entrypoint),
                "entrypoint": entrypoint,
                "xtz": xtz,
                "usd_rate": usd_rate,
                "usd": usd_val,
                "target": target_addr,
                "target_alias": target_alias,
                "hash": op["hash"],
            })

    csv_path = Path(__file__).parent / "nft_spend_tezos.csv"
    if not rows:
        print("No NFT-related transactions found.", file=sys.stderr)
        return
    with csv_path.open("w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
        w.writeheader()
        w.writerows(rows)
    print(f"Wrote {csv_path} ({len(rows)} rows)\n")

    total_xtz = sum(r["xtz"] for r in rows)
    total_usd = sum(r["usd"] for r in rows if r["usd"] is not None)
    print(f"=== TOTAL NFT SPEND (Tezos, both wallets, since 2021-01-01) ===")
    print(f"  {len(rows):5d} tx   {total_xtz:>12,.2f} XTZ   ${total_usd:>12,.2f}\n")

    def grouped(key_fn):
        d: dict = defaultdict(lambda: {"xtz": 0.0, "usd": 0.0, "count": 0})
        for r in rows:
            k = key_fn(r)
            d[k]["xtz"] += r["xtz"]
            d[k]["usd"] += r["usd"] or 0
            d[k]["count"] += 1
        return d

    def print_group(title: str, d: dict, sort_by_usd: bool = True):
        print(f"=== {title} ===")
        items = sorted(d.items(), key=lambda x: -x[1]["usd"]) if sort_by_usd else sorted(d.items())
        for k, v in items:
            print(f"  {str(k):42s} {v['count']:>5d} tx  {v['xtz']:>10,.2f} XTZ  ${v['usd']:>10,.2f}")
        print()

    print_group("BY KIND", grouped(lambda r: r["kind"]))
    print_group("BY MARKETPLACE", grouped(lambda r: r["marketplace"]))
    print_group("BY YEAR", grouped(lambda r: r["timestamp"][:4]), sort_by_usd=False)
    print_group("BY WALLET", grouped(lambda r: r["wallet"]))

    if unknown_contracts:
        unk_xtz_total = sum(v["xtz"] for v in unknown_contracts.values())
        print(f"=== UNCLASSIFIED KT1 TARGETS ({len(unknown_contracts)} contracts, {unk_xtz_total:,.2f} XTZ total) ===")
        print(f"    Top 40 by XTZ — review and extend MARKETPLACES / EXCLUDED_NON_NFT in source")
        top = sorted(unknown_contracts.items(), key=lambda x: -x[1]["xtz"])[:40]
        for addr, v in top:
            alias = v["alias"] or "(no alias)"
            eps = ", ".join(f"{e}({c})" for e, c in sorted(v["entrypoints"].items(), key=lambda x: -x[1])[:3])
            print(f"  {addr}  {alias[:34]:34s}  {v['count']:>4d} tx  {v['xtz']:>8,.2f} XTZ   eps: {eps}")


if __name__ == "__main__":
    main()