pycryptoscan 2.0.2

Professional Real-Time Crypto Payment Monitoring Library for Python

CryptoScan 🚀

Professional Real-Time Crypto Payment Monitoring Library for Python

A fast, intelligent, production-ready Python library for monitoring cryptocurrency payments in real-time across multiple blockchain networks. Built with WebSocket subscriptions, automatic fallback to polling, HTTP/2 support, and enterprise-grade reliability.

✨ Features

Real-Time Monitoring

• 📡 WebSocket Subscriptions: Instant notifications when new blocks arrive
• 🧠 Smart Auto-Detection: Uses WebSocket if available, falls back to polling
• ⚡ Zero Configuration: Create a monitor - it selects the best mode automatically
• 🔄 Auto-Reconnect: Graceful reconnection with exponential backoff

Network Support

• 🌐 Major Blockchains: Ethereum, BSC, Polygon, Arbitrum, Avalanche, Base, Optimism, Solana, Sui, Osmosis, and more
• 🔥 Real-Time Capable: Major EVM chains, Solana, and Cosmos-based networks
• 🎯 Universal Provider: Single provider works with any blockchain type
• 📡 PublicNode Powered: Access to PublicNode's network infrastructure
• ✨ Easy to Extend: Add any network with simple configuration

Developer Experience

• ⚡ Async/Await: Built for high-performance async applications
• 🚀 HTTP/2: Optimized API calls with HTTP/2 support
• 🔒 Proxy Support: Full proxy configuration (HTTPS, HTTP, auth)
• 🎯 Exact Matching: Precise payment amount detection with Decimal
• 📊 Event System: Payment and error callbacks with decorators
• 🛡️ Production Ready: Tenacity retry logic, comprehensive error handling

🚀 Quick Start

Installation

pip install pycryptoscan

Basic Usage

import asyncio
from cryptoscan import create_monitor

async def main():
    # Create a payment monitor
    # Real-time mode is auto-detected (Ethereum has WebSocket support)
    monitor = create_monitor(
        network="ethereum",
        wallet_address="0xD45F36545b373585a2213427C12AD9af2bEFCE18",
        expected_amount="1.0",
        auto_stop=True,
        min_confirmations=3  # Wait for 3 confirmations before accepting payment
    )

    # Set up payment handler
    @monitor.on_payment
    async def handle_payment(event):
        payment = event.payment_info
        print(f"💰 Payment received: {payment.amount} {payment.currency}")
        print(f"📝 Transaction: {payment.transaction_id}")
        print(f"👤 From: {payment.from_address}")
        print(f"📦 Block: #{payment.block_height}")

    # Start monitoring (uses WebSocket automatically)
    await monitor.start()

if __name__ == "__main__":
    asyncio.run(main())

🌐 Supported Networks

🔥 Popular Networks (Real-Time WebSocket)

Network	Symbol	Usage	Network Type
Ethereum	ETH	ethereum, eth	EVM (Layer 1)
BSC	BNB	bsc, bnb	EVM (Binance)
Polygon	MATIC	polygon, matic	EVM (Layer 2)
Arbitrum	ETH	arbitrum, arb	EVM (Layer 2)
Avalanche	AVAX	avalanche, avax	EVM (Layer 1)
Base	ETH	base	EVM (Coinbase Layer 2)
Optimism	ETH	optimism, op	EVM (Layer 2)
Solana	SOL	solana, sol	Non-EVM (High Performance)
Sui	SUI	sui	Move VM
Osmosis	OSMO	osmosis	Cosmos (DEX)
Injective	INJ	injective, inj	Cosmos (DeFi)

🔵 Additional Networks (Polling Mode)

Network	Symbol	Usage	Network Type
Aptos	APT	aptos, apt	Move VM
Bitcoin	BTC	bitcoin, btc	UTXO
TON	TON	ton	TON VM
USDT-Tron	USDT	usdt_tron, trc20	Tron (TRC-20)

✨ More Networks Available

CryptoScan leverages PublicNode's infrastructure with 100+ blockchain networks including:

