Cryptoscan - Python Library for Monitoring (Reception) Crypto Payments

С приветом други!

CryptoScan 2.0 - профессиональная библиотека для мониторинга крипто платежей
Мощный инструмент для отслеживания криптовалютных платежей в РЕАЛЬНОМ ВРЕМЕНИ через WebSocket подписки с автоматическим переключением на polling режим.

Особенности CryptoScan 2.0:

• WebSocket мониторинг в реальном времени - мгновенные уведомления при поступлении платежей
• Умное автоопределение - автоматически выбирает WebSocket или polling режим
• Поддержка 15+ блокчейнов: Ethereum, BSC, Polygon, Arbitrum, Avalanche, Base, Optimism, Solana, Sui, USDT Tron, TON, Bitcoin и другие
• HTTP/2 поддержка для оптимизированных API вызовов
• Автоматическое переподключение с экспоненциальной задержкой
• Контроль подтверждений (confirmations) для критичных платежей
• Регистрация кастомных сетей с PublicNode инфраструктурой (100+ блокчейнов)
• Асинхронная архитектура с полной поддержкой async/await
• Точное определение сумм платежей с Decimal точностью
• Полная поддержка ******-серверов (HTTPS, HTTP, auth)
• Система событий с декораторами для payment и error обработчиков
• Корпоративный уровень надёжности с Tenacity retry логикой
• Мониторинг нескольких кошельков и сетей одновременно
• Интеграция с любыми фреймворками, например Aiogram v3

Установка через pip:

pip install pycryptoscan

Code

pip install pycryptoscan

Базовый пример - реальное время через WebSocket:
​

Мониторинг Ethereum в реальном времени

import asyncio
from cryptoscan import create_monitor

async def main():
# Создаём монитор с автоопределением режима (автоматически использует WebSocket)
monitor = create_monitor(
network="ethereum", # Сеть с WebSocket поддержкой
wallet_address="0xD45F36545b373585a2213427C12AD9af2bEFCE18",
expected_amount="1.0",
min_confirmations=3, # Ждём 3 подтверждения
auto_stop=True
)

# Обработчик платежей
@monitor.on_payment
async def handle_payment(event):
payment = event.payment_info
print(f" Платёж получен: {payment.amount} {payment.currency}")
print(f" Транзакция: {payment.transaction_id}")
print(f" От: {payment.from_address}")
print(f" Блок: #{payment.block_height}")
print(f" Подтверждения: {payment.confirmations}")

# Запускаем мониторинг (автоматически использует WebSocket)
await monitor.start()

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

Python

import asyncio

from cryptoscan import create_monitor



async def main():

    # Создаём монитор с автоопределением режима (автоматически использует WebSocket)

    monitor = create_monitor(

        network="ethereum",  # Сеть с WebSocket поддержкой

        wallet_address="0xD45F36545b373585a2213427C12AD9af2bEFCE18",

        expected_amount="1.0",

        min_confirmations=3,  # Ждём 3 подтверждения

        auto_stop=True

    )



    # Обработчик платежей

    @monitor.on_payment

    async def handle_payment(event):

        payment = event.payment_info

        print(f" Платёж получен: {payment.amount} {payment.currency}")

        print(f" Транзакция: {payment.transaction_id}")

        print(f" От: {payment.from_address}")

        print(f" Блок: #{payment.block_height}")

        print(f" Подтверждения: {payment.confirmations}")



    # Запускаем мониторинг (автоматически использует WebSocket)

    await monitor.start()



if __name__ == "__main__":

    asyncio.run(main())

Поддерживаемые сети:

Популярные сети (Реальное время WebSocket):
Ethereum (ETH) - `ethereum`, `eth`
BSC (BNB) - `bsc`, `bnb`
Polygon (MATIC) - `polygon`, `matic`
Arbitrum (ETH) - `arbitrum`, `arb`
Avalanche (AVAX) - `avalanche`, `avax`
Base (ETH) - `base`
Optimism (ETH) - `optimism`, `op`
Solana (SOL) - `solana`, `sol`
Sui (SUI) - `sui`
Osmosis (OSMO) - `osmosis`
Injective (INJ) - `injective`, `inj`

Дополнительные сети (Polling режим):
Bitcoin (BTC) - `bitcoin`, `btc`
TON - `ton`
USDT Tron (TRC-20) - `usdt_tron`, `trc20`
Aptos (APT) - `aptos`, `apt`

