""" Indicator Definitions and Configuration This module defines indicator configurations and provides integration with the existing data/common/indicators.py technical indicators module. """ from typing import Dict, List, Any, Optional, Union from dataclasses import dataclass from datetime import datetime, timezone from decimal import Decimal from data.common.indicators import TechnicalIndicators, IndicatorResult, create_default_indicators_config, validate_indicator_config from data.common.data_types import OHLCVCandle from utils.logger import get_logger # Initialize logger logger = get_logger("indicator_defs") @dataclass class ChartIndicatorConfig: """ Configuration for chart indicators with display properties. Extends the base indicator config with chart-specific properties like colors, line styles, and subplot placement. """ name: str indicator_type: str parameters: Dict[str, Any] display_type: str # 'overlay', 'subplot' color: str line_style: str = 'solid' # 'solid', 'dash', 'dot' line_width: int = 2 opacity: float = 1.0 visible: bool = True subplot_height_ratio: float = 0.3 # For subplot indicators def to_indicator_config(self) -> Dict[str, Any]: """Convert to format expected by TechnicalIndicators.""" config = {'type': self.indicator_type} config.update(self.parameters) return config # Built-in indicator definitions with chart display properties INDICATOR_DEFINITIONS = { 'sma_20': ChartIndicatorConfig( name='SMA (20)', indicator_type='sma', parameters={'period': 20, 'price_column': 'close'}, display_type='overlay', color='#007bff', line_width=2 ), 'sma_50': ChartIndicatorConfig( name='SMA (50)', indicator_type='sma', parameters={'period': 50, 'price_column': 'close'}, display_type='overlay', color='#28a745', line_width=2 ), 'ema_12': ChartIndicatorConfig( name='EMA (12)', indicator_type='ema', parameters={'period': 12, 'price_column': 'close'}, display_type='overlay', color='#ff6b35', line_width=2 ), 'ema_26': ChartIndicatorConfig( name='EMA (26)', indicator_type='ema', parameters={'period': 26, 'price_column': 'close'}, display_type='overlay', color='#dc3545', line_width=2 ), 'rsi_14': ChartIndicatorConfig( name='RSI (14)', indicator_type='rsi', parameters={'period': 14, 'price_column': 'close'}, display_type='subplot', color='#20c997', line_width=2, subplot_height_ratio=0.25 ), 'macd_default': ChartIndicatorConfig( name='MACD', indicator_type='macd', parameters={'fast_period': 12, 'slow_period': 26, 'signal_period': 9, 'price_column': 'close'}, display_type='subplot', color='#fd7e14', line_width=2, subplot_height_ratio=0.3 ), 'bollinger_bands': ChartIndicatorConfig( name='Bollinger Bands', indicator_type='bollinger_bands', parameters={'period': 20, 'std_dev': 2.0, 'price_column': 'close'}, display_type='overlay', color='#6f42c1', line_width=1, opacity=0.7 ) } def convert_database_candles_to_ohlcv(candles: List[Dict[str, Any]]) -> List[OHLCVCandle]: """ Convert database candle dictionaries to OHLCVCandle objects. Args: candles: List of candle dictionaries from database operations Returns: List of OHLCVCandle objects for technical indicators """ ohlcv_candles = [] for candle in candles: try: # Handle timestamp conversion timestamp = candle['timestamp'] if isinstance(timestamp, str): timestamp = datetime.fromisoformat(timestamp.replace('Z', '+00:00')) elif timestamp.tzinfo is None: timestamp = timestamp.replace(tzinfo=timezone.utc) # For database candles, start_time and end_time are the same # as we store right-aligned timestamps ohlcv_candle = OHLCVCandle( symbol=candle['symbol'], timeframe=candle['timeframe'], start_time=timestamp, end_time=timestamp, open=Decimal(str(candle['open'])), high=Decimal(str(candle['high'])), low=Decimal(str(candle['low'])), close=Decimal(str(candle['close'])), volume=Decimal(str(candle.get('volume', 0))), trade_count=candle.get('trades_count', 0), exchange=candle.get('exchange', 'okx'), is_complete=True ) ohlcv_candles.append(ohlcv_candle) except Exception as e: logger.error(f"Error converting candle to OHLCV: {e}") continue logger.debug(f"Converted {len(ohlcv_candles)} database candles to OHLCV format") return ohlcv_candles def calculate_indicators(candles: List[Dict[str, Any]], indicator_configs: List[str], custom_configs: Optional[Dict[str, ChartIndicatorConfig]] = None) -> Dict[str, List[IndicatorResult]]: """ Calculate technical indicators for chart display. Args: candles: List of candle dictionaries from database indicator_configs: List of indicator names to calculate custom_configs: Optional custom indicator configurations Returns: Dictionary mapping indicator names to their calculation results """ if not candles: logger.warning("No candles provided for indicator calculation") return {} # Convert to OHLCV format ohlcv_candles = convert_database_candles_to_ohlcv(candles) if not ohlcv_candles: logger.error("Failed to convert candles to OHLCV format") return {} # Initialize technical indicators calculator indicators_calc = TechnicalIndicators(logger) # Prepare configurations configs_to_calculate = {} all_configs = {**INDICATOR_DEFINITIONS} if custom_configs: all_configs.update(custom_configs) for indicator_name in indicator_configs: if indicator_name in all_configs: chart_config = all_configs[indicator_name] configs_to_calculate[indicator_name] = chart_config.to_indicator_config() else: logger.warning(f"Unknown indicator configuration: {indicator_name}") if not configs_to_calculate: logger.warning("No valid indicator configurations found") return {} # Calculate indicators try: results = indicators_calc.calculate_multiple_indicators(ohlcv_candles, configs_to_calculate) logger.debug(f"Calculated {len(results)} indicators successfully") return results except Exception as e: logger.error(f"Error calculating indicators: {e}") return {} def get_indicator_display_config(indicator_name: str) -> Optional[ChartIndicatorConfig]: """ Get display configuration for an indicator. Args: indicator_name: Name of the indicator Returns: Chart indicator configuration or None if not found """ return INDICATOR_DEFINITIONS.get(indicator_name) def get_available_indicators() -> Dict[str, str]: """ Get list of available indicators with descriptions. Returns: Dictionary mapping indicator names to descriptions """ return {name: config.name for name, config in INDICATOR_DEFINITIONS.items()} def get_overlay_indicators() -> List[str]: """Get list of indicators that display as overlays on the price chart.""" return [name for name, config in INDICATOR_DEFINITIONS.items() if config.display_type == 'overlay'] def get_subplot_indicators() -> List[str]: """Get list of indicators that display in separate subplots.""" return [name for name, config in INDICATOR_DEFINITIONS.items() if config.display_type == 'subplot'] def get_default_indicator_params(indicator_type: str) -> Dict[str, Any]: """ Get default parameters for an indicator type. Args: indicator_type: Type of indicator ('sma', 'ema', 'rsi', etc.) Returns: Dictionary of default parameters """ defaults = { 'sma': {'period': 20, 'price_column': 'close'}, 'ema': {'period': 20, 'price_column': 'close'}, 'rsi': {'period': 14, 'price_column': 'close'}, 'macd': {'fast_period': 12, 'slow_period': 26, 'signal_period': 9, 'price_column': 'close'}, 'bollinger_bands': {'period': 20, 'std_dev': 2.0, 'price_column': 'close'} } return defaults.get(indicator_type, {})