almanak-strategy-builder

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Almanak Strategy Builder

Almanak 策略构建器

You are helping a quant build DeFi strategies using the Almanak SDK. Strategies are Python classes that return Intent objects. The framework handles compilation to transactions, execution, and state management.

你正在帮助量化分析师使用Almanak SDK构建DeFi策略。策略是返回Intent对象的Python类，该框架会自动处理交易编译、执行和状态管理。

Quick Start

快速开始

bash

undefined

bash

undefined

Install

安装

pip install almanak

Scaffold a new strategy

生成新策略脚手架

almanak strat new --template mean_reversion --name my_rsi --chain arbitrum

Run on local Anvil fork (auto-starts gateway + Anvil)

在本地Anvil分叉链上运行（自动启动gateway + Anvil）

cd my_rsi almanak strat run --network anvil --once

Run a single iteration on mainnet

在主网运行单次迭代

almanak strat run --once

Browse and copy a working demo strategy

浏览并复制可运行的演示策略

almanak strat demo


**Minimal strategy (3 files):**

my_strategy/ strategy.py # IntentStrategy subclass with decide() method config.json # Runtime parameters (chain, tokens, thresholds) .env # Secrets (ALMANAK_PRIVATE_KEY, ALCHEMY_API_KEY)


For Anvil testing, add `anvil_funding` to `config.json` so your wallet is auto-funded on fork start
(see [Configuration](#configuration) below).

```python

almanak strat demo


**最小策略结构（3个文件）：**

my_strategy/ strategy.py # 实现decide()方法的IntentStrategy子类 config.json # 运行时参数（链、代币、阈值等） .env # 敏感信息（ALMANAK_PRIVATE_KEY、ALCHEMY_API_KEY）


如果要在Anvil上测试，可以在`config.json`中添加`anvil_funding`配置，这样分叉链启动时会自动为你的钱包充值（参考下文[配置](#configuration)部分）。

```python

strategy.py

from decimal import Decimal from almanak.framework.strategies import IntentStrategy, MarketSnapshot, almanak_strategy from almanak.framework.intents import Intent

@almanak_strategy( name="my_strategy", version="1.0.0", supported_chains=["arbitrum"], supported_protocols=["uniswap_v3"], intent_types=["SWAP", "HOLD"], ) class MyStrategy(IntentStrategy): def init(self, *args, **kwargs): super().init(*args, **kwargs) self.trade_size = Decimal(str(self.config.get("trade_size_usd", "100")))

def decide(self, market: MarketSnapshot) -> Intent | None:
    rsi = market.rsi("WETH", period=14)
    if rsi.value < 30:
        return Intent.swap(
            from_token="USDC", to_token="WETH",
            amount_usd=self.trade_size, max_slippage=Decimal("0.005"),
        )
    return Intent.hold(reason=f"RSI={rsi.value:.1f}, waiting")


> **Note:** `amount_usd=` requires a live price oracle from the gateway. If swaps revert with
> "Too little received", switch to `amount=` (token units) which bypasses USD-to-token conversion.
> Always verify pricing on first live run with `--dry-run --once`.

<!-- almanak-sdk-end: quick-start -->

<!-- almanak-sdk-start: core-concepts -->

from decimal import Decimal from almanak.framework.strategies import IntentStrategy, MarketSnapshot, almanak_strategy from almanak.framework.intents import Intent

def decide(self, market: MarketSnapshot) -> Intent | None:
    rsi = market.rsi("WETH", period=14)
    if rsi.value < 30:
        return Intent.swap(
            from_token="USDC", to_token="WETH",
            amount_usd=self.trade_size, max_slippage=Decimal("0.005"),
        )
    return Intent.hold(reason=f"RSI={rsi.value:.1f}, waiting")


> **注意：** `amount_usd=` 需要gateway提供实时价格预言机。如果兑换交易因为「到账金额过少」回滚，可以改用`amount=`（代币单位），这样会跳过USD到代币的转换。首次上线运行时一定要先使用`--dry-run --once`参数验证定价逻辑。

<!-- almanak-sdk-end: quick-start -->

<!-- almanak-sdk-start: core-concepts -->

Core Concepts

核心概念

IntentStrategy

All strategies inherit from

IntentStrategy

and implement one method:

python

def decide(self, market: MarketSnapshot) -> Intent | None

The framework calls

decide()

on each iteration with a fresh

MarketSnapshot

. Return an

Intent

object (swap, LP, borrow, etc.) or

Intent.hold()

所有策略都继承自

IntentStrategy

，并且需要实现一个方法：

python

def decide(self, market: MarketSnapshot) -> Intent | None

框架每次迭代都会传入最新的

MarketSnapshot

调用

decide()

方法，你可以返回一个

Intent

对象（兑换、LP做市、借贷等）或者

Intent.hold()

。

Lifecycle

生命周期

```
__init__
```
: Extract config parameters, set up state
```
decide(market)
```
: Called each iteration - return an Intent

on_intent_executed(intent, success, result)

: Optional callback after execution

```
get_status()
```
: Optional - return dict for monitoring dashboards

supports_teardown()

generate_teardown_intents()

: Optional safe shutdown

```
__init__
```
: 提取配置参数，初始化状态
```
decide(market)
```
: 每次迭代调用，返回一个Intent

on_intent_executed(intent, success, result)

: 可选回调，在Intent执行后触发

```
get_status()
```
: 可选，返回字典用于监控仪表盘展示

supports_teardown()

generate_teardown_intents()

: 可选的安全关闭逻辑

@almanak_strategy Decorator

@almanak_strategy 装饰器

Attaches metadata used by the framework and CLI:

python

@almanak_strategy(
    name="my_strategy",              # Unique identifier
    description="What it does",      # Human-readable description
    version="1.0.0",                 # Strategy version
    author="Your Name",              # Optional
    tags=["trading", "rsi"],         # Optional tags for discovery
    supported_chains=["arbitrum"],   # Which chains this runs on
    supported_protocols=["uniswap_v3"],  # Which protocols it uses
    intent_types=["SWAP", "HOLD"],   # Intent types it may return
)

用于附加框架和CLI需要用到的元数据：

python

@almanak_strategy(
    name="my_strategy",              # 唯一标识符
    description="What it does",      # 人类可读的描述
    version="1.0.0",                 # 策略版本
    author="Your Name",              # 可选
    tags=["trading", "rsi"],         # 可选标签，用于发现
    supported_chains=["arbitrum"],   # 策略支持的链
    supported_protocols=["uniswap_v3"],  # 策略用到的协议
    intent_types=["SWAP", "HOLD"],   # 可能返回的Intent类型
)

Config Access

配置访问

__init__

, read parameters from

self.config

(dict loaded from config.json):

python

def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)
    self.trade_size = Decimal(str(self.config.get("trade_size_usd", "100")))
    self.rsi_period = int(self.config.get("rsi_period", 14))
    self.base_token = self.config.get("base_token", "WETH")

Also available:

self.chain

(str),

self.wallet_address

(str).

在

__init__

中可以从

self.config

读取参数（从config.json加载的字典）：

python

def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)
    self.trade_size = Decimal(str(self.config.get("trade_size_usd", "100")))
    self.rsi_period = int(self.config.get("rsi_period", 14))
    self.base_token = self.config.get("base_token", "WETH")

还可以访问：

self.chain

（字符串，当前链）、

self.wallet_address

（字符串，钱包地址）。

Intent Reference

Intent 参考

All intents are created via

Intent

factory methods. Import:

python

from almanak.framework.intents import Intent

所有Intent都通过

Intent

工厂方法创建，导入方式：

python

from almanak.framework.intents import Intent

Trading

交易类

Intent.swap - Exchange tokens on a DEX

python

Intent.swap(
    from_token="USDC",           # Token to sell
    to_token="WETH",             # Token to buy
    amount_usd=Decimal("1000"),  # Amount in USD (use amount_usd OR amount)
    amount=Decimal("500"),       # Amount in token units (alternative to amount_usd)
    max_slippage=Decimal("0.005"),  # Max slippage (0.5%)
    protocol="uniswap_v3",      # Optional: specific DEX
    chain="arbitrum",            # Optional: override chain
    destination_chain="base",    # Optional: cross-chain swap
)

Use

amount="all"

to swap the entire balance.

amount=
vs
amount_usd=
: Use

amount_usd=

to specify trade size in USD (requires a live price oracle from the gateway). Use

amount=

to specify exact token units (more reliable for live trading since it bypasses USD-to-token conversion). When in doubt, prefer

amount=

for mainnet.

Intent.swap - 在DEX上兑换代币

python

Intent.swap(
    from_token="USDC",           # 要卖出的代币
    to_token="WETH",             # 要买入的代币
    amount_usd=Decimal("1000"),  # USD计价的金额（二选一使用amount_usd或amount）
    amount=Decimal("500"),       # 代币单位的金额（amount_usd的替代选项）
    max_slippage=Decimal("0.005"),  # 最大滑点（0.5%）
    protocol="uniswap_v3",      # 可选：指定DEX
    chain="arbitrum",            # 可选：覆盖默认链
    destination_chain="base",    # 可选：跨链兑换
)

使用

amount="all"

可以兑换全部余额。

amount=
和
amount_usd=
的区别：使用

amount_usd

可以指定USD计价的交易规模（需要gateway提供实时价格预言机）。使用

amount

可以指定精确的代币单位（实盘交易更可靠，因为跳过了USD到代币的转换）。如果不确定，主网环境优先使用

amount

。

Liquidity Provision

流动性提供

Intent.lp_open - Open a concentrated LP position

python

Intent.lp_open(
    pool="WETH/USDC",               # Pool identifier
    amount0=Decimal("1.0"),          # Amount of token0
    amount1=Decimal("2000"),         # Amount of token1
    range_lower=Decimal("1800"),     # Lower price bound
    range_upper=Decimal("2200"),     # Upper price bound
    protocol="uniswap_v3",          # Default: uniswap_v3
    chain=None,                      # Optional override
)

