Market Snapshot

almanak.framework.market

Canonical MarketSnapshot package — VIB-4062.

The single source of truth for the strategy-facing market-data interface. All strategies and runtime surfaces consume MarketSnapshot through this package; legacy import paths under almanak.framework.strategies and almanak.framework.data.market_snapshot are removed in commit 6 of the VIB-4062 migration.

Public surface:

• MarketSnapshot — the only concrete snapshot class.
• MarketSnapshotBuilder — builder with factories per runtime surface.
• MultiChainMarketSnapshot — type alias for MarketSnapshot (see PRD §4.2).
• Typed errors — MarketSnapshotError, ChainNotConfiguredError, AmbiguousChainError, PriceUnavailableError, …
• Typed return models — TokenBalance, PriceData, RSIData, MACDData, …

AmbiguousChainError

AmbiguousChainError(
    *args: Any, reason: str = "", **fields: Any
)

Bases: MarketSnapshotError

A multi-chain snapshot was queried with chain=None.

Raised when len(self.chains) > 1 and no explicit chain= was given. Multi-chain callers must be explicit; default-to-primary policy hides chain-routing bugs (PRD §4.2 R2, Codex finding HIGH).

ChainNotConfiguredError

ChainNotConfiguredError(
    *args: Any, reason: str = "", **fields: Any
)

Bases: MarketSnapshotError

A specific chain= argument was supplied but is not in chains.

Raised in BOTH single-chain (mismatch with the only configured chain) and multi-chain (chain not in the configured set) cases. The runner has no recovery path — strategy code must update its chain handling.

IndicatorUnavailableError

IndicatorUnavailableError(
    *args: Any, reason: str = "", **fields: Any
)

Bases: MarketSnapshotError

Generic indicator (sma, ema, macd, …) unavailable.

MarketSnapshotError

MarketSnapshotError(
    *args: Any, reason: str = "", **fields: Any
)

Bases: Exception

Base class for typed snapshot errors.

Subclasses carry structured fields. Stringifying gives a stable shape for log lines: "<ClassName>(<key>=<value>, …): <reason>".

For backward compat with the legacy data-layer error idiom, positional arguments are accepted as (chain, reason) or just (reason,); callers using kwargs (the canonical post-VIB-4062 form) continue to work.

StaleDataError

StaleDataError(*args: Any, reason: str = '', **fields: Any)

Bases: MarketSnapshotError

A provider returned data older than the freshness policy permits.

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.

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.

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.

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.

IchimokuData dataclass

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

Ichimoku Cloud data for a token.

IndicatorProvider

IndicatorProvider(
    macd: Callable[..., MACDData] | None = None,
    bollinger: Callable[..., BollingerBandsData]
    | None = None,
    stochastic: Callable[..., StochasticData] | None = None,
    atr: Callable[..., ATRData] | None = None,
    sma: Callable[..., MAData] | None = None,
    ema: Callable[..., MAData] | None = None,
    adx: Callable[..., ADXData] | None = None,
    obv: Callable[..., OBVData] | None = None,
    cci: Callable[..., CCIData] | None = None,
    ichimoku: Callable[..., IchimokuData] | None = None,
)

Provider that wraps all indicator calculators for synchronous access.

Each attribute corresponds to a MarketSnapshot accessor. The runner creates this once and injects it into the strategy, so all indicators work out of the box without strategy authors managing calculators.

MACDData dataclass

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

MACD data for a token.

MAData dataclass

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

Moving Average data for a token.

OBVData dataclass

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

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

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))(),
    source: str = "",
)

Price data for a token.

Includes source (the named provider that produced the datum) so accounting writers can stamp transaction_ledger.price_inputs_json with the real provider name (VIB-3889).

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: ... round(rsi, 2) f"{rsi:.2f}"

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.

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).

MarketSnapshot

MarketSnapshot(
    chain: str | None = None,
    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,
    multi_dex_service: Any | None = None,
    rate_monitor: Any | None = None,
    funding_rate_provider: Any | None = None,
    gateway_client: Any | None = None,
    default_timeframe: str | None = None,
    runtime_surface: str = "unit_test",
    chains: tuple[str, ...] | list[str] | None = None,
    gas_oracle: Any | None = None,
    aave_health_factor_provider: Any | None = None,
    pool_reader_registry: Any | None = None,
    pool_reader: Any | None = None,
    price_aggregator: Any | None = None,
    ohlcv_router: Any | None = None,
    ohlcv_module: Any | None = None,
    pool_history_reader: Any | None = None,
    liquidity_depth_reader: Any | None = None,
    slippage_estimator: Any | None = None,
    pool_analytics_reader: Any | None = None,
    yield_aggregator: Any | None = None,
    il_calculator: Any | None = None,
    volatility_calculator: Any | None = None,
    risk_calculator: Any | None = None,
    rate_history_reader: Any | None = None,
    solana_lst_provider: Any | None = None,
    data_router: Any | 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 | None	Chain name (e.g., "arbitrum", "ethereum")	None
wallet_address	str	Wallet address for balance queries	''
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	PredictionProvider for prediction market data	None
indicator_provider	IndicatorProvider | None	IndicatorProvider for calculator-backed TA indicators	None
multi_dex_service	Any | None	MultiDexService for cross-DEX price comparison	None
rate_monitor	Any | None	RateMonitor instance for lending rate queries	None
funding_rate_provider	Any | None	GatewayFundingRateProvider for perpetual funding rate queries	None
gateway_client	Any | None	Connected GatewayClient for gateway-routed on-chain reads (used by position_health and other methods that need eth_call).	None
default_timeframe	str | None	Default OHLCV timeframe from strategy config (e.g., "15m", "1h"). Used as the default for all indicator methods (rsi, macd, sma, etc.) when no explicit timeframe is passed. Falls back to DEFAULT_TIMEFRAME if not set.	None

chain property

chain: str

Get the (primary) chain name.

For multi-chain snapshots this returns the first chain in self.chains. Strategy code that cares about the active chain in a multi-chain context must pass chain= explicitly to each method (PRD §4.2).

chains property

chains: tuple[str, ...]

Configured chain set (PRD §4.2).

Single-chain snapshots return (self.chain,). Multi-chain snapshots — folded in from MultiChainMarketSnapshot in commit 3 — return all configured chains. Multi-chain helpers consult len(self.chains) to decide whether chain=None defaults to the primary or raises AmbiguousChainError.

wallet_address property

wallet_address: str

Get the wallet address.

timestamp property

timestamp: datetime

Get the snapshot timestamp.

fork_rpc_url property

fork_rpc_url: str | None

Get the Anvil fork RPC URL for on-chain reads (paper trading only).

Returns the fork's JSON-RPC endpoint when running in paper trading mode, allowing strategies to perform protocol-level reads directly against the fork. Returns None when not in paper trading mode.

VIB-1956: Enables strategies to do protocol-level reads (e.g., Aave getReserveData, DEX pool state) during paper trading.

WARNING: This is a paper-trading-only escape hatch. In production, this returns None. Do NOT gate trading logic on fork_rpc_url availability — strategies that behave differently based on this property will diverge between paper trading and production.

fork_block property

fork_block: int | None

Get the current fork block number (paper trading only).

prices property

prices: _PricesAccessor

Hybrid dict-callable accessor for prices.

Two usage shapes (both supported, see ALM-2696 deferred-batch NOTE):

``market.prices.get("WETH")`` — legacy dict-style access on the
internal ``_prices`` map (used by uniswap_rsi-style backtest
adapters that probe ``hasattr(market, "prices")``).

``market.prices(["WETH", "USDC"], chain="arbitrum")`` — batch
fetch via ``price()``, returning ``{symbol: Decimal}``. Per-symbol
failures are logged and skipped.

Returning a hybrid accessor preserves both contracts; making prices a plain method broke hasattr+get callers ('function' object has no attribute 'get'), and making it a plain dict broke the batch-fetch tests.

price

price(
    token: str,
    quote: str = "USD",
    *,
    chain: str | None = None,
) -> 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'
chain	str | None	Optional chain override. chain is keyword-only (PRD §4.2 R1). Strict resolution applies for chain=None (single-chain returns the only chain; multi-chain raises AmbiguousChainError). When chain= is explicitly supplied AND the configured price oracle accepts a chain argument, the value is passed through to the oracle as-is (the oracle is the authoritative chain-routing surface in that case).	None

Returns:

Type	Description
Decimal	Token price in quote currency

Raises:

Type	Description
AmbiguousChainError	If chain is None on a multi-chain snapshot.
ChainNotConfiguredError	If chain is not in this snapshot's configured chains AND the oracle cannot route by chain.
ValueError	If price cannot be determined.

price_data

price_data(
    token: str,
    quote: str = "USD",
    *,
    chain: str | None = None,
) -> 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'
chain	str | None	Optional chain override (keyword-only, PRD §4.2 R1).	None

Returns:

Type	Description
PriceData	PriceData with current price and historical data

Raises:

Type	Description
ChainNotConfiguredError / AmbiguousChainError	same rules as :meth:price.

rsi

rsi(
    token: str,
    period: int = 14,
    timeframe: str | None = None,
) -> 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 | None	OHLCV candle timeframe. Defaults to strategy's data_granularity config, or "4h" if not configured.	None

Returns:

Type	Description
RSIData	RSIData with current RSI value and signal

Raises:

Type	Description
ValueError	If RSI cannot be calculated

price_across_dexs

price_across_dexs(
    token_in: str,
    token_out: str,
    amount: Decimal,
    dexs: list[str] | None = None,
) -> Any

Get prices from multiple DEXs for comparison.

Fetches quotes from all configured DEXs and returns a comparison of prices and execution details.

Parameters:

Name	Type	Description	Default
token_in	str	Input token symbol (e.g., "USDC", "WETH")	required
token_out	str	Output token symbol (e.g., "WETH", "USDC")	required
amount	Decimal	Input amount (human-readable)	required
dexs	list[str] | None	DEXs to query (default: all available on chain)	None

Returns:

Type	Description
Any	MultiDexPriceResult with quotes from each DEX

Raises:

Type	Description
NotImplementedError	If multi-DEX service is not configured

best_dex_price

best_dex_price(
    token_in: str,
    token_out: str,
    amount: Decimal,
    dexs: list[str] | None = None,
) -> Any

Get the best DEX for a trade.

Compares prices from all configured DEXs and returns the one with the highest output amount (best execution).

Parameters:

Name	Type	Description	Default
token_in	str	Input token symbol (e.g., "USDC", "WETH")	required
token_out	str	Output token symbol (e.g., "WETH", "USDC")	required
amount	Decimal	Input amount (human-readable)	required
dexs	list[str] | None	DEXs to compare (default: all available on chain)	None

Returns:

Type	Description
Any	BestDexResult with the best DEX and quote

Raises:

Type	Description
NotImplementedError	If multi-DEX service is not configured

macd

macd(
    token: str,
    fast_period: int = 12,
    slow_period: int = 26,
    signal_period: int = 9,
    timeframe: str | None = None,
) -> 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 | None	OHLCV candle timeframe. Defaults to strategy's data_granularity config, or "4h" if not configured.	None

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 | None = None,
) -> 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 | None	OHLCV candle timeframe. Defaults to strategy's data_granularity config, or "4h" if not configured.	None

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 | None = None,
) -> 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 | None	OHLCV candle timeframe. Defaults to strategy's data_granularity config, or "4h" if not configured.	None

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 | None = None,
) -> 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 | None	OHLCV candle timeframe. Defaults to strategy's data_granularity config, or "4h" if not configured.	None

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 | None = None,
) -> 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 | None	OHLCV candle timeframe. Defaults to strategy's data_granularity config, or "4h" if not configured.	None

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 | None = None,
) -> 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 | None	OHLCV candle timeframe. Defaults to strategy's data_granularity config, or "4h" if not configured.	None

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 | None = None,
) -> 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 | None	OHLCV candle timeframe. Defaults to strategy's data_granularity config, or "4h" if not configured.	None

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 | None = None,
) -> 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 | None	OHLCV candle timeframe. Defaults to strategy's data_granularity config, or "4h" if not configured.	None

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 | None = None,
) -> 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 | None	OHLCV candle timeframe. Defaults to strategy's data_granularity config, or "4h" if not configured.	None

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 | None = None,
) -> 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 | None	OHLCV candle timeframe. Defaults to strategy's data_granularity config, or "4h" if not configured.	None

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"))

