PowerChannelPool

class PowerChannelPool

Bases: MomentumPool

A class for power channel strategies run as TFMM liquidity pools.

This class implements a “power channel” strategy for asset allocation within a TFMM framework. It uses price data to generate power channel signals, which are then translated into weight adjustments.

Parameters:

None

calculate_rule_outputs(params, run_fingerprint, prices, additional_oracle_input)

Calculate the raw weight outputs based on power channel signals.

Parameters:

• params (Dict[str, Any])

• run_fingerprint (Dict[str, Any])

• prices (Array)

• additional_oracle_input (Array | None)

Return type:

Array

Notes

The PowerChannelPool implements a “power channel” strategy for asset allocation within a TFMM framework. It uses price data to generate power channel signals, which are then translated into weight adjustments. The class provides methods to calculate raw weight outputs based on these 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), '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', 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), 'sp_exponents': ParamSpec(initial=0.0, optuna=OptunaRange(low=-2.0, high=4.0, log_scale=False, scalar=False), transform=None, description='Squareplus-space exponents (typically gives 0.3-5)', trainable=True), 'sp_k': ParamSpec(initial=19.5, optuna=OptunaRange(low=-1.0, high=100.0, log_scale=False, scalar=False), transform=None, description='Squareplus-space k factor (squareplus gives k_per_day)', trainable=True), 'sp_pre_exp_scaling': ParamSpec(initial=-1.0, optuna=OptunaRange(low=-3.0, high=2.0, log_scale=False, scalar=False), transform=None, description='Squareplus-space pre-exp scaling (gives 0.09-2.4)', trainable=True)}

classmethod get_param_schema()

Get the full parameter schema for PowerChannelPool.

Return type:

dict

__init__()

Initialize a new PowerChannelPool instance.

Parameters:

None

calculate_rule_outputs(params, run_fingerprint, prices, additional_oracle_input=None)

Calculate the raw weight outputs based on power channel signals.

This method computes the raw weight adjustments for the power channel 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 power channel 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_rule_output_step(carry, price, params, run_fingerprint)

Calculate a single step of power channel 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 power channel 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, sp_k, sp_exponents, etc.)

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

Returns:

(new_carry, rule_output)

Return type:

tuple

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

Initialize parameters for a power channel pool.

This method sets up the initial parameters for the power channel pool strategy, including weights, memory length (lambda), the update aggressiveness (k) and the exponents.

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), the update agressiveness (k) and the exponents 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).