Skip to main content

Create a Strategy

REVIEW: Ari

Before jumping into the code, let's clarify Almanak's Agent & Strategy Stack Framework.

There are three (3) main components at play:

  1. The "main": as it name implies, it refers to the main.py or simply the entry point orchestrating the agents/strategies.
  2. The Strategy: the strategy represents the logic the user is trying to convey, that what the user is coding.
  3. The Executioner: the execution layer handling all the abstraction for the strategy.

There are obviously many subparts to the Strategy and the Executioner but conceptually the Strategy's goal is to produce Actions based on various conditions and the main will relay such Actions to the Executioner to execute on-chain transactions.

As a user, you won't be modifying the main nor the Executioner, only the Strategy. So let's focus on that part.

info

If you want the simplest working strategy look for the Tutorial - Hello World! strategy. You can clone it and start from there.

Strategy Framework​

For a robust and reliable Strategy framework we opted for a state machine with a persistent state. For conveniency and scalability we abstracted away all the execution, smart contract and on-chain transactions related stuff. For flexibility we implemented a configuration schema to quickly (hot) change important parameters of the strategy. For clarity and consistency we use Pydantic models across the stack.

Strategy Class​

Creating a strategy means creating a folder with the following structure:

.
├── __init__.py
├── states
│   ├── __init__.py
│   ├── initialization.py
│   └── teardown.py
└── templates
    ├── config_template.json
    └── persistent_state_template.json
└── presets
    ├── ... [ **coming soon** ]
├── models.py
├── strategy.py
├── strategy.toml [ **coming soon** ]

models.py​

Contains all the Pydantic models, namely:

  • State: all the states of the state machine for the strategy.
  • Substate: should any state have substates (a state machine within a state machine).
  • PersistentState: the persistent state class.
  • StrategyConfig: the configuration class.

Any other Pydantic needed across the Strategy should be added here.

# models.py

from enum import Enum

from src.almanak_library.enums import Chain, Network, Protocol
from src.strategy.models import PersistentStateBase, StrategyConfigBase


class State(Enum):
    """Enum representing the state of the strategy."""

    INITIALIZATION = "INITIALIZATION"
    COMPLETED = "COMPLETED"  # A "Cycle" is completed in between Checks for Rebalance
    TEARDOWN = "TEARDOWN"
    TERMINATED = "TERMINATED"  # The Strategy is terminated (manual intervention required)


class SubState(Enum):
    """Enum representing the substates of some of the strategy states. A state machine within a state machine."""

    NO_SUBSTATE = "NO_SUBSTATE"


class PersistentState(PersistentStateBase):
    current_state: State
    current_substate: SubState


class StrategyConfig(StrategyConfigBase):
    id: str
    network: Network
    chain: Chain
    protocol: Protocol

strategy.py​

That's the main file of the strategy, defining the class and implementing the run() function moving throught the states. That's the function that the main will call to enter the strategy. It is a re-entrant function, therefore the states themselves need to be developped as re-entrant (e.g. avoiding persistent_state.var += value).

import json
import os
import sys
from time import sleep

from src.almanak_library.models.action_bundle import ActionBundle
from src.strategy.strategies.template_hello_world.models import (
    PersistentState,
    State,
    StrategyConfig,
    SubState,
)
from src.strategy.strategy_base import StrategyUniV3

from .states.display_message import display_message
from .states.initialization import initialization
from .states.teardown import teardown


