Claiming & Swapping Accumulated Fees

Once fees pile up in the collector, the operator can either:

1. Claim Raw Fees

• Function: claim_fees(e, operator: Address, token: Address) -> u128

• Effect: Transfers entire token balance from the contract to fee_destination.

from stellar_sdk import Asset, Keypair, Network, scval, Server, SorobanServer, TransactionBuilder
from stellar_sdk.xdr import Int128Parts, TransactionMeta, UInt128Parts

# =========================================
# Configuration & Setup
# =========================================

# swap provider contract operator
operator_secret_key = "SA..........."
keypair = Keypair.from_secret(operator_secret_key)

# Input and output tokens
token_to_claim = Asset.native()

# swap provider contract for AMM swaps on mainnet
# IMPORTANT: Replace with your deployed contract ID. This contract address is only for test purposes.
provider_swap_contract_id = "CDDIKSXZ4QWQIOQKWJXPLP7QI5XHFVTEF3VG7FOJZPPYWDFNGW7RYJIP"

# Soroban and Horizon servers
soroban_server = SorobanServer("https://mainnet.sorobanrpc.com")
horizon_server = Server("https://horizon.stellar.org/")
network = Network.PUBLIC_NETWORK_PASSPHRASE


# =========================================
# Utility Function
# =========================================

def u128_to_int(value: UInt128Parts) -> int:
    """Convert Uint128Parts to Python int."""
    return (value.hi.uint64 << 64) + value.lo.uint64


def i128_to_int(value: Int128Parts) -> int:
    """Convert int128Parts to Python int."""
    return (value.hi.int64 << 64) + value.lo.uint64


# =========================================
# Entry Point
# =========================================

print("Starting claim process...")
print(f"Claiming {token_to_claim.code}...")

tx = (
    TransactionBuilder(
        source_account=horizon_server.load_account(keypair.public_key),
        network_passphrase=network,
        base_fee=10000
    )
    .set_timeout(300)
    .append_invoke_contract_function_op(
        contract_id=provider_swap_contract_id,
        function_name='claim_fees',
        parameters=[
            scval.to_address(keypair.public_key),
            scval.to_address(token_to_claim.contract_id(network)),
        ],
    )
    .build()
)

# Prepare transaction to get Soroban-specific data (footprint, etc.)
print("Preparing transaction on Soroban...")
prepared_tx = soroban_server.prepare_transaction(tx)

# Sign the prepared transaction
print("Signing transaction...")
prepared_tx.sign(keypair)

# Submit the transaction to Horizon
print("Submitting transaction to Horizon...")
submit_response = horizon_server.submit_transaction(prepared_tx)

if not submit_response.get('successful', False):
    raise Exception("Transaction failed: " + str(submit_response))

print("Transaction submitted successfully. Fetching result...")

# Get the transaction result from Soroban server to access Soroban metadata
tx_info = soroban_server.get_transaction(submit_response['id'])
if not tx_info or not tx_info.result_meta_xdr:
    raise Exception("No transaction metadata found.")

# Extract the result from the Soroban metadata
transaction_meta = TransactionMeta.from_xdr(tx_info.result_meta_xdr)
return_val = transaction_meta.v3.soroban_meta.return_value
final_amount = u128_to_int(return_val.u128)

print("Claim completed successfully!")
print(f"Amount out: {final_amount / 10 ** 7} {token_to_claim.code}")

1. Claim & Immediately Swap

• Function: claim_fees_and_swap(e, operator: Address, swaps_chain: Vec<…>, token: Address, out_min: u128) -> u128

• Effect:

  1. Claims all of token.

  2. Swaps via your router along swaps_chain.

  3. Sends resulting tokens to fee_destination.

• Returns: Final output amount in the target asset.

import requests
from stellar_sdk import Asset, Keypair, Network, scval, Server, SorobanServer, TransactionBuilder
from stellar_sdk.xdr import Int128Parts, SCVal, TransactionMeta, UInt128Parts

# =========================================
# Configuration & Setup
# =========================================

# swap provider contract operator
operator_secret_key = "SA..........."
keypair = Keypair.from_secret(operator_secret_key)

# Input and output tokens
token_to_claim = Asset("AQUA", "GBNZILSTVQZ4R7IKQDGHYGY2QXL5QOFJYQMXPKWRRM5PAV7Y4M67AQUA")
token_out = Asset.native()  # XLM
slippage = "0.005"  # .5% slippage

# swap provider contract for AMM swaps on mainnet
# IMPORTANT: Replace with your deployed contract ID. This contract address is only for test purposes.
provider_swap_contract_id = "CDDIKSXZ4QWQIOQKWJXPLP7QI5XHFVTEF3VG7FOJZPPYWDFNGW7RYJIP"

# Soroban and Horizon servers
soroban_server = SorobanServer("https://mainnet.sorobanrpc.com")
horizon_server = Server("https://horizon.stellar.org/")
network = Network.PUBLIC_NETWORK_PASSPHRASE

# AQUA AMM API endpoint
base_api = 'https://amm-api.aqua.network/api/external/v1'


# =========================================
# Utility Function
# =========================================

def u128_to_int(value: UInt128Parts) -> int:
    """Convert Uint128Parts to Python int."""
    return (value.hi.uint64 << 64) + value.lo.uint64


def i128_to_int(value: Int128Parts) -> int:
    """Convert int128Parts to Python int."""
    return (value.hi.int64 << 64) + value.lo.uint64


# =========================================
# Functions
# =========================================

