Мониторинг кошельков маркет-мейкеров в Solana на Python через RPC

import asyncio
import aiohttp
import logging
import sys

# Настройка логирования для вывода в консоль
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger(__name__)

# URL RPC-ноды (для продакшена рекомендуется использовать приватные эндпоинты от Helius, QuickNode или Triton)
RPC_URL = 'https://api.mainnet-beta.solana.com'

# Известные адреса маркет-мейкеров (публичные ключи Solana)
TARGET_WALLETS = {
    'Wintermute_1': 'C98996g5Sg6a6vA1J9vD6vA1J9vD6vA1J9vD6vA1J9vD',
    'Jump_Trading_1': 'JUMP111111111111111111111111111111111111111'
}

async def call_rpc(session: aiohttp.ClientSession, method: str, params: list) -> dict:
    '''Универсальный метод для отправки JSON-RPC запросов.'''
    payload = {
        'jsonrpc': '2.0',
        'id': 1,
        'method': method,
        'params': params
    }
    try:
        async with session.post(RPC_URL, json=payload, timeout=10) as response:
            if response.status == 200:
                result = await response.json()
                if 'error' in result:
                    logger.error(f'Ошибка RPC в методе {method}: {result["error"]}')
                    return {}
                return result.get('result', {})
            else:
                logger.error(f'HTTP ошибка {response.status} при вызове {method}')
    except Exception as e:
        logger.error(f'Исключение при вызове {method}: {e}')
    return {}

async def get_latest_signatures(session: aiohttp.ClientSession, address: str, limit: int = 5) -> list:
    '''Получение последних сигнатур транзакций для адреса.'''
    params = [address, {'limit': limit}]
    return await call_rpc(session, 'getSignaturesForAddress', params) or []

async def get_transaction_details(session: aiohttp.ClientSession, signature: str) -> dict:
    '''Получение подробной информации о транзакции по ее сигнатуре.'''
    params = [
        signature,
        {
            'encoding': 'json',
            'maxSupportedTransactionVersion': 0
        }
    ]
    return await call_rpc(session, 'getTransaction', params) or {}

async def monitor_wallet(session: aiohttp.ClientSession, name: str, address: str):
    '''Асинхронный цикл мониторинга конкретного кошелька.'''
    logger.info(f'Запуск мониторинга кошелька {name} ({address})')
    last_signature = None

    # Первичный запрос для фиксации текущего состояния
    initial_sigs = await get_latest_signatures(session, address, limit=1)
    if initial_sigs:
        last_signature = initial_sigs[0].get('signature')
        logger.info(f'Начальная сигнатура для {name}: {last_signature}')

    while True:
        await asyncio.sleep(2)  # Интервал опроса RPC-ноды
        try:
            signatures = await get_latest_signatures(session, address, limit=5)
            if not signatures:
                continue

            new_signatures = []
            for sig_info in signatures:
                sig = sig_info.get('signature')
                if sig == last_signature:
                    break
                new_signatures.append(sig)

            if new_signatures:
                # Обрабатываем транзакции от старых к новым
                for sig in reversed(new_signatures):
                    logger.info(f'[ОБНАРУЖЕНА АКТИВНОСТЬ] {name} совершил транзакцию!')
                    tx_details = await get_transaction_details(session, sig)
                    if tx_details:
                        logger.info(f'Успешно получены детали транзакции {sig}')
                last_signature = new_signatures[0]

        except Exception as e:
            logger.error(f'Ошибка в цикле мониторинга {name}: {e}')

async def main():
    # Настройка сессии с поддержкой keep-alive соединений
    connector = aiohttp.TCPConnector(limit=100, keepalive_timeout=30)
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [
            monitor_wallet(session, name, addr) 
            for name, addr in TARGET_WALLETS.items()
        ]
        await asyncio.gather(*tasks)

if __name__ == '__main__':
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logger.info('Мониторинг остановлен пользователем.')

pip install aiohttp

python solana_monitor.py