suix_queryTransactionBlocks - Query Tran...

# Query transactions from a specific address
curl -X POST https://api-sui-mainnet-full.n.dwellir.com/YOUR_API_KEY \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "method": "suix_queryTransactionBlocks",
    "params": [
      {
        "FromAddress": "0xd77955e670601c2c2e6e8637e383695c166aac0a86b741c266bdfb23c2e3369f"
      },
      {
        "showInput": true,
        "showEffects": true,
        "showEvents": true
      },
      null,
      20,
      true
    ],
    "id": 1
  }'

# Query transactions between two addresses
curl -X POST https://api-sui-mainnet-full.n.dwellir.com/YOUR_API_KEY \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "method": "suix_queryTransactionBlocks",
    "params": [
      {
        "FromAndToAddress": {
          "from": "0xd77955e670601c2c2e6e8637e383695c166aac0a86b741c266bdfb23c2e3369f",
          "to": "0x1a2b3c4d5e6f789012345678901234567890123456789012345678901234567890"
        }
      },
      {
        "showEffects": true,
        "showBalanceChanges": true
      },
      null,
      10,
      true
    ],
    "id": 1
  }'

# Query Move function calls
curl -X POST https://api-sui-mainnet-full.n.dwellir.com/YOUR_API_KEY \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "method": "suix_queryTransactionBlocks",
    "params": [
      {
        "MoveFunction": {
          "package": "0x2",
          "module": "coin",
          "function": "split"
        }
      },
      {
        "showInput": true,
        "showEffects": true
      },
      null,
      50,
      true
    ],
    "id": 1
  }'

import { SuiClient } from '@mysten/sui.js/client';

const client = new SuiClient({ 
  url: 'https://api-sui-mainnet-full.n.dwellir.com/YOUR_API_KEY' 
});

// Query transactions from specific address
async function getTransactionsFromAddress(address, limit = 50) {
  const result = await client.queryTransactionBlocks({
    filter: { FromAddress: address },
    options: {
      showInput: true,
      showEffects: true,
      showEvents: true,
      showBalanceChanges: true
    },
    limit: limit,
    order: 'descending'
  });
  
  console.log(`Found ${result.data.length} transactions`);
  return result;
}

// Get transaction history with pagination
async function getCompleteTransactionHistory(address, maxPages = 10) {
  let allTransactions = [];
  let cursor = null;
  let pageCount = 0;
  
  while (pageCount < maxPages) {
    const result = await client.queryTransactionBlocks({
      filter: { FromAddress: address },
      options: {
        showEffects: true,
        showBalanceChanges: true
      },
      cursor: cursor,
      limit: 50,
      order: 'descending'
    });
    
    allTransactions.push(...result.data);
    
    console.log(`Page ${pageCount + 1}: ${result.data.length} transactions`);
    
    if (!result.hasNextPage) break;
    
    cursor = result.nextCursor;
    pageCount++;
  }
  
  return allTransactions;
}

// Query transactions between two addresses
async function getTransactionsBetweenAddresses(fromAddress, toAddress) {
  const result = await client.queryTransactionBlocks({
    filter: {
      FromAndToAddress: {
        from: fromAddress,
        to: toAddress
      }
    },
    options: {
      showInput: true,
      showEffects: true,
      showBalanceChanges: true
    },
    limit: 100,
    order: 'descending'
  });
  
  return result.data;
}

// Query specific Move function calls
async function getMoveCallTransactions(packageId, module, functionName) {
  const result = await client.queryTransactionBlocks({
    filter: {
      MoveFunction: {
        package: packageId,
        module: module,
        function: functionName
      }
    },
    options: {
      showInput: true,
      showEvents: true
    },
    limit: 50,
    order: 'descending'
  });
  
  return result.data;
}

