ec8f5514bb
- Updated all technical indicators to return pandas DataFrames instead of lists, improving consistency and usability. - Modified the `calculate` method in `TechnicalIndicators` to directly return DataFrames with relevant indicator values. - Enhanced the `data_integration.py` to utilize the new DataFrame outputs for better integration with charting. - Updated documentation to reflect the new DataFrame-centric approach, including usage examples and output structures. - Improved error handling to ensure empty DataFrames are returned when insufficient data is available. These changes streamline the indicator calculations and improve the overall architecture, aligning with project standards for maintainability and performance.
788 lines
26 KiB
Python
788 lines
26 KiB
Python
"""
|
|
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
|
|
import pandas as pd
|
|
|
|
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("default_logger")
|
|
|
|
|
|
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"Indicator Definitions: Error converting candle to OHLCV: {e}")
|
|
continue
|
|
|
|
logger.debug(f"Indicator Definitions: 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, pd.DataFrame]:
|
|
"""
|
|
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 as DataFrames
|
|
"""
|
|
if not candles:
|
|
logger.warning("Indicator Definitions: 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("Indicator Definitions: 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"Indicator Definitions: Unknown indicator configuration: {indicator_name}")
|
|
|
|
if not configs_to_calculate:
|
|
logger.warning("Indicator Definitions: No valid indicator configurations found")
|
|
return {}
|
|
|
|
# Calculate indicators
|
|
try:
|
|
results = indicators_calc.calculate_multiple_indicators(ohlcv_candles, configs_to_calculate)
|
|
# results is now a dict of DataFrames
|
|
logger.debug(f"Indicator Definitions: Calculated {len(results)} indicators successfully")
|
|
return results
|
|
|
|
except Exception as e:
|
|
logger.error(f"Indicator Definitions: 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}"] |