MomentumPool

class MomentumPool[source]

Bases: TFMMBasePool

A class for momentum strategies run as TFMM (Temporal Function Market Making) liquidity pools, extending the TFMMBasePool class.

This class implements a momentum-based strategy for asset allocation within a TFMM framework. It uses price data to generate momentum signals, which are then translated into weight adjustments.

Parameters:

None

calculate_rule_outputs(params, run_fingerprint, prices, additional_oracle_input)[source]

Calculate the raw weight outputs based on momentum signals.

Parameters:
Return type:

Array

calculate_fine_weights(rule_output, initial_weights, run_fingerprint, params)[source]

Refine the raw weight outputs to produce final weights.

Parameters:
Return type:

Array

calculate_weights(params, run_fingerprint, prices, additional_oracle_input)

Orchestrate the weight calculation process.

Notes

The class provides methods to calculate raw weight outputs based on momentum signals and refine them into final asset weights, taking into account various parameters and constraints defined in the pool setup.

PARAM_SCHEMA = {'initial_weights_logits': ParamSpec(initial=1.0, optuna=OptunaRange(low=-10, high=10, log_scale=False, scalar=False), transform=None, description='Logit-space initial portfolio weights', trainable=False), 'log_k': ParamSpec(initial=4.32, optuna=OptunaRange(low=-6.6, high=12.0, log_scale=False, scalar=False), transform=None, description='Log2 of momentum sensitivity factor (k) per day', trainable=True), 'logit_delta_lamb': ParamSpec(initial=0.0, optuna=OptunaRange(low=-5.0, high=5.0, log_scale=False, scalar=False), transform=None, description='Delta in logit space for alternative lambda calculation', trainable=True), 'logit_lamb': ParamSpec(initial=4.0, optuna=OptunaRange(low=-4.0, high=8.0, log_scale=False, scalar=False), transform=None, description='Logit of decay parameter lambda (memory length)', trainable=True)}
classmethod get_param_schema()[source]

Get the full parameter schema for this pool.

Returns the pool-specific schema merged with common parameters from the base class.

Returns:

Complete parameter schema for this pool

Return type:

Dict[str, ParamSpec]

__init__()[source]

Initialize a new MomentumPool instance.

Parameters:

None

calculate_rule_outputs(params, run_fingerprint, prices, additional_oracle_input=None)[source]

Calculate the raw weight outputs based on momentum signals.

This method computes the raw weight adjustments for the momentum strategy. It processes the input prices to calculate gradients, which are then used to determine weight updates.

Parameters:

  • params (Dict[str, Any]) – A dictionary of strategy parameters.

  • run_fingerprint (Dict[str, Any]) – A dictionary containing run-specific settings.

  • prices (jnp.ndarray) – An array of asset prices over time.

  • additional_oracle_input (Optional[jnp.ndarray], optional) – Additional input data, if any.

Returns:

Raw weight outputs representing the suggested weight adjustments.

Return type:

jnp.ndarray

Notes

The method performs the following steps: 1. Calculates the memory days based on the lambda parameter. 2. Computes the ‘k’ factor which scales the weight updates. 3. Extracts chunkwise price values from the input prices. 4. Calculates price gradients using the calc_gradients function. 5. Applies the momentum weight update formula to get raw weight outputs.

The raw weight outputs are not the final weights, but rather the changes to be applied to the previous weights. These will be refined in subsequent steps.

calculate_readouts(params, run_fingerprint, prices, start_index, additional_oracle_input=None)[source]

Calculate readouts (internal gradient estimator variables) for the pool, based on price history.

This method gives the readout values for the gradient estimator (the ewma of prices and the running a), sliced in the same way that the raw weight outputs are sliced.

Parameters:

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

  • run_fingerprint (Dict[str, Any]) – Simulation settings

  • prices (jnp.ndarray) – Historical price data

  • additional_oracle_input (Optional[jnp.ndarray]) – Extra data for weight calculation

  • start_index (Array)

Returns:

Dict containing readout value from the gradient estimator

Return type:

dict

get_initial_rule_state(initial_price, params, run_fingerprint)[source]

Initialize the carry state for scanning.

For MomentumPool, the carry consists of: - ewma: initialized to the first price - running_a: initialized to zeros (steady-state for constant input)

Parameters:

  • initial_price (jnp.ndarray) – First price observation (shape: n_assets,)

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

  • run_fingerprint (Dict[str, Any]) – Simulation settings

Returns:

Initial carry state with ‘ewma’ and ‘running_a’ keys.

Return type:

Dict[str, jnp.ndarray]

calculate_rule_output_step(carry, price, params, run_fingerprint)[source]

Calculate a single step of momentum weight update.

This mirrors the production implementation where we: 1. Update the gradient estimator state (ewma, running_a) 2. Compute the gradient from the updated state 3. Apply the momentum weight update formula

Parameters:

  • carry (Dict[str, jnp.ndarray]) – Current state with ‘ewma’ and ‘running_a’

  • price (jnp.ndarray) – Current price observation (shape: n_assets,)

  • params (Dict[str, Any]) – Pool parameters (logit_lamb, log_k, etc.)

  • run_fingerprint (Dict[str, Any]) – Simulation settings (chunk_period, max_memory_days, etc.)

Returns:

(new_carry, rule_output)

Return type:

tuple

calculate_fine_weights(rule_output, initial_weights, run_fingerprint, params)[source]

Refine raw weight outputs to produce final weights for the momentum pool.

This method takes the raw weight outputs calculated from momentum signals and refines them into final asset weights. It applies various constraints and adjustments defined in the pool parameters and run fingerprint.

Parameters:

  • rule_output (jnp.ndarray) – Raw weight changes or outputs from momentum calculations.

  • initial_weights (jnp.ndarray) – Initial weights of assets in the pool.

  • run_fingerprint (Dict[str, Any]) – Dictionary containing run-specific parameters and settings.

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

Returns:

Refined weights for each asset in the pool over the specified time period.

Return type:

jnp.ndarray

Notes

Uses the calc_fine_weight_output_from_weight_changes function to perform the actual refinement. The implementation of this function should handle details such as weight interpolation, maximum change limits, and ensuring weights sum to 1.

init_base_parameters(initial_values_dict, run_fingerprint, n_assets, n_parameter_sets=1, noise='gaussian')[source]

Initialize parameters for the momentum pool.

This method sets up the initial parameters for the momentum pool strategy, including weights, memory length (lambda), and the momentum factor (k).

Parameters:

  • initial_values_dict (Dict[str, Any]) – Dictionary containing initial values for various parameters.

  • run_fingerprint (Dict[str, Any]) – Dictionary containing run-specific settings and parameters.

  • n_assets (int) – The number of assets in the pool.

  • n_parameter_sets (int, optional) – The number of parameter sets to initialize, by default 1.

  • noise (str, optional) – The type of noise to apply during initialization, by default “gaussian”.

Returns:

Dictionary containing the initialized parameters for the momentum pool.

Return type:

Dict[str, jnp.array]

Raises:

ValueError – If required initial values are missing or in an incorrect format.

Notes

This method handles the initialization of parameters for initial weights, lambda (memory length parameter), and k (momentum factor) for each asset and parameter set. It processes the initial values to ensure they are in the correct format and applies any necessary transformations (e.g., logit transformations for lambda).