3.4 Implement user-defined indicator management system and enhance chart capabilities
- Introduced a comprehensive user indicator management system in `components/charts/indicator_manager.py`, allowing users to create, edit, and manage custom indicators with JSON persistence. - Added new default indicators in `components/charts/indicator_defaults.py` to provide users with immediate options for technical analysis. - Enhanced the chart rendering capabilities by implementing the `create_chart_with_indicators` function in `components/charts/builder.py`, supporting both overlay and subplot indicators. - Updated the main application layout in `app.py` to include a modal for adding and editing indicators, improving user interaction. - Enhanced documentation to cover the new indicator system, including a quick guide for adding new indicators and detailed usage examples. - Added unit tests to ensure the reliability and functionality of the new indicator management features.
This commit is contained in:
Vasily.onl
@@ -14,6 +14,7 @@ Main Components:
|
||||
"""
|
||||
|
||||
import plotly.graph_objects as go
|
||||
from typing import List
|
||||
from .builder import ChartBuilder
|
||||
from .utils import (
|
||||
validate_market_data,
|
||||
@@ -137,7 +138,8 @@ __all__ = [
|
||||
|
||||
# Convenience functions
|
||||
"create_basic_chart",
|
||||
"create_indicator_chart"
|
||||
"create_indicator_chart",
|
||||
"create_chart_with_indicators"
|
||||
]
|
||||
|
||||
# Initialize logger
|
||||
@@ -447,4 +449,27 @@ def create_indicator_chart(symbol: str, data: list,
|
||||
return create_basic_chart(symbol, data, indicators=[indicator_config])
|
||||
|
||||
except Exception as e:
|
||||
return create_basic_chart(symbol, data, indicators=[]) # Fallback to basic chart
|
||||
return create_basic_chart(symbol, data, indicators=[]) # Fallback to basic chart
|
||||
|
||||
def create_chart_with_indicators(symbol: str, timeframe: str,
|
||||
overlay_indicators: List[str] = None,
|
||||
subplot_indicators: List[str] = None,
|
||||
days_back: int = 7, **kwargs) -> go.Figure:
|
||||
"""
|
||||
Create a chart with dynamically selected indicators.
|
||||
|
||||
Args:
|
||||
symbol: Trading pair (e.g., 'BTC-USDT')
|
||||
timeframe: Timeframe (e.g., '1h', '1d')
|
||||
overlay_indicators: List of overlay indicator names
|
||||
subplot_indicators: List of subplot indicator names
|
||||
days_back: Number of days to look back
|
||||
**kwargs: Additional chart parameters
|
||||
|
||||
Returns:
|
||||
Plotly figure with selected indicators
|
||||
"""
|
||||
builder = ChartBuilder()
|
||||
return builder.create_chart_with_indicators(
|
||||
symbol, timeframe, overlay_indicators, subplot_indicators, days_back, **kwargs
|
||||
)
|
||||
@@ -349,4 +349,253 @@ class ChartBuilder:
|
||||
'data_age_minutes': None,
|
||||
'sufficient_for_indicators': False,
|
||||
'message': f"Error checking data: {str(e)}"
|
||||
}
|
||||
}
|
||||
|
||||
def create_chart_with_indicators(self, symbol: str, timeframe: str,
|
||||
overlay_indicators: List[str] = None,
|
||||
subplot_indicators: List[str] = None,
|
||||
days_back: int = 7, **kwargs) -> go.Figure:
|
||||
"""
|
||||
Create a chart with dynamically selected indicators.
|
||||
|
||||
Args:
|
||||
symbol: Trading pair
|
||||
timeframe: Timeframe
|
||||
overlay_indicators: List of overlay indicator names
|
||||
subplot_indicators: List of subplot indicator names
|
||||
days_back: Number of days to look back
|
||||
**kwargs: Additional chart parameters
|
||||
|
||||
Returns:
|
||||
Plotly Figure object with selected indicators
|
||||
"""
|
||||
try:
|
||||
# Fetch market data
|
||||
candles = self.fetch_market_data_enhanced(symbol, timeframe, days_back)
|
||||
|
||||
if not candles:
|
||||
self.logger.warning(f"No data available for {symbol} {timeframe}")
|
||||
return self._create_empty_chart(f"No data available for {symbol} {timeframe}")
|
||||
|
||||
# Validate and prepare data
|
||||
if not validate_market_data(candles):
|
||||
self.logger.error(f"Invalid market data for {symbol} {timeframe}")
|
||||
return self._create_error_chart("Invalid market data format")
|
||||
|
||||
df = prepare_chart_data(candles)
|
||||
|
||||
# Import layer classes
|
||||
from .layers import (
|
||||
LayerManager, CandlestickLayer, VolumeLayer,
|
||||
SMALayer, EMALayer, BollingerBandsLayer,
|
||||
RSILayer, MACDLayer, IndicatorLayerConfig
|
||||
)
|
||||
from .indicator_manager import get_indicator_manager
|
||||
|
||||
# Get user indicators instead of default configurations
|
||||
indicator_manager = get_indicator_manager()
|
||||
|
||||
# Calculate subplot requirements
|
||||
subplot_count = 0
|
||||
volume_enabled = 'volume' in df.columns and df['volume'].sum() > 0
|
||||
if volume_enabled:
|
||||
subplot_count += 1
|
||||
|
||||
if subplot_indicators:
|
||||
subplot_count += len(subplot_indicators)
|
||||
|
||||
# Create subplot structure if needed
|
||||
if subplot_count > 0:
|
||||
# Calculate height ratios
|
||||
main_height = 0.7 # Main chart gets 70%
|
||||
subplot_height = 0.3 / subplot_count if subplot_count > 0 else 0
|
||||
|
||||
# Create subplot specifications
|
||||
subplot_specs = [[{"secondary_y": False}]] # Main chart
|
||||
row_heights = [main_height]
|
||||
|
||||
if volume_enabled:
|
||||
subplot_specs.append([{"secondary_y": False}])
|
||||
row_heights.append(subplot_height)
|
||||
|
||||
if subplot_indicators:
|
||||
for _ in subplot_indicators:
|
||||
subplot_specs.append([{"secondary_y": False}])
|
||||
row_heights.append(subplot_height)
|
||||
|
||||
# Create subplots figure
|
||||
from plotly.subplots import make_subplots
|
||||
fig = make_subplots(
|
||||
rows=len(subplot_specs),
|
||||
cols=1,
|
||||
shared_xaxes=True,
|
||||
vertical_spacing=0.02,
|
||||
row_heights=row_heights,
|
||||
specs=subplot_specs,
|
||||
subplot_titles=[f"{symbol} - {timeframe}"] + [""] * (len(subplot_specs) - 1)
|
||||
)
|
||||
else:
|
||||
# Create simple figure for main chart only
|
||||
fig = go.Figure()
|
||||
|
||||
current_row = 1
|
||||
|
||||
# Add candlestick layer (always included)
|
||||
candlestick_trace = go.Candlestick(
|
||||
x=df['timestamp'],
|
||||
open=df['open'],
|
||||
high=df['high'],
|
||||
low=df['low'],
|
||||
close=df['close'],
|
||||
name=symbol,
|
||||
increasing_line_color=self.default_colors['bullish'],
|
||||
decreasing_line_color=self.default_colors['bearish'],
|
||||
showlegend=False
|
||||
)
|
||||
fig.add_trace(candlestick_trace, row=current_row, col=1)
|
||||
|
||||
# Add overlay indicators
|
||||
if overlay_indicators:
|
||||
for indicator_id in overlay_indicators:
|
||||
try:
|
||||
# Load user indicator
|
||||
user_indicator = indicator_manager.load_indicator(indicator_id)
|
||||
|
||||
if user_indicator is None:
|
||||
self.logger.warning(f"Overlay indicator {indicator_id} not found")
|
||||
continue
|
||||
|
||||
# Create appropriate indicator layer using user configuration
|
||||
if user_indicator.type == 'sma':
|
||||
period = user_indicator.parameters.get('period', 20)
|
||||
layer_config = IndicatorLayerConfig(
|
||||
name=user_indicator.name,
|
||||
indicator_type='sma',
|
||||
color=user_indicator.styling.color,
|
||||
parameters={'period': period},
|
||||
line_width=user_indicator.styling.line_width
|
||||
)
|
||||
sma_layer = SMALayer(layer_config)
|
||||
traces = sma_layer.create_traces(df.to_dict('records'))
|
||||
for trace in traces:
|
||||
fig.add_trace(trace, row=current_row, col=1)
|
||||
|
||||
elif user_indicator.type == 'ema':
|
||||
period = user_indicator.parameters.get('period', 12)
|
||||
layer_config = IndicatorLayerConfig(
|
||||
name=user_indicator.name,
|
||||
indicator_type='ema',
|
||||
color=user_indicator.styling.color,
|
||||
parameters={'period': period},
|
||||
line_width=user_indicator.styling.line_width
|
||||
)
|
||||
ema_layer = EMALayer(layer_config)
|
||||
traces = ema_layer.create_traces(df.to_dict('records'))
|
||||
for trace in traces:
|
||||
fig.add_trace(trace, row=current_row, col=1)
|
||||
|
||||
elif user_indicator.type == 'bollinger_bands':
|
||||
period = user_indicator.parameters.get('period', 20)
|
||||
std_dev = user_indicator.parameters.get('std_dev', 2.0)
|
||||
layer_config = IndicatorLayerConfig(
|
||||
name=user_indicator.name,
|
||||
indicator_type='bollinger_bands',
|
||||
color=user_indicator.styling.color,
|
||||
parameters={'period': period, 'std_dev': std_dev},
|
||||
line_width=user_indicator.styling.line_width,
|
||||
show_middle_line=True
|
||||
)
|
||||
bb_layer = BollingerBandsLayer(layer_config)
|
||||
traces = bb_layer.create_traces(df.to_dict('records'))
|
||||
for trace in traces:
|
||||
fig.add_trace(trace, row=current_row, col=1)
|
||||
|
||||
self.logger.debug(f"Added overlay indicator: {user_indicator.name}")
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error adding overlay indicator {indicator_id}: {e}")
|
||||
|
||||
# Move to next row for volume if enabled
|
||||
if volume_enabled:
|
||||
current_row += 1
|
||||
volume_colors = [self.default_colors['bullish'] if close >= open else self.default_colors['bearish']
|
||||
for close, open in zip(df['close'], df['open'])]
|
||||
|
||||
volume_trace = go.Bar(
|
||||
x=df['timestamp'],
|
||||
y=df['volume'],
|
||||
name='Volume',
|
||||
marker_color=volume_colors,
|
||||
opacity=0.7,
|
||||
showlegend=False
|
||||
)
|
||||
fig.add_trace(volume_trace, row=current_row, col=1)
|
||||
fig.update_yaxes(title_text="Volume", row=current_row, col=1)
|
||||
|
||||
# Add subplot indicators
|
||||
if subplot_indicators:
|
||||
for indicator_id in subplot_indicators:
|
||||
current_row += 1
|
||||
try:
|
||||
# Load user indicator
|
||||
user_indicator = indicator_manager.load_indicator(indicator_id)
|
||||
|
||||
if user_indicator is None:
|
||||
self.logger.warning(f"Subplot indicator {indicator_id} not found")
|
||||
continue
|
||||
|
||||
# Create appropriate subplot indicator layer
|
||||
if user_indicator.type == 'rsi':
|
||||
period = user_indicator.parameters.get('period', 14)
|
||||
rsi_layer = RSILayer(period=period, color=user_indicator.styling.color, name=user_indicator.name)
|
||||
|
||||
# Use the render method
|
||||
fig = rsi_layer.render(fig, df, row=current_row, col=1)
|
||||
|
||||
# Add RSI reference lines
|
||||
fig.add_hline(y=70, line_dash="dash", line_color="red", opacity=0.5, row=current_row, col=1)
|
||||
fig.add_hline(y=30, line_dash="dash", line_color="green", opacity=0.5, row=current_row, col=1)
|
||||
fig.update_yaxes(title_text="RSI", range=[0, 100], row=current_row, col=1)
|
||||
|
||||
elif user_indicator.type == 'macd':
|
||||
fast_period = user_indicator.parameters.get('fast_period', 12)
|
||||
slow_period = user_indicator.parameters.get('slow_period', 26)
|
||||
signal_period = user_indicator.parameters.get('signal_period', 9)
|
||||
macd_layer = MACDLayer(fast_period=fast_period, slow_period=slow_period,
|
||||
signal_period=signal_period, color=user_indicator.styling.color, name=user_indicator.name)
|
||||
|
||||
# Use the render method
|
||||
fig = macd_layer.render(fig, df, row=current_row, col=1)
|
||||
|
||||
# Add zero line for MACD
|
||||
fig.add_hline(y=0, line_dash="dash", line_color="gray", opacity=0.5, row=current_row, col=1)
|
||||
fig.update_yaxes(title_text="MACD", row=current_row, col=1)
|
||||
|
||||
self.logger.debug(f"Added subplot indicator: {user_indicator.name}")
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error adding subplot indicator {indicator_id}: {e}")
|
||||
|
||||
# Update layout
|
||||
height = kwargs.get('height', self.default_height)
|
||||
template = kwargs.get('template', self.default_template)
|
||||
|
||||
fig.update_layout(
|
||||
title=f"{symbol} - {timeframe} Chart",
|
||||
template=template,
|
||||
height=height,
|
||||
showlegend=True,
|
||||
legend=dict(yanchor="top", y=0.99, xanchor="left", x=0.01),
|
||||
xaxis_rangeslider_visible=False,
|
||||
hovermode='x unified'
|
||||
)
|
||||
|
||||
# Update x-axis for all subplots
|
||||
fig.update_xaxes(title_text="Time", row=current_row, col=1)
|
||||
fig.update_yaxes(title_text="Price (USDT)", row=1, col=1)
|
||||
|
||||
indicator_count = len(overlay_indicators or []) + len(subplot_indicators or [])
|
||||
self.logger.debug(f"Created chart for {symbol} {timeframe} with {indicator_count} indicators")
|
||||
return fig
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error creating chart with indicators: {e}")
|
||||
return self._create_error_chart(f"Chart creation failed: {str(e)}")
|
||||
133
components/charts/indicator_defaults.py
Normal file
133
components/charts/indicator_defaults.py
Normal file
@@ -0,0 +1,133 @@
|
||||
"""
|
||||
Default Indicator Creation
|
||||
|
||||
This module creates a set of default indicators that users can start with.
|
||||
These are common indicator configurations that are immediately useful.
|
||||
"""
|
||||
|
||||
from .indicator_manager import get_indicator_manager, IndicatorType, DisplayType
|
||||
|
||||
|
||||
def create_default_indicators():
|
||||
"""Create default indicators if they don't exist."""
|
||||
manager = get_indicator_manager()
|
||||
|
||||
# Check if we already have indicators
|
||||
existing_indicators = manager.list_indicators()
|
||||
if existing_indicators:
|
||||
manager.logger.info(f"Found {len(existing_indicators)} existing indicators, skipping defaults creation")
|
||||
return
|
||||
|
||||
# Define default indicators
|
||||
default_indicators = [
|
||||
# Moving Averages
|
||||
{
|
||||
"name": "SMA 20",
|
||||
"description": "20-period Simple Moving Average for short-term trend",
|
||||
"type": IndicatorType.SMA.value,
|
||||
"parameters": {"period": 20},
|
||||
"color": "#007bff"
|
||||
},
|
||||
{
|
||||
"name": "SMA 50",
|
||||
"description": "50-period Simple Moving Average for medium-term trend",
|
||||
"type": IndicatorType.SMA.value,
|
||||
"parameters": {"period": 50},
|
||||
"color": "#6c757d"
|
||||
},
|
||||
{
|
||||
"name": "EMA 12",
|
||||
"description": "12-period Exponential Moving Average for fast signals",
|
||||
"type": IndicatorType.EMA.value,
|
||||
"parameters": {"period": 12},
|
||||
"color": "#ff6b35"
|
||||
},
|
||||
{
|
||||
"name": "EMA 26",
|
||||
"description": "26-period Exponential Moving Average for slower signals",
|
||||
"type": IndicatorType.EMA.value,
|
||||
"parameters": {"period": 26},
|
||||
"color": "#28a745"
|
||||
},
|
||||
|
||||
# Oscillators
|
||||
{
|
||||
"name": "RSI 14",
|
||||
"description": "14-period RSI for momentum analysis",
|
||||
"type": IndicatorType.RSI.value,
|
||||
"parameters": {"period": 14},
|
||||
"color": "#20c997"
|
||||
},
|
||||
{
|
||||
"name": "RSI 21",
|
||||
"description": "21-period RSI for less sensitive momentum signals",
|
||||
"type": IndicatorType.RSI.value,
|
||||
"parameters": {"period": 21},
|
||||
"color": "#17a2b8"
|
||||
},
|
||||
|
||||
# MACD Variants
|
||||
{
|
||||
"name": "MACD Standard",
|
||||
"description": "Standard MACD (12, 26, 9) for trend changes",
|
||||
"type": IndicatorType.MACD.value,
|
||||
"parameters": {"fast_period": 12, "slow_period": 26, "signal_period": 9},
|
||||
"color": "#fd7e14"
|
||||
},
|
||||
{
|
||||
"name": "MACD Fast",
|
||||
"description": "Fast MACD (5, 13, 4) for quick signals",
|
||||
"type": IndicatorType.MACD.value,
|
||||
"parameters": {"fast_period": 5, "slow_period": 13, "signal_period": 4},
|
||||
"color": "#dc3545"
|
||||
},
|
||||
|
||||
# Bollinger Bands
|
||||
{
|
||||
"name": "Bollinger Bands",
|
||||
"description": "Standard Bollinger Bands (20, 2) for volatility analysis",
|
||||
"type": IndicatorType.BOLLINGER_BANDS.value,
|
||||
"parameters": {"period": 20, "std_dev": 2.0},
|
||||
"color": "#6f42c1"
|
||||
},
|
||||
{
|
||||
"name": "Bollinger Tight",
|
||||
"description": "Tight Bollinger Bands (20, 1.5) for sensitive volatility",
|
||||
"type": IndicatorType.BOLLINGER_BANDS.value,
|
||||
"parameters": {"period": 20, "std_dev": 1.5},
|
||||
"color": "#e83e8c"
|
||||
}
|
||||
]
|
||||
|
||||
# Create indicators
|
||||
created_count = 0
|
||||
for indicator_config in default_indicators:
|
||||
indicator = manager.create_indicator(
|
||||
name=indicator_config["name"],
|
||||
indicator_type=indicator_config["type"],
|
||||
parameters=indicator_config["parameters"],
|
||||
description=indicator_config["description"],
|
||||
color=indicator_config["color"]
|
||||
)
|
||||
|
||||
if indicator:
|
||||
created_count += 1
|
||||
manager.logger.info(f"Created default indicator: {indicator.name}")
|
||||
else:
|
||||
manager.logger.error(f"Failed to create indicator: {indicator_config['name']}")
|
||||
|
||||
manager.logger.info(f"Created {created_count} default indicators")
|
||||
|
||||
|
||||
def ensure_default_indicators():
|
||||
"""Ensure default indicators exist (called during app startup)."""
|
||||
try:
|
||||
create_default_indicators()
|
||||
except Exception as e:
|
||||
manager = get_indicator_manager()
|
||||
manager.logger.error(f"Error creating default indicators: {e}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Create defaults when run directly
|
||||
create_default_indicators()
|
||||
446
components/charts/indicator_manager.py
Normal file
446
components/charts/indicator_manager.py
Normal file
@@ -0,0 +1,446 @@
|
||||
"""
|
||||
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("indicator_manager")
|
||||
|
||||
# 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
|
||||
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),
|
||||
'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,
|
||||
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 directories created/verified")
|
||||
except Exception as e:
|
||||
self.logger.error(f"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"Saved indicator: {indicator.name} ({indicator.id})")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"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 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"Loaded indicator: {indicator.name} ({indicator.id})")
|
||||
return indicator
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"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"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"Deleted indicator: {indicator_id}")
|
||||
return True
|
||||
else:
|
||||
self.logger.warning(f"Indicator file not found for deletion: {indicator_id}")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"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) -> 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)
|
||||
|
||||
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
|
||||
)
|
||||
|
||||
# Save to file
|
||||
if self.save_indicator(indicator):
|
||||
self.logger.info(f"Created new indicator: {name} ({indicator_id})")
|
||||
return indicator
|
||||
else:
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"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 field, value in updates.items():
|
||||
if hasattr(indicator, field):
|
||||
if field == 'styling' and isinstance(value, dict):
|
||||
# Update styling fields
|
||||
for style_field, style_value in value.items():
|
||||
if hasattr(indicator.styling, style_field):
|
||||
setattr(indicator.styling, style_field, style_value)
|
||||
else:
|
||||
setattr(indicator, field, value)
|
||||
|
||||
return self.save_indicator(indicator)
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"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"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"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
|
||||
@@ -32,6 +32,7 @@ class IndicatorLayerConfig(LayerConfig):
|
||||
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__()
|
||||
@@ -341,9 +342,7 @@ class SMALayer(BaseIndicatorLayer):
|
||||
line=dict(
|
||||
color=self.config.color or '#2196F3',
|
||||
width=self.config.line_width
|
||||
),
|
||||
row=subplot_row,
|
||||
col=1
|
||||
)
|
||||
)
|
||||
|
||||
self.traces = [sma_trace]
|
||||
@@ -442,9 +441,7 @@ class EMALayer(BaseIndicatorLayer):
|
||||
line=dict(
|
||||
color=self.config.color or '#FF9800',
|
||||
width=self.config.line_width
|
||||
),
|
||||
row=subplot_row,
|
||||
col=1
|
||||
)
|
||||
)
|
||||
|
||||
self.traces = [ema_trace]
|
||||
@@ -550,8 +547,6 @@ class BollingerBandsLayer(BaseIndicatorLayer):
|
||||
mode='lines',
|
||||
name=f'BB Upper({period})',
|
||||
line=dict(color=self.config.color or '#9C27B0', width=1),
|
||||
row=subplot_row,
|
||||
col=1,
|
||||
showlegend=True
|
||||
)
|
||||
traces.append(upper_trace)
|
||||
@@ -565,8 +560,6 @@ class BollingerBandsLayer(BaseIndicatorLayer):
|
||||
line=dict(color=self.config.color or '#9C27B0', width=1),
|
||||
fill='tonexty',
|
||||
fillcolor='rgba(156, 39, 176, 0.1)',
|
||||
row=subplot_row,
|
||||
col=1,
|
||||
showlegend=True
|
||||
)
|
||||
traces.append(lower_trace)
|
||||
@@ -579,8 +572,6 @@ class BollingerBandsLayer(BaseIndicatorLayer):
|
||||
mode='lines',
|
||||
name=f'BB Middle({period})',
|
||||
line=dict(color=self.config.color or '#9C27B0', width=1, dash='dash'),
|
||||
row=subplot_row,
|
||||
col=1,
|
||||
showlegend=True
|
||||
)
|
||||
traces.append(middle_trace)
|
||||
|
||||
Reference in New Issue
Block a user