Более 100+ сетей доступно через PublicNode:
Layer 2: Scroll, Linea, Blast, Mantle, Taiko, opBNB
Cosmos: Terra, Kava, Neutron, Celestia, Sei, dYdX
EVM: Gnosis, Moonbeam, Celo, Cronos, PulseChain, Sonic

Мультисетевой мониторинг в реальном времени:
​

Одновременный мониторинг Ethereum, BSC и Polygon

import asyncio
from cryptoscan import create_monitor

async def multi_chain():
# Мониторим несколько сетей одновременно (все через WebSocket)
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"),
]

# Единый обработчик для всех сетей
async def on_payment(event):
print(f" Платёж на {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())

Python

import asyncio

from cryptoscan import create_monitor



async def multi_chain():

    # Мониторим несколько сетей одновременно (все через WebSocket)

    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"),

    ]

   

    # Единый обработчик для всех сетей

    async def on_payment(event):

        print(f" Платёж на {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())

Регистрация кастомных сетей:
​

Добавление новых сетей (Scroll, Blast, Linea)

from cryptoscan import register_network, create_network_config, create_monitor

# Регистрируем Scroll (EVM Layer 2)
scroll_config = create_network_config(
name="scroll",
symbol="ETH",
rpc_url="https://scroll-rpc.publicnode.com",
ws_url="wss://scroll-rpc.publicnode.com", # WebSocket для реального времени
aliases=["scrl"],
address_pattern=r'^0x[a-fA-F0-9]{40}$', # EVM формат
decimals=18,
chain_type="evm"
)

register_network(scroll_config)

# Теперь можно использовать где угодно
monitor = create_monitor("scroll", "0x...", "1.0") # Автоматически использует WebSocket
monitor = create_monitor("scrl", "0x...", "1.0") # Или по алиасу

# Или используйте без регистрации - укажите rpc_url напрямую
monitor = create_monitor(
network="celestia",
wallet_address="celestia1...",
expected_amount="1.0",
rpc_url="https://celestia-rpc.publicnode.com",
ws_url="wss://celestia-rpc.publicnode.com" # Реальное время
)

Python

from cryptoscan import register_network, create_network_config, create_monitor



# Регистрируем Scroll (EVM Layer 2)

scroll_config = create_network_config(

    name="scroll",

    symbol="ETH",

    rpc_url="https://scroll-rpc.publicnode.com",

    ws_url="wss://scroll-rpc.publicnode.com",  # WebSocket для реального времени

    aliases=["scrl"],

    address_pattern=r'^0x[a-fA-F0-9]{40}$',  # EVM формат

    decimals=18,

    chain_type="evm"

)



register_network(scroll_config)



# Теперь можно использовать где угодно

monitor = create_monitor("scroll", "0x...", "1.0")  # Автоматически использует WebSocket

monitor = create_monitor("scrl", "0x...", "1.0")    # Или по алиасу



# Или используйте без регистрации - укажите rpc_url напрямую

monitor = create_monitor(

    network="celestia",

    wallet_address="celestia1...",

    expected_amount="1.0",

    rpc_url="https://celestia-rpc.publicnode.com",

    ws_url="wss://celestia-rpc.publicnode.com"  # Реальное время

)

Продвинутая конфигурация с ******:
​

Настройка ******, HTTP/2 и таймаутов

from cryptoscan import create_monitor, create_user_config

# Создаём конфигурацию с ****** и HTTP/2
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="0xD45F36545b373585a2213427C12AD9af2bEFCE18",
expected_amount="1.0",
user_config=user_config # HTTP/2 включён по умолчанию
)

Python

from cryptoscan import create_monitor, create_user_config



# Создаём конфигурацию с ****** и HTTP/2

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="0xD45F36545b373585a2213427C12AD9af2bEFCE18",

    expected_amount="1.0",

    user_config=user_config  # HTTP/2 включён по умолчанию

)

Интеграция с Telegram ботом (Aiogram v3):
​

Telegram бот с реальным временем

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

logging.basicConfig(level=logging.INFO)

bot = Bot(token="YOUR_BOT_TOKEN")
dp = Dispatcher()

@dp.message(Command("start"))
async def start_handler(message: Message):
await message.answer(
" CryptoScan 2.0 Bot\n\n"
"Мониторинг крипто платежей в реальном времени!\n"
"Использование: /monitor <сеть> <адрес> <сумма>\n\n"
" WebSocket: ethereum, bsc, polygon, arbitrum, solana\n"
" Polling: bitcoin, ton, usdt_tron, aptos"
)