invalidate_balance

invalidate_balance(
    token: str, protocol: str | None = None
) -> None

Evict any memoized wallet balance for token so the next :meth:balance call re-queries the provider.

Sequential execution lanes mutate wallet balances mid-snapshot: a teardown staircase's REPAY consumes the debt token before a later amount="all" sweep resolves against this same snapshot, and the stale memo then over-resolves by exactly the repaid amount (compile fails on insufficient balance). The gateway-level cache is already invalidated by the commit pipeline after each intent — this clears the snapshot-level memo so the fresh value is actually read.

No-op on provider-less snapshots (paper / dry-run inject simulated balances directly into the memo maps): with no provider there is no fresher source to re-query, so evicting would turn every subsequent read into a ValueError instead of serving the (correct, simulated) memo.

balance

balance(
    token: str,
    protocol: str | None = None,
    *,
    chain: str | None = None,
    price: Decimal | None = None,
) -> TokenBalance

Get wallet balance for a token.

Parameters:

Name	Type	Description	Default
token	str	Token symbol	required
protocol	str | None	Optional protocol name for variant disambiguation (VIB-3138). When set, resolves generic symbols to the protocol's preferred variant via the connector's settlement_token_variants capability (read through CapabilitiesRegistry; e.g., balance("USDC", protocol="polymarket") on Polygon returns the pUSD balance — the V2 spendable trading collateral wrapped from USDC.e or native USDC at the on-chain Onramp. See VIB-3770). When unset, returns the balance for the symbol as given.	None
chain	str | None	Optional chain override (keyword-only, PRD §4.2 R1). Required on multi-chain snapshots; on single-chain it must match self.chain or be None.	None
price	Decimal | None	Optional already-known USD price (keyword-only, VIB-4843 FR-5003). When supplied, balance_usd is computed as balance * price WITHOUT a re-fetch. When omitted, the warm _price_cache is consulted (still no oracle call); only when neither is available is balance_usd left as the coerced Decimal("0") for callers to fill via price(). This removes the redundant price re-fetch the portfolio valuation lane previously incurred per token.	None

Returns:

Type	Description
TokenBalance	TokenBalance with current balance (and balance_usd filled from
TokenBalance	price / _price_cache when derivable).

Raises:

Type	Description
ChainNotConfiguredError / AmbiguousChainError	same rules as :meth:price.
ValueError	If balance cannot be determined.

clear_critical_data_failures

clear_critical_data_failures() -> None

Clear all tracked critical data failures.

Called by the runner after pre-warming the price cache and before strategy.decide() runs, so that pre-warm failures (which are retried inside decide()) do not incorrectly poison the HOLD-escalation check.

has_critical_data_failures

has_critical_data_failures() -> bool

Return True when this snapshot observed any critical data failures.

critical_data_failure_count

critical_data_failure_count() -> int

Number of currently tracked critical failures for this snapshot.

classify_critical_data_failures

classify_critical_data_failures() -> str

Classify observed data failures as transient, permanent, or mixed.

summarize_critical_data_failures

summarize_critical_data_failures(*, limit: int = 3) -> str

Create a concise summary for logs/lifecycle error messages.

is_quiet_pool_hold

is_quiet_pool_hold() -> bool

Liveness backstop for a DEX pool with no recent swaps.

Returns True iff every recorded critical data failure is a DEX quiet-pool staleness miss and the affected token is still priceable from the 24/7 aggregated oracle.

Rationale: a DEX pool that simply hasn't traded recently returns stale (not absent) trade-derived OHLCV, so swap-based indicators (RSI, MACD, …) can't be computed. But the asset itself is continuously priceable — the pool is alive, just quiet. Holding through that is correct and must not be escalated to a DATA_ERROR (which trips the circuit breaker on a live pool). A genuinely dead/unreachable feed has no oracle price and stays critical, so the dead-pool guard is preserved.

The check is conservative: any non-quiet-pool failure, any unparseable detail, or any token that is not priceable returns False (escalate).

balance_usd

balance_usd(
    token: str, protocol: str | None = None
) -> Decimal

Get wallet balance in USD terms.

Parameters:

Name	Type	Description	Default
token	str	Token symbol	required
protocol	str | None	Optional protocol for variant disambiguation (see balance).	None

Returns:

Type	Description
Decimal	Balance in USD

collateral_value_usd

collateral_value_usd(
    token: str, amount: Decimal
) -> Decimal

Get the USD value of a given amount of collateral.

