nibiru package

Subpackages

Submodules

nibiru.client module

nibiru.common module

pysdk.exceptions module

nibiru.network module

nibiru.sdk module

nibiru.tx module

Classes:: TxClient: A client for building, simulating, and broadcasting transactions. Transaction: Transactions trigger state changes based on messages. Each

message must be cryptographically signed before being broadcasted to the network.

class nibiru.tx.Transaction(msgs: Tuple[Message, ...] | None = None, account_num: int | None = None, priv_key: PrivateKey | None = None, sequence: int | None = None, chain_id: str | None = None, fee: List[Coin] | None = None, gas_limit: int = 0, memo: str = '', timeout_height: int = 0)

Bases: Jsonable

Transactions trigger state changes based on messages (‘msgs’). Each message must be signed before being broadcasted to the network, included in a block, validated, and approved through the consensus process.

account_num

Number of the account in state. Account numbers increment every time an account is created so that each account has its own number, and the highest account number is equivalent to the number of accounts in the ‘auth’ module (but not necessarily the store).

Type:: int

msgs: A List of messages to be executed.

sequence

A per sender “nonce” that acts as a security measure to prevent replay attacks on transactions. Each transaction request must have a different sequence number from all previously executed transactions so that no transaction can be replayed.

Type:: int

chain_id

The unique identifier for the blockchain that this transaction targets. Inclusion of a ‘chain_id’ prevents potential attackers from using signed transactions on other blockchains.

Type:: str

fee

Coins to be paid in fees. The ‘fee’ helps prevents end users from spamming the network. Gas cosumed during message execution is typically priced from a fee equal to ‘gas_consumed * gas_prices’. Here, ‘gas_prices’ is the minimum gas price, and it’s a parameter local to each node.

Type:: List[Coin]

gas_limit

Maximum gas to be allowed for the transaction. The transaction execution fails if the gas limit is exceeded.

Type:: int

priv_key

Primary signer for the transaction. By convention, the signer from the first message is referred to as the primary signer and pays the fee for the whole transaction. We refer to this primary signer with ‘priv_key’.

Type:: wallet.PrivateKey

memo

Memo is a note or comment to be added to the transaction.

Type:: str

timeout_height

Timeout height is the block height after which the transaction will not be processed by the chain.

Type:: int

get_sign_doc(public_key: PublicKey | None = None) → SignDoc

get_signed_tx_data() → bytes

get_tx_data(signature: bytes, public_key: PublicKey | None = None) → bytes

property raw_bytes: bytes

with_account_num(account_num: int) → Transaction

with_chain_id(chain_id: str) → Transaction

with_fee(fee: List[Coin]) → Transaction

with_gas_limit(gas: Number) → Transaction

with_memo(memo: str) → Transaction

with_messages(msgs: Iterable[Message]) → Transaction

with_sender(client: GrpcClient, sender: str) → Transaction

with_sequence(sequence: int) → Transaction

with_signer(priv_key: PrivateKey)

with_timeout_height(timeout_height: int) → Transaction

class nibiru.tx.TxClient(network: Network, client: GrpcClient, config: TxConfig, priv_key: PrivateKey | None = None)

Bases: object

A client for building, simulating, and broadcasting transactions.

address

Type:: Optional[wallet.Address]

client

Type:: GrpcClient

network

Type:: pt.Network

priv_key

Type:: wallet.PrivateKey

tx_config

Type:: pt.TxConfig

address: Address | None

build_tx(msgs: PythonMsg | List[PythonMsg], sequence: int | None = None) → Tuple[Transaction, Address]

build_tx_with_node_sequence(msgs: PythonMsg | List[PythonMsg])

client: GrpcClient

ensure_address_info() → Address: Guarantees that the TxClient.address has been set and returns it. If the wallet address has not been set prior to this function call, (1) the address is derived from the ‘priv_key’ and (2) the sequence is derived from the ‘network.lcd_endpoint’.

