Ejemplo Robusto en Python para Utilizar get_signatures_for_address en el Cliente de Solana

Introducción

Obtener transacciones específicas de una dirección en la blockchain de Solana es una tarea común para desarrolladores que buscan analizar o monitorear actividades recientes. La función get_signatures_for_address del cliente de Solana es una herramienta poderosa para este propósito. Este tutorial detallado te guiará a través de un ejemplo en Python que demuestra un uso robusto de get_signatures_for_address para obtener transacciones en un intervalo de tiempo reciente, específicamente las últimas 48 horas.

Requisitos Previos

Antes de comenzar, asegúrate de tener lo siguiente:

• Python 3.7 o superior: Puedes descargarlo desde python.org.
• Librerías de Python necesarias: Utilizaremos solana-py y asyncio. Instálalas usando pip:

  pip install solana
  pip install asyncio

• Acceso a un nodo RPC de Solana: Puedes usar el endpoint público https://api.mainnet-beta.solana.com o un servicio de terceros como Chainstack para mayor fiabilidad.
• Dirección de Solana: La dirección de la cuenta de Solana de la cual deseas obtener las transacciones.

Configuración del Cliente de Solana

Primero, configuraremos el cliente de Solana que se comunicará con el nodo RPC.

from solana.rpc.api import Client

# Reemplaza con tu endpoint RPC preferido
rpc_url = "https://api.mainnet-beta.solana.com"
client = Client(rpc_url)

Definir la Dirección de la Cuenta

Especifica la dirección de la cuenta de Solana que deseas consultar.

address = "TU_DIRECCIÓN_SOLANA_AQUÍ"

Calcular el Intervalo de Tiempo

Calcularemos el timestamp actual y el de hace 48 horas para filtrar las transacciones dentro de este rango.

from datetime import datetime, timedelta

now = datetime.utcnow()
time_delta = timedelta(hours=48)
start_time = now - time_delta

# Convertir a timestamp en segundos
start_timestamp = int(start_time.timestamp())
end_timestamp = int(now.timestamp())

Obtener Firmas de Transacciones

Utilizaremos get_signatures_for_address para obtener las firmas de las transacciones asociadas a la dirección especificada. Debido a que la API puede devolver un número limitado de resultados por solicitud, implementaremos paginación para asegurar que obtenemos todas las transacciones dentro del intervalo de tiempo deseado.

Implementación Síncrona

A continuación se muestra un ejemplo de cómo implementar esta funcionalidad de manera síncrona:

def get_recent_transactions(client, address, start_timestamp, end_timestamp, limit=1000):
    signatures = []
    before = None

    while True:
        response = client.get_signatures_for_address(
            address,
            limit=limit,
            before=before
        )
        result = response['result']

        if not result:
            break

        for sig in result:
            if sig['blockTime'] is None:
                continue
            if sig['blockTime'] < start_timestamp:
                return signatures
            if start_timestamp <= sig['blockTime'] <= end_timestamp:
                signatures.append(sig)

        before = result[-1]['signature']

    return signatures

# Uso de la función
recent_signatures = get_recent_transactions(client, address, start_timestamp, end_timestamp)
print(f"Transacciones encontradas: {len(recent_signatures)}")
for sig in recent_signatures:
    print(f"Firma: {sig['signature']}, Hora: {datetime.fromtimestamp(sig['blockTime'])}

Implementación Asíncrona

Para mejorar la eficiencia, especialmente cuando se manejan grandes cantidades de transacciones, podemos utilizar una implementación asíncrona:

import asyncio
from solana.rpc.async_api import AsyncClient
from solana.rpc.types import Commitment

async def get_recent_transactions_async(client, address, start_timestamp, end_timestamp, limit=1000):
    signatures = []
    before = None

    while True:
        response = await client.get_signatures_for_address(
            address,
            limit=limit,
            before=before,
            commitment=Commitment("confirmed")
        )
        result = response['result']

        if not result:
            break

        for sig in result:
            if sig['blockTime'] is None:
                continue
            if sig['blockTime'] < start_timestamp:
                return signatures
            if start_timestamp <= sig['blockTime'] <= end_timestamp:
                signatures.append(sig)

        before = result[-1]['signature']

    return signatures

async def main():
    async_client = AsyncClient("https://api.mainnet-beta.solana.com")
    try:
        recent_signatures = await get_recent_transactions_async(
            async_client,
            address,
            start_timestamp,
            end_timestamp
        )
        print(f"Transacciones encontradas: {len(recent_signatures)}")
        for sig in recent_signatures:
            print(f"Firma: {sig['signature']}, Hora: {datetime.fromtimestamp(sig['blockTime'])}")
    except Exception as e:
        print(f"Error: {e}")
    finally:
        await async_client.close()

# Ejecutar el script asíncrono
if __name__ == "__main__":
    asyncio.run(main())

Filtrar Transacciones por Intervalo de Tiempo

Después de obtener las firmas de las transacciones, es esencial filtrar aquellas que se encuentran dentro del intervalo de 48 horas definido anteriormente.

filtered_signatures = [
    sig for sig in recent_signatures
    if start_timestamp <= sig['blockTime'] <= end_timestamp
]

print(f"Transacciones en las últimas 48 horas: {len(filtered_signatures)}")
for sig in filtered_signatures:
    print(f"Firma: {sig['signature']}, Hora: {datetime.fromtimestamp(sig['blockTime'])}")

Consideraciones Importantes

Manejo de Paginación

La función get_signatures_for_address puede devolver un número limitado de resultados por solicitud (hasta 1000). Es crucial implementar paginación correctamente para asegurarse de obtener todas las transacciones dentro del intervalo de tiempo deseado. Esto se logra utilizando el parámetro before para indicar desde dónde continuar la búsqueda en solicitudes posteriores.

Manejo de Errores

Siempre es recomendable implementar manejo de errores para capturar y gestionar posibles fallos en las solicitudes a la API RPC de Solana. Esto incluye manejar excepciones de red, respuestas inválidas o cualquier otro error inesperado.

Limitaciones de la API

Ten en cuenta que las API públicas pueden tener limitaciones de tasa (rate limiting). Si necesitas realizar un gran número de solicitudes, considera la posibilidad de utilizar un servicio de nodo RPC dedicado o implementar mecanismos de control de velocidad para evitar ser bloqueado.

Consistencia de los Datos

Asegúrate de comprender la consistencia de las transacciones obtenidas. Utilizar niveles de compromiso adecuados (como confirmed) garantiza un mayor grado de certeza sobre la inclusión de las transacciones en la blockchain.

Ejecución de los Ejemplos

Una vez que hayas configurado tu entorno y definido tus parámetros, puedes ejecutar los ejemplos de código proporcionados para obtener y filtrar las transacciones de la dirección especificada.

Recursos Adicionales

Conclusión

Obtener transacciones recientes de una dirección en la blockchain de Solana es una tarea esencial para muchos desarrolladores y analistas. Utilizando la función get_signatures_for_address de manera robusta, junto con una implementación adecuada de paginación y filtrado por tiempo, puedes asegurarte de obtener datos precisos y completos. Además, considerar aspectos como el manejo de errores y las limitaciones de la API contribuirá a la fiabilidad y eficiencia de tus aplicaciones.