""" Bollinger Bands indicator implementation. """ from typing import List import pandas as pd from ..base import BaseIndicator from ..result import IndicatorResult class BollingerBandsIndicator(BaseIndicator): """ Bollinger Bands technical indicator. Calculates a set of lines plotted two standard deviations away from a simple moving average. Handles sparse data appropriately without interpolation. """ def calculate(self, df: pd.DataFrame, period: int = 20, std_dev: float = 2.0, price_column: str = 'close') -> List[IndicatorResult]: """ Calculate Bollinger Bands. Args: df: DataFrame with OHLCV data period: Number of periods for moving average (default 20) std_dev: Number of standard deviations (default 2.0) price_column: Price column to use ('open', 'high', 'low', 'close') Returns: List of indicator results with upper band, middle band (SMA), and lower band """ # Validate input data if not self.validate_dataframe(df, period): return [] try: # Calculate middle band (SMA) df['middle_band'] = df[price_column].rolling(window=period, min_periods=period).mean() # Calculate standard deviation df['std'] = df[price_column].rolling(window=period, min_periods=period).std() # Calculate upper and lower bands df['upper_band'] = df['middle_band'] + (std_dev * df['std']) df['lower_band'] = df['middle_band'] - (std_dev * df['std']) # Calculate bandwidth and %B df['bandwidth'] = (df['upper_band'] - df['lower_band']) / df['middle_band'] df['percent_b'] = (df[price_column] - df['lower_band']) / (df['upper_band'] - df['lower_band']) # Convert results to IndicatorResult objects results = [] for timestamp, row in df.iterrows(): if not pd.isna(row['middle_band']): result = IndicatorResult( timestamp=timestamp, symbol=row['symbol'], timeframe=row['timeframe'], values={ 'upper_band': row['upper_band'], 'middle_band': row['middle_band'], 'lower_band': row['lower_band'], 'bandwidth': row['bandwidth'], 'percent_b': row['percent_b'] }, metadata={ 'period': period, 'std_dev': std_dev, 'price_column': price_column } ) results.append(result) return results except Exception as e: if self.logger: self.logger.error(f"Error calculating Bollinger Bands: {e}") return []