Solana Rpc

#!/usr/bin/env python3
"""Get top holders and concentration metrics for any Solana token.

Uses direct RPC calls (getTokenLargestAccounts, getTokenSupply) to
fetch holder data and compute concentration metrics: top-N percentage,
Gini coefficient, HHI, and Nakamoto coefficient.

Usage:
    python scripts/token_holders.py
    TOKEN_ADDRESS="TokenMint..." python scripts/token_holders.py

Dependencies:
    uv pip install httpx

Environment Variables:
    SOLANA_RPC_URL: Solana RPC endpoint (default: public mainnet)
    TOKEN_ADDRESS: Token mint address (default: BONK)
"""

import os
import sys
import time
from typing import Any, Optional

import httpx

# ── Configuration ───────────────────────────────────────────────────

RPC_URL = os.getenv("SOLANA_RPC_URL", "https://api.mainnet-beta.solana.com")
TOKEN_ADDRESS = os.getenv(
    "TOKEN_ADDRESS",
    "DezXAZ8z7PnrnRJjz3wXBoRgixCa6xjnB7YaB1pPB263",  # BONK
)

# ── RPC Helper ──────────────────────────────────────────────────────


def rpc_call(method: str, params: Optional[list] = None) -> dict[str, Any]:
    """Make a JSON-RPC call to Solana with retry.

    Args:
        method: RPC method name.
        params: Method parameters.

    Returns:
        The 'result' field from the response.

    Raises:
        RuntimeError: On persistent RPC error.
    """
    payload = {"jsonrpc": "2.0", "id": 1, "method": method, "params": params or []}

    for attempt in range(3):
        try:
            resp = httpx.post(RPC_URL, json=payload, timeout=30.0)

            if resp.status_code == 429:
                time.sleep(2.0 * (attempt + 1))
                continue

            resp.raise_for_status()
            data = resp.json()

            if "error" in data:
                err = data["error"]
                raise RuntimeError(f"RPC error: {err.get('message')}")

            return data.get("result", {})

        except httpx.TimeoutException:
            if attempt < 2:
                time.sleep(2.0)
                continue
            raise

    raise RuntimeError(f"RPC {method} failed after retries")


# ── Data Fetching ───────────────────────────────────────────────────


def get_token_supply(mint: str) -> dict:
    """Get total supply of a token.

    Args:
        mint: Token mint address.

    Returns:
        Supply dict with amount, decimals, uiAmount.
    """
    result = rpc_call("getTokenSupply", [mint])
    return result.get("value", {})


def get_largest_accounts(mint: str) -> list[dict]:
    """Get top 20 largest token accounts.

    Args:
        mint: Token mint address.

    Returns:
        List of holder dicts sorted by amount descending.
    """
    result = rpc_call("getTokenLargestAccounts", [mint])
    return result.get("value", [])


# ── Concentration Metrics ───────────────────────────────────────────


def top_n_percentage(amounts: list[int], total_supply: int, n: int) -> float:
    """Percentage held by top N holders.

    Args:
        amounts: Sorted list of holder amounts (largest first).
        total_supply: Total token supply.
        n: Number of top holders.

    Returns:
        Percentage (0-100).
    """
    if total_supply == 0:
        return 0.0
    top_n = sum(amounts[:n])
    return top_n / total_supply * 100


def gini_coefficient(amounts: list[int]) -> float:
    """Calculate Gini coefficient for holder distribution.

    Args:
        amounts: List of holder amounts (any order).

    Returns:
        Gini coefficient (0 = equal, 1 = maximally unequal).
    """
    if not amounts or all(a == 0 for a in amounts):
        return 0.0
    sorted_a = sorted(amounts)
    n = len(sorted_a)
    total = sum(sorted_a)
    if total == 0:
        return 0.0
    cumsum = sum((i + 1) * a for i, a in enumerate(sorted_a))
    return (2 * cumsum) / (n * total) - (n + 1) / n