Convenience helper for perp position sizing. Multiplies the given amount by the token's current price.

Parameters:

Name	Type	Description	Default
token	str	Token symbol (e.g., "WETH", "USDC", "WBTC")	required
amount	Decimal	Token amount in human-readable units (not wei)	required

Returns:

Type	Description
Decimal	USD value as a Decimal

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).

_balance_cache keys are either the bare symbol (legacy single-arg provider path) or f"{symbol}@{chain}" (multi-chain path). Strip the chain suffix before dedup so a token populated via both set_balance(symbol, tb) (which writes both maps) is counted once.

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_or_chain: PriceData | str,
    price_data_or_quote: PriceData | str | None = None,
    quote: str = "USD",
    chain: str | None = None,
) -> None

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

Accepts both call shapes during the VIB-4062 transition:

• set_price_data(token, price_data, quote="USD", chain=None) — canonical strategy-layer shape.
• set_price_data(token, chain, price_data, quote="USD") — legacy multichain shape; chain is the second positional argument.

The ambiguity is resolved by inspecting the type of the second argument: PriceData → canonical; str → legacy multichain.

set_balance

set_balance(
    token: str,
    balance_data_or_chain: Any,
    balance_data: TokenBalance | None = None,
) -> None

Pre-populate balance for a token.

Two call shapes:

