Distribution module¶

The distribution module provides functionality for managing staking rewards, validator commissions, and community pool operations on the Akash Network.

Transaction parameters

All transaction functions support optional parameters like fee_amount, gas_limit, and gas_adjustment. See Transaction parameters for details.

DistributionClient class¶

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")

distribution = client.distribution

Transactions¶

withdraw_delegator_reward()¶

Withdraw rewards from a specific validator.

def withdraw_delegator_reward(wallet, validator_address: str, memo: str = "",
                              fee_amount: str = "5000", gas_limit: int = 200000,
                              gas_adjustment: float = 1.2, use_simulation: bool = True) -> BroadcastResult

Required parameters:

wallet (AkashWallet): Wallet to sign the transaction
validator_address (str): Validator address to withdraw rewards from

Optional parameters:

memo (str): Transaction memo (default: "")
fee_amount (str): Fee amount in uakt (default: "5000")
gas_limit (int): Gas limit (default: 200000)
gas_adjustment (float): Multiplier for gas estimation (default: 1.2)
use_simulation (bool): Enable gas simulation (default: True)

Returns: BroadcastResult - Transaction result with the following fields:

tx_hash (str): Transaction hash
code (int): Response code (0 for success)
raw_log (str): Raw transaction log/error message
success (bool): Whether transaction succeeded
events (List[Dict]): Transaction events
- type (str): Event type
- attributes (List[Dict]): Event attributes
  - key (str): Attribute key
  - value (str): Attribute value

Example:

from akash import AkashClient, AkashWallet

wallet = AkashWallet.from_mnemonic("your mnemonic here")
client = AkashClient("https://akash-rpc.polkachu.com:443")

result = client.distribution.withdraw_delegator_reward(
    wallet=wallet,
    validator_address="akashvaloper1..."
)

if result.success:
    print(f"Rewards withdrawn: {result.tx_hash}")
else:
    print(f"Failed to withdraw rewards: {result.raw_log}")

withdraw_all_rewards()¶

Withdraw all delegation rewards for a delegator from all validators.

def withdraw_all_rewards(wallet, memo: str = "", fee_amount: str = "8000",
                         gas_limit: int = 300000, gas_adjustment: float = 1.2,
                         use_simulation: bool = True) -> BroadcastResult

Required parameters:

wallet (AkashWallet): Wallet to sign the transaction

Optional parameters:

memo (str): Transaction memo (default: "")
fee_amount (str): Fee amount in uakt (default: "8000")
gas_limit (int): Gas limit (default: 300000)
gas_adjustment (float): Multiplier for gas estimation (default: 1.2)
use_simulation (bool): Enable gas simulation (default: True)

Returns: BroadcastResult - Transaction result with the following fields:

tx_hash (str): Transaction hash
code (int): Response code (0 for success)
raw_log (str): Raw transaction log/error message
success (bool): Whether transaction succeeded
events (List[Dict]): Transaction events
- type (str): Event type
- attributes (List[Dict]): Event attributes
  - key (str): Attribute key
  - value (str): Attribute value

Example:

from akash import AkashClient, AkashWallet

client = AkashClient("https://akash-rpc.polkachu.com:443")
wallet = AkashWallet.from_mnemonic("your mnemonic here")

result = client.distribution.withdraw_all_rewards(wallet=wallet)

if result.success:
    print(f"All rewards withdrawn: {result.tx_hash}")
else:
    print(f"Failed to withdraw all rewards: {result.raw_log}")

withdraw_validator_commission()¶

Withdraw validator commission (only for validator operators).

def withdraw_validator_commission(wallet, validator_address: str, memo: str = "",
                                  fee_amount: str = "5000", gas_limit: int = 200000,
                                  gas_adjustment: float = 1.2, use_simulation: bool = True) -> BroadcastResult

Required parameters:

wallet (AkashWallet): Wallet to sign the transaction (must be validator operator)
validator_address (str): Validator address to withdraw commission from

Optional parameters:

memo (str): Transaction memo (default: "")
fee_amount (str): Fee amount in uakt (default: "5000")
gas_limit (int): Gas limit (default: 200000)
gas_adjustment (float): Multiplier for gas estimation (default: 1.2)
use_simulation (bool): Enable gas simulation (default: True)

Returns: BroadcastResult - Transaction result with the following fields:

tx_hash (str): Transaction hash
code (int): Response code (0 for success)
raw_log (str): Raw transaction log/error message
success (bool): Whether transaction succeeded
events (List[Dict]): Transaction events
- type (str): Event type
- attributes (List[Dict]): Event attributes
  - key (str): Attribute key
  - value (str): Attribute value

Example:

from akash import AkashClient, AkashWallet

client = AkashClient("https://akash-rpc.polkachu.com:443")
validator_wallet = AkashWallet.from_mnemonic("validator mnemonic here")

result = client.distribution.withdraw_validator_commission(
    wallet=validator_wallet,
    validator_address="akashvaloper1..."
)

if result.success:
    print(f"Commission withdrawn: {result.tx_hash}")

set_withdraw_address()¶

Set the withdrawal address for rewards.

def set_withdraw_address(wallet, withdraw_address: str, memo: str = "",
                         fee_amount: str = "5000", gas_limit: int = 200000,
                         gas_adjustment: float = 1.2, use_simulation: bool = True) -> BroadcastResult

Required parameters:

wallet (AkashWallet): Wallet to sign the transaction
withdraw_address (str): Address to receive rewards

Optional parameters:

memo (str): Transaction memo (default: "")
fee_amount (str): Fee amount in uakt (default: "5000")
gas_limit (int): Gas limit (default: 200000)
gas_adjustment (float): Multiplier for gas estimation (default: 1.2)
use_simulation (bool): Enable gas simulation (default: True)

Returns: BroadcastResult - Transaction result with the following fields:

tx_hash (str): Transaction hash
code (int): Response code (0 for success)
raw_log (str): Raw transaction log/error message
success (bool): Whether transaction succeeded
events (List[Dict]): Transaction events
- type (str): Event type
- attributes (List[Dict]): Event attributes
  - key (str): Attribute key
  - value (str): Attribute value

Example:

from akash import AkashClient, AkashWallet

client = AkashClient("https://akash-rpc.polkachu.com:443")
wallet = AkashWallet.from_mnemonic("your mnemonic here")

result = client.distribution.set_withdraw_address(
    wallet=wallet,
    withdraw_address="akash1different..."
)

if result.success:
    print(f"Withdraw address set: {result.tx_hash}")

fund_community_pool()¶

Fund the community pool with specified amount.

def fund_community_pool(wallet, amount: str, denom: str = "uakt",
                        memo: str = "", fee_amount: str = "5000",
                        gas_limit: int = 200000, gas_adjustment: float = 1.2,
                        use_simulation: bool = True) -> BroadcastResult

Required parameters:

wallet (AkashWallet): Wallet to sign the transaction
amount (str): Amount to fund

Optional parameters:

denom (str): Token denomination (default: "uakt")
memo (str): Transaction memo (default: "")
fee_amount (str): Fee amount in uakt (default: "5000")
gas_limit (int): Gas limit (default: 200000)
gas_adjustment (float): Multiplier for gas estimation (default: 1.2)
use_simulation (bool): Enable gas simulation (default: True)

Returns: BroadcastResult - Transaction result with the following fields:

tx_hash (str): Transaction hash
code (int): Response code (0 for success)
raw_log (str): Raw transaction log/error message
success (bool): Whether transaction succeeded
events (List[Dict]): Transaction events
- type (str): Event type
- attributes (List[Dict]): Event attributes
  - key (str): Attribute key
  - value (str): Attribute value

Example:

from akash import AkashClient, AkashWallet

client = AkashClient("https://akash-rpc.polkachu.com:443")
wallet = AkashWallet.from_mnemonic("your mnemonic here")

result = client.distribution.fund_community_pool(
    wallet=wallet,
    amount="10000000",
    denom="uakt"
)

if result.success:
    print(f"Community pool funded: {result.tx_hash}")

Queries¶

get_delegator_rewards()¶

Query delegator rewards from a specific validator or all validators.

def get_delegator_rewards(delegator_address: str, validator_address: str = "") -> List[Dict[str, Any]]

Required parameters:

delegator_address (str): Address of the delegator

Optional parameters:

validator_address (str): Validator address to filter by (default: "")

Returns: List[Dict[str, Any]] with the following structure:

If validator_address is provided: - List[Dict]: List of reward coins - denom (str): Token denomination - amount (str): Reward amount

If validator_address is empty (all rewards): - List[Dict]: List of validator reward entries - validator_address (str): Validator address - rewards (List[Dict]): List of reward coins - denom (str): Token denomination - amount (str): Reward amount

Example:

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")