def hhi(amounts: list[int]) -> float:
    """Herfindahl-Hirschman Index for concentration.

    Args:
        amounts: List of holder amounts.

    Returns:
        HHI value (0-10000). Higher = more concentrated.
    """
    total = sum(amounts)
    if total == 0:
        return 0.0
    shares = [a / total * 100 for a in amounts]
    return sum(s ** 2 for s in shares)


def nakamoto_coefficient(amounts: list[int]) -> int:
    """Minimum holders needed for >50% control.

    Args:
        amounts: Sorted list of holder amounts (largest first).

    Returns:
        Number of holders needed for majority control.
    """
    total = sum(amounts)
    if total == 0:
        return 0
    threshold = total * 0.51
    cumulative = 0
    for i, amount in enumerate(amounts):
        cumulative += amount
        if cumulative >= threshold:
            return i + 1
    return len(amounts)


def classify_risk(top_10_pct: float, gini_val: float, hhi_val: float) -> str:
    """Classify concentration risk level.

    Args:
        top_10_pct: Percentage held by top 10.
        gini_val: Gini coefficient.
        hhi_val: HHI value.

    Returns:
        Risk level string.
    """
    if top_10_pct > 80 or hhi_val > 5000:
        return "EXTREME"
    if top_10_pct > 50 or hhi_val > 2500:
        return "HIGH"
    if top_10_pct > 30 or hhi_val > 1500:
        return "MODERATE"
    return "LOW"


# ── Display ─────────────────────────────────────────────────────────


def print_report(
    mint: str,
    supply: dict,
    holders: list[dict],
    total_supply: int,
    amounts: list[int],
) -> None:
    """Print formatted holder analysis report.

    Args:
        mint: Token mint address.
        supply: Token supply info.
        holders: Top holder list from RPC.
        total_supply: Total supply in raw units.
        amounts: Sorted amounts (largest first).
    """
    decimals = supply.get("decimals", 0)
    ui_supply = supply.get("uiAmount", 0)

    print(f"\n{'='*60}")
    print(f"TOKEN HOLDER ANALYSIS")
    print(f"{'='*60}")
    print(f"  Mint:     {mint}")
    print(f"  Supply:   {ui_supply:,.0f} ({decimals} decimals)")

    # Top holders table
    print(f"\n--- Top 20 Holders ---")
    print(f"  {'#':>3}  {'Account':<20} {'Amount':>16} {'% Supply':>10}")
    print(f"  {'─'*3}  {'─'*20} {'─'*16} {'─'*10}")

    for i, h in enumerate(holders, 1):
        addr = h.get("address", "?")
        addr_short = addr[:8] + "..." + addr[-4:]
        amount = int(h.get("amount", 0))
        ui = h.get("uiAmount", 0)
        pct = amount / total_supply * 100 if total_supply > 0 else 0

        amount_str = f"{ui:,.2f}" if ui < 1e9 else f"{ui:,.0f}"
        print(f"  {i:>3}  {addr_short:<20} {amount_str:>16} {pct:>9.2f}%")

    # Concentration metrics
    t1 = top_n_percentage(amounts, total_supply, 1)
    t5 = top_n_percentage(amounts, total_supply, 5)
    t10 = top_n_percentage(amounts, total_supply, 10)
    t20 = top_n_percentage(amounts, total_supply, 20)
    gini_val = gini_coefficient(amounts)
    hhi_val = hhi(amounts)
    naka = nakamoto_coefficient(amounts)

    print(f"\n--- Concentration Metrics ---")
    print(f"  Top 1 holder:   {t1:.2f}%")
    print(f"  Top 5 holders:  {t5:.2f}%")
    print(f"  Top 10 holders: {t10:.2f}%")
    print(f"  Top 20 holders: {t20:.2f}%")
    print(f"  Gini coeff:     {gini_val:.4f}")
    print(f"  HHI:            {hhi_val:.1f}")
    print(f"  Nakamoto coeff: {naka} (holders for >50%)")

    # Risk assessment
    risk = classify_risk(t10, gini_val, hhi_val)
    print(f"\n--- Risk Assessment ---")
    print(f"  Concentration Risk: {risk}")

    if risk == "EXTREME":
        print("  [!!] Extremely concentrated — few wallets control majority")
        print("       High rug/dump risk. Not suitable for significant positions.")
    elif risk == "HIGH":
        print("  [!] Highly concentrated — top holders can significantly impact price")
        print("      Use small position sizes and tight stops.")
    elif risk == "MODERATE":
        print("  [i] Moderate concentration — typical for newer tokens")
        print("      Monitor top holder activity for large sells.")
    else:
        print("  [ok] Well distributed — lower concentration risk")

    # Note about limitations
    print(f"\n--- Notes ---")
    print("  - RPC returns max 20 holders (getTokenLargestAccounts)")
    print("  - Some holders may be pool/program accounts, not individuals")
    print("  - For deeper analysis (100+ holders, bundlers), use SolanaTracker API")

    print()


