Technical Indicators¶

Technical analysis indicator calculators available through the data layer.

RSICalculator¶

almanak.framework.data.indicators.RSICalculator ¶

RSICalculator(
    ohlcv_provider: OHLCVProvider, default_period: int = 14
)

RSI (Relative Strength Index) Calculator using Wilder's Smoothing Method.

Implements the standard RSI formula with Wilder's exponential moving average for smoothing, which is the industry-standard approach.

RSI Formula

RSI = 100 - (100 / (1 + RS)) RS = Average Gain / Average Loss

Wilder's Smoothing: - First average: Simple average of first N periods - Subsequent: ((Previous Avg * (N-1)) + Current Value) / N

属性：

名称	类型	描述
`ohlcv_provider`		Provider for OHLCV data (implements OHLCVProvider protocol)
`default_period`		Default RSI period (default 14)

Example

provider = CoinGeckoOHLCVProvider() calculator = RSICalculator(ohlcv_provider=provider)

Calculate 14-period RSI for WETH¶

rsi = await calculator.calculate_rsi("WETH", period=14) print(f"RSI: {rsi:.2f}")

Interpret the signal¶

if rsi < 30: print("Oversold - potential buy signal") elif rsi > 70: print("Overbought - potential sell signal")

Initialize the RSI Calculator.

参数：

名称	类型	描述	默认
`ohlcv_provider`	`OHLCVProvider`	Provider implementing OHLCVProvider protocol	必需
`default_period`	`int`	Default RSI calculation period (default 14)	`14`

calculate_rsi_from_prices `staticmethod` ¶

calculate_rsi_from_prices(
    close_prices: list[Decimal], period: int = 14
) -> float

Calculate RSI from a list of close prices using Wilder's smoothing.

This is a pure calculation function that can be used independently of the data provider for testing or alternative data sources.

参数：

名称	类型	描述	默认
`close_prices`	`list[Decimal]`	List of closing prices (oldest first)	必需
`period`	`int`	RSI calculation period (default 14)	`14`

返回：

类型	描述
`float`	RSI value from 0 to 100

引发：

类型	描述
`InsufficientDataError`	If not enough price data (need period + 1)

calculate_rsi `async` ¶

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

Calculate RSI for a token.

Fetches OHLCV data from the configured provider and calculates RSI using Wilder's smoothing method.

参数：

名称	类型	描述	默认
`token`	`str`	Token symbol (e.g., "WETH", "ETH")	必需
`period`	`int`	RSI calculation period (default 14)	`14`
`timeframe`	`str`	OHLCV candle timeframe (default "4h") Supported: "1m", "5m", "15m", "1h", "4h", "1d" Note: 1m/5m/15m may return 30-min candles (CoinGecko limitation)	`'4h'`

返回：

类型	描述
`float`	RSI value from 0 to 100

引发：

类型	描述
`InsufficientDataError`	If not enough historical data
`DataSourceError`	If data cannot be fetched

Example

Default 4-hour candles¶

rsi = await calculator.calculate_rsi("WETH", period=14)

1-hour candles for shorter-term analysis¶

rsi_1h = await calculator.calculate_rsi("WETH", period=14, timeframe="1h")

Daily candles for longer-term analysis¶

rsi_1d = await calculator.calculate_rsi("WETH", period=14, timeframe="1d")

get_ohlcv_provider_health ¶

get_ohlcv_provider_health() -> dict[str, Any]

Get health metrics from the OHLCV provider if available.

BollingerBandsCalculator¶

almanak.framework.data.indicators.BollingerBandsCalculator ¶

BollingerBandsCalculator(ohlcv_provider: OHLCVProvider)

Bollinger Bands Calculator.

Calculates Bollinger Bands with configurable period and standard deviation multiplier.

Default Parameters

Period: 20 (industry standard)
Standard Deviation Multiplier: 2.0 (captures ~95% of price movement)

属性：

名称	类型	描述
`ohlcv_provider`		Provider for OHLCV data (implements OHLCVProvider protocol)

Example

provider = CoinGeckoOHLCVProvider() bb_calc = BollingerBandsCalculator(ohlcv_provider=provider)

bb = await bb_calc.calculate_bollinger_bands("WETH", period=20, std_dev=2.0) print(f"Upper: {bb.upper_band}, Middle: {bb.middle_band}, Lower: {bb.lower_band}")

Check price position¶

if bb.percent_b < 0.2: print("Near lower band - potential buy")

Initialize the Bollinger Bands Calculator.

参数：

名称	类型	描述	默认
`ohlcv_provider`	`OHLCVProvider`	Provider implementing OHLCVProvider protocol	必需

name `property` ¶

name: str

Return indicator name.

min_data_points `property` ¶

min_data_points: int

Return minimum data points (default period of 20).

calculate_bollinger_from_prices `staticmethod` ¶

calculate_bollinger_from_prices(
    close_prices: list[Decimal],
    period: int = 20,
    std_dev_multiplier: float = 2.0,
) -> BollingerBandsResult

Calculate Bollinger Bands from a list of close prices.