Intent.lp_close - Close an LP position

python

Intent.lp_close(
    position_id="12345",     # NFT token ID from lp_open result
    pool="WETH/USDC",        # Optional pool identifier
    collect_fees=True,       # Collect accumulated fees
    protocol="uniswap_v3",
)

Intent.collect_fees - Harvest LP fees without closing

python

Intent.collect_fees(
    pool="WETH/USDC",
    protocol="traderjoe_v2",
)

Intent.lp_open - 开一个集中流动性LP仓位

python

Intent.lp_open(
    pool="WETH/USDC",               # 池标识符
    amount0=Decimal("1.0"),          # token0的数量
    amount1=Decimal("2000"),         # token1的数量
    range_lower=Decimal("1800"),     # 价格区间下限
    range_upper=Decimal("2200"),     # 价格区间上限
    protocol="uniswap_v3",          # 默认：uniswap_v3
    chain=None,                      # 可选覆盖链
)

Intent.lp_close - 关闭LP仓位

python

Intent.lp_close(
    position_id="12345",     # lp_open返回的NFT代币ID
    pool="WETH/USDC",        # 可选池标识符
    collect_fees=True,       # 归集累积的手续费
    protocol="uniswap_v3",
)

Intent.collect_fees - 提取LP手续费但不关闭仓位

python

Intent.collect_fees(
    pool="WETH/USDC",
    protocol="traderjoe_v2",
)

Lending / Borrowing

借贷类

Intent.supply - Deposit collateral into a lending protocol

python

Intent.supply(
    protocol="aave_v3",
    token="WETH",
    amount=Decimal("10"),
    use_as_collateral=True,   # Enable as collateral (default: True)
    market_id=None,           # Required for Morpho Blue
)

Intent.borrow - Borrow tokens against collateral

python

Intent.borrow(
    protocol="aave_v3",
    collateral_token="WETH",
    collateral_amount=Decimal("10"),
    borrow_token="USDC",
    borrow_amount=Decimal("5000"),
    interest_rate_mode="variable",  # Aave: "variable" or "stable"
    market_id=None,                 # Required for Morpho Blue
)

Intent.repay - Repay borrowed tokens

python

Intent.repay(
    protocol="aave_v3",
    token="USDC",
    amount=Decimal("5000"),
    repay_full=False,        # Set True to repay entire debt
    market_id=None,
)

Intent.withdraw - Withdraw from lending protocol

python

Intent.withdraw(
    protocol="aave_v3",
    token="WETH",
    amount=Decimal("10"),
    withdraw_all=False,      # Set True to withdraw everything
    market_id=None,
)

Intent.supply - 向借贷协议存入抵押品

python

Intent.supply(
    protocol="aave_v3",
    token="WETH",
    amount=Decimal("10"),
    use_as_collateral=True,   # 启用作为抵押品（默认：True）
    market_id=None,           # Morpho Blue需要填此项
)

Intent.borrow - 用抵押品借入代币

python

Intent.borrow(
    protocol="aave_v3",
    collateral_token="WETH",
    collateral_amount=Decimal("10"),
    borrow_token="USDC",
    borrow_amount=Decimal("5000"),
    interest_rate_mode="variable",  # Aave可选："variable"浮动 / "stable"固定
    market_id=None,                 # Morpho Blue需要填此项
)

Intent.repay - 偿还借入的代币

python

Intent.repay(
    protocol="aave_v3",
    token="USDC",
    amount=Decimal("5000"),
    repay_full=False,        # 设为True偿还全部债务
    market_id=None,
)

Intent.withdraw - 从借贷协议提取资金

python

Intent.withdraw(
    protocol="aave_v3",
    token="WETH",
    amount=Decimal("10"),
    withdraw_all=False,      # 设为True提取全部资金
    market_id=None,
)

Perpetuals

永续合约

Intent.perp_open - Open a perpetual futures position

python

Intent.perp_open(
    market="ETH/USD",
    collateral_token="USDC",
    collateral_amount=Decimal("1000"),
    size_usd=Decimal("5000"),
    is_long=True,
    leverage=Decimal("5"),
    max_slippage=Decimal("0.01"),
    protocol="gmx_v2",
)

Intent.perp_close - Close a perpetual futures position

python

Intent.perp_close(
    market="ETH/USD",
    collateral_token="USDC",
    is_long=True,
    size_usd=None,               # None = close full position
    max_slippage=Decimal("0.01"),
    protocol="gmx_v2",
)

Intent.perp_open - 开永续合约仓位

python

Intent.perp_open(
    market="ETH/USD",
    collateral_token="USDC",
    collateral_amount=Decimal("1000"),
    size_usd=Decimal("5000"),
    is_long=True,
    leverage=Decimal("5"),
    max_slippage=Decimal("0.01"),
    protocol="gmx_v2",
)

Intent.perp_close - 平永续合约仓位

python

Intent.perp_close(
    market="ETH/USD",
    collateral_token="USDC",
    is_long=True,
    size_usd=None,               # None = 平掉全部仓位
    max_slippage=Decimal("0.01"),
    protocol="gmx_v2",
)

Bridging

跨链桥

Intent.bridge - Cross-chain token transfer

python

Intent.bridge(
    token="USDC",
    amount=Decimal("1000"),
    from_chain="arbitrum",
    to_chain="base",
    max_slippage=Decimal("0.005"),
    preferred_bridge=None,       # Optional: specific bridge protocol
)

Intent.bridge - 跨链转移代币

python

Intent.bridge(
    token="USDC",
    amount=Decimal("1000"),
    from_chain="arbitrum",
    to_chain="base",
    max_slippage=Decimal("0.005"),
    preferred_bridge=None,       # 可选：指定跨链桥协议
)

Staking

质押

Intent.stake - Liquid staking deposit

python

Intent.stake(
    protocol="lido",
    token_in="ETH",
    amount=Decimal("10"),
    receive_wrapped=True,    # Receive wrapped token (e.g., wstETH)
)

Intent.unstake - Withdraw from liquid staking

python

Intent.unstake(
    protocol="lido",
    token_in="wstETH",
    amount=Decimal("10"),
)

Intent.stake - 流动性质押存款

python

Intent.stake(
    protocol="lido",
    token_in="ETH",
    amount=Decimal("10"),
    receive_wrapped=True,    # 接收包装后的代币（比如wstETH）
)

Intent.unstake - 从流动性质押中提取

python

Intent.unstake(
    protocol="lido",
    token_in="wstETH",
    amount=Decimal("10"),
)

Flash Loans

闪电贷

Intent.flash_loan - Borrow and repay in a single transaction

python

Intent.flash_loan(
    provider="aave",         # "aave", "balancer", "morpho", or "auto"
    token="USDC",
    amount=Decimal("100000"),
    callback_intents=[...],  # Intents to execute with the borrowed funds
)

Intent.flash_loan - 在同一笔交易中借入并偿还

python

Intent.flash_loan(
    provider="aave",         # "aave", "balancer", "morpho", 或 "auto"
    token="USDC",
    amount=Decimal("100000"),
    callback_intents=[...],  # 用借入资金执行的Intent列表
)

Vaults (ERC-4626)

金库（ERC-4626）

Intent.vault_deposit - Deposit into an ERC-4626 vault

python

Intent.vault_deposit(
    vault="0x...",           # Vault contract address
    asset_token="USDC",
    amount=Decimal("1000"),
)

Intent.vault_redeem - Redeem shares from an ERC-4626 vault

python

Intent.vault_redeem(
    vault="0x...",
    shares_amount=Decimal("1000"),
)

Intent.vault_deposit - 存入ERC-4626金库

python

Intent.vault_deposit(
    vault="0x...",           # 金库合约地址
    asset_token="USDC",
    amount=Decimal("1000"),
)

Intent.vault_redeem - 从ERC-4626金库赎回份额

python

Intent.vault_redeem(
    vault="0x...",
    shares_amount=Decimal("1000"),
)

Prediction Markets

预测市场

python

Intent.prediction_buy(protocol="polymarket", market="...", amount_usd=Decimal("100"))
Intent.prediction_sell(protocol="polymarket", market="...", amount_shares=Decimal("50"))
Intent.prediction_redeem(protocol="polymarket", market="...")

python

Intent.prediction_buy(protocol="polymarket", market="...", amount_usd=Decimal("100"))
Intent.prediction_sell(protocol="polymarket", market="...", amount_shares=Decimal("50"))
Intent.prediction_redeem(protocol="polymarket", market="...")

Cross-Chain

跨链

Intent.ensure_balance - Meta-intent that resolves to a

BridgeIntent

(if balance is insufficient) or

HoldIntent

(if already met). Call

.resolve(market)

before returning from

decide()

python

intent = Intent.ensure_balance(
    token="USDC",
    min_amount=Decimal("1000"),
    target_chain="arbitrum",
    max_slippage=Decimal("0.005"),
    preferred_bridge=None,
)

Intent.ensure_balance - 元Intent，如果目标链余额不足会解析为

BridgeIntent

，如果余额足够则返回

HoldIntent

。在

decide()

返回前需要调用

.resolve(market)

。

python

intent = Intent.ensure_balance(
    token="USDC",
    min_amount=Decimal("1000"),
    target_chain="arbitrum",
    max_slippage=Decimal("0.005"),
    preferred_bridge=None,
)

Must resolve before returning - returns BridgeIntent or HoldIntent

返回前必须解析，返回BridgeIntent或HoldIntent

resolved = intent.resolve(market) return resolved

undefined

resolved = intent.resolve(market) return resolved

undefined

Control Flow

控制流

Intent.hold - Do nothing this iteration

python

Intent.hold(reason="RSI in neutral zone")

Intent.sequence - Execute multiple intents in order

python

Intent.sequence(
    intents=[
        Intent.swap(from_token="USDC", to_token="WETH", amount_usd=Decimal("1000")),
        Intent.supply(protocol="aave_v3", token="WETH", amount=Decimal("0.5")),
    ],
    description="Buy WETH then supply to Aave",
)

Intent.hold - 本次迭代不执行任何操作

python

Intent.hold(reason="RSI处于中性区间")

Intent.sequence - 按顺序执行多个Intent

python

Intent.sequence(
    intents=[
        Intent.swap(from_token="USDC", to_token="WETH", amount_usd=Decimal("1000")),
        Intent.supply(protocol="aave_v3", token="WETH", amount=Decimal("0.5")),
    ],
    description="买入WETH然后存入Aave",
)

Chained Amounts

链式金额传递

Use

"all"

to reference the full output of a prior intent:

python

Intent.sequence(intents=[
    Intent.swap(from_token="USDC", to_token="WETH", amount_usd=Decimal("1000")),
    Intent.supply(protocol="aave_v3", token="WETH", amount="all"),  # Uses swap output
])

使用

"all"

可以引用上一个Intent的全部输出：

python

Intent.sequence(intents=[
    Intent.swap(from_token="USDC", to_token="WETH", amount_usd=Decimal("1000")),
    Intent.supply(protocol="aave_v3", token="WETH", amount="all"),  # 使用兑换的全部输出
])

Market Data API

市场数据API

The

MarketSnapshot

passed to

decide()

provides these methods:

传入

decide()

的

MarketSnapshot

提供以下方法：

Prices

价格

python

price = market.price("WETH")                    # Decimal, USD price
price = market.price("WETH", quote="USDC")      # Price in USDC terms

pd = market.price_data("WETH")                  # PriceData object
pd.price             # Decimal - current price
pd.price_24h_ago     # Decimal
pd.change_24h_pct    # Decimal
pd.high_24h          # Decimal
pd.low_24h           # Decimal
pd.timestamp         # datetime

python

price = market.price("WETH")                    # Decimal类型，USD计价价格
price = market.price("WETH", quote="USDC")      # USDC计价的价格

pd = market.price_data("WETH")                  # PriceData对象
pd.price             # Decimal - 当前价格
pd.price_24h_ago     # Decimal
pd.change_24h_pct    # Decimal
pd.high_24h          # Decimal
pd.low_24h           # Decimal
pd.timestamp         # datetime

Balances

余额

python

bal = market.balance("USDC")
bal.balance       # Decimal - token amount
bal.balance_usd   # Decimal - USD value
bal.symbol        # str
bal.address       # str - token contract address

TokenBalance

supports numeric comparisons:

bal > Decimal("100")

python

bal = market.balance("USDC")
bal.balance       # Decimal - 代币数量
bal.balance_usd   # Decimal - USD价值
bal.symbol        # str
bal.address       # str - 代币合约地址

TokenBalance

支持数值比较：

bal > Decimal("100")

。

Technical Indicators

技术指标

All indicators accept

token

period

(int), and

timeframe

(str, default

"4h"

python

rsi = market.rsi("WETH", period=14, timeframe="4h")
rsi.value          # Decimal (0-100)
rsi.is_oversold    # bool (value < 30)
rsi.is_overbought  # bool (value > 70)
rsi.signal         # "BUY" | "SELL" | "HOLD"

macd = market.macd("WETH", fast_period=12, slow_period=26, signal_period=9)
macd.macd_line     # Decimal
macd.signal_line   # Decimal
macd.histogram     # Decimal
macd.is_bullish_crossover  # bool
macd.is_bearish_crossover  # bool

bb = market.bollinger_bands("WETH", period=20, std_dev=2.0)
bb.upper_band      # Decimal
bb.middle_band     # Decimal
bb.lower_band      # Decimal
bb.bandwidth        # Decimal
bb.percent_b        # Decimal (0.0 = at lower band, 1.0 = at upper band)
bb.is_squeeze       # bool

stoch = market.stochastic("WETH", k_period=14, d_period=3)
stoch.k_value       # Decimal
stoch.d_value       # Decimal
stoch.is_oversold   # bool
stoch.is_overbought # bool

atr_val = market.atr("WETH", period=14)
atr_val.value       # Decimal (absolute)
atr_val.value_percent  # Decimal, percentage points (2.62 means 2.62%, not 0.0262)
atr_val.is_high_volatility  # bool

sma = market.sma("WETH", period=20)
ema = market.ema("WETH", period=12)

所有指标都接受

token

、

period

（整数）和

timeframe

（字符串，默认

"4h"

）参数。

python

rsi = market.rsi("WETH", period=14, timeframe="4h")
rsi.value          # Decimal (0-100)
rsi.is_oversold    # bool (值 < 30)
rsi.is_overbought  # bool (值 > 70)
rsi.signal         # "BUY" | "SELL" | "HOLD"

macd = market.macd("WETH", fast_period=12, slow_period=26, signal_period=9)
macd.macd_line     # Decimal
macd.signal_line   # Decimal
macd.histogram     # Decimal
macd.is_bullish_crossover  # bool
macd.is_bearish_crossover  # bool

bb = market.bollinger_bands("WETH", period=20, std_dev=2.0)
bb.upper_band      # Decimal
bb.middle_band     # Decimal
bb.lower_band      # Decimal
bb.bandwidth        # Decimal
bb.percent_b        # Decimal (0.0 = 在下轨, 1.0 = 在上轨)
bb.is_squeeze       # bool

stoch = market.stochastic("WETH", k_period=14, d_period=3)
stoch.k_value       # Decimal
stoch.d_value       # Decimal
stoch.is_oversold   # bool
stoch.is_overbought # bool

atr_val = market.atr("WETH", period=14)
atr_val.value       # Decimal (绝对值)
atr_val.value_percent  # Decimal, 百分点（2.62代表2.62%，不是0.0262）
atr_val.is_high_volatility  # bool

sma = market.sma("WETH", period=20)
ema = market.ema("WETH", period=12)

Both return MAData with: .value, .is_price_above, .is_price_below, .signal

两者都返回MAData对象，包含：.value, .is_price_above, .is_price_below, .signal

adx = market.adx("WETH", period=14) adx.adx # Decimal (0-100, trend strength) adx.plus_di # Decimal (+DI) adx.minus_di # Decimal (-DI) adx.is_strong_trend # bool (adx >= 25) adx.is_uptrend # bool (+DI > -DI) adx.is_downtrend # bool (-DI > +DI)

obv = market.obv("WETH") obv.obv # Decimal (OBV value) obv.signal_line # Decimal (SMA of OBV) obv.is_bullish # bool (OBV > signal) obv.is_bearish # bool (OBV < signal)

cci = market.cci("WETH", period=20) cci.value # Decimal cci.is_oversold # bool (value <= -100) cci.is_overbought # bool (value >= 100)

ich = market.ichimoku("WETH") ich.tenkan_sen # Decimal (conversion line) ich.kijun_sen # Decimal (base line) ich.senkou_span_a # Decimal (leading span A) ich.senkou_span_b # Decimal (leading span B) ich.cloud_top # Decimal ich.cloud_bottom # Decimal ich.is_bullish_crossover # bool (tenkan > kijun) ich.is_above_cloud # bool ich.signal # "BUY" | "SELL" | "HOLD"

undefined

adx = market.adx("WETH", period=14) adx.adx # Decimal (0-100, 趋势强度) adx.plus_di # Decimal (+DI) adx.minus_di # Decimal (-DI) adx.is_strong_trend # bool (adx >= 25) adx.is_uptrend # bool (+DI > -DI) adx.is_downtrend # bool (-DI > +DI)

obv = market.obv("WETH") obv.obv # Decimal (OBV值) obv.signal_line # Decimal (OBV的SMA) obv.is_bullish # bool (OBV > 信号线) obv.is_bearish # bool (OBV < 信号线)

cci = market.cci("WETH", period=20) cci.value # Decimal cci.is_oversold # bool (值 <= -100) cci.is_overbought # bool (值 >= 100)

ich = market.ichimoku("WETH") ich.tenkan_sen # Decimal (转换线) ich.kijun_sen # Decimal (基准线) ich.senkou_span_a # Decimal (先行带A) ich.senkou_span_b # Decimal (先行带B) ich.cloud_top # Decimal ich.cloud_bottom # Decimal ich.is_bullish_crossover # bool (转换线上穿基准线) ich.is_above_cloud # bool ich.signal # "BUY" | "SELL" | "HOLD"

undefined

Multi-Token Queries

多代币查询

python

prices = market.prices(["WETH", "WBTC"])           # dict[str, Decimal]
balances = market.balances(["USDC", "WETH"])        # dict[str, Decimal]
usd_val = market.balance_usd("WETH")               # Decimal - USD value of holdings
total = market.total_portfolio_usd()                # Decimal

python

prices = market.prices(["WETH", "WBTC"])           # dict[str, Decimal]
balances = market.balances(["USDC", "WETH"])        # dict[str, Decimal]
usd_val = market.balance_usd("WETH")               # Decimal - 持仓的USD价值
total = market.total_portfolio_usd()                # Decimal

OHLCV Data

OHLCV数据

python

df = market.ohlcv("WETH", timeframe="1h", limit=100)  # pd.DataFrame

python

df = market.ohlcv("WETH", timeframe="1h", limit=100)  # pd.DataFrame

Columns: open, high, low, close, volume

列：open, high, low, close, volume

undefined

undefined

Pool and DEX Data

池和DEX数据

python

pool = market.pool_price("0x...")                   # DataEnvelope[PoolPrice]
pool = market.pool_price_by_pair("WETH", "USDC")   # DataEnvelope[PoolPrice]
reserves = market.pool_reserves("0x...")            # PoolReserves
history = market.pool_history("0x...", resolution="1h")  # DataEnvelope[list[PoolSnapshot]]
analytics = market.pool_analytics("0x...")          # DataEnvelope[PoolAnalytics]
best = market.best_pool("WETH", "USDC", metric="fee_apr")  # DataEnvelope[PoolAnalyticsResult]

python

pool = market.pool_price("0x...")                   # DataEnvelope[PoolPrice]
pool = market.pool_price_by_pair("WETH", "USDC")   # DataEnvelope[PoolPrice]
reserves = market.pool_reserves("0x...")            # PoolReserves
history = market.pool_history("0x...", resolution="1h")  # DataEnvelope[list[PoolSnapshot]]
analytics = market.pool_analytics("0x...")          # DataEnvelope[PoolAnalytics]
best = market.best_pool("WETH", "USDC", metric="fee_apr")  # DataEnvelope[PoolAnalyticsResult]

Price Aggregation and Slippage

价格聚合和滑点

python

twap = market.twap("WETH/USDC", window_seconds=300)       # DataEnvelope[AggregatedPrice]
lwap = market.lwap("WETH/USDC")                           # DataEnvelope[AggregatedPrice]
depth = market.liquidity_depth("0x...")                    # DataEnvelope[LiquidityDepth]
slip = market.estimate_slippage("WETH", "USDC", Decimal("10000"))  # DataEnvelope[SlippageEstimate]
best_dex = market.best_dex_price("WETH", "USDC", Decimal("1"))    # BestDexResult

python

twap = market.twap("WETH/USDC", window_seconds=300)       # DataEnvelope[AggregatedPrice]
lwap = market.lwap("WETH/USDC")                           # DataEnvelope[AggregatedPrice]
depth = market.liquidity_depth("0x...")                    # DataEnvelope[LiquidityDepth]
slip = market.estimate_slippage("WETH", "USDC", Decimal("10000"))  # DataEnvelope[SlippageEstimate]
best_dex = market.best_dex_price("WETH", "USDC", Decimal("1"))    # BestDexResult

Lending and Funding Rates

借贷和资金费率

python

rate = market.lending_rate("aave_v3", "USDC", side="supply")   # LendingRate
best = market.best_lending_rate("USDC", side="supply")         # BestRateResult
fr = market.funding_rate("binance", "ETH-PERP")               # FundingRate
spread = market.funding_rate_spread("ETH-PERP", "binance", "hyperliquid")  # FundingRateSpread

python

rate = market.lending_rate("aave_v3", "USDC", side="supply")   # LendingRate
best = market.best_lending_rate("USDC", side="supply")         # BestRateResult
fr = market.funding_rate("binance", "ETH-PERP")               # FundingRate
spread = market.funding_rate_spread("ETH-PERP", "binance", "hyperliquid")  # FundingRateSpread

Impermanent Loss

无常损失

python

il = market.il_exposure("position_id", fees_earned=Decimal("50"))  # ILExposure
proj = market.projected_il("WETH", "USDC", price_change_pct=Decimal("0.1"))  # ProjectedILResult

python

il = market.il_exposure("position_id", fees_earned=Decimal("50"))  # ILExposure
proj = market.projected_il("WETH", "USDC", price_change_pct=Decimal("0.1"))  # ProjectedILResult

Prediction Markets

预测市场

python

mkt = market.prediction("market_id")                    # PredictionMarket
price = market.prediction_price("market_id", "YES")     # Decimal
positions = market.prediction_positions("market_id")     # list[PredictionPosition]
orders = market.prediction_orders("market_id")           # list[PredictionOrder]

python

mkt = market.prediction("market_id")                    # PredictionMarket
price = market.prediction_price("market_id", "YES")     # Decimal
positions = market.prediction_positions("market_id")     # list[PredictionPosition]
orders = market.prediction_orders("market_id")           # list[PredictionOrder]

Yield and Analytics

收益和分析

python

yields = market.yield_opportunities("USDC", min_tvl=100_000, sort_by="apy")  # DataEnvelope[list[YieldOpportunity]]
gas = market.gas_price()                                # GasPrice
health = market.health()                                # HealthReport
signals = market.wallet_activity(action_types=["SWAP", "LP_OPEN"])  # list

python

yields = market.yield_opportunities("USDC", min_tvl=100_000, sort_by="apy")  # DataEnvelope[list[YieldOpportunity]]
gas = market.gas_price()                                # GasPrice
health = market.health()                                # HealthReport
signals = market.wallet_activity(action_types=["SWAP", "LP_OPEN"])  # list

Context Properties

上下文属性

python

market.chain            # str - current chain name
market.wallet_address   # str - wallet address
market.timestamp        # datetime - snapshot timestamp

python

market.chain            # str - 当前链名称
market.wallet_address   # str - 钱包地址
market.timestamp        # datetime - 快照时间戳

State Management

状态管理

self.state

is a dict that persists across iterations and restarts. The runner calls

load_state()

on startup, restoring previously saved state from the StateManager (SQLite locally, gateway when deployed). Write to

self.state

during

decide()

on_intent_executed()

and changes are automatically persisted after each iteration.

python

def decide(self, market):
    # Read state (survives restarts)
    last_trade = self.state.get("last_trade_time")
    position_id = self.state.get("lp_position_id")

    # Write state (persisted automatically after each iteration)
    self.state["last_trade_time"] = datetime.now().isoformat()
    self.state["consecutive_holds"] = self.state.get("consecutive_holds", 0) + 1

Use

--fresh

to clear saved state when starting over:

almanak strat run --fresh --once

For strategies that need custom serialization (e.g., Decimal fields, complex objects), override

get_persistent_state()

and

load_persistent_state()

python

def get_persistent_state(self) -> dict:
    """Called by framework to serialize state for persistence."""
    return {
        "phase": self.state.get("phase", "idle"),
        "position_id": self.state.get("position_id"),
        "entry_price": str(self.state.get("entry_price", "0")),
    }

def load_persistent_state(self, saved: dict) -> None:
    """Called by framework on startup to restore state."""
    self.state["phase"] = saved.get("phase", "idle")
    self.state["position_id"] = saved.get("position_id")
    self.state["entry_price"] = Decimal(saved.get("entry_price", "0"))

self.state

是一个字典，会在迭代之间甚至重启后持久化。运行器在启动时会调用

load_state()

，从StateManager恢复之前保存的状态（本地是SQLite，部署后是gateway）。在

decide()

或

on_intent_executed()

中写入

self.state

，每次迭代后变更会自动持久化。

python

def decide(self, market):
    # 读取状态（重启后仍然存在）
    last_trade = self.state.get("last_trade_time")
    position_id = self.state.get("lp_position_id")

    # 写入状态（每次迭代后自动持久化）
    self.state["last_trade_time"] = datetime.now().isoformat()
    self.state["consecutive_holds"] = self.state.get("consecutive_holds", 0) + 1

重新开始时可以使用

--fresh

清除保存的状态：

almanak strat run --fresh --once

。

如果策略需要自定义序列化（比如Decimal字段、复杂对象），可以重写

get_persistent_state()

和

load_persistent_state()

：

python

def get_persistent_state(self) -> dict:
    """框架调用此方法序列化状态用于持久化。"""
    return {
        "phase": self.state.get("phase", "idle"),
        "position_id": self.state.get("position_id"),
        "entry_price": str(self.state.get("entry_price", "0")),
    }

def load_persistent_state(self, saved: dict) -> None:
    """框架在启动时调用此方法恢复状态。"""
    self.state["phase"] = saved.get("phase", "idle")
    self.state["position_id"] = saved.get("position_id")
    self.state["entry_price"] = Decimal(saved.get("entry_price", "0"))

on_intent_executed Callback

on_intent_executed 回调

After execution, access results (position IDs, swap amounts) via the callback. The framework automatically enriches

result

with protocol-specific data - no manual receipt parsing needed.

python

undefined

执行完成后，可以通过回调访问结果（仓位ID、兑换金额等）。框架会自动为

result

填充协议特定的数据，不需要手动解析交易收据。

python

undefined

In your strategy file, import logging at the top:

在策略文件顶部导入logging：

import logging

logger = logging.getLogger(name)

def on_intent_executed(self, intent, success: bool, result):

if not success:
    logger.warning(f"Intent failed: {intent.intent_type}")
    return

# Capture LP position ID (enriched automatically by ResultEnricher)
if result.position_id is not None:
    self.state["lp_position_id"] = result.position_id
    logger.info(f"Opened LP position {result.position_id}")

    # Store range bounds for rebalancing strategies (keep as Decimal)
    if (
        hasattr(intent, "range_lower") and intent.range_lower is not None
        and hasattr(intent, "range_upper") and intent.range_upper is not None
    ):
        self.state["range_lower"] = intent.range_lower
        self.state["range_upper"] = intent.range_upper

# Capture swap amounts
if result.swap_amounts:
    self.state["last_swap"] = {
        "amount_in": str(result.swap_amounts.amount_in),
        "amount_out": str(result.swap_amounts.amount_out),
    }
    logger.info(
        f"Swapped {result.swap_amounts.amount_in} -> {result.swap_amounts.amount_out}"
    )


<!-- almanak-sdk-end: state-management -->

<!-- almanak-sdk-start: configuration -->

def on_intent_executed(self, intent, success: bool, result):

if not success:
    logger.warning(f"Intent执行失败：{intent.intent_type}")
    return

# 捕获LP仓位ID（由ResultEnricher自动填充）
if result.position_id is not None:
    self.state["lp_position_id"] = result.position_id
    logger.info(f"已开LP仓位 {result.position_id}")

    # 存储区间上下限用于再平衡策略（保留Decimal类型）
    if (
        hasattr(intent, "range_lower") and intent.range_lower is not None
        and hasattr(intent, "range_upper") and intent.range_upper is not None
    ):
        self.state["range_lower"] = intent.range_lower
        self.state["range_upper"] = intent.range_upper

# 捕获兑换金额
if result.swap_amounts:
    self.state["last_swap"] = {
        "amount_in": str(result.swap_amounts.amount_in),
        "amount_out": str(result.swap_amounts.amount_out),
    }
    logger.info(
        f"兑换完成 {result.swap_amounts.amount_in} -> {result.swap_amounts.amount_out}"
    )


<!-- almanak-sdk-end: state-management -->

<!-- almanak-sdk-start: configuration -->

Configuration

配置

config.json

json

{
    "strategy_id": "my_rsi_strategy",
    "chain": "arbitrum",
    "base_token": "WETH",
    "quote_token": "USDC",
    "rsi_period": 14,
    "rsi_oversold": 30,
    "rsi_overbought": 70,
    "trade_size_usd": 1000,
    "max_slippage_bps": 50,
    "anvil_funding": {
        "USDC": "10000",
        "WETH": "5"
    }
}

Required fields:

strategy_id

chain

. All other fields are strategy-specific and accessed via

self.config.get(key, default)

json

{
    "strategy_id": "my_rsi_strategy",
    "chain": "arbitrum",
    "base_token": "WETH",
    "quote_token": "USDC",
    "rsi_period": 14,
    "rsi_oversold": 30,
    "rsi_overbought": 70,
    "trade_size_usd": 1000,
    "max_slippage_bps": 50,
    "anvil_funding": {
        "USDC": "10000",
        "WETH": "5"
    }
}

必填字段：

strategy_id

、

chain

。其他字段都是策略特定的，可以通过

self.config.get(key, default)

访问。

.env

bash

undefined

bash

undefined

Required

必填

ALMANAK_PRIVATE_KEY=0x...

RPC access (set at least one)

RPC访问（至少配置一个）

ALCHEMY_API_KEY=your_key

RPC_URL=https://...

Optional

可选

ENSO_API_KEY=

COINGECKO_API_KEY=

ALMANAK_API_KEY=

undefined

undefined

anvil_funding

When running on Anvil (

--network anvil

), the framework auto-funds the wallet with tokens specified in

anvil_funding

. Values are in token units (not USD).

在Anvil上运行时（

--network anvil

），框架会自动为钱包充值

anvil_funding

中指定的代币，值是代币单位（不是USD）。

Token Resolution

代币解析

Use

get_token_resolver()

for all token lookups. Never hardcode addresses.

python

from almanak.framework.data.tokens import get_token_resolver

resolver = get_token_resolver()

所有代币查询都应该使用

get_token_resolver()

，永远不要硬编码地址。

python

from almanak.framework.data.tokens import get_token_resolver

resolver = get_token_resolver()

Resolve by symbol

通过符号解析

token = resolver.resolve("USDC", "arbitrum")

-> ResolvedToken(symbol="USDC", address="0xaf88...", decimals=6, chain="arbitrum")

Resolve by address

通过地址解析

token = resolver.resolve("0xaf88d065e77c8cC2239327C5EDb3A432268e5831", "arbitrum")

Convenience

便捷方法

decimals = resolver.get_decimals("arbitrum", "USDC") # -> 6 address = resolver.get_address("arbitrum", "USDC") # -> "0xaf88..."

For DEX swaps (auto-wraps native tokens: ETH->WETH, MATIC->WMATIC)

用于DEX兑换（自动包装原生代币：ETH->WETH, MATIC->WMATIC）

token = resolver.resolve_for_swap("ETH", "arbitrum") # -> WETH

Resolve trading pair

解析交易对

usdc, weth = resolver.resolve_pair("USDC", "WETH", "arbitrum")


Resolution order: memory cache -> disk cache -> static registry -> gateway on-chain lookup.

Never default to 18 decimals. If the token is unknown, `TokenNotFoundError` is raised.

<!-- almanak-sdk-end: token-resolution -->

<!-- almanak-sdk-start: backtesting -->

usdc, weth = resolver.resolve_pair("USDC", "WETH", "arbitrum")


解析顺序：内存缓存 -> 磁盘缓存 -> 静态注册表 -> gateway链上查询。

永远不要默认用18位小数，如果代币不存在会抛出`TokenNotFoundError`。

<!-- almanak-sdk-end: token-resolution -->

<!-- almanak-sdk-start: backtesting -->

Backtesting

回测

PnL Backtest (historical prices, no on-chain execution)

PnL回测（历史价格，无链上执行）

bash

almanak strat backtest pnl -s my_strategy \
    --start 2024-01-01 --end 2024-06-01 \
    --initial-capital 10000

bash

almanak strat backtest pnl -s my_strategy \
    --start 2024-01-01 --end 2024-06-01 \
    --initial-capital 10000

Paper Trading (Anvil fork with real execution, PnL tracking)

模拟交易（Anvil分叉链真实执行，PnL跟踪）

bash

almanak strat backtest paper -s my_strategy \
    --duration 3600 --interval 60 \
    --initial-capital 10000

Paper trading runs the full strategy loop on an Anvil fork with real transaction execution, equity curve tracking, and JSON result logs.

bash

almanak strat backtest paper -s my_strategy \
    --duration 3600 --interval 60 \
    --initial-capital 10000

模拟交易在Anvil分叉链上运行完整的策略循环，包含真实交易执行、权益曲线跟踪和JSON结果日志。

Parameter Sweep

参数扫描

bash

almanak strat backtest sweep -s my_strategy \
    --start 2024-01-01 --end 2024-06-01 \
    --param "rsi_oversold:20,25,30" \
    --param "rsi_overbought:70,75,80"

Runs the PnL backtest across all parameter combinations and ranks by Sharpe ratio.

bash

almanak strat backtest sweep -s my_strategy \
    --start 2024-01-01 --end 2024-06-01 \
    --param "rsi_oversold:20,25,30" \
    --param "rsi_overbought:70,75,80"

会在所有参数组合上运行PnL回测，并按夏普比率排序。

Programmatic Backtesting

编程式回测

python

from almanak.framework.backtesting import BacktestEngine

engine = BacktestEngine(
    strategy_class=MyStrategy,
    config={...},
    start_date="2024-01-01",
    end_date="2024-06-01",
    initial_capital=10000,
)
results = engine.run()
results.sharpe_ratio
results.max_drawdown
results.total_return
results.plot()  # Matplotlib equity curve

python

from almanak.framework.backtesting import BacktestEngine

engine = BacktestEngine(
    strategy_class=MyStrategy,
    config={...},
    start_date="2024-01-01",
    end_date="2024-06-01",
    initial_capital=10000,
)
results = engine.run()
results.sharpe_ratio
results.max_drawdown
results.total_return
results.plot()  # Matplotlib权益曲线

Backtesting Limitations

回测限制

OHLCV data: The PnL backtester uses historical close prices from CoinGecko. Indicators that require OHLCV data (ATR, Stochastic, Ichimoku) need a paid CoinGecko tier or an external data source.
RPC for paper trading: Paper trading requires an RPC endpoint. Alchemy free tier is recommended for performance; public RPCs work but are slow.
No CWD auto-discovery: Backtest CLI commands (
```
backtest pnl
```
,
```
backtest paper
```
,
```
backtest sweep
```
) require an explicit
```
-s strategy_name
```
flag. They do not auto-discover strategies from the current directory like
```
strat run
```
does.
Percentage fields:
```
total_return_pct
```
and similar
```
_pct
```
result fields are decimal fractions (0.33 = 33%), not percentages.

OHLCV数据：PnL回测器使用CoinGecko的历史收盘价。需要OHLCV数据的指标（ATR、Stochastic、Ichimoku）需要付费版CoinGecko或者外部数据源。
模拟交易的RPC：模拟交易需要RPC端点，推荐使用Alchemy免费层以获得更好的性能，公共RPC也可以运行但速度较慢。
无CWD自动发现：回测CLI命令（
```
backtest pnl
```
、
```
backtest paper
```
、
```
backtest sweep
```
）需要显式指定
```
-s strategy_name
```
参数，不会像
```
strat run
```
那样从当前目录自动发现策略。
百分比字段：
```
total_return_pct
```
和类似的
```
_pct
```
结果字段是小数形式（0.33 = 33%），不是百分比数值。

CLI Commands

CLI命令

Strategy Management

策略管理

bash

almanak strat new                     # Interactive strategy scaffolding
almanak strat new -t mean_reversion -n my_rsi -c arbitrum  # Non-interactive
almanak strat demo                    # Browse and copy a working demo strategy

Templates:

blank

dynamic_lp

mean_reversion

bollinger

basis_trade

lending_loop

copy_trader

bash

almanak strat new                     # 交互式生成策略脚手架
almanak strat new -t mean_reversion -n my_rsi -c arbitrum  # 非交互式生成
almanak strat demo                    # 浏览并复制可运行的演示策略

模板：

blank

、

dynamic_lp

、

mean_reversion

、

bollinger

、

basis_trade

、

lending_loop

、

copy_trader

Running Strategies

运行策略

bash

almanak strat run --once              # Single iteration (from strategy dir)
almanak strat run -d path/to/strat --once  # Explicit directory
almanak strat run --network anvil --once   # Local Anvil fork
almanak strat run --interval 30       # Continuous (30s between iterations)
almanak strat run --dry-run --once    # No transactions submitted
almanak strat run --fresh --once      # Clear state before running
almanak strat run --id abc123 --once  # Resume previous run
almanak strat run --dashboard         # Launch live monitoring dashboard

bash

almanak strat run --once              # 单次迭代（在策略目录下运行）
almanak strat run -d path/to/strat --once  # 指定策略目录
almanak strat run --network anvil --once   # 本地Anvil分叉链运行
almanak strat run --interval 30       # 连续运行（迭代间隔30秒）
almanak strat run --dry-run --once    # 不提交交易
almanak strat run --fresh --once      # 运行前清除状态
almanak strat run --id abc123 --once  # 恢复之前的运行

Backtesting

回测

bash

almanak strat backtest pnl -s my_strategy            # Historical PnL simulation
almanak strat backtest paper -s my_strategy            # Paper trading on Anvil fork
almanak strat backtest sweep -s my_strategy           # Parameter sweep optimization

bash

almanak strat backtest pnl -s my_strategy            # 历史PnL模拟
almanak strat backtest paper -s my_strategy            # Anvil分叉链模拟交易
almanak strat backtest sweep -s my_strategy           # 参数扫描优化

Teardown

关闭策略

bash

almanak strat teardown plan           # Preview teardown intents
almanak strat teardown execute        # Execute teardown

bash

almanak strat teardown plan           # 预览关闭Intent
almanak strat teardown execute        # 执行关闭

Gateway

bash

almanak gateway                       # Start standalone gateway
almanak gateway --network anvil       # Gateway for local Anvil testing
almanak gateway --port 50052          # Custom port

bash

almanak gateway                       # 启动独立gateway
almanak gateway --network anvil       # 用于本地Anvil测试的gateway
almanak gateway --port 50052          # 自定义端口

Agent Skill Management

Agent技能管理

bash

almanak agent install                 # Auto-detect platforms and install
almanak agent install -p claude       # Install for specific platform
almanak agent install -p all          # Install for all 9 platforms
almanak agent update                  # Update installed skill files
almanak agent status                  # Check installation status

bash

almanak agent install                 # 自动检测平台并安装
almanak agent install -p claude       # 为指定平台安装
almanak agent install -p all          # 为所有9个平台安装
almanak agent update                  # 更新已安装的技能文件
almanak agent status                  # 检查安装状态

Documentation

文档

bash

almanak docs path                     # Path to bundled LLM docs
almanak docs dump                     # Print full LLM docs
almanak docs agent-skill              # Path to bundled agent skill
almanak docs agent-skill --dump       # Print agent skill content

bash

almanak docs path                     # 内置LLM文档的路径
almanak docs dump                     # 打印完整LLM文档
almanak docs agent-skill              # 内置Agent技能的路径
almanak docs agent-skill --dump       # 打印Agent技能内容

Supported Chains and Protocols

支持的链和协议

Chains

链

Chain	Enum Value	Config Name
Ethereum	`ETHEREUM`	`ethereum`
Arbitrum	`ARBITRUM`	`arbitrum`
Optimism	`OPTIMISM`	`optimism`
Base	`BASE`	`base`
Avalanche	`AVALANCHE`	`avalanche`
Polygon	`POLYGON`	`polygon`
BSC	`BSC`	`bsc`
Sonic	`SONIC`	`sonic`
Plasma	`PLASMA`	`plasma`
Blast	`BLAST`	`blast`
Mantle	`MANTLE`	`mantle`
Berachain	`BERACHAIN`	`berachain`

区块链	枚举值	配置名称
Ethereum	`ETHEREUM`	`ethereum`
Arbitrum	`ARBITRUM`	`arbitrum`
Optimism	`OPTIMISM`	`optimism`
Base	`BASE`	`base`
Avalanche	`AVALANCHE`	`avalanche`
Polygon	`POLYGON`	`polygon`
BSC	`BSC`	`bsc`
Sonic	`SONIC`	`sonic`
Plasma	`PLASMA`	`plasma`
Blast	`BLAST`	`blast`
Mantle	`MANTLE`	`mantle`
Berachain	`BERACHAIN`	`berachain`

Protocols

协议

Protocol	Enum Value	Type	Config Name
Uniswap V3	`UNISWAP_V3`	DEX / LP	`uniswap_v3`
PancakeSwap V3	`PANCAKESWAP_V3`	DEX / LP	`pancakeswap_v3`
SushiSwap V3	`SUSHISWAP_V3`	DEX / LP	`sushiswap_v3`
TraderJoe V2	`TRADERJOE_V2`	DEX / LP	`traderjoe_v2`
Aerodrome	`AERODROME`	DEX / LP	`aerodrome`
Enso	`ENSO`	Aggregator	`enso`
Pendle	`PENDLE`	Yield	`pendle`
MetaMorpho	`METAMORPHO`	Lending	`metamorpho`
LiFi	`LIFI`	Bridge	`lifi`
Vault	`VAULT`	ERC-4626	`vault`
Curve	`CURVE`	DEX / LP	`curve`
Balancer	`BALANCER`	DEX / LP	`balancer`
Aave V3	*	Lending	`aave_v3`
Morpho Blue	*	Lending	`morpho_blue`
Compound V3	*	Lending	`compound_v3`
GMX V2	*	Perps	`gmx_v2`
Hyperliquid	*	Perps	`hyperliquid`
Polymarket	*	Prediction	`polymarket`
Kraken	*	CEX	`kraken`
Lido	*	Staking	`lido`
Lagoon	*	Vault	`lagoon`

* These protocols do not have a

Protocol

enum value. Use the string config name (e.g.,

protocol="aave_v3"

) in intents. They are resolved by the intent compiler and transaction builder directly.

协议	枚举值	类型	配置名称
Uniswap V3	`UNISWAP_V3`	DEX / LP	`uniswap_v3`
PancakeSwap V3	`PANCAKESWAP_V3`	DEX / LP	`pancakeswap_v3`
SushiSwap V3	`SUSHISWAP_V3`	DEX / LP	`sushiswap_v3`
TraderJoe V2	`TRADERJOE_V2`	DEX / LP	`traderjoe_v2`
Aerodrome	`AERODROME`	DEX / LP	`aerodrome`
Enso	`ENSO`	聚合器	`enso`
Pendle	`PENDLE`	收益	`pendle`
MetaMorpho	`METAMORPHO`	借贷	`metamorpho`
LiFi	`LIFI`	跨链桥	`lifi`
Vault	`VAULT`	ERC-4626	`vault`
Curve	`CURVE`	DEX / LP	`curve`
Balancer	`BALANCER`	DEX / LP	`balancer`
Aave V3	*	借贷	`aave_v3`
Morpho Blue	*	借贷	`morpho_blue`
Compound V3	*	借贷	`compound_v3`
GMX V2	*	永续合约	`gmx_v2`
Hyperliquid	*	永续合约	`hyperliquid`
Polymarket	*	预测市场	`polymarket`
Kraken	*	CEX	`kraken`
Lido	*	质押	`lido`
Lagoon	*	金库	`lagoon`

* 这些协议没有

Protocol

枚举值，在Intent中使用字符串配置名称即可（比如

protocol="aave_v3"

），它们会由Intent编译器和交易构建器直接解析。

Networks

网络

Network	Enum Value	Description
Mainnet	`MAINNET`	Production chains
Anvil	`ANVIL`	Local fork for testing
Sepolia	`SEPOLIA`	Testnet

网络	枚举值	描述
主网	`MAINNET`	生产链
Anvil	`ANVIL`	本地测试分叉链
Sepolia	`SEPOLIA`	测试网

Protocol-Specific Notes

协议特定说明

GMX V2 (Perpetuals)

Market format: Use slash separator:
```
"BTC/USD"
```
,
```
"ETH/USD"
```
,
```
"LINK/USD"
```
(not dash).
Two-step execution: GMX V2 uses a keeper-based execution model. When you call
```
Intent.perp_open()
```
, the SDK submits an order creation transaction. A GMX keeper then executes the actual position change in a separate transaction.
```
on_intent_executed(success=True)
```
fires when the order creation TX confirms, not when the keeper executes the position. Strategies should poll position state before relying on it.
Minimum position size: GMX V2 enforces a minimum position size of approximately $11 net of fees. Orders below this threshold are silently rejected by the keeper with no on-chain error.
Collateral approvals: Handled automatically by the intent compiler (same as LP opens).
Position monitoring:
```
get_all_positions()
```
may not return positions immediately after opening due to keeper delay. Allow a few seconds before querying.
Supported chains: Arbitrum, Avalanche.
Collateral tokens: USDC, USDT (chain-dependent).

GMX V2（永续合约）

市场格式：使用斜杠分隔：
```
"BTC/USD"
```
、
```
"ETH/USD"
```
、
```
"LINK/USD"
```
（不要用短横线）。
两步执行：GMX V2使用keeper执行模型，当你调用
```
Intent.perp_open()
```
时，SDK提交订单创建交易，然后GMX keeper会在单独的交易中执行实际的仓位变更。
```
on_intent_executed(success=True)
```
会在订单创建交易确认时触发，不是在keeper执行仓位变更时触发。策略在依赖仓位状态前应该轮询仓位状态。
最小仓位规模：GMX V2要求最小仓位规模约为11美元（扣除手续费后），低于此阈值的订单会被keeper静默拒绝，链上不会有错误提示。
抵押品授权：由Intent编译器自动处理（和开LP仓位一样）。
仓位监控：开仓后
```
get_all_positions()
```
可能不会立即返回仓位，因为keeper执行有延迟，查询前请等待几秒。
支持的链：Arbitrum、Avalanche。
抵押代币：USDC、USDT（依链而定）。

Common Patterns

常用模式

RSI Mean Reversion (Trading)

RSI均值回归（交易）

python

def decide(self, market):
    rsi = market.rsi(self.base_token, period=self.rsi_period)
    quote_bal = market.balance(self.quote_token)
    base_bal = market.balance(self.base_token)

    if rsi.is_oversold and quote_bal.balance_usd >= self.trade_size:
        return Intent.swap(
            from_token=self.quote_token, to_token=self.base_token,
            amount_usd=self.trade_size, max_slippage=Decimal("0.005"),
        )
    if rsi.is_overbought and base_bal.balance_usd >= self.trade_size:
        return Intent.swap(
            from_token=self.base_token, to_token=self.quote_token,
            amount_usd=self.trade_size, max_slippage=Decimal("0.005"),
        )
    return Intent.hold(reason=f"RSI={rsi.value:.1f} in neutral zone")

python

def decide(self, market):
    rsi = market.rsi(self.base_token, period=self.rsi_period)
    quote_bal = market.balance(self.quote_token)
    base_bal = market.balance(self.base_token)

    if rsi.is_oversold and quote_bal.balance_usd >= self.trade_size:
        return Intent.swap(
            from_token=self.quote_token, to_token=self.base_token,
            amount_usd=self.trade_size, max_slippage=Decimal("0.005"),
        )
    if rsi.is_overbought and base_bal.balance_usd >= self.trade_size:
        return Intent.swap(
            from_token=self.base_token, to_token=self.quote_token,
            amount_usd=self.trade_size, max_slippage=Decimal("0.005"),
        )
    return Intent.hold(reason=f"RSI={rsi.value:.1f} in neutral zone")

LP Rebalancing

LP再平衡

python

def decide(self, market):
    price = market.price(self.base_token)
    position_id = self.state.get("lp_position_id")

    if position_id:
        # Check if price is out of range - close and reopen
        if price < self.state.get("range_lower") or price > self.state.get("range_upper"):
            return Intent.lp_close(position_id=position_id, protocol="uniswap_v3")

    # Open new position centered on current price
    atr = market.atr(self.base_token)
    half_range = price * (atr.value_percent / Decimal("100")) * 2  # value_percent is percentage points
    return Intent.lp_open(
        pool="WETH/USDC",
        amount0=Decimal("1"), amount1=Decimal("2000"),
        range_lower=price - half_range,
        range_upper=price + half_range,
    )

python

def decide(self, market):
    price = market.price(self.base_token)
    position_id = self.state.get("lp_position_id")

    if position_id:
        # 检查价格是否超出区间，如果是则关闭并重开
        if price < self.state.get("range_lower") or price > self.state.get("range_upper"):
            return Intent.lp_close(position_id=position_id, protocol="uniswap_v3")

    # 开以当前价格为中心的新仓位
    atr = market.atr(self.base_token)
    half_range = price * (atr.value_percent / Decimal("100")) * 2  # value_percent是百分点
    return Intent.lp_open(
        pool="WETH/USDC",
        amount0=Decimal("1"), amount1=Decimal("2000"),
        range_lower=price - half_range,
        range_upper=price + half_range,
    )

Multi-Step with IntentSequence

多步IntentSequence

python

def decide(self, market):
    return Intent.sequence(
        intents=[
            Intent.swap(from_token="USDC", to_token="WETH", amount_usd=Decimal("5000")),
            Intent.supply(protocol="aave_v3", token="WETH", amount="all"),
            Intent.borrow(
                protocol="aave_v3",
                collateral_token="WETH", collateral_amount=Decimal("0"),
                borrow_token="USDC", borrow_amount=Decimal("3000"),
            ),
        ],
        description="Leverage loop: buy WETH, supply, borrow USDC",
    )

python

def decide(self, market):
    return Intent.sequence(
        intents=[
            Intent.swap(from_token="USDC", to_token="WETH", amount_usd=Decimal("5000")),
            Intent.supply(protocol="aave_v3", token="WETH", amount="all"),
            Intent.borrow(
                protocol="aave_v3",
                collateral_token="WETH", collateral_amount=Decimal("0"),
                borrow_token="USDC", borrow_amount=Decimal("3000"),
            ),
        ],
        description="Leverage loop: buy WETH, supply, borrow USDC",
    )

Alerting

告警

python

from almanak.framework.alerting import AlertManager

class MyStrategy(IntentStrategy):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.alerts = AlertManager.from_config(self.config.get("alerting", {}))

    def decide(self, market):
        rsi = market.rsi("WETH")
        if rsi.value < 20:
            self.alerts.send("Extreme oversold: RSI={:.1f}".format(rsi.value), level="warning")
        # ... trading logic

python

from almanak.framework.alerting import AlertManager

class MyStrategy(IntentStrategy):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.alerts = AlertManager.from_config(self.config.get("alerting", {}))

    def decide(self, market):
        rsi = market.rsi("WETH")
        if rsi.value < 20:
            self.alerts.send("Extreme oversold: RSI={:.1f}".format(rsi.value), level="warning")
        # ... 交易逻辑

Safe Teardown

安全关闭

Implement teardown so the strategy can cleanly exit positions:

python

def supports_teardown(self) -> bool:
    return True

def generate_teardown_intents(self, mode, market=None) -> list[Intent]:
    intents = []
    position_id = self.state.get("lp_position_id")
    if position_id:
        intents.append(Intent.lp_close(position_id=position_id))
    # Swap all base token back to quote
    intents.append(Intent.swap(
        from_token=self.base_token, to_token=self.quote_token,
        amount="all", max_slippage=Decimal("0.03"),
    ))
    return intents

实现关闭逻辑，让策略可以干净地平仓退出：

python

def supports_teardown(self) -> bool:
    return True

def generate_teardown_intents(self, mode, market=None) -> list[Intent]:
    intents = []
    position_id = self.state.get("lp_position_id")
    if position_id:
        intents.append(Intent.lp_close(position_id=position_id))
    # 将所有基础代币换回计价代币
    intents.append(Intent.swap(
        from_token=self.base_token, to_token=self.quote_token,
        amount="all", max_slippage=Decimal("0.03"),
    ))
    return intents

Error Handling

错误处理

Always wrap

decide()

in try/except and return

Intent.hold()

on error:

python

def decide(self, market):
    try:
        # ... strategy logic
    except Exception as e:
        logger.exception(f"Error in decide(): {e}")
        return Intent.hold(reason=f"Error: {e}")

总是用try/except包裹

decide()

，出错时返回

Intent.hold()

：

python

def decide(self, market):
    try:
        # ... 策略逻辑
    except Exception as e:
        logger.exception(f"Error in decide(): {e}")
        return Intent.hold(reason=f"Error: {e}")

Execution Failure Tracking (Circuit Breaker)

执行失败跟踪（断路器）

The framework retries each failed intent up to

max_retries

(default: 3) with exponential backoff. However, after all retries are exhausted the strategy continues running and will attempt the same trade on the next iteration. Without a circuit breaker, this creates an infinite loop of reverted transactions that burn gas without any hope of success.

Always track consecutive execution failures in persistent state and stop trading (or enter an extended cooldown) after a threshold is reached:

python

MAX_CONSECUTIVE_FAILURES = 3     # Stop after 3 rounds of failed intents
FAILURE_COOLDOWN_SECONDS = 1800  # 30-min cooldown before retrying

def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)
    self.consecutive_failures = 0
    self.failure_cooldown_until = 0.0

def decide(self, market):
    try:
        now = time.time()

        # Circuit breaker: skip trading while in cooldown
        if now < self.failure_cooldown_until:
            remaining = int(self.failure_cooldown_until - now)
            return Intent.hold(
                reason=f"Circuit breaker active, cooldown {remaining}s remaining"
            )

        # Circuit breaker: enter cooldown after too many failures
        if self.consecutive_failures >= MAX_CONSECUTIVE_FAILURES:
            self.failure_cooldown_until = now + FAILURE_COOLDOWN_SECONDS
            self.consecutive_failures = 0
            logger.warning(
                f"Circuit breaker tripped after {MAX_CONSECUTIVE_FAILURES} "
                f"consecutive failures, cooling down {FAILURE_COOLDOWN_SECONDS}s"
            )
            return Intent.hold(reason="Circuit breaker tripped")

        # ... normal strategy logic ...

    except Exception as e:
        logger.exception(f"Error in decide(): {e}")
        return Intent.hold(reason=f"Error: {e}")

def on_intent_executed(self, intent, success: bool, result):
    if success:
        self.consecutive_failures = 0   # Reset on success
    else:
        self.consecutive_failures += 1
        logger.warning(
            f"Intent failed ({self.consecutive_failures}/{MAX_CONSECUTIVE_FAILURES})"
        )

def get_persistent_state(self) -> dict:
    return {
        "consecutive_failures": self.consecutive_failures,
        "failure_cooldown_until": self.failure_cooldown_until,
    }

def load_persistent_state(self, state: dict) -> None:
    self.consecutive_failures = int(state.get("consecutive_failures", 0))
    self.failure_cooldown_until = float(state.get("failure_cooldown_until", 0))

Important: Only update trade-timing state (e.g.

last_trade_ts

) inside

on_intent_executed

when

success=True

, not when the intent is created. Setting it at creation time means a failed trade still resets the interval timer, causing the strategy to wait before retrying — or worse, to keep retrying on a fixed schedule with no failure awareness.

框架会对每个失败的Intent最多重试

max_retries

次（默认3次），采用指数退避。但是所有重试耗尽后，策略会继续运行，并在下一次迭代中尝试执行同样的交易。如果没有断路器，会产生无限循环的回滚交易，白白消耗gas却不可能成功。

一定要在持久化状态中跟踪连续执行失败次数，达到阈值后停止交易（或者进入较长的冷却期）：

python

MAX_CONSECUTIVE_FAILURES = 3     # 连续3次Intent失败后停止
FAILURE_COOLDOWN_SECONDS = 1800  # 冷却30分钟后再重试

def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)
    self.consecutive_failures = 0
    self.failure_cooldown_until = 0.0

def decide(self, market):
    try:
        now = time.time()

        # 断路器：冷却期内跳过交易
        if now < self.failure_cooldown_until:
            remaining = int(self.failure_cooldown_until - now)
            return Intent.hold(
                reason=f"Circuit breaker active, cooldown {remaining}s remaining"
            )

        # 断路器：失败次数过多进入冷却期
        if self.consecutive_failures >= MAX_CONSECUTIVE_FAILURES:
            self.failure_cooldown_until = now + FAILURE_COOLDOWN_SECONDS
            self.consecutive_failures = 0
            logger.warning(
                f"Circuit breaker tripped after {MAX_CONSECUTIVE_FAILURES} "
                f"consecutive failures, cooling down {FAILURE_COOLDOWN_SECONDS}s"
            )
            return Intent.hold(reason="Circuit breaker tripped")

        # ... 正常策略逻辑 ...

    except Exception as e:
        logger.exception(f"Error in decide(): {e}")
        return Intent.hold(reason=f"Error: {e}")

def on_intent_executed(self, intent, success: bool, result):
    if success:
        self.consecutive_failures = 0   # 成功时重置计数
    else:
        self.consecutive_failures += 1
        logger.warning(
            f"Intent failed ({self.consecutive_failures}/{MAX_CONSECUTIVE_FAILURES})"
        )

def get_persistent_state(self) -> dict:
    return {
        "consecutive_failures": self.consecutive_failures,
        "failure_cooldown_until": self.failure_cooldown_until,
    }

def load_persistent_state(self, state: dict) -> None:
    self.consecutive_failures = int(state.get("consecutive_failures", 0))
    self.failure_cooldown_until = float(state.get("failure_cooldown_until", 0))

重要： 仅在

on_intent_executed

中

success=True

时更新交易时间状态（比如

last_trade_ts

），不要在创建Intent时更新。如果创建时就更新，失败的交易仍然会重置间隔计时器，导致策略等待后才重试，更糟的是会按固定 schedule 不断重试，完全感知不到失败。

Handling Gas and Slippage Errors (Sadflow Hook)

处理Gas和滑点错误（Sadflow钩子）

Override

on_sadflow_enter

to react to specific error types during intent retries. This hook is called before each retry attempt and lets you modify the transaction (e.g. increase gas or slippage) or abort early:

python

from almanak.framework.intents.state_machine import SadflowAction

class MyStrategy(IntentStrategy):
    def on_sadflow_enter(self, error_type, attempt, context):
        # Abort immediately on insufficient funds — retrying won't help
        if error_type == "INSUFFICIENT_FUNDS":
            return SadflowAction.abort("Insufficient funds, stopping retries")

        # Increase gas limit for gas-related 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")

        # For slippage errors ("Too little received"), abort after 1 attempt
        # since retrying with the same parameters will produce the same result
        if error_type == "SLIPPAGE" and attempt >= 1:
            return SadflowAction.abort("Slippage error persists, aborting")

        # Default: let the framework retry with backoff
        return None

Error types passed to

on_sadflow_enter

(from

_categorize_error

state_machine.py

```
GAS_ERROR
```
— gas estimation failed or gas limit exceeded
```
INSUFFICIENT_FUNDS
```
— wallet balance too low
```
SLIPPAGE
```
— "Too little received" or similar DEX revert
```
TIMEOUT
```
— transaction confirmation timed out
```
NONCE_ERROR
```
— nonce mismatch or conflict
```
REVERT
```
— generic transaction revert
```
RATE_LIMIT
```
— RPC or API rate limit hit
```
NETWORK_ERROR
```
— connection or network failure
```
COMPILATION_PERMANENT
```
— unsupported protocol/chain (non-retriable)
```
None
```
— unclassified error

重写

on_sadflow_enter

可以在Intent重试时响应特定的错误类型，这个钩子会在每次重试前调用，让你可以修改交易（比如增加gas或滑点）或者提前中止：

python

from almanak.framework.intents.state_machine import SadflowAction

class MyStrategy(IntentStrategy):
    def on_sadflow_enter(self, error_type, attempt, context):
        # 余额不足直接中止，重试没用
        if error_type == "INSUFFICIENT_FUNDS":
            return SadflowAction.abort("Insufficient funds, stopping retries")

        # gas相关错误增加gas limit
        if error_type == "GAS_ERROR" and context.action_bundle:
            modified = self._increase_gas(context.action_bundle)
            return SadflowAction.modify(modified, reason="Increased gas limit")

        # 滑点错误（「到账金额过少」）尝试1次后中止，因为用同样参数重试结果一样
        if error_type == "SLIPPAGE" and attempt >= 1:
            return SadflowAction.abort("Slippage error persists, aborting")

        # 默认：让框架按退避策略重试
        return None

传入
on_sadflow_enter
的错误类型（来自

state_machine.py

中的

_categorize_error

）：

```
GAS_ERROR
```
— gas预估失败或gas limit超出
```
INSUFFICIENT_FUNDS
```
— 钱包余额不足
```
SLIPPAGE
```
— 「到账金额过少」或类似的DEX回滚
```
TIMEOUT
```
— 交易确认超时
```
NONCE_ERROR
```
— nonce不匹配或冲突
```
REVERT
```
— 通用交易回滚
```
RATE_LIMIT
```
— 触发RPC或API限流
```
NETWORK_ERROR
```
— 连接或网络故障
```
COMPILATION_PERMANENT
```
— 不支持的协议/链（不可重试）
```
None
```
— 未分类错误

Going Live Checklist

上线检查清单

Before deploying to mainnet:

Test on Anvil with
```
--network anvil --once
```
until
```
decide()
```
works correctly
Run
```
--dry-run --once
```
on mainnet to verify compilation without submitting transactions
Use
```
amount=
```
(token units) for swaps if
```
amount_usd=
```
causes reverts (see swap reference above)
Override
```
get_persistent_state()
```
/
```
load_persistent_state()
```
if your strategy tracks positions or phase state
Verify token approvals for all protocols used (auto-handled for most, but verify on first run)
Fund wallet on the correct chain with sufficient tokens plus gas (ETH/AVAX/MATIC)
Note your instance ID after first successful iteration (needed for
```
--id
```
resume)
Start with small amounts and monitor the first few iterations

部署到主网前：

先用
```
--network anvil --once
```
在Anvil上测试，直到
```
decide()
```
逻辑完全正确
在主网运行
```
--dry-run --once
```
，验证编译逻辑正常，不会提交交易
如果
```
amount_usd=
```
导致回滚，兑换交易改用
```
amount=
```
（代币单位）（参考上文兑换参考）
如果策略跟踪仓位或阶段状态，重写
```
get_persistent_state()
```
/
```
load_persistent_state()
```
验证所有用到的协议的代币授权（大部分会自动处理，但首次运行需要验证）
为对应链的钱包充值足够的代币和gas（ETH/AVAX/MATIC）
首次成功迭代后记录实例ID（后续用
```
--id
```
恢复运行需要）
先用小额资金运行，监控前几次迭代

Troubleshooting

问题排查

Error	Cause	Fix
`TokenNotFoundError`	Token symbol not in registry	Use exact symbol (e.g., "WETH" not "ETH" for swaps). Check `resolver.resolve("TOKEN", "chain")` .
`Gateway not available`	Gateway not running	Use `almanak strat run` (auto-starts gateway) or start manually with `almanak gateway` .
`ALMANAK_PRIVATE_KEY not set`	Missing .env	Add `ALMANAK_PRIVATE_KEY=0x...` to your `.env` file.
`Anvil not found`	Foundry not installed	Install: `curl -L https://foundry.paradigm.xyz \| bash && foundryup`
`RSI data unavailable`	Insufficient price history	The gateway needs time to accumulate data. Try a longer timeframe or wait.
`Insufficient balance`	Wallet doesn't have enough tokens	For Anvil: add `anvil_funding` to config.json. For mainnet: fund the wallet.
`Slippage exceeded`	Trade too large or pool illiquid	Increase `max_slippage` or reduce trade size.
`Too little received` (repeated reverts)	Placeholder prices used for slippage calculation, or stale price data	Ensure real price feeds are active (not placeholder). Implement `on_sadflow_enter` to abort on persistent slippage errors. Add a circuit breaker to stop retrying the same failing trade.
Transactions keep reverting after max retries	Strategy re-emits the same failing intent on subsequent iterations	Track `consecutive_failures` in persistent state and enter cooldown after a threshold. See the "Execution Failure Tracking" pattern.
Gas wasted on reverted transactions	No circuit breaker; framework retries 3x per intent, then strategy retries next iteration indefinitely	Implement `on_intent_executed` callback to count failures and `on_sadflow_enter` to abort non-recoverable errors early.
Intent compilation fails	Wrong parameter types	Ensure amounts are `Decimal` , not `float` . Use `Decimal(str(value))` .

错误	原因	修复方案
`TokenNotFoundError`	代币符号不在注册表中	使用准确的符号（比如兑换用"WETH"不是"ETH"），检查 `resolver.resolve("TOKEN", "chain")` 。
`Gateway not available`	Gateway未运行	使用 `almanak strat run` （会自动启动gateway）或者手动运行 `almanak gateway` 启动。
`ALMANAK_PRIVATE_KEY not set`	缺少.env配置	在 `.env` 文件中添加 `ALMANAK_PRIVATE_KEY=0x...` 。
`Anvil not found`	未安装Foundry	安装命令： `curl -L https://foundry.paradigm.xyz \| bash && foundryup`
`RSI data unavailable`	价格历史不足	Gateway需要时间积累数据，尝试更长的时间周期或者等待。
`Insufficient balance`	钱包代币不足	Anvil环境：在config.json中添加 `anvil_funding` ；主网环境：给钱包充值。
`Slippage exceeded`	交易金额太大或者池流动性不足	提高 `max_slippage` 或者减小交易规模。
`Too little received` （反复回滚）	滑点计算用了占位价格，或者价格数据过时	确保真实价格源正常运行（不是占位数据）。实现 `on_sadflow_enter` 在持续滑点错误时中止。添加断路器停止重试相同的失败交易。
达到最大重试次数后交易仍然反复回滚	策略在后续迭代中重复发出相同的失败Intent	在持久化状态中跟踪 `consecutive_failures` ，达到阈值后进入冷却期，参考「执行失败跟踪」模式。
回滚交易浪费gas	没有断路器；框架每个Intent重试3次，之后策略下次迭代会无限重试	实现 `on_intent_executed` 回调统计失败次数，实现 `on_sadflow_enter` 提前中止不可恢复的错误。
Intent编译失败	参数类型错误	确保金额是 `Decimal` 类型，不是 `float` ，使用 `Decimal(str(value))` 转换。

Debugging Tips

调试技巧

Use

--verbose

flag for detailed logging:

almanak strat run --once --verbose

Use
```
--dry-run
```
to test decide() without submitting transactions
Use
```
--log-file out.json
```
for machine-readable JSON logs
Check strategy state:
```
self.state
```
persists between iterations
Paper trade first:
```
almanak strat backtest paper -s my_strategy
```
runs real execution on Anvil

使用

--verbose

参数获取详细日志：

almanak strat run --once --verbose

使用
```
--dry-run
```
测试decide()逻辑，不会提交交易
使用
```
--log-file out.json
```
生成机器可读的JSON日志
检查策略状态：
```
self.state
```
会在迭代之间持久化
先做模拟交易：
```
almanak strat backtest paper -s my_strategy
```
在Anvil上运行真实执行逻辑