c8d8d980aa
- Introduced a dedicated sub-package for technical indicators under `data/common/indicators/`, improving modularity and maintainability. - Moved `TechnicalIndicators` and `IndicatorResult` classes to their respective files, along with utility functions for configuration management. - Updated import paths throughout the codebase to reflect the new structure, ensuring compatibility. - Added comprehensive safety net tests for the indicators module to verify core functionality and prevent regressions during refactoring. - Enhanced documentation to provide clear usage examples and details on the new package structure. These changes improve the overall architecture of the technical indicators module, making it more scalable and easier to manage.
713 lines
28 KiB
Python
713 lines
28 KiB
Python
"""
|
|
Technical Indicator Chart Layers
|
|
|
|
This module implements overlay indicator layers for technical analysis visualization
|
|
including SMA, EMA, and Bollinger Bands with comprehensive error handling.
|
|
"""
|
|
|
|
import pandas as pd
|
|
import plotly.graph_objects as go
|
|
from typing import Dict, Any, Optional, List, Union, Callable
|
|
from dataclasses import dataclass
|
|
|
|
from ..error_handling import (
|
|
ChartErrorHandler, ChartError, ErrorSeverity, DataRequirements,
|
|
InsufficientDataError, DataValidationError, IndicatorCalculationError,
|
|
ErrorRecoveryStrategies, create_error_annotation, get_error_message
|
|
)
|
|
|
|
from .base import BaseLayer, LayerConfig
|
|
from data.common.indicators import TechnicalIndicators
|
|
from data.common.data_types import OHLCVCandle
|
|
from components.charts.utils import get_indicator_colors
|
|
from utils.logger import get_logger
|
|
|
|
# Initialize logger
|
|
logger = get_logger("default_logger")
|
|
|
|
|
|
@dataclass
|
|
class IndicatorLayerConfig(LayerConfig):
|
|
"""Extended configuration for indicator layers"""
|
|
id: str = ""
|
|
indicator_type: str = "" # e.g., 'sma', 'ema', 'rsi'
|
|
parameters: Dict[str, Any] = None # Indicator-specific parameters
|
|
line_width: int = 2
|
|
opacity: float = 1.0
|
|
show_middle_line: bool = True # For indicators like Bollinger Bands
|
|
|
|
def __post_init__(self):
|
|
super().__post_init__()
|
|
if self.parameters is None:
|
|
self.parameters = {}
|
|
|
|
|
|
class BaseIndicatorLayer(BaseLayer):
|
|
"""
|
|
Enhanced base class for all indicator layers with comprehensive error handling.
|
|
"""
|
|
|
|
def __init__(self, config: IndicatorLayerConfig):
|
|
"""
|
|
Initialize base indicator layer.
|
|
|
|
Args:
|
|
config: Indicator layer configuration
|
|
"""
|
|
super().__init__(config)
|
|
self.indicators = TechnicalIndicators()
|
|
self.colors = get_indicator_colors()
|
|
self.calculated_data = None
|
|
self.calculation_errors = []
|
|
|
|
def prepare_indicator_data(self, data: pd.DataFrame) -> List[OHLCVCandle]:
|
|
"""
|
|
Convert DataFrame to OHLCVCandle format for indicator calculations.
|
|
|
|
Args:
|
|
data: Chart data (OHLCV format)
|
|
|
|
Returns:
|
|
List of OHLCVCandle objects
|
|
"""
|
|
try:
|
|
candles = []
|
|
for _, row in data.iterrows():
|
|
# Calculate start_time (assuming 1-minute candles for now)
|
|
start_time = row['timestamp']
|
|
end_time = row['timestamp']
|
|
|
|
candle = OHLCVCandle(
|
|
symbol="BTCUSDT", # Default symbol for testing
|
|
timeframe="1m", # Default timeframe
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
open=Decimal(str(row['open'])),
|
|
high=Decimal(str(row['high'])),
|
|
low=Decimal(str(row['low'])),
|
|
close=Decimal(str(row['close'])),
|
|
volume=Decimal(str(row.get('volume', 0))),
|
|
trade_count=1, # Default trade count
|
|
exchange="test", # Test exchange
|
|
is_complete=True # Mark as complete for testing
|
|
)
|
|
candles.append(candle)
|
|
|
|
return candles
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Indicators: Error preparing indicator data: {e}")
|
|
return []
|
|
|
|
def validate_indicator_data(self, data: Union[pd.DataFrame, List[Dict[str, Any]]],
|
|
required_columns: List[str] = None) -> bool:
|
|
"""
|
|
Validate data specifically for indicator calculations.
|
|
|
|
Args:
|
|
data: Input data
|
|
required_columns: Required columns for this indicator
|
|
|
|
Returns:
|
|
True if data is valid for indicator calculation
|
|
"""
|
|
try:
|
|
# Use parent validation first
|
|
if not super().validate_data(data):
|
|
return False
|
|
|
|
# Convert to DataFrame if needed
|
|
if isinstance(data, list):
|
|
df = pd.DataFrame(data)
|
|
else:
|
|
df = data.copy()
|
|
|
|
# Check required columns for indicator
|
|
if required_columns:
|
|
missing_columns = [col for col in required_columns if col not in df.columns]
|
|
if missing_columns:
|
|
error = ChartError(
|
|
code='MISSING_INDICATOR_COLUMNS',
|
|
message=f'Missing columns for {self.config.indicator_type}: {missing_columns}',
|
|
severity=ErrorSeverity.ERROR,
|
|
context={
|
|
'indicator_type': self.config.indicator_type,
|
|
'missing_columns': missing_columns,
|
|
'available_columns': list(df.columns)
|
|
},
|
|
recovery_suggestion=f'Ensure data contains required columns: {required_columns}'
|
|
)
|
|
self.error_handler.errors.append(error)
|
|
return False
|
|
|
|
# Check data sufficiency for indicator
|
|
indicator_config = {
|
|
'type': self.config.indicator_type,
|
|
'parameters': self.config.parameters or {}
|
|
}
|
|
|
|
indicator_error = DataRequirements.check_indicator_requirements(
|
|
self.config.indicator_type,
|
|
len(df),
|
|
self.config.parameters or {}
|
|
)
|
|
|
|
if indicator_error.severity == ErrorSeverity.WARNING:
|
|
self.error_handler.warnings.append(indicator_error)
|
|
elif indicator_error.severity in [ErrorSeverity.ERROR, ErrorSeverity.CRITICAL]:
|
|
self.error_handler.errors.append(indicator_error)
|
|
return False
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Indicators: Error validating indicator data: {e}")
|
|
error = ChartError(
|
|
code='INDICATOR_VALIDATION_ERROR',
|
|
message=f'Indicator validation failed: {str(e)}',
|
|
severity=ErrorSeverity.ERROR,
|
|
context={'exception': str(e), 'indicator_type': self.config.indicator_type}
|
|
)
|
|
self.error_handler.errors.append(error)
|
|
return False
|
|
|
|
def safe_calculate_indicator(self, data: pd.DataFrame,
|
|
calculation_func: Callable,
|
|
**kwargs) -> Optional[pd.DataFrame]:
|
|
"""
|
|
Safely calculate indicator with error handling.
|
|
|
|
Args:
|
|
data: Input data
|
|
calculation_func: Function to calculate indicator
|
|
**kwargs: Additional arguments for calculation
|
|
|
|
Returns:
|
|
Calculated indicator data or None if failed
|
|
"""
|
|
try:
|
|
# Validate data first
|
|
if not self.validate_indicator_data(data):
|
|
return None
|
|
|
|
# Try calculation with recovery strategies
|
|
result = calculation_func(data, **kwargs)
|
|
|
|
# Validate result
|
|
if result is None or (isinstance(result, pd.DataFrame) and result.empty):
|
|
error = ChartError(
|
|
code='EMPTY_INDICATOR_RESULT',
|
|
message=f'Indicator calculation returned no data: {self.config.indicator_type}',
|
|
severity=ErrorSeverity.WARNING,
|
|
context={'indicator_type': self.config.indicator_type, 'input_length': len(data)},
|
|
recovery_suggestion='Check calculation parameters or input data range'
|
|
)
|
|
self.error_handler.warnings.append(error)
|
|
return None
|
|
|
|
# Check for sufficient calculated data
|
|
if isinstance(result, pd.DataFrame) and len(result) < len(data) * 0.1:
|
|
error = ChartError(
|
|
code='INSUFFICIENT_INDICATOR_OUTPUT',
|
|
message=f'Very few indicator values calculated: {len(result)}/{len(data)}',
|
|
severity=ErrorSeverity.WARNING,
|
|
context={
|
|
'indicator_type': self.config.indicator_type,
|
|
'output_length': len(result),
|
|
'input_length': len(data)
|
|
},
|
|
recovery_suggestion='Consider adjusting indicator parameters'
|
|
)
|
|
self.error_handler.warnings.append(error)
|
|
|
|
self.calculated_data = result
|
|
return result
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Indicators: Error calculating {self.config.indicator_type}: {e}")
|
|
|
|
# Try to apply error recovery
|
|
recovery_strategy = ErrorRecoveryStrategies.handle_insufficient_data(
|
|
ChartError(
|
|
code='INDICATOR_CALCULATION_ERROR',
|
|
message=f'Calculation failed for {self.config.indicator_type}: {str(e)}',
|
|
severity=ErrorSeverity.ERROR,
|
|
context={'exception': str(e), 'indicator_type': self.config.indicator_type}
|
|
),
|
|
fallback_options={'data_length': len(data)}
|
|
)
|
|
|
|
if recovery_strategy['can_proceed'] and recovery_strategy['fallback_action'] == 'adjust_parameters':
|
|
# Try with adjusted parameters
|
|
try:
|
|
modified_config = recovery_strategy.get('modified_config', {})
|
|
self.logger.info(f"Indicators: Retrying indicator calculation with adjusted parameters: {modified_config}")
|
|
|
|
# Update parameters temporarily
|
|
original_params = self.config.parameters.copy() if self.config.parameters else {}
|
|
self.config.parameters.update(modified_config)
|
|
|
|
# Retry calculation
|
|
result = calculation_func(data, **kwargs)
|
|
|
|
# Restore original parameters
|
|
self.config.parameters = original_params
|
|
|
|
if result is not None and not (isinstance(result, pd.DataFrame) and result.empty):
|
|
# Add warning about parameter adjustment
|
|
warning = ChartError(
|
|
code='INDICATOR_PARAMETERS_ADJUSTED',
|
|
message=recovery_strategy['user_message'],
|
|
severity=ErrorSeverity.WARNING,
|
|
context={'original_params': original_params, 'adjusted_params': modified_config}
|
|
)
|
|
self.error_handler.warnings.append(warning)
|
|
self.calculated_data = result
|
|
return result
|
|
|
|
except Exception as retry_error:
|
|
self.logger.error(f"Indicators: Retry with adjusted parameters also failed: {retry_error}")
|
|
|
|
# Final error if all recovery attempts fail
|
|
error = ChartError(
|
|
code='INDICATOR_CALCULATION_FAILED',
|
|
message=f'Failed to calculate {self.config.indicator_type}: {str(e)}',
|
|
severity=ErrorSeverity.ERROR,
|
|
context={'exception': str(e), 'indicator_type': self.config.indicator_type}
|
|
)
|
|
self.error_handler.errors.append(error)
|
|
return None
|
|
|
|
def create_indicator_traces(self, data: pd.DataFrame, subplot_row: int = 1) -> List[go.Scatter]:
|
|
"""
|
|
Create indicator traces with error handling.
|
|
Must be implemented by subclasses.
|
|
"""
|
|
raise NotImplementedError("Subclasses must implement create_indicator_traces")
|
|
|
|
def is_enabled(self) -> bool:
|
|
"""Check if the layer is enabled."""
|
|
return self.config.enabled
|
|
|
|
def is_overlay(self) -> bool:
|
|
"""Check if this layer is an overlay (main chart) or subplot."""
|
|
return self.config.subplot_row is None
|
|
|
|
def get_subplot_row(self) -> Optional[int]:
|
|
"""Get the subplot row for this layer."""
|
|
return self.config.subplot_row
|
|
|
|
|
|
class SMALayer(BaseIndicatorLayer):
|
|
"""Simple Moving Average layer with enhanced error handling"""
|
|
|
|
def __init__(self, config: IndicatorLayerConfig = None):
|
|
"""Initialize SMA layer"""
|
|
if config is None:
|
|
config = IndicatorLayerConfig(
|
|
indicator_type='sma',
|
|
parameters={'period': 20}
|
|
)
|
|
super().__init__(config)
|
|
|
|
def create_traces(self, data: List[Dict[str, Any]], subplot_row: int = 1) -> List[go.Scatter]:
|
|
"""Create SMA traces with comprehensive error handling"""
|
|
try:
|
|
# Convert to DataFrame
|
|
df = pd.DataFrame(data) if isinstance(data, list) else data.copy()
|
|
|
|
# Validate data
|
|
if not self.validate_indicator_data(df, required_columns=['close', 'timestamp']):
|
|
if self.error_handler.errors:
|
|
return [self.create_error_trace(f"SMA Error: {self._error_message}")]
|
|
|
|
# Calculate SMA with error handling
|
|
period = self.config.parameters.get('period', 20)
|
|
sma_data = self.safe_calculate_indicator(
|
|
df,
|
|
self._calculate_sma,
|
|
period=period
|
|
)
|
|
|
|
if sma_data is None:
|
|
if self.error_handler.errors:
|
|
return [self.create_error_trace(f"SMA calculation failed")]
|
|
else:
|
|
return [] # Skip layer gracefully
|
|
|
|
# Create trace
|
|
sma_trace = go.Scatter(
|
|
x=sma_data['timestamp'],
|
|
y=sma_data['sma'],
|
|
mode='lines',
|
|
name=f'SMA({period})',
|
|
line=dict(
|
|
color=self.config.color or '#2196F3',
|
|
width=self.config.line_width
|
|
)
|
|
)
|
|
|
|
self.traces = [sma_trace]
|
|
return self.traces
|
|
|
|
except Exception as e:
|
|
error_msg = f"Indicators: Error creating SMA traces: {str(e)}"
|
|
self.logger.error(error_msg)
|
|
return [self.create_error_trace(error_msg)]
|
|
|
|
def _calculate_sma(self, data: pd.DataFrame, period: int) -> pd.DataFrame:
|
|
"""Calculate SMA with validation"""
|
|
try:
|
|
result_df = data.copy()
|
|
result_df['sma'] = result_df['close'].rolling(window=period, min_periods=period).mean()
|
|
|
|
# Remove NaN values
|
|
result_df = result_df.dropna(subset=['sma'])
|
|
|
|
if result_df.empty:
|
|
raise IndicatorCalculationError(ChartError(
|
|
code='SMA_NO_VALUES',
|
|
message=f'SMA calculation produced no values (period={period}, data_length={len(data)})',
|
|
severity=ErrorSeverity.ERROR,
|
|
context={'period': period, 'data_length': len(data)}
|
|
))
|
|
|
|
return result_df[['timestamp', 'sma']]
|
|
|
|
except Exception as e:
|
|
raise IndicatorCalculationError(ChartError(
|
|
code='SMA_CALCULATION_ERROR',
|
|
message=f'SMA calculation failed: {str(e)}',
|
|
severity=ErrorSeverity.ERROR,
|
|
context={'period': period, 'data_length': len(data), 'exception': str(e)}
|
|
))
|
|
|
|
def render(self, fig: go.Figure, data: pd.DataFrame, **kwargs) -> go.Figure:
|
|
"""Render SMA layer for compatibility with base interface"""
|
|
try:
|
|
traces = self.create_traces(data.to_dict('records'), **kwargs)
|
|
for trace in traces:
|
|
if hasattr(fig, 'add_trace'):
|
|
fig.add_trace(trace, **kwargs)
|
|
else:
|
|
fig.add_trace(trace)
|
|
return fig
|
|
except Exception as e:
|
|
self.logger.error(f"Indicators: Error rendering SMA layer: {e}")
|
|
return fig
|
|
|
|
|
|
class EMALayer(BaseIndicatorLayer):
|
|
"""Exponential Moving Average layer with enhanced error handling"""
|
|
|
|
def __init__(self, config: IndicatorLayerConfig = None):
|
|
"""Initialize EMA layer"""
|
|
if config is None:
|
|
config = IndicatorLayerConfig(
|
|
indicator_type='ema',
|
|
parameters={'period': 20}
|
|
)
|
|
super().__init__(config)
|
|
|
|
def create_traces(self, data: List[Dict[str, Any]], subplot_row: int = 1) -> List[go.Scatter]:
|
|
"""Create EMA traces with comprehensive error handling"""
|
|
try:
|
|
# Convert to DataFrame
|
|
df = pd.DataFrame(data) if isinstance(data, list) else data.copy()
|
|
|
|
# Validate data
|
|
if not self.validate_indicator_data(df, required_columns=['close', 'timestamp']):
|
|
if self.error_handler.errors:
|
|
return [self.create_error_trace(f"EMA Error: {self._error_message}")]
|
|
|
|
# Calculate EMA with error handling
|
|
period = self.config.parameters.get('period', 20)
|
|
ema_data = self.safe_calculate_indicator(
|
|
df,
|
|
self._calculate_ema,
|
|
period=period
|
|
)
|
|
|
|
if ema_data is None:
|
|
if self.error_handler.errors:
|
|
return [self.create_error_trace(f"EMA calculation failed")]
|
|
else:
|
|
return [] # Skip layer gracefully
|
|
|
|
# Create trace
|
|
ema_trace = go.Scatter(
|
|
x=ema_data['timestamp'],
|
|
y=ema_data['ema'],
|
|
mode='lines',
|
|
name=f'EMA({period})',
|
|
line=dict(
|
|
color=self.config.color or '#FF9800',
|
|
width=self.config.line_width
|
|
)
|
|
)
|
|
|
|
self.traces = [ema_trace]
|
|
return self.traces
|
|
|
|
except Exception as e:
|
|
error_msg = f"Indicators: Error creating EMA traces: {str(e)}"
|
|
self.logger.error(error_msg)
|
|
return [self.create_error_trace(error_msg)]
|
|
|
|
def _calculate_ema(self, data: pd.DataFrame, period: int) -> pd.DataFrame:
|
|
"""Calculate EMA with validation"""
|
|
try:
|
|
result_df = data.copy()
|
|
result_df['ema'] = result_df['close'].ewm(span=period, adjust=False).mean()
|
|
|
|
# For EMA, we can start from the first value, but remove obvious outliers
|
|
# Skip first few values for stability
|
|
warmup_period = max(1, period // 4)
|
|
result_df = result_df.iloc[warmup_period:]
|
|
|
|
if result_df.empty:
|
|
raise IndicatorCalculationError(ChartError(
|
|
code='EMA_NO_VALUES',
|
|
message=f'EMA calculation produced no values (period={period}, data_length={len(data)})',
|
|
severity=ErrorSeverity.ERROR,
|
|
context={'period': period, 'data_length': len(data)}
|
|
))
|
|
|
|
return result_df[['timestamp', 'ema']]
|
|
|
|
except Exception as e:
|
|
raise IndicatorCalculationError(ChartError(
|
|
code='EMA_CALCULATION_ERROR',
|
|
message=f'EMA calculation failed: {str(e)}',
|
|
severity=ErrorSeverity.ERROR,
|
|
context={'period': period, 'data_length': len(data), 'exception': str(e)}
|
|
))
|
|
|
|
def render(self, fig: go.Figure, data: pd.DataFrame, **kwargs) -> go.Figure:
|
|
"""Render EMA layer for compatibility with base interface"""
|
|
try:
|
|
traces = self.create_traces(data.to_dict('records'), **kwargs)
|
|
for trace in traces:
|
|
if hasattr(fig, 'add_trace'):
|
|
fig.add_trace(trace, **kwargs)
|
|
else:
|
|
fig.add_trace(trace)
|
|
return fig
|
|
except Exception as e:
|
|
self.logger.error(f"Indicators: Error rendering EMA layer: {e}")
|
|
return fig
|
|
|
|
|
|
class BollingerBandsLayer(BaseIndicatorLayer):
|
|
"""Bollinger Bands layer with enhanced error handling"""
|
|
|
|
def __init__(self, config: IndicatorLayerConfig = None):
|
|
"""Initialize Bollinger Bands layer"""
|
|
if config is None:
|
|
config = IndicatorLayerConfig(
|
|
indicator_type='bollinger_bands',
|
|
parameters={'period': 20, 'std_dev': 2},
|
|
show_middle_line=True
|
|
)
|
|
super().__init__(config)
|
|
|
|
def create_traces(self, data: List[Dict[str, Any]], subplot_row: int = 1) -> List[go.Scatter]:
|
|
"""Create Bollinger Bands traces with comprehensive error handling"""
|
|
try:
|
|
# Convert to DataFrame
|
|
df = pd.DataFrame(data) if isinstance(data, list) else data.copy()
|
|
|
|
# Validate data
|
|
if not self.validate_indicator_data(df, required_columns=['close', 'timestamp']):
|
|
if self.error_handler.errors:
|
|
return [self.create_error_trace(f"Bollinger Bands Error: {self._error_message}")]
|
|
|
|
# Calculate Bollinger Bands with error handling
|
|
period = self.config.parameters.get('period', 20)
|
|
std_dev = self.config.parameters.get('std_dev', 2)
|
|
|
|
bb_data = self.safe_calculate_indicator(
|
|
df,
|
|
self._calculate_bollinger_bands,
|
|
period=period,
|
|
std_dev=std_dev
|
|
)
|
|
|
|
if bb_data is None:
|
|
if self.error_handler.errors:
|
|
return [self.create_error_trace(f"Bollinger Bands calculation failed")]
|
|
else:
|
|
return [] # Skip layer gracefully
|
|
|
|
# Create traces
|
|
traces = []
|
|
|
|
# Upper band
|
|
upper_trace = go.Scatter(
|
|
x=bb_data['timestamp'],
|
|
y=bb_data['upper_band'],
|
|
mode='lines',
|
|
name=f'BB Upper({period})',
|
|
line=dict(color=self.config.color or '#9C27B0', width=1),
|
|
showlegend=True
|
|
)
|
|
traces.append(upper_trace)
|
|
|
|
# Lower band with fill
|
|
lower_trace = go.Scatter(
|
|
x=bb_data['timestamp'],
|
|
y=bb_data['lower_band'],
|
|
mode='lines',
|
|
name=f'BB Lower({period})',
|
|
line=dict(color=self.config.color or '#9C27B0', width=1),
|
|
fill='tonexty',
|
|
fillcolor='rgba(156, 39, 176, 0.1)',
|
|
showlegend=True
|
|
)
|
|
traces.append(lower_trace)
|
|
|
|
# Middle line (SMA)
|
|
if self.config.show_middle_line:
|
|
middle_trace = go.Scatter(
|
|
x=bb_data['timestamp'],
|
|
y=bb_data['middle_band'],
|
|
mode='lines',
|
|
name=f'BB Middle({period})',
|
|
line=dict(color=self.config.color or '#9C27B0', width=1, dash='dash'),
|
|
showlegend=True
|
|
)
|
|
traces.append(middle_trace)
|
|
|
|
self.traces = traces
|
|
return self.traces
|
|
|
|
except Exception as e:
|
|
error_msg = f"Indicators: Error creating Bollinger Bands traces: {str(e)}"
|
|
self.logger.error(error_msg)
|
|
return [self.create_error_trace(error_msg)]
|
|
|
|
def _calculate_bollinger_bands(self, data: pd.DataFrame, period: int, std_dev: float) -> pd.DataFrame:
|
|
"""Calculate Bollinger Bands with validation"""
|
|
try:
|
|
result_df = data.copy()
|
|
|
|
# Calculate middle band (SMA)
|
|
result_df['middle_band'] = result_df['close'].rolling(window=period, min_periods=period).mean()
|
|
|
|
# Calculate standard deviation
|
|
result_df['std'] = result_df['close'].rolling(window=period, min_periods=period).std()
|
|
|
|
# Calculate upper and lower bands
|
|
result_df['upper_band'] = result_df['middle_band'] + (result_df['std'] * std_dev)
|
|
result_df['lower_band'] = result_df['middle_band'] - (result_df['std'] * std_dev)
|
|
|
|
# Remove NaN values
|
|
result_df = result_df.dropna(subset=['middle_band', 'upper_band', 'lower_band'])
|
|
|
|
if result_df.empty:
|
|
raise IndicatorCalculationError(ChartError(
|
|
code='BB_NO_VALUES',
|
|
message=f'Bollinger Bands calculation produced no values (period={period}, data_length={len(data)})',
|
|
severity=ErrorSeverity.ERROR,
|
|
context={'period': period, 'std_dev': std_dev, 'data_length': len(data)}
|
|
))
|
|
|
|
return result_df[['timestamp', 'upper_band', 'middle_band', 'lower_band']]
|
|
|
|
except Exception as e:
|
|
raise IndicatorCalculationError(ChartError(
|
|
code='BB_CALCULATION_ERROR',
|
|
message=f'Bollinger Bands calculation failed: {str(e)}',
|
|
severity=ErrorSeverity.ERROR,
|
|
context={'period': period, 'std_dev': std_dev, 'data_length': len(data), 'exception': str(e)}
|
|
))
|
|
|
|
def render(self, fig: go.Figure, data: pd.DataFrame, **kwargs) -> go.Figure:
|
|
"""Render Bollinger Bands layer for compatibility with base interface"""
|
|
try:
|
|
traces = self.create_traces(data.to_dict('records'), **kwargs)
|
|
for trace in traces:
|
|
if hasattr(fig, 'add_trace'):
|
|
fig.add_trace(trace, **kwargs)
|
|
else:
|
|
fig.add_trace(trace)
|
|
return fig
|
|
except Exception as e:
|
|
self.logger.error(f"Indicators: Error rendering Bollinger Bands layer: {e}")
|
|
return fig
|
|
|
|
|
|
def create_sma_layer(period: int = 20, **kwargs) -> SMALayer:
|
|
"""
|
|
Convenience function to create an SMA layer.
|
|
|
|
Args:
|
|
period: SMA period
|
|
**kwargs: Additional configuration options
|
|
|
|
Returns:
|
|
Configured SMA layer
|
|
"""
|
|
return SMALayer(period=period, **kwargs)
|
|
|
|
|
|
def create_ema_layer(period: int = 12, **kwargs) -> EMALayer:
|
|
"""
|
|
Convenience function to create an EMA layer.
|
|
|
|
Args:
|
|
period: EMA period
|
|
**kwargs: Additional configuration options
|
|
|
|
Returns:
|
|
Configured EMA layer
|
|
"""
|
|
return EMALayer(period=period, **kwargs)
|
|
|
|
|
|
def create_bollinger_bands_layer(period: int = 20, std_dev: float = 2.0, **kwargs) -> BollingerBandsLayer:
|
|
"""
|
|
Convenience function to create a Bollinger Bands layer.
|
|
|
|
Args:
|
|
period: BB period (default: 20)
|
|
std_dev: Standard deviation multiplier (default: 2.0)
|
|
**kwargs: Additional configuration options
|
|
|
|
Returns:
|
|
Configured Bollinger Bands layer
|
|
"""
|
|
return BollingerBandsLayer(period=period, std_dev=std_dev, **kwargs)
|
|
|
|
|
|
def create_common_ma_layers() -> List[BaseIndicatorLayer]:
|
|
"""
|
|
Create commonly used moving average layers.
|
|
|
|
Returns:
|
|
List of configured MA layers (SMA 20, SMA 50, EMA 12, EMA 26)
|
|
"""
|
|
colors = get_indicator_colors()
|
|
|
|
return [
|
|
SMALayer(20, color=colors.get('sma', '#007bff'), name="SMA(20)"),
|
|
SMALayer(50, color='#6c757d', name="SMA(50)"), # Gray for longer SMA
|
|
EMALayer(12, color=colors.get('ema', '#ff6b35'), name="EMA(12)"),
|
|
EMALayer(26, color='#28a745', name="EMA(26)") # Green for longer EMA
|
|
]
|
|
|
|
|
|
def create_common_overlay_indicators() -> List[BaseIndicatorLayer]:
|
|
"""
|
|
Create commonly used overlay indicators including moving averages and Bollinger Bands.
|
|
|
|
Returns:
|
|
List of configured overlay indicator layers
|
|
"""
|
|
colors = get_indicator_colors()
|
|
|
|
return [
|
|
SMALayer(20, color=colors.get('sma', '#007bff'), name="SMA(20)"),
|
|
EMALayer(12, color=colors.get('ema', '#ff6b35'), name="EMA(12)"),
|
|
BollingerBandsLayer(20, 2.0, color=colors.get('bb_upper', '#6f42c1'), name="BB(20,2)")
|
|
] |