Find who is buying a token, and who they are

export VIKE_API_KEY=...
python oc-early-buyers.py             # real run: needs the key
python oc-early-buyers.py --dry-run   # prints the requests it would make, calls nothing

"""Who is buying the token smart money is moving into? Screen, take the top token, label its biggest buyers.

Run:  VIKE_API_KEY=... python oc-early-buyers.py    (--dry-run lists the requests and makes none)
"""
import os
import sys
from urllib.parse import urlencode

NAME = "oc-early-buyers"
BASE = "https://data.vike.io"
DRY = "--dry-run" in sys.argv
# Wallet types that are intermediaries, not an owner deciding to buy; everything else is worth a look.
PLUMBING = {"cex", "defi", "mev"}


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():
    # Ethereum by default. A token is identified by address only: the screener's symbol is empty.
    screen = call("/v1/onchain/tokens/screener", sort="smart_inflow", window="7d", limit=5)
    # --dry-run has no response to read
    rows = [{"token_address": "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48", "volume_usd": 0}] if DRY else screen
    # The zero address is the chain's native coin (ETH), which often tops the screen; /transfers lists no buyers for it.
    rows = [r for r in rows or [] if int(r["token_address"], 16)]
    if not rows:
        fail("the screener returned no token contracts")

    # Ten rows, one side per call, and no limit parameter. usd_spent and net_usd are plain USD.
    for row in rows:
        token = row["token_address"]
        buyers = call(f"/v1/onchain/tokens/{token}/transfers", period="7d", tab="bought")
        if DRY or buyers:
            break
        print(f"{token} has no buyers in the window, trying the next token")
    else:
        fail(f"none of the top {len(rows)} screened tokens has buyers in the window")
    wallets = ["0xbbbbbbbbbb9cc5e90e3b3af64bdaf62c37eeffcb", "0x06cff7088619c7178f5e14f0b119458d08d2f5ef"]
    if not DRY:
        print(f"top smart-money inflow over 7d: {token}  (${row['volume_usd']:,.0f} volume)")
        wallets = [b["address"] for b in buyers]

    # A POST with no body: the addresses ride in the query string, at most 100 per call.
    labels = call("/v1/onchain/labels/batch", method="POST", addresses=",".join(wallets))
    if DRY:
        return
    if len(labels) != len(wallets):
        fail(f"/labels/batch returned {len(labels)} rows for {len(wallets)} wallets")
    by_wallet = {row["wallet"]: row for row in labels}
    for b in buyers:
        row = by_wallet.get(b["address"].lower(), {})
        kind = (row.get("wallet_type") or "") if row.get("found") else ""
        who = f"{row['entity']} ({kind})" if kind else "unlabelled"
        # bought counts every transfer in, so a flash-loan loop or a router "buys" billions and keeps nothing
        passed_on = b["net_tokens"] < 0.01 * b["bought"]
        flag = "   (passed it on)" if passed_on else "" if kind in PLUMBING else "   <- look here"
        print(f"  {b['address']}  bought ${b['usd_spent']:,.0f}  kept {b['net_tokens']:,.0f}  {who}{flag}")


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