""" 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, Literal from dataclasses import dataclass, field from datetime import datetime, timezone from decimal import Decimal import json from enum import Enum 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") class IndicatorType(str, Enum): """Supported indicator types.""" SMA = "sma" EMA = "ema" RSI = "rsi" MACD = "macd" BOLLINGER_BANDS = "bollinger_bands" class DisplayType(str, Enum): """Chart display types for indicators.""" OVERLAY = "overlay" SUBPLOT = "subplot" class LineStyle(str, Enum): """Available line styles for chart display.""" SOLID = "solid" DASH = "dash" DOT = "dot" DASHDOT = "dashdot" class PriceColumn(str, Enum): """Available price columns for calculations.""" OPEN = "open" HIGH = "high" LOW = "low" CLOSE = "close" @dataclass class IndicatorParameterSchema: """ Schema definition for an indicator parameter. """ name: str type: type required: bool = True default: Any = None min_value: Optional[Union[int, float]] = None max_value: Optional[Union[int, float]] = None description: str = "" def validate(self, value: Any) -> tuple[bool, str]: """ Validate a parameter value against this schema. Returns: Tuple of (is_valid, error_message) """ if value is None: if self.required: return False, f"Parameter '{self.name}' is required" return True, "" # Type validation if not isinstance(value, self.type): return False, f"Parameter '{self.name}' must be of type {self.type.__name__}, got {type(value).__name__}" # Range validation for numeric types if isinstance(value, (int, float)): if self.min_value is not None and value < self.min_value: return False, f"Parameter '{self.name}' must be >= {self.min_value}, got {value}" if self.max_value is not None and value > self.max_value: return False, f"Parameter '{self.name}' must be <= {self.max_value}, got {value}" return True, "" @dataclass class IndicatorSchema: """ Complete schema definition for an indicator type. """ indicator_type: IndicatorType display_type: DisplayType required_parameters: List[IndicatorParameterSchema] optional_parameters: List[IndicatorParameterSchema] = field(default_factory=list) min_data_points: int = 1 description: str = "" def get_parameter_schema(self, param_name: str) -> Optional[IndicatorParameterSchema]: """Get schema for a specific parameter.""" for param in self.required_parameters + self.optional_parameters: if param.name == param_name: return param return None def validate_parameters(self, parameters: Dict[str, Any]) -> tuple[bool, List[str]]: """ Validate all parameters against this schema. Returns: Tuple of (is_valid, list_of_error_messages) """ errors = [] # Check required parameters for param_schema in self.required_parameters: value = parameters.get(param_schema.name) is_valid, error = param_schema.validate(value) if not is_valid: errors.append(error) # Check optional parameters if provided for param_schema in self.optional_parameters: if param_schema.name in parameters: value = parameters[param_schema.name] is_valid, error = param_schema.validate(value) if not is_valid: errors.append(error) # Check for unknown parameters known_params = {p.name for p in self.required_parameters + self.optional_parameters} for param_name in parameters: if param_name not in known_params: errors.append(f"Unknown parameter '{param_name}' for {self.indicator_type.value} indicator") return len(errors) == 0, errors # Define schema for each indicator type INDICATOR_SCHEMAS = { IndicatorType.SMA: IndicatorSchema( indicator_type=IndicatorType.SMA, display_type=DisplayType.OVERLAY, required_parameters=[ IndicatorParameterSchema( name="period", type=int, min_value=1, max_value=200, description="Number of periods for moving average" ) ], optional_parameters=[ IndicatorParameterSchema( name="price_column", type=str, required=False, default="close", description="Price column to use (open, high, low, close)" ) ], min_data_points=1, description="Simple Moving Average - arithmetic mean of closing prices over a specified period" ), IndicatorType.EMA: IndicatorSchema( indicator_type=IndicatorType.EMA, display_type=DisplayType.OVERLAY, required_parameters=[ IndicatorParameterSchema( name="period", type=int, min_value=1, max_value=200, description="Number of periods for exponential moving average" ) ], optional_parameters=[ IndicatorParameterSchema( name="price_column", type=str, required=False, default="close", description="Price column to use (open, high, low, close)" ) ], min_data_points=1, description="Exponential Moving Average - gives more weight to recent prices" ), IndicatorType.RSI: IndicatorSchema( indicator_type=IndicatorType.RSI, display_type=DisplayType.SUBPLOT, required_parameters=[ IndicatorParameterSchema( name="period", type=int, min_value=2, max_value=100, description="Number of periods for RSI calculation" ) ], optional_parameters=[ IndicatorParameterSchema( name="price_column", type=str, required=False, default="close", description="Price column to use (open, high, low, close)" ) ], min_data_points=2, description="Relative Strength Index - momentum oscillator measuring speed and magnitude of price changes" ), IndicatorType.MACD: IndicatorSchema( indicator_type=IndicatorType.MACD, display_type=DisplayType.SUBPLOT, required_parameters=[ IndicatorParameterSchema( name="fast_period", type=int, min_value=1, max_value=50, description="Fast EMA period" ), IndicatorParameterSchema( name="slow_period", type=int, min_value=1, max_value=100, description="Slow EMA period" ), IndicatorParameterSchema( name="signal_period", type=int, min_value=1, max_value=50, description="Signal line EMA period" ) ], optional_parameters=[ IndicatorParameterSchema( name="price_column", type=str, required=False, default="close", description="Price column to use (open, high, low, close)" ) ], min_data_points=3, description="Moving Average Convergence Divergence - trend-following momentum indicator" ), IndicatorType.BOLLINGER_BANDS: IndicatorSchema( indicator_type=IndicatorType.BOLLINGER_BANDS, display_type=DisplayType.OVERLAY, required_parameters=[ IndicatorParameterSchema( name="period", type=int, min_value=2, max_value=100, description="Number of periods for moving average" ), IndicatorParameterSchema( name="std_dev", type=float, min_value=0.1, max_value=5.0, description="Number of standard deviations for bands" ) ], optional_parameters=[ IndicatorParameterSchema( name="price_column", type=str, required=False, default="close", description="Price column to use (open, high, low, close)" ) ], min_data_points=2, description="Bollinger Bands - volatility bands placed above and below a moving average" ) } @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 def validate(self) -> tuple[bool, List[str]]: """ Validate this indicator configuration against its schema. Returns: Tuple of (is_valid, list_of_error_messages) """ errors = [] # Check if indicator type is supported try: indicator_type = IndicatorType(self.indicator_type) except ValueError: return False, [f"Unsupported indicator type: {self.indicator_type}"] # Get schema for this indicator type schema = INDICATOR_SCHEMAS.get(indicator_type) if not schema: return False, [f"No schema found for indicator type: {self.indicator_type}"] # Validate parameters against schema is_valid, param_errors = schema.validate_parameters(self.parameters) if not is_valid: errors.extend(param_errors) # Validate display properties if self.display_type not in [DisplayType.OVERLAY.value, DisplayType.SUBPLOT.value]: errors.append(f"Invalid display_type: {self.display_type}") if self.line_style not in [style.value for style in LineStyle]: errors.append(f"Invalid line_style: {self.line_style}") if not isinstance(self.line_width, int) or self.line_width < 1: errors.append("line_width must be a positive integer") if not isinstance(self.opacity, (int, float)) or not (0.0 <= self.opacity <= 1.0): errors.append("opacity must be a number between 0.0 and 1.0") if self.display_type == DisplayType.SUBPLOT.value: if not isinstance(self.subplot_height_ratio, (int, float)) or not (0.1 <= self.subplot_height_ratio <= 1.0): errors.append("subplot_height_ratio must be a number between 0.1 and 1.0") return len(errors) == 0, errors # 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, {}) def validate_indicator_configuration(config: ChartIndicatorConfig) -> tuple[bool, List[str]]: """ Validate an indicator configuration against its schema. Args: config: Chart indicator configuration to validate Returns: Tuple of (is_valid, list_of_error_messages) """ return config.validate() def create_indicator_config( name: str, indicator_type: str, parameters: Dict[str, Any], display_type: Optional[str] = None, color: str = "#007bff", **display_options ) -> tuple[Optional[ChartIndicatorConfig], List[str]]: """ Create and validate a new indicator configuration. Args: name: Display name for the indicator indicator_type: Type of indicator (sma, ema, rsi, etc.) parameters: Indicator parameters display_type: Optional override for display type color: Color for chart display **display_options: Additional display configuration options Returns: Tuple of (config_object_or_None, list_of_error_messages) """ errors = [] # Validate indicator type try: indicator_enum = IndicatorType(indicator_type) except ValueError: return None, [f"Unsupported indicator type: {indicator_type}"] # Get schema for validation schema = INDICATOR_SCHEMAS.get(indicator_enum) if not schema: return None, [f"No schema found for indicator type: {indicator_type}"] # Use schema display type if not overridden if display_type is None: display_type = schema.display_type.value # Fill in default parameters final_parameters = {} # Add required parameters with defaults if missing for param_schema in schema.required_parameters: if param_schema.name in parameters: final_parameters[param_schema.name] = parameters[param_schema.name] elif param_schema.default is not None: final_parameters[param_schema.name] = param_schema.default # Required parameters without defaults will be caught by validation # Add optional parameters for param_schema in schema.optional_parameters: if param_schema.name in parameters: final_parameters[param_schema.name] = parameters[param_schema.name] elif param_schema.default is not None: final_parameters[param_schema.name] = param_schema.default # Create configuration config = ChartIndicatorConfig( name=name, indicator_type=indicator_type, parameters=final_parameters, display_type=display_type, color=color, line_style=display_options.get('line_style', 'solid'), line_width=display_options.get('line_width', 2), opacity=display_options.get('opacity', 1.0), visible=display_options.get('visible', True), subplot_height_ratio=display_options.get('subplot_height_ratio', 0.3) ) # Validate the configuration is_valid, validation_errors = config.validate() if not is_valid: return None, validation_errors return config, [] def get_indicator_schema(indicator_type: str) -> Optional[IndicatorSchema]: """ Get the schema for an indicator type. Args: indicator_type: Type of indicator Returns: IndicatorSchema object or None if not found """ try: indicator_enum = IndicatorType(indicator_type) return INDICATOR_SCHEMAS.get(indicator_enum) except ValueError: return None def get_available_indicator_types() -> List[str]: """ Get list of available indicator types. Returns: List of supported indicator type strings """ return [indicator_type.value for indicator_type in IndicatorType] def get_indicator_parameter_info(indicator_type: str) -> Dict[str, Dict[str, Any]]: """ Get detailed parameter information for an indicator type. Args: indicator_type: Type of indicator Returns: Dictionary with parameter information including types, ranges, and descriptions """ schema = get_indicator_schema(indicator_type) if not schema: return {} param_info = {} for param in schema.required_parameters + schema.optional_parameters: param_info[param.name] = { 'type': param.type.__name__, 'required': param.required, 'default': param.default, 'min_value': param.min_value, 'max_value': param.max_value, 'description': param.description } return param_info def validate_parameters_for_type(indicator_type: str, parameters: Dict[str, Any]) -> tuple[bool, List[str]]: """ Validate parameters for a specific indicator type. Args: indicator_type: Type of indicator parameters: Parameters to validate Returns: Tuple of (is_valid, list_of_error_messages) """ schema = get_indicator_schema(indicator_type) if not schema: return False, [f"Unknown indicator type: {indicator_type}"] return schema.validate_parameters(parameters) def create_configuration_from_json(json_data: Union[str, Dict[str, Any]]) -> tuple[Optional[ChartIndicatorConfig], List[str]]: """ Create indicator configuration from JSON data. Args: json_data: JSON string or dictionary with configuration data Returns: Tuple of (config_object_or_None, list_of_error_messages) """ try: if isinstance(json_data, str): data = json.loads(json_data) else: data = json_data required_fields = ['name', 'indicator_type', 'parameters'] missing_fields = [field for field in required_fields if field not in data] if missing_fields: return None, [f"Missing required fields: {', '.join(missing_fields)}"] return create_indicator_config( name=data['name'], indicator_type=data['indicator_type'], parameters=data['parameters'], display_type=data.get('display_type'), color=data.get('color', '#007bff'), **{k: v for k, v in data.items() if k not in ['name', 'indicator_type', 'parameters', 'display_type', 'color']} ) except json.JSONDecodeError as e: return None, [f"Invalid JSON: {e}"] except Exception as e: return None, [f"Error creating configuration: {e}"]