ensure_tx_config(new_tx_config: TxConfig | None = None) → TxConfig

Guarantees that the TxClient.tx_config has been set and returns it.

Parameters:: new_tx_config (Optional[pytypes.TxConfig]) – Becomes the new value for the tx config if given. Defaults to None.
Returns:: The new value for the TxClient.tx_config.
Return type:: (pt.TxConfig)

execute_msgs(msgs: PythonMsg | List[PythonMsg], sequence: int | None = None, tx_config: TxConfig | None = None) → ExecuteTxResp

Broadcasts messages to a node in a single transaction. This function first simulates the corresponding transaction to estimate the amount of gas needed.

If the transaction fails because of account sequence mismatch, we try to query the sequence from the LCD endpoint and broadcast with the updated sequence value.

Parameters:

msgs (Union[pt.PythonMsg, List[pt.PythonMsg]]) –
sequence (Optional[int]) – Account sequence for the tx. Sequence is used to enforce tx ordering and prevent double-spending. Each time a tx is procesed and committed to the blockchain, the account sequence number is incremented.
tx_config (Optional[pt.TxConfig] = None) –
get_sequence_from_node (bool, optional) – Specifies whether the sequence comes from the local value or the lcd endpoint. Defaults to False.

Raises:

SimulationError – If broadcasting fails during the simulation.
TxError – If the response code is nonzero, the ‘TxError’ includes the raw error logs from the blockchain.

Returns:

The transaction response as: a dict in proto3 JSON format.

Return type:

Union[RawSyncTxResp, Dict[str, Any]]

execute_tx(tx: Transaction, gas_estimate: float, use_tmrpc: bool = True, tx_config: TxConfig | None = None) → JsonRPCResponse | TxResponse

network: Network

priv_key: PrivateKey

simulate(tx: Transaction) → SimulationResponse

Parameters:

tx (Transaction) – The transaction being simulated.

Returns:

SimulationResponse defines the response: generated when a transaction is simulated successfully.

Return type:

SimulationResponse

Raises:

SimulationError –

tx_config: TxConfig

nibiru.utils module

class nibiru.utils.ColoredFormatter(fmt=None, datefmt=None, style='%', validate=True)

Bases: Formatter

FORMATS = {10: '%(asctime)s|%(levelname)s|%(funcName)s| %(message)s', 20: '%(asctime)s|%(levelname)s|%(funcName)s| %(message)s', 30: '%(asctime)s|%(levelname)s|%(funcName)s| %(message)s', 40: '%(asctime)s|%(levelname)s|%(funcName)s| %(message)s', 50: '%(asctime)s|%(levelname)s|%(funcName)s| %(message)s'}

bold_red = '\x1b[31;1m'

cyan = '\x1b[36;20m'

fmt = '%(asctime)s|%(levelname)s|%(funcName)s| %(message)s'

format(record: LogRecord)

Formats a record for the logging handler.

Parameters:: record (logging.LogRecord) – Represents an instance of an event being logged.

green = '\x1b[32;20m'

grey = '\x1b[38;20m'

red = '\x1b[31;20m'

reset = '\x1b[0m'

white = '\x1b[97;20m'

yellow = '\x1b[33;20m'

nibiru.utils.assert_subset(result, expected, path=None)

Check if all values in expected are equal to the values in result.

This function iteratively checks if the expected values match the values in result. It traverses nested dictionaries and lists, checking for matching values while ignoring missing keys in the expected dictionary.

Parameters:

result – The result dictionary.
expected – The expected dictionary.
path – The path to the current key (used for error messages).

Raises:

AssertionError – If a value in expected does not match the corresponding value in result.

nibiru.utils.clean_nested_dict(dictionary: List | Dict | str) → Dict

Takes a nested dictionnary with some values being string json values and convert it into a proper nested dictionary.

Eg