// Advanced transaction analysis
async function analyzeAddressActivity(address, days = 30) {
  const transactions = await getCompleteTransactionHistory(address, 20);
  
  const analysis = {
    totalTransactions: transactions.length,
    successfulTransactions: 0,
    failedTransactions: 0,
    totalGasUsed: 0,
    uniqueRecipients: new Set(),
    transactionTypes: {},
    dailyActivity: {},
    averageGasPerTransaction: 0
  };
  
  const now = Date.now();
  const cutoffTime = now - (days * 24 * 60 * 60 * 1000);
  
  transactions.forEach(txn => {
    const timestamp = parseInt(txn.timestampMs || '0');
    
    // Skip transactions outside time window
    if (timestamp < cutoffTime) return;
    
    // Status analysis
    if (txn.effects?.status?.status === 'success') {
      analysis.successfulTransactions++;
    } else {
      analysis.failedTransactions++;
    }
    
    // Gas analysis
    if (txn.effects?.gasUsed) {
      const gasUsed = parseInt(txn.effects.gasUsed.computationCost) + 
                     parseInt(txn.effects.gasUsed.storageCost) - 
                     parseInt(txn.effects.gasUsed.storageRebate);
      analysis.totalGasUsed += gasUsed;
    }
    
    // Recipients analysis
    txn.balanceChanges?.forEach(change => {
      if (change.owner?.AddressOwner && change.owner.AddressOwner !== address) {
        analysis.uniqueRecipients.add(change.owner.AddressOwner);
      }
    });
    
    // Transaction type analysis
    const txType = txn.transaction?.data?.transaction?.kind || 'Unknown';
    analysis.transactionTypes[txType] = (analysis.transactionTypes[txType] || 0) + 1;
    
    // Daily activity
    const day = new Date(timestamp).toISOString().split('T')[0];
    analysis.dailyActivity[day] = (analysis.dailyActivity[day] || 0) + 1;
  });
  
  analysis.averageGasPerTransaction = analysis.totalGasUsed / 
    Math.max(1, analysis.successfulTransactions + analysis.failedTransactions);
  analysis.uniqueRecipients = analysis.uniqueRecipients.size;
  
  return analysis;
}

// Monitor recent transactions
async function monitorRecentTransactions(address, callback, intervalMs = 10000) {
  let lastSeenDigest = null;
  
  console.log(`Starting transaction monitoring for ${address}`);
  
  const checkForNewTransactions = async () => {
    try {
      const result = await client.queryTransactionBlocks({
        filter: { FromAddress: address },
        options: {
          showEffects: true,
          showBalanceChanges: true
        },
        limit: 10,
        order: 'descending'
      });
      
      if (result.data.length === 0) return;
      
      const latestDigest = result.data[0].digest;
      
      if (lastSeenDigest === null) {
        lastSeenDigest = latestDigest;
        return;
      }
      
      if (latestDigest !== lastSeenDigest) {
        // Find new transactions
        const newTransactions = [];
        for (const txn of result.data) {
          if (txn.digest === lastSeenDigest) break;
          newTransactions.push(txn);
        }
        
        if (newTransactions.length > 0) {
          callback(newTransactions);
          lastSeenDigest = latestDigest;
        }
      }
    } catch (error) {
      console.error('Error monitoring transactions:', error);
    }
  };
  
  // Initial check
  await checkForNewTransactions();
  
  // Set up interval
  const intervalId = setInterval(checkForNewTransactions, intervalMs);
  
  return () => {
    clearInterval(intervalId);
    console.log('Transaction monitoring stopped');
  };
}

// Usage examples
const address = '0xd77955e670601c2c2e6e8637e383695c166aac0a86b741c266bdfb23c2e3369f';

// Get recent transactions
const recentTransactions = await getTransactionsFromAddress(address, 20);
console.log('Recent transactions:', recentTransactions.data.length);

// Analyze activity
const activity = await analyzeAddressActivity(address, 7); // Last 7 days
console.log('Address Activity Analysis:', activity);

// Monitor for new transactions
const stopMonitoring = await monitorRecentTransactions(address, (newTxns) => {
  console.log(`📩 ${newTxns.length} new transactions detected`);
  newTxns.forEach(txn => {
    console.log(`- ${txn.digest}: ${txn.effects?.status?.status || 'unknown'}`);
  });
});

// Stop monitoring after 5 minutes
setTimeout(stopMonitoring, 5 * 60 * 1000);

import requests
import json
from datetime import datetime, timedelta
from typing import List, Dict, Any, Optional, Callable
from dataclasses import dataclass
from collections import defaultdict
import time

@dataclass
class TransactionFilter:
    from_address: Optional[str] = None
    to_address: Optional[str] = None
    input_object: Optional[str] = None
    changed_object: Optional[str] = None
    from_and_to: Optional[Dict[str, str]] = None
    transaction_kind: Optional[str] = None
    move_function: Optional[Dict[str, str]] = None