# ── Main ────────────────────────────────────────────────────────────


def main() -> None:
    """Run token holder analysis."""
    print(f"Analyzing holders for: {TOKEN_ADDRESS}")
    print(f"RPC: {RPC_URL[:40]}...")

    print("Fetching supply...")
    supply = get_token_supply(TOKEN_ADDRESS)
    total_supply = int(supply.get("amount", "0"))

    if total_supply == 0:
        print("Could not fetch token supply. Check the mint address.")
        sys.exit(1)

    time.sleep(0.5)

    print("Fetching top holders...")
    holders = get_largest_accounts(TOKEN_ADDRESS)

    if not holders:
        print("No holder data returned.")
        sys.exit(1)

    amounts = sorted([int(h.get("amount", 0)) for h in holders], reverse=True)

    print_report(TOKEN_ADDRESS, supply, holders, total_supply, amounts)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Scan a Solana wallet for all token holdings using direct RPC.

Fetches SOL balance and all SPL token accounts, then displays a
summary of holdings sorted by value (if price data available via
a simple DexScreener lookup).

Usage:
    python scripts/wallet_scanner.py
    WALLET_ADDRESS="YourWallet..." python scripts/wallet_scanner.py

Dependencies:
    uv pip install httpx

Environment Variables:
    SOLANA_RPC_URL: Solana RPC endpoint (default: public mainnet)
    WALLET_ADDRESS: Wallet public key to scan