• Layer 2s: Scroll, Linea, Blast, Mantle, Taiko, opBNB, Fraxtal
• Cosmos Ecosystem: Cosmos Hub, Terra, Kava, Neutron, Celestia, Sei, dYdX
• Other EVMs: Gnosis, Moonbeam, Celo, Cronos, PulseChain, Sonic
• And many more...

🔥 Real-Time: Instant WebSocket notifications when new blocks arrive
🔵 Polling: HTTP checks every N seconds (fast and reliable)
✨ Easy to Add: Configure any PublicNode network using registration functions

🔧 Network Registration

CryptoScan comes with common networks pre-registered (Ethereum, BSC, Polygon, Solana). You can add custom networks in several ways:

🚀 Pre-Registered Networks (Ready to Use)

These networks are automatically available:

from cryptoscan import list_networks
print("Available networks:", list_networks())
# Output: ['binance', 'bnb', 'bsc', 'eth', 'ethereum', 'matic', 'polygon', 'sol', 'solana']

# Use directly by name or alias
monitor = create_monitor("ethereum", "0x...", "1.0")  # Name
monitor = create_monitor("eth", "0x...", "1.0")       # Alias
monitor = create_monitor("polygon", "0x...", "1.0")   # Name
monitor = create_monitor("matic", "0x...", "1.0")     # Alias

📝 Register Custom Networks

Method 1: Quick Registration

from cryptoscan import register_network, create_network_config

# Register once, use everywhere
scroll_config = create_network_config(
    name="scroll",
    symbol="ETH",
    rpc_url="https://scroll-rpc.publicnode.com",
    ws_url="wss://scroll-rpc.publicnode.com",  # Optional: enables real-time
    aliases=["scrl"],
    address_pattern=r'^0x[a-fA-F0-9]{40}$',  # EVM address format
    decimals=18,
    chain_type="evm"
)

register_network(scroll_config)

# Now use anywhere in your app
monitor = create_monitor("scroll", "0x...", "1.0")  # By name
monitor = create_monitor("scrl", "0x...", "1.0")    # By alias

Method 2: Register Multiple Networks

from cryptoscan import register_network, NetworkConfig

# Define multiple networks
custom_networks = [
    NetworkConfig(
        name="blast", symbol="ETH",
        rpc_url="https://blast-rpc.publicnode.com",
        ws_url="wss://blast-rpc.publicnode.com",
        aliases=["blst"], chain_type="evm", decimals=18
    ),
    NetworkConfig(
        name="linea", symbol="ETH",
        rpc_url="https://linea-rpc.publicnode.com",
        ws_url="wss://linea-rpc.publicnode.com",
        chain_type="evm", decimals=18
    )
]

# Register all at once
for network in custom_networks:
    register_network(network)

# Use them
monitor = create_monitor("blast", "0x...", "1.0")
monitor = create_monitor("linea", "0x...", "1.0")

⚡ Alternative: Direct Usage Without Registration

Skip registration by providing rpc_url directly:

# No registration needed - works immediately
monitor = create_monitor(
    network="any-name",  # Can be anything
    wallet_address="0x...",
    expected_amount="1.0",
    rpc_url="https://your-rpc-endpoint.com",
    ws_url="wss://your-ws-endpoint.com"  # Optional
)

💡 Pro Tips:

• Registration is optional - use rpc_url parameter to skip it
• Pre-registered networks are available immediately after import cryptoscan
• Use list_networks() to see all available networks
• Aliases let you use short names (e.g., "eth" instead of "ethereum")
• Find RPC endpoints at PublicNode.com

🌐 Real-World Examples

🔥 Scroll (EVM Layer 2)

monitor = create_monitor(
    network="scroll",
    wallet_address="0xYourAddress",
    expected_amount="1.0",
    rpc_url="https://scroll-rpc.publicnode.com",
    ws_url="wss://scroll-rpc.publicnode.com"
)

🌟 Celestia (Cosmos)

monitor = create_monitor(
    network="celestia",
    wallet_address="celestia1...",  # Cosmos address format
    expected_amount="1.0",
    rpc_url="https://celestia-rpc.publicnode.com",
    ws_url="wss://celestia-rpc.publicnode.com"
)

