Strategies¶

Strategy base classes and the market snapshot interface.

Implementing Teardown¶

Every strategy must implement three teardown methods so operators can safely close positions. Without them, close-requests are silently ignored.

Required Methods¶

Method	Purpose
`supports_teardown() -> bool`	Return `True` to enable teardown
`get_open_positions()`	Return a `TeardownPositionSummary` listing all open positions
`generate_teardown_intents(mode, market)`	Return an ordered list of `Intent` objects that unwind positions

Execution Order¶

If your strategy holds multiple position types, teardown intents must follow this order:

PERP -- close perpetual positions first (highest risk)
BORROW -- repay borrows to free collateral
SUPPLY -- withdraw supplied collateral
LP -- close liquidity positions
TOKEN -- swap remaining tokens to stable

Example: Swap Strategy Teardown¶

from decimal import Decimal
from almanak import IntentStrategy, Intent

class MyStrategy(IntentStrategy):
    def supports_teardown(self) -> bool:
        return True

    def get_open_positions(self) -> "TeardownPositionSummary":
        from datetime import UTC, datetime
        from almanak.framework.teardown import (
            PositionInfo, PositionType, TeardownPositionSummary,
        )
        return TeardownPositionSummary(
            strategy_id=getattr(self, "strategy_id", "my_strategy"),
            timestamp=datetime.now(UTC),
            positions=[
                PositionInfo(
                    position_type=PositionType.TOKEN,
                    position_id="my_strategy_eth",
                    chain=self.chain,
                    protocol="uniswap_v3",
                    value_usd=Decimal("1000"),  # query on-chain balance, not cache
                    details={"asset": "WETH"},
                )
            ],
        )

    def generate_teardown_intents(self, mode: "TeardownMode", market=None) -> list[Intent]:
        from almanak.framework.teardown import TeardownMode
        max_slippage = Decimal("0.03") if mode == TeardownMode.HARD else Decimal("0.005")
        return [
            Intent.swap(
                from_token="WETH", to_token="USDC",
                amount="all", max_slippage=max_slippage, protocol="uniswap_v3",
            )
        ]

Always query on-chain state

get_open_positions() must query live on-chain balances, not cached values. Stale data can cause teardown to skip positions or attempt to close positions that no longer exist.

IntentStrategy¶

The primary base class for writing strategies. Implement the decide() method to return an Intent.

almanak.framework.strategies.IntentStrategy ¶

IntentStrategy(
    config: ConfigT,
    chain: str,
    wallet_address: str,
    risk_guard_config: RiskGuardConfig | None = None,
    notification_callback: NotificationCallback
    | None = None,
    compiler: IntentCompiler | None = None,
    state_machine_config: StateMachineConfig | None = None,
    price_oracle: PriceOracle | None = None,
    rsi_provider: RSIProvider | None = None,
    balance_provider: BalanceProvider | None = None,
    rpc_url: str | None = None,
    wallet_activity_provider: WalletActivityProvider
    | None = None,
)

Bases: StrategyBase[ConfigT]

Base class for Intent-based strategies.

IntentStrategy simplifies strategy development by allowing developers to write just a decide() method that returns an Intent. The framework handles:

Market data access via MarketSnapshot
Intent compilation to ActionBundle
State machine generation for execution
Hot-reloadable configuration
Error handling and retries

Subclasses must implement the abstract decide() method.

Example

@almanak_strategy(name="simple_strategy") class SimpleStrategy(IntentStrategy): def decide(self, market: MarketSnapshot) -> Optional[Intent]: if market.rsi("ETH").is_oversold: return Intent.swap("USDC", "ETH", amount_usd=Decimal("100")) return Intent.hold()

Attributes:

Name	Type	Description
`compiler`	`IntentCompiler`	IntentCompiler for converting intents to action bundles
`state_machine_config`		Configuration for state machine execution
`_current_intent`	`AnyIntent \| None`	Currently executing intent (if any)
`_current_state_machine`	`IntentStateMachine \| None`	Current state machine (if any)

Initialize the intent strategy.

Parameters:

Name	Type	Description	Default
`config`	`ConfigT`	Hot-reloadable configuration	required
`chain`	`str`	Chain to operate on (e.g., "arbitrum")	required
`wallet_address`	`str`	Wallet address for transactions	required
`risk_guard_config`	`RiskGuardConfig \| None`	Risk guard configuration	`None`
`notification_callback`	`NotificationCallback \| None`	Callback for operator notifications	`None`
`compiler`	`IntentCompiler \| None`	Intent compiler (required for direct run() calls, optional for runner)	`None`
`state_machine_config`	`StateMachineConfig \| None`	State machine configuration	`None`
`price_oracle`	`PriceOracle \| None`	Function to fetch prices	`None`
`rsi_provider`	`RSIProvider \| None`	Function to calculate RSI (token, period[, timeframe=]) -> RSIData	`None`
`balance_provider`	`BalanceProvider \| None`	Function to fetch balances	`None`
`rpc_url`	`str \| None`	RPC URL for on-chain queries (needed for LP close)	`None`
`wallet_activity_provider`	`WalletActivityProvider \| None`	Provider for leader wallet activity signals	`None`

chain `property` ¶

chain: str

Get the chain name.

wallet_address `property` ¶

wallet_address: str

Get the wallet address.

compiler `property` `writable` ¶

compiler: IntentCompiler

Get the intent compiler.

Raises:

Type	Description
`RuntimeError`	If compiler was not provided and is accessed directly. The StrategyRunner creates its own compiler with real prices, so this is only needed for direct run() calls.

current_intent `property` ¶

current_intent: AnyIntent | None

Get the currently executing intent.

current_state_machine `property` ¶

current_state_machine: IntentStateMachine | None

Get the current state machine.

set_state_manager ¶

set_state_manager(
    state_manager: Any, strategy_id: str
) -> None

Set the state manager for persistence.

Called by the runner to inject the state manager.

Parameters:

Name	Type	Description	Default
`state_manager`	`Any`	StateManager instance	required
`strategy_id`	`str`	Unique ID for this strategy instance	required

get_persistent_state ¶

get_persistent_state() -> dict[str, Any]

Get strategy state to persist.

Override this method to define what state should be persisted. Default implementation returns empty dict (no state).

Returns:

Type	Description
`dict[str, Any]`	Dict of state key-value pairs to persist

load_persistent_state ¶

load_persistent_state(state: dict[str, Any]) -> None

Load persisted state into the strategy.

Override this method to restore state from persistence. Default implementation does nothing.

Parameters:

Name	Type	Description	Default
`state`	`dict[str, Any]`	Dict of state key-value pairs loaded from storage	required

save_state ¶

save_state() -> None

Save current strategy state to persistence.

Called by runner after each iteration.

flush_pending_saves `async` ¶

flush_pending_saves() -> None

Wait for any pending save operations to complete.

This should be called before disconnecting from the gateway to ensure all state saves have completed. Handles both successful completion and errors gracefully.

load_state ¶

load_state() -> bool

Load strategy state from persistence.

Called by runner on startup.

Returns:

Type	Description
`bool`	True if state was found and loaded, False otherwise

decide `abstractmethod` ¶

decide(market: MarketSnapshot) -> DecideResult

Decide what action to take based on current market conditions.

This is the main method that strategy developers need to implement. It receives a MarketSnapshot with current market data and should return an Intent, IntentSequence, list of intents, or None.

Parameters:

Name	Type	Description	Default
`market`	`MarketSnapshot`	Current market snapshot with prices, balances, RSI, etc.	required

Returns:

Type	Description
`DecideResult`	One of:
`DecideResult`	Single Intent: Execute one action
`DecideResult`	IntentSequence: Execute multiple actions sequentially (dependent)
`DecideResult`	list[Intent \| IntentSequence]: Execute items in parallel
`DecideResult`	None: Take no action (equivalent to Intent.hold())
`DecideResult`	Returning None is equivalent to returning Intent.hold().

Example

def decide(self, market: MarketSnapshot) -> DecideResult: # Single intent if market.rsi("ETH").is_oversold: return Intent.swap("USDC", "ETH", amount_usd=Decimal("1000"))

# Sequence of dependent actions (execute in order)
if should_move_funds:
    return Intent.sequence([
        Intent.swap("USDC", "ETH", amount=Decimal("1000"), chain="base"),
        Intent.supply(protocol="aave_v3", token="WETH", amount=Decimal("0.5"), chain="arbitrum"),
    ])

# Multiple independent actions (execute in parallel)
if should_rebalance:
    return [
        Intent.swap("USDC", "ETH", amount=Decimal("500"), chain="arbitrum"),
        Intent.swap("USDC", "ETH", amount=Decimal("500"), chain="optimism"),
    ]

# No action
return Intent.hold(reason="RSI in neutral zone")

on_intent_executed ¶

on_intent_executed(
    intent: Any, success: bool, result: Any
) -> None

Called after each intent execution completes.

Override this method to react to execution results, e.g., to track position IDs, log swap amounts, or update state based on results.