@dp.message(Command("monitor"))
async def monitor_payment(message: Message):
args = message.text.split()[1:]
if len(args) != 3:
await message.answer(
" Неверный формат!\n"
"Использование: /monitor <сеть> <адрес> <сумма>\n\n"
"Пример: /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" Платёж получен!\n\n"
f" Сумма: {payment.amount} {payment.currency}\n"
f" Транзакция: {payment.transaction_id[:16]}...\n"
f" От: {payment.from_address[:16]}...\n"
f" Время: {payment.timestamp}"
)

@monitor.on_error
async def on_error(event):
await message.answer(f" Ошибка мониторинга: {event.error}")

await message.answer(
f" Мониторинг запущен!\n\n"
f"Сеть: {network.upper()}\n"
f"Сумма: {amount}\n"
f"Адрес: {address[:16]}...\n\n"
f"Я уведомлю вас при поступлении платежа!"
)

# Запускаем мониторинг в фоне
asyncio.create_task(monitor.start())

except Exception as e:
await message.answer(f" Ошибка: {str(e)}")

async def main():
await dp.start_polling(bot)

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

Python

from aiogram import Bot, Dispatcher

from aiogram.types import Message

from aiogram.filters import Command

from cryptoscan import create_monitor

import asyncio

import logging



logging.basicConfig(level=logging.INFO)



bot = Bot(token="YOUR_BOT_TOKEN")

dp = Dispatcher()



@dp.message(Command("start"))

async def start_handler(message: Message):

    await message.answer(

        " CryptoScan 2.0 Bot\n\n"

        "Мониторинг крипто платежей в реальном времени!\n"

        "Использование: /monitor <сеть> <адрес> <сумма>\n\n"

        " WebSocket: ethereum, bsc, polygon, arbitrum, solana\n"

        " Polling: bitcoin, ton, usdt_tron, aptos"

    )



@dp.message(Command("monitor"))

async def monitor_payment(message: Message):

    args = message.text.split()[1:]

    if len(args) != 3:

        await message.answer(

            " Неверный формат!\n"

            "Использование: /monitor <сеть> <адрес> <сумма>\n\n"

            "Пример: /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" Платёж получен!\n\n"

                f" Сумма: {payment.amount} {payment.currency}\n"

                f" Транзакция: {payment.transaction_id[:16]}...\n"

                f" От: {payment.from_address[:16]}...\n"

                f" Время: {payment.timestamp}"

            )



        @monitor.on_error

        async def on_error(event):

            await message.answer(f" Ошибка мониторинга: {event.error}")



        await message.answer(

            f" Мониторинг запущен!\n\n"

            f"Сеть: {network.upper()}\n"

            f"Сумма: {amount}\n"

            f"Адрес: {address[:16]}...\n\n"

            f"Я уведомлю вас при поступлении платежа!"

        )



        # Запускаем мониторинг в фоне

        asyncio.create_task(monitor.start())



    except Exception as e:

        await message.answer(f" Ошибка: {str(e)}")



async def main():

    await dp.start_polling(bot)



if __name__ == "__main__":

    asyncio.run(main())

Обработка ошибок и надёжность:
​

Надёжный мониторинг с автоповтором

import asyncio
from cryptoscan import create_monitor, NetworkError, PaymentNotFoundError

async def reliable_monitoring():
monitor = create_monitor(
network="ethereum", # Использует WebSocket с автопереподключением
wallet_address="0xD45F36545b373585a2213427C12AD9af2bEFCE18",
expected_amount="1.0",
max_transactions=20,
min_confirmations=3, # Ждём 3 подтверждения
poll_interval=30.0
)

@monitor.on_payment
async def on_payment(event):
print(f" Платёж подтверждён: {event.payment_info.amount} ETH")

@monitor.on_error
async def on_error(event):
error = event.error
if isinstance(error, NetworkError):
print(f" Сетевая ошибка: {error.message}")
print(" Автоматический повтор (Tenacity retry)...")
else:
print(f" Неожиданная ошибка: {error}")

try:
await monitor.start()
except Exception as e:
print(f" Монитор упал: {e}")
finally:
await monitor.stop()

asyncio.run(reliable_monitoring())

Python

import asyncio

from cryptoscan import create_monitor, NetworkError, PaymentNotFoundError



