🛠️ solana_wallet_watcher: Monitor a Solana wallet and ping Discord on new transactions.

#solana #discord #blockchain

Solana Wallet Watcher

This tool monitors a specified Solana wallet address for new transactions and sends a notification to a Discord webhook whenever a transaction is detected. It's ideal for tracking wallet activity in real time—useful for traders, investors, or anyone managing Solana-based assets.

Features

Lightweight and efficient polling of the Solana blockchain via public RPC.
Detects new transactions by comparing recent transaction signatures.
Sends richly formatted alerts to a Discord channel using webhooks.
Configurable polling interval and wallet address.
No external dependencies beyond standard Python libraries and requests.

Setup

Install Dependencies

   pip install requests

Get Discord Webhook
- Open your Discord server settings.
- Go to 'Integrations' > 'Webhooks'.
- Create or select a webhook and copy its URL.
Run the Script

   python main.py --wallet YOUR_SOLANA_WALLET --webhook DISCORD_WEBHOOK_URL --interval 10

Parameters

--wallet: The Solana wallet address to monitor.
--webhook: The Discord webhook URL for sending alerts.
--interval: Polling interval in seconds (default: 10).

Example Alert

The Discord message includes:

Wallet address (with Solana Explorer link)
Transaction signature (linked)
Timestamp

Notes

This tool uses the Solana public RPC (https://api.mainnet-beta.solana.com). For high-frequency use, consider using a dedicated RPC provider.
Only the five most recent transactions are checked per poll.
Does not support parsing token transfers in detail—only detects transaction activity.

License

MIT

import argparse
import json
import time
import urllib.parse
import urllib.request
from typing import List, Dict

# Configuration
def send_discord_webhook(webhook_url: str, message: str):
    data = {"content": message}
    body = json.dumps(data).encode('utf-8')
    headers = {"Content-Type": "application/json"}
    req = urllib.request.Request(webhook_url, data=body, headers=headers, method='POST')
    with urllib.request.urlopen(req) as f:
        if f.status != 204:
            print(f"Failed to send webhook: {f.status}")

def get_wallet_txs(wallet: str) -> List[str]:
    url = "https://api.mainnet-beta.solana.com"
    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "getSignaturesForAddress",
        "params": [wallet, {"limit": 5}]
    }
    body = json.dumps(payload).encode('utf-8')
    req = urllib.request.Request(url, data=body, headers={"Content-Type": "application/json"}, method='POST')
    with urllib.request.urlopen(req) as f:
        response = json.loads(f.read().decode('utf-8'))
        return [entry['signature'] for entry in response['result']]

def main(wallet: str, webhook: str, interval: int):
    print(f"Monitoring wallet: {wallet}")
    seen_txs = set()

    try:
        seen_txs.update(get_wallet_txs(wallet))
        print(f"Initialized with {len(seen_txs)} recent transactions.")
    except Exception as e:
        print(f"Initial fetch failed: {e}")
        return

    while True:
        time.sleep(interval)
        try:
            current_txs = get_wallet_txs(wallet)
            explorer_base = "https://solscan.io/tx/"
            for tx in current_txs:
                if tx not in seen_txs:
                    link = f"<a href="{explorer_base}{tx}">View Transaction</a>"
                    msg = f"🔔 **New Transaction** on {wallet}\n{link}"
                    send_discord_webhook(webhook, msg)
                    seen_txs.add(tx)
        except Exception as e:
            print(f"Polling error: {e}")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='Watch Solana wallet and ping Discord on tx.')
    parser.add_argument('--wallet', required=True, help='Solana wallet address')
    parser.add_argument('--webhook', required=True, help='Discord webhook URL')
    parser.add_argument('--interval', type=int, default=10, help='Polling interval in seconds')
    args = parser.parse_args()
    main(args.wallet, args.webhook, args.interval)

DEV Community