The result object is enriched by the framework with extracted data that "just appears" based on intent type: - SWAP: result.swap_amounts (SwapAmounts) - LP_OPEN: result.position_id, result.extracted_data["liquidity"] - LP_CLOSE: result.lp_close_data (LPCloseData) - PERP_OPEN: result.extracted_data["entry_price"], ["leverage"]

Parameters:

Name	Type	Description	Default
`intent`	`Any`	The intent that was executed	required
`success`	`bool`	Whether execution succeeded	required
`result`	`Any`	ExecutionResult with enriched data	required

valuate ¶

valuate(market: MarketSnapshot) -> Decimal

Calculate the total portfolio value in USD for vault settlement.

Called by the framework during vault settlement to determine the current value of the strategy's holdings. The returned value is converted to underlying token units and proposed as the new totalAssets for the vault.

The default implementation sums balance_usd for all known token balances in the market snapshot. Override this method for custom valuation logic (e.g., including LP positions, pending rewards, or off-chain assets).

Parameters:

Name	Type	Description	Default
`market`	`MarketSnapshot`	Current market snapshot with prices and balances	required

Returns:

Type	Description
`Decimal`	Total portfolio value in USD as a Decimal

on_vault_settled ¶

on_vault_settled(settlement: SettlementResult) -> None

Called after a vault settlement cycle completes.

Override this method to react to settlement results, e.g., to log deposit/redemption amounts or update internal state.

Parameters:

Name	Type	Description	Default
`settlement`	`SettlementResult`	SettlementResult with deposit/redemption data	required

set_multi_chain_providers ¶

set_multi_chain_providers(
    price_oracle: MultiChainPriceOracle | None = None,
    balance_provider: MultiChainBalanceProvider
    | None = None,
    aave_health_factor_provider: AaveHealthFactorProvider
    | None = None,
) -> None

Set multi-chain data providers for cross-chain strategies.

Call this method before running a multi-chain strategy to enable MultiChainMarketSnapshot creation.

Parameters:

Name	Type	Description	Default
`price_oracle`	`MultiChainPriceOracle \| None`	Multi-chain price oracle	`None`
`balance_provider`	`MultiChainBalanceProvider \| None`	Multi-chain balance provider	`None`
`aave_health_factor_provider`	`AaveHealthFactorProvider \| None`	Aave health factor provider	`None`

is_multi_chain ¶

is_multi_chain() -> bool

Check if this strategy is multi-chain.

Returns:

Type	Description
`bool`	True if SUPPORTED_CHAINS has multiple chains

get_supported_chains ¶

get_supported_chains() -> list[str]

Get the chains supported by this strategy.

Returns:

Type	Description
`list[str]`	List of supported chain names

create_market_snapshot ¶

create_market_snapshot() -> MarketSnapshot

Create a market snapshot for the current iteration.

Automatically creates MultiChainMarketSnapshot for multi-chain strategies if multi-chain providers have been set. Otherwise returns single-chain MarketSnapshot.

Override this method to customize how market data is populated.

Returns:

Type	Description
`MarketSnapshot`	MarketSnapshot (or MultiChainMarketSnapshot for multi-chain strategies)

run ¶

run() -> ActionBundle | None

Execute one iteration of the strategy.

This method: 1. Creates a MarketSnapshot 2. Calls decide() to get an intent or DecideResult 3. Compiles single intents to an ActionBundle 4. Returns the ActionBundle for execution

Note: For multi-intent results (list or IntentSequence), this method only compiles the first intent. Use run_multi() for full multi-intent execution with proper parallel/sequential handling.

Returns:

Type	Description
`ActionBundle \| None`	ActionBundle to execute, or None if HOLD intent or no action

run_multi ¶

run_multi() -> DecideResult

Execute one iteration of the strategy, returning the full DecideResult.

Unlike run(), this method returns the full DecideResult from decide() without compiling to ActionBundle. This is useful for multi-chain execution via MultiChainOrchestrator.

Returns:

Name	Type	Description
`DecideResult`	`DecideResult`	The raw result from decide() (may be None, single intent,
	`DecideResult`	IntentSequence, or list of intents/sequences)

run_with_state_machine ¶

run_with_state_machine(
    receipt_provider: Callable[
        [ActionBundle], TransactionReceipt
    ]
    | None = None,
) -> ExecutionResult

Execute strategy with full state machine lifecycle.

This method provides full state machine execution including: - Intent compilation - Transaction execution (via receipt_provider) - Validation - Retry logic on failure

Note: This method only handles single intents for backward compatibility. For multi-intent execution, use run_multi() with MultiChainOrchestrator.

Parameters:

Name	Type	Description	Default
`receipt_provider`	`Callable[[ActionBundle], TransactionReceipt] \| None`	Function that executes an ActionBundle and returns a TransactionReceipt. If not provided, returns after compilation.	`None`

Returns:

Type	Description
`ExecutionResult`	ExecutionResult with full execution details

get_metadata ¶

get_metadata() -> StrategyMetadata | None

Get strategy metadata if available.

Returns:

Type	Description
`StrategyMetadata \| None`	StrategyMetadata if set via decorator, otherwise None

to_dict ¶

to_dict() -> dict[str, Any]

Serialize strategy state to dictionary.

Returns:

Type	Description
`dict[str, Any]`	Dictionary representation of strategy state

on_sadflow_enter ¶

on_sadflow_enter(
    error_type: str | None,
    attempt: int,
    context: SadflowContext,
) -> SadflowAction | None

Hook called when entering sadflow state.

Override this method to customize sadflow behavior for your strategy. This is called once when first entering sadflow, before any retry attempts.

Parameters:

Name	Type	Description	Default
`error_type`	`str \| None`	Categorized error type (e.g., "INSUFFICIENT_FUNDS", "TIMEOUT", "SLIPPAGE", "REVERT"). May be None for uncategorized errors.	required
`attempt`	`int`	Current attempt number (1-indexed).	required
`context`	`SadflowContext`	SadflowContext with error details and execution state.	required

Returns:

Type	Description
`SadflowAction \| None`	Optional[SadflowAction]: Action to take. Return None to use default
`SadflowAction \| None`	retry behavior. Return SadflowAction to customize:
`SadflowAction \| None`	SadflowAction.retry(): Continue with default retry
`SadflowAction \| None`	SadflowAction.abort(reason): Stop immediately and fail
`SadflowAction \| None`	SadflowAction.modify(bundle): Retry with modified ActionBundle
`SadflowAction \| None`	SadflowAction.skip(reason): Skip intent and mark as completed

Example

def on_sadflow_enter(self, error_type, attempt, context): # Abort immediately on insufficient funds if error_type == "INSUFFICIENT_FUNDS": return SadflowAction.abort("Not enough funds for transaction")

# Increase gas for gas errors
if error_type == "GAS_ERROR" and context.action_bundle:
    modified = self._increase_gas(context.action_bundle)
    return SadflowAction.modify(modified, reason="Increased gas limit")

# Use default retry for other errors
return None

supports_teardown ¶

supports_teardown() -> bool

Check if this strategy supports the teardown system.

Override to return True when your strategy implements: - get_open_positions() - generate_teardown_intents()

Returns:

Type	Description
`bool`	True if teardown is supported, False otherwise

Example

def supports_teardown(self) -> bool: return True # Enable teardown for this strategy

get_portfolio_snapshot ¶

get_portfolio_snapshot(
    market: MarketSnapshot | None = None,
) -> PortfolioSnapshot

Get current portfolio value and positions.

This method is called by the StrategyRunner after each iteration to capture portfolio snapshots for: - Dashboard value display (Total Value, PnL) - Historical PnL charts - Position breakdown by type

Default implementation: 1. Calls get_open_positions() for position values (LP, lending, perps) 2. Adds wallet token balances not captured by positions

Override for strategies needing custom value calculation (CEX, prediction).

Parameters:

Name	Type	Description	Default
`market`	`MarketSnapshot \| None`	Optional MarketSnapshot. If None, creates one internally.	`None`

Returns:

Type	Description
`PortfolioSnapshot`	PortfolioSnapshot with current values and confidence level.
`PortfolioSnapshot`	If value cannot be computed, returns snapshot with
`PortfolioSnapshot`	value_confidence=UNAVAILABLE instead of $0.

Example

def get_portfolio_snapshot(self, market=None) -> PortfolioSnapshot: if market is None: market = self.create_market_snapshot()

# Custom CEX balance fetch
cex_balance = self._fetch_cex_balance()

return PortfolioSnapshot(
    timestamp=datetime.now(UTC),
    strategy_id=self.strategy_id,
    total_value_usd=cex_balance,
    available_cash_usd=cex_balance,
    value_confidence=ValueConfidence.ESTIMATED,
    chain=self.chain,
)

get_open_positions ¶

get_open_positions() -> TeardownPositionSummary

Get all open positions for this strategy.

MUST query on-chain state - do not use cached state for safety. Called during teardown preview and execution to determine what positions need to be closed.

Returns:

Type	Description
`TeardownPositionSummary`	TeardownPositionSummary with all current positions

Raises:

Type	Description
`NotImplementedError`	If teardown is not implemented

Example

from almanak.framework.teardown import TeardownPositionSummary, PositionInfo, PositionType

