b49e39dcb4
- Enhanced the `UserIndicator` class to include an optional `timeframe` attribute for custom indicator timeframes. - Updated the `get_indicator_data` method in `MarketDataIntegrator` to fetch and calculate indicators based on the specified timeframe, ensuring proper data alignment and handling. - Modified the `ChartBuilder` to pass the correct DataFrame for plotting indicators with different timeframes. - Added UI elements in the indicator modal for selecting timeframes, improving user experience. - Updated relevant JSON templates to include the new `timeframe` field for all indicators. - Refactored the `prepare_chart_data` function to ensure it returns a DataFrame with a `DatetimeIndex` for consistent calculations. This commit enhances the flexibility and usability of the indicator system, allowing users to analyze data across various timeframes.
454 lines
17 KiB
Python
454 lines
17 KiB
Python
"""
|
|
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("default_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 |