参数：

名称	类型	描述	默认
`close_prices`	`list[Decimal]`	List of closing prices (oldest first)	必需
`period`	`int`	SMA period (default 20)	`20`
`std_dev_multiplier`	`float`	Standard deviation multiplier (default 2.0)	`2.0`

返回：

类型	描述
`BollingerBandsResult`	BollingerBandsResult with all band values

引发：

类型	描述
`InsufficientDataError`	If not enough price data

calculate_bollinger_bands `async` ¶

calculate_bollinger_bands(
    token: str,
    period: int = 20,
    std_dev: float = 2.0,
    timeframe: str = "1h",
) -> BollingerBandsResult

Calculate Bollinger Bands for a token.

参数：

名称	类型	描述	默认
`token`	`str`	Token symbol (e.g., "WETH", "ETH")	必需
`period`	`int`	SMA period (default 20)	`20`
`std_dev`	`float`	Standard deviation multiplier (default 2.0)	`2.0`
`timeframe`	`str`	OHLCV candle timeframe (default "1h")	`'1h'`

返回：

类型	描述
`BollingerBandsResult`	BollingerBandsResult with upper_band, middle_band, lower_band,
`BollingerBandsResult`	bandwidth, and percent_b

引发：

类型	描述
`InsufficientDataError`	If not enough historical data
`DataSourceError`	If data cannot be fetched

Example

bb = await bb_calc.calculate_bollinger_bands("WETH", period=20, std_dev=2.0)

Trading logic¶

if bb.percent_b < 0: # Price below lower band - potential buy return SwapIntent(token_in="USDC", token_out="WETH", ...) elif bb.percent_b > 1: # Price above upper band - potential sell return SwapIntent(token_in="WETH", token_out="USDC", ...)

Volatility check¶

if bb.bandwidth < 0.05: print("Low volatility - squeeze detected")

calculate `async` ¶

calculate(
    token: str, timeframe: str = "1h", **params: Any
) -> dict[str, float]

Calculate Bollinger Bands (BaseIndicator protocol implementation).

参数：

名称	类型	描述	默认
`token`	`str`	Token symbol	必需
`timeframe`	`str`	OHLCV candle timeframe	`'1h'`
`**params`	`Any`	period (default 20), std_dev (default 2.0)	`{}`

返回：

类型	描述
`dict[str, float]`	Dictionary with all Bollinger Bands values

MACDCalculator¶

almanak.framework.data.indicators.MACDCalculator ¶

MACDCalculator(ohlcv_provider: OHLCVProvider)

MACD (Moving Average Convergence Divergence) Calculator.

Calculates MACD with configurable fast, slow, and signal periods.

Default Parameters (industry standard): - Fast EMA: 12 periods - Slow EMA: 26 periods - Signal EMA: 9 periods

属性：

名称	类型	描述
`ohlcv_provider`		Provider for OHLCV data (implements OHLCVProvider protocol)

Example

provider = CoinGeckoOHLCVProvider() macd_calc = MACDCalculator(ohlcv_provider=provider)

macd = await macd_calc.calculate_macd("WETH", fast=12, slow=26, signal=9) print(f"MACD: {macd.macd_line}, Signal: {macd.signal_line}")

Check for bullish crossover¶

if macd.histogram > 0: print("MACD above signal - bullish")

Initialize the MACD Calculator.

参数：

名称	类型	描述	默认
`ohlcv_provider`	`OHLCVProvider`	Provider implementing OHLCVProvider protocol	必需

name `property` ¶

name: str

Return indicator name.

min_data_points `property` ¶

min_data_points: int

Return minimum data points (slow period + signal + buffer).

calculate_macd_from_prices `staticmethod` ¶

calculate_macd_from_prices(
    close_prices: list[Decimal],
    fast_period: int = 12,
    slow_period: int = 26,
    signal_period: int = 9,
) -> MACDResult

Calculate MACD from a list of close prices.

参数：

名称	类型	描述	默认
`close_prices`	`list[Decimal]`	List of closing prices (oldest first)	必需
`fast_period`	`int`	Fast EMA period (default 12)	`12`
`slow_period`	`int`	Slow EMA period (default 26)	`26`
`signal_period`	`int`	Signal line EMA period (default 9)	`9`

返回：

类型	描述
`MACDResult`	MACDResult with macd_line, signal_line, and histogram

引发：

类型	描述
`InsufficientDataError`	If not enough price data

calculate_macd `async` ¶

calculate_macd(
    token: str,
    fast_period: int = 12,
    slow_period: int = 26,
    signal_period: int = 9,
    timeframe: str = "1h",
) -> MACDResult

Calculate MACD for a token.

参数：

名称	类型	描述	默认
`token`	`str`	Token symbol (e.g., "WETH", "ETH")	必需
`fast_period`	`int`	Fast EMA period (default 12)	`12`
`slow_period`	`int`	Slow EMA period (default 26)	`26`
`signal_period`	`int`	Signal line EMA period (default 9)	`9`
`timeframe`	`str`	OHLCV candle timeframe (default "1h")	`'1h'`