def get_open_positions(self) -> TeardownPositionSummary: positions = []

# Query on-chain LP position
lp_data = self._query_lp_position()
if lp_data:
    positions.append(PositionInfo(
        position_type=PositionType.LP,
        position_id=lp_data["token_id"],
        chain=self.chain,
        protocol="uniswap_v3",
        value_usd=Decimal(str(lp_data["value_usd"])),
    ))

return TeardownPositionSummary(
    strategy_id=self.STRATEGY_NAME,
    timestamp=datetime.now(timezone.utc),
    positions=positions,
)

generate_teardown_intents ¶

generate_teardown_intents(
    mode: TeardownMode, market: MarketSnapshot | None = None
) -> list[Intent]

Generate intents to close all positions.

Return intents in the correct execution order: 1. PERP - Close perpetuals first (highest liquidation risk) 2. BORROW - Repay borrowed amounts (frees collateral) 3. SUPPLY - Withdraw supplied collateral 4. LP - Close LP positions and collect fees 5. TOKEN - Swap all tokens to target token (USDC)

Parameters:

Name	Type	Description	Default
`mode`	`TeardownMode`	TeardownMode.SOFT (graceful) or TeardownMode.HARD (emergency)	required
`market`	`MarketSnapshot \| None`	Optional market snapshot with real prices. When called from the runner, this is the same snapshot used for normal decide() iterations. May be None for backward compatibility or when called outside the runner.	`None`

Returns:

Type	Description
`list[Intent]`	List of intents to execute in order

Raises:

Type	Description
`NotImplementedError`	If teardown is not implemented

Example

from almanak.framework.teardown import TeardownMode

def generate_teardown_intents(self, mode: TeardownMode, market=None) -> list[Intent]: intents = []

# Get current positions
positions = self.get_open_positions()

# Use market data if available for smarter teardown
if market:
    eth_price = market.price("ETH")

# Close LP position first
for pos in positions.positions_by_type(PositionType.LP):
    intents.append(Intent.lp_close(
        position_id=pos.position_id,
        pool=pos.details.get("pool"),
        collect_fees=True,
        protocol="uniswap_v3",
    ))

# Swap remaining tokens to USDC
intents.append(Intent.swap(
    from_token="WETH",
    to_token="USDC",
    amount=Decimal("0"),  # All remaining
    swap_all=True,
))

return intents

on_teardown_started ¶

on_teardown_started(mode: TeardownMode) -> None

Hook called when teardown starts.

Override to perform any setup before teardown begins. This is called after the cancel window expires.

Parameters:

Name	Type	Description	Default
`mode`	`TeardownMode`	The teardown mode (SOFT or HARD)	required

Example

def on_teardown_started(self, mode: TeardownMode) -> None: logger.info(f"Teardown starting in {mode.value} mode") self._pause_monitoring()

on_teardown_completed ¶

on_teardown_completed(
    success: bool, recovered_usd: Decimal
) -> None

Hook called when teardown completes.

Override to perform cleanup after teardown.

Parameters:

Name	Type	Description	Default
`success`	`bool`	Whether all positions were closed successfully	required
`recovered_usd`	`Decimal`	Total USD value recovered	required

Example

def on_teardown_completed(self, success: bool, recovered_usd: Decimal) -> None: if success: logger.info(f"Teardown complete. Recovered ${recovered_usd:,.2f}") else: logger.error("Teardown failed - manual intervention required")

get_teardown_profile ¶

get_teardown_profile() -> TeardownProfile

Get teardown profile metadata for UX display.

Override to provide better information about teardown expectations. This helps the dashboard show more accurate previews.

Returns:

Type	Description
`TeardownProfile`	TeardownProfile with strategy-specific metadata

Example

from almanak.framework.teardown import TeardownProfile

def get_teardown_profile(self) -> TeardownProfile: return TeardownProfile( natural_exit_assets=["WETH", "USDC"], original_entry_assets=["USDC"], recommended_target="USDC", estimated_steps=3, chains_involved=[self.chain], has_lp_positions=True, )

acknowledge_teardown_request ¶

acknowledge_teardown_request() -> bool

Acknowledge a pending teardown request.

Called when the strategy picks up a teardown request and starts processing it.

Returns:

Type	Description
`bool`	True if request was acknowledged, False otherwise

should_teardown ¶

should_teardown() -> bool

Check if the strategy should enter teardown mode.

Checks for: 1. Pending teardown request (from CLI, dashboard, config) 2. Auto-protect triggers (health factor, loss limits)

Returns:

Type	Description
`bool`	True if teardown should be initiated

on_sadflow_exit ¶

on_sadflow_exit(success: bool, total_attempts: int) -> None

Hook called when exiting sadflow (on completion or final failure).

Override this method to perform cleanup or logging after sadflow resolution. This is called once when the intent completes (success or failure) after having been in sadflow.

Parameters:

Name	Type	Description	Default
`success`	`bool`	Whether the intent eventually succeeded after retries.	required
`total_attempts`	`int`	Total number of attempts made (including the final one).	required

Example

def on_sadflow_exit(self, success, total_attempts): if success: logger.info(f"Recovered after {total_attempts} attempts") else: logger.error(f"Failed after {total_attempts} attempts") self.notify_operator("Intent failed after all retries")

on_retry ¶

on_retry(
    context: SadflowContext, action: SadflowAction
) -> SadflowAction

Hook called before each retry attempt.

Override this method to customize individual retry behavior. This is called before each retry, after the initial on_sadflow_enter call.

Parameters:

Name	Type	Description	Default
`context`	`SadflowContext`	SadflowContext with current error details and state.	required
`action`	`SadflowAction`	The default SadflowAction (RETRY with calculated delay).	required

Returns:

Name	Type	Description
`SadflowAction`	`SadflowAction`	The action to take. Return the input action unchanged
	`SadflowAction`	for default behavior, or return a modified action:
	`SadflowAction`	SadflowAction.retry(custom_delay=5.0): Retry with custom delay
	`SadflowAction`	SadflowAction.abort(reason): Stop retrying and fail
	`SadflowAction`	SadflowAction.modify(bundle): Retry with modified ActionBundle
	`SadflowAction`	SadflowAction.skip(reason): Skip and mark as completed

Example

def on_retry(self, context, action): # After 2 attempts, try with higher gas if context.attempt_number > 2 and context.action_bundle: modified = self._increase_gas(context.action_bundle) return SadflowAction.modify(modified)

# Abort if we've been retrying too long
if context.total_duration_seconds > 120:
    return SadflowAction.abort("Retry timeout exceeded")

# Use default retry
return action

StrategyBase¶

Lower-level base class for strategies that need direct action control.

almanak.framework.strategies.StrategyBase ¶

StrategyBase(
    config: ConfigT,
    risk_guard_config: RiskGuardConfig | None = None,
    notification_callback: NotificationCallback
    | None = None,
)

Bases: ABC

Base class for all strategies with hot-reload configuration support.

This class provides: - Hot-reload configuration updates via update_config() - RiskGuard validation to prevent dangerous config changes - Atomic config application with rollback on failure - CONFIG_UPDATED event emission to timeline - Operator notification support

Strategies should inherit from this class and implement the abstract run() method.

Attributes:

Name	Type	Description
`config`		Hot-reloadable configuration
`risk_guard`		RiskGuard instance for validation
`persistent_state`	`dict[str, Any]`	Dict containing strategy state including config snapshots
`notification_callback`	`dict[str, Any]`	Optional callback for operator notifications

Initialize the strategy base.

Parameters:

Name	Type	Description	Default
`config`	`ConfigT`	Hot-reloadable configuration	required
`risk_guard_config`	`RiskGuardConfig \| None`	Optional RiskGuard configuration	`None`
`notification_callback`	`NotificationCallback \| None`	Optional callback for sending notifications	`None`

strategy_id `property` ¶

strategy_id: str

Get the strategy ID.

chain `property` ¶

chain: str

Get the chain.

run `abstractmethod` ¶

run() -> Any

Execute one iteration of the strategy.

Must be implemented by subclasses.

Returns:

Type	Description
`Any`	ActionBundle or None if no action needed

update_config ¶

update_config(
    updates: dict[str, Any], updated_by: str = "operator"
) -> ConfigUpdateResult

Update configuration with validation and persistence.

This method: 1. Validates new config values against the config's schema 2. Validates changes against RiskGuard (can't bypass risk limits) 3. Applies changes atomically 4. Persists config snapshot to persistent_state 5. Emits CONFIG_UPDATED event with old and new values 6. Sends notification to operator

Parameters:

Name	Type	Description	Default
`updates`	`dict[str, Any]`	Dict of field -> new value to update	required
`updated_by`	`str`	Who is making the update (for audit trail)	`'operator'`

Returns:

Type	Description
`ConfigUpdateResult`	ConfigUpdateResult indicating success or failure

set_notification_callback ¶

set_notification_callback(
    callback: NotificationCallback | None,
) -> None

Set the notification callback for operator alerts.

Parameters:

Name	Type	Description	Default
`callback`	`NotificationCallback \| None`	Callback function that takes an OperatorCard	required