class SuiTransactionQueryClient:
    def __init__(self, rpc_url: str):
        self.rpc_url = rpc_url
    
    def _build_filter_dict(self, filter_obj: TransactionFilter) -> Dict[str, Any]:
        """Convert TransactionFilter to RPC filter format"""
        filter_dict = {}
        
        if filter_obj.from_address:
            filter_dict["FromAddress"] = filter_obj.from_address
        elif filter_obj.to_address:
            filter_dict["ToAddress"] = filter_obj.to_address
        elif filter_obj.input_object:
            filter_dict["InputObject"] = filter_obj.input_object
        elif filter_obj.changed_object:
            filter_dict["ChangedObject"] = filter_obj.changed_object
        elif filter_obj.from_and_to:
            filter_dict["FromAndToAddress"] = {
                "from": filter_obj.from_and_to["from"],
                "to": filter_obj.from_and_to["to"]
            }
        elif filter_obj.transaction_kind:
            filter_dict["TransactionKind"] = filter_obj.transaction_kind
        elif filter_obj.move_function:
            filter_dict["MoveFunction"] = filter_obj.move_function
        
        return filter_dict
    
    def query_transactions(
        self,
        transaction_filter: TransactionFilter,
        cursor: Optional[str] = None,
        limit: int = 50,
        descending: bool = True,
        show_all: bool = True
    ) -> Dict[str, Any]:
        """Query transactions with filters"""
        
        # Build options
        options = {}
        if show_all:
            options = {
                "showInput": True,
                "showEffects": True,
                "showEvents": True,
                "showObjectChanges": True,
                "showBalanceChanges": True
            }
        
        # Build filter
        filter_dict = self._build_filter_dict(transaction_filter)
        
        payload = {
            "jsonrpc": "2.0",
            "method": "suix_queryTransactionBlocks",
            "params": [
                filter_dict,
                options,
                cursor,
                limit,
                descending
            ],
            "id": 1
        }
        
        response = requests.post(
            self.rpc_url,
            headers={'Content-Type': 'application/json'},
            data=json.dumps(payload)
        )
        
        result = response.json()
        if 'error' in result:
            raise Exception(f"RPC Error: {result['error']}")
        
        return result['result']
    
    def get_all_transactions_paginated(
        self,
        transaction_filter: TransactionFilter,
        max_pages: int = 100,
        limit: int = 50
    ) -> List[Dict[str, Any]]:
        """Get all matching transactions using pagination"""
        
        all_transactions = []
        cursor = None
        page = 0
        
        while page < max_pages:
            result = self.query_transactions(
                transaction_filter,
                cursor=cursor,
                limit=limit,
                descending=True
            )
            
            all_transactions.extend(result['data'])
            print(f"Page {page + 1}: {len(result['data'])} transactions")
            
            if not result['hasNextPage']:
                break
            
            cursor = result['nextCursor']
            page += 1
        
        print(f"Total transactions retrieved: {len(all_transactions)}")
        return all_transactions
    
    def analyze_address_activity(
        self, 
        address: str, 
        days: int = 30
    ) -> Dict[str, Any]:
        """Comprehensive address activity analysis"""
        
        # Get transactions from the address
        filter_obj = TransactionFilter(from_address=address)
        transactions = self.get_all_transactions_paginated(
            filter_obj, 
            max_pages=20
        )
        
        # Filter by time range
        cutoff_time = datetime.now() - timedelta(days=days)
        cutoff_ms = int(cutoff_time.timestamp() * 1000)
        
        recent_transactions = [
            txn for txn in transactions
            if int(txn.get('timestampMs', '0')) > cutoff_ms
        ]
        
        analysis = {
            'address': address,
            'analysis_period_days': days,
            'total_transactions': len(recent_transactions),
            'successful_transactions': 0,
            'failed_transactions': 0,
            'total_gas_used': 0,
            'unique_recipients': set(),
            'transaction_types': defaultdict(int),
            'daily_activity': defaultdict(int),
            'hourly_activity': defaultdict(int),
            'gas_statistics': {
                'min': float('inf'),
                'max': 0,
                'avg': 0,
                'total': 0
            },
            'event_types': defaultdict(int),
            'balance_changes': defaultdict(float)
        }
        
        total_gas = 0
        gas_transactions = 0
        
        for txn in recent_transactions:
            # Status analysis
            status = txn.get('effects', {}).get('status', {}).get('status')
            if status == 'success':
                analysis['successful_transactions'] += 1
            else:
                analysis['failed_transactions'] += 1
            
            # Gas analysis
            gas_used = txn.get('effects', {}).get('gasUsed', {})
            if gas_used:
                computation = int(gas_used.get('computationCost', 0))
                storage = int(gas_used.get('storageCost', 0))
                rebate = int(gas_used.get('storageRebate', 0))
                
                total_gas_cost = computation + storage - rebate
                total_gas += total_gas_cost
                gas_transactions += 1
                
                # Update gas statistics
                gas_stats = analysis['gas_statistics']
                gas_stats['min'] = min(gas_stats['min'], total_gas_cost)
                gas_stats['max'] = max(gas_stats['max'], total_gas_cost)
                gas_stats['total'] += total_gas_cost
            
            # Recipients analysis
            balance_changes = txn.get('balanceChanges', [])
            for change in balance_changes:
                owner = change.get('owner', {})
                address_owner = owner.get('AddressOwner')
                
                if address_owner and address_owner != address:
                    analysis['unique_recipients'].add(address_owner)
                
                # Track balance changes by coin type
                coin_type = change.get('coinType', 'Unknown')
                amount = float(change.get('amount', 0))
                analysis['balance_changes'][coin_type] += amount
            
            # Transaction type analysis
            tx_data = txn.get('transaction', {}).get('data', {})
            tx_kind = tx_data.get('transaction', {}).get('kind', 'Unknown')
            analysis['transaction_types'][tx_kind] += 1
            
            # Event analysis
            events = txn.get('events', [])
            for event in events:
                event_type = event.get('type', 'Unknown')
                analysis['event_types'][event_type] += 1
            
            # Time-based analysis
            timestamp_ms = int(txn.get('timestampMs', 0))
            if timestamp_ms > 0:
                dt = datetime.fromtimestamp(timestamp_ms / 1000)
                day_key = dt.strftime('%Y-%m-%d')
                hour_key = dt.strftime('%H')
                
                analysis['daily_activity'][day_key] += 1
                analysis['hourly_activity'][hour_key] += 1
        
        # Finalize gas statistics
        if gas_transactions > 0:
            analysis['gas_statistics']['avg'] = total_gas / gas_transactions
        else:
            analysis['gas_statistics']['min'] = 0
        
        # Convert sets to counts
        analysis['unique_recipients'] = len(analysis['unique_recipients'])
        analysis['total_gas_used'] = total_gas
        
        # Convert defaultdicts to regular dicts for JSON serialization
        analysis['transaction_types'] = dict(analysis['transaction_types'])
        analysis['daily_activity'] = dict(analysis['daily_activity'])
        analysis['hourly_activity'] = dict(analysis['hourly_activity'])
        analysis['event_types'] = dict(analysis['event_types'])
        analysis['balance_changes'] = dict(analysis['balance_changes'])
        
        return analysis
    
    def find_transactions_between_addresses(
        self,
        from_address: str,
        to_address: str,
        limit: int = 100
    ) -> List[Dict[str, Any]]:
        """Find all transactions between two specific addresses"""
        
        filter_obj = TransactionFilter(
            from_and_to={"from": from_address, "to": to_address}
        )
        
        result = self.query_transactions(
            filter_obj,
            limit=limit,
            descending=True,
            show_all=True
        )
        
        return result['data']
    
    def track_object_interactions(
        self,
        object_id: str,
        interaction_type: str = "changed"
    ) -> List[Dict[str, Any]]:
        """Track all transactions that interacted with a specific object"""
        
        if interaction_type == "input":
            filter_obj = TransactionFilter(input_object=object_id)
        else:  # changed
            filter_obj = TransactionFilter(changed_object=object_id)
        
        transactions = self.get_all_transactions_paginated(filter_obj, max_pages=10)
        
        # Analyze interaction patterns
        interaction_analysis = {
            'object_id': object_id,
            'interaction_type': interaction_type,
            'total_interactions': len(transactions),
            'unique_addresses': set(),
            'interaction_timeline': [],
            'transaction_types': defaultdict(int)
        }
        
        for txn in transactions:
            # Track unique addresses
            sender = txn.get('transaction', {}).get('data', {}).get('sender')
            if sender:
                interaction_analysis['unique_addresses'].add(sender)
            
            # Timeline
            timestamp = int(txn.get('timestampMs', 0))
            if timestamp > 0:
                interaction_analysis['interaction_timeline'].append({
                    'timestamp': timestamp,
                    'digest': txn['digest'],
                    'sender': sender
                })
            
            # Transaction types
            tx_kind = txn.get('transaction', {}).get('data', {}).get('transaction', {}).get('kind', 'Unknown')
            interaction_analysis['transaction_types'][tx_kind] += 1
        
        interaction_analysis['unique_addresses'] = len(interaction_analysis['unique_addresses'])
        interaction_analysis['transaction_types'] = dict(interaction_analysis['transaction_types'])
        
        return transactions, interaction_analysis
    
    def monitor_transactions(
        self,
        transaction_filter: TransactionFilter,
        callback: Callable[[List[Dict[str, Any]]], None],
        interval_seconds: int = 30,
        max_duration_minutes: int = 60
    ):
        """Monitor for new transactions matching filter"""
        
        last_seen_digest = None
        start_time = time.time()
        max_duration_seconds = max_duration_minutes * 60
        
        print(f"Starting transaction monitoring (max {max_duration_minutes} minutes)")
        
        while (time.time() - start_time) < max_duration_seconds:
            try:
                result = self.query_transactions(
                    transaction_filter,
                    limit=10,
                    descending=True,
                    show_all=True
                )
                
                if not result['data']:
                    time.sleep(interval_seconds)
                    continue
                
                latest_digest = result['data'][0]['digest']
                
                if last_seen_digest is None:
                    last_seen_digest = latest_digest
                    print("Monitoring initialized")
                    time.sleep(interval_seconds)
                    continue
                
                if latest_digest != last_seen_digest:
                    # Find new transactions
                    new_transactions = []
                    for txn in result['data']:
                        if txn['digest'] == last_seen_digest:
                            break
                        new_transactions.append(txn)
                    
                    if new_transactions:
                        print(f"📩 {len(new_transactions)} new transactions detected")
                        callback(new_transactions)
                        last_seen_digest = latest_digest
                
                time.sleep(interval_seconds)
                
            except Exception as e:
                print(f"Error in monitoring: {e}")
                time.sleep(interval_seconds)
        
        print("Transaction monitoring completed")

