AbstractPool

class AbstractPool

Bases: ABC

Abstract base class for implementing various types of liquidity pools.

This class defines the basic structure and interface for different pool implementations in the quantammsim simulator. It provides abstract methods that must be implemented by concrete subclasses to define specific pool behaviors.

calculate_reserves_with_fees(params, run_fingerprint, prices, start_index, additional_oracle_input)

Calculate reserve changes with fees and arbitrage enabled.

Used when fees are non-zero and arbitrage is enabled. Handles arbitrage thresholds, trading costs, and fee calculations. Less performant than zero-fees case.

Parameters:

• params (Dict[str, Any])

• run_fingerprint (Dict[str, Any])

• prices (Array)

• start_index (Array)

• additional_oracle_input (Array | None)

Return type:

Array

calculate_reserves_zero_fees(params, run_fingerprint, prices, start_index, additional_oracle_input)

Calculate reserve changes assuming zero fees.

Fast, vectorized implementation for the zero-fees case. Uses parallel computation since arbitrageurs will always trade to exactly match external market prices. Should be overridden with fees=0 version of calculate_reserves_with_fees if no faster implementation exists.

Parameters:

• params (Dict[str, Any])

• run_fingerprint (Dict[str, Any])

• prices (Array)

• start_index (Array)

• additional_oracle_input (Array | None)

Return type:

Array

calculate_reserves_with_dynamic_inputs(params, run_fingerprint, prices, start_index, additional_oracle_input)

Calculate reserve changes with time-varying parameters.

Handles cases where pool properties like fees, arbitrage thresholds, or weights can change over time. Required for pools with dynamic parameters.

Parameters:

• params (Dict[str, Any])

• run_fingerprint (Dict[str, Any])

• prices (Array)

• start_index (Array)

• fees_array (Array)

• arb_thresh_array (Array)

• arb_fees_array (Array)

• trade_array (Array)

• lp_supply_array (Array | None)

• additional_oracle_input (Array | None)

Return type:

Array

Notes

Subclasses of AbstractPool should implement the abstract methods to define specific behaviors for different types of liquidity pools.

__init__()

extend_parameters(base_params, initial_values_dict, n_assets, n_parameter_sets)

Default null implementation of parameter extension.

Parameters:

• base_params (Dict[str, Any])

• initial_values_dict (Dict[str, Any])

• n_assets (int)

• n_parameter_sets (int)

Return type:

Dict[str, Any]

abstract calculate_reserves_with_fees(params, run_fingerprint, prices, start_index, additional_oracle_input=None)

Parameters:

• params (Dict[str, Any])

• run_fingerprint (Dict[str, Any])

• prices (Array)

• start_index (Array)

• additional_oracle_input (Array | None)

Return type:

Array

abstract calculate_reserves_zero_fees(params, run_fingerprint, prices, start_index, additional_oracle_input=None)

Parameters:

• params (Dict[str, Any])

• run_fingerprint (Dict[str, Any])

• prices (Array)

• start_index (Array)

• additional_oracle_input (Array | None)

Return type:

Array

abstract calculate_reserves_with_dynamic_inputs(params, run_fingerprint, prices, start_index, fees_array, arb_thresh_array, arb_fees_array, trade_array, lp_supply_array=None, additional_oracle_input=None)

Parameters:

• params (Dict[str, Any])

• run_fingerprint (Dict[str, Any])

• prices (Array)

• start_index (Array)

• fees_array (Array)

• arb_thresh_array (Array)

• arb_fees_array (Array)

• trade_array (Array)

• lp_supply_array (Array | None)

• additional_oracle_input (Array | None)

Return type:

Array

init_parameters(initial_values_dict, run_fingerprint, n_assets, n_parameter_sets=1, noise='gaussian')

Initialize pool parameters and apply any extensions from mixins.

Parameters:

• initial_values_dict (Dict[str, Any])

• run_fingerprint (Dict[str, Any])

• n_assets (int)

• n_parameter_sets (int)

• noise (str)

Return type:

Dict[str, Any]

calculate_initial_weights(params, *args, **kwargs)

Calculate initial pool weights from initial logits or from directly-provided weights. If both are provided, the weights calculated from logits take precedence.

Uses softmax with stop_gradient to ensure weights remain constant during any optimization.

Parameters:

• params (Dict[str, jnp.ndarray]) – Must contain ‘initial_weights_logits’ key or ‘initial_weights’ key

• *args – Not used, kept for interface compatibility

• **kwargs – Not used, kept for interface compatibility

Returns:

Fixed normalized weights

Return type:

jnp.ndarray

Notes

Using ‘initial_weights_logits’ means that the calculated initial weights have +ve entries and sum to one by construction. If ‘initial_weights’ is used the values are used unchecked.

calculate_weights(params, *args, **kwargs)

This function will be overridden for any pools that a) have weights and b) have weights that vary. As so many of the pools modelled in this package have weights (Balancer [G3M], Cow [FM-AMM], QuantAMM [TFMM]) this is helpful to have here (though this method is overriden for QuantAMM [TFMM] pools).

This method is used by some hooks that rely on having access to a pools weights over time. If a pool is to work with all hooks, this method should be ensured to implement the correct logic for that pool. See GyroscopePool for an example where a custom implementation was needed for the sake of hook compatibility.

Parameters:

• params (Dict[str, jnp.ndarray]) – Must contain ‘initial_weights_logits’ key or ‘initial_weights’ key

• *args – Not used, kept for interface compatibility

• **kwargs – Not used, kept for interface compatibility

Returns:

Fixed normalized weights

Return type:

jnp.ndarray

Notes

Using ‘initial_weights_logits’ means that the calculated initial weights have +ve entries and sum to one by construction. If ‘initial_weights’ is used the values are used unchecked.

abstract init_base_parameters(initial_values_dict, run_fingerprint, n_assets, n_parameter_sets=1, noise='gaussian')

Initialize base parameters specific to this pool type.

Parameters:

• initial_values_dict (Dict[str, Any])

• run_fingerprint (Dict[str, Any])

• n_assets (int)

• n_parameter_sets (int)

• noise (str)

Return type:

Dict[str, Any]

add_noise(params, noise, n_parameter_sets, noise_scale=1.0)

Parameters:

• params (Dict[str, ndarray])

• noise (str)

• n_parameter_sets (int)

• noise_scale (float)

Return type:

Dict[str, Array]

make_vmap_in_axes(params, n_repeats_of_recurred=0)

Configure JAX vectorization axes for pool parameters.

Parameters:

• params (Dict[str, Any]) – Pool parameters

• n_repeats_of_recurred (int) – Number of times to repeat recurrent parameters

Returns:

vmap axes configuration

Return type:

Dict[str, Any]

abstract is_trainable()

classmethod process_parameters(update_rule_parameters, run_fingerprint)

Default implementation for processing pool parameters from web interface input.

Performs simple conversion of parameter values to numpy arrays while preserving names. Override this method in subclasses that need custom parameter processing.

Parameters:

• update_rule_parameters (Dict[str, Any]) – Dict of parameters from the web interface

• run_fingerprint (Dict[str, Any]) – Run fingerprint dictionary

Returns:

Processed parameters ready for pool initialization

Return type:

Dict[str, np.ndarray]

weights_needs_original_methods()

Indicates if calculate_weights needs access to original pool methods.

Returns:

False by default - most pools don’t need original methods. Override in subclasses if they do.

Return type:

bool