{
    "transaction_fee": "{"denom":"unusd","amount":"0"}",
    "funding_payment": "0.000000000000000000",
    "realized_pnl": "0.000000000000000000",
    "bad_debt": "{"denom":"unusd","amount":"0"}",
    "trader_address": "nibi1zaavvzxez0elundtn32qnk9lkm8kmcsz44g7xl",
    "margin": "{"denom":"unusd","amount":"10"}",
    "exchanged_position_size": "0.004999999999999500",
    "tx_hash": "12E496C996E39820B0807857CB7C19674BDD12DC8D789647D68C50BBB8C7D9CF"
}

becomes

{
    "transaction_fee": {
        "denom": "unusd",
        "amount": "0"
    },
    "funding_payment": "0.000000000000000000",
    "realized_pnl": "0.000000000000000000",
    "bad_debt": {
        "denom": "unusd",
        "amount": "0"
    },
    "trader_address": "nibi1zaavvzxez0elundtn32qnk9lkm8kmcsz44g7xl",
    "margin": {
        "denom": "unusd",
        "amount": "10"
    },
    "exchanged_position_size": "0.004999999999999500",
    "tx_hash": "12E496C996E39820B0807857CB7C19674BDD12DC8D789647D68C50BBB8C7D9CF"
}

Parameters:: dictionary (Union[List, Dict, str]) – The dictionary to be converted.
Returns:: A converted dictionary.
Return type:: Dict

nibiru.utils.dict_keys_must_match(dict_: dict, keys: List[str])

Asserts that two iterables have the same elements, the same number of times, without regard to order. Alias for the ‘element_counts_are_equal’ function.

dict_keys_must_match(dict_, keys)

Example: - [0, 1, 1] and [1, 0, 1] compare equal. - [0, 0, 1] and [0, 1] compare unequal.

nibiru.utils.element_counts_are_equal(first: Iterable[Any], second: Iterable[Any]) → bool | None

Asserts that two iterables have the same elements, the same number of times, without regard to order.

Parameters:

first (Iterable[Any]) –
second (Iterable[Any]) –

Returns:

“passed” status. If this is True, first and second share: the same element counts. If they don’t the function will raise an AssertionError and return ‘None’.

Return type:

Optional[bool]

nibiru.utils.format_fields_nested(object: list | dict, fn: Callable[[Any], Any], fields: List[str]) → list | dict

Format the fields inside a nested dictionary with the function provided

Parameters:

object (Union[list, dict]) – The object to format
fn (Callable[[Any], Any]) – The function to format objects with
fields (list[str]) – The fields to format

Returns:

The output formatted

Return type:

Union[list, dict]

nibiru.utils.from_sdk_dec(dec_str: str) → float

nibiru.utils.from_sdk_dec_n(dec_str: str, n: int = 6) → float

nibiru.utils.from_sdk_int(int_str: str) → int

nibiru.utils.toPbTimestamp(dt: datetime)

nibiru.utils.to_sdk_dec(dec: float) → str

create a decimal from an input decimal. valid must come in the form:

(-) whole integers (.) decimal integers

examples of acceptable input include:: -123.456 456.7890 345 -456789

NOTE - An error will return if more decimal places are provided in the string than the constant Precision.

CONTRACT - This function does not mutate the input str.

nibiru.utils.to_sdk_int(i: float) → str

nibiru.utils.update_nested_fields(d: dict, fields: List[str], func: Callable[[Any], Any]) → dict

Update specified fields in a nested dictionary.

This function recursively traverses a dictionary and updates specified fields using a provided function. Field names should be specified as a list of strings, where each string represents a nested path to the key that should be updated (e.g., “key1.key2.key3”).

Parameters:

d – The dictionary to update.
fields – A list of fields to update, specified as strings representing key paths.
func – A function to apply to each specified field.

Returns:

The updated dictionary.

nibiru.wallet module

class nibiru.wallet.Address(addr: bytes)

Bases: object

addr: bytes

decrease_sequence(): If a tx failed the sequence should not increase

