Helius Api

#!/usr/bin/env python3
"""Look up token metadata and holder information via Helius DAS API.

Queries the DAS API for comprehensive token information including metadata,
supply, authorities, and top holders. Useful for pre-trade due diligence
on new tokens.

Usage:
    python scripts/token_lookup.py
    TOKEN_MINT="So11111111111111111111111111111111111111112" python scripts/token_lookup.py

Dependencies:
    uv pip install httpx python-dotenv

Environment Variables:
    HELIUS_API_KEY: Your Helius API key (free tier works)
    TOKEN_MINT: Token mint address to look up
"""

import os
import sys
import time
from typing import Optional

import httpx

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

API_KEY = os.getenv("HELIUS_API_KEY", "")
if not API_KEY:
    print("Set HELIUS_API_KEY environment variable")
    print("  Get a free key at https://dashboard.helius.dev")
    sys.exit(1)

TOKEN_MINT = os.getenv(
    "TOKEN_MINT", "So11111111111111111111111111111111111111112"
)

RPC_URL = f"https://mainnet.helius-rpc.com/?api-key={API_KEY}"
RATE_LIMIT_DELAY = 0.5

# ── API Helper ──────────────────────────────────────────────────────


def das_request(method: str, params: dict, retries: int = 2) -> dict:
    """Make a DAS API request with retry logic.

    Args:
        method: DAS method name.
        params: Method parameters.
        retries: Number of retry attempts.

    Returns:
        The 'result' field from the JSON-RPC response.

    Raises:
        RuntimeError: On DAS error or exhausted retries.
    """
    for attempt in range(retries + 1):
        try:
            resp = httpx.post(
                RPC_URL,
                json={"jsonrpc": "2.0", "id": 1, "method": method, "params": params},
                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:
                raise RuntimeError(f"DAS error: {data['error']}")
            return data["result"]
        except httpx.TimeoutException:
            if attempt < retries:
                time.sleep(1.0)
                continue
            raise
    raise RuntimeError("Max retries exceeded")


# ── Token Analysis ──────────────────────────────────────────────────


def get_token_metadata(mint: str) -> dict:
    """Fetch comprehensive token metadata via getAsset.

    Args:
        mint: Token mint address.

    Returns:
        Parsed metadata dict.
    """
    result = das_request("getAsset", {
        "id": mint,
        "options": {
            "showFungible": True,
            "showCollectionMetadata": True,
        }
    })

    metadata = result.get("content", {}).get("metadata", {})
    token_info = result.get("token_info", {})
    authorities = result.get("authorities", [])
    creators = result.get("creators", [])

    return {
        "mint": mint,
        "name": metadata.get("name", "Unknown"),
        "symbol": metadata.get("symbol", "?"),
        "description": metadata.get("description", ""),
        "interface": result.get("interface", "Unknown"),
        "decimals": token_info.get("decimals", 0),
        "supply": token_info.get("supply", 0),
        "token_program": token_info.get("token_program", ""),
        "mint_authority": token_info.get("mint_authority"),
        "freeze_authority": token_info.get("freeze_authority"),
        "mutable": result.get("mutable", False),
        "burnt": result.get("burnt", False),
        "authorities": [
            {"address": a["address"], "scopes": a.get("scopes", [])}
            for a in authorities
        ],
        "creators": [
            {"address": c["address"], "share": c.get("share", 0), "verified": c.get("verified", False)}
            for c in creators
        ],
        "image": result.get("content", {}).get("links", {}).get("image", ""),
    }


def get_top_holders(mint: str, limit: int = 20) -> list[dict]:
    """Fetch top token holders via getTokenAccounts.

    Args:
        mint: Token mint address.
        limit: Max holders to return.

    Returns:
        List of holder dicts sorted by balance (descending).
    """
    result = das_request("getTokenAccounts", {
        "mint": mint,
        "limit": limit,
        "options": {"showZeroBalance": False},
    })

    holders = []
    for account in result.get("token_accounts", []):
        holders.append({
            "owner": account.get("owner", ""),
            "amount": float(account.get("amount", 0)),
            "address": account.get("address", ""),
        })

    # Sort by amount descending
    holders.sort(key=lambda x: x["amount"], reverse=True)
    return holders[:limit]


def compute_concentration(holders: list[dict], total_supply: int, decimals: int) -> dict:
    """Compute holder concentration metrics.

    Args:
        holders: List of holder dicts with 'amount' field (raw amounts).
        total_supply: Total token supply (raw, not decimalized).
        decimals: Token decimals.

    Returns:
        Concentration metrics dict.
    """
    if not holders or total_supply == 0:
        return {"top10_pct": 0, "top20_pct": 0, "largest_holder_pct": 0}

    supply_float = total_supply / (10 ** decimals)
    amounts = [h["amount"] / (10 ** decimals) for h in holders]

    top10_sum = sum(amounts[:10])
    top20_sum = sum(amounts[:20])

    return {
        "top10_pct": round(top10_sum / supply_float * 100, 2) if supply_float > 0 else 0,
        "top20_pct": round(top20_sum / supply_float * 100, 2) if supply_float > 0 else 0,
        "largest_holder_pct": round(amounts[0] / supply_float * 100, 2) if supply_float > 0 and amounts else 0,
        "holder_count_sampled": len(holders),
    }


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


def format_supply(raw_supply: int, decimals: int) -> str:
    """Format raw supply into human-readable string."""
    value = raw_supply / (10 ** decimals)
    if value >= 1_000_000_000:
        return f"{value / 1_000_000_000:.2f}B"
    elif value >= 1_000_000:
        return f"{value / 1_000_000:.2f}M"
    elif value >= 1_000:
        return f"{value / 1_000:.2f}K"
    return f"{value:.4f}"


def print_token_report(
    metadata: dict,
    holders: list[dict],
    concentration: dict,
) -> None:
    """Print a formatted token analysis report.

    Args:
        metadata: Token metadata dict.
        holders: Top holders list.
        concentration: Concentration metrics.
    """
    print(f"{'=' * 60}")
    print(f"TOKEN REPORT: {metadata['symbol']} ({metadata['name']})")
    print(f"{'=' * 60}")
    print()

    # Metadata
    print("Metadata:")
    print(f"  Mint:            {metadata['mint']}")
    print(f"  Interface:       {metadata['interface']}")
    print(f"  Decimals:        {metadata['decimals']}")
    print(f"  Supply:          {format_supply(metadata['supply'], metadata['decimals'])}")
    print(f"  Mutable:         {metadata['mutable']}")
    print(f"  Token Program:   {metadata['token_program'][:20]}...")

    # Authorities
    print()
    print("Authorities:")
    mint_auth = metadata.get("mint_authority")
    freeze_auth = metadata.get("freeze_authority")
    print(f"  Mint authority:   {mint_auth if mint_auth else 'None (supply locked)'}")
    print(f"  Freeze authority: {freeze_auth if freeze_auth else 'None (cannot freeze)'}")

    if metadata["creators"]:
        print()
        print("Creators:")
        for c in metadata["creators"]:
            verified = "verified" if c["verified"] else "unverified"
            print(f"  {c['address'][:20]}... ({c['share']}%, {verified})")

    # Safety flags
    print()
    print("Safety Indicators:")
    if mint_auth:
        print("  [!] Mint authority exists — supply can be increased")
    else:
        print("  [ok] No mint authority — supply is fixed")
    if freeze_auth:
        print("  [!] Freeze authority exists — tokens can be frozen")
    else:
        print("  [ok] No freeze authority")
    if metadata["mutable"]:
        print("  [!] Metadata is mutable — can be changed")
    else:
        print("  [ok] Metadata is immutable")

    # Holders
    if holders:
        print()
        print(f"Top Holders (sampled {concentration['holder_count_sampled']}):")
        print(f"  Top 10 own:       {concentration['top10_pct']}%")
        print(f"  Top 20 own:       {concentration['top20_pct']}%")
        print(f"  Largest holder:   {concentration['largest_holder_pct']}%")
        print()
        print(f"  {'#':>3} {'Owner':>20} {'Amount':>20}")
        print(f"  {'—'*3} {'—'*20} {'—'*20}")
        for i, h in enumerate(holders[:10], 1):
            owner = h["owner"][:20] if h["owner"] else "unknown"
            amt = format_supply(int(h["amount"]), metadata["decimals"])
            print(f"  {i:>3} {owner:>20} {amt:>20}")

    if concentration["top10_pct"] > 80:
        print()
        print("  [!] HIGH CONCENTRATION — top 10 holders own >80% of supply")
    elif concentration["top10_pct"] > 50:
        print()
        print("  [!] MODERATE CONCENTRATION — top 10 holders own >50% of supply")


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


def main() -> None:
    """Look up token and print analysis report."""
    print(f"Looking up token: {TOKEN_MINT}\n")

    # 1. Get metadata
    print("Fetching metadata...")
    metadata = get_token_metadata(TOKEN_MINT)
    time.sleep(RATE_LIMIT_DELAY)

    # 2. Get holders
    print("Fetching top holders...")
    holders = get_top_holders(TOKEN_MINT, limit=20)
    time.sleep(RATE_LIMIT_DELAY)

    # 3. Compute concentration
    concentration = compute_concentration(
        holders, metadata["supply"], metadata["decimals"]
    )

    # 4. Print report
    print()
    print_token_report(metadata, holders, concentration)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Analyze a Solana wallet using Helius DAS and Enhanced Transactions APIs.

Fetches a wallet's token portfolio via DAS API, retrieves parsed transaction
history, and generates a profile including top holdings, recent swap activity,
preferred DEXes, and trading frequency.

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

Dependencies:
    uv pip install httpx python-dotenv

Environment Variables:
    HELIUS_API_KEY: Your Helius API key (free tier works)
    WALLET_ADDRESS: Solana wallet to analyze (optional, uses example if not set)
"""

import os
import sys
import time
from collections import Counter
from datetime import datetime, timezone
from typing import Optional

import httpx

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

API_KEY = os.getenv("HELIUS_API_KEY", "")
if not API_KEY:
    print("Set HELIUS_API_KEY environment variable")
    print("  Get a free key at https://dashboard.helius.dev")
    sys.exit(1)

WALLET = os.getenv("WALLET_ADDRESS", "")
if not WALLET:
    print("Set WALLET_ADDRESS environment variable")
    print("  export WALLET_ADDRESS='YourSolanaWalletAddress'")
    sys.exit(1)

RPC_URL = f"https://mainnet.helius-rpc.com/?api-key={API_KEY}"
API_URL = f"https://api-mainnet.helius-rpc.com/v0"

# Rate limit delay between API calls (seconds)
RATE_LIMIT_DELAY = 0.5

# ── API Helpers ─────────────────────────────────────────────────────


def das_request(method: str, params: dict) -> dict:
    """Make a DAS API request via JSON-RPC.

    Args:
        method: DAS method name (e.g., 'getAssetsByOwner').
        params: Method parameters.

    Returns:
        The 'result' field from the JSON-RPC response.

    Raises:
        httpx.HTTPStatusError: On non-2xx response.
        KeyError: If response has no 'result'.
    """
    resp = httpx.post(
        RPC_URL,
        json={"jsonrpc": "2.0", "id": 1, "method": method, "params": params},
        timeout=30.0,
    )
    resp.raise_for_status()
    data = resp.json()
    if "error" in data:
        raise RuntimeError(f"DAS error: {data['error']}")
    return data["result"]


def get_enhanced_transactions(
    address: str,
    limit: int = 50,
    tx_type: Optional[str] = None,
) -> list[dict]:
    """Fetch parsed transaction history for an address.

    Args:
        address: Solana address to query.
        limit: Max transactions to return (max 100 per request).
        tx_type: Optional filter (e.g., 'SWAP', 'TRANSFER').

    Returns:
        List of EnhancedTransaction dicts.
    """
    params: dict = {"api-key": API_KEY, "limit": min(limit, 100)}
    if tx_type:
        params["type"] = tx_type

    resp = httpx.get(
        f"{API_URL}/addresses/{address}/transactions",
        params=params,
        timeout=30.0,
    )
    resp.raise_for_status()
    return resp.json()


# ── Analysis Functions ──────────────────────────────────────────────


def fetch_portfolio(wallet: str) -> list[dict]:
    """Fetch all fungible token holdings for a wallet.

    Args:
        wallet: Solana wallet address.

    Returns:
        List of asset dicts with token info.
    """
    result = das_request("getAssetsByOwner", {
        "ownerAddress": wallet,
        "page": 1,
        "limit": 1000,
        "displayOptions": {
            "showFungible": True,
            "showNativeBalance": True,
            "showZeroBalance": False,
        },
    })
    return result.get("items", [])


def analyze_portfolio(assets: list[dict]) -> dict:
    """Analyze a wallet's token portfolio.

    Args:
        assets: List of asset dicts from DAS API.

    Returns:
        Analysis dict with holdings summary.
    """
    fungible = []
    nfts = []

    for asset in assets:
        interface = asset.get("interface", "")
        if interface in ("FungibleToken", "FungibleAsset"):
            name = asset.get("content", {}).get("metadata", {}).get("name", "Unknown")
            symbol = asset.get("content", {}).get("metadata", {}).get("symbol", "?")
            token_info = asset.get("token_info", {})
            decimals = token_info.get("decimals", 0)

            fungible.append({
                "name": name,
                "symbol": symbol,
                "mint": asset.get("id", ""),
                "decimals": decimals,
            })
        else:
            nfts.append(asset)

    return {
        "fungible_count": len(fungible),
        "nft_count": len(nfts),
        "top_holdings": fungible[:20],
    }


def analyze_swap_history(swaps: list[dict]) -> dict:
    """Analyze swap transaction history.

    Args:
        swaps: List of enhanced swap transactions.

    Returns:
        Analysis dict with swap patterns.
    """
    if not swaps:
        return {"total_swaps": 0}

    sources = Counter()
    tokens_traded = Counter()
    hourly_activity = Counter()
    total_fee = 0

    for tx in swaps:
        sources[tx.get("source", "UNKNOWN")] += 1
        total_fee += tx.get("fee", 0)

        # Count tokens involved
        for tt in tx.get("tokenTransfers", []):
            mint = tt.get("mint", "unknown")
            tokens_traded[mint] += 1

        # Hourly distribution
        ts = tx.get("timestamp")
        if ts:
            hour = datetime.fromtimestamp(ts, tz=timezone.utc).hour
            hourly_activity[hour] += 1

    # Time span
    timestamps = [tx.get("timestamp", 0) for tx in swaps if tx.get("timestamp")]
    time_span_hours = 0
    if len(timestamps) >= 2:
        time_span_hours = (max(timestamps) - min(timestamps)) / 3600

    swaps_per_hour = len(swaps) / time_span_hours if time_span_hours > 0 else 0

    return {
        "total_swaps": len(swaps),
        "time_span_hours": round(time_span_hours, 1),
        "swaps_per_hour": round(swaps_per_hour, 2),
        "preferred_dexes": sources.most_common(5),
        "unique_tokens_traded": len(tokens_traded),
        "most_traded_mints": tokens_traded.most_common(5),
        "peak_hours_utc": hourly_activity.most_common(3),
        "total_fees_sol": round(total_fee / 1e9, 6),
    }


def analyze_transfers(transfers: list[dict], wallet: str) -> dict:
    """Analyze transfer patterns.

    Args:
        transfers: List of enhanced transfer transactions.
        wallet: The wallet address for determining direction.

    Returns:
        Analysis dict with transfer patterns.
    """
    if not transfers:
        return {"total_transfers": 0}

    inbound = 0
    outbound = 0

    for tx in transfers:
        for nt in tx.get("nativeTransfers", []):
            if nt.get("toUserAccount") == wallet:
                inbound += nt.get("amount", 0)
            elif nt.get("fromUserAccount") == wallet:
                outbound += nt.get("amount", 0)

    return {
        "total_transfers": len(transfers),
        "sol_received": round(inbound / 1e9, 4),
        "sol_sent": round(outbound / 1e9, 4),
        "net_sol_flow": round((inbound - outbound) / 1e9, 4),
    }


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


def main() -> None:
    """Run wallet analysis and print results."""
    print(f"Analyzing wallet: {WALLET}")
    print(f"{'=' * 60}\n")

    # 1. Fetch portfolio
    print("Fetching portfolio...")
    assets = fetch_portfolio(WALLET)
    portfolio = analyze_portfolio(assets)
    time.sleep(RATE_LIMIT_DELAY)

    print(f"\nPortfolio: {portfolio['fungible_count']} tokens, {portfolio['nft_count']} NFTs")
    if portfolio["top_holdings"]:
        print("\nTop Holdings:")
        for h in portfolio["top_holdings"][:10]:
            print(f"  {h['symbol']:>10} | {h['name']}")

    # 2. Fetch recent swaps
    print("\nFetching swap history...")
    swaps = get_enhanced_transactions(WALLET, limit=100, tx_type="SWAP")
    swap_analysis = analyze_swap_history(swaps)
    time.sleep(RATE_LIMIT_DELAY)

    print(f"\nSwap Activity ({swap_analysis['total_swaps']} swaps):")
    if swap_analysis["total_swaps"] > 0:
        print(f"  Time span: {swap_analysis['time_span_hours']}h")
        print(f"  Rate: {swap_analysis['swaps_per_hour']} swaps/hr")
        print(f"  Unique tokens: {swap_analysis['unique_tokens_traded']}")
        print(f"  Total fees: {swap_analysis['total_fees_sol']} SOL")
        print(f"  Preferred DEXes:")
        for dex, count in swap_analysis["preferred_dexes"]:
            print(f"    {dex}: {count}")
        if swap_analysis["peak_hours_utc"]:
            print(f"  Peak hours (UTC):")
            for hour, count in swap_analysis["peak_hours_utc"]:
                print(f"    {hour:02d}:00 — {count} swaps")

    # 3. Fetch recent transfers
    print("\nFetching transfer history...")
    transfers = get_enhanced_transactions(WALLET, limit=100, tx_type="TRANSFER")
    transfer_analysis = analyze_transfers(transfers, WALLET)
    time.sleep(RATE_LIMIT_DELAY)

    print(f"\nTransfer Activity ({transfer_analysis['total_transfers']} transfers):")
    if transfer_analysis["total_transfers"] > 0:
        print(f"  SOL received: {transfer_analysis['sol_received']}")
        print(f"  SOL sent: {transfer_analysis['sol_sent']}")
        print(f"  Net flow: {transfer_analysis['net_sol_flow']} SOL")

    # 4. Summary
    print(f"\n{'=' * 60}")
    print("WALLET PROFILE SUMMARY")
    print(f"{'=' * 60}")
    print(f"  Address:        {WALLET[:20]}...")
    print(f"  Token holdings: {portfolio['fungible_count']}")
    print(f"  NFT holdings:   {portfolio['nft_count']}")
    print(f"  Recent swaps:   {swap_analysis['total_swaps']}")
    if swap_analysis.get("preferred_dexes"):
        print(f"  Primary DEX:    {swap_analysis['preferred_dexes'][0][0]}")
    print(f"  Trade frequency: {swap_analysis.get('swaps_per_hour', 0)} swaps/hr")


if __name__ == "__main__":
    main()