import requests import json from datetime import datetime from typing import Dict, List, Optional, Tuple from dataclasses import dataclass @dataclass class UndelegationRequest: receiver: str resource: str amount: Optional[float] = None # If None, undelegate all class TronUndelegationManager: def __init__(self, api_key: str): self.api_key = api_key self.base_url = f"https://api-tron-mainnet.n.dwellir.com/{api_key}" def undelegate_resources( self, owner_address: str, receiver_address: str, amount_trx: float, resource: str ) -> Dict: """Create an undelegation transaction""" response = requests.post( f"{self.base_url}/wallet/undelegateresource", json={ "owner_address": owner_address, "receiver_address": receiver_address, "balance": int(amount_trx * 1_000_000), # Convert to SUN "resource": resource, "visible": True } ) response.raise_for_status() return response.json() def get_delegation_index(self, owner_address: str) -> List[str]: """Get list of accounts that have received delegations""" response = requests.post( f"{self.base_url}/wallet/getdelegatedresourceaccountindex", json={ "value": owner_address, "visible": True } ) response.raise_for_status() data = response.json() return data.get('toAccounts', []) def get_delegation_detail( self, from_address: str, to_address: str ) -> Tuple[Dict, Dict]: """Get detailed delegation info between accounts""" response = requests.post( f"{self.base_url}/wallet/getdelegatedresource", json={ "fromAddress": from_address, "toAddress": to_address, "visible": True } ) response.raise_for_status() data = response.json() delegated_resources = data.get('delegatedResource', []) energy_info = {'amount': 0, 'locked': False, 'expire_time': None} bandwidth_info = {'amount': 0, 'locked': False, 'expire_time': None} for resource in delegated_resources: if 'frozen_balance_for_energy' in resource: energy_info['amount'] = resource['frozen_balance_for_energy'] / 1_000_000 if 'expire_time_for_energy' in resource: expire_time = datetime.fromtimestamp(resource['expire_time_for_energy'] / 1000) energy_info['locked'] = expire_time > datetime.now() energy_info['expire_time'] = expire_time if 'frozen_balance_for_bandwidth' in resource: bandwidth_info['amount'] = resource['frozen_balance_for_bandwidth'] / 1_000_000 if 'expire_time_for_bandwidth' in resource: expire_time = datetime.fromtimestamp(resource['expire_time_for_bandwidth'] / 1000) bandwidth_info['locked'] = expire_time > datetime.now() bandwidth_info['expire_time'] = expire_time return energy_info, bandwidth_info def check_undelegation_status( self, owner_address: str, receiver_address: str, resource: str ) -> Dict: """Check if undelegation is possible""" energy_info, bandwidth_info = self.get_delegation_detail( owner_address, receiver_address ) info = energy_info if resource == "ENERGY" else bandwidth_info if info['amount'] == 0: return { 'can_undelegate': False, 'reason': f'No {resource} delegated to this address', 'amount': 0 } if info['locked']: days_remaining = (info['expire_time'] - datetime.now()).days return { 'can_undelegate': False, 'reason': f'Locked for {days_remaining} more days', 'amount': info['amount'], 'unlock_time': info['expire_time'] } return { 'can_undelegate': True, 'reason': 'Ready to undelegate', 'amount': info['amount'] } def get_all_delegations(self, owner_address: str) -> List[Dict]: """Get all active delegations with details""" receivers = self.get_delegation_index(owner_address) delegations = [] for receiver in receivers: energy_info, bandwidth_info = self.get_delegation_detail( owner_address, receiver ) delegations.append({ 'receiver': receiver, 'energy': energy_info, 'bandwidth': bandwidth_info, 'total_delegated': energy_info['amount'] + bandwidth_info['amount'] }) return delegations def batch_undelegate( self, owner_address: str, requests: List[UndelegationRequest] ) -> List[Dict]: """Execute multiple undelegations""" results = [] for req in requests: try: # Check status first status = self.check_undelegation_status( owner_address, req.receiver, req.resource ) if not status['can_undelegate']: results.append({ 'success': False, 'receiver': req.receiver, 'resource': req.resource, 'reason': status['reason'] }) continue # Determine amount amount = req.amount if req.amount else status['amount'] # Execute undelegation tx = self.undelegate_resources( owner_address, req.receiver, amount, req.resource ) results.append({ 'success': True, 'receiver': req.receiver, 'resource': req.resource, 'amount': amount, 'txID': tx['txID'] }) except Exception as e: results.append({ 'success': False, 'receiver': req.receiver, 'resource': req.resource, 'error': str(e) }) return results def reclaim_all_unlocked(self, owner_address: str) -> Dict: """Reclaim all unlocked delegations""" delegations = self.get_all_delegations(owner_address) requests = [] for delegation in delegations: # Check energy if delegation['energy']['amount'] > 0 and not delegation['energy']['locked']: requests.append(UndelegationRequest( receiver=delegation['receiver'], resource='ENERGY', amount=delegation['energy']['amount'] )) # Check bandwidth if delegation['bandwidth']['amount'] > 0 and not delegation['bandwidth']['locked']: requests.append(UndelegationRequest( receiver=delegation['receiver'], resource='BANDWIDTH', amount=delegation['bandwidth']['amount'] )) if not requests: return { 'message': 'No unlocked delegations to reclaim', 'total_reclaimed': 0, 'transactions': [] } results = self.batch_undelegate(owner_address, requests) total_reclaimed = sum( r['amount'] for r in results if r['success'] ) return { 'message': f'Reclaimed {len(requests)} delegations', 'total_reclaimed': total_reclaimed, 'transactions': results } class DelegationAnalyzer: """Analyze and optimize delegation efficiency""" def __init__(self, manager: TronUndelegationManager): self.manager = manager def analyze_utilization( self, owner_address: str ) -> List[Dict]: """Analyze delegation utilization rates""" delegations = self.manager.get_all_delegations(owner_address) analysis = [] for delegation in delegations: # Get receiver's resource usage response = requests.post( f"{self.manager.base_url}/wallet/getaccountresource", json={ "address": delegation['receiver'], "visible": True } ) resources = response.json() # Calculate utilization energy_used = resources.get('EnergyUsed', 0) energy_limit = resources.get('EnergyLimit', 0) net_used = resources.get('NetUsed', 0) net_limit = resources.get('NetLimit', 0) energy_utilization = (energy_used / energy_limit * 100) if energy_limit > 0 else 0 bandwidth_utilization = (net_used / net_limit * 100) if net_limit > 0 else 0 analysis.append({ 'receiver': delegation['receiver'], 'energy': { 'delegated_trx': delegation['energy']['amount'], 'utilization': f"{energy_utilization:.1f}%", 'utilization_rate': energy_utilization, 'locked': delegation['energy']['locked'] }, 'bandwidth': { 'delegated_trx': delegation['bandwidth']['amount'], 'utilization': f"{bandwidth_utilization:.1f}%", 'utilization_rate': bandwidth_utilization, 'locked': delegation['bandwidth']['locked'] } }) return analysis def recommend_optimizations( self, owner_address: str, min_utilization: float = 30.0 ) -> List[Dict]: """Recommend delegations to reclaim based on utilization""" analysis = self.analyze_utilization(owner_address) recommendations = [] for item in analysis: # Energy recommendations if (item['energy']['delegated_trx'] > 0 and item['energy']['utilization_rate'] < min_utilization and not item['energy']['locked']): recommendations.append({ 'action': 'RECLAIM', 'receiver': item['receiver'], 'resource': 'ENERGY', 'amount': item['energy']['delegated_trx'], 'reason': f"Low utilization: {item['energy']['utilization']}", 'potential_savings': item['energy']['delegated_trx'] }) # Bandwidth recommendations if (item['bandwidth']['delegated_trx'] > 0 and item['bandwidth']['utilization_rate'] < min_utilization and not item['bandwidth']['locked']): recommendations.append({ 'action': 'RECLAIM', 'receiver': item['receiver'], 'resource': 'BANDWIDTH', 'amount': item['bandwidth']['delegated_trx'], 'reason': f"Low utilization: {item['bandwidth']['utilization']}", 'potential_savings': item['bandwidth']['delegated_trx'] }) total_savings = sum(r['potential_savings'] for r in recommendations) return { 'recommendations': recommendations, 'total_potential_savings': total_savings, 'summary': f"Can reclaim {total_savings:.2f} TRX from {len(recommendations)} inefficient delegations" } # Example usage def main(): manager = TronUndelegationManager('YOUR_API_KEY') analyzer = DelegationAnalyzer(manager) owner_address = 'TOwnerAddress...' # 1. View all delegations print("=== Current Delegations ===") delegations = manager.get_all_delegations(owner_address) for d in delegations: print(f"\nTo: {d['receiver']}") print(f" Energy: {d['energy']['amount']:.2f} TRX", "(LOCKED)" if d['energy']['locked'] else "(UNLOCKED)") print(f" Bandwidth: {d['bandwidth']['amount']:.2f} TRX", "(LOCKED)" if d['bandwidth']['locked'] else "(UNLOCKED)") print(f" Total: {d['total_delegated']:.2f} TRX") # 2. Check specific undelegation print("\n=== Undelegation Status Check ===") status = manager.check_undelegation_status( owner_address, 'TReceiverAddress...', 'ENERGY' ) print(f"Can undelegate: {status['can_undelegate']}") print(f"Reason: {status['reason']}") print(f"Amount available: {status['amount']:.2f} TRX") # 3. Analyze utilization print("\n=== Delegation Utilization Analysis ===") utilization = analyzer.analyze_utilization(owner_address) for u in utilization: print(f"\n{u['receiver']}:") print(f" Energy: {u['energy']['delegated_trx']:.2f} TRX @ {u['energy']['utilization']} usage") print(f" Bandwidth: {u['bandwidth']['delegated_trx']:.2f} TRX @ {u['bandwidth']['utilization']} usage") # 4. Get optimization recommendations print("\n=== Optimization Recommendations ===") recommendations = analyzer.recommend_optimizations(owner_address, 30.0) print(recommendations['summary']) for rec in recommendations['recommendations']: print(f"\n • {rec['action']}: {rec['resource']} from {rec['receiver']}") print(f" Amount: {rec['amount']:.2f} TRX") print(f" Reason: {rec['reason']}") # 5. Reclaim all unlocked print("\n=== Reclaiming Unlocked Delegations ===") reclaim_result = manager.reclaim_all_unlocked(owner_address) print(reclaim_result['message']) print(f"Total reclaimed: {reclaim_result['total_reclaimed']:.2f} TRX") for tx in reclaim_result['transactions']: if tx['success']: print(f" ✓ {tx['resource']} from {tx['receiver']}: {tx['amount']:.2f} TRX") else: print(f" ✗ {tx['resource']} from {tx['receiver']}: {tx.get('reason', tx.get('error'))}") if __name__ == "__main__": main()
