Skip to main content

strategy.strategy_base

functools​

json​

ABC​

abstractmethod​

datetime​

timezone​

Decimal​

Path​

pprint​

Optional​

Tuple​

GoogleAPICallError​

NotFound​

storage​

retry​

ExecutionStatus​

MetricAggType​

MetricsAggTable​

MetricsLogger​

ActionBundle​

ProfileType​

time_block​

ExecutionManager​

InternalFlowStatus​

Mode​

StateBase​

SubStateBase​

get_current_price​

to_readable​

Config​

get_logger​

download_action_bundle_from_storage​

STORAGE_DIR​

READ_ONLY_MODE​

PERSISTENT_STATE_FILENAME​

IS_AGENT_DEPLOYMENT​

logger​

Strategy Objects​

class Strategy(ABC)

N_NOT_INCLUDED_UNLOCK_SOFT_RAISE​

NotIncludedUnlockError Objects​

class NotIncludedUnlockError(Exception)

Exception raised when a NOT_INCLUDED still fails after N_NOT_INCLUDED_UNLOCK_SOFT_RAISE retries.

__init__​

def __init__(
    message="NOT_INCLUDED max retries reached. Soft raise to unblock main thread."
)

__init__​

def __init__()

run​

@abstractmethod
def run()

restart_cycle​

@abstractmethod
def restart_cycle()

is_locked​

@property
def is_locked()

Check if the strategy cycle has completed and now waiting the next.

is_paused​

@property
def is_paused()

is_initialized​

@property
def is_initialized()

Check if the strategy is initialized.

is_terminated​

@property
def is_terminated()

Check if the strategy is terminated (teardown finished)

__repr__​

@abstractmethod
def __repr__() -> str

get_persistent_state_model​

@classmethod
@abstractmethod
def get_persistent_state_model(cls)

Pydandic model defined in child's class.

get_config_model​

@classmethod
@abstractmethod
def get_config_model(cls)

Pydandic model defined in child's class.

save_persistent_state​

@retry(tries=3, delay=2, backoff=2, jitter=(1, 3), exceptions=(Exception, ))
def save_persistent_state() -> None

save_persistent_state_local​

def save_persistent_state_local() -> None

save_persistent_state_gcs​

def save_persistent_state_gcs() -> None

Saves the strategy's persistent state to GCS ensuring that the strategy can resume operation from where it last left off after a restart or shutdown.

Raises:

  • FileNotFoundError - If the configuration file is not found at the specified path.
  • ValueError - If there is an error decoding JSON from the configuration file.
  • EnvironmentError - If required environment variables are not set.
  • PermissionsError - If there are insufficient permissions to access the file.
  • TimeoutError - If the request to download the configuration file times out.
  • GoogleAPICallError - For network-related errors or issues on the backend from Google Cloud services.

load_persistent_state​

@retry(tries=3, delay=2, backoff=2, jitter=(1, 3), exceptions=(Exception, ))
def load_persistent_state() -> None

load_persistent_state_bytes_local​

def load_persistent_state_bytes_local() -> None

load_persistent_state_bytes_gcs​

def load_persistent_state_bytes_gcs() -> None

Loads the strategy's state from persistent storage in GCS.

This method is responsible for restoring the state of the strategy from GCS, ensuring that the strategy can resume operation from where it last left off after a restart or shutdown.

check_for_persistent_state_file​

def check_for_persistent_state_file() -> bool

check_for_persistent_state_file_local​

def check_for_persistent_state_file_local() -> bool

check_for_persistent_state_file_gcs​

def check_for_persistent_state_file_gcs() -> bool

Mainly a utility function to check if the persistent state file exists.

upload_persistent_state​

def upload_persistent_state(template_path: str, force_upload=False)

Loads the local json template and saves it as the persistent state. That step is meant to be done manually when deploying a strategy!

Arguments:

  • template_path str - Path to the JSON template file.
  • force_upload bool - Whether to force upload the persistent state.

initialize_persistent_state​

def initialize_persistent_state(template_path: str)

load_executioner_state​

