Cycles/IncrementalTrader/docs/indicators/volatility.md

Volatility Indicators

Overview

Volatility indicators measure the rate of price change and market uncertainty. IncrementalTrader provides Average True Range (ATR) implementations that help assess market volatility and set appropriate stop-loss levels.

ATRState (Average True Range)

Full ATR implementation that maintains a moving average of True Range values.

Features

• True Range Calculation: Accounts for gaps between trading sessions
• Volatility Measurement: Provides absolute volatility measurement
• Stop-Loss Guidance: Helps set dynamic stop-loss levels
• Trend Strength: Indicates trend strength through volatility

Mathematical Formula

True Range = max(
    High - Low,
    |High - Previous_Close|,
    |Low - Previous_Close|
)

ATR = Moving_Average(True_Range, period)

Class Definition

from IncrementalTrader.strategies.indicators import ATRState

class ATRState(OHLCIndicatorState):
    def __init__(self, period: int):
        super().__init__(period)
        self.true_ranges = []
        self.tr_sum = 0.0
        self.previous_close = None
    
    def _process_ohlc_data(self, high: float, low: float, close: float):
        # Calculate True Range
        if self.previous_close is not None:
            tr = max(
                high - low,
                abs(high - self.previous_close),
                abs(low - self.previous_close)
            )
        else:
            tr = high - low
        
        # Update True Range moving average
        self.true_ranges.append(tr)
        self.tr_sum += tr
        
        if len(self.true_ranges) > self.period:
            old_tr = self.true_ranges.pop(0)
            self.tr_sum -= old_tr
        
        self.previous_close = close
    
    def get_value(self) -> float:
        if not self.is_ready():
            return 0.0
        return self.tr_sum / len(self.true_ranges)

Usage Examples

Basic ATR Calculation

# Create 14-period ATR
atr_14 = ATRState(period=14)

# OHLC data: (high, low, close)
ohlc_data = [
    (105.0, 102.0, 104.0),
    (106.0, 103.0, 105.5),
    (107.0, 104.0, 106.0),
    (108.0, 105.0, 107.5)
]

for high, low, close in ohlc_data:
    atr_14.update_ohlc(high, low, close)
    if atr_14.is_ready():
        print(f"ATR(14): {atr_14.get_value():.2f}")

Dynamic Stop-Loss with ATR

class ATRStopLoss:
    def __init__(self, atr_period: int = 14, atr_multiplier: float = 2.0):
        self.atr = ATRState(atr_period)
        self.atr_multiplier = atr_multiplier
    
    def update(self, high: float, low: float, close: float):
        self.atr.update_ohlc(high, low, close)
    
    def get_stop_loss(self, entry_price: float, position_type: str) -> float:
        if not self.atr.is_ready():
            return entry_price * 0.95 if position_type == "LONG" else entry_price * 1.05
        
        atr_value = self.atr.get_value()
        
        if position_type == "LONG":
            return entry_price - (atr_value * self.atr_multiplier)
        else:  # SHORT
            return entry_price + (atr_value * self.atr_multiplier)
    
    def get_position_size(self, account_balance: float, risk_percent: float, entry_price: float, position_type: str) -> float:
        """Calculate position size based on ATR risk."""
        if not self.atr.is_ready():
            return 0.0
        
        risk_amount = account_balance * (risk_percent / 100)
        stop_loss = self.get_stop_loss(entry_price, position_type)
        risk_per_share = abs(entry_price - stop_loss)
        
        if risk_per_share == 0:
            return 0.0
        
        return risk_amount / risk_per_share

# Usage
atr_stop = ATRStopLoss(atr_period=14, atr_multiplier=2.0)

for high, low, close in ohlc_stream:
    atr_stop.update(high, low, close)
    
    # Calculate stop loss for a long position
    entry_price = close
    stop_loss = atr_stop.get_stop_loss(entry_price, "LONG")
    position_size = atr_stop.get_position_size(10000, 2.0, entry_price, "LONG")
    
    print(f"Entry: {entry_price:.2f}, Stop: {stop_loss:.2f}, Size: {position_size:.0f}")

Performance Characteristics

• Time Complexity: O(1) per update
• Space Complexity: O(period)
• Memory Usage: ~8 bytes per period + constant overhead

SimpleATRState

Simplified ATR implementation using exponential smoothing instead of simple moving average.

Features

• O(1) Memory: Constant memory usage regardless of period
• Exponential Smoothing: Uses Wilder's smoothing method
• Faster Computation: No need to maintain historical True Range values
• Traditional ATR: Follows Wilder's original ATR calculation

Mathematical Formula

True Range = max(
    High - Low,
    |High - Previous_Close|,
    |Low - Previous_Close|
)

ATR = (Previous_ATR × (period - 1) + True_Range) / period

Class Definition

class SimpleATRState(OHLCIndicatorState):
    def __init__(self, period: int):
        super().__init__(period)
        self.atr_value = 0.0
        self.previous_close = None
        self.is_first_value = True
    
    def _process_ohlc_data(self, high: float, low: float, close: float):
        # Calculate True Range
        if self.previous_close is not None:
            tr = max(
                high - low,
                abs(high - self.previous_close),
                abs(low - self.previous_close)
            )
        else:
            tr = high - low
        
        # Update ATR using Wilder's smoothing
        if self.is_first_value:
            self.atr_value = tr
            self.is_first_value = False
        else:
            self.atr_value = ((self.atr_value * (self.period - 1)) + tr) / self.period
        
        self.previous_close = close
    
    def get_value(self) -> float:
        return self.atr_value

Usage Examples

Memory-Efficient ATR

# Create memory-efficient ATR
simple_atr = SimpleATRState(period=14)

# Process large amounts of data with constant memory
for i, (high, low, close) in enumerate(large_ohlc_dataset):
    simple_atr.update_ohlc(high, low, close)
    
    if i % 1000 == 0:  # Print every 1000 updates
        print(f"ATR after {i} updates: {simple_atr.get_value():.4f}")

Volatility Breakout Strategy

class VolatilityBreakout:
    def __init__(self, atr_period: int = 14, breakout_multiplier: float = 1.5):
        self.atr = SimpleATRState(atr_period)
        self.breakout_multiplier = breakout_multiplier
        self.previous_close = None
    
    def update(self, high: float, low: float, close: float):
        self.atr.update_ohlc(high, low, close)
        self.previous_close = close
    
    def get_breakout_levels(self, current_close: float) -> tuple:
        """Get upper and lower breakout levels."""
        if not self.atr.is_ready() or self.previous_close is None:
            return current_close * 1.01, current_close * 0.99
        
        atr_value = self.atr.get_value()
        breakout_distance = atr_value * self.breakout_multiplier
        
        upper_breakout = self.previous_close + breakout_distance
        lower_breakout = self.previous_close - breakout_distance
        
        return upper_breakout, lower_breakout
    
    def check_breakout(self, current_high: float, current_low: float, current_close: float) -> str:
        """Check if current price breaks out of volatility range."""
        upper_level, lower_level = self.get_breakout_levels(current_close)
        
        if current_high > upper_level:
            return "BULLISH_BREAKOUT"
        elif current_low < lower_level:
            return "BEARISH_BREAKOUT"
        
        return "NO_BREAKOUT"

# Usage
breakout_detector = VolatilityBreakout(atr_period=14, breakout_multiplier=1.5)

for high, low, close in ohlc_data:
    breakout_detector.update(high, low, close)
    breakout_signal = breakout_detector.check_breakout(high, low, close)
    
    if breakout_signal != "NO_BREAKOUT":
        print(f"Breakout detected: {breakout_signal} at {close:.2f}")

Performance Characteristics

• Time Complexity: O(1) per update
• Space Complexity: O(1)
• Memory Usage: ~32 bytes (constant)

Comparison: ATRState vs SimpleATRState

Aspect	ATRState	SimpleATRState
Memory Usage	O(period)	O(1)
Calculation Method	Simple Moving Average	Exponential Smoothing
Accuracy	Higher (true SMA)	Good (Wilder's method)
Responsiveness	Moderate	Slightly more responsive
Historical Compatibility	Modern	Traditional (Wilder's)

When to Use ATRState

• Precise Calculations: When you need exact simple moving average of True Range
• Backtesting: For historical analysis where memory isn't constrained
• Research: When studying exact ATR behavior
• Small Periods: When period is small (< 20) and memory isn't an issue

When to Use SimpleATRState

• Memory Efficiency: When processing large amounts of data
• Real-time Systems: For high-frequency trading applications
• Traditional Analysis: When following Wilder's original methodology
• Large Periods: When using large ATR periods (> 50)

Advanced Usage Patterns

Multi-Timeframe ATR Analysis

class MultiTimeframeATR:
    def __init__(self):
        self.atr_short = SimpleATRState(period=7)   # Short-term volatility
        self.atr_medium = SimpleATRState(period=14) # Medium-term volatility
        self.atr_long = SimpleATRState(period=28)   # Long-term volatility
    
    def update(self, high: float, low: float, close: float):
        self.atr_short.update_ohlc(high, low, close)
        self.atr_medium.update_ohlc(high, low, close)
        self.atr_long.update_ohlc(high, low, close)
    
    def get_volatility_regime(self) -> str:
        """Determine current volatility regime."""
        if not all([self.atr_short.is_ready(), self.atr_medium.is_ready(), self.atr_long.is_ready()]):
            return "UNKNOWN"
        
        short_atr = self.atr_short.get_value()
        medium_atr = self.atr_medium.get_value()
        long_atr = self.atr_long.get_value()
        
        # Compare short-term to long-term volatility
        volatility_ratio = short_atr / long_atr if long_atr > 0 else 1.0
        
        if volatility_ratio > 1.5:
            return "HIGH_VOLATILITY"
        elif volatility_ratio < 0.7:
            return "LOW_VOLATILITY"
        else:
            return "NORMAL_VOLATILITY"
    
    def get_adaptive_stop_multiplier(self) -> float:
        """Get adaptive stop-loss multiplier based on volatility regime."""
        regime = self.get_volatility_regime()
        
        if regime == "HIGH_VOLATILITY":
            return 2.5  # Wider stops in high volatility
        elif regime == "LOW_VOLATILITY":
            return 1.5  # Tighter stops in low volatility
        else:
            return 2.0  # Standard stops in normal volatility

# Usage
multi_atr = MultiTimeframeATR()

for high, low, close in ohlc_data:
    multi_atr.update(high, low, close)
    
    regime = multi_atr.get_volatility_regime()
    stop_multiplier = multi_atr.get_adaptive_stop_multiplier()
    
    print(f"Volatility Regime: {regime}, Stop Multiplier: {stop_multiplier:.1f}")

ATR-Based Position Sizing

class ATRPositionSizer:
    def __init__(self, atr_period: int = 14):
        self.atr = SimpleATRState(atr_period)
        self.price_history = []
    
    def update(self, high: float, low: float, close: float):
        self.atr.update_ohlc(high, low, close)
        self.price_history.append(close)
        
        # Keep only recent price history
        if len(self.price_history) > 100:
            self.price_history.pop(0)
    
    def calculate_position_size(self, account_balance: float, risk_percent: float, 
                              entry_price: float, stop_loss_atr_multiplier: float = 2.0) -> dict:
        """Calculate position size based on ATR risk management."""
        
        if not self.atr.is_ready():
            return {"position_size": 0, "risk_amount": 0, "stop_loss": entry_price * 0.95}
        
        atr_value = self.atr.get_value()
        risk_amount = account_balance * (risk_percent / 100)
        
        # Calculate stop loss based on ATR
        stop_loss = entry_price - (atr_value * stop_loss_atr_multiplier)
        risk_per_share = entry_price - stop_loss
        
        # Calculate position size
        if risk_per_share > 0:
            position_size = risk_amount / risk_per_share
        else:
            position_size = 0
        
        return {
            "position_size": position_size,
            "risk_amount": risk_amount,
            "stop_loss": stop_loss,
            "atr_value": atr_value,
            "risk_per_share": risk_per_share
        }
    
    def get_volatility_percentile(self) -> float:
        """Get current ATR percentile compared to recent history."""
        if not self.atr.is_ready() or len(self.price_history) < 20:
            return 50.0  # Default to median
        
        current_atr = self.atr.get_value()
        
        # Calculate ATR for recent periods
        recent_atrs = []
        for i in range(len(self.price_history) - 14):
            if i + 14 < len(self.price_history):
                # Simplified ATR calculation for comparison
                price_range = max(self.price_history[i:i+14]) - min(self.price_history[i:i+14])
                recent_atrs.append(price_range)
        
        if not recent_atrs:
            return 50.0
        
        # Calculate percentile
        sorted_atrs = sorted(recent_atrs)
        position = sum(1 for atr in sorted_atrs if atr <= current_atr)
        percentile = (position / len(sorted_atrs)) * 100
        
        return percentile

# Usage
position_sizer = ATRPositionSizer(atr_period=14)

for high, low, close in ohlc_data:
    position_sizer.update(high, low, close)
    
    # Calculate position for a potential trade
    trade_info = position_sizer.calculate_position_size(
        account_balance=10000,
        risk_percent=2.0,
        entry_price=close,
        stop_loss_atr_multiplier=2.0
    )
    
    volatility_percentile = position_sizer.get_volatility_percentile()
    
    print(f"Price: {close:.2f}, Position Size: {trade_info['position_size']:.0f}, "
          f"ATR Percentile: {volatility_percentile:.1f}%")

Integration with Strategies

ATR-Enhanced Strategy Example

class ATRTrendStrategy(IncStrategyBase):
    def __init__(self, name: str, params: dict = None):
        super().__init__(name, params)
        
        # Initialize indicators
        self.atr = SimpleATRState(self.params.get('atr_period', 14))
        self.sma = MovingAverageState(self.params.get('sma_period', 20))
        
        # ATR parameters
        self.atr_stop_multiplier = self.params.get('atr_stop_multiplier', 2.0)
        self.atr_entry_multiplier = self.params.get('atr_entry_multiplier', 0.5)
    
    def _process_aggregated_data(self, timestamp: int, ohlcv: tuple) -> IncStrategySignal:
        open_price, high, low, close, volume = ohlcv
        
        # Update indicators
        self.atr.update_ohlc(high, low, close)
        self.sma.update(close)
        
        # Wait for indicators to be ready
        if not all([self.atr.is_ready(), self.sma.is_ready()]):
            return IncStrategySignal.HOLD()
        
        atr_value = self.atr.get_value()
        sma_value = self.sma.get_value()
        
        # Calculate dynamic entry threshold based on ATR
        entry_threshold = atr_value * self.atr_entry_multiplier
        
        # Generate signals based on trend and volatility
        if close > sma_value + entry_threshold:
            # Strong uptrend with sufficient volatility
            confidence = min(0.9, (close - sma_value) / atr_value * 0.1)
            
            # Calculate stop loss
            stop_loss = close - (atr_value * self.atr_stop_multiplier)
            
            return IncStrategySignal.BUY(
                confidence=confidence,
                metadata={
                    'atr_value': atr_value,
                    'sma_value': sma_value,
                    'stop_loss': stop_loss,
                    'entry_threshold': entry_threshold
                }
            )
        
        elif close < sma_value - entry_threshold:
            # Strong downtrend with sufficient volatility
            confidence = min(0.9, (sma_value - close) / atr_value * 0.1)
            
            # Calculate stop loss
            stop_loss = close + (atr_value * self.atr_stop_multiplier)
            
            return IncStrategySignal.SELL(
                confidence=confidence,
                metadata={
                    'atr_value': atr_value,
                    'sma_value': sma_value,
                    'stop_loss': stop_loss,
                    'entry_threshold': entry_threshold
                }
            )
        
        return IncStrategySignal.HOLD()

Performance Optimization Tips

1. Choose the Right ATR Implementation

# For memory-constrained environments
atr = SimpleATRState(period=14)  # O(1) memory

# For precise calculations
atr = ATRState(period=14)  # O(period) memory

2. Batch Processing for Multiple ATRs

def update_multiple_atrs(atrs: list, high: float, low: float, close: float):
    """Efficiently update multiple ATR indicators."""
    for atr in atrs:
        atr.update_ohlc(high, low, close)
    
    return [atr.get_value() for atr in atrs if atr.is_ready()]

3. Cache ATR Values for Complex Calculations

class CachedATR:
    def __init__(self, period: int):
        self.atr = SimpleATRState(period)
        self._cached_value = 0.0
        self._cache_valid = False
    
    def update_ohlc(self, high: float, low: float, close: float):
        self.atr.update_ohlc(high, low, close)
        self._cache_valid = False
    
    def get_value(self) -> float:
        if not self._cache_valid:
            self._cached_value = self.atr.get_value()
            self._cache_valid = True
        return self._cached_value

---

ATR indicators are essential for risk management and volatility analysis. Use ATRState for precise calculations or SimpleATRState for memory efficiency in high-frequency applications.