"""

import os
import sys
import time
from typing import Any, Optional

import httpx

# ── Configuration ───────────────────────────────────────────────────

RPC_URL = os.getenv("SOLANA_RPC_URL", "https://api.mainnet-beta.solana.com")
WALLET_ADDRESS = os.getenv("WALLET_ADDRESS", "")

if not WALLET_ADDRESS:
    print("Set WALLET_ADDRESS environment variable")
    print("  Example: WALLET_ADDRESS=\"YourWallet...\" python scripts/wallet_scanner.py")
    sys.exit(1)

# ── RPC Helper ──────────────────────────────────────────────────────


def rpc_call(method: str, params: Optional[list] = None) -> dict[str, Any]:
    """Make a JSON-RPC call to Solana.

    Args:
        method: RPC method name.
        params: Method parameters.

    Returns:
        The 'result' field from the response.

    Raises:
        RuntimeError: On RPC error.
    """
    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": method,
        "params": params or [],
    }

    for attempt in range(3):
        try:
            resp = httpx.post(RPC_URL, json=payload, timeout=30.0)

            if resp.status_code == 429:
                time.sleep(2.0 * (attempt + 1))
                continue

            resp.raise_for_status()
            data = resp.json()

            if "error" in data:
                err = data["error"]
                raise RuntimeError(f"RPC error: {err.get('message', 'unknown')}")

            return data.get("result", {})

        except httpx.TimeoutException:
            if attempt < 2:
                time.sleep(2.0)
                continue
            raise

    raise RuntimeError(f"RPC call {method} failed after retries")


def rpc_batch(calls: list[tuple[str, list]]) -> list[dict]:
    """Execute multiple RPC calls in a single HTTP request.

    Args:
        calls: List of (method, params) tuples.

    Returns:
        List of result dicts in the same order.
    """
    payload = [
        {"jsonrpc": "2.0", "id": i, "method": method, "params": params}
        for i, (method, params) in enumerate(calls)
    ]

    resp = httpx.post(RPC_URL, json=payload, timeout=30.0)
    resp.raise_for_status()
    results = resp.json()
    results.sort(key=lambda r: r.get("id", 0))
    return [r.get("result") for r in results]


# ── Token Scanning ──────────────────────────────────────────────────


def get_sol_balance(wallet: str) -> float:
    """Get SOL balance for a wallet.

    Args:
        wallet: Wallet public key.

    Returns:
        SOL balance.
    """
    result = rpc_call("getBalance", [wallet, {"commitment": "confirmed"}])
    lamports = result.get("value", 0)
    return lamports / 1e9


def get_token_accounts(wallet: str) -> list[dict]:
    """Get all SPL token accounts for a wallet.

    Args:
        wallet: Wallet public key.

    Returns:
        List of token account dicts with mint, amount, decimals.
    """
    # SPL Token program
    result = rpc_call("getTokenAccountsByOwner", [
        wallet,
        {"programId": "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA"},
        {"encoding": "jsonParsed"},
    ])

    tokens = []
    for acct in result.get("value", []):
        parsed = acct.get("account", {}).get("data", {}).get("parsed", {}).get("info", {})
        amount_info = parsed.get("tokenAmount", {})
        raw_amount = int(amount_info.get("amount", "0"))

        if raw_amount == 0:
            continue

        tokens.append({
            "mint": parsed.get("mint", ""),
            "account": acct.get("pubkey", ""),
            "amount": raw_amount,
            "decimals": amount_info.get("decimals", 0),
            "ui_amount": amount_info.get("uiAmount", 0),
        })

    # Also check Token-2022
    try:
        result_2022 = rpc_call("getTokenAccountsByOwner", [
            wallet,
            {"programId": "TokenzQdBNbLqP5VEhdkAS6EPFLC1PHnBqCXEpPxuEb"},
            {"encoding": "jsonParsed"},
        ])
        for acct in result_2022.get("value", []):
            parsed = acct.get("account", {}).get("data", {}).get("parsed", {}).get("info", {})
            amount_info = parsed.get("tokenAmount", {})
            raw_amount = int(amount_info.get("amount", "0"))
            if raw_amount == 0:
                continue
            tokens.append({
                "mint": parsed.get("mint", ""),
                "account": acct.get("pubkey", ""),
                "amount": raw_amount,
                "decimals": amount_info.get("decimals", 0),
                "ui_amount": amount_info.get("uiAmount", 0),
                "token_2022": True,
            })
    except Exception:
        pass  # Token-2022 query may fail on some providers

    return tokens


def get_prices_dexscreener(mints: list[str]) -> dict[str, float]:
    """Get USD prices for tokens from DexScreener (free, no auth).

    Args:
        mints: List of token mint addresses.

    Returns:
        Dict of mint -> USD price.
    """
    prices = {}
    # DexScreener supports up to 30 addresses per call
    for i in range(0, len(mints), 30):
        batch = mints[i:i + 30]
        joined = ",".join(batch)
        try:
            resp = httpx.get(
                f"https://api.dexscreener.com/tokens/v1/solana/{joined}",
                timeout=15.0,
            )
            if resp.status_code == 200:
                pairs = resp.json() if isinstance(resp.json(), list) else resp.json().get("pairs", [])
                # Group by base token, take highest liquidity pair
                for pair in pairs:
                    mint = pair.get("baseToken", {}).get("address", "")
                    price = float(pair.get("priceUsd", "0") or "0")
                    liq = (pair.get("liquidity") or {}).get("usd", 0) or 0
                    if mint and price > 0:
                        if mint not in prices or liq > prices.get(f"_liq_{mint}", 0):
                            prices[mint] = price
                            prices[f"_liq_{mint}"] = liq
            time.sleep(1.0)
        except Exception:
            pass

    # Clean up liquidity tracking keys
    return {k: v for k, v in prices.items() if not k.startswith("_liq_")}


# ── Display ─────────────────────────────────────────────────────────


def format_usd(value: float) -> str:
    """Format a USD value for display."""
    if abs(value) >= 1_000_000:
        return f"${value / 1e6:.2f}M"
    elif abs(value) >= 1_000:
        return f"${value / 1e3:.2f}K"
    return f"${value:.2f}"


def print_report(
    wallet: str,
    sol_balance: float,
    tokens: list[dict],
    prices: dict[str, float],
) -> None:
    """Print formatted wallet scan report.

    Args:
        wallet: Wallet address.
        sol_balance: SOL balance.
        tokens: List of token holdings.
        prices: Dict of mint -> USD price.
    """
    # SOL price
    sol_mint = "So11111111111111111111111111111111111111112"
    sol_price = prices.get(sol_mint, 0)
    sol_value = sol_balance * sol_price

    print(f"\n{'='*60}")
    print(f"WALLET SCAN")
    print(f"{'='*60}")
    print(f"  Address: {wallet}")
    print(f"  SOL:     {sol_balance:.4f} SOL", end="")
    if sol_value:
        print(f" ({format_usd(sol_value)})")
    else:
        print()
    print(f"  Tokens:  {len(tokens)} non-zero holdings")

    if not tokens:
        print("\n  No token holdings found.")
        return

    # Enrich with prices and sort by value
    enriched = []
    for t in tokens:
        price = prices.get(t["mint"], 0)
        value = t["ui_amount"] * price if price else 0
        enriched.append({**t, "price": price, "value": value})

    enriched.sort(key=lambda t: t["value"], reverse=True)

    total_value = sol_value + sum(t["value"] for t in enriched)

    print(f"\n--- Token Holdings ---")
    print(f"  {'Mint':<20} {'Amount':>14} {'Price':>12} {'Value':>12}")
    print(f"  {'─'*20} {'─'*14} {'─'*12} {'─'*12}")

    displayed = 0
    for t in enriched:
        mint_short = t["mint"][:8] + "..." + t["mint"][-4:]
        amount_str = f"{t['ui_amount']:.4f}" if t["ui_amount"] < 1e6 else f"{t['ui_amount']:,.0f}"

        if t["price"]:
            price_str = f"${t['price']:.6f}" if t["price"] < 0.01 else f"${t['price']:.4f}"
            value_str = format_usd(t["value"])
        else:
            price_str = "—"
            value_str = "—"

        t2022 = " [T22]" if t.get("token_2022") else ""
        print(f"  {mint_short:<20} {amount_str:>14} {price_str:>12} {value_str:>12}{t2022}")

        displayed += 1
        if displayed >= 20:
            remaining = len(enriched) - displayed
            if remaining > 0:
                print(f"\n  ... and {remaining} more tokens (showing top 20 by value)")
            break

    if total_value > 0:
        print(f"\n  Total Portfolio Value: {format_usd(total_value)}")
        print(f"    SOL:    {format_usd(sol_value)} ({sol_value / total_value * 100:.1f}%)")
        token_value = total_value - sol_value
        print(f"    Tokens: {format_usd(token_value)} ({token_value / total_value * 100:.1f}%)")

    print()


# ── Main ────────────────────────────────────────────────────────────


def main() -> None:
    """Run wallet scanner."""
    print(f"Scanning wallet: {WALLET_ADDRESS}")
    print(f"RPC: {RPC_URL[:40]}...")

    print("Fetching SOL balance...")
    sol_balance = get_sol_balance(WALLET_ADDRESS)

    print("Fetching token accounts...")
    tokens = get_token_accounts(WALLET_ADDRESS)

    # Get prices from DexScreener (free, no auth)
    prices: dict[str, float] = {}
    if tokens:
        mints = list(set(t["mint"] for t in tokens))
        # Add SOL for price reference
        mints.append("So11111111111111111111111111111111111111112")
        print(f"Fetching prices for {len(mints)} tokens...")
        prices = get_prices_dexscreener(mints)

    print_report(WALLET_ADDRESS, sol_balance, tokens, prices)


if __name__ == "__main__":
    main()