from typing import List
from datetime import datetime
import json
import requests
from loguru import logger
from dataclasses import dataclass
from datetime import timezone
import time
import random
# Configure loguru logger
logger.add(
"solana_transactions.log",
rotation="500 MB",
retention="10 days",
level="INFO",
format="{time} {level} {message}",
)
# Most reliable RPC endpoints
RPC_ENDPOINTS = [
"https://api.mainnet-beta.solana.com",
"https://rpc.ankr.com/solana",
"https://solana.getblock.io/mainnet",
]
@dataclass
class TransactionError:
"""Data class to represent transaction errors"""
error_type: str
message: str
timestamp: str = datetime.now(timezone.utc).isoformat()
class SolanaAPIException(Exception):
"""Custom exception for Solana API related errors"""
pass
class RPCEndpointManager:
"""Manages RPC endpoints and handles switching between them"""
def __init__(self, endpoints: List[str]):
self.endpoints = endpoints.copy()
self.current_endpoint = self.endpoints[0]
self.last_request_time = 0
self.min_request_interval = 0.2 # Increased minimum interval
self.total_requests = 0
self.max_requests_per_endpoint = 3
def get_endpoint(self) -> str:
"""Get current endpoint with rate limiting"""
now = time.time()
time_since_last = now - self.last_request_time
if time_since_last < self.min_request_interval:
time.sleep(self.min_request_interval - time_since_last)
self.total_requests += 1
if self.total_requests >= self.max_requests_per_endpoint:
self.switch_endpoint()
self.total_requests = 0
self.last_request_time = time.time()
return self.current_endpoint
def switch_endpoint(self) -> str:
"""Switch to next available endpoint"""
current = self.current_endpoint
available_endpoints = [
ep for ep in self.endpoints if ep != current
]
if not available_endpoints:
raise SolanaAPIException("All endpoints exhausted")
self.current_endpoint = random.choice(available_endpoints)
logger.info(f"Switched to endpoint: {self.current_endpoint}")
return self.current_endpoint
def make_request(
endpoint_manager: RPCEndpointManager,
payload: dict,
retry_count: int = 3,
) -> dict:
"""
Makes a request with automatic endpoint switching and error handling.
"""
last_error = None
for attempt in range(retry_count):
try:
endpoint = endpoint_manager.get_endpoint()
response = requests.post(
endpoint,
json=payload,
timeout=10,
headers={"Content-Type": "application/json"},
verify=True, # Ensure SSL verification
)
if response.status_code != 200:
raise SolanaAPIException(
f"HTTP {response.status_code}: {response.text}"
)
data = response.json()
if "error" in data:
error_code = data["error"].get("code")
if error_code == 429: # Rate limit
logger.warning(
"Rate limit hit, switching endpoint..."
)
endpoint_manager.switch_endpoint()
time.sleep(2**attempt) # Exponential backoff
continue
if "message" in data["error"]:
raise SolanaAPIException(
f"RPC error: {data['error']['message']}"
)
return data
except (
requests.exceptions.SSLError,
requests.exceptions.ConnectionError,
) as e:
logger.warning(
f"Connection error with {endpoint}: {str(e)}"
)
endpoint_manager.switch_endpoint()
continue
except Exception as e:
last_error = e
logger.warning(f"Request failed: {str(e)}")
endpoint_manager.switch_endpoint()
time.sleep(1)
continue
raise SolanaAPIException(
f"All retry attempts failed. Last error: {str(last_error)}"
)
def fetch_wallet_transactions(
wallet_address: str, max_transactions: int = 10
) -> str:
"""
Fetches recent transactions for a given Solana wallet address.
Args:
wallet_address (str): The Solana wallet address to fetch transactions for
max_transactions (int, optional): Maximum number of transactions to fetch. Defaults to 10.
Returns:
str: JSON string containing transaction details
"""
try:
if (
not isinstance(wallet_address, str)
or len(wallet_address) != 44
):
raise ValueError(
f"Invalid Solana wallet address format: {wallet_address}"
)
if (
not isinstance(max_transactions, int)
or max_transactions < 1
):
raise ValueError(
"max_transactions must be a positive integer"
)
logger.info(
f"Fetching up to {max_transactions} transactions for wallet: {wallet_address}"
)
endpoint_manager = RPCEndpointManager(RPC_ENDPOINTS)
# Get transaction signatures
signatures_payload = {
"jsonrpc": "2.0",
"id": str(random.randint(1, 1000)),
"method": "getSignaturesForAddress",
"params": [wallet_address, {"limit": max_transactions}],
}
signatures_data = make_request(
endpoint_manager, signatures_payload
)
transactions = signatures_data.get("result", [])
if not transactions:
logger.info("No transactions found for this wallet")
return json.dumps(
{
"success": True,
"transactions": [],
"error": None,
"transaction_count": 0,
},
indent=2,
)
logger.info(f"Found {len(transactions)} transactions")
# Process transactions
enriched_transactions = []
for tx in transactions:
try:
tx_payload = {
"jsonrpc": "2.0",
"id": str(random.randint(1, 1000)),
"method": "getTransaction",
"params": [
tx["signature"],
{
"encoding": "json",
"maxSupportedTransactionVersion": 0,
},
],
}
tx_data = make_request(endpoint_manager, tx_payload)
if "result" in tx_data and tx_data["result"]:
result = tx_data["result"]
enriched_tx = {
"signature": tx["signature"],
"slot": tx["slot"],
"timestamp": tx.get("blockTime"),
"success": not tx.get("err"),
}
if "meta" in result:
enriched_tx["fee"] = result["meta"].get("fee")
if (
"preBalances" in result["meta"]
and "postBalances" in result["meta"]
):
enriched_tx["balance_change"] = sum(
result["meta"]["postBalances"]
) - sum(result["meta"]["preBalances"])
enriched_transactions.append(enriched_tx)
logger.info(
f"Processed transaction {tx['signature'][:8]}..."
)
except Exception as e:
logger.warning(
f"Failed to process transaction {tx['signature']}: {str(e)}"
)
continue
logger.info(
f"Successfully processed {len(enriched_transactions)} transactions"
)
return json.dumps(
{
"success": True,
"transactions": enriched_transactions,
"error": None,
"transaction_count": len(enriched_transactions),
},
indent=2,
)
except Exception as e:
error = TransactionError(
error_type="API_ERROR", message=str(e)
)
logger.error(f"Error: {error.message}")
return json.dumps(
{
"success": False,
"transactions": [],
"error": error.__dict__,
"transaction_count": 0,
},
indent=2,
)
if __name__ == "__main__":
# Example wallet address
wallet = "CtBLg4AX6LQfKVtPPUWqJyQ5cRfHydUwuZZ87rmojA1P"
try:
result = fetch_wallet_transactions(wallet)
print(result)
except Exception as e:
logger.error(f"Failed to fetch transactions: {str(e)}")