""" Indicator Management System This module provides functionality to manage user-defined indicators with file-based storage. Each indicator is saved as a separate JSON file for portability and easy sharing. """ import json import os import uuid from datetime import datetime, timezone from pathlib import Path from typing import Dict, List, Optional, Any, Tuple from dataclasses import dataclass, asdict from enum import Enum from utils.logger import get_logger # Initialize logger logger = get_logger() # Base directory for indicators INDICATORS_DIR = Path("config/indicators") USER_INDICATORS_DIR = INDICATORS_DIR / "user_indicators" TEMPLATES_DIR = INDICATORS_DIR / "templates" 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" @dataclass class IndicatorStyling: """Styling configuration for indicators.""" color: str = "#007bff" line_width: int = 2 opacity: float = 1.0 line_style: str = "solid" # solid, dash, dot, dashdot @dataclass class UserIndicator: """User-defined indicator configuration.""" id: str name: str description: str type: str # IndicatorType display_type: str # DisplayType parameters: Dict[str, Any] styling: IndicatorStyling timeframe: Optional[str] = None visible: bool = True created_date: str = "" modified_date: str = "" def __post_init__(self): """Initialize timestamps if not provided.""" current_time = datetime.now(timezone.utc).isoformat() if not self.created_date: self.created_date = current_time if not self.modified_date: self.modified_date = current_time def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for JSON serialization.""" return { 'id': self.id, 'name': self.name, 'description': self.description, 'type': self.type, 'display_type': self.display_type, 'parameters': self.parameters, 'styling': asdict(self.styling), 'timeframe': self.timeframe, 'visible': self.visible, 'created_date': self.created_date, 'modified_date': self.modified_date } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'UserIndicator': """Create UserIndicator from dictionary.""" styling_data = data.get('styling', {}) styling = IndicatorStyling(**styling_data) return cls( id=data['id'], name=data['name'], description=data.get('description', ''), type=data['type'], display_type=data['display_type'], parameters=data.get('parameters', {}), styling=styling, timeframe=data.get('timeframe'), visible=data.get('visible', True), created_date=data.get('created_date', ''), modified_date=data.get('modified_date', '') ) class IndicatorManager: """Manager for user-defined indicators with file-based storage.""" def __init__(self): """Initialize the indicator manager.""" self.logger = logger self._ensure_directories() self._create_default_templates() def _ensure_directories(self): """Ensure indicator directories exist.""" try: USER_INDICATORS_DIR.mkdir(parents=True, exist_ok=True) TEMPLATES_DIR.mkdir(parents=True, exist_ok=True) self.logger.debug("Indicator manager: Indicator directories created/verified") except Exception as e: self.logger.error(f"Indicator manager: Error creating indicator directories: {e}") def _get_indicator_file_path(self, indicator_id: str) -> Path: """Get file path for an indicator.""" return USER_INDICATORS_DIR / f"{indicator_id}.json" def _get_template_file_path(self, indicator_type: str) -> Path: """Get file path for an indicator template.""" return TEMPLATES_DIR / f"{indicator_type}_template.json" def save_indicator(self, indicator: UserIndicator) -> bool: """ Save an indicator to file. Args: indicator: UserIndicator instance to save Returns: True if saved successfully, False otherwise """ try: # Update modified date indicator.modified_date = datetime.now(timezone.utc).isoformat() file_path = self._get_indicator_file_path(indicator.id) with open(file_path, 'w', encoding='utf-8') as f: json.dump(indicator.to_dict(), f, indent=2, ensure_ascii=False) self.logger.info(f"Indicator manager: Saved indicator: {indicator.name} ({indicator.id})") return True except Exception as e: self.logger.error(f"Indicator manager: Error saving indicator {indicator.id}: {e}") return False def load_indicator(self, indicator_id: str) -> Optional[UserIndicator]: """ Load an indicator from file. Args: indicator_id: ID of the indicator to load Returns: UserIndicator instance or None if not found/error """ try: file_path = self._get_indicator_file_path(indicator_id) if not file_path.exists(): self.logger.warning(f"Indicator manager: Indicator file not found: {indicator_id}") return None with open(file_path, 'r', encoding='utf-8') as f: data = json.load(f) indicator = UserIndicator.from_dict(data) self.logger.debug(f"Indicator manager: Loaded indicator: {indicator.name} ({indicator.id})") return indicator except Exception as e: self.logger.error(f"Indicator manager: Error loading indicator {indicator_id}: {e}") return None def list_indicators(self, visible_only: bool = False) -> List[UserIndicator]: """ List all user indicators. Args: visible_only: If True, only return visible indicators Returns: List of UserIndicator instances """ indicators = [] try: for file_path in USER_INDICATORS_DIR.glob("*.json"): indicator_id = file_path.stem indicator = self.load_indicator(indicator_id) if indicator: if not visible_only or indicator.visible: indicators.append(indicator) # Sort by name indicators.sort(key=lambda x: x.name.lower()) self.logger.debug(f"Listed {len(indicators)} indicators") except Exception as e: self.logger.error(f"Indicator manager: Error listing indicators: {e}") return indicators def delete_indicator(self, indicator_id: str) -> bool: """ Delete an indicator. Args: indicator_id: ID of the indicator to delete Returns: True if deleted successfully, False otherwise """ try: file_path = self._get_indicator_file_path(indicator_id) if file_path.exists(): file_path.unlink() self.logger.info(f"Indicator manager: Deleted indicator: {indicator_id}") return True else: self.logger.warning(f"Indicator manager: Indicator file not found for deletion: {indicator_id}") return False except Exception as e: self.logger.error(f"Indicator manager: Error deleting indicator {indicator_id}: {e}") return False def create_indicator(self, name: str, indicator_type: str, parameters: Dict[str, Any], description: str = "", color: str = "#007bff", display_type: str = None, timeframe: Optional[str] = None) -> Optional[UserIndicator]: """ Create a new indicator. Args: name: Display name for the indicator indicator_type: Type of indicator (sma, ema, etc.) parameters: Indicator parameters description: Optional description color: Color for chart display display_type: overlay or subplot (auto-detected if None) timeframe: Optional timeframe for the indicator Returns: Created UserIndicator instance or None if error """ try: # Generate unique ID indicator_id = f"{indicator_type}_{uuid.uuid4().hex[:8]}" # Auto-detect display type if not provided if display_type is None: display_type = self._get_default_display_type(indicator_type) # Create styling styling = IndicatorStyling(color=color) # Create indicator indicator = UserIndicator( id=indicator_id, name=name, description=description, type=indicator_type, display_type=display_type, parameters=parameters, styling=styling, timeframe=timeframe ) # Save to file if self.save_indicator(indicator): self.logger.info(f"Indicator manager: Created new indicator: {name} ({indicator_id})") return indicator else: return None except Exception as e: self.logger.error(f"Indicator manager: Error creating indicator: {e}") return None def update_indicator(self, indicator_id: str, **updates) -> bool: """ Update an existing indicator. Args: indicator_id: ID of indicator to update **updates: Fields to update Returns: True if updated successfully, False otherwise """ try: indicator = self.load_indicator(indicator_id) if not indicator: return False # Update fields for key, value in updates.items(): if hasattr(indicator, key): if key == 'styling' and isinstance(value, dict): # Update nested styling fields for style_key, style_value in value.items(): if hasattr(indicator.styling, style_key): setattr(indicator.styling, style_key, style_value) elif key == 'parameters' and isinstance(value, dict): indicator.parameters.update(value) else: setattr(indicator, key, value) # Save updated indicator return self.save_indicator(indicator) except Exception as e: self.logger.error(f"Indicator manager: Error updating indicator {indicator_id}: {e}") return False def get_indicators_by_type(self, display_type: str) -> List[UserIndicator]: """Get indicators by display type (overlay/subplot).""" indicators = self.list_indicators(visible_only=True) return [ind for ind in indicators if ind.display_type == display_type] def get_available_indicator_types(self) -> List[str]: """Get list of available indicator types.""" return [t.value for t in IndicatorType] def _get_default_display_type(self, indicator_type: str) -> str: """Get default display type for an indicator type.""" overlay_types = {IndicatorType.SMA, IndicatorType.EMA, IndicatorType.BOLLINGER_BANDS} subplot_types = {IndicatorType.RSI, IndicatorType.MACD} if indicator_type in [t.value for t in overlay_types]: return DisplayType.OVERLAY.value elif indicator_type in [t.value for t in subplot_types]: return DisplayType.SUBPLOT.value else: return DisplayType.OVERLAY.value # Default def _create_default_templates(self): """Create default indicator templates if they don't exist.""" templates = { IndicatorType.SMA.value: { "name": "Simple Moving Average", "description": "Simple Moving Average indicator", "type": IndicatorType.SMA.value, "display_type": DisplayType.OVERLAY.value, "default_parameters": {"period": 20}, "parameter_schema": { "period": {"type": "int", "min": 1, "max": 200, "default": 20, "description": "Period for SMA calculation"} }, "default_styling": {"color": "#007bff", "line_width": 2} }, IndicatorType.EMA.value: { "name": "Exponential Moving Average", "description": "Exponential Moving Average indicator", "type": IndicatorType.EMA.value, "display_type": DisplayType.OVERLAY.value, "default_parameters": {"period": 12}, "parameter_schema": { "period": {"type": "int", "min": 1, "max": 200, "default": 12, "description": "Period for EMA calculation"} }, "default_styling": {"color": "#ff6b35", "line_width": 2} }, IndicatorType.RSI.value: { "name": "Relative Strength Index", "description": "RSI oscillator indicator", "type": IndicatorType.RSI.value, "display_type": DisplayType.SUBPLOT.value, "default_parameters": {"period": 14}, "parameter_schema": { "period": {"type": "int", "min": 2, "max": 50, "default": 14, "description": "Period for RSI calculation"} }, "default_styling": {"color": "#20c997", "line_width": 2} }, IndicatorType.MACD.value: { "name": "MACD", "description": "Moving Average Convergence Divergence", "type": IndicatorType.MACD.value, "display_type": DisplayType.SUBPLOT.value, "default_parameters": {"fast_period": 12, "slow_period": 26, "signal_period": 9}, "parameter_schema": { "fast_period": {"type": "int", "min": 2, "max": 50, "default": 12, "description": "Fast EMA period"}, "slow_period": {"type": "int", "min": 5, "max": 100, "default": 26, "description": "Slow EMA period"}, "signal_period": {"type": "int", "min": 2, "max": 30, "default": 9, "description": "Signal line period"} }, "default_styling": {"color": "#fd7e14", "line_width": 2} }, IndicatorType.BOLLINGER_BANDS.value: { "name": "Bollinger Bands", "description": "Bollinger Bands volatility indicator", "type": IndicatorType.BOLLINGER_BANDS.value, "display_type": DisplayType.OVERLAY.value, "default_parameters": {"period": 20, "std_dev": 2.0}, "parameter_schema": { "period": {"type": "int", "min": 5, "max": 100, "default": 20, "description": "Period for middle line (SMA)"}, "std_dev": {"type": "float", "min": 0.5, "max": 5.0, "default": 2.0, "description": "Standard deviation multiplier"} }, "default_styling": {"color": "#6f42c1", "line_width": 1} } } for indicator_type, template_data in templates.items(): template_path = self._get_template_file_path(indicator_type) if not template_path.exists(): try: with open(template_path, 'w', encoding='utf-8') as f: json.dump(template_data, f, indent=2, ensure_ascii=False) self.logger.debug(f"Created template: {indicator_type}") except Exception as e: self.logger.error(f"Indicator manager: Error creating template {indicator_type}: {e}") def get_template(self, indicator_type: str) -> Optional[Dict[str, Any]]: """Get indicator template by type.""" try: template_path = self._get_template_file_path(indicator_type) if template_path.exists(): with open(template_path, 'r', encoding='utf-8') as f: return json.load(f) else: self.logger.warning(f"Template not found: {indicator_type}") return None except Exception as e: self.logger.error(f"Indicator manager: Error loading template {indicator_type}: {e}") return None # Global instance indicator_manager = IndicatorManager() def get_indicator_manager() -> IndicatorManager: """Get the global indicator manager instance.""" return indicator_manager