On-Chain Analytics

import os
import requests

ETHERSCAN_KEY = os.getenv("ETHERSCAN_KEY", "YourApiKeyToken")
BASE = "https://api.etherscan.io/api"

def eth_get(module, action, **params):
    """Base Etherscan API call."""
    resp = requests.get(BASE, params={
        "module": module,
        "action": action,
        "apikey": ETHERSCAN_KEY,
        **params
    }, timeout=10)
    resp.raise_for_status()
    data = resp.json()
    if data["status"] != "1":
        raise ValueError(f"Etherscan error: {data.get('message', 'Unknown')}")
    return data["result"]

# Test: ETH balance of Ethereum Foundation wallet
ETHFOUND = "0xde0B295669a9FD93d5F28D9Ec85E40f4cb697BAe"
wei = int(eth_get("account", "balance", address=ETHFOUND, tag="latest"))
eth = wei / 1e18
print(f"ETH Foundation balance: {eth:,.4f} ETH")

from datetime import datetime

WATCHED_WALLETS = {
    "ETH Foundation": "0xde0B295669a9FD93d5F28D9Ec85E40f4cb697BAe",
    "Vitalik.eth":    "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045",
    "Binance Hot":    "0x28C6c06298d514Db089934071355E5743bf21d60",
}

def watch_balances(wallets: dict, alert_threshold_eth: float = 1000.0):
    """Check balances for a set of labeled wallets."""
    # Batch request: up to 20 addresses at once
    addrs = ",".join(wallets.values())
    results = eth_get("account", "balancemulti", address=addrs, tag="latest")

    print(f"\n=== Wallet Balances | {datetime.now().strftime('%Y-%m-%d %H:%M')} ===")
    label_map = {v.lower(): k for k, v in wallets.items()}

    for item in results:
        addr = item["account"]
        eth = int(item["balance"]) / 1e18
        label = label_map.get(addr.lower(), addr[:10] + "…")
        flag = " 🚨 LARGE BALANCE" if eth > alert_threshold_eth else ""
        print(f"  {label:<20} {eth:>12,.4f} ETH{flag}")

watch_balances(WATCHED_WALLETS)

=== Wallet Balances | 2026-03-28 14:32 ===
  ETH Foundation       2,145,223.1234 ETH 🚨 LARGE BALANCE
  Vitalik.eth             47,392.5678 ETH 🚨 LARGE BALANCE
  Binance Hot            563,128.4321 ETH 🚨 LARGE BALANCE

import time

def get_recent_txns(address: str, block_count: int = 100) -> list:
    """Fetch recent normal transactions for a wallet."""
    txns = eth_get("account", "txlist",
        address=address,
        startblock=0,
        endblock=99999999,
        page=1,
        offset=block_count,
        sort="desc"
    )
    return txns

def whale_monitor(min_eth: float = 500.0, poll_seconds: int = 300):
    """
    Monitor public large wallets for big outflows.
    Flags any single tx above min_eth.
    """
    WHALE_WALLETS = [
        "0x28C6c06298d514Db089934071355E5743bf21d60",  # Binance Hot
        "0x21a31Ee1afC51d94C2efcCaa2092aD1028285549",  # Binance Cold
    ]
    seen_hashes = set()

    print(f"🐋 Whale monitor started — alerting on txns > {min_eth} ETH")
    while True:
        for addr in WHALE_WALLETS:
            txns = get_recent_txns(addr, block_count=20)
            for tx in txns:
                if tx["hash"] in seen_hashes:
                    continue
                seen_hashes.add(tx["hash"])
                eth_val = int(tx["value"]) / 1e18
                if eth_val >= min_eth:
                    direction = "OUT" if tx["from"].lower() == addr.lower() else "IN"
                    print(f"🚨 WHALE {direction}: {eth_val:,.2f} ETH")
                    print(f"   From: {tx['from'][:20]}…")
                    print(f"   To:   {tx['to'][:20]}…")
                    print(f"   Hash: {tx['hash'][:20]}…\n")
        time.sleep(poll_seconds)

whale_monitor(min_eth=500)

def get_gas_prices() -> dict:
    """
    Returns safe/proposed/fast gas prices in Gwei.
    """
    data = eth_get("gastracker", "gasoracle")
    return {
        "safe":     int(data["SafeGasPrice"]),
        "standard": int(data["ProposeGasPrice"]),
        "fast":     int(data["FastGasPrice"]),
        "base_fee": float(data.get("suggestBaseFee", 0)),
    }

def gas_alert_loop(target_gwei: int = 15, check_interval: int = 60):
    """Alert when gas drops to target level."""
    import smtplib
    from email.mime.text import MIMEText

    print(f"⛽ Gas monitor: alerting when safe gas ≤ {target_gwei} Gwei")
    alerted = False

    while True:
        gas = get_gas_prices()
        print(f"  Gas: safe={gas['safe']} | std={gas['standard']} | fast={gas['fast']} Gwei | "
              f"base={gas['base_fee']:.2f}")

        if gas["safe"] <= target_gwei and not alerted:
            print(f"🟢 LOW GAS ALERT: {gas['safe']} Gwei — good time to transact!")
            alerted = True
        elif gas["safe"] > target_gwei + 5:
            alerted = False  # Reset after gas rises

        time.sleep(check_interval)

gas_alert_loop(target_gwei=20)

def get_token_transfers(address: str, contract_address: str = None) -> list:
    """Fetch ERC-20 token transfer events for a wallet."""
    params = dict(address=address, page=1, offset=50, sort="desc")
    if contract_address:
        params["contractaddress"] = contract_address
    return eth_get("account", "tokentx", **params)

def get_token_balances_via_transfers(address: str) -> dict:
    """
    Infer current token holdings from transfer history.
    Note: Use Moralis/Alchemy for real-time balances.
    This approach works with free Etherscan API.
    """
    txns = get_token_transfers(address)
    holdings = {}
    for tx in txns:
        symbol = tx["tokenSymbol"]
        decimals = int(tx["tokenDecimal"])
        value = int(tx["value"]) / (10 ** decimals)
        if symbol not in holdings:
            holdings[symbol] = {"in": 0.0, "out": 0.0, "contract": tx["contractAddress"]}
        if tx["to"].lower() == address.lower():
            holdings[symbol]["in"] += value
        else:
            holdings[symbol]["out"] += value

    print(f"\nToken flow for {address[:10]}…:")
    for sym, data in sorted(holdings.items()):
        net = data["in"] - data["out"]
        print(f"  {sym:<12} net: {net:>15,.4f}")
    return holdings

vitalik = "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045"
get_token_balances_via_transfers(vitalik)

Token flow for 0xd8dA6BF…:
  USDC         net:           5,234.5000
  DAI          net:          12,100.0000
  USDT         net:             500.2500
  UNI          net:           1,050.0000
  SHIB         net:     100,000,000.0000

{
  "status": "1",        // "1" = success, "0" = error
  "message": "OK",      // "OK" on success, error details on failure
  "result": [...]       // Data array or object
}

def eth_get_safe(module, action, max_retries=3, **params):
    """Robust Etherscan call with retries and error handling."""
    for attempt in range(max_retries):
        try:
            return eth_get(module, action, **params)
        except requests.exceptions.Timeout:
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)  # Exponential backoff
                continue
            raise ValueError("API timeout after retries")
        except ValueError as e:
            if "Max rate limit reached" in str(e):
                time.sleep(1)  # Rate limit; wait before retry
                continue
            raise  # Other errors, re-raise immediately
    raise RuntimeError("Unexpected retry failure")