async def reliable_monitoring():

    monitor = create_monitor(

        network="ethereum",  # Использует WebSocket с автопереподключением

        wallet_address="0xD45F36545b373585a2213427C12AD9af2bEFCE18",

        expected_amount="1.0",

        max_transactions=20,

        min_confirmations=3,  # Ждём 3 подтверждения

        poll_interval=30.0

    )



    @monitor.on_payment

    async def on_payment(event):

        print(f" Платёж подтверждён: {event.payment_info.amount} ETH")



    @monitor.on_error

    async def on_error(event):

        error = event.error

        if isinstance(error, NetworkError):

            print(f" Сетевая ошибка: {error.message}")

            print(" Автоматический повтор (Tenacity retry)...")

        else:

            print(f" Неожиданная ошибка: {error}")



    try:

        await monitor.start()

    except Exception as e:

        print(f" Монитор упал: {e}")

    finally:

        await monitor.stop()



asyncio.run(reliable_monitoring())

Основные функции и классы:

create_monitor()

create_monitor(
network: str, # Имя сети: "ethereum", "polygon", "solana" и т.д.
wallet_address: str, # Адрес кошелька
expected_amount: str | Decimal, # Ожидаемая сумма
poll_interval: float = 15.0, # Интервал опроса (polling режим)
max_transactions: int = 10, # Макс. транзакций для проверки
auto_stop: bool = False, # Остановка после нахождения платежа
rpc_url: str = None, # Кастомный RPC URL (может быть wss://)
realtime: bool = None, # None=авто, True=WebSocket, False=polling
min_confirmations: int = 1, # Мин. подтверждения (по умолчанию: 1)
**kwargs
) -> PaymentMonitor

Code

create_monitor(

    network: str,                    # Имя сети: "ethereum", "polygon", "solana" и т.д.

    wallet_address: str,             # Адрес кошелька

    expected_amount: str | Decimal,  # Ожидаемая сумма

    poll_interval: float = 15.0,     # Интервал опроса (polling режим)

    max_transactions: int = 10,      # Макс. транзакций для проверки

    auto_stop: bool = False,         # Остановка после нахождения платежа

    rpc_url: str = None,             # Кастомный RPC URL (может быть wss://)

    realtime: bool = None,           # None=авто, True=WebSocket, False=polling

    min_confirmations: int = 1,      # Мин. подтверждения (по умолчанию: 1)

    **kwargs

) -> PaymentMonitor

PaymentMonitor

• `async start()` - Запуск мониторинга (автоматически выбирает WebSocket или polling)
• `async stop()` - Остановка мониторинга
• `on_payment(callback)` - Регистрация обработчика платежей (декоратор)
• `on_error(callback)` - Регистрация обработчика ошибок (декоратор)
• `provider` - Доступ к UniversalProvider
• `is_running` - Статус мониторинга
• `monitor_id` - Уникальный идентификатор

PaymentInfo

• `transaction_id` - Хэш транзакции
• `amount` - Сумма платежа (Decimal)
• `currency` - Символ валюты (ETH, BTC, SOL, и т.д.)
• `status` - Статус (PENDING, CONFIRMED, FAILED)
• `timestamp` - Время транзакции
• `block_height` - Номер блока
• `confirmations` - Количество подтверждений
• `fee` - Комиссия транзакции
• `from_address` - Адрес отправителbя
• `to_address` - Адрес получателbя
• `raw_data` - Сырые данные API

Регистрация сетей

• `register_network(config)` - Регистрация новой сети
• `create_network_config(...)` - Создание конфигурации сети
• `list_networks()` - Список доступных сетей

Производительность - WebSocket vs Polling:

WebSocket (Реальное время):
• Задержка: <1с (мгновенно)
• Эффективность: Push-подписки
• Нагрузка: Одно соединение
• Сети: Ethereum, BSC, Polygon, Arbitrum, Solana, Sui, и др.

Polling (HTTP запросы):
• Задержка: 15-30с (интервал опроса)
• Эффективность: Pull-запросы
• Нагрузка: Множественные запросы
• Сети: Bitcoin, TON, USDT-Tron, Aptos

Особенности:
• HTTP/2 для всех HTTP вызовов
• Connection pooling для повторного использования
• Tenacity retry с экспоненциальной задержкой
• Async/await для параллельного мониторинга

Полезные ссылки:

• PyPI:
  

  Client Challenge

  pypi.org
• GitHub: DedInc/cryptoscan
• PublicNode: publicnode.com - 100+ блокчейнов

CryptoScan 2.0 - профессиональный мониторинг крипто платежей в реальном времени!

​