get_config_history ¶

get_config_history() -> list[dict[str, Any]]

Get the configuration update history.

Returns:

Type	Description
`list[dict[str, Any]]`	List of config snapshots in chronological order

get_current_config_version ¶

get_current_config_version() -> int

Get the current config version number.

Returns:

Type	Description
`int`	Current config version

get_config ¶

get_config(key: str, default: Any = None) -> Any

Get a configuration value by key, regardless of config type.

Works transparently with: - Plain dict configs (from tests or raw JSON) - DictConfigWrapper objects (from the CLI) - Dataclasses and Pydantic models (attribute access)

This eliminates the per-strategy boilerplate of checking isinstance(self.config, dict) before every config.get() call.

Parameters:

Name	Type	Description	Default
`key`	`str`	Configuration key to retrieve	required
`default`	`Any`	Value to return if the key is not found	`None`

Returns:

Type	Description
`Any`	The configuration value, or `default` if not found

Example::

# In decide() -- no more boilerplate!
trade_size = self.get_config("trade_size_usd", "100")
rsi_period = self.get_config("rsi_period", 14)

restore_config_from_snapshot ¶

restore_config_from_snapshot(
    version: int,
) -> ConfigUpdateResult

Restore configuration from a previous snapshot.

Parameters:

Name	Type	Description	Default
`version`	`int`	The config version to restore to	required

Returns:

Type	Description
`ConfigUpdateResult`	ConfigUpdateResult indicating success or failure

MarketSnapshot¶

Unified interface for accessing market data within decide().

almanak.framework.strategies.MarketSnapshot ¶

MarketSnapshot(
    chain: str,
    wallet_address: str,
    price_oracle: PriceOracle | None = None,
    rsi_provider: RSIProvider | None = None,
    balance_provider: BalanceProvider | None = None,
    timestamp: datetime | None = None,
    wallet_activity_provider: WalletActivityProvider
    | None = None,
    prediction_provider: Any | None = None,
    indicator_provider: IndicatorProvider | None = None,
)

Helper class providing market data access for strategy decisions.

MarketSnapshot provides a simple interface for strategies to access: - Token prices - RSI values - Wallet balances - Position information

The snapshot is populated with data at the start of each iteration, allowing strategies to make decisions based on current market conditions.

Example

def decide(self, market: MarketSnapshot) -> Optional[Intent]: # Get ETH price eth_price = market.price("ETH")

# Get RSI
rsi = market.rsi("ETH", period=14)
if rsi.is_oversold:
    return Intent.swap("USDC", "ETH", amount_usd=Decimal("1000"))

# Check balance
balance = market.balance("USDC")
if balance.balance_usd < Decimal("100"):
    return Intent.hold(reason="Insufficient balance")

return Intent.hold()

Initialize market snapshot.

Parameters:

Name	Type	Description	Default
`chain`	`str`	Chain name (e.g., "arbitrum", "ethereum")	required
`wallet_address`	`str`	Wallet address for balance queries	required
`price_oracle`	`PriceOracle \| None`	Function to fetch prices (token, quote) -> price	`None`
`rsi_provider`	`RSIProvider \| None`	Function to calculate RSI (token, period[, timeframe=]) -> RSIData	`None`
`balance_provider`	`BalanceProvider \| None`	Function to fetch balances (token) -> TokenBalance	`None`
`timestamp`	`datetime \| None`	Snapshot timestamp (defaults to now)	`None`
`wallet_activity_provider`	`WalletActivityProvider \| None`	Provider for leader wallet activity signals	`None`
`prediction_provider`	`Any \| None`	PredictionMarketDataProvider for prediction market data	`None`
`indicator_provider`	`IndicatorProvider \| None`	IndicatorProvider for calculator-backed TA indicators	`None`

chain `property` ¶

chain: str

Get the chain name.

wallet_address `property` ¶

wallet_address: str

Get the wallet address.

timestamp `property` ¶

timestamp: datetime

Get the snapshot timestamp.

price ¶

price(token: str, quote: str = 'USD') -> Decimal

Get the price of a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol (e.g., "ETH", "WBTC")	required
`quote`	`str`	Quote currency (default "USD")	`'USD'`

Returns:

Type	Description
`Decimal`	Token price in quote currency

Raises:

Type	Description
`ValueError`	If price cannot be determined

price_data ¶

price_data(token: str, quote: str = 'USD') -> PriceData

Get full price data for a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required
`quote`	`str`	Quote currency (default "USD")	`'USD'`

Returns:

Type	Description
`PriceData`	PriceData with current price and historical data

rsi ¶

rsi(
    token: str, period: int = 14, timeframe: str = "4h"
) -> RSIData

Get RSI (Relative Strength Index) for a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required
`period`	`int`	RSI calculation period (default 14)	`14`
`timeframe`	`str`	OHLCV candle timeframe (default "4h")	`'4h'`

Returns:

Type	Description
`RSIData`	RSIData with current RSI value and signal

Raises:

Type	Description
`ValueError`	If RSI cannot be calculated

macd ¶

macd(
    token: str,
    fast_period: int = 12,
    slow_period: int = 26,
    signal_period: int = 9,
    timeframe: str = "4h",
) -> MACDData

Get MACD (Moving Average Convergence Divergence) for a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required
`fast_period`	`int`	Fast EMA period (default 12)	`12`
`slow_period`	`int`	Slow EMA period (default 26)	`26`
`signal_period`	`int`	Signal EMA period (default 9)	`9`
`timeframe`	`str`	OHLCV candle timeframe (default "4h")	`'4h'`

Returns:

Type	Description
`MACDData`	MACDData with MACD line, signal line, and histogram

Raises:

Type	Description
`ValueError`	If MACD data is not available

Example

macd = market.macd("WETH") if macd.is_bullish_crossover: return Intent.swap("USDC", "WETH", amount_usd=Decimal("100"))

bollinger_bands ¶

bollinger_bands(
    token: str,
    period: int = 20,
    std_dev: float = 2.0,
    timeframe: str = "4h",
) -> BollingerBandsData

Get Bollinger Bands for a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required
`period`	`int`	SMA period (default 20)	`20`
`std_dev`	`float`	Standard deviation multiplier (default 2.0)	`2.0`
`timeframe`	`str`	OHLCV candle timeframe (default "4h")	`'4h'`

Returns:

Type	Description
`BollingerBandsData`	BollingerBandsData with upper, middle, lower bands and position metrics

Raises:

Type	Description
`ValueError`	If Bollinger Bands data is not available

Example

bb = market.bollinger_bands("WETH") if bb.is_oversold: return Intent.swap("USDC", "WETH", amount_usd=Decimal("100"))

stochastic ¶

stochastic(
    token: str,
    k_period: int = 14,
    d_period: int = 3,
    timeframe: str = "4h",
) -> StochasticData

Get Stochastic Oscillator for a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required
`k_period`	`int`	%K period (default 14)	`14`
`d_period`	`int`	%D period (default 3)	`3`
`timeframe`	`str`	OHLCV candle timeframe (default "4h")	`'4h'`

Returns:

Type	Description
`StochasticData`	StochasticData with %K and %D values

Raises:

Type	Description
`ValueError`	If Stochastic data is not available

Example

stoch = market.stochastic("WETH") if stoch.is_oversold and stoch.k_value > stoch.d_value: return Intent.swap("USDC", "WETH", amount_usd=Decimal("100"))

atr ¶

atr(
    token: str, period: int = 14, timeframe: str = "4h"
) -> ATRData

Get ATR (Average True Range) for a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required
`period`	`int`	ATR period (default 14)	`14`
`timeframe`	`str`	OHLCV candle timeframe (default "4h")	`'4h'`

Returns:

Type	Description
`ATRData`	ATRData with ATR value and volatility assessment

Raises:

Type	Description
`ValueError`	If ATR data is not available

Example

atr = market.atr("WETH") if atr.is_low_volatility: # Safe to trade return Intent.swap("USDC", "WETH", amount_usd=Decimal("100"))

sma ¶

sma(
    token: str, period: int = 20, timeframe: str = "4h"
) -> MAData

Get Simple Moving Average for a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required
`period`	`int`	SMA period (default 20)	`20`
`timeframe`	`str`	OHLCV candle timeframe (default "4h")	`'4h'`

Returns:

Type	Description
`MAData`	MAData with SMA value

Raises:

Type	Description
`ValueError`	If SMA data is not available

Example

sma = market.sma("WETH", period=50) if sma.is_price_above: print("Bullish - price above 50 SMA")

ema ¶

ema(
    token: str, period: int = 12, timeframe: str = "4h"
) -> MAData

Get Exponential Moving Average for a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required
`period`	`int`	EMA period (default 12)	`12`
`timeframe`	`str`	OHLCV candle timeframe (default "4h")	`'4h'`

Returns:

Type	Description
`MAData`	MAData with EMA value

Raises:

Type	Description
`ValueError`	If EMA data is not available

Example

ema_12 = market.ema("WETH", period=12) ema_26 = market.ema("WETH", period=26) if ema_12.value > ema_26.value: print("Golden cross - bullish")

