Crypto wallet lookup & checker: who is behind an address

"""Put a name to an Ethereum address: label, who said so, and its ENS name.

Run:  VIKE_API_KEY=... python oc-whose-wallet.py [0xADDRESS]    (--dry-run lists the requests and makes none)
"""
import os
import sys
from urllib.parse import urlencode

NAME = "oc-whose-wallet"
BASE = "https://data.vike.io"
DRY = "--dry-run" in sys.argv
ARGS = [a for a in sys.argv[1:] if not a.startswith("--")]
WALLET = ARGS[0] if ARGS else "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045"


def fail(why):
    sys.exit(f"{NAME}: {why}")


def call(path, method="GET", **params):
    url = BASE + path + ("?" + urlencode(params) if params else "")
    if DRY:
        print(method, url)
        return None
    key = os.environ.get("VIKE_API_KEY") or fail("set VIKE_API_KEY")
    import requests

    r = requests.request(method, url, headers={"X-API-Key": key}, timeout=60)
    if r.status_code != 200:
        fail(f"{method} {path} answered HTTP {r.status_code}: {r.text[:200]}")
    return r.json()


def main():
    # A POST with no body; one address is a one-element list. An address that is not valid comes back as [].
    batch = call("/v1/onchain/labels/batch", method="POST", addresses=WALLET)
    # Who said it: the consensus plus every source's own claim. A thin list can mean a source is rebuilding.
    prov = call(f"/v1/onchain/labels/{WALLET}/provenance")
    # Reverse ENS. source says which path answered: local (our store) or rpc (a fresh lookup).
    ens = call(f"/v1/onchain/wallets/{WALLET}/ens")
    if DRY:
        return
    if not batch:
        fail(f"{WALLET} is not a valid address")
    row = batch[0]
    print(f"{WALLET}")
    if row["found"]:
        print(f"  label    {row['entity']} ({row['wallet_type']}), confidence {row['confidence']}")
        print(f"  30 days  ${row['volume_30d_usd']:,.0f} volume, {row['tx_30d']} txs, risk flag {row['risk_flag'] or 'none'}")
    else:
        print("  label    none on record")
    claims = "; ".join(s["label"] for s in prov.get("sources", []))
    print(f"  sources  {prov.get('source_count', 0)}: {claims or 'none name it'}")
    name = ens["ens"]
    if name:
        print(f"  ens      {name}")
    elif ens["source"] == "rpc":
        print("  ens      none found, but an rpc miss may only mean the resolvers were unreachable: retry before concluding")
    else:
        print("  ens      no name set")


