""" Average True Range (ATR) Indicator State This module implements incremental ATR calculation that maintains constant memory usage and provides identical results to traditional batch calculations. ATR is used by Supertrend and other volatility-based indicators. """ from typing import Dict, Union, Optional from .base import OHLCIndicatorState from .moving_average import ExponentialMovingAverageState class ATRState(OHLCIndicatorState): """ Incremental Average True Range calculation state. ATR measures market volatility by calculating the average of true ranges over a specified period. True Range is the maximum of: 1. Current High - Current Low 2. |Current High - Previous Close| 3. |Current Low - Previous Close| This implementation uses exponential moving average for smoothing, which is more responsive than simple moving average and requires less memory. Attributes: period (int): The ATR period ema_state (ExponentialMovingAverageState): EMA state for smoothing true ranges previous_close (float): Previous period's close price Example: atr = ATRState(period=14) # Add OHLC data incrementally ohlc = {'open': 100, 'high': 105, 'low': 98, 'close': 103} atr_value = atr.update(ohlc) # Returns current ATR value # Check if warmed up if atr.is_warmed_up(): current_atr = atr.get_current_value() """ def __init__(self, period: int = 14): """ Initialize ATR state. Args: period: Number of periods for ATR calculation (default: 14) Raises: ValueError: If period is not a positive integer """ super().__init__(period) self.ema_state = ExponentialMovingAverageState(period) self.previous_close = None self.is_initialized = True def update(self, ohlc_data: Dict[str, float]) -> float: """ Update ATR with new OHLC data. Args: ohlc_data: Dictionary with 'open', 'high', 'low', 'close' keys Returns: Current ATR value Raises: ValueError: If OHLC data is invalid TypeError: If ohlc_data is not a dictionary """ # Validate input if not isinstance(ohlc_data, dict): raise TypeError(f"ohlc_data must be a dictionary, got {type(ohlc_data)}") self.validate_input(ohlc_data) high = float(ohlc_data['high']) low = float(ohlc_data['low']) close = float(ohlc_data['close']) # Calculate True Range if self.previous_close is None: # First period - True Range is just High - Low true_range = high - low else: # True Range is the maximum of: # 1. Current High - Current Low # 2. |Current High - Previous Close| # 3. |Current Low - Previous Close| tr1 = high - low tr2 = abs(high - self.previous_close) tr3 = abs(low - self.previous_close) true_range = max(tr1, tr2, tr3) # Update EMA with the true range atr_value = self.ema_state.update(true_range) # Store current close as previous close for next calculation self.previous_close = close self.values_received += 1 # Store current ATR value self._current_values = {'atr': atr_value} return atr_value def is_warmed_up(self) -> bool: """ Check if ATR has enough data for reliable values. Returns: True if EMA state is warmed up (has enough true range values) """ return self.ema_state.is_warmed_up() def reset(self) -> None: """Reset ATR state to initial conditions.""" self.ema_state.reset() self.previous_close = None self.values_received = 0 self._current_values = {} def get_current_value(self) -> Optional[float]: """ Get current ATR value without updating. Returns: Current ATR value, or None if not warmed up """ if not self.is_warmed_up(): return None return self.ema_state.get_current_value() def get_state_summary(self) -> dict: """Get detailed state summary for debugging.""" base_summary = super().get_state_summary() base_summary.update({ 'previous_close': self.previous_close, 'ema_state': self.ema_state.get_state_summary(), 'current_atr': self.get_current_value() }) return base_summary class SimpleATRState(OHLCIndicatorState): """ Simple ATR implementation using simple moving average instead of EMA. This version uses a simple moving average for smoothing true ranges, which matches some traditional ATR implementations but requires more memory. """ def __init__(self, period: int = 14): """ Initialize simple ATR state. Args: period: Number of periods for ATR calculation (default: 14) """ super().__init__(period) from collections import deque self.true_ranges = deque(maxlen=period) self.tr_sum = 0.0 self.previous_close = None self.is_initialized = True def update(self, ohlc_data: Dict[str, float]) -> float: """ Update simple ATR with new OHLC data. Args: ohlc_data: Dictionary with 'open', 'high', 'low', 'close' keys Returns: Current ATR value """ # Validate input if not isinstance(ohlc_data, dict): raise TypeError(f"ohlc_data must be a dictionary, got {type(ohlc_data)}") self.validate_input(ohlc_data) high = float(ohlc_data['high']) low = float(ohlc_data['low']) close = float(ohlc_data['close']) # Calculate True Range if self.previous_close is None: true_range = high - low else: tr1 = high - low tr2 = abs(high - self.previous_close) tr3 = abs(low - self.previous_close) true_range = max(tr1, tr2, tr3) # Update rolling sum if len(self.true_ranges) == self.period: self.tr_sum -= self.true_ranges[0] # Remove oldest value self.true_ranges.append(true_range) self.tr_sum += true_range # Calculate ATR as simple moving average atr_value = self.tr_sum / len(self.true_ranges) # Store state self.previous_close = close self.values_received += 1 self._current_values = {'atr': atr_value} return atr_value def is_warmed_up(self) -> bool: """Check if simple ATR is warmed up.""" return len(self.true_ranges) >= self.period def reset(self) -> None: """Reset simple ATR state.""" self.true_ranges.clear() self.tr_sum = 0.0 self.previous_close = None self.values_received = 0 self._current_values = {} def get_current_value(self) -> Optional[float]: """Get current simple ATR value.""" if not self.is_warmed_up(): return None return self.tr_sum / len(self.true_ranges) def get_state_summary(self) -> dict: """Get detailed state summary for debugging.""" base_summary = super().get_state_summary() base_summary.update({ 'previous_close': self.previous_close, 'tr_window_size': len(self.true_ranges), 'tr_sum': self.tr_sum, 'current_atr': self.get_current_value() }) return base_summary