adx ¶

adx(
    token: str, period: int = 14, timeframe: str = "4h"
) -> ADXData

Get ADX (Average Directional Index) for a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required
`period`	`int`	ADX period (default 14)	`14`
`timeframe`	`str`	OHLCV candle timeframe (default "4h")	`'4h'`

Returns:

Type	Description
`ADXData`	ADXData with ADX, +DI, and -DI values

Raises:

Type	Description
`ValueError`	If ADX data is not available

Example

adx = market.adx("WETH") if adx.is_uptrend: return Intent.swap("USDC", "WETH", amount_usd=Decimal("100"))

obv ¶

obv(
    token: str,
    signal_period: int = 21,
    timeframe: str = "4h",
) -> OBVData

Get OBV (On-Balance Volume) for a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required
`signal_period`	`int`	OBV signal line period (default 21)	`21`
`timeframe`	`str`	OHLCV candle timeframe (default "4h")	`'4h'`

Returns:

Type	Description
`OBVData`	OBVData with OBV and signal line values

Raises:

Type	Description
`ValueError`	If OBV data is not available

Example

obv = market.obv("WETH") if obv.is_bullish: return Intent.swap("USDC", "WETH", amount_usd=Decimal("100"))

cci ¶

cci(
    token: str, period: int = 20, timeframe: str = "4h"
) -> CCIData

Get CCI (Commodity Channel Index) for a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required
`period`	`int`	CCI period (default 20)	`20`
`timeframe`	`str`	OHLCV candle timeframe (default "4h")	`'4h'`

Returns:

Type	Description
`CCIData`	CCIData with CCI value and overbought/oversold status

Raises:

Type	Description
`ValueError`	If CCI data is not available

Example

cci = market.cci("WETH") if cci.is_oversold: return Intent.swap("USDC", "WETH", amount_usd=Decimal("100"))

ichimoku ¶

ichimoku(
    token: str,
    tenkan_period: int = 9,
    kijun_period: int = 26,
    senkou_b_period: int = 52,
    timeframe: str = "4h",
) -> IchimokuData

Get Ichimoku Cloud data for a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required
`tenkan_period`	`int`	Conversion line period (default 9)	`9`
`kijun_period`	`int`	Base line period (default 26)	`26`
`senkou_b_period`	`int`	Leading span B period (default 52)	`52`
`timeframe`	`str`	OHLCV candle timeframe (default "4h")	`'4h'`

Returns:

Type	Description
`IchimokuData`	IchimokuData with all Ichimoku components

Raises:

Type	Description
`ValueError`	If Ichimoku data is not available

Example

ich = market.ichimoku("WETH") if ich.is_bullish_crossover and ich.is_above_cloud: return Intent.swap("USDC", "WETH", amount_usd=Decimal("100"))

balance ¶

balance(token: str) -> TokenBalance

Get wallet balance for a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required

Returns:

Type	Description
`TokenBalance`	TokenBalance with current balance

Raises:

Type	Description
`ValueError`	If balance cannot be determined

balance_usd ¶

balance_usd(token: str) -> Decimal

Get wallet balance in USD terms.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required

Returns:

Type	Description
`Decimal`	Balance in USD

total_portfolio_usd ¶

total_portfolio_usd() -> Decimal

Calculate total portfolio value in USD across all known balances.

Sums balance_usd for all tokens in pre-populated balances and cached balances (tokens queried via balance() in this snapshot).

Returns:

Type	Description
`Decimal`	Total portfolio value in USD

set_price ¶

set_price(token: str, price_value: Decimal) -> None

Pre-populate price for a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required
`price_value`	`Decimal`	Price value in USD	required

set_price_data ¶

set_price_data(
    token: str, price_data: PriceData, quote: str = "USD"
) -> None

Pre-populate enriched price data for a token (useful for testing).

Unlike set_price() which only sets a scalar price, this sets the full PriceData object including change_24h_pct, high_24h, low_24h, etc.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required
`price_data`	`PriceData`	PriceData with price, change_24h_pct, etc.	required
`quote`	`str`	Quote currency (default "USD")	`'USD'`

set_balance ¶

set_balance(token: str, balance_data: TokenBalance) -> None

Pre-populate balance for a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required
`balance_data`	`TokenBalance`	Balance data	required

set_rsi ¶

set_rsi(
    token: str,
    rsi_data: RSIData,
    timeframe: str | None = None,
) -> None

Pre-populate RSI for a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required
`rsi_data`	`RSIData`	RSI data	required
`timeframe`	`str \| None`	OHLCV timeframe this data was computed from (None matches any)	`None`

set_macd ¶

set_macd(
    token: str,
    macd_data: MACDData,
    timeframe: str | None = None,
) -> None

Pre-populate MACD data for a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required
`macd_data`	`MACDData`	MACDData instance	required
`timeframe`	`str \| None`	OHLCV timeframe this data was computed from (None matches any)	`None`

Example

market.set_macd("WETH", MACDData( macd_line=Decimal("0.5"), signal_line=Decimal("0.3"), histogram=Decimal("0.2"), ))

set_bollinger_bands ¶

set_bollinger_bands(
    token: str,
    bb_data: BollingerBandsData,
    timeframe: str | None = None,
) -> None

Pre-populate Bollinger Bands data for a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required
`bb_data`	`BollingerBandsData`	BollingerBandsData instance	required
`timeframe`	`str \| None`	OHLCV timeframe this data was computed from (None matches any)	`None`

Example

market.set_bollinger_bands("WETH", BollingerBandsData( upper_band=Decimal("3100"), middle_band=Decimal("3000"), lower_band=Decimal("2900"), percent_b=Decimal("0.5"), ))

set_stochastic ¶

set_stochastic(
    token: str,
    stoch_data: StochasticData,
    timeframe: str | None = None,
) -> None

Pre-populate Stochastic data for a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required
`stoch_data`	`StochasticData`	StochasticData instance	required
`timeframe`	`str \| None`	OHLCV timeframe this data was computed from (None matches any)	`None`

Example

market.set_stochastic("WETH", StochasticData( k_value=Decimal("25"), d_value=Decimal("30"), ))

set_atr ¶

set_atr(
    token: str,
    atr_data: ATRData,
    timeframe: str | None = None,
) -> None

Pre-populate ATR data for a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required
`atr_data`	`ATRData`	ATRData instance	required
`timeframe`	`str \| None`	OHLCV timeframe this data was computed from (None matches any)	`None`

Example

market.set_atr("WETH", ATRData( value=Decimal("50"), value_percent=Decimal("2.5"), ))

set_ma ¶

set_ma(
    token: str,
    ma_data: MAData,
    ma_type: str = "SMA",
    period: int = 20,
    timeframe: str | None = None,
) -> None

Pre-populate Moving Average data for a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required
`ma_data`	`MAData`	MAData instance	required
`ma_type`	`str`	Type of MA ("SMA" or "EMA")	`'SMA'`
`period`	`int`	MA period	`20`
`timeframe`	`str \| None`	OHLCV timeframe this data was computed from (None matches any)	`None`

Example

market.set_ma("WETH", MAData( value=Decimal("3000"), ma_type="SMA", period=20, current_price=Decimal("3050"), ), ma_type="SMA", period=20)

set_adx ¶

set_adx(
    token: str,
    adx_data: ADXData,
    timeframe: str | None = None,
) -> None

Pre-populate ADX data for a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required
`adx_data`	`ADXData`	ADXData instance	required
`timeframe`	`str \| None`	Optional timeframe (None matches any query)	`None`

Example

market.set_adx("WETH", ADXData( adx=Decimal("30"), plus_di=Decimal("25"), minus_di=Decimal("15"), ))

set_obv ¶

set_obv(
    token: str,
    obv_data: OBVData,
    timeframe: str | None = None,
) -> None

Pre-populate OBV data for a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required
`obv_data`	`OBVData`	OBVData instance	required
`timeframe`	`str \| None`	Optional timeframe (None matches any query)	`None`

Example

market.set_obv("WETH", OBVData( obv=Decimal("1000000"), signal_line=Decimal("950000"), ))

set_cci ¶

set_cci(
    token: str,
    cci_data: CCIData,
    timeframe: str | None = None,
) -> None

Pre-populate CCI data for a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required
`cci_data`	`CCIData`	CCIData instance	required
`timeframe`	`str \| None`	Optional timeframe (None matches any query)	`None`

Example

market.set_cci("WETH", CCIData( value=Decimal("-120"), ))

set_ichimoku ¶

set_ichimoku(
    token: str,
    ichimoku_data: IchimokuData,
    timeframe: str | None = None,
) -> None

Pre-populate Ichimoku data for a token.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol	required
`ichimoku_data`	`IchimokuData`	IchimokuData instance	required
`timeframe`	`str \| None`	Optional timeframe (None matches any query)	`None`

Example

market.set_ichimoku("WETH", IchimokuData( tenkan_sen=Decimal("3050"), kijun_sen=Decimal("3000"), senkou_span_a=Decimal("3025"), senkou_span_b=Decimal("2950"), current_price=Decimal("3100"), ))

