Async cryptocurrency payment detection and wallet tracking for Python.
Unlike heavy full-node SDKs (web3.py, solana-py), CryptoScan does exactly one thing: it watches a wallet address and fires a Python callback when money arrives. Built for payment gateways, Telegram bots, and automated billing systems.
- Stablecoin Support: Native parsing for ERC-20 (USDT, USDC on EVM) and SPL tokens (Solana).
- Flexible Matching: Trigger on exact amounts, "at least X", or track ALL incoming transfers.
- Dual Engine: Uses lightweight WebSocket log subscriptions where possible (EVM), falls back to optimized HTTP polling for others (Solana, TRON, Bitcoin).
- Confirmation Control: Set
min_confirmationsto wait for block finality before triggering callbacks. - Enterprise Security: Built-in SSRF protection (blocks private IPs/metadata endpoints) and automated API key masking in logs.
- Resilient: Fully async (
httpx,asyncio) with exponential backoff and auto-reconnects viatenacity.
pip install pycryptoscanFor WebSocket-based real-time monitoring:
pip install pycryptoscan[realtime]If you just want to track a wallet and be notified of any incoming transfer:
import asyncio
from cryptoscan import create_monitor, MatchMode
async def main():
monitor = create_monitor(
network="ethereum",
wallet_address="0xD45F36545b373585a2213427C12AD9af2bEFCE18",
match_mode=MatchMode.ANY
)
@monitor.on_payment
async def handle_payment(event):
info = event.payment_info
print(f"Received: {info.amount} {info.currency} from {info.from_address}")
print(f"TX: {info.transaction_id}")
try:
await monitor.start()
except KeyboardInterrupt:
await monitor.stop()
if __name__ == "__main__":
asyncio.run(main())Wait for a specific stablecoin payment (e.g., user needs to pay at least 50 USDT). The library will automatically scan Transfer logs instead of heavy blocks.
from cryptoscan import create_monitor, MatchMode, TokenConfig
usdt_config = TokenConfig(
contract_address="0xdAC17F958D2ee523a2206206994597C13D831ec7",
symbol="USDT",
decimals=6
)
monitor = create_monitor(
network="ethereum",
wallet_address="0xYOUR_MERCHANT_ADDRESS",
expected_amount="50.0",
match_mode=MatchMode.AT_LEAST,
token_contract=usdt_config,
auto_stop=True
)
@monitor.on_payment
async def release_product(event):
print("Payment confirmed! Releasing digital goods...")CryptoScan supports Solana natively and can monitor SPL token transfers (e.g., USDC on Solana) by parsing preTokenBalances/postTokenBalances.
from cryptoscan import create_monitor, MatchMode, TokenConfig
usdc_solana = TokenConfig(
contract_address="EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v",
symbol="USDC",
decimals=6
)
monitor = create_monitor(
network="solana",
wallet_address="9WzDXwBbmkg8ZTbNMqUxvQRAyrZzDsGYdLVL9zYtAWWM",
expected_amount="25.0",
match_mode=MatchMode.AT_LEAST,
token_contract=usdc_solana,
auto_stop=True
)For chains without WebSocket support (or when you prefer polling), force polling mode and set confirmation thresholds.
monitor = create_monitor(
network="bitcoin",
wallet_address="3DVSCqZdrNJHyu9Le7Sepdh1KgQTNR8reG",
expected_amount="0.00611813",
poll_interval=30.0,
min_confirmations=3,
realtime=False,
auto_stop=True
)The library includes pre-configurations for PublicNode endpoints, but you can inject ANY standard RPC.
| Protocol | Networks | Token Support | Method |
|---|---|---|---|
| EVM | Ethereum, BSC, Polygon (+ any EVM chain via custom config) | Native + ERC-20 | WebSockets (eth_subscribe logs) |
| SVM | Solana | Native + SPL | Polling (getSignaturesForAddress) |
| TRON | TRON | Native + TRC-20 | Polling (TRON Grid REST) |
| TON | TON | Native | Polling (TON Center REST) |
| Other | Bitcoin | Native only | Polling (RPC) |
Want to add any EVM-compatible chain? Just pass rpc_url/ws_url or a NetworkConfig — see below.
Pass custom endpoints directly to create_monitor() without registering anything:
monitor = create_monitor(
network="scroll",
wallet_address="0x...",
expected_amount="1.0",
rpc_url="https://scroll-rpc.publicnode.com",
ws_url="wss://scroll-rpc.publicnode.com"
)You can also pass a NetworkConfig object directly:
from cryptoscan import create_monitor, NetworkConfig
my_chain = NetworkConfig(
name="my-chain",
symbol="ETH",
rpc_url="https://my-chain-rpc.example.com",
ws_url="wss://my-chain-rpc.example.com",
chain_type="evm",
decimals=18,
)
monitor = create_monitor(
network=my_chain,
wallet_address="0x...",
expected_amount="1.0",
)Register custom networks globally so they can be referenced by name (useful for private chains or testnets):
from cryptoscan import register_network, create_network_config
config = create_network_config(
name="local-testnet",
symbol="ETH",
rpc_url="http://localhost:8545",
chain_type="evm",
decimals=18,
)
register_network(config)
monitor = create_monitor("local-testnet", "0x...", "1.0")Use UserConfig to control connection pooling, timeouts, proxy routing, and WebSocket behavior.
from cryptoscan import create_monitor, UserConfig, ProxyConfig
proxy_settings = ProxyConfig(
https_proxy="http://10.10.1.10:3128",
proxy_auth="user:pass",
)
config = UserConfig(
proxy_config=proxy_settings,
timeout=10.0,
max_retries=5,
retry_delay=2.0,
connector_limit=20,
ssl_verify=True,
)
monitor = create_monitor(
network="solana",
wallet_address="...",
expected_amount="1.5",
user_config=config,
)You can also pass timeout and max_retries directly to create_monitor() as shortcuts — they override the corresponding UserConfig fields:
monitor = create_monitor(
network="ethereum",
wallet_address="0x...",
expected_amount="1.0",
timeout=15.0,
max_retries=5,
)The library provides domain-specific exceptions for fine-grained error handling. Strategies catch errors, emit ErrorEvent, and continue retrying — your callbacks are for logging and external integrations.
from cryptoscan import (
create_monitor, NetworkError, CSConnectionError,
CSTimeoutError, PaymentNotFoundError, RPCError,
)
monitor = create_monitor(...)
@monitor.on_error
async def on_error(event):
error = event.error
if isinstance(error, NetworkError):
print(f"Connection instability (auto-retrying): {error}")
elif isinstance(error, RPCError):
print(f"RPC error (code={error.code}): {error}")
else:
print(f"Unexpected error: {error}")Full exception hierarchy:
CryptoScanError
├── NetworkError
│ ├── CSConnectionError
│ ├── CSTimeoutError
│ └── BlockFetchError
├── PaymentNotFoundError
├── ValidationError
├── RPCError
├── ParserError
└── AdapterError
Use get_provider() for direct RPC interaction without running a full monitor (e.g., balance checks, block queries):
import asyncio
from cryptoscan import get_provider
async def main():
provider = get_provider("ethereum")
await provider.connect()
try:
block = await provider.get_block_number()
print(f"Latest block: {block}")
finally:
await provider.close()
asyncio.run(main())If you are running CryptoScan in a long-lived billing microservice, you can collect per-method request metrics and export them as Prometheus text format:
from cryptoscan import enable_global_metrics, get_global_metrics
enable_global_metrics()
# ... monitor runs ...
collector = get_global_metrics()
summary = collector.get_summary()
print(f"Requests: {summary.total_requests} | Failed: {summary.failed_requests}")
print(f"Avg Latency: {summary.avg_response_time_ms:.2f}ms")
print(f"Error Rate: {summary.error_rate:.1%}")
print(collector.export_prometheus())Since Litecoin and Dogecoin are Bitcoin forks, their nodes expose the same RPC API. No custom parser is needed — just create a NetworkConfig with chain_type="bitcoin":
import asyncio
from cryptoscan import create_network_config, register_network, create_monitor, MatchMode
doge_config = create_network_config(
name="dogecoin",
symbol="DOGE",
rpc_url="https://rpc.your-doge-node.com",
chain_type="bitcoin",
decimals=8
)
ltc_config = create_network_config(
name="litecoin",
symbol="LTC",
rpc_url="https://rpc.your-ltc-node.com",
chain_type="bitcoin",
decimals=8
)
register_network(doge_config)
register_network(ltc_config)
async def main():
monitor = create_monitor(
network="dogecoin",
wallet_address="D7q...your_doge_address",
expected_amount="150.0",
match_mode=MatchMode.AT_LEAST,
realtime=False
)
@monitor.on_payment
async def on_payment(event):
info = event.payment_info
print(f"Received {info.amount} {info.currency}! TX: {info.transaction_id}")
await monitor.start()
if __name__ == "__main__":
asyncio.run(main())The library supports any ERC-20 token (Ethereum, BSC, Polygon, etc.) and SPL token (Solana) via TokenConfig. Find the token's contract address and decimals on Etherscan or CoinMarketCap:
from cryptoscan import create_monitor, MatchMode, TokenConfig
pepe_token = TokenConfig(
contract_address="0x6982508145454Ce325dDbE47a25d4ec3d2311933",
symbol="PEPE",
decimals=18
)
monitor = create_monitor(
network="ethereum",
wallet_address="0xYOUR_WALLET_ADDRESS",
expected_amount="1000000.0",
match_mode=MatchMode.AT_LEAST,
token_contract=pepe_token,
auto_stop=True
)For chains that are neither Bitcoin-like nor EVM-compatible, write a custom parser and register it:
from cryptoscan import ChainParser, register_parser, create_network_config
class RippleParser(ChainParser):
async def get_transactions(self, address, limit, expected_amount=None, match_mode=None, token_contract=None):
# Your XRP API logic here
pass
async def get_transaction(self, tx_id):
# Your XRP API logic here
pass
async def get_block_number(self):
return 0
async def get_block_for_payment(self, block_identifier, wallet_address, expected_amount, latest_block_num=None, match_mode=None, token_contract=None):
return None
register_parser("ripple", RippleParser)
xrp_net = create_network_config(
name="ripple_mainnet",
symbol="XRP",
rpc_url="...",
chain_type="ripple"
)CryptoScan includes built-in protections for production deployments:
- SSRF Protection
- Log Masking
- Response Limits: 1MB for WebSocket messages, 10MB for HTTP responses.
- SSL Verification: Enabled by default; can be disabled via
UserConfig(ssl_verify=False).
MIT License. See LICENSE for details.