返回：

类型	描述
`MACDResult`	MACDResult with macd_line, signal_line, and histogram

引发：

类型	描述
`InsufficientDataError`	If not enough historical data
`DataSourceError`	If data cannot be fetched

Example

macd = await macd_calc.calculate_macd("WETH", fast=12, slow=26, signal=9)

Trading logic¶

if macd.histogram > 0: # MACD above signal - bullish momentum print("Bullish crossover") elif macd.histogram < 0: # MACD below signal - bearish momentum print("Bearish crossover")

calculate `async` ¶

calculate(
    token: str, timeframe: str = "1h", **params: Any
) -> dict[str, float]

Calculate MACD (BaseIndicator protocol implementation).

参数：

名称	类型	描述	默认
`token`	`str`	Token symbol	必需
`timeframe`	`str`	OHLCV candle timeframe	`'1h'`
`**params`	`Any`	fast_period (12), slow_period (26), signal_period (9)	`{}`

返回：

类型	描述
`dict[str, float]`	Dictionary with all MACD values

StochasticCalculator¶

almanak.framework.data.indicators.StochasticCalculator ¶

StochasticCalculator(ohlcv_provider: OHLCVProvider)

Stochastic Oscillator Calculator.

Calculates the Stochastic Oscillator with configurable %K and %D periods.

Default Parameters

%K Period: 14 (industry standard)
%D Period: 3 (smoothing period)

属性：

名称	类型	描述
`ohlcv_provider`		Provider for OHLCV data (implements OHLCVProvider protocol)

Example

provider = CoinGeckoOHLCVProvider() stoch_calc = StochasticCalculator(ohlcv_provider=provider)

stoch = await stoch_calc.calculate_stochastic("WETH", k_period=14, d_period=3) print(f"%K: {stoch.k_value:.2f}, %D: {stoch.d_value:.2f}")

if stoch.k_value < 20: print("Oversold territory")

Initialize the Stochastic Calculator.

参数：

名称	类型	描述	默认
`ohlcv_provider`	`OHLCVProvider`	Provider implementing OHLCVProvider protocol	必需

name `property` ¶

name: str

Return indicator name.

min_data_points `property` ¶

min_data_points: int

Return minimum data points (k_period + d_period).

calculate_stochastic_from_candles `staticmethod` ¶

calculate_stochastic_from_candles(
    candles: list[OHLCVCandle],
    k_period: int = 14,
    d_period: int = 3,
) -> StochasticResult

Calculate Stochastic Oscillator from OHLCV candles.

Formula: %K = 100 * (Close - Lowest Low) / (Highest High - Lowest Low) %D = SMA(%K, d_period)

参数：

名称	类型	描述	默认
`candles`	`list[OHLCVCandle]`	List of OHLCVCandle objects (oldest first)	必需
`k_period`	`int`	Lookback period for %K (default 14)	`14`
`d_period`	`int`	SMA period for %D (default 3)	`3`

返回：

类型	描述
`StochasticResult`	StochasticResult with k_value and d_value

引发：

类型	描述
`InsufficientDataError`	If not enough candle data

calculate_stochastic `async` ¶

calculate_stochastic(
    token: str,
    k_period: int = 14,
    d_period: int = 3,
    timeframe: str = "1h",
) -> StochasticResult

Calculate Stochastic Oscillator for a token.

参数：

名称	类型	描述	默认
`token`	`str`	Token symbol (e.g., "WETH", "ETH")	必需
`k_period`	`int`	Lookback period for %K (default 14)	`14`
`d_period`	`int`	SMA period for %D (default 3)	`3`
`timeframe`	`str`	OHLCV candle timeframe (default "1h")	`'1h'`

返回：

类型	描述
`StochasticResult`	StochasticResult with k_value and d_value (both 0-100 scale)

引发：

类型	描述
`InsufficientDataError`	If not enough historical data
`DataSourceError`	If data cannot be fetched

Example

stoch = await stoch_calc.calculate_stochastic("WETH", k_period=14, d_period=3)

Trading logic¶

if stoch.k_value < 20: print("Oversold - potential buy signal") elif stoch.k_value > 80: print("Overbought - potential sell signal")

Crossover signals¶

if stoch.k_value > stoch.d_value: print("Bullish - %K above %D")

calculate `async` ¶

calculate(
    token: str, timeframe: str = "1h", **params: Any
) -> dict[str, float]

Calculate Stochastic (BaseIndicator protocol implementation).

参数：

名称	类型	描述	默认
`token`	`str`	Token symbol	必需
`timeframe`	`str`	OHLCV candle timeframe	`'1h'`
`**params`	`Any`	k_period (default 14), d_period (default 3)	`{}`

返回：

类型	描述
`dict[str, float]`	Dictionary with k_value and d_value

ATRCalculator¶

almanak.framework.data.indicators.ATRCalculator ¶

ATRCalculator(ohlcv_provider: OHLCVProvider)

ATR (Average True Range) Calculator.

Calculates ATR with configurable period using Wilder's smoothing method.