@retry(tries=3, delay=2, backoff=2, jitter=(1, 3), exceptions=(Exception, ))
def load_executioner_state(action_id: str) -> dict

Loads the executioner's state from persistent storage in GCS.

Note: Only supports Pickle format for now as serializing Web3 objects to JSON is not straightforward.

validate_executioner_action_bundle​

def validate_executioner_action_bundle(action_bundle: ActionBundle)

This function is called while loading the persistent state, when re-entering the strategy. First the Strategy Persistent State is loaded, then the Executioner Status is loaded based on the current action_id.

handle_state_with_actions​

def handle_state_with_actions(prepare_fn,
                              validate_fn,
                              sadflow_fn,
                              next_state,
                              next_substate=None)

log_strategy_balance_metrics​

@abstractmethod
def log_strategy_balance_metrics(action_id: str)

StrategyUniV3 Objects​

class StrategyUniV3(Strategy)

__init__​

def __init__()

get_wallet_active_positions​

def get_wallet_active_positions()

Retrieves and identifies all active positions. Returns only the one(s) that match the strategy's pool and with liquidity (i.e. active).

IMPORTANT: This is wallet-level, not strategy-level.

Returns:

Dict[int, Tuple]: A dictionary of active positions where keys are position IDs and values are the position details. The position details are the default one returned by Uniswap V3 enhanced with 2 additional values appended: token0 and token1 amount (from liquidity value to token amount).

get_active_position_info​

def get_active_position_info(position_id: int)

calculate_desired_amounts​

def calculate_desired_amounts(amount0_initial: int, amount1_initial: int,
                              ratio: float, spot_price: float,
                              token0_decimals: int,
                              token1_decimals: int) -> Tuple[int, int]

Calculates the desired amounts based on a specified ratio and the current spot price, using Decimal for intermediate calculations and converting the result back to int.

Arguments:

  • amount0_initial int - The current amount of token0 in native units (e.g., Wei).
  • amount1_initial int - The current amount of token1 in native units (e.g., Wei).
  • ratio float - The desired ratio of token0 to token1 in the pool.
  • spot_price float - The current market spot price of token1 measured in units of token0.
  • token0_decimals int - The number of decimals for token0.
  • token1_decimals int - The number of decimals for token1.

Returns:

Tuple[int, int]: The desired amounts of token0 and token1 in native units (e.g., Wei).

calculate_reswap_amounts​

def calculate_reswap_amounts(
        amount0_initial: int,
        amount1_initial: int,
        ratio: float,
        spot_price: float,
        token0_decimals: int,
        token1_decimals: int,
        amount0_desired: Optional[int] = None,
        amount1_desired: Optional[int] = None
) -> Tuple[Optional[bool], int, int]

Calculates the specific amounts of tokens to swap to achieve the desired ratio, using Decimal for intermediate calculations and converting the result back to int.

Arguments:

  • amount0_initial int - The current amount of token0 in native units (e.g., Wei).
  • amount1_initial int - The current amount of token1 in native units (e.g., Wei).
  • ratio float - The target ratio of token0 to token1 that the portfolio should aim to achieve.
  • spot_price float - The current market price of token1 in terms of token0.
  • token0_decimals int - The number of decimals for token0.
  • token1_decimals int - The number of decimals for token1.
  • amount0_desired Optional[int] - The desired amount of token0 in native units (e.g., Wei).
  • amount1_desired Optional[int] - The desired amount of token1 in native units (e.g., Wei).

Returns:

Tuple[Optional[bool], int, int]: A tuple containing:

  • A boolean indicating whether token0 needs to be swapped for token1 (True) or vice versa (False).
  • The amount of the input token to be swapped in native units.
  • The amount of the output token to be received in native units.

get_balances​

def get_balances(native_format=False) -> Tuple[float, float, float]

show_balances​

def show_balances()

show_positions​

def show_positions()

Display the current positions of the wallet. Debug Function.

show_state​

def show_state(show_persistent_state: bool = False)

Display the current state and flowstatus of the strategy.

log_snapshot​

def log_snapshot(action_id: Optional[str] = None,
                 bundle_id: Optional[str] = None,
                 block_number: Optional[int] = None)