def find_swap_path(base_api: str, token_in_address: str, token_out_address: str, amount: int, is_send: bool) -> (int, str):
    """
    Call the Find Path API to retrieve the swap chain and estimated amount.
    """
    print("Requesting swap path from AMM API...")
    data = {
        'token_in_address': token_in_address,
        'token_out_address': token_out_address,
        'amount': amount,
        'slippage': slippage,
    }
    endpoint = '/find-path/' if is_send else '/find-path-strict-receive/'
    response = requests.post(f'{base_api}{endpoint}', json=data)
    swap_result = response.json()
    print(swap_result)
    """
        {
          'success': True,
          'swap_chain_xdr': 'AAAAEAAAAAEAAAABAAAAEAAAAAEAAAADAAAAEAAAAAEAAAACAAAAEgAAAAEltPzYWa7C+mNIQ4xImzw8EMmLbSG+T9PLMMtolT75dwAAABIAAAABKIUvaMGYSI40b7EhLtUCkFN2HMJPRTOS41OYIBsIJecAAAANAAAAILLgL8/KbJb4rVy9hOd4Snd7NtnJaiRZQCxPRYRiqrfwAAAAEgAAAAEohS9owZhIjjRvsSEu1QKQU3Ycwk9FM5LjU5ggGwgl5w==',
          'pools': [
            'CDE57N6XTUPBKYYDGQMXX7E7SLNOLFY3JEQB4MULSMR2AKTSAENGX2HC'
          ],
          'tokens': [
            'native',
            'AQUA:GBNZILSTVQZ4R7IKQDGHYGY2QXL5QOFJYQMXPKWRRM5PAV7Y4M67AQUA'
          ],
          'amount': 3627808902,
          'amount_with_fee': 3627808002,
        }
    """

    if not swap_result.get('success', False):
        raise Exception("Failed to retrieve swap path from the API.")

    print("Swap path retrieved. Estimated amount:", swap_result['amount'])
    print("Estimated amount with fee:", swap_result['amount_with_fee'])
    return int(swap_result['amount_with_fee']), swap_result['swap_chain_xdr']


def get_contract_balance(
        network: str,
        soroban_rpc_server: SorobanServer,
        keypair: Keypair,
        contract_address: str,
        token_address: str,
) -> int:
    """
    Retrieves the balance of a specific token for the given account.
    """
    tx = (
        TransactionBuilder(
            source_account=soroban_rpc_server.load_account(keypair.public_key),
            network_passphrase=network,
            base_fee=10000
        )
        .set_timeout(300)
        .append_invoke_contract_function_op(
            contract_id=token_address,
            function_name="balance",
            parameters=[
                scval.to_address(contract_address),
            ],
        )
        .build()
    )
    simulation = soroban_rpc_server.simulate_transaction(tx)
    return i128_to_int(SCVal.from_xdr(simulation.results[0].xdr).i128)


def execute_claim_with_swap(
        network: str,
        soroban_rpc_server: SorobanServer,
        horizon_server: Server,
        keypair: Keypair,
        contract_id: str,
        token_in_address: str,
        amount_with_slippage: int,
        swap_path: str,
) -> int:
    """
    Executes the chained swap transaction on Soroban and returns the final amount out.
    """
    print("Preparing and building swap transaction...")
    source_account = horizon_server.load_account(keypair.public_key)

    # Build the transaction to invoke `swap_chained`
    tx = (
        TransactionBuilder(
            source_account=source_account,
            network_passphrase=network,
            base_fee=10000
        )
        .set_timeout(300)
        .append_invoke_contract_function_op(
            contract_id=contract_id,
            function_name='claim_fees_and_swap',
            parameters=[
                scval.to_address(keypair.public_key),
                SCVal.from_xdr(swap_path),
                scval.to_address(token_in_address),
                scval.to_uint128(amount_with_slippage),
            ],
        )
        .build()
    )

    # Prepare transaction to get Soroban-specific data (footprint, etc.)
    print("Preparing transaction on Soroban...")
    prepared_tx = soroban_rpc_server.prepare_transaction(tx)

    # Sign the prepared transaction
    print("Signing transaction...")
    prepared_tx.sign(keypair)

    # Submit the transaction to Horizon
    print("Submitting transaction to Horizon...")
    submit_response = horizon_server.submit_transaction(prepared_tx)

    if not submit_response.get('successful', False):
        raise Exception("Transaction failed: " + str(submit_response))

    print("Transaction submitted successfully. Fetching result...")

    # Get the transaction result from Soroban server to access Soroban metadata
    tx_info = soroban_server.get_transaction(submit_response['id'])
    if not tx_info or not tx_info.result_meta_xdr:
        raise Exception("No transaction metadata found.")

    # Extract the result from the Soroban metadata
    transaction_meta = TransactionMeta.from_xdr(tx_info.result_meta_xdr)
    return_val = transaction_meta.v3.soroban_meta.return_value
    final_amount = u128_to_int(return_val.u128)
    print("Swap executed successfully.")
    return final_amount


# =========================================
# Entry Point
# =========================================

print("Starting claim process...")
print(f"Swapping {token_to_claim.code} for {token_out.code}...")

# 0. Get the balance of the token to claim
amount = get_contract_balance(
    network,
    soroban_server,
    keypair,
    provider_swap_contract_id,
    token_to_claim.contract_id(network),
)

# 1. Find the swap path and estimated output
amount_with_slippage, swap_path_xdr = find_swap_path(
    base_api,
    token_to_claim.contract_id(network),
    token_out.contract_id(network),
    amount,
    True,
)

# 2. Execute the claim
amount = execute_claim_with_swap(
    network,
    soroban_server,
    horizon_server,
    keypair,
    provider_swap_contract_id,
    token_to_claim.contract_id(network),
    amount_with_slippage,
    swap_path_xdr,
)

print("Claim with swap completed successfully!")
print(f"Amount out: {amount / 10 ** 7} {token_out.code}")