# Usage examples
client = SuiTransactionQueryClient('https://api-sui-mainnet-full.n.dwellir.com/YOUR_API_KEY')

# Example 1: Get transactions from address
address = '0xd77955e670601c2c2e6e8637e383695c166aac0a86b741c266bdfb23c2e3369f'
filter_obj = TransactionFilter(from_address=address)
result = client.query_transactions(filter_obj, limit=20)
print(f"Found {len(result['data'])} transactions from address")

# Example 2: Comprehensive activity analysis
analysis = client.analyze_address_activity(address, days=7)
print(f"\nActivity Analysis for last 7 days:")
print(f"Total transactions: {analysis['total_transactions']}")
print(f"Success rate: {analysis['successful_transactions'] / max(1, analysis['total_transactions']):.2%}")
print(f"Average gas per transaction: {analysis['gas_statistics']['avg']:.0f}")
print(f"Unique recipients: {analysis['unique_recipients']}")

# Example 3: Track Move function calls
move_filter = TransactionFilter(
    move_function={
        "package": "0x2",
        "module": "coin", 
        "function": "split"
    }
)
coin_split_txns = client.query_transactions(move_filter, limit=50)
print(f"\nFound {len(coin_split_txns['data'])} coin split transactions")

# Example 4: Monitor new transactions
def on_new_transactions(transactions):
    for txn in transactions:
        status = txn.get('effects', {}).get('status', {}).get('status', 'unknown')
        print(f"New transaction: {txn['digest'][:16]}... - {status}")

# Monitor for 2 minutes
client.monitor_transactions(
    TransactionFilter(from_address=address),
    on_new_transactions,
    interval_seconds=10,
    max_duration_minutes=2
)