🌊 Sei (Parallel Execution)

monitor = create_monitor(
    network="sei",
    wallet_address="sei1...",
    expected_amount="1.0",
    rpc_url="https://sei-rpc.publicnode.com",
    ws_url="wss://sei-rpc.publicnode.com"
)

📚 Examples

Real-Time Monitoring (Auto-Detected)

import asyncio
from cryptoscan import create_monitor

async def realtime_example():
    # WebSocket real-time monitoring (auto-detected)
    monitor = create_monitor(
        network="polygon",  # Has wss:// - uses real-time
        wallet_address="0xD45F36545b373585a2213427C12AD9af2bEFCE18",
        expected_amount="10.0",
        auto_stop=True
    )
    
    @monitor.on_payment
    async def on_payment(event):
        print(f"⚡ Instant notification from new block")
        print(f"💰 {event.payment_info.amount} MATIC received")
        print(f"📦 Block #{event.payment_info.block_height}")
    
    await monitor.start()

asyncio.run(realtime_example())

Multi-Chain Real-Time Monitoring

import asyncio
from cryptoscan import create_monitor

async def multi_chain():
    # Monitor multiple chains simultaneously
    monitors = [
        create_monitor("ethereum", "0x...", "1.0", monitor_id="eth"),
        create_monitor("bsc", "0x...", "0.5", monitor_id="bsc"),
        create_monitor("polygon", "0x...", "10.0", monitor_id="matic"),
    ]
    
    # Unified handler
    async def on_payment(event):
        print(f"💰 Payment on {event.monitor_id}: {event.payment_info.amount}")
    
    for m in monitors:
        m.on_payment(on_payment)
    
    await asyncio.gather(*[m.start() for m in monitors])

asyncio.run(multi_chain())

Basic Payment Monitoring

import asyncio
from decimal import Decimal
from cryptoscan import create_monitor

async def bitcoin_example():
    monitor = create_monitor(
        network="bitcoin",  # or "btc"
        wallet_address="3DVSCqZdrNJHyu9Le7Sepdh1KgQTNR8reG",
        expected_amount=Decimal("0.00611813"),
        poll_interval=30.0,
        auto_stop=True
    )

    @monitor.on_payment
    async def on_payment(event):
        payment = event.payment_info
        print(f"💰 Payment received: {payment.amount} {payment.currency}")
        print(f"   Transaction: {payment.transaction_id}")
        print(f"   From: {payment.from_address}")

    await monitor.start()

asyncio.run(bitcoin_example())

Payment with Confirmation Requirements

import asyncio
from cryptoscan import create_monitor

async def confirmation_example():
    # Monitor payment and wait for 6 confirmations for security
    monitor = create_monitor(
        network="ethereum",
        wallet_address="0xD45F36545b373585a2213427C12AD9af2bEFCE18",
        expected_amount="100.0",
        min_confirmations=6,  # Require 6 confirmations
        auto_stop=True
    )

    @monitor.on_payment
    async def on_payment(event):
        payment = event.payment_info
        print(f"✅ Payment confirmed with {payment.confirmations} confirmations")
        print(f"💰 Amount: {payment.amount} {payment.currency}")
        print(f"🔒 Secure payment received!")

    await monitor.start()

asyncio.run(confirmation_example())

Multi-Network Monitoring

import asyncio
from cryptoscan import create_monitor

async def multi_network_example():
    # Monitor multiple networks simultaneously
    btc_monitor = create_monitor(
        network="bitcoin",
        wallet_address="3DVSCqZdrNJHyu9Le7Sepdh1KgQTNR8reG",
        expected_amount="0.00611813"
    )

    usdt_monitor = create_monitor(
        network="usdt_tron",
        wallet_address="TVRzaRqX9soeRpcJVT6zCAZjGtLtQXacCR",
        expected_amount="200.0"
    )

    # Unified payment handler
    async def handle_payment(event):
        payment = event.payment_info
        print(f"💰 {payment.currency} payment: {payment.amount}")

    btc_monitor.on_payment(handle_payment)
    usdt_monitor.on_payment(handle_payment)

    # Start all monitors
    await asyncio.gather(
        btc_monitor.start(),
        usdt_monitor.start()
    )