wallet_activity ¶

wallet_activity(
    leader_address: str | None = None,
    action_types: list[str] | None = None,
    min_usd_value: Decimal | None = None,
    protocols: list[str] | None = None,
) -> list

Get leader wallet activity signals for copy trading.

Returns filtered signals from the WalletActivityProvider. If no provider is configured, returns an empty list (graceful degradation).

Parameters:

Name	Type	Description	Default
`leader_address`	`str \| None`	Filter by specific leader wallet address	`None`
`action_types`	`list[str] \| None`	Filter by action types (e.g., ["SWAP"])	`None`
`min_usd_value`	`Decimal \| None`	Minimum USD value filter	`None`
`protocols`	`list[str] \| None`	Filter by protocol names (e.g., ["uniswap_v3"])	`None`

Returns:

Type	Description
`list`	List of CopySignal objects matching the filters

prediction_price ¶

prediction_price(
    market_id: str, outcome: str
) -> Decimal | None

Get current price for a prediction market outcome.

Convenience method that extracts the YES or NO price from a market.

Parameters:

Name	Type	Description	Default
`market_id`	`str`	Prediction market ID or URL slug	required
`outcome`	`str`	"YES" or "NO"	required

Returns:

Type	Description
`Decimal \| None`	Current price as Decimal (0.01 to 0.99), or None if unavailable

Example

yes_price = market.prediction_price("btc-100k", "YES") if yes_price is not None and yes_price < Decimal("0.3"): return BuyIntent(...)

get_price_oracle_dict ¶

get_price_oracle_dict() -> dict[str, Decimal]

Get all prices as a dict suitable for IntentCompiler.

Combines pre-populated prices and cached prices from oracle calls. This is used to pass real prices to the IntentCompiler for accurate slippage calculations.

Returns:

Type	Description
`dict[str, Decimal]`	Dict mapping token symbols to USD prices

to_dict ¶

to_dict() -> dict[str, Any]

Convert snapshot to dictionary.

RiskGuard¶

Non-bypassable risk validation that runs before every execution.

almanak.framework.strategies.RiskGuard ¶

RiskGuard(config: RiskGuardConfig | None = None)

Validates configuration changes against risk limits.

RiskGuard ensures that configuration updates cannot bypass critical risk limits defined by the operator or system. It provides human-readable guidance when validation fails.

Initialize RiskGuard with configuration.

Parameters:

Name	Type	Description	Default
`config`	`RiskGuardConfig \| None`	Risk guard configuration (uses defaults if not provided)	`None`

generate_guidance ¶

generate_guidance(
    field_name: str,
    requested_value: Decimal,
    limit_value: Decimal,
) -> RiskGuardGuidance

Generate human-readable guidance for a failed risk check.

This method creates a RiskGuardGuidance object with: - What the limit is and what value was requested - Clear explanation of what the limit protects against - Actionable suggestion for how to proceed

Parameters:

Name	Type	Description	Default
`field_name`	`str`	The configuration field that failed validation	required
`requested_value`	`Decimal`	The value the operator tried to set	required
`limit_value`	`Decimal`	The maximum/minimum allowed value	required

Returns:

Type	Description
`RiskGuardGuidance`	RiskGuardGuidance with human-readable explanation and suggestion

validate_config_update ¶

validate_config_update(
    current_config: HotReloadableConfig,
    updates: dict[str, Any],
) -> RiskGuardResult

Validate proposed configuration updates against risk limits.

This method checks each proposed update against the risk guard's limits to ensure operators cannot accidentally configure dangerous parameters. When validation fails, it includes human-readable guidance explaining what went wrong and how to fix it.

Parameters:

Name	Type	Description	Default
`current_config`	`HotReloadableConfig`	Current configuration	required
`updates`	`dict[str, Any]`	Dict of field -> new value proposed updates	required

Returns:

Type	Description
`RiskGuardResult`	RiskGuardResult indicating if validation passed, with guidance if not

RiskGuardConfig¶

almanak.framework.strategies.RiskGuardConfig `dataclass` ¶

RiskGuardConfig(
    max_slippage_limit: Decimal = Decimal("0.1"),
    max_leverage_limit: Decimal = Decimal("10"),
    max_daily_loss_limit_usd: Decimal = Decimal("100000"),
    min_health_factor_floor: Decimal = Decimal("1.05"),
)

Configuration for RiskGuard validation.

Defines the maximum allowed values for risk-sensitive parameters. These limits cannot be exceeded through hot-reload updates.

Attributes:

Name	Type	Description
`max_slippage_limit`	`Decimal`	Maximum allowed slippage (e.g., 0.05 = 5%)
`max_leverage_limit`	`Decimal`	Maximum allowed leverage (e.g., 10x)
`max_daily_loss_limit_usd`	`Decimal`	Maximum allowed daily loss limit
`min_health_factor_floor`	`Decimal`	Minimum allowed health factor setting

DecideResult¶

almanak.framework.strategies.DecideResult `module-attribute` ¶

DecideResult = (
    AnyIntent
    | IntentSequence
    | list[AnyIntent | IntentSequence]
    | None
)

IntentSequence¶

almanak.framework.strategies.IntentSequence `dataclass` ¶

IntentSequence(
    intents: list[AnyIntent],
    sequence_id: str = (lambda: str(uuid.uuid4()))(),
    created_at: datetime = (lambda: datetime.now(UTC))(),
    description: str | None = None,
)

A sequence of intents that must execute in order (dependent actions).

IntentSequence wraps a list of intents that have dependencies between them and must execute sequentially. This is used when the output of one intent feeds into the input of the next (e.g., swap output -> bridge input).

Intents that are NOT in a sequence can execute in parallel if they are independent (e.g., two swaps on different chains).

Attributes:

Name	Type	Description
`intents`	`list[AnyIntent]`	List of intents to execute in order
`sequence_id`	`str`	Unique identifier for this sequence
`created_at`	`datetime`	Timestamp when the sequence was created
`description`	`str \| None`	Optional description of the sequence purpose

Example

Create a sequence of dependent actions¶

sequence = Intent.sequence([ Intent.swap("USDC", "ETH", amount=Decimal("1000"), chain="base"), Intent.bridge(token="ETH", amount="all", from_chain="base", to_chain="arbitrum"), Intent.supply(protocol="aave_v3", token="WETH", amount="all", chain="arbitrum"), ])

Return from decide() - will execute sequentially¶

return sequence

first `property` ¶

first: AnyIntent

Get the first intent in the sequence.

last `property` ¶

last: AnyIntent

Get the last intent in the sequence.

__post_init__ ¶

__post_init__() -> None

Validate the sequence.

len ¶

__len__() -> int

Return the number of intents in the sequence.

iter ¶

__iter__()

Iterate over intents in the sequence.

getitem ¶

__getitem__(index: int) -> AnyIntent

Get intent at index.

serialize ¶

serialize() -> dict[str, Any]

Serialize the sequence to a dictionary.

deserialize `classmethod` ¶

deserialize(data: dict[str, Any]) -> IntentSequence

Deserialize a dictionary to an IntentSequence.

Note: This requires the Intent.deserialize function to be available, which creates a circular dependency. The actual deserialization is done in the Intent class.

ExecutionResult¶

almanak.framework.strategies.ExecutionResult `dataclass` ¶

ExecutionResult(
    intent: AnyIntent | None,
    action_bundle: ActionBundle | None = None,
    state_machine_result: StepResult | None = None,
    success: bool = False,
    error: str | None = None,
    execution_time_ms: float = 0.0,
)

Result of strategy execution.

Attributes:

Name	Type	Description
`intent`	`AnyIntent \| None`	The intent that was executed (or None if HOLD)
`action_bundle`	`ActionBundle \| None`	The compiled action bundle (or None)
`state_machine_result`	`StepResult \| None`	Final state machine step result
`success`	`bool`	Whether execution was successful
`error`	`str \| None`	Error message if failed
`execution_time_ms`	`float`	Time taken for execution in milliseconds

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

Market Data Types¶

Data types returned by MarketSnapshot getters and accepted by set_* methods for unit testing.

TokenBalance¶

almanak.framework.strategies.TokenBalance `dataclass` ¶

TokenBalance(
    symbol: str,
    balance: Decimal,
    balance_usd: Decimal,
    address: str = "",
)

Balance information for a single token.

Supports numeric comparisons so strategy authors can write

if market.balance("ETH") > Decimal("1.0"): ... amount = min(trade_size, market.balance("USDC"))

Comparisons delegate to the balance field (native units).

Attributes:

Name	Type	Description
`symbol`	`str`	Token symbol (e.g., "ETH", "USDC")
`balance`	`Decimal`	Token balance in native units
`balance_usd`	`Decimal`	Token balance in USD terms
`address`	`str`	Token contract address

PriceData¶

almanak.framework.strategies.PriceData `dataclass` ¶

PriceData(
    price: Decimal,
    price_24h_ago: Decimal = Decimal("0"),
    change_24h_pct: Decimal = Decimal("0"),
    high_24h: Decimal = Decimal("0"),
    low_24h: Decimal = Decimal("0"),
    timestamp: datetime = (lambda: datetime.now(UTC))(),
)

Price data for a token.

Attributes:

Name	Type	Description
`price`	`Decimal`	Current price in USD
`price_24h_ago`	`Decimal`	Price 24 hours ago in USD
`change_24h_pct`	`Decimal`	24-hour price change percentage
`high_24h`	`Decimal`	24-hour high
`low_24h`	`Decimal`	24-hour low
`timestamp`	`datetime`	When the price was fetched

RSIData¶

almanak.framework.strategies.RSIData `dataclass` ¶

RSIData(
    value: Decimal,
    period: int = 14,
    overbought: Decimal = Decimal("70"),
    oversold: Decimal = Decimal("30"),
)

RSI (Relative Strength Index) data for a token.

Supports numeric operations so strategy authors can write

rsi = market.rsi("ETH") if rsi > 70: ... # comparison against int/float f"{rsi:.2f}" # f-string formatting float(rsi) # explicit float conversion

Attributes:

Name	Type	Description
`value`	`Decimal`	Current RSI value (0-100)
`period`	`int`	RSI period used (e.g., 14)
`overbought`	`Decimal`	Overbought threshold (default 70)
`oversold`	`Decimal`	Oversold threshold (default 30)
`signal`	`str`	Signal based on RSI (BUY, SELL, or HOLD)

signal `property` ¶

signal: str

Get signal based on RSI value.

is_oversold `property` ¶

is_oversold: bool

Check if RSI indicates oversold condition.

is_overbought `property` ¶

is_overbought: bool

Check if RSI indicates overbought condition.

MACDData¶

almanak.framework.strategies.MACDData `dataclass` ¶

MACDData(
    macd_line: Decimal,
    signal_line: Decimal,
    histogram: Decimal,
    fast_period: int = 12,
    slow_period: int = 26,
    signal_period: int = 9,
)

MACD (Moving Average Convergence Divergence) data for a token.

Attributes:

Name	Type	Description
`macd_line`	`Decimal`	MACD line value (fast EMA - slow EMA)
`signal_line`	`Decimal`	Signal line value (EMA of MACD line)
`histogram`	`Decimal`	MACD histogram (MACD line - signal line)
`fast_period`	`int`	Fast EMA period (default 12)
`slow_period`	`int`	Slow EMA period (default 26)
`signal_period`	`int`	Signal EMA period (default 9)

is_bullish_crossover `property` ¶

is_bullish_crossover: bool

Check if MACD line is above signal line (bullish).

is_bearish_crossover `property` ¶

is_bearish_crossover: bool

Check if MACD line is below signal line (bearish).

signal `property` ¶

signal: str

Get signal based on MACD histogram.

BollingerBandsData¶

almanak.framework.strategies.BollingerBandsData `dataclass` ¶

BollingerBandsData(
    upper_band: Decimal,
    middle_band: Decimal,
    lower_band: Decimal,
    bandwidth: Decimal = Decimal("0"),
    percent_b: Decimal = Decimal("0.5"),
    period: int = 20,
    std_dev: float = 2.0,
)

Bollinger Bands data for a token.

Numeric operations use percent_b (price position within bands, 0-1): bb = market.bollinger_bands("ETH") f"{bb:.2f}" # formats percent_b float(bb) # returns percent_b as float

Attributes:

Name	Type	Description
`upper_band`	`Decimal`	Upper band value (middle + std_dev * multiplier)
`middle_band`	`Decimal`	Middle band value (SMA)
`lower_band`	`Decimal`	Lower band value (middle - std_dev * multiplier)
`bandwidth`	`Decimal`	Band width as percentage ((upper - lower) / middle)
`percent_b`	`Decimal`	Price position relative to bands (0 = lower, 1 = upper)
`period`	`int`	SMA period (default 20)
`std_dev`	`float`	Standard deviation multiplier (default 2.0)

is_oversold `property` ¶

is_oversold: bool

Check if price is below lower band (oversold).

is_overbought `property` ¶

is_overbought: bool

Check if price is above upper band (overbought).

is_squeeze `property` ¶

is_squeeze: bool

Check if bands are tight (low volatility squeeze).

signal `property` ¶

signal: str

Get signal based on Bollinger Bands position.

StochasticData¶

almanak.framework.strategies.StochasticData `dataclass` ¶

StochasticData(
    k_value: Decimal,
    d_value: Decimal,
    k_period: int = 14,
    d_period: int = 3,
    overbought: Decimal = Decimal("80"),
    oversold: Decimal = Decimal("20"),
)

Stochastic Oscillator data for a token.

Attributes:

Name	Type	Description
`k_value`	`Decimal`	%K value (fast stochastic, 0-100)
`d_value`	`Decimal`	%D value (slow stochastic, SMA of %K, 0-100)
`k_period`	`int`	%K period (default 14)
`d_period`	`int`	%D period (default 3)
`overbought`	`Decimal`	Overbought threshold (default 80)
`oversold`	`Decimal`	Oversold threshold (default 20)

is_oversold `property` ¶

is_oversold: bool

Check if stochastic indicates oversold condition.

is_overbought `property` ¶

is_overbought: bool

Check if stochastic indicates overbought condition.

is_bullish_crossover `property` ¶

is_bullish_crossover: bool

Check if %K crossed above %D (bullish signal).

is_bearish_crossover `property` ¶

is_bearish_crossover: bool

Check if %K crossed below %D (bearish signal).

signal `property` ¶

signal: str

Get signal based on Stochastic values.

ATRData¶

almanak.framework.strategies.ATRData `dataclass` ¶

ATRData(
    value: Decimal,
    value_percent: Decimal = Decimal("0"),
    period: int = 14,
    volatility_threshold: Decimal = Decimal("5.0"),
)

ATR (Average True Range) data for a token.

Numeric operations use value (ATR in price units): atr = market.atr("ETH") f"{atr:.2f}" # formats ATR value float(atr) # returns ATR value as float

Attributes:

Name	Type	Description
`value`	`Decimal`	ATR value in price units
`value_percent`	`Decimal`	ATR as percentage points of current price (e.g., 2.62 means 2.62%, not 0.0262)
`period`	`int`	ATR period (default 14)
`volatility_threshold`	`Decimal`	Max volatility threshold in percentage points (e.g., 5.0 means 5.0%)

is_high_volatility `property` ¶

is_high_volatility: bool

Check if volatility is above threshold (risky to trade).

is_low_volatility `property` ¶

is_low_volatility: bool

Check if volatility is below threshold (safe to trade).

signal `property` ¶

signal: str

Get signal based on ATR volatility gate.

MAData¶

almanak.framework.strategies.MAData `dataclass` ¶

MAData(
    value: Decimal,
    ma_type: str = "SMA",
    period: int = 20,
    current_price: Decimal = Decimal("0"),
)

Moving Average data for a token.

Attributes:

Name	Type	Description
`value`	`Decimal`	Moving average value
`ma_type`	`str`	Type of moving average ("SMA", "EMA", "WMA")
`period`	`int`	MA period
`current_price`	`Decimal`	Current price for comparison

is_price_above `property` ¶

is_price_above: bool

Check if current price is above the MA.

is_price_below `property` ¶

is_price_below: bool

Check if current price is below the MA.

signal `property` ¶

signal: str

Get signal based on price vs MA.

ADXData¶

almanak.framework.strategies.ADXData `dataclass` ¶

ADXData(
    adx: Decimal,
    plus_di: Decimal,
    minus_di: Decimal,
    period: int = 14,
    trend_threshold: Decimal = Decimal("25"),
)

ADX (Average Directional Index) data for a token.

Attributes:

Name	Type	Description
`adx`	`Decimal`	ADX value (0-100, measures trend strength)
`plus_di`	`Decimal`	+DI value (positive directional indicator)
`minus_di`	`Decimal`	-DI value (negative directional indicator)
`period`	`int`	ADX period (default 14)
`trend_threshold`	`Decimal`	Threshold for strong trend (default 25)

is_strong_trend `property` ¶

is_strong_trend: bool

Check if ADX indicates a strong trend.

is_uptrend `property` ¶

is_uptrend: bool

Check if in uptrend (+DI > -DI with strong trend).

is_downtrend `property` ¶

is_downtrend: bool

Check if in downtrend (-DI > +DI with strong trend).

signal `property` ¶

signal: str

Get signal based on ADX trend analysis.

OBVData¶

almanak.framework.strategies.OBVData `dataclass` ¶

OBVData(
    obv: Decimal,
    signal_line: Decimal,
    signal_period: int = 21,
)

OBV (On-Balance Volume) data for a token.

Attributes:

Name	Type	Description
`obv`	`Decimal`	Current OBV value
`signal_line`	`Decimal`	Signal line (SMA of OBV)
`signal_period`	`int`	Signal period (default 21)

is_bullish `property` ¶

is_bullish: bool

Check if OBV is above signal (buying pressure).

is_bearish `property` ¶

is_bearish: bool

Check if OBV is below signal (selling pressure).

signal `property` ¶

signal: str

Get signal based on OBV analysis.

CCIData¶

almanak.framework.strategies.CCIData `dataclass` ¶

CCIData(
    value: Decimal,
    period: int = 20,
    upper_level: Decimal = Decimal("100"),
    lower_level: Decimal = Decimal("-100"),
)

CCI (Commodity Channel Index) data for a token.

Attributes:

Name	Type	Description
`value`	`Decimal`	CCI value
`period`	`int`	CCI period (default 20)
`upper_level`	`Decimal`	Overbought level (default 100)
`lower_level`	`Decimal`	Oversold level (default -100)

is_oversold `property` ¶

is_oversold: bool

Check if CCI indicates oversold condition.

is_overbought `property` ¶

is_overbought: bool

Check if CCI indicates overbought condition.

signal `property` ¶

signal: str

Get signal based on CCI.

IchimokuData¶

almanak.framework.strategies.IchimokuData `dataclass` ¶

IchimokuData(
    tenkan_sen: Decimal,
    kijun_sen: Decimal,
    senkou_span_a: Decimal,
    senkou_span_b: Decimal,
    chikou_span: Decimal = Decimal("0"),
    current_price: Decimal = Decimal("0"),
    tenkan_period: int = 9,
    kijun_period: int = 26,
    senkou_b_period: int = 52,
)

Ichimoku Cloud data for a token.

Attributes:

Name	Type	Description
`tenkan_sen`	`Decimal`	Conversion line (9-period midpoint)
`kijun_sen`	`Decimal`	Base line (26-period midpoint)
`senkou_span_a`	`Decimal`	Leading span A
`senkou_span_b`	`Decimal`	Leading span B
`chikou_span`	`Decimal`	Lagging span (current close plotted 26 periods back)
`current_price`	`Decimal`	Current price for cloud position check
`tenkan_period`	`int`	Tenkan-sen period (default 9)
`kijun_period`	`int`	Kijun-sen period (default 26)
`senkou_b_period`	`int`	Senkou Span B period (default 52)

cloud_top `property` ¶

cloud_top: Decimal

Get the top of the cloud.

cloud_bottom `property` ¶

cloud_bottom: Decimal

Get the bottom of the cloud.

is_bullish_crossover `property` ¶

is_bullish_crossover: bool

Check if Tenkan crossed above Kijun (bullish).

is_bearish_crossover `property` ¶

is_bearish_crossover: bool

Check if Tenkan crossed below Kijun (bearish).

is_above_cloud `property` ¶

is_above_cloud: bool

Check if price is above the cloud.

is_below_cloud `property` ¶

is_below_cloud: bool

Check if price is below the cloud.

signal `property` ¶

signal: str

Get signal based on Ichimoku analysis.

ChainHealthStatus¶

almanak.framework.strategies.ChainHealthStatus ¶

Bases: Enum

Status of a chain's data health.

Attributes:

Name	Type	Description
`HEALTHY`		Chain data is fresh and available
`DEGRADED`		Chain data is stale but still usable (between threshold and 2x threshold)
`UNAVAILABLE`		Chain data could not be fetched
`STALE`		Chain data is too old to be trusted (beyond staleness threshold)

ChainHealth¶

almanak.framework.strategies.ChainHealth `dataclass` ¶

ChainHealth(
    chain: str,
    status: ChainHealthStatus,
    last_updated: datetime | None = None,
    staleness_seconds: float | None = None,
    stale_threshold_seconds: float = 30.0,
    error: str | None = None,
)

Health status and staleness information for a single chain.

This dataclass provides detailed information about the health of market data for a specific chain, including when data was last fetched, staleness metrics, and any error information.

Attributes:

Name	Type	Description
`chain`	`str`	Chain name (e.g., "arbitrum", "optimism")
`status`	`ChainHealthStatus`	Current health status of the chain
`last_updated`	`datetime \| None`	When the chain's data was last successfully fetched
`staleness_seconds`	`float \| None`	How old the data is in seconds (None if unavailable)
`stale_threshold_seconds`	`float`	The threshold used to determine staleness
`error`	`str \| None`	Error message if data fetch failed
`is_stale`	`bool`	Whether the data is considered stale
`is_available`	`bool`	Whether the data is available for use

Example

health = ChainHealth( chain="arbitrum", status=ChainHealthStatus.HEALTHY, last_updated=datetime.now(timezone.utc), staleness_seconds=5.2, stale_threshold_seconds=30.0, )

if health.is_stale: logger.warning(f"Chain {health.chain} data is stale")

is_stale `property` ¶

is_stale: bool

Check if the chain data is stale.

Returns:

Type	Description
`bool`	True if staleness exceeds threshold or data is unavailable

is_available `property` ¶

is_available: bool

Check if the chain data is available for use.

Returns:

Type	Description
`bool`	True if data is healthy or degraded (but still usable)

is_healthy `property` ¶

is_healthy: bool

Check if the chain data is fully healthy.

Returns:

Type	Description
`bool`	True if status is HEALTHY

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary representation.

Returns:

Type	Description
`dict[str, Any]`	Dictionary with health information

from_dict `classmethod` ¶

from_dict(data: dict[str, Any]) -> ChainHealth

Create ChainHealth from dictionary.

Parameters:

Name	Type	Description	Default
`data`	`dict[str, Any]`	Dictionary with health information	required

Returns:

Type	Description
`ChainHealth`	ChainHealth instance

Strategies¶

Implementing Teardown¶

Required Methods¶

Execution Order¶

Example: Swap Strategy Teardown¶

IntentStrategy¶

almanak.framework.strategies.IntentStrategy ¶

chain property ¶

wallet_address property ¶

compiler property writable ¶

current_intent property ¶

current_state_machine property ¶

set_state_manager ¶

get_persistent_state ¶

load_persistent_state ¶

save_state ¶

flush_pending_saves async ¶

load_state ¶

decide abstractmethod ¶

on_intent_executed ¶

valuate ¶

on_vault_settled ¶

set_multi_chain_providers ¶

is_multi_chain ¶

get_supported_chains ¶

create_market_snapshot ¶

run ¶

run_multi ¶

run_with_state_machine ¶

get_metadata ¶

to_dict ¶

on_sadflow_enter ¶

supports_teardown ¶

get_portfolio_snapshot ¶

get_open_positions ¶

generate_teardown_intents ¶

on_teardown_started ¶

on_teardown_completed ¶

get_teardown_profile ¶

acknowledge_teardown_request ¶

should_teardown ¶

on_sadflow_exit ¶

on_retry ¶

StrategyBase¶

almanak.framework.strategies.StrategyBase ¶

strategy_id property ¶

chain property ¶

run abstractmethod ¶

update_config ¶

set_notification_callback ¶

get_config_history ¶

get_current_config_version ¶

get_config ¶

restore_config_from_snapshot ¶

MarketSnapshot¶

almanak.framework.strategies.MarketSnapshot ¶

chain property ¶

wallet_address property ¶

timestamp property ¶

price ¶

price_data ¶

rsi ¶

macd ¶

bollinger_bands ¶

stochastic ¶

atr ¶

sma ¶

ema ¶

adx ¶

obv ¶

cci ¶

ichimoku ¶

balance ¶

balance_usd ¶

total_portfolio_usd ¶

set_price ¶

set_price_data ¶

set_balance ¶

set_rsi ¶

set_macd ¶

chain `property` ¶

wallet_address `property` ¶

compiler `property` `writable` ¶

current_intent `property` ¶

current_state_machine `property` ¶

flush_pending_saves `async` ¶

decide `abstractmethod` ¶

strategy_id `property` ¶

chain `property` ¶

run `abstractmethod` ¶

chain `property` ¶

wallet_address `property` ¶

timestamp `property` ¶

almanak.framework.strategies.RiskGuardConfig `dataclass` ¶

almanak.framework.strategies.DecideResult `module-attribute` ¶

almanak.framework.strategies.IntentSequence `dataclass` ¶

first `property` ¶

last `property` ¶

len ¶

iter ¶

getitem ¶

deserialize `classmethod` ¶

almanak.framework.strategies.ExecutionResult `dataclass` ¶

almanak.framework.strategies.TokenBalance `dataclass` ¶

almanak.framework.strategies.PriceData `dataclass` ¶

almanak.framework.strategies.RSIData `dataclass` ¶

signal `property` ¶

is_oversold `property` ¶

is_overbought `property` ¶

almanak.framework.strategies.MACDData `dataclass` ¶

is_bullish_crossover `property` ¶

is_bearish_crossover `property` ¶

signal `property` ¶

almanak.framework.strategies.BollingerBandsData `dataclass` ¶

is_oversold `property` ¶

is_overbought `property` ¶

is_squeeze `property` ¶

signal `property` ¶

almanak.framework.strategies.StochasticData `dataclass` ¶

is_oversold `property` ¶

is_overbought `property` ¶

is_bullish_crossover `property` ¶

is_bearish_crossover `property` ¶

signal `property` ¶

almanak.framework.strategies.ATRData `dataclass` ¶

is_high_volatility `property` ¶

is_low_volatility `property` ¶

signal `property` ¶

almanak.framework.strategies.MAData `dataclass` ¶