all_rewards = client.distribution.get_delegator_rewards(
    delegator_address="akash1..."
)

for validator_reward in all_rewards:
    validator = validator_reward["validator_address"]
    total_rewards = sum(float(r["amount"]) for r in validator_reward["rewards"] if r["denom"] == "uakt")
    print(f"Validator {validator}: {total_rewards / 1000000:.6f} AKT")

validator_rewards = client.distribution.get_delegator_rewards(
    delegator_address="akash1...",
    validator_address="akashvaloper1..."
)

for reward in validator_rewards:
    if reward["denom"] == "uakt":
        akt_amount = float(reward["amount"]) / 1000000
        print(f"Pending rewards: {akt_amount:.6f} AKT")

get_validator_commission()¶

Query validator commission.

def get_validator_commission(validator_address: str) -> Dict[str, Any]

Required parameters:

validator_address (str): Address of the validator

Returns: Dict[str, Any] with the following fields:

commission (List[Dict]): List of commission coins
- denom (str): Token denomination
- amount (str): Commission amount

Example:

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")

commission = client.distribution.get_validator_commission("akashvaloper1...")

for coin in commission["commission"]:
    if coin["denom"] == "uakt":
        akt_commission = float(coin["amount"]) / 1000000
        print(f"Validator commission: {akt_commission:.6f} AKT")

get_validator_outstanding_rewards()¶

Query validator outstanding rewards.

def get_validator_outstanding_rewards(validator_address: str) -> Dict[str, Any]

Required parameters:

validator_address (str): Address of the validator

Returns: Dict[str, Any] with the following fields:

rewards (List[Dict]): List of outstanding reward coins
- denom (str): Token denomination
- amount (str): Outstanding reward amount

Example:

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")

outstanding = client.distribution.get_validator_outstanding_rewards("akashvaloper1...")

for reward in outstanding["rewards"]:
    if reward["denom"] == "uakt":
        akt_rewards = float(reward["amount"]) / 1000000
        print(f"Outstanding rewards: {akt_rewards:.6f} AKT")

get_distribution_params()¶

Query distribution parameters.

def get_distribution_params() -> Dict[str, Any]

Returns: Dict[str, Any] with the following fields:

community_tax (str): Community tax rate
base_proposer_reward (str): Base proposer reward rate
bonus_proposer_reward (str): Bonus proposer reward rate
withdraw_addr_enabled (bool): Whether withdraw address changes are enabled

Example:

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")

params = client.distribution.get_distribution_params()

print(f"Community tax: {float(params['community_tax']) * 100:.2f}%")
print(f"Base proposer reward: {float(params['base_proposer_reward']) * 100:.2f}%")
print(f"Bonus proposer reward: {float(params['bonus_proposer_reward']) * 100:.2f}%")
print(f"Withdraw address enabled: {params['withdraw_addr_enabled']}")

get_community_pool()¶

Query the community pool coins.

def get_community_pool() -> Dict[str, Any]

Returns: Dict[str, Any] with the following fields:

pool (List[Dict]): List of community pool coins
- denom (str): Token denomination
- amount (str): Pool amount

Example:

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")

pool = client.distribution.get_community_pool()

for coin in pool["pool"]:
    if coin["denom"] == "uakt":
        akt_pool = float(coin["amount"]) / 1000000
        print(f"Community pool: {akt_pool:,.2f} AKT")

get_validator_slashes()¶

Query validator slashes within a height range.

def get_validator_slashes(validator_address: str, starting_height: int = 1,
                          ending_height: int = 0) -> List[Dict[str, Any]]

Required parameters:

validator_address (str): Validator operator address

Optional parameters:

starting_height (int): Starting block height (default: 1)
ending_height (int): Ending block height (default: 0 = current height)

Returns: List[Dict[str, Any]] - List of validator slashes with the following fields:

validator_period (str): Validator period when slash occurred
fraction (str): Slash fraction

Example:

from akash import AkashClient

client = AkashClient("https://akash-rpc.polkachu.com:443")

slashes = client.distribution.get_validator_slashes(
    validator_address="akashvaloper1...",
    starting_height=1000000,
    ending_height=2000000
)

if slashes:
    print(f"Found {len(slashes)} slashes:")
    for slash in slashes:
        fraction_pct = float(slash["fraction"]) * 100
        print(f"Period {slash['validator_period']}: {fraction_pct:.4f}% slashed")
else:
    print("No slashes found in the specified range")