如何使用Python订阅Solana地址日志？

前置准备 #

首先安装Solana Python SDK及依赖：

pip install solana websockets

核心实现代码 #

以下是异步订阅指定地址日志的完整示例，支持过滤仅包含目标地址的日志：

import asyncio
from solana.rpc.websocket_api import connect, LogsSubscribeHandler
from solana.rpc.types import LogsSubscribeOpts, Commitment

# 替换成你要监控的Solana地址
TARGET_ADDRESS = "YOUR_SOLANA_ADDRESS_HERE"

async def subscribe_address_logs():
    # 连接Solana主网RPC节点（也可以替换为测试网：wss://api.devnet.solana.com）
    async with connect("wss://api.mainnet-beta.solana.com") as websocket:
        # 配置订阅选项：仅推送包含目标地址的已确认日志
        opts = LogsSubscribeOpts(
            commitment=Commitment("confirmed"),
            mentions=[TARGET_ADDRESS]
        )
        # 定义日志处理函数
        async def handle_logs(logs_response):
            log_data = logs_response.result.value
            print(f"=== 新日志（Slot: {log_data.slot}） ===")
            print(f"交易签名: {log_data.signature}")
            print(f"日志内容: {log_data.logs}")
            print("------------------------")

        # 启动订阅
        handler = LogsSubscribeHandler(handle_logs, opts)
        await websocket.add_handler(handler)
        
        # 保持连接，持续接收日志
        await asyncio.Future()

if __name__ == "__main__":
    try:
        asyncio.run(subscribe_address_logs())
    except KeyboardInterrupt:
        print("\n订阅已停止")

关键细节说明 #

• 节点选择：示例用主网WebSocket节点，测试阶段可以改用测试网节点地址
• 过滤逻辑：通过mentions参数指定目标地址，RPC节点会只推送涉及该地址的交易日志，避免无关数据
• Commitment级别：设置为confirmed确保日志来自已确认的交易，也可根据需求改为finalized（最终确认）或processed（已处理）
• 日志内容解析：log_data.logs包含交易执行过程中的所有日志信息，包括程序调用、错误提示等，可根据业务需求进一步解析

测试方法 #

1. 替换TARGET_ADDRESS为你的Solana地址
2. 运行代码后，给该地址发送一笔测试交易（主网或测试网）
3. 终端会实时打印该交易的相关日志信息

内容的提问来源于stack exchange，提问作者Shahbaz Ali