nqs_sdk package

Subpackages

Submodules

Module contents

class nqs_sdk.Simulation(protocols, config, namespace=None)[source]

Bases: object

Main orchestrator for running simulations.

This class serves as the primary interface for users to configure, build, and execute simulations involving multiple DeFi protocols. It handles the coordination between protocol factories, configuration loading, and simulation execution.

Protocols and configuration are provided during initialization, and the simulation can be run multiple times with different parameters.

protocols

List of protocol factories or protocol factory adapters that define the DeFi protocols to simulate

config

Configuration data (file path, dict, or YAML/JSON content)

simulator

Internal rust-based simulator instance (built lazily)

__init__(protocols, config, namespace=None)[source]

Initialize a new simulation with specified protocols and configuration.

Parameters:

  • protocols (Union[ProtocolFactory, ProtocolFactoryAdapter, List[ProtocolFactory | ProtocolFactoryAdapter]]) – Single ProtocolFactory or ProtocolFactoryAdapter, or list of ProtocolFactorys or ProtocolFactoryAdapters defining the protocols to include in the simulation

  • config (Union[str, dict, Path]) – Configuration for the simulation. Can be: - Path to a YAML/JSON configuration file - Dictionary containing configuration parameters - String containing YAML/JSON configuration content

Example

>>> from nqs_sdk import Simulation
>>> from nqs_sdk.protocols import UniswapV3Factory
>>> from nqs_sdk_extension.protocols import CompoundV2Factory
>>> uniswap = UniswapV3Factory()
>>> compound = CompoundV2Factory()
>>> sim = Simulation([uniswap, compound], "config.yaml")
get_protocol(protocol_id)[source]

Retrieve a protocol instance by its identifier.

Parameters:

protocol_id (str) – Unique identifier for the protocol (e.g., “uniswap_v3”)

Return type:

Any

Returns:

The protocol instance corresponding to the given ID

Raises:

  • RuntimeError – If simulation hasn’t been built yet

  • KeyError – If protocol_id doesn’t exist in the simulation

Example

>>> uniswap_protocol = sim.get_protocol("uniswap_v3")
>>> current_price = uniswap_protocol.get_current_price()
run()[source]

Execute the simulation and return results.

This method runs the entire simulation from start to end block/timestamp, processing all transactions and collecting metrics along the way.

Returns:

Object containing all simulation data including:

  • Protocol states at each block

  • Agent portfolio values over time

  • Transaction logs and fees

  • Observable metrics and KPIs

Return type:

SimulationResults

Raises:

RuntimeError – If simulation hasn’t been built yet

Example

>>> results = sim.run()
>>> portfolio_value = results.get_agent_metric("alice", "total_holding")
class nqs_sdk.BlockNumberOrTimestamp

Bases: object

static block_number(n)
static timestamp(t)
class nqs_sdk.LPTokenUniv3(*args, **kwargs)

Bases: object

fee_growth_inside_0_last_x128
fee_growth_inside_1_last_x128
liquidity
pool_name
tick_lower
tick_upper
token_id
tokens_owed_0
tokens_owed_1
class nqs_sdk.MetricName(parameters, name)

Bases: object

static deserialize(data)
class nqs_sdk.Metrics

Bases: object

mirror of [simulator::ObserverOutputs]

get(name)
class nqs_sdk.MutBuilderSharedState

Bases: object

mirror of [simulator::SharedState]

builder_spot_oracle()
class nqs_sdk.MutSharedState

Bases: object

agent_name_to_addr(agent_name)
ensure_loaded_for_block(py_clock)
get_parameters()
get_wallet(addr)
insert_wallet(addr, py_wallet)
list_wallets()
py_wrapper_spot_oracle(py_clock)
class nqs_sdk.ObservableDescription(protocol_ids, aggregation_targets, aggregation_outputs)

Bases: object

aggregation_outputs

all the different metrics that will be output into aggregations

aggregation_targets

all the different exclusively owned fields that are legal to aggregate into, and which algo they should use

protocol_ids

all the different protocol ids for metrics, should generally be the protocol id

class nqs_sdk.Parameters

Bases: object

mirror of [simulator::Parameters]

add_common(tokens)
add_compound_v2_protocol(tokens)
add_json_protocol(config, name, params)
add_uniswap_v3_protocol(name, token0, token1)
add_yaml_protocol(config, name, params)
class nqs_sdk.ProtocolFactoryAdapter(py_object)

Bases: object

class nqs_sdk.RefSharedState

Bases: object

agent_name_to_addr(agent_name)
get_parameters()
get_wallet(addr)
list_wallets()
py_wrapper_spot_oracle(py_clock)
class nqs_sdk.SealedParameters

Bases: object

mirror of [simulator::SealedParameters]

all_agents()
metric_to_str(metric)
str_to_metric(str)
class nqs_sdk.SimulationClock

Bases: object

mirror of [simulator::SimulationClock]

current_block()
current_time()
elapsed_time()
previous_observable_block()
simulation_time()
class nqs_sdk.SimulationTime

Bases: object

mirror of [simulator::SimulationClock]

block_at(time)
block_at_timestamp(timestamp)
clock_at(block)
first_block()
first_time()
start_block()
start_time()
stop_block()
time_at(block)
timestamp_at(block)
class nqs_sdk.Simulator

Bases: object

collect_observables(_metrics, _block)
static from_json(json)
get_py_protocol(id)
list_protocols()
run()

Will run all events of the simulation

run_to_dict()
run_with_gil()

Will run all events of the simulation

to_json()
class nqs_sdk.SimulatorBuilder

Bases: object

add_factory(factory)

Add a new factory to the list of factories

build()

Build the simulator

static from_json(config)

Create a new instance of the simulator builder from a json config file

static from_yaml(config)

Create a new instance of the simulator builder from a json config file

static new()

Create a new empty instance of the simulator builder

class nqs_sdk.TokenMetadata(*args, **kwargs)

Bases: object

decimals
name
symbol
class nqs_sdk.TxRequest(protocol, source, sender, payload)

Bases: object

mirror of [simulator::TxRequest]

static new_random(protocol, source, payload)
static new_with_order(protocol, source, sender, order, payload)
order
payload
protocol
sender
source
class nqs_sdk.Wallet(*args, **kwargs)

Bases: object

agent_name
drop_token(token)
get_balance_of(token)
get_balance_of_float(token)
get_cerc20_tokens()
get_erc721_token(token_id, action_name=None)
get_erc721_tokens()
get_erc721_tokens_for_pool_name(pool_name)
get_float_representation(amount, identifier)
get_list_tokens()
get_next_token_id(token_id, pool_name)
get_py_metadata(symbol)
get_scaled_tokens()
get_state()
get_token_id_tick_range(token_id, action_name=None)
has_enough_balance(amount, identifier, action_name=None)
has_enough_liquidity(token, amount, action_name=None)
holdings
is_empty()

Check if the holdings is empty this method could be counterintuitive we add it for backward compatibility with python

mint_erc721(token)
replace_erc721_token(erc721_token)
restore_from_state(state)
tokens_metadata
transfer_from(token, amount, action_name=None)
transfer_to(token, amount, action_name=None)