asyncio.run(multi_network_example())

🔧 Advanced Configuration

Smart Real-Time Detection

from cryptoscan import create_monitor

# Auto-detects real-time (default behavior)
monitor = create_monitor(
    network="ethereum",  # Has wss:// - uses real-time
    wallet_address="0x...",
    expected_amount="1.0"
)  # Real-time mode enabled

# Custom WebSocket endpoint
monitor = create_monitor(
    network="ethereum",
    wallet_address="0x...",
    expected_amount="1.0",
    rpc_url="wss://eth.llamarpc.com"  # Detects wss:// - enables real-time
)

# Force polling mode (if needed)
monitor = create_monitor(
    network="ethereum",
    wallet_address="0x...",
    expected_amount="1.0",
    realtime=False  # Explicitly use polling
)

User Configuration (Recommended)

from cryptoscan import create_monitor, create_user_config, ProxyConfig

# Create user configuration with proxy and custom settings
user_config = create_user_config(
    proxy_url="https://proxy.example.com:8080",
    proxy_auth="username:password",
    timeout=60,
    max_retries=5,
    ssl_verify=True
)

monitor = create_monitor(
    network="ethereum",
    wallet_address="0x...",
    expected_amount="1.0",
    user_config=user_config
)

Direct UserConfig Creation

from cryptoscan import create_monitor, UserConfig, ProxyConfig

# Create proxy configuration
proxy_config = ProxyConfig(
    https_proxy="https://proxy.example.com:8080",
    proxy_auth="username:password",
    proxy_headers={"Custom-Header": "value"}
)

# Create user configuration
user_config = UserConfig(
    proxy_config=proxy_config,
    timeout=60,
    max_retries=5,
    retry_delay=2.0,
    ssl_verify=True,
    connector_limit=50
)

monitor = create_monitor(
    network="solana",
    wallet_address="39eda9Jzabcr1HPkmjt7sZPCznZqngkfXZn1utwE8uwk",
    expected_amount="0.000542353",
    user_config=user_config
)

Multiple Payment Monitoring

import asyncio
from cryptoscan import create_monitor

async def multi_network_monitoring():
    # Monitor multiple networks simultaneously
    monitors = []

    # Bitcoin monitor
    btc_monitor = create_monitor(
        network="bitcoin",
        wallet_address="3DVSCqZdrNJHyu9Le7Sepdh1KgQTNR8reG",
        expected_amount="0.001",
        monitor_id="btc-payment-1"
    )

    # Ethereum monitor
    eth_monitor = create_monitor(
        network="ethereum",
        wallet_address="0xD45F36545b373585a2213427C12AD9af2bEFCE18",
        expected_amount="0.15",
        monitor_id="eth-payment-1"
    )

    # Unified payment handler
    async def handle_any_payment(event):
        payment = event.payment_info
        monitor_id = event.monitor_id
        print(f"💰 Payment on {monitor_id}: {payment.amount} {payment.currency}")

    btc_monitor.on_payment(handle_any_payment)
    eth_monitor.on_payment(handle_any_payment)

    # Start all monitors
    await asyncio.gather(
        btc_monitor.start(),
        eth_monitor.start()
    )

asyncio.run(multi_network_monitoring())

🛡️ Error Handling & Reliability

Robust Error Handling

import asyncio
from cryptoscan import create_monitor, NetworkError, PaymentNotFoundError

async def reliable_monitoring():
    monitor = create_monitor(
        network="bitcoin",
        wallet_address="3DVSCqZdrNJHyu9Le7Sepdh1KgQTNR8reG",
        expected_amount="0.001",
        max_transactions=20,  # Check more transactions
        poll_interval=30.0
    )

    @monitor.on_payment
    async def on_payment(event):
        print(f"✅ Payment confirmed: {event.payment_info.amount} BTC")

    @monitor.on_error
    async def on_error(event):
        error = event.error
        if isinstance(error, NetworkError):
            print(f"🌐 Network error: {error.message}")
            print("🔄 Will retry automatically...")
        else:
            print(f"❌ Unexpected error: {error}")

    try:
        await monitor.start()
    except Exception as e:
        print(f"💥 Monitor failed: {e}")
    finally:
        await monitor.stop()