1. Canonical (post-VIB-4062): set_balance(token, balance_data).
2. Legacy multichain: set_balance(token, chain, balance_data) — kept so multi-chain runners (runner_teardown's simulated-balances injector, paper-engine cross-chain seeding) keep working.

In the multichain shape, _balances (the chain-agnostic short cache) is updated only when chain == self._chain so cross-chain reads stay distinct.

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"), ))

lending_rate

lending_rate(
    protocol: str,
    token: str,
    side: str = "supply",
    *,
    chain: str | None = None,
) -> Any

Get the lending rate for a specific protocol and token.

Canonical strategy-side accessor for live lending rates (VIB-4859 / W7). Delegates to the gateway's RateHistoryService.GetLendingRateCurrent RPC via :class:almanak.framework.data.rates.RateMonitor — all HTTP / Web3 egress for rate lookups happens server-side via :class:GatewayLendingRateHistoryCapability implementations on the corresponding connectors.

Use this in preference to the deprecated :class:almanak.framework.data.rates.RateMonitor direct surface (caller migration tracked in VIB-4869).

Resolution order:

1. Pre-populated cache (set_lending_rate(...)) — hit first for strategies that inject rates synthetically (backtests / tests).
2. The constructor-injected rate_monitor (legacy path), if set.
3. A lazily-constructed default RateMonitor (calls the gateway).

Parameters:

Name	Type	Description	Default
protocol	str	Protocol identifier ("aave_v3", "compound_v3", "morpho_blue"). Must be registered with GatewayLendingRateHistoryCapability on the gateway side.	required
token	str	Token symbol (e.g. "USDC", "WETH").	required
side	str	Rate side — "supply" (default) or "borrow".	'supply'
chain	str | None	Optional chain override (keyword-only, PRD §4.2 R1). When omitted the snapshot's resolved chain is used.	None

Returns:

Type	Description
Any	class:LendingRate dataclass with apy_percent, apy_ray,
Any	utilization_percent, etc.

Raises:

Type	Description
ChainNotConfiguredError / AmbiguousChainError	same rules as :meth:price.
ValueError	If the gateway returns no rate and no fallback is configured.

Example

rate = market.lending_rate("aave_v3", "USDC", "supply") print(f"Aave USDC Supply APY: {rate.apy_percent:.2f}%")

best_lending_rate

best_lending_rate(
    token: str,
    side: str = "supply",
    protocols: list[str] | None = None,
    *,
    chain: str | None = None,
) -> Any

Get the best lending rate for a token across protocols.

Canonical strategy-side accessor for the best live lending rate (VIB-4869), symmetric with :meth:lending_rate. For supply rates, returns highest APY; for borrow rates, returns lowest APY.

Resolution order mirrors :meth:lending_rate:

1. The constructor-injected rate_monitor (legacy path), if set — preserves test surfaces that mock the monitor.
2. A lazily-constructed framework-internal RateMonitor (_internal=True) that fans out to the gateway. This keeps direct instantiations (README-style flows, unit / backtest harnesses calling create_market_snapshot() directly) working without an injected monitor, instead of unconditionally raising.

As with :meth:lending_rate, the lazy lane queries the gateway directly and raises loudly when the gateway is unreachable rather than silently substituting placeholder rates — surfacing hardcoded numbers for a strategy that asked for the best live rate is a bigger footgun than raising.

Parameters:

Name	Type	Description	Default
token	str	Token symbol (e.g., "USDC", "WETH")	required
side	str	Rate side - "supply" or "borrow" (default "supply")	'supply'
protocols	list[str] | None	Protocols to compare (default: all available on chain)	None
chain	str | None	Optional chain override (keyword-only). When omitted the snapshot's resolved chain is used.	None

Returns:

Type	Description
Any	BestRateResult with best_rate, all_rates, etc.

Raises:

Type	Description
ValueError	If no monitor is injected and the gateway is unavailable, or if the best-rate lookup otherwise fails.

Example

result = market.best_lending_rate("USDC", "supply") if result.best_rate: print(f"Best: {result.best_rate.protocol} at {result.best_rate.apy_percent:.2f}%")

set_lending_rate

set_lending_rate(
    protocol: str, token: str, side: str, rate: Any
) -> None

Pre-populate a lending rate for a protocol/token/side.

Useful for backtesting and testing where you want to inject known rates without needing a live RateMonitor.

Parameters:

Name	Type	Description	Default
protocol	str	Protocol identifier (e.g., "aave_v3")	required
token	str	Token symbol (e.g., "USDC")	required
side	str	Rate side ("supply" or "borrow")	required
rate	Any	LendingRate dataclass instance	required

Example

from almanak.framework.data.rates import LendingRate market.set_lending_rate("aave_v3", "USDC", "supply", LendingRate( protocol="aave_v3", token="USDC", side="supply", apy_ray=Decimal("0"), apy_percent=Decimal("4.25"), ))

position_health

position_health(
    protocol: str,
    market_id: str,
    rpc_url: str | None = None,
    collateral_price_usd: Decimal | None = None,
    debt_price_usd: Decimal | None = None,
) -> Any

Get health factor for a lending position.

Reads on-chain position data and computes the health factor for Aave V3, Morpho Blue, or Compound V3 positions. Uses gateway-routed eth_calls when a gateway_client was wired into this MarketSnapshot.

Parameters:

Name	Type	Description	Default
protocol	str	"aave_v3", "morpho_blue", or "compound_v3"	required
market_id	str	Protocol-specific market identifier. For Aave V3 this is informational (one pool per chain). For Morpho Blue this is the bytes32 market id. For Compound V3 this is the Comet market key (e.g. "usdc", "weth").	required
rpc_url	str | None	Optional explicit RPC URL. Strategies should leave this None so the gateway-routed path is used. Paper-trading code may pass a local anvil URL.	None
collateral_price_usd	Decimal | None	Optional override for collateral price (Morpho cross-asset markets require this).	None
debt_price_usd	Decimal | None	Optional override for debt-token price (same).	None

Returns:

Type	Description
Any	PositionHealth with .health_factor (Decimal), .collateral_value_usd,
Any	.debt_value_usd, .max_borrow_usd, .protocol, .market_id.

Raises:

Type	Description
ValueError	If health data cannot be retrieved.

set_position_health

set_position_health(
    protocol: str, market_id: str, health: Any
) -> None

Pre-populate position health for tests and backtesting.

Useful where you want to inject a known PositionHealth without needing a live gateway_client. Mirrors set_lending_rate.

Parameters:

Name	Type	Description	Default
protocol	str	Protocol identifier (e.g., "aave_v3")	required
market_id	str	Protocol-specific market identifier	required
health	Any	PositionHealth dataclass instance	required

aave_health_factor

aave_health_factor(
    *, chain: str | None = None
) -> Decimal | None

Aave V3 account health factor for chain via the wired provider.

Returns None when no aave_health_factor_provider was wired into this snapshot (the default multi-chain CLI path wires only a balance provider) OR when the provider reports no live Aave position. Callers treat None as "no confirmed live position" and must not infer a healthy or unhealthy position from it. This mirrors the None-on-no-provider convention of :meth:prediction_price / :meth:wallet_activity so decide() can branch with if hf is None (see docs/internal/blueprints/01-data-layer.md §MarketSnapshot — provider-driven methods).

The provider (an AaveHealthFactorProvider — a (chain) -> Decimal | None callable) owns any gateway-routed read; this accessor is pure delegation and opens no network connection itself, so it does not cross the gateway boundary. Exceptions raised by the provider propagate unchanged: a gateway failure must surface, never be silently coerced to None and mistaken for "no position" — the leverage-stacking guard in cross-chain strategies depends on that distinction.

Parameters:

Name	Type	Description	Default
chain	str | None	Chain to query (keyword-only, PRD §4.2). Required on a multi-chain snapshot; on a single-chain snapshot it must match self.chain or be None.	None

Returns:

Type	Description
Decimal | None	The Aave V3 health factor as a Decimal, or None when no
Decimal | None	provider is wired / no live position exists.

funding_rate

funding_rate(venue: str, market: str) -> FundingRate

Get the current funding rate for a perpetual market on a specific venue.

Parameters:

Name	Type	Description	Default
venue	str	Venue identifier (e.g., "gmx_v2", "hyperliquid")	required
market	str	Market symbol (e.g., "ETH-USD")	required

Returns:

Type	Description
FundingRate	FundingRate dataclass with rate_hourly, rate_8h, rate_annualized, etc.

Raises:

Type	Description
ValueError	If no funding rate provider is configured or venue is unsupported

funding_rate_spread

funding_rate_spread(
    market: str, venue_a: str, venue_b: str
) -> FundingRateSpread

Get the funding rate spread between two venues.

Parameters:

Name	Type	Description	Default
market	str	Market symbol (e.g., "ETH-USD")	required
venue_a	str	First venue identifier	required
venue_b	str	Second venue identifier	required

Returns:

Type	Description
FundingRateSpread	FundingRateSpread dataclass with spread_hourly, spread_annualized, rate_a, rate_b

Raises:

Type	Description
ValueError	If no funding rate provider is configured or venue is unsupported

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(
    with_sources: bool = False,
) -> dict[str, Any]

Get all prices as a dict suitable for IntentCompiler.

Combines pre-populated prices and cached prices from oracle calls. Keys are normalized to uppercase to match Token.symbol (which is always uppercased by Token.post_init). This prevents case-mismatch lookup failures for mixed-case tokens like cbETH, wstETH, crvUSD, sUSDe, etc.

Parameters:

Name	Type	Description	Default
with_sources	bool	When True (VIB-3889), return the canonical nested shape {symbol: {price_usd, oracle_source, fetched_at, confidence}} so downstream writers (transaction_ledger.price_inputs_json) carry the actual provider name rather than "unknown". Default False preserves the legacy flat {symbol: price} return.	False

Returns:

Type	Description
dict[str, Any]	Flat dict[str, Decimal] (default) or nested
dict[str, Any]	dict[str, dict] when with_sources=True.

gas_price

gas_price(chain: str | None = None) -> Any

Get current gas price for a chain (data-layer port).

Returns GasPrice from the configured _gas_oracle (provider attribute). Caches per _gas_cache_ttl_seconds (default 12s).

Raises:

Type	Description
GasUnavailableError	If the gas price cannot be fetched.
ValueError	If no gas oracle is configured.

estimate_swap_gas_cost_usd

estimate_swap_gas_cost_usd(
    chain: str | None = None,
) -> Decimal

Estimate USD cost of a typical DEX swap on the given chain.

Returns Decimal("0") when the gas oracle is unconfigured or the underlying estimated_cost_usd is zero — strategy authors who need a fail-closed gate should call gas_price() directly and handle the typed errors.

is_trade_worthwhile

is_trade_worthwhile(
    amount_usd: Decimal,
    chain: str | None = None,
    max_gas_ratio: Decimal = Decimal("0.05"),
) -> bool

Whether amount_usd is worth paying gas for on chain.

Fail-open: if a gas estimate cannot be obtained (oracle missing, returns 0, or raises GasUnavailableError), returns True so the strategy is not silently halted by a transient infrastructure issue.

pool_price

pool_price(
    pool_address: str, chain: str | None = None
) -> DataEnvelope[PoolPrice]

Get the live price from an on-chain DEX pool.

Reads slot0() from the pool contract and decodes sqrtPriceX96 into a human-readable price. Classification is EXECUTION_GRADE (fail-closed, no off-chain fallback).

Parameters:

Name	Type	Description	Default
pool_address	str	Pool contract address.	required
chain	str | None	Chain name. Defaults to this snapshot's chain.	None

Returns:

Type	Description
DataEnvelope[PoolPrice]	DataEnvelope[PoolPrice] with provenance metadata.

Raises:

Type	Description
ValueError	If no pool reader registry is configured.
PoolPriceUnavailableError	If the pool price cannot be retrieved.

pool_price_by_pair

pool_price_by_pair(
    token_a: str,
    token_b: str,
    chain: str | None = None,
    protocol: str | None = None,
    fee_tier: int = 3000,
) -> DataEnvelope[PoolPrice]

Get the live pool price for a token pair.

Resolves the pool address for the given pair and reads the price.

Parameters:

Name	Type	Description	Default
token_a	str	Token A symbol or address.	required
token_b	str	Token B symbol or address.	required
chain	str | None	Chain name. Defaults to this snapshot's chain.	None
protocol	str | None	Protocol name (e.g. "uniswap_v3"). If None, tries all registered protocols.	None
fee_tier	int	Fee tier in basis points (default 3000 = 0.3%).	3000

Returns:

Type	Description
DataEnvelope[PoolPrice]	DataEnvelope[PoolPrice] with provenance metadata.

Raises:

Type	Description
ValueError	If no pool reader registry is configured.
PoolPriceUnavailableError	If the pool cannot be found or the price cannot be read.

pool_reserves

pool_reserves(
    pool_address: str, chain: str | None = None
) -> PoolReserves

Get DEX pool reserves and state.

Fetches the current state of a DEX liquidity pool from the blockchain.

Parameters:

Name	Type	Description	Default
pool_address	str	Pool contract address.	required
chain	str | None	Chain identifier. Defaults to this snapshot's chain.	None

Returns:

Type	Description
PoolReserves	PoolReserves dataclass with reserves, fee tier, sqrtPrice, etc.

Raises:

Type	Description
ValueError	If no pool reader is configured.
PoolReservesUnavailableError	If pool data cannot be retrieved.

twap

twap(
    token_pair: str | Instrument,
    chain: str | None = None,
    window_seconds: int = 300,
    pool_address: str | None = None,
    protocol: str = "uniswap_v3",
    token0_decimals: int | None = None,
    token1_decimals: int | None = None,
) -> DataEnvelope[AggregatedPrice]

Get the time-weighted average price (TWAP) for a token pair.

Uses the Uniswap V3 oracle's observe() function to compute the TWAP over the specified time window. Classification: EXECUTION_GRADE (fail-closed, no off-chain fallback).

Parameters:

Name	Type	Description	Default
token_pair	str | Instrument	Token pair as "BASE/QUOTE" string or Instrument.	required
chain	str | None	Chain name. Defaults to this snapshot's chain.	None
window_seconds	int	Time window in seconds (default 300 = 5 min).	300
pool_address	str | None	Explicit pool address. If None, resolves from pair.	None
protocol	str	Protocol to use (default "uniswap_v3").	'uniswap_v3'
token0_decimals	int | None	Decimals for token0. If omitted and pool_address is given, the registry is consulted to resolve pool metadata; if neither is available, raises ValueError.	None
token1_decimals	int | None	Decimals for token1. Same fallback as above.	None

Returns:

Type	Description
DataEnvelope[AggregatedPrice]	DataEnvelope[AggregatedPrice] with TWAP price and provenance.

Raises:

Type	Description
ValueError	If no price aggregator is configured, or if token decimals cannot be derived for an explicit pool_address.
PoolPriceUnavailableError	If TWAP cannot be calculated.

lwap

lwap(
    token_pair: str | Instrument,
    chain: str | None = None,
    fee_tiers: list[int] | None = None,
    protocols: list[str] | None = None,
) -> DataEnvelope[AggregatedPrice]

Get the liquidity-weighted average price (LWAP) for a token pair.

Reads live prices from all known pools for the pair, filters by minimum liquidity, and computes a liquidity-weighted average.

Parameters:

Name	Type	Description	Default
token_pair	str | Instrument	Token pair as "BASE/QUOTE" string or Instrument.	required
chain	str | None	Chain name. Defaults to this snapshot's chain.	None
fee_tiers	list[int] | None	Fee tiers to search (default: [100, 500, 3000, 10000]).	None
protocols	list[str] | None	Protocols to search (default: all registered for chain).	None

Returns:

Type	Description
DataEnvelope[AggregatedPrice]	DataEnvelope[AggregatedPrice] with LWAP price and provenance.

Raises:

Type	Description
ValueError	If no price aggregator is configured.
PoolPriceUnavailableError	If LWAP cannot be calculated.

pool_history

pool_history(
    pool_address: str,
    chain: str | None = None,
    start_date: datetime | None = None,
    end_date: datetime | None = None,
    resolution: str = "1h",
    *,
    protocol: str,
) -> DataEnvelope[list[PoolSnapshot]]

Get historical pool state snapshots for backtesting and analytics.

Parameters:

Name	Type	Description	Default
pool_address	str	Pool contract address.	required
chain	str | None	Chain name. Defaults to this snapshot's chain.	None
start_date	datetime | None	Start of the history window. Defaults to 90 days before end_date.	None
end_date	datetime | None	End of the history window. Defaults to the snapshot's iteration timestamp (NOT datetime.now()) so backtests, paper runs, and historical snapshots stay deterministic and never leak future data.	None
resolution	str	"1h" / "4h" / "1d". Default "1h".	'1h'
protocol	str	REQUIRED keyword-only. Protocol slug — e.g. "uniswap_v3", "aerodrome". NO default — closes the silent cross-protocol surface flagged by VIB-4755 Phase 0b Round-4: a defaulted protocol="uniswap_v3" would let a caller on a Base Aerodrome pool address get CoinGecko Onchain- served Aerodrome data labelled as uniswap_v3 in any audit log. Forgetting this kwarg raises TypeError at the framework boundary BEFORE any gateway round-trip. See docs/internal/uat-cards/VIB-4755.md §D-2.	required

Returns:

Type	Description
DataEnvelope[list[PoolSnapshot]]	DataEnvelope[list[PoolSnapshot]] with INFORMATIONAL classification.

Raises:

Type	Description
ValueError	If no pool history reader is configured.
PoolHistoryUnavailableError	If historical data cannot be retrieved.
TypeError	If protocol is omitted (Python's missing-required- keyword-only signal).

liquidity_depth

liquidity_depth(
    pool_address: str, chain: str | None = None
) -> DataEnvelope[LiquidityDepth]

Get tick-level liquidity depth for a concentrated-liquidity pool.

Parameters:

Name	Type	Description	Default
pool_address	str	Pool contract address.	required
chain	str | None	Chain name. Defaults to this snapshot's chain.	None

Returns:

Type	Description
DataEnvelope[LiquidityDepth]	DataEnvelope[LiquidityDepth] with tick-level liquidity data.

Raises:

Type	Description
ValueError	If no liquidity depth reader is configured.
LiquidityDepthUnavailableError	If liquidity data cannot be read.

estimate_slippage

estimate_slippage(
    token_in: str,
    token_out: str,
    amount: Decimal,
    chain: str | None = None,
    protocol: str | None = None,
) -> DataEnvelope[SlippageEstimate]

Estimate price impact and slippage for a potential swap.

Simulates the swap through tick ranges using actual on-chain liquidity.

Parameters:

Name	Type	Description	Default
token_in	str	Input token symbol or address.	required
token_out	str	Output token symbol or address.	required
amount	Decimal	Amount of token_in to swap (human-readable units).	required
chain	str | None	Chain name. Defaults to this snapshot's chain.	None
protocol	str | None	Protocol name. Auto-detected if None.	None

Returns:

Type	Description
DataEnvelope[SlippageEstimate]	DataEnvelope[SlippageEstimate] with price impact data.

Raises:

Type	Description
ValueError	If no slippage estimator is configured.
SlippageEstimateUnavailableError	If slippage cannot be estimated.

pool_analytics

pool_analytics(
    pool_address: str,
    chain: str | None = None,
    protocol: str | None = None,
) -> DataEnvelope[PoolAnalytics]

Get real-time analytics for a pool (TVL, volume, fee APR/APY).

Parameters:

Name	Type	Description	Default
pool_address	str	Pool contract address.	required
chain	str | None	Chain name. Defaults to this snapshot's chain.	None
protocol	str | None	Optional protocol hint (e.g. "uniswap_v3").	None

Returns:

Type	Description
DataEnvelope[PoolAnalytics]	DataEnvelope[PoolAnalytics] with INFORMATIONAL classification.

Raises:

Type	Description
ValueError	If no pool analytics reader is configured.
PoolAnalyticsUnavailableError	If analytics cannot be retrieved.

best_pool

best_pool(
    token_a: str,
    token_b: str,
    chain: str | None = None,
    metric: str = "fee_apr",
    protocols: list[str] | None = None,
) -> DataEnvelope[PoolAnalyticsResult]

Find the best pool for a token pair based on a metric.

Parameters:

Name	Type	Description	Default
token_a	str	First token symbol.	required
token_b	str	Second token symbol.	required
chain	str | None	Chain name. Defaults to this snapshot's chain.	None
metric	str	"fee_apr" / "fee_apy" / "tvl_usd" / "volume_24h_usd".	'fee_apr'
protocols	list[str] | None	Optional list of protocols to filter by.	None

Returns:

Type	Description
DataEnvelope[PoolAnalyticsResult]	DataEnvelope[PoolAnalyticsResult] with the best pool.

Raises:

Type	Description
ValueError	If no pool analytics reader is configured.
PoolAnalyticsUnavailableError	If no pools found or all providers fail.

yield_opportunities

yield_opportunities(
    token: str,
    chains: list[str] | None = None,
    min_tvl: float = 100000,
    sort_by: str = "apy",
) -> DataEnvelope[list[YieldOpportunity]]

Find yield opportunities for a token across protocols and chains.

Parameters:

Name	Type	Description	Default
token	str	Token symbol.	required
chains	list[str] | None	Optional list of chains to filter. None means all.	None
min_tvl	float	Minimum TVL in USD. Default $100k.	100000
sort_by	str	"apy" / "tvl" / "risk_score". Default "apy".	'apy'

Returns:

Type	Description
DataEnvelope[list[YieldOpportunity]]	DataEnvelope[list[YieldOpportunity]] sorted by chosen metric.

Raises:

Type	Description
ValueError	If no yield aggregator is configured.
YieldOpportunitiesUnavailableError	If data cannot be retrieved.

lending_rate_history

lending_rate_history(
    protocol: str,
    token: str,
    chain: str | None = None,
    days: int = 90,
) -> DataEnvelope[list[LendingRateSnapshot]]

Get historical lending rate snapshots for backtesting.

Parameters:

Name	Type	Description	Default
protocol	str	Lending protocol (e.g. "aave_v3").	required
token	str	Token symbol.	required
chain	str | None	Chain name. Defaults to this snapshot's chain.	None
days	int	Number of days of history. Default 90.	90

Returns:

Type	Description
DataEnvelope[list[LendingRateSnapshot]]	DataEnvelope[list[LendingRateSnapshot]] sorted ascending.

Raises:

Type	Description
ValueError	If no rate history reader is configured.
LendingRateHistoryUnavailableError	If data cannot be retrieved.

funding_rate_history

funding_rate_history(
    venue: str, market_symbol: str, hours: int = 168
) -> DataEnvelope[list[FundingRateSnapshot]]

Get historical funding rate snapshots for backtesting.

Parameters:

Name	Type	Description	Default
venue	str	Perps venue (e.g. "hyperliquid").	required
market_symbol	str	Market symbol (e.g. "ETH-USD").	required
hours	int	Number of hours of history. Default 168 (7 days).	168

Returns:

Type	Description
DataEnvelope[list[FundingRateSnapshot]]	DataEnvelope[list[FundingRateSnapshot]] sorted ascending.

Raises:

Type	Description
ValueError	If no rate history reader is configured.
FundingRateHistoryUnavailableError	If data cannot be retrieved.

il_exposure

il_exposure(
    position_id: str, fees_earned: Decimal = Decimal("0")
) -> ILExposure

Get impermanent loss exposure for a tracked LP position.

Requires an ILCalculator with the position already registered via add_position().

Parameters:

Name	Type	Description	Default
position_id	str	Unique identifier for the LP position.	required
fees_earned	Decimal	Optional fees earned (for net PnL).	Decimal('0')

Returns:

Type	Description
ILExposure	ILExposure with current IL metrics.

Raises:

Type	Description
ValueError	If no IL calculator is configured.
ILExposureUnavailableError	If exposure cannot be calculated.

projected_il

projected_il(
    token_a: str,
    token_b: str,
    price_change_pct: Decimal,
    weight_a: Decimal = Decimal("0.5"),
    weight_b: Decimal = Decimal("0.5"),
) -> ProjectedILResult

Project impermanent loss for a hypothetical price change.

Parameters:

Name	Type	Description	Default
token_a	str	Symbol of token A (the volatile token).	required
token_b	str	Symbol of token B (often a stablecoin).	required
price_change_pct	Decimal	Price change percentage (50 for +50%).	required
weight_a	Decimal	Weight of token A in the pool (default 0.5).	Decimal('0.5')
weight_b	Decimal	Weight of token B in the pool (default 0.5).	Decimal('0.5')

Returns:

Type	Description
ProjectedILResult	ProjectedILResult with il_ratio / il_percent / il_bps.

Raises:

Type	Description
ValueError	If no IL calculator is configured or invalid args.

realized_vol

realized_vol(
    token: str,
    window_days: int = 30,
    timeframe: str = "1h",
    estimator: str = "close_to_close",
    *,
    ohlcv_limit: int | None = None,
) -> DataEnvelope[VolatilityResult]

Calculate realized volatility for a token.

Parameters:

Name	Type	Description	Default
token	str	Token symbol.	required
window_days	int	Lookback window in calendar days. Default 30.	30
timeframe	str	Candle timeframe (1m, 5m, 15m, 1h, 4h, 1d). Default "1h".	'1h'
estimator	str	"close_to_close" (default) or "parkinson".	'close_to_close'
ohlcv_limit	int | None	Override for number of candles to fetch.	None

Returns:

Type	Description
DataEnvelope[VolatilityResult]	DataEnvelope[VolatilityResult] with INFORMATIONAL classification.

Raises:

Type	Description
ValueError	If no volatility calculator is configured.
VolatilityUnavailableError	If volatility cannot be calculated.

vol_cone

vol_cone(
    token: str,
    windows: list[int] | None = None,
    timeframe: str = "1h",
    estimator: str = "close_to_close",
    *,
    ohlcv_limit: int | None = None,
) -> DataEnvelope[VolConeResult]

Compute volatility cone: current vol vs historical percentile.

Parameters:

Name	Type	Description	Default
token	str	Token symbol.	required
windows	list[int] | None	Lookback windows in days. Default [7, 14, 30, 90].	None
timeframe	str	Candle timeframe. Default "1h".	'1h'
estimator	str	"close_to_close" or "parkinson".	'close_to_close'
ohlcv_limit	int | None	Override for number of candles to fetch.	None

Returns:

Type	Description
DataEnvelope[VolConeResult]	DataEnvelope[VolConeResult] with INFORMATIONAL classification.

Raises:

Type	Description
ValueError	If no volatility calculator is configured.
VolConeUnavailableError	If vol cone cannot be calculated.

portfolio_risk

portfolio_risk(
    pnl_series: list[float],
    total_value_usd: Decimal | None = None,
    return_interval: str = "1d",
    risk_free_rate: Decimal = Decimal("0"),
    var_method: str = "parametric",
    timestamps: list[datetime] | None = None,
    benchmark_eth_returns: list[float] | None = None,
    benchmark_btc_returns: list[float] | None = None,
) -> DataEnvelope[PortfolioRisk]

Calculate portfolio risk metrics from a PnL return series.

Computes Sharpe ratio, Sortino ratio, VaR, CVaR, and drawdown.

Parameters:

Name	Type	Description	Default
pnl_series	list[float]	Periodic returns as fractions (0.01 = 1% gain).	required
total_value_usd	Decimal | None	Current portfolio value in USD.	None
return_interval	str	Periodicity of returns (1d, 1h, etc.).	'1d'
risk_free_rate	Decimal	Risk-free rate per period as a decimal.	Decimal('0')
var_method	str	"parametric" / "historical" / "cornish_fisher".	'parametric'
timestamps	list[datetime] | None	Optional timestamps for each return.	None
benchmark_eth_returns	list[float] | None	Optional ETH returns for beta.	None
benchmark_btc_returns	list[float] | None	Optional BTC returns for beta.	None

Returns:

Type	Description
DataEnvelope[PortfolioRisk]	DataEnvelope[PortfolioRisk] with INFORMATIONAL classification.

Raises:

Type	Description
ValueError	If no risk calculator is configured.
PortfolioRiskUnavailableError	If risk metrics cannot be calculated.

rolling_sharpe

rolling_sharpe(
    pnl_series: list[float],
    window_days: int = 30,
    return_interval: str = "1d",
    risk_free_rate: Decimal = Decimal("0"),
    timestamps: list[datetime] | None = None,
) -> DataEnvelope[RollingSharpeResult]

Compute rolling Sharpe ratio over a PnL series.

Parameters:

Name	Type	Description	Default
pnl_series	list[float]	Periodic returns as fractions.	required
window_days	int	Rolling window in days. Default 30.	30
return_interval	str	Periodicity of returns.	'1d'
risk_free_rate	Decimal	Risk-free rate per period.	Decimal('0')
timestamps	list[datetime] | None	Optional timestamps aligned with pnl_series.	None

Returns:

Type	Description
DataEnvelope[RollingSharpeResult]	DataEnvelope[RollingSharpeResult] with INFORMATIONAL classification.

Raises:

Type	Description
ValueError	If no risk calculator is configured.
RollingSharpeUnavailableError	If rolling Sharpe cannot be computed.

ohlcv

ohlcv(
    token: str | Instrument,
    timeframe: str = "1h",
    limit: int = 100,
    quote: str = "USD",
    gap_strategy: GapStrategy = "nan",
    *,
    pool_address: str | None = None,
) -> pd.DataFrame

Get OHLCV (candlestick) data for a token.

Routes through ohlcv_router (preferred) or legacy ohlcv_module.

Parameters:

Name	Type	Description	Default
token	str | Instrument	Token symbol, "BASE/QUOTE" string, or Instrument.	required
timeframe	str	Candle timeframe (1m, 5m, 15m, 1h, 4h, 1d). Default "1h".	'1h'
limit	int	Maximum number of candles. Default 100.	100
quote	str	Quote currency. Default "USD".	'USD'
gap_strategy	GapStrategy	"nan" / "ffill" / "drop". Default "nan".	'nan'
pool_address	str | None	Explicit pool address for DEX providers.	None

Returns:

Type	Description
DataFrame	pandas DataFrame with columns timestamp, open, high, low, close, volume.

Raises:

Type	Description
ValueError	If no OHLCV module/router is configured.
OHLCVUnavailableError	If OHLCV data cannot be retrieved.

health

health() -> HealthReport

Get a health report for all registered data providers.

Returns:

Type	Description
HealthReport	HealthReport with per-source health, cache stats, overall status.

prediction

prediction(market_id: str) -> PredictionMarket

Get prediction market data.

Parameters:

Name	Type	Description	Default
market_id	str	Prediction market ID or URL slug.	required

Returns:

Type	Description
PredictionMarket	PredictionMarket with full market details.

Raises:

Type	Description
ValueError	If no prediction provider is configured.
PredictionUnavailableError	If market data cannot be retrieved.

prediction_positions

prediction_positions(
    market_id: str | None = None,
) -> list[PredictionPosition]

Get all open prediction market positions.

Parameters:

Name	Type	Description	Default
market_id	str | None	Optional market ID or slug to filter by.	None

Returns:

Type	Description
list[PredictionPosition]	List of PredictionPosition objects.

Raises:

Type	Description
ValueError	If no prediction provider is configured.
PredictionUnavailableError	If positions cannot be retrieved.

prediction_orders

prediction_orders(
    market_id: str | None = None,
) -> list[PredictionOrder]

Get all open prediction market orders.

Parameters:

Name	Type	Description	Default
market_id	str | None	Optional market ID or slug to filter by.	None

Returns:

Type	Description
list[PredictionOrder]	List of PredictionOrder objects.

Raises:

Type	Description
ValueError	If no prediction provider is configured.
PredictionUnavailableError	If orders cannot be retrieved.

lst_exchange_rate

lst_exchange_rate(symbol: str) -> LSTExchangeRate

Get Solana LST exchange rate vs SOL.

Parameters:

Name	Type	Description	Default
symbol	str	LST symbol (e.g. "jitoSOL", "mSOL", "bSOL", "INF").	required

Returns:

Type	Description
LSTExchangeRate	LSTExchangeRate with rate vs SOL and APY data.

Raises:

Type	Description
ValueError	If no LST provider is configured.
LSTDataUnavailableError	If data cannot be retrieved.

lst_all_rates

lst_all_rates() -> dict[str, LSTExchangeRate]

Get exchange rates for all tracked Solana LSTs.

Returns:

Type	Description
dict[str, LSTExchangeRate]	dict mapping symbol -> LSTExchangeRate.

Raises:

Type	Description
ValueError	If no LST provider is configured.
LSTDataUnavailableError	If data cannot be retrieved.

pt_position_health

pt_position_health(
    morpho_market_id: str,
    principal_token_market_address: str | None = None,
    rpc_url: str | None = None,
    collateral_price_usd: Decimal | None = None,
    debt_price_usd: Decimal | None = None,
    *,
    principal_token_protocol: str | None = None,
    pendle_market_address: str | None = None,
) -> PTPositionHealth

Get extended health data for a PT-collateral position.

Combines Morpho Blue position data with principal-token market metrics (implied APY, maturity risk) for comprehensive risk assessment.

Parameters:

Name	Type	Description	Default
morpho_market_id	str	Morpho Blue market ID.	required
principal_token_market_address	str | None	Principal-token market address for the PT collateral.	None
rpc_url	str | None	RPC endpoint (uses gateway-routed path when None).	None
collateral_price_usd	Decimal | None	Override for PT collateral price.	None
debt_price_usd	Decimal | None	Override for debt token price.	None
principal_token_protocol	str | None	Optional connector protocol key. When omitted, the sole registered principal-token reader is used for backward compatibility.	None
pendle_market_address	str | None	Deprecated alias for principal_token_market_address.	None

Returns:

Type	Description
PTPositionHealth	PTPositionHealth with Morpho + principal-token risk metrics.

Raises:

Type	Description
HealthUnavailableError	If health data cannot be retrieved or if neither a connected GatewayClient nor an explicit rpc_url is available (the provider has no transport otherwise — fail fast with an actionable contract error instead of letting a downstream provider call surface a less specific exception).

to_dict

to_dict() -> dict[str, Any]

Convert snapshot to dictionary.

seed_price

seed_price(
    token: str, price: Decimal | float | int
) -> None

Seed a scalar price into the snapshot's cache.

Goes through the same internal chokepoint as set_price so the validation behavior is identical.

seed_price_data

seed_price_data(
    token: str,
    data: PriceData,
    *,
    quote: str = "USD",
    chain: str | None = None,
) -> None

Seed a full PriceData record (with source / timestamp / block).

chain is keyword-only and lets multi-chain tests / backtests seed different PriceData for the same symbol on different chains — matches the legacy multichain set_price_data shape.

seed_balance

seed_balance(token: str, balance: TokenBalance) -> None

Seed a TokenBalance for a token.

seed_rsi

seed_rsi(
    token: str, data: RSIData, timeframe: str | None = None
) -> None

Seed an RSIData instance (timeframe-keyed).

MarketSnapshotBuilder

Factory class — never instantiated. All entry points are classmethods.

for_strategy_runner classmethod

for_strategy_runner(
    *,
    strategy: Any,
    runtime_context: Any | None = None,
    gateway_client: Any | None = None,
    chain: str | None = None,
    wallet_address: str | None = None,
    runtime_surface: str | None = None,
    chains: tuple[str, ...] | list[str] | None = None,
    multi_chain_price_oracle: Any | None = None,
    multi_chain_balance_provider: Any | None = None,
    aave_health_factor_provider: Any | None = None,
    default_timeframe: str | None = None,
) -> MarketSnapshot

Build a snapshot for the live / hosted runner.

Pulls providers off the strategy's wired services (price oracle, balance provider, indicator provider, …) and wraps the gateway client into the snapshot. Returns a snapshot with runtime_surface="local_sdk" or "hosted" based on runtime_context (the caller resolves which is which — the snapshot does not read env vars).

For multi-chain strategies the caller passes chains= and the multi_chain_* providers; the resulting snapshot has the same canonical class as a single-chain build.

for_pnl_backtest_state classmethod

for_pnl_backtest_state(
    *, chain: str, wallet_address: str, state: Any
) -> MarketSnapshot

Build a snapshot wired to a PnL backtest engine's state.

The state object exposes price_oracle / balance_provider / indicator_provider interfaces; the builder forwards them as-is.

VIB-4727: backtest factories inject NullPoolAnalyticsReader so any market.pool_analytics(...) call raises DataSourceUnavailable deterministically. Live gateway HTTP at backtest time would break reproducibility — strategies must take a deterministic code path (static fee assumption, fixture-backed analytics, or HOLD) inside backtests.

VIB-4728 / POOL-7 (VIB-4755) extends the same pattern to pool history: NullPoolHistoryReader is injected for the same determinism reason. for_strategy_runner does NOT auto-construct the live PoolHistoryReader (per VIB-4755 D-4: the cut-over is gated on VIB-4730 hosted-egress + VIB-4863 TheGraph API key landing).

for_paper_fork classmethod

for_paper_fork(
    *,
    chain: str,
    wallet_address: str,
    fork_manager: Any,
    gateway_client: Any | None = None,
) -> MarketSnapshot

Build a snapshot for paper-trading on an Anvil fork.

Strategies must NOT see fork_rpc_url directly — the fork-aware market service adapters consume it internally. PRD §4.7.

for_http_backtest_spec classmethod

for_http_backtest_spec(*, spec: Any) -> MarketSnapshot

Build a snapshot for the HTTP-backtest service path.

seeded classmethod

seeded(
    *,
    chain: str = DEFAULT_CHAIN,
    wallet_address: str = "0x" + "0" * 40,
    prices: Mapping[str, Decimal] | None = None,
    price_data: Mapping[str, PriceData] | None = None,
    balances: Mapping[str, TokenBalance] | None = None,
    indicators: Mapping[str, Any] | None = None,
    timestamp: Any | None = None,
) -> MarketSnapshot

Build a snapshot pre-seeded for unit tests.

Goes through the public seed_* API on the snapshot — never writes private cache attributes directly (PRD §5.6).

Overview

almanak.framework.market is the canonical home for MarketSnapshot — the strategy-facing market-data interface. It replaces the two legacy locations (almanak.framework.strategies.intent_strategy.MarketSnapshot and almanak.framework.data.market_snapshot.MarketSnapshot) that silently diverged before VIB-4062.

Builder factories

almanak.framework.market.builders.MarketSnapshotBuilder

Factory class — never instantiated. All entry points are classmethods.

for_strategy_runner classmethod

for_strategy_runner(
    *,
    strategy: Any,
    runtime_context: Any | None = None,
    gateway_client: Any | None = None,
    chain: str | None = None,
    wallet_address: str | None = None,
    runtime_surface: str | None = None,
    chains: tuple[str, ...] | list[str] | None = None,
    multi_chain_price_oracle: Any | None = None,
    multi_chain_balance_provider: Any | None = None,
    aave_health_factor_provider: Any | None = None,
    default_timeframe: str | None = None,
) -> MarketSnapshot

Build a snapshot for the live / hosted runner.

Pulls providers off the strategy's wired services (price oracle, balance provider, indicator provider, …) and wraps the gateway client into the snapshot. Returns a snapshot with runtime_surface="local_sdk" or "hosted" based on runtime_context (the caller resolves which is which — the snapshot does not read env vars).

For multi-chain strategies the caller passes chains= and the multi_chain_* providers; the resulting snapshot has the same canonical class as a single-chain build.

for_pnl_backtest_state classmethod

for_pnl_backtest_state(
    *, chain: str, wallet_address: str, state: Any
) -> MarketSnapshot

Build a snapshot wired to a PnL backtest engine's state.

The state object exposes price_oracle / balance_provider / indicator_provider interfaces; the builder forwards them as-is.

VIB-4727: backtest factories inject NullPoolAnalyticsReader so any market.pool_analytics(...) call raises DataSourceUnavailable deterministically. Live gateway HTTP at backtest time would break reproducibility — strategies must take a deterministic code path (static fee assumption, fixture-backed analytics, or HOLD) inside backtests.

VIB-4728 / POOL-7 (VIB-4755) extends the same pattern to pool history: NullPoolHistoryReader is injected for the same determinism reason. for_strategy_runner does NOT auto-construct the live PoolHistoryReader (per VIB-4755 D-4: the cut-over is gated on VIB-4730 hosted-egress + VIB-4863 TheGraph API key landing).

for_paper_fork classmethod

for_paper_fork(
    *,
    chain: str,
    wallet_address: str,
    fork_manager: Any,
    gateway_client: Any | None = None,
) -> MarketSnapshot

Build a snapshot for paper-trading on an Anvil fork.

Strategies must NOT see fork_rpc_url directly — the fork-aware market service adapters consume it internally. PRD §4.7.

for_http_backtest_spec classmethod

for_http_backtest_spec(*, spec: Any) -> MarketSnapshot

Build a snapshot for the HTTP-backtest service path.

seeded classmethod

seeded(
    *,
    chain: str = DEFAULT_CHAIN,
    wallet_address: str = "0x" + "0" * 40,
    prices: Mapping[str, Decimal] | None = None,
    price_data: Mapping[str, PriceData] | None = None,
    balances: Mapping[str, TokenBalance] | None = None,
    indicators: Mapping[str, Any] | None = None,
    timestamp: Any | None = None,
) -> MarketSnapshot

Build a snapshot pre-seeded for unit tests.

Goes through the public seed_* API on the snapshot — never writes private cache attributes directly (PRD §5.6).

Typed errors

almanak.framework.market.errors

Typed errors for VIB-4062 MarketSnapshot.

Every error inherits from MarketSnapshotError and carries source/chain/instrument/severity/retryability metadata so the runner can classify the failure without parsing exception messages (PRD §4.4).

The snapshot always raises typed errors on missing/stale/invalid data and records the failure in its critical-data-failure register before the exception bubbles up. The runner/runtime decides halt-vs-continue. The snapshot is mode-agnostic.

MarketSnapshotError

MarketSnapshotError(
    *args: Any, reason: str = "", **fields: Any
)

Bases: Exception

Base class for typed snapshot errors.

Subclasses carry structured fields. Stringifying gives a stable shape for log lines: "<ClassName>(<key>=<value>, …): <reason>".

For backward compat with the legacy data-layer error idiom, positional arguments are accepted as (chain, reason) or just (reason,); callers using kwargs (the canonical post-VIB-4062 form) continue to work.

ChainNotConfiguredError

ChainNotConfiguredError(
    *args: Any, reason: str = "", **fields: Any
)

Bases: MarketSnapshotError

A specific chain= argument was supplied but is not in chains.

Raised in BOTH single-chain (mismatch with the only configured chain) and multi-chain (chain not in the configured set) cases. The runner has no recovery path — strategy code must update its chain handling.

AmbiguousChainError

AmbiguousChainError(
    *args: Any, reason: str = "", **fields: Any
)

Bases: MarketSnapshotError

A multi-chain snapshot was queried with chain=None.

Raised when len(self.chains) > 1 and no explicit chain= was given. Multi-chain callers must be explicit; default-to-primary policy hides chain-routing bugs (PRD §4.2 R2, Codex finding HIGH).

StaleDataError

StaleDataError(*args: Any, reason: str = '', **fields: Any)

Bases: MarketSnapshotError

A provider returned data older than the freshness policy permits.

IndicatorUnavailableError

IndicatorUnavailableError(
    *args: Any, reason: str = "", **fields: Any
)

Bases: MarketSnapshotError

Generic indicator (sma, ema, macd, …) unavailable.

CriticalDataFailureRecord dataclass

CriticalDataFailureRecord(
    error_type: str,
    severity: str,
    retryable: bool,
    chain: str | None,
    instrument: str | None,
    reason: str,
)

One entry in MarketSnapshot's critical-data-failure register.

Written from inside the snapshot's _record_critical_failure chokepoint; read from outside via the public has_critical_data_failures / critical_data_failure_count / classify_critical_data_failures / summarize_critical_data_failures accessors.

HOLD contract for data-unavailable errors

Some MarketSnapshot accessors call out to off-chain services through the gateway (e.g. pool_analytics(...), which routes to PoolAnalyticsService over gRPC). When the gateway is unreachable, the strategy container has no fallback, so the accessor raises a typed error such as PoolAnalyticsUnavailableError whose __cause__ chain preserves the underlying DataSourceUnavailable.

The runner's classify_failure walks __cause__ to depth 8 and treats DataSourceUnavailable as DATA_UNAVAILABLE, which the iteration loop interprets as HOLD-worthy. Strategy authors must either let these errors propagate, or catch them and explicitly return Intent.hold(...). A bare except (swallowing the typed error without re-raising or returning HOLD) breaks the runner's HOLD inference and the strategy will appear to "succeed with no signal" while losing the safety contract.

def decide(self, market: MarketSnapshot) -> Intent:
    try:
        analytics = market.pool_analytics(pool_address, protocol="uniswap_v3")
    except PoolAnalyticsUnavailableError:
        # Correct: surface as HOLD so the runner's data-unavailable path fires.
        return Intent.hold(reason="pool analytics unavailable")
    # ... use analytics

The same HOLD contract applies to market.pool_history(...) — backed by PoolHistoryReader over the gateway-side PoolHistoryService (VIB-4728). In a backtest the injected NullPoolHistoryReader raises DataSourceUnavailable("backtest") on every call; in a live run a gateway outage or "pool not found across all providers" surfaces as the same typed error. The same catch-and-HOLD or let-it-propagate discipline applies:

def decide(self, market: MarketSnapshot) -> Intent:
    try:
        history = market.pool_history(
            pool_address,
            chain=self.chain,
            start_date=self.lookback_start,
            resolution="1h",
            protocol="uniswap_v3",
        )
    except DataSourceUnavailable:
        # Correct: HOLD on backtest or gateway-down; do NOT default to zeros.
        return Intent.hold(reason="pool history unavailable")
    # ... use history.value (list[PoolSnapshot])

Return-type DTOs

almanak.framework.market.models

Typed return models for VIB-4062 MarketSnapshot.

Public surface for TokenBalance, PriceData, RSIData, MACDData, all *Data indicator types, and config dataclasses.

This module is intentionally lean (PRD §4.8) — only stdlib imports at module level. The legacy locations almanak.framework.strategies.strategy_models and almanak.framework.strategies.indicator_models re-export from this module during the VIB-4062 transition; commit 6 deletes those re-export shims.

If you are adding a new typed return model for MarketSnapshot, put it HERE — not in strategies/ or data/. The AST uniqueness gate (PRD §5.1) enforces this.

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).

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))(),
    source: str = "",
)