if __name__ == "__main__":
    try:
        main()
    except Exception as e:  # a shape the example does not expect fails by name, not by traceback
        fail(f"{type(e).__name__}: {e}")

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# entity, type and risk flag for five addresses in one call
r = requests.post(
    "https://data.vike.io/v1/onchain/labels/batch?addresses=0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045,0xeae7380dd4cef6fbd1144f49e4d1e6964258a4f4,0xbbbbbbbbbb9cc5e90e3b3af64bdaf62c37eeffcb,0x47666Fab8bd0Ac7003bce3f5C3585383F09486E2,0xA160cdAB225685dA1d56aa342Ad8841c3b53f291",
    headers={"X-API-Key": API_KEY},
)

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# the flagged addresses of a list, with their flag
r = requests.post(
    "https://data.vike.io/v1/onchain/labels/batch?addresses=0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045,0x47666Fab8bd0Ac7003bce3f5C3585383F09486E2,0xA160cdAB225685dA1d56aa342Ad8841c3b53f291",
    headers={"X-API-Key": API_KEY},
)
# then: jq '[.[] | select((.risk_flag // "") != "") | {wallet, entity, risk_flag}]'

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# every source's claim about one flagged address, plus the consensus
r = requests.get(
    "https://data.vike.io/v1/onchain/labels/0xd90e2f925da726b50c4ed8d0fb90ad053324f31b/provenance",
    headers={"X-API-Key": API_KEY},
)

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# the same lookup against Base's labels
r = requests.post(
    "https://data.vike.io/v1/onchain/labels/batch?chain=base&addresses=0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045,0xeae7380dd4cef6fbd1144f49e4d1e6964258a4f4",
    headers={"X-API-Key": API_KEY},
)

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# addresses 101 to 200, in a second call
# the list here: every address in the newest 500 transfers of a Coinbase hot wallet, in order of appearance
transfers = requests.get(
    "https://data.vike.io/v1/onchain/wallets/0xa9d1e08c7793af67e9d92fe308d5697fb81d3e43/transactions?limit=500",
    headers={"X-API-Key": API_KEY},
).json()
ADDRESSES = list(dict.fromkeys(a for t in transfers for a in (t["from_address"], t["to_address"])))
BATCH_2 = ",".join(ADDRESSES[100:200])
r = requests.post(
    f"https://data.vike.io/v1/onchain/labels/batch?addresses={BATCH_2}",
    headers={"X-API-Key": API_KEY},
)

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# the first coin and the first priced token the wallet received, each with its sender's label
r = requests.get(
    "https://data.vike.io/v1/onchain/wallets/0x556e9d5d9e64638edde632bbd4810c9fd0d88397/first-funder",
    headers={"X-API-Key": API_KEY},
)

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# the same question on Base
r = requests.get(
    "https://data.vike.io/v1/onchain/wallets/0x2b1e8e051d85a65306910aef136d2aebf2eaccc5/first-funder?chain=base",
    headers={"X-API-Key": API_KEY},
)

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# the very first inbound coin, whatever its size
r = requests.get(
    "https://data.vike.io/v1/onchain/wallets/0x556e9d5d9e64638edde632bbd4810c9fd0d88397/first-funder?min_native=0",
    headers={"X-API-Key": API_KEY},
)

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# only a first coin of at least 0.5 ETH and a first token worth at least $1000
r = requests.get(
    "https://data.vike.io/v1/onchain/wallets/0x556e9d5d9e64638edde632bbd4810c9fd0d88397/first-funder?min_native=0.5&min_usd=1000",
    headers={"X-API-Key": API_KEY},
)

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# which exchanges a market-maker wallet used over 90 days, and how much went each way
r = requests.get(
    "https://data.vike.io/v1/onchain/wallets/0x84b38bc60f3bd82640ecefa320dab2be62e2da15/exchange-usage?window=90d",
    headers={"X-API-Key": API_KEY},
)

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# one line per exchange: deposits and withdrawals in USD
r = requests.get(
    "https://data.vike.io/v1/onchain/wallets/0x84b38bc60f3bd82640ecefa320dab2be62e2da15/exchange-usage?window=90d",
    headers={"X-API-Key": API_KEY},
)
# then: jq '.exchanges[] | {entity, sent_usd, recv_usd, sent_transfers, recv_transfers}'

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# a small wallet's exchanges over the last year
r = requests.get(
    "https://data.vike.io/v1/onchain/wallets/0x556e9d5d9e64638edde632bbd4810c9fd0d88397/exchange-usage",
    headers={"X-API-Key": API_KEY},
)

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# exchanges a wallet used on BSC over all recorded history
r = requests.get(
    "https://data.vike.io/v1/onchain/wallets/0x40c9b7d91762f9c715478876310c554f4dc10435/exchange-usage?chain=bsc&window=all",
    headers={"X-API-Key": API_KEY},
)

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# does this address forward its money to one exchange?
r = requests.get(
    "https://data.vike.io/v1/onchain/wallets/0x556e9d5d9e64638edde632bbd4810c9fd0d88397/exchange-usage",
    headers={"X-API-Key": API_KEY},
)

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# the address's inbound transfers over the last year, newest first
r = requests.get(
    "https://data.vike.io/v1/onchain/wallets/0x556e9d5d9e64638edde632bbd4810c9fd0d88397/transactions?direction=in&days=365&limit=50",
    headers={"X-API-Key": API_KEY},
)

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# the pattern and the exchange it points to, or null
r = requests.get(
    "https://data.vike.io/v1/onchain/wallets/0x556e9d5d9e64638edde632bbd4810c9fd0d88397/exchange-usage",
    headers={"X-API-Key": API_KEY},
)
# then: jq '{pattern, pattern_basis}'

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# the same check over the last 90 days
r = requests.get(
    "https://data.vike.io/v1/onchain/wallets/0x556e9d5d9e64638edde632bbd4810c9fd0d88397/exchange-usage?window=90d",
    headers={"X-API-Key": API_KEY},
)

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# has vitalik.eth been active on HyperEVM?
r = requests.get(
    "https://data.vike.io/v1/onchain/wallets/0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045?chain=hyperevm",
    headers={"X-API-Key": API_KEY},
)

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# the Ethereum figures
r = requests.get(
    "https://data.vike.io/v1/onchain/wallets/0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045?chain=ethereum",
    headers={"X-API-Key": API_KEY},
)

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# the BNB Chain figures
r = requests.get(
    "https://data.vike.io/v1/onchain/wallets/0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045?chain=bsc",
    headers={"X-API-Key": API_KEY},
)

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# the Base figures
r = requests.get(
    "https://data.vike.io/v1/onchain/wallets/0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045?chain=base",
    headers={"X-API-Key": API_KEY},
)

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# chain and value of every holding worth at least 1 USD
r = requests.get(
    "https://data.vike.io/v1/onchain/wallets/0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045/balances",
    headers={"X-API-Key": API_KEY},
)
# then: jq '.holdings | map({chain, symbol, usd_value})'

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# which exchanges the wallet used over the last year, and whether it sweeps to one of them
r = requests.get(
    "https://data.vike.io/v1/onchain/wallets/0x556e9d5d9e64638edde632bbd4810c9fd0d88397/exchange-usage",
    headers={"X-API-Key": API_KEY},
)

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# the first inbound ETH and the first priced inbound token
r = requests.get(
    "https://data.vike.io/v1/onchain/wallets/0x556e9d5d9e64638edde632bbd4810c9fd0d88397/first-funder",
    headers={"X-API-Key": API_KEY},
)

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# first_seen, last_active and lifetime transfer count
r = requests.get(
    "https://data.vike.io/v1/onchain/wallets/0x556e9d5d9e64638edde632bbd4810c9fd0d88397",
    headers={"X-API-Key": API_KEY},
)
# then: jq '{chain, all}'

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# the same check over 90 days
r = requests.get(
    "https://data.vike.io/v1/onchain/wallets/0x556e9d5d9e64638edde632bbd4810c9fd0d88397/exchange-usage?window=90d",
    headers={"X-API-Key": API_KEY},
)

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# who sent this wallet its first ETH and its first priced token
r = requests.get(
    "https://data.vike.io/v1/onchain/wallets/0x556e9d5d9e64638edde632bbd4810c9fd0d88397/first-funder",
    headers={"X-API-Key": API_KEY},
)

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# both senders and their labels, nothing else
r = requests.get(
    "https://data.vike.io/v1/onchain/wallets/0x556e9d5d9e64638edde632bbd4810c9fd0d88397/first-funder",
    headers={"X-API-Key": API_KEY},
)
# then: jq '{address, coin_from: .first_native_in.from, coin_label: .first_native_in.wallet_type, token_from: .first_token_in.from, token_label: .first_token_in.wallet_type}'

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# WALLET_A and WALLET_B are two wallets of one group: what they share over all history
WALLETS = [  # your list; one first-funder call per wallet, grouped by first_native_in.from
    "0x556e9d5d9e64638edde632bbd4810c9fd0d88397",
    "0x690c3eebf6ee237f00fc96f9766282f096c191cc",
    "0x20bdd9bc103293f20a142f5118c00434e94cae4c",
]
groups = {}
for w in WALLETS:
    first = requests.get(
        f"https://data.vike.io/v1/onchain/wallets/{w}/first-funder",
        headers={"X-API-Key": API_KEY},
    ).json().get("first_native_in")
    if first and first["wallet_type"] != "cex":  # an exchange hot wallet links nobody
        groups.setdefault(first["from"], []).append(w)
