291 lines
10 KiB
Python
291 lines
10 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("chart_builder")
|
||
|
|
|
||
|
|
|
||
|
|
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)
|
||
|
|
|
||
|
|
# 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"Fetched {len(candles)} candles for {symbol} {timeframe}")
|
||
|
|
return candles
|
||
|
|
|
||
|
|
except DatabaseOperationError as e:
|
||
|
|
self.logger.error(f"Database error fetching market data: {e}")
|
||
|
|
return []
|
||
|
|
except Exception as e:
|
||
|
|
self.logger.error(f"Unexpected error fetching market data: {e}")
|
||
|
|
return []
|
||
|
|
|
||
|
|
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"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")
|
||
|
|
|
||
|
|
# 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"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"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"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"Creating strategy chart for {strategy_name} (basic implementation)")
|
||
|
|
return self.create_candlestick_chart(symbol, timeframe, **kwargs)
|