class StrategyTemplateHelloWorld(StrategyUniV3):
    # TODO: Replace Strategy_Template with your strategy name, both here as well as in config.json
    STRATEGY_NAME = "Template_Hello_World"

    def __init__(self, **kwargs):
        """
        Initialize the strategy with given configuration parameters.

        Args:
            **kwargs: Strategy-specific configuration parameters.
        """
        super().__init__()
        self.name = self.STRATEGY_NAME.replace("_", " ")

        # Overwrite the States and SubStates for this Strategy
        self.State = State
        self.SubState = SubState

        # Get configuration from kwargs
        try:
            self.config = StrategyConfig(**kwargs)
        except Exception as e:
            raise ValueError(f"Invalid Strategy Configuration. {e}")

        self.id = self.config.id
        self.chain = self.config.chain
        self.network = self.config.network
        self.protocol = self.config.protocol

        # TODO: Move me to parent class if this becomes permanent.
        def _upload_persistent_state(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!
            """
            if not self.check_for_persistent_state_file() or force_upload:
                current_dir = os.path.dirname(os.path.abspath(__file__))
                local_file_path = os.path.join(current_dir, "templates", "persistent_state_template.json")
                try:
                    with open(local_file_path, "r") as file:
                        json_data = json.load(file)

                    self.persistent_state = self.PersistentStateModel(**json_data)
                    self.save_persistent_state()
                    print("New Persistent State uploaded.")

                except Exception as e:
                    raise Exception(f"Unexpected error occurred: {str(e)}")

        _upload_persistent_state()

        try:
            self.load_persistent_state()
        except Exception as e:
            raise ValueError(f"Unable to load persistent state. {e}")

    def __repr__(self):
        return (
            f"{self.__class__.__name__}(id={self.id}, "
            f"chain={self.chain}, network={self.network}, protocol={self.protocol}, "
            f"wallet_address={self.wallet_address}, mode={self.mode})"
        )

    @classmethod
    def get_persistent_state_model(cls):
        return PersistentState

    @classmethod
    def get_config_model(cls):
        return StrategyConfig

    @property
    def is_locked(self):
        """Check if the strategy is locked."""
        # TODO: Implement the strategy lock logic
        return True

    def restart_cycle(self) -> None:
        """A Strategy should only be restarted when the full cycle is completed."""
        if self.persistent_state.current_state == self.State.COMPLETED:
            # Properly restart the cycle
            self.persistent_state.current_flowstatus = self.InternalFlowStatus.PREPARING_ACTION
            self.persistent_state.current_state = self.State.DISPLAY_MESSAGE
            self.persistent_state.completed = False

            # Dump the state to the persistent state because we load it when called.
            self.save_persistent_state()
        elif self.persistent_state.current_state == self.State.TERMINATED:
            print("Strategy is terminated, nothing to restart.")
        else:
            raise ValueError("The strategy is not completed yet, can't restart.")

    def run(self):
        """
        Executes the strategy by progressing through its state machine based on the current state.

        This method orchestrates the transitions between different states of the strategy,
        performing actions as defined in each state, and moves to the next state based on the
        actions' results and strategy's configuration.

        Returns:
            dict: A dictionary containing the current state, next state, and actions taken or
                recommended, depending on the execution mode.

        Raises:
            ValueError: If an unknown state is encountered, indicating a potential issue in state management.

        Notes:
            - This method is central to the strategy's operational logic, calling other methods
            associated with specific states like initialization, rebalancing, or closing positions.
            - It integrates debugging features to display balances and positions if enabled.
        """
        # TODO: Implement the main strategy execution logic
        print("Running the strategy")
        print(self.persistent_state)

        # NOTE: Here we are cheating a little by changing the state directly from here.
        #       Usually, the state change would be handled in the state's function/file.
        actions = None
        match self.persistent_state.current_state:
            case State.INITIALIZATION:
                actions = initialization(self)
                self.persistent_state.current_state = State.DISPLAY_MESSAGE
            case State.DISPLAY_MESSAGE:
                actions = display_message(self)
                self.persistent_state.current_state = State.COMPLETED
            case State.COMPLETED:
                self.complete()
            case State.TEARDOWN:
                teardown(self)
                # Here we would set the state to TERMINATED, but for a nicer demo we loop.
                self.persistent_state.current_state = State.INITIALIZATION
            case self.State.TERMINATED:
                print("Strategy is terminated.")
                actions = None
            case _:
                raise ValueError(f"Unknown state: {self.persistent_state.current_state}")

        # Save actions to current state to load the executioner status from them when re-entering.
        if actions is None:
            self.persistent_state.current_actions = []
        elif isinstance(actions, ActionBundle):
            self.persistent_state.current_actions = [actions.id]
        else:
            raise ValueError(f"Invalid actions type. {type(actions)} : {actions}")

        # Always save the persistent state before leaving the Strategy's state machine
        self.save_persistent_state()
        return actions

    def complete(self) -> None:
        self.persistent_state.current_state = self.State.COMPLETED
        self.persistent_state.current_flowstatus = self.InternalFlowStatus.PREPARING_ACTION
        self.persistent_state.completed = True

    def log_strategy_balance_metrics(self, action_id: str):
        """Logs strategy balance metrics per action. It is called in the StrategyBase class."""
        pass

States Folder​

Each state, except for very simple states, should have it's own my_state.py file in that folder. That makes it easier to navigate the code for improving, scaling and troubleshooting.

The name of the state in the enum, the name of the file and the entry function for that state, should be the same or closely related to make is easier to nagivate and understand.

For example:

class State(Enum):
    ...
    DISPLAY_MESSAGE = "DISPLAY_MESSAGE"
# display_message.py

def display_message(...):
    ...

Persistent State​

All important variables of the Strategy class, like state, accounting, etc. needs to be saved in the persistent state and not directly in the class variables. The reason for such a design is to be robust towards the machine shutting down at any moment. Should the machine crash for any reason, the strategy needs to be able to re-instanciate itself and retrieve its memory. When developping a strategy, assume that anything outside of the pesistent_state can be lost at any moment. Would that affect the behaviour of the strategy?

# This
self.current_state = ...
# Becomes
self.persistent_state.current_state = ...

Each Strategy must have a ./templates/persistent_state_template.json. The platform will use this template as a starting point when launching the Strategy.

{
  "current_state": "INITIALIZATION",
  "current_flowstatus": "PREPARING_ACTION",
  "current_actions": [],
  "current_substate": "NO_SUBSTATE",
  "initialized": false,
  "completed": false,
  "last_message": "N/A"
}

Config​

Most strategy will have parameters for the user to configure. This is achieve via the StrategyConfig class in the models.py. It is a Pydantic class which will load a config.json (taken from the /templates/config_templates.json).

The minimum mandatory fields in the config are the following:

{
  "strategy_configs": {
    "HelloWorld-1": {
      "strategy_name": "Template_Hello_World",
      "parameters": {
        "network": "MAINNET",
        "chain": "ARBITRUM",
        "protocol": "UNISWAP_V3",
        "message": "Almanak rocks!"
      }
    }
  }
}

State Machine​

As mentioned before the strategy is built as a state machine, however the robustness of the system really comes from our Prepare / Validate / Sadflow design.

Each Action state, i.e. performing actions, not just updating internal states or displaying messages, must use the Prepare -> Validate | Sadflow design.

Prepare: The function responsible to preparing (i.e. creating) the Action(s).
Validate: The function responsible to validating that the action executed as intented. For example by looking at the Transaction details (i.e. the receipt) to confirm that the swap amount is what was intented and then update internal accounting accordingly. The only way for the state machine to move on to the next state from an action state is after the validate was successful (returned True).
Sadflow: This will fail, transactions will revert. In these situations the strategy will enter into a Sadflow. More on Sadflows below.

The implementation is the following: the entry function of the state will call the handle_state_with_actions()

# Example of Close Position state
def close_position(strat: "StrategyDegenLPChaser") -> ActionBundle:
    """
    ...
    """
    return strat.handle_state_with_actions(
        prepare_fn=partial(prepare_close_position, strat),
        validate_fn=partial(validate_close_position, strat),
        sadflow_fn=partial(sadflow_close_position, strat),
        next_state=strat.State.SWAP_ASSETS,
    )

Actions​

For more information about Actions and Action Bundle please see [INSERT LINK].

Any on-chain operation is an Action, bundled into an Action Bundle. Up to 3 actions per bundle.

The Action Bundle is what will be passed across the stack and different layer will use it in different ways adding status information across the layers (e.g. constructing the action, then building the transaction(s), then adding the TX has, then the execution status, then parsing the receipt, etc.)

action_approve = Action(
    type=ActionType.APPROVE,
    params=ApproveParams(
        token_address=token_address,
        spender_address=spender_address,
        from_address=from_address,
        amount=int(amount),
    ),
    protocol=protocol,
)

action_swap = Action(
    type=ActionType.SWAP,
    params=SwapParams(
        side=SwapSide.SELL,
        tokenIn=strat.token1.address,
        tokenOut=strat.token0.address,
        fee=strat.fee,
        recipient=strat.wallet_address,
        amount=int(swap_amount_in),
        slippage=strat.slippage_swap,
    ),
    protocol=strat.protocol,
)

return ActionBundle(
    actions=[action_approve, action_swap],
    chain=strat.chain,
    network=strat.network,
    strategy_id=strat.id,
    config=strat.config,
    persistent_state=strat.persistent_state,
)

Init & Teardown​

The minimum required states of a strategy are:

class StateBase(Enum):
    INITIALIZATION = "INITIALIZATION"
    COMPLETED = "COMPLETED"
    TEARDOWN = "TEARDOWN"
    TERMINATED = "TERMINATED"

Initialization is where initial early steps are required before the strategy actual starts. In an LP strategy for example, that could mean obtaining Token0 and Token1 from ETH or USDT. The Initilization is likely to have Substates if different Actions have to be performed as part of the initialization process.

Initialization happens once in the life of a deployed strategy. To "restart", it has to be teared down and then re-deployed.

Teardown is where the last steps to terminate the strategy occur. The teardown is a clean process doing the necessary action In an LP strategy this could mean closing the position and selling Token0 and Token1 to get back into ETH or USDT.

Once a strategy has validated the teardown state, it must go into Terminated state. This will let the system know that the strategy is fully done and cleanly terminated.

Completed vs Terminated​

In the strategy there is the concept of cycle meaning a sequence of Actions that should not be interrupted. For example, in an LP strategy a rebalance sequence (Close, Open) should not be interrupted. This is called a cycle. After a cycle is completed, the strategy should go into the Completed state. This will "release" the main and allow it to execute

Pausing the Strategy can only be done when a cycle is completed and therefore the strategy is in a Completed state.

Tearing down the Strategy can only be done when a cycle is completed and there the strategy is in a Completed state.

Sadflow​

This part is probably the most complicated one of the strategy since it requires a good understanding of what can go wrong within each different state. Each state has its own custom Sadflow.

important

At Almanak when developing a strategy we have 2 rules: 1- The Strategy must never do anything stupid. 2- In doubt, refer to rule #1.

What it means is to put raise everywhere in the code to make sure the strategy doesn't end up in a state it shouldn't be. Money at stake.