asyncio.run(reliable_monitoring())

Timeout and Retry Configuration

from cryptoscan import create_monitor

# High-reliability configuration
monitor = create_monitor(
    network="ethereum",
    wallet_address="0x...",
    expected_amount="1.0",
    poll_interval=15.0,
    timeout=60,  # 60 second timeout
    max_retries=5,  # Retry failed requests 5 times
    auto_stop=True
)

🔌 Integration Examples

Aiogram v3.x (Telegram Bot) Integration

from aiogram import Bot, Dispatcher
from aiogram.types import Message
from aiogram.filters import Command
from cryptoscan import create_monitor
import asyncio
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)

# Initialize bot and dispatcher
bot = Bot(token="YOUR_BOT_TOKEN")
dp = Dispatcher()

@dp.message(Command("start"))
async def start_handler(message: Message):
    await message.answer(
        "🚀 CryptoScan Bot\n\n"
        "Monitor crypto payments with ease!\n"
        "Usage: /monitor <network> <address> <amount>\n\n"
        "Pre-registered: ethereum, bsc, polygon, solana\n"
        "Or use any network with custom RPC URL"
    )

@dp.message(Command("monitor"))
async def monitor_payment(message: Message):
    # Parse command: /monitor ethereum 0x... 1.0
    args = message.text.split()[1:]
    if len(args) != 3:
        await message.answer(
            "❌ Invalid format!\n"
            "Usage: /monitor <network> <address> <amount>\n\n"
            "Example: /monitor ethereum 0xD45F36545b373585a2213427C12AD9af2bEFCE18 1.0"
        )
        return

    network, address, amount = args

    try:
        monitor = create_monitor(
            network=network,
            wallet_address=address,
            expected_amount=amount,
            auto_stop=True
        )

        @monitor.on_payment
        async def on_payment(event):
            payment = event.payment_info
            await message.answer(
                f"✅ Payment Received!\n\n"
                f"💰 Amount: {payment.amount} {payment.currency}\n"
                f"🔗 Transaction: {payment.transaction_id[:16]}...\n"
                f"👤 From: {payment.from_address[:16]}...\n"
                f"⏰ Time: {payment.timestamp}"
            )

        @monitor.on_error
        async def on_error(event):
            await message.answer(f"❌ Monitoring error: {event.error}")

        await message.answer(
            f"🔍 Monitoring started!\n\n"
            f"Network: {network.upper()}\n"
            f"Amount: {amount}\n"
            f"Address: {address[:16]}...\n\n"
            f"I'll notify you when payment is received!"
        )

        # Start monitoring in background
        asyncio.create_task(monitor.start())

    except Exception as e:
        await message.answer(f"❌ Error: {str(e)}")

async def main():
    # Start polling
    await dp.start_polling(bot)

if __name__ == "__main__":
    asyncio.run(main())

📚 API Reference

Core Functions

create_monitor()

Creates a payment monitor for any supported network with smart real-time detection.

def create_monitor(
    network: str,                    # Network name: "ethereum", "polygon", "solana", etc.
    wallet_address: str,             # Wallet address to monitor
    expected_amount: str | Decimal,  # Expected payment amount (exact match)
    poll_interval: float = 15.0,     # Seconds between checks (polling mode)
    max_transactions: int = 10,      # Max transactions to check per poll
    auto_stop: bool = False,         # Stop after finding payment
    rpc_url: str = None,             # Custom RPC URL (can be wss://)
    realtime: bool = None,           # None=auto-detect, True=force, False=polling
    min_confirmations: int = 1,      # Minimum confirmations required (default: 1)
    **kwargs                         # Additional configuration
) -> PaymentMonitor

Real-Time Detection Logic:

• realtime=None (default): Auto-detects based on network WebSocket availability
• rpc_url starts with wss://: Uses real-time mode
• Network has WebSocket configured: Uses real-time mode
• Otherwise: Uses polling mode
• Force mode with realtime=True or realtime=False

Confirmation Handling:

• min_confirmations=1 (default): Accepts payment after 1 confirmation
• Set higher values for critical payments (e.g., min_confirmations=3 or 6)
• Monitor will only trigger payment callback once confirmations meet the threshold

PaymentMonitor Class

Methods

• async start() - Start monitoring for payments
• async stop() - Stop monitoring
• on_payment(callback) - Register payment event handler
• on_error(callback) - Register error event handler

Properties

• provider - Access to the underlying network provider
• is_running - Check if monitor is currently running
• monitor_id - Unique identifier for this monitor

PaymentInfo Class

Payment information returned when a payment is detected.

@dataclass
class PaymentInfo:
    transaction_id: str      # Transaction hash/ID
    wallet_address: str      # Receiving wallet address
    amount: Decimal         # Payment amount in main units
    currency: str           # Currency symbol (BTC, ETH, etc.)
    status: PaymentStatus   # PENDING, CONFIRMED, FAILED
    timestamp: datetime     # Transaction timestamp
    block_height: int       # Block number (if available)
    confirmations: int      # Number of confirmations
    fee: Decimal           # Transaction fee (if available)
    from_address: str      # Sender address
    to_address: str        # Receiver address
    raw_data: dict         # Raw API response data

UniversalProvider

Direct access to the universal provider for advanced use cases.

from cryptoscan import UniversalProvider, NetworkConfig

# Create provider for any network
network_config = NetworkConfig(
    name="ethereum",
    symbol="ETH",
    rpc_url="https://ethereum-rpc.publicnode.com",
    ws_url="wss://ethereum-rpc.publicnode.com",
    chain_type="evm",
    decimals=18
)

provider = UniversalProvider(network=network_config)
await provider.connect()

# Get recent transactions
transactions = await provider.get_recent_transactions(
    "0xD45F36545b373585a2213427C12AD9af2bEFCE18",
    limit=10
)

await provider.close()

🔧 Configuration

Proxy Configuration

from cryptoscan import create_monitor, create_user_config

# Simple proxy configuration
monitor = create_monitor(
    network="ethereum",
    wallet_address="0x...",
    expected_amount="1.0",
    user_config=create_user_config(
        proxy_url="https://proxy.example.com:8080",
        proxy_auth="username:password",
        timeout=60,
        max_retries=5
    )
)

# Advanced proxy configuration
from cryptoscan import UserConfig, ProxyConfig

proxy_config = ProxyConfig(
    https_proxy="https://proxy.example.com:8080",
    http_proxy="http://proxy.example.com:8080",
    proxy_auth="username:password",
    proxy_headers={"Custom-Header": "value"}
)

user_config = UserConfig(
    proxy_config=proxy_config,
    timeout=60,
    max_retries=5
)

monitor = create_monitor(
    network="solana",
    wallet_address="39eda9Jzabcr1HPkmjt7sZPCznZqngkfXZn1utwE8uwk",
    expected_amount="0.1",
    user_config=user_config
)

🚀 Performance

Real-Time vs Polling

Feature	Real-Time (WebSocket)	Polling (HTTP)
Latency	<1s (instant)	15-30s (poll interval)
Efficiency	Push-based	Pull-based
Load	Single connection	Multiple requests
Networks	When ws_url provided	All networks
Auto-Detect	✅ Yes	✅ Yes

Performance Tips

1. Real-Time First: Use networks with WebSocket for instant notifications
2. HTTP/2: All HTTP calls use HTTP/2 for better performance
3. Connection Pooling: Automatic connection reuse reduces overhead
4. Tenacity Retry: Intelligent exponential backoff for high reliability
5. Async Concurrent: Use asyncio.gather() for multi-chain monitoring
6. Optimize Polling: Balance poll interval with responsiveness needs

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

---

Made with ❤️ for the crypto community