Default Parameters

Period: 14 (industry standard)

属性：

名称	类型	描述
`ohlcv_provider`		Provider for OHLCV data (implements OHLCVProvider protocol)

Example

provider = CoinGeckoOHLCVProvider() atr_calc = ATRCalculator(ohlcv_provider=provider)

atr = await atr_calc.calculate_atr("WETH", period=14, timeframe="4h") print(f"ATR: ${atr:.2f}")

Position sizing¶

if atr > 100: print("High volatility - reduce position size")

Initialize the ATR Calculator.

参数：

名称	类型	描述	默认
`ohlcv_provider`	`OHLCVProvider`	Provider implementing OHLCVProvider protocol	必需

name `property` ¶

name: str

Return indicator name.

min_data_points `property` ¶

min_data_points: int

Return minimum data points (period + 1).

calculate_atr_from_candles `staticmethod` ¶

calculate_atr_from_candles(
    candles: list[OHLCVCandle], period: int = 14
) -> float

Calculate ATR from OHLCV candles using Wilder's smoothing.

Wilder's ATR uses a modified EMA: - First ATR = Simple average of first N True Ranges - Subsequent ATR = ((Prior ATR * (N-1)) + Current TR) / N

参数：

名称	类型	描述	默认
`candles`	`list[OHLCVCandle]`	List of OHLCVCandle objects (oldest first)	必需
`period`	`int`	ATR period (default 14)	`14`

返回：

类型	描述
`float`	ATR value

引发：

类型	描述
`InsufficientDataError`	If not enough candle data

calculate_atr_from_prices `staticmethod` ¶

calculate_atr_from_prices(
    prices: list, period: int = 14
) -> float

Calculate ATR from close prices only using Wilder's smoothing.

When only close prices are available (e.g., synthetic Monte Carlo paths), True Range is approximated as |close[i] - close[i-1]|. This is mathematically equivalent to ATR computed from OHLCV data where open = high = low = close for each candle.

参数：

名称	类型	描述	默认
`prices`	`list`	List of close prices (oldest first), as Decimal or float	必需
`period`	`int`	ATR period (default 14)	`14`

返回：

类型	描述
`float`	ATR value (float)

引发：

类型	描述
`InsufficientDataError`	If not enough price data
`ValueError`	If period < 1

calculate_atr `async` ¶

calculate_atr(
    token: str, period: int = 14, timeframe: str = "1h"
) -> float

Calculate ATR for a token.

参数：

名称	类型	描述	默认
`token`	`str`	Token symbol (e.g., "WETH", "ETH")	必需
`period`	`int`	ATR period (default 14)	`14`
`timeframe`	`str`	OHLCV candle timeframe (default "1h")	`'1h'`

返回：

类型	描述
`float`	ATR value (in the same units as the token price)

引发：

类型	描述
`InsufficientDataError`	If not enough historical data
`DataSourceError`	If data cannot be fetched

Example

atr = await atr_calc.calculate_atr("WETH", period=14, timeframe="4h")

Stop-loss calculation¶

current_price = 2500 stop_loss = current_price - (2 * atr) print(f"Stop loss at ${stop_loss:.2f} (2 ATR below)")

Position sizing with 1% risk¶

risk_per_trade = 10000 * 0.01 # $100 position_size = risk_per_trade / atr print(f"Position size: {position_size:.2f} units")

calculate `async` ¶

calculate(
    token: str, timeframe: str = "1h", **params: Any
) -> dict[str, float]

Calculate ATR (BaseIndicator protocol implementation).

参数：

名称	类型	描述	默认
`token`	`str`	Token symbol	必需
`timeframe`	`str`	OHLCV candle timeframe	`'1h'`
`**params`	`Any`	period (default 14)	`{}`

返回：

类型	描述
`dict[str, float]`	Dictionary with atr value

MovingAverageCalculator¶

almanak.framework.data.indicators.MovingAverageCalculator ¶

MovingAverageCalculator(ohlcv_provider: OHLCVProvider)

Calculator for Simple, Exponential, and Weighted Moving Averages.

Provides three types of moving averages commonly used in technical analysis: - SMA: Simple average of last N closing prices - EMA: Exponential moving average with configurable smoothing - WMA: Weighted moving average with linear weighting

属性：

名称	类型	描述
`ohlcv_provider`		Provider for OHLCV data (implements OHLCVProvider protocol)

Example

provider = CoinGeckoOHLCVProvider() ma_calc = MovingAverageCalculator(ohlcv_provider=provider)

sma = await ma_calc.sma("WETH", period=20, timeframe="4h") ema = await ma_calc.ema("WETH", period=12, timeframe="1h")

Initialize the Moving Average Calculator.

参数：

名称	类型	描述	默认
`ohlcv_provider`	`OHLCVProvider`	Provider implementing OHLCVProvider protocol	必需

name `property` ¶

name: str

Return indicator name.

min_data_points `property` ¶

min_data_points: int

Return minimum data points (depends on period, default 20).

calculate_sma_from_prices `staticmethod` ¶

calculate_sma_from_prices(
    close_prices: list[Decimal], period: int
) -> float

Calculate SMA from a list of close prices.

参数：

名称	类型	描述	默认
`close_prices`	`list[Decimal]`	List of closing prices (oldest first)	必需
`period`	`int`	Number of periods for the average	必需

返回：

类型	描述
`float`	Simple Moving Average value

引发：

类型	描述
`InsufficientDataError`	If not enough price data

calculate_ema_from_prices `staticmethod` ¶

calculate_ema_from_prices(
    close_prices: list[Decimal],
    period: int,
    smoothing: float = 2.0,
) -> float

Calculate EMA from a list of close prices.

Uses the formula: EMA = Price(t) * k + EMA(y) * (1 - k) where k = smoothing / (period + 1)

参数：

名称	类型	描述	默认
`close_prices`	`list[Decimal]`	List of closing prices (oldest first)	必需
`period`	`int`	Number of periods for the average	必需
`smoothing`	`float`	Smoothing factor (default 2.0 for standard EMA)	`2.0`

返回：

类型	描述
`float`	Exponential Moving Average value

引发：

类型	描述
`InsufficientDataError`	If not enough price data

calculate_wma_from_prices `staticmethod` ¶

calculate_wma_from_prices(
    close_prices: list[Decimal], period: int
) -> float

Calculate WMA from a list of close prices.

Weighted Moving Average: More recent prices have higher weights. Weight for position i (1-indexed from oldest): i / sum(1..period)

参数：

名称	类型	描述	默认
`close_prices`	`list[Decimal]`	List of closing prices (oldest first)	必需
`period`	`int`	Number of periods for the average	必需

返回：

类型	描述
`float`	Weighted Moving Average value

引发：

类型	描述
`InsufficientDataError`	If not enough price data

sma `async` ¶

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

Calculate Simple Moving Average for a token.

SMA is the unweighted mean of the last N closing prices. Commonly used periods: 10, 20, 50, 100, 200

参数：

名称	类型	描述	默认
`token`	`str`	Token symbol (e.g., "WETH", "ETH")	必需
`period`	`int`	Number of periods (default 20)	`20`
`timeframe`	`str`	OHLCV candle timeframe (default "1h")	`'1h'`

返回：

类型	描述
`float`	SMA value as float

引发：

类型	描述
`InsufficientDataError`	If not enough historical data
`DataSourceError`	If data cannot be fetched

Example

sma_20 = await ma_calc.sma("WETH", period=20, timeframe="1h") sma_200 = await ma_calc.sma("WETH", period=200, timeframe="1d")

ema `async` ¶

ema(
    token: str,
    period: int = 12,
    timeframe: str = "1h",
    smoothing: float = 2.0,
) -> float

Calculate Exponential Moving Average for a token.

EMA gives more weight to recent prices using exponential decay. Commonly used periods: 12, 26 (for MACD), 9, 21

参数：

名称	类型	描述	默认
`token`	`str`	Token symbol (e.g., "WETH", "ETH")	必需
`period`	`int`	Number of periods (default 12)	`12`
`timeframe`	`str`	OHLCV candle timeframe (default "1h")	`'1h'`
`smoothing`	`float`	Smoothing factor (default 2.0)	`2.0`

返回：

类型	描述
`float`	EMA value as float

引发：

类型	描述
`InsufficientDataError`	If not enough historical data
`DataSourceError`	If data cannot be fetched

Example

ema_12 = await ma_calc.ema("WETH", period=12, timeframe="1h") ema_26 = await ma_calc.ema("WETH", period=26, timeframe="1h")

wma `async` ¶

wma(
    token: str, period: int = 20, timeframe: str = "1h"
) -> float

Calculate Weighted Moving Average for a token.

WMA assigns higher weights to more recent data points linearly. Most recent price has weight N, oldest has weight 1.

参数：

名称	类型	描述	默认
`token`	`str`	Token symbol (e.g., "WETH", "ETH")	必需
`period`	`int`	Number of periods (default 20)	`20`
`timeframe`	`str`	OHLCV candle timeframe (default "1h")	`'1h'`

返回：

类型	描述
`float`	WMA value as float

引发：

类型	描述
`InsufficientDataError`	If not enough historical data
`DataSourceError`	If data cannot be fetched

Example

wma_20 = await ma_calc.wma("WETH", period=20, timeframe="1h")

calculate `async` ¶

calculate(
    token: str, timeframe: str = "1h", **params: Any
) -> dict[str, float]

Calculate moving average (BaseIndicator protocol implementation).

参数：

名称	类型	描述	默认
`token`	`str`	Token symbol	必需
`timeframe`	`str`	OHLCV candle timeframe	`'1h'`
`**params`	`Any`	Must include 'type' (sma/ema/wma) and 'period'	`{}`

返回：

类型	描述
`dict[str, float]`	Dictionary with calculated value

Example

result = await ma_calc.calculate("WETH", type="sma", period=20)

{"sma": 2500.0}¶

Result Types¶

BollingerBandsResult¶

almanak.framework.data.indicators.BollingerBandsResult `dataclass` ¶

BollingerBandsResult(
    upper_band: float,
    middle_band: float,
    lower_band: float,
    bandwidth: float,
    percent_b: float,
)

Result from Bollinger Bands calculation.

属性：

名称	类型	描述
`upper_band`	`float`	Upper band (SMA + std_dev * multiplier)
`middle_band`	`float`	Middle band (Simple Moving Average)
`lower_band`	`float`	Lower band (SMA - std_dev * multiplier)
`bandwidth`	`float`	Band width as ratio ((upper - lower) / middle). Typical ranges for crypto assets (1h candles, 20-period, 2 std dev): - Squeeze: < 0.02 (low volatility, breakout likely) - Normal: 0.02 - 0.06 (typical trading range) - Expansion: > 0.06 (high volatility, trend in progress) These are starting guidelines; actual thresholds vary by asset and timeframe. Shorter timeframes and lower-cap assets trend higher.
`percent_b`	`float`	Price position relative to bands (0 = lower, 1 = upper). Values above 1.0 indicate price above the upper band; values below 0.0 indicate price below the lower band.

to_dict ¶

to_dict() -> dict[str, float]

Convert to dictionary format.

MACDResult¶

almanak.framework.data.indicators.MACDResult `dataclass` ¶

MACDResult(
    macd_line: float, signal_line: float, histogram: float
)

Result from MACD calculation.

属性：

名称	类型	描述
`macd_line`	`float`	MACD line (fast EMA - slow EMA)
`signal_line`	`float`	Signal line (EMA of MACD line)
`histogram`	`float`	MACD histogram (MACD line - signal line)

to_dict ¶

to_dict() -> dict[str, float]

Convert to dictionary format.

StochasticResult¶

almanak.framework.data.indicators.StochasticResult `dataclass` ¶

StochasticResult(k_value: float, d_value: float)

Result from Stochastic Oscillator calculation.

属性：

名称	类型	描述
`k_value`	`float`	%K (fast stochastic) - current position in price range
`d_value`	`float`	%D (slow stochastic) - SMA of %K

to_dict ¶

to_dict() -> dict[str, float]

Convert to dictionary format.

OHLCV Data¶

OHLCVData¶

almanak.framework.data.indicators.OHLCVData `dataclass` ¶

OHLCVData(
    timestamp: datetime,
    open: Decimal,
    high: Decimal,
    low: Decimal,
    close: Decimal,
    volume: Decimal | None = None,
)

OHLCV candlestick data point.

属性：

名称	类型	描述
`timestamp`	`datetime`	Candle open time
`open`	`Decimal`	Opening price
`high`	`Decimal`	Highest price
`low`	`Decimal`	Lowest price
`close`	`Decimal`	Closing price
`volume`	`Decimal \| None`	Trading volume (optional, CoinGecko OHLC doesn't include volume)

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary for serialization.

CoinGeckoOHLCVProvider¶

almanak.framework.data.indicators.CoinGeckoOHLCVProvider ¶

CoinGeckoOHLCVProvider(
    api_key: str = "",
    cache_ttl: int = 300,
    request_timeout: float = 30.0,
)

CoinGecko OHLCV data provider implementing the OHLCVProvider protocol.

Fetches historical OHLCV data from CoinGecko API for technical analysis. Implements the OHLCVProvider protocol from src/data/interfaces.py.

Note on timeframe support

CoinGecko's OHLC API has granularity limitations based on the day range: - 1-2 days: Returns 30-minute candles (supports 1m, 5m, 15m approximation) - 3-30 days: Returns 4-hour candles (supports 1h, 4h) - 31+ days: Returns daily candles (supports 1d)

For finer timeframes (1m, 5m, 15m), we request 1-2 day ranges and return the 30-minute candles. Strategies requiring exact minute-level candles should use a more granular data source.

属性：

名称	类型	描述
`api_key`		Optional CoinGecko API key (uses pro API if provided)
`cache_ttl`		Cache time-to-live in seconds (default 300 = 5 minutes)
`request_timeout`		HTTP request timeout in seconds (default 30)

Example

provider = CoinGeckoOHLCVProvider(api_key="optional-key") candles = await provider.get_ohlcv("WETH", timeframe="1h", limit=100) print(f"Got {len(candles)} candles") for candle in candles[-3:]: print(f" {candle.timestamp}: close={candle.close}")

Initialize the CoinGecko OHLCV provider.

参数：

名称	类型	描述	默认
`api_key`	`str`	Optional CoinGecko API key. If provided, uses pro API.	`''`
`cache_ttl`	`int`	Cache time-to-live in seconds. Default 300 (5 minutes).	`300`
`request_timeout`	`float`	HTTP request timeout in seconds. Default 30.	`30.0`

supported_timeframes `property` ¶

supported_timeframes: list[str]

Return the list of timeframes this provider supports.

返回：

名称	类型	描述
	`list[str]`	List of supported timeframe strings.
`Note`	`list[str]`	1m, 5m, 15m return 30-minute candles (CoinGecko's finest granularity)

close `async` ¶

close() -> None

Close the HTTP session.

get_ohlcv `async` ¶

get_ohlcv(
    token: str,
    quote: str = "USD",
    timeframe: str = "1h",
    limit: int = 100,
) -> list[OHLCVCandle]

Get OHLCV data for a token.

Uses the CoinGecko OHLC endpoint for historical candlestick data. Implements the OHLCVProvider protocol.

参数：

名称	类型	描述	默认
`token`	`str`	Token symbol (e.g., "WETH", "ETH")	必需
`quote`	`str`	Quote currency (default "USD")	`'USD'`
`timeframe`	`str`	Candle timeframe. Supported: "1m", "5m", "15m", "1h", "4h", "1d" Note: 1m, 5m, 15m return 30-minute candles (CoinGecko's finest granularity)	`'1h'`
`limit`	`int`	Number of candles to fetch	`100`

返回：

类型	描述
`list[OHLCVCandle]`	List of OHLCVCandle objects sorted by timestamp ascending.
`list[OHLCVCandle]`	Volume is always None as CoinGecko OHLC API does not provide volume data.

引发：

类型	描述
`DataSourceUnavailable`	If data cannot be fetched
`ValueError`	If timeframe is not supported

get_health_metrics ¶

get_health_metrics() -> dict[str, Any]

Get health metrics for observability.

clear_cache ¶

clear_cache() -> None

Clear the OHLCV cache.

aenter `async` ¶

__aenter__() -> CoinGeckoOHLCVProvider

Async context manager entry.

aexit `async` ¶

__aexit__(exc_type: Any, exc_val: Any, exc_tb: Any) -> None

Async context manager exit.

Registry¶

IndicatorRegistry¶

almanak.framework.data.indicators.IndicatorRegistry ¶

Registry for technical indicator discovery and instantiation.

This class provides a centralized registry for indicator classes, enabling dynamic lookup and factory-style creation.

The registry uses class methods so it can be used without instantiation, acting as a singleton-like pattern for global indicator registration.

Example

Register an indicator¶

IndicatorRegistry.register("rsi", RSICalculator) IndicatorRegistry.register("bollinger", BollingerBandsCalculator)

List all registered indicators¶

indicators = IndicatorRegistry.list_all()

['rsi', 'bollinger']¶

Get an indicator class by name¶

RSIClass = IndicatorRegistry.get("rsi") if RSIClass: calculator = RSIClass(ohlcv_provider=provider)

Check if an indicator is registered¶

if IndicatorRegistry.has("macd"): MACDClass = IndicatorRegistry.get("macd")

register `classmethod` ¶

register(
    name: str,
    indicator_class: type,
    metadata: dict[str, Any] | None = None,
) -> None

Register an indicator class.

参数：

名称	类型	描述	默认
`name`	`str`	Unique name for the indicator (case-insensitive)	必需
`indicator_class`	`type`	The indicator class to register	必需
`metadata`	`dict[str, Any] \| None`	Optional metadata about the indicator (description, version, etc.)	`None`

引发：

类型	描述
`ValueError`	If name is already registered

Example

IndicatorRegistry.register("rsi", RSICalculator, metadata={ "description": "Relative Strength Index", "version": "1.0.0", "category": "momentum", })

get `classmethod` ¶

get(name: str) -> type | None

Get an indicator class by name.

参数：

名称	类型	描述	默认
`name`	`str`	Indicator name (case-insensitive)	必需

返回：

类型	描述
`type \| None`	The indicator class, or None if not found

Example

RSIClass = IndicatorRegistry.get("rsi") if RSIClass: calculator = RSIClass(ohlcv_provider=provider)

has `classmethod` ¶

has(name: str) -> bool

Check if an indicator is registered.

参数：

名称	类型	描述	默认
`name`	`str`	Indicator name (case-insensitive)	必需

返回：

类型	描述
`bool`	True if registered, False otherwise

list_all `classmethod` ¶

list_all() -> list[str]

List all registered indicator names.

返回：

类型	描述
`list[str]`	Sorted list of registered indicator names

Example

indicators = IndicatorRegistry.list_all()

['atr', 'bollinger', 'macd', 'rsi', 'stochastic']¶

get_metadata `classmethod` ¶

get_metadata(name: str) -> dict[str, Any]

Get metadata for an indicator.

参数：

名称	类型	描述	默认
`name`	`str`	Indicator name (case-insensitive)	必需

返回：

类型	描述
`dict[str, Any]`	Metadata dictionary, or empty dict if not found

unregister `classmethod` ¶

unregister(name: str) -> bool

Unregister an indicator.

参数：

名称	类型	描述	默认
`name`	`str`	Indicator name (case-insensitive)	必需

返回：

类型	描述
`bool`	True if unregistered, False if not found

clear `classmethod` ¶

clear() -> None

Clear all registered indicators.

Primarily used for testing.

create `classmethod` ¶

create(
    name: str, *args: Any, **kwargs: Any
) -> BaseIndicator | None

Factory method to create an indicator instance.

参数：

名称	类型	描述	默认
`name`	`str`	Indicator name (case-insensitive)	必需
`*args`	`Any`	Positional arguments for indicator constructor	`()`
`**kwargs`	`Any`	Keyword arguments for indicator constructor	`{}`

返回：

类型	描述
`BaseIndicator \| None`	Indicator instance, or None if not found

Example

calculator = IndicatorRegistry.create("rsi", ohlcv_provider=provider) if calculator: rsi = await calculator.calculate("WETH", timeframe="1h", period=14)

Technical Indicators¶

RSICalculator¶

almanak.framework.data.indicators.RSICalculator ¶

Calculate 14-period RSI for WETH¶

Interpret the signal¶

calculate_rsi_from_prices staticmethod ¶

calculate_rsi async ¶

Default 4-hour candles¶

1-hour candles for shorter-term analysis¶

Daily candles for longer-term analysis¶

get_ohlcv_provider_health ¶

BollingerBandsCalculator¶

almanak.framework.data.indicators.BollingerBandsCalculator ¶

Check price position¶

name property ¶

min_data_points property ¶

calculate_bollinger_from_prices staticmethod ¶

calculate_bollinger_bands async ¶

Trading logic¶

Volatility check¶

calculate async ¶

MACDCalculator¶

almanak.framework.data.indicators.MACDCalculator ¶

Check for bullish crossover¶

name property ¶

min_data_points property ¶

calculate_macd_from_prices staticmethod ¶

calculate_macd async ¶

Trading logic¶

calculate async ¶

StochasticCalculator¶

almanak.framework.data.indicators.StochasticCalculator ¶

name property ¶

min_data_points property ¶

calculate_stochastic_from_candles staticmethod ¶

calculate_stochastic async ¶

Trading logic¶

Crossover signals¶

calculate async ¶

ATRCalculator¶

almanak.framework.data.indicators.ATRCalculator ¶

Position sizing¶

name property ¶

min_data_points property ¶

calculate_atr_from_candles staticmethod ¶

calculate_atr_from_prices staticmethod ¶

calculate_atr async ¶

Stop-loss calculation¶

Position sizing with 1% risk¶

calculate async ¶

MovingAverageCalculator¶

almanak.framework.data.indicators.MovingAverageCalculator ¶

name property ¶

min_data_points property ¶

calculate_sma_from_prices staticmethod ¶

calculate_ema_from_prices staticmethod ¶

calculate_wma_from_prices staticmethod ¶

sma async ¶

ema async ¶

wma async ¶

calculate async ¶

{"sma": 2500.0}¶

Result Types¶

BollingerBandsResult¶

almanak.framework.data.indicators.BollingerBandsResult dataclass ¶

to_dict ¶

MACDResult¶

almanak.framework.data.indicators.MACDResult dataclass ¶

to_dict ¶

StochasticResult¶

almanak.framework.data.indicators.StochasticResult dataclass ¶

to_dict ¶

OHLCV Data¶

OHLCVData¶

almanak.framework.data.indicators.OHLCVData dataclass ¶

to_dict ¶

CoinGeckoOHLCVProvider¶

almanak.framework.data.indicators.CoinGeckoOHLCVProvider ¶

supported_timeframes property ¶

close async ¶

calculate_rsi_from_prices `staticmethod` ¶

calculate_rsi `async` ¶

name `property` ¶

min_data_points `property` ¶

calculate_bollinger_from_prices `staticmethod` ¶

calculate_bollinger_bands `async` ¶

calculate `async` ¶

name `property` ¶

min_data_points `property` ¶

calculate_macd_from_prices `staticmethod` ¶

calculate_macd `async` ¶

calculate `async` ¶

name `property` ¶

min_data_points `property` ¶

calculate_stochastic_from_candles `staticmethod` ¶

calculate_stochastic `async` ¶

calculate `async` ¶

name `property` ¶

min_data_points `property` ¶

calculate_atr_from_candles `staticmethod` ¶

calculate_atr_from_prices `staticmethod` ¶

calculate_atr `async` ¶

calculate `async` ¶

name `property` ¶

min_data_points `property` ¶

calculate_sma_from_prices `staticmethod` ¶

calculate_ema_from_prices `staticmethod` ¶

calculate_wma_from_prices `staticmethod` ¶

sma `async` ¶

ema `async` ¶

wma `async` ¶

calculate `async` ¶

almanak.framework.data.indicators.BollingerBandsResult `dataclass` ¶

almanak.framework.data.indicators.MACDResult `dataclass` ¶

almanak.framework.data.indicators.StochasticResult `dataclass` ¶

almanak.framework.data.indicators.OHLCVData `dataclass` ¶

supported_timeframes `property` ¶

close `async` ¶

get_ohlcv `async` ¶

aenter `async` ¶

aexit `async` ¶

register `classmethod` ¶

get `classmethod` ¶

has `classmethod` ¶

list_all `classmethod` ¶

get_metadata `classmethod` ¶

unregister `classmethod` ¶

clear `classmethod` ¶

create `classmethod` ¶