Executing Swaps with Provider Fees

Replace your direct swap_chained call with the wrapper’s version that accepts a fee fraction:

• Function: swap_chained(e, user: Address, swaps_chain: Vec<…>, token_in: Address, in_amount: u128, out_min: u128, fee_fraction: u32) -> u128

• Args:

  • fee_fraction: Fee in bps to deduct on this swap.

  • Validated on‑chain: if fee_fraction > max_swap_fee_fraction the call reverts.

• Returns: Net output amount delivered to user after deducting provider fee.

from decimal import Decimal

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

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

# This account must have at least 3 XLM and a trustline to AQUA.
user_secret_key = "SA..........."
keypair = Keypair.from_secret(user_secret_key)

# Input and output tokens
token_in = Asset.native()  # XLM
token_out = Asset("AQUA", "GBNZILSTVQZ4R7IKQDGHYGY2QXL5QOFJYQMXPKWRRM5PAV7Y4M67AQUA")

# If True, the swap behaves like strict-send: the amount of the sending asset is fixed.
# If False, the swap behaves like strict-receive: the amount of the receiving asset is fixed.
is_send = True
# Amount of 1 XLM or 1 AQUA in stroops (depending on is_send)
amount = 1_0000000
slippage = Decimal("0.005")  # .5% slippage
provider_fee = Decimal("0.003")  # .3% provider fee

# 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


# =========================================
# 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': str(slippage),
        'provider_fee': str(provider_fee),
    }
    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 execute_swap(
        network: str,
        soroban_rpc_server: SorobanServer,
        horizon_server: Server,
        keypair: Keypair,
        router_contract_id: str,
        token_in_address: str,
        amount: int,
        amount_with_slippage: int,
        swap_path: str,
        provider_fee_bps: int,
        is_send: bool,
) -> 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)

    function_name = 'swap_chained' if is_send else 'swap_chained_strict_receive'

    # 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=router_contract_id,
            function_name=function_name,
            parameters=[
                scval.to_address(keypair.public_key),
                SCVal.from_xdr(swap_path),
                scval.to_address(token_in_address),
                scval.to_uint128(amount),
                scval.to_uint128(amount_with_slippage),
                scval.to_uint32(provider_fee_bps),
            ],
        )
        .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 swap process...")
print(f"Swapping {token_in.code} for {token_out.code}...")

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

# 2. Execute the swap
amount = execute_swap(
    network,
    soroban_server,
    horizon_server,
    keypair,
    provider_swap_contract_id,
    token_in.contract_id(network),
    amount,
    amount_with_slippage,
    swap_path_xdr,
    int(provider_fee * 10000),  # Convert to basis points
    is_send,
)

print("Swap completed successfully!")

if is_send:
    print(f"Amount out: {amount / 10 ** 7} {token_out.code}")  # If it's strict-send, show output amount
else:
    print(f"Amount in: {amount / 10 ** 7} {token_in.code}")  # If it's strict-receive, show input amount