classmethod from_acc_bech32(bech: str) → Address: Create an address instance from a bech32-encoded account address

classmethod from_cons_bech32(bech: str) → Address: Create an address instance from a bech32-encoded consensus address

classmethod from_val_bech32(bech: str) → Address: Create an address instance from a bech32-encoded validator address

get_number()

get_sequence(from_node=False, lcd_endpoint: str | None = None) → int

Get the sequence number from the module. If from node is set to true, update our local indice with a query to the node.

Parameters:

from_node (bool, optional) – Whether to query or not from the node. Defaults to False.
lcd_endpoint (str, optional) – The lcd endoint, needed for when from_node is set to true. Defaults to None.

Returns:

the current sequence number

Return type:

int

get_subaccount_id(index: int) → str: Return a hex representation of address

init_num_seq(lcd_endpoint: str) → Address

number: int

sequence: int

to_acc_bech32() → str: Return a bech32-encoded account address

to_cons_bech32() → str: Return a bech32-encoded with consensus address

to_hex() → str: Return a hex representation of address

to_val_bech32() → str: Return a bech32-encoded validator address

class nibiru.wallet.PrivateKey(_error_do_not_use_init_directly=None)

Bases: object

Class for wrapping SigningKey that is used for signature creation and public key derivation.

Variables:: signing_key (ecdsa.SigningKey) – the ecdsa SigningKey instance

classmethod from_hex(priv: str) → PrivateKey

classmethod from_mnemonic(words: str, path="m/44'/118'/0'/0/0") → PrivateKey

Create a PrivateKey instance from a given mnemonic phrase and a HD derivation path. If path is not given, default to Band’s HD prefix 494 and all other indexes being zeroes.

Parameters:

words – the mnemonic phrase for recover private key
path – the HD path that follows the BIP32 standard

Returns:

Initialized PrivateKey object

classmethod generate(path="m/44'/118'/0'/0/0") → Tuple[str, PrivateKey]

Generate new private key with random mnemonic phrase

Parameters:: path – the HD path that follows the BIP32 standard
Returns:: A tuple of mnemonic phrase and PrivateKey instance
Return type:: Tuple[Mnemonic, ‘PrivateKey’]

sign(msg: bytes) → bytes

Sign the given message using the edcsa sign_deterministic function.

Parameters:: msg – the message that will be hashed and signed
Returns:: a signature of this private key over the given message

to_hex() → str: Return a hex representation of signing key.

to_public_key() → PublicKey

Return the PublicKey associated with this private key.

Returns:: a PublicKey that can be used to verify the signatures made with this PrivateKey

class nibiru.wallet.PublicKey(_error_do_not_use_init_directly=None)

Bases: object

Class for wrapping VerifyKey using for signature verification. Adding method to encode/decode to Bech32 format.

Variables:: verify_key (ecdsa.VerifyingKey) – the ecdsa VerifyingKey instance

classmethod from_acc_bech32(bech: str) → PublicKey

classmethod from_cons_bech32(bech: str) → PublicKey

classmethod from_val_bech32(bech: str) → PublicKey

to_acc_bech32() → str: Return bech32-encoded with account public key prefix

to_address() → Address: Return address instance from this public key

to_cons_bech32() → str: Return bech32-encoded with validator consensus public key prefix

to_hex() → str: Return a hex representation of verified key.

to_public_key_proto() → PubKey

to_val_bech32() → str: Return bech32-encoded with validator public key prefix

verify(msg: bytes, sig: bytes) → bool

Verify a signature made from the given message.

Parameters:

msg – data signed by the signature, will be hashed using sha256 function
sig – encoding of the signature

Raises:

BadSignatureError – if the signature is invalid or malformed

Returns:

True if the verification was successful

nibiru package

Subpackages

Submodules

nibiru.client module

nibiru.common module

pysdk.exceptions module

nibiru.network module

nibiru.sdk module

nibiru.tx module

nibiru.utils module

nibiru.wallet module

Module contents