WALLET_A, WALLET_B = next(g for g in groups.values() if len(g) > 1)[:2]
r = requests.get(
    f"https://data.vike.io/v1/onchain/wallets/compare?address_a={WALLET_A}&address_b={WALLET_B}&window=all",
    headers={"X-API-Key": API_KEY},
)

import os
import requests

API_KEY = os.environ["VIKE_API_KEY"]  # your key, read from the environment

# FUNDER is the shared sender from the grouping; sent_usd > 0 marks the wallets it paid
WALLETS = [  # your list; one first-funder call per wallet, grouped by first_native_in.from
    "0x556e9d5d9e64638edde632bbd4810c9fd0d88397",
    "0x690c3eebf6ee237f00fc96f9766282f096c191cc",
    "0x20bdd9bc103293f20a142f5118c00434e94cae4c",
]
groups = {}
for w in WALLETS:
    first = requests.get(
        f"https://data.vike.io/v1/onchain/wallets/{w}/first-funder",
        headers={"X-API-Key": API_KEY},
    ).json().get("first_native_in")
    if first and first["wallet_type"] != "cex":  # an exchange hot wallet links nobody
        groups.setdefault(first["from"], []).append(w)
FUNDER = next(f for f, g in groups.items() if len(g) > 1)
r = requests.get(
    f"https://data.vike.io/v1/onchain/wallets/{FUNDER}/counterparties?window=all&limit=50",
    headers={"X-API-Key": API_KEY},
)