Price data for a token.

Includes source (the named provider that produced the datum) so accounting writers can stamp transaction_ledger.price_inputs_json with the real provider name (VIB-3889).

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: ... round(rsi, 2) f"{rsi:.2f}"

MACDData dataclass

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

MACD data for a token.

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.

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.

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.

MAData dataclass

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

Moving Average data for a token.

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.

OBVData dataclass

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

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

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.

IchimokuData dataclass

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

Ichimoku Cloud data for a token.

IndicatorProvider

IndicatorProvider(
    macd: Callable[..., MACDData] | None = None,
    bollinger: Callable[..., BollingerBandsData]
    | None = None,
    stochastic: Callable[..., StochasticData] | None = None,
    atr: Callable[..., ATRData] | None = None,
    sma: Callable[..., MAData] | None = None,
    ema: Callable[..., MAData] | None = None,
    adx: Callable[..., ADXData] | None = None,
    obv: Callable[..., OBVData] | None = None,
    cci: Callable[..., CCIData] | None = None,
    ichimoku: Callable[..., IchimokuData] | None = None,
)

Provider that wraps all indicator calculators for synchronous access.

Each attribute corresponds to a MarketSnapshot accessor. The runner creates this once and injects it into the strategy, so all indicators work out of the box without strategy authors managing calculators.

Provider Protocols (sync adapters)

almanak.framework.market.services

Provider Protocol layer for VIB-4062 MarketSnapshot.

PRD §4.3 — every builder factory normalizes async data-layer providers into the canonical sync service Protocols at the builder boundary. MarketSnapshot itself never sees an async provider, never decides whether to await or not, never re-implements the sync bridge.

Production sync adapters wrap the async data-layer providers (under framework/data); the wrap centralizes the sync bridge logic in framework/market/sync_bridge.py.