601 lines
25 KiB
Python
601 lines
25 KiB
Python
"""
|
|
ChartBuilder - Main orchestrator for chart creation
|
|
|
|
This module contains the ChartBuilder class which serves as the main entry point
|
|
for creating charts with various configurations, indicators, and layers.
|
|
"""
|
|
|
|
import plotly.graph_objects as go
|
|
from plotly.subplots import make_subplots
|
|
import pandas as pd
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import List, Dict, Any, Optional, Union
|
|
from decimal import Decimal
|
|
|
|
from database.operations import get_database_operations, DatabaseOperationError
|
|
from utils.logger import get_logger
|
|
from .utils import validate_market_data, prepare_chart_data, get_indicator_colors
|
|
|
|
# Initialize logger
|
|
logger = get_logger("default_logger")
|
|
|
|
|
|
class ChartBuilder:
|
|
"""
|
|
Main chart builder class for creating modular, configurable charts.
|
|
|
|
This class orchestrates the creation of charts by coordinating between
|
|
data fetching, layer rendering, and configuration management.
|
|
"""
|
|
|
|
def __init__(self, logger_instance: Optional = None):
|
|
"""
|
|
Initialize the ChartBuilder.
|
|
|
|
Args:
|
|
logger_instance: Optional logger instance
|
|
"""
|
|
self.logger = logger_instance or logger
|
|
self.db_ops = get_database_operations(self.logger)
|
|
|
|
# Initialize market data integrator
|
|
from .data_integration import get_market_data_integrator
|
|
self.data_integrator = get_market_data_integrator()
|
|
|
|
# Chart styling defaults
|
|
self.default_colors = get_indicator_colors()
|
|
self.default_height = 600
|
|
self.default_template = "plotly_white"
|
|
|
|
def fetch_market_data(self, symbol: str, timeframe: str,
|
|
days_back: int = 7, exchange: str = "okx") -> List[Dict[str, Any]]:
|
|
"""
|
|
Fetch market data from the database.
|
|
|
|
Args:
|
|
symbol: Trading pair (e.g., 'BTC-USDT')
|
|
timeframe: Timeframe (e.g., '1h', '1d')
|
|
days_back: Number of days to look back
|
|
exchange: Exchange name
|
|
|
|
Returns:
|
|
List of candle data dictionaries
|
|
"""
|
|
try:
|
|
# Calculate time range
|
|
end_time = datetime.now(timezone.utc)
|
|
start_time = end_time - timedelta(days=days_back)
|
|
|
|
# Fetch candles using the database operations API
|
|
candles = self.db_ops.market_data.get_candles(
|
|
symbol=symbol,
|
|
timeframe=timeframe,
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
exchange=exchange
|
|
)
|
|
|
|
self.logger.debug(f"Chart builder: Fetched {len(candles)} candles for {symbol} {timeframe}")
|
|
return candles
|
|
|
|
except DatabaseOperationError as e:
|
|
self.logger.error(f"Chart builder: Database error fetching market data: {e}")
|
|
return []
|
|
except Exception as e:
|
|
self.logger.error(f"Chart builder: Unexpected error fetching market data: {e}")
|
|
return []
|
|
|
|
def fetch_market_data_enhanced(self, symbol: str, timeframe: str,
|
|
days_back: int = 7, exchange: str = "okx") -> List[Dict[str, Any]]:
|
|
"""
|
|
Enhanced market data fetching with validation and caching.
|
|
|
|
Args:
|
|
symbol: Trading pair (e.g., 'BTC-USDT')
|
|
timeframe: Timeframe (e.g., '1h', '1d')
|
|
days_back: Number of days to look back
|
|
exchange: Exchange name
|
|
|
|
Returns:
|
|
List of validated candle data dictionaries
|
|
"""
|
|
try:
|
|
# Use the data integrator for enhanced data handling
|
|
raw_candles, ohlcv_candles = self.data_integrator.get_market_data_for_indicators(
|
|
symbol, timeframe, days_back, exchange
|
|
)
|
|
|
|
if not raw_candles:
|
|
self.logger.warning(f"Chart builder: No market data available for {symbol} {timeframe}")
|
|
return []
|
|
|
|
self.logger.debug(f"Chart builder: Enhanced fetch: {len(raw_candles)} candles for {symbol} {timeframe}")
|
|
return raw_candles
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Chart builder: Error in enhanced market data fetch: {e}")
|
|
# Fallback to original method
|
|
return self.fetch_market_data(symbol, timeframe, days_back, exchange)
|
|
|
|
def create_candlestick_chart(self, symbol: str, timeframe: str,
|
|
days_back: int = 7, **kwargs) -> go.Figure:
|
|
"""
|
|
Create a basic candlestick chart.
|
|
|
|
Args:
|
|
symbol: Trading pair
|
|
timeframe: Timeframe
|
|
days_back: Number of days to look back
|
|
**kwargs: Additional chart parameters
|
|
|
|
Returns:
|
|
Plotly Figure object with candlestick chart
|
|
"""
|
|
try:
|
|
# Fetch market data
|
|
candles = self.fetch_market_data(symbol, timeframe, days_back)
|
|
|
|
# Handle empty data
|
|
if not candles:
|
|
self.logger.warning(f"Chart builder: 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"Chart builder: Invalid market data for {symbol} {timeframe}")
|
|
return self._create_error_chart("Invalid market data format")
|
|
|
|
# Prepare chart data
|
|
df = prepare_chart_data(candles)
|
|
|
|
# Determine if we need volume subplot
|
|
has_volume = 'volume' in df.columns and df['volume'].sum() > 0
|
|
include_volume = kwargs.get('include_volume', has_volume)
|
|
|
|
if include_volume and has_volume:
|
|
return self._create_candlestick_with_volume(df, symbol, timeframe, **kwargs)
|
|
else:
|
|
return self._create_basic_candlestick(df, symbol, timeframe, **kwargs)
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Chart builder: Error creating candlestick chart for {symbol} {timeframe}: {e}")
|
|
return self._create_error_chart(f"Error loading chart: {str(e)}")
|
|
|
|
def _create_basic_candlestick(self, df: pd.DataFrame, symbol: str,
|
|
timeframe: str, **kwargs) -> go.Figure:
|
|
"""Create a basic candlestick chart without volume."""
|
|
|
|
# Get custom parameters
|
|
height = kwargs.get('height', self.default_height)
|
|
template = kwargs.get('template', self.default_template)
|
|
|
|
# Create candlestick chart
|
|
fig = go.Figure(data=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']
|
|
))
|
|
|
|
# Update layout
|
|
fig.update_layout(
|
|
title=f"{symbol} - {timeframe} Chart",
|
|
xaxis_title="Time",
|
|
yaxis_title="Price (USDT)",
|
|
template=template,
|
|
showlegend=False,
|
|
height=height,
|
|
xaxis_rangeslider_visible=False,
|
|
hovermode='x unified'
|
|
)
|
|
|
|
self.logger.debug(f"Chart builder: Created basic candlestick chart for {symbol} {timeframe} with {len(df)} candles")
|
|
return fig
|
|
|
|
def _create_candlestick_with_volume(self, df: pd.DataFrame, symbol: str,
|
|
timeframe: str, **kwargs) -> go.Figure:
|
|
"""Create a candlestick chart with volume subplot."""
|
|
|
|
# Get custom parameters
|
|
height = kwargs.get('height', 700) # Taller for volume subplot
|
|
template = kwargs.get('template', self.default_template)
|
|
|
|
# Create subplots
|
|
fig = make_subplots(
|
|
rows=2, cols=1,
|
|
shared_xaxes=True,
|
|
vertical_spacing=0.03,
|
|
subplot_titles=(f'{symbol} Price', 'Volume'),
|
|
row_heights=[0.7, 0.3] # 70% for price, 30% for volume
|
|
)
|
|
|
|
# Add candlestick chart
|
|
fig.add_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']
|
|
),
|
|
row=1, col=1
|
|
)
|
|
|
|
# Add volume bars with color coding
|
|
colors = [self.default_colors['bullish'] if close >= open else self.default_colors['bearish']
|
|
for close, open in zip(df['close'], df['open'])]
|
|
|
|
fig.add_trace(
|
|
go.Bar(
|
|
x=df['timestamp'],
|
|
y=df['volume'],
|
|
name='Volume',
|
|
marker_color=colors,
|
|
opacity=0.7
|
|
),
|
|
row=2, col=1
|
|
)
|
|
|
|
# Update layout
|
|
fig.update_layout(
|
|
title=f"{symbol} - {timeframe} Chart with Volume",
|
|
template=template,
|
|
showlegend=False,
|
|
height=height,
|
|
xaxis_rangeslider_visible=False,
|
|
hovermode='x unified'
|
|
)
|
|
|
|
# Update axes
|
|
fig.update_yaxes(title_text="Price (USDT)", row=1, col=1)
|
|
fig.update_yaxes(title_text="Volume", row=2, col=1)
|
|
fig.update_xaxes(title_text="Time", row=2, col=1)
|
|
|
|
self.logger.debug(f"Chart builder: Created candlestick chart with volume for {symbol} {timeframe}")
|
|
return fig
|
|
|
|
def _create_empty_chart(self, message: str = "No data available") -> go.Figure:
|
|
"""Create an empty chart with a message."""
|
|
fig = go.Figure()
|
|
|
|
fig.add_annotation(
|
|
text=message,
|
|
xref="paper", yref="paper",
|
|
x=0.5, y=0.5,
|
|
xanchor='center', yanchor='middle',
|
|
showarrow=False,
|
|
font=dict(size=16, color="#7f8c8d")
|
|
)
|
|
|
|
fig.update_layout(
|
|
template=self.default_template,
|
|
height=self.default_height,
|
|
showlegend=False,
|
|
xaxis=dict(visible=False),
|
|
yaxis=dict(visible=False)
|
|
)
|
|
|
|
return fig
|
|
|
|
def _create_error_chart(self, error_message: str) -> go.Figure:
|
|
"""Create an error chart with error message."""
|
|
fig = go.Figure()
|
|
|
|
fig.add_annotation(
|
|
text=f"⚠️ {error_message}",
|
|
xref="paper", yref="paper",
|
|
x=0.5, y=0.5,
|
|
xanchor='center', yanchor='middle',
|
|
showarrow=False,
|
|
font=dict(size=16, color="#e74c3c")
|
|
)
|
|
|
|
fig.update_layout(
|
|
template=self.default_template,
|
|
height=self.default_height,
|
|
showlegend=False,
|
|
xaxis=dict(visible=False),
|
|
yaxis=dict(visible=False)
|
|
)
|
|
|
|
return fig
|
|
|
|
def create_strategy_chart(self, symbol: str, timeframe: str,
|
|
strategy_name: str, **kwargs) -> go.Figure:
|
|
"""
|
|
Create a strategy-specific chart (placeholder for future implementation).
|
|
|
|
Args:
|
|
symbol: Trading pair
|
|
timeframe: Timeframe
|
|
strategy_name: Name of the strategy configuration
|
|
**kwargs: Additional parameters
|
|
|
|
Returns:
|
|
Plotly Figure object
|
|
"""
|
|
# For now, return a basic candlestick chart
|
|
# This will be enhanced in later tasks with strategy configurations
|
|
self.logger.info(f"Chart builder: Creating strategy chart for {strategy_name} (basic implementation)")
|
|
return self.create_candlestick_chart(symbol, timeframe, **kwargs)
|
|
|
|
def check_data_quality(self, symbol: str, timeframe: str,
|
|
exchange: str = "okx") -> Dict[str, Any]:
|
|
"""
|
|
Check data quality and availability for chart creation.
|
|
|
|
Args:
|
|
symbol: Trading pair
|
|
timeframe: Timeframe
|
|
exchange: Exchange name
|
|
|
|
Returns:
|
|
Dictionary with data quality information
|
|
"""
|
|
try:
|
|
return self.data_integrator.check_data_availability(symbol, timeframe, exchange)
|
|
except Exception as e:
|
|
self.logger.error(f"Chart builder: Error checking data quality: {e}")
|
|
return {
|
|
'available': False,
|
|
'latest_timestamp': None,
|
|
'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"Chart builder: 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"Chart builder: 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"Chart builder: 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"Chart builder: 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"Chart builder: Error creating chart with indicators: {e}")
|
|
return self._create_error_chart(f"Chart creation failed: {str(e)}") |