""" Modular Chart System for Crypto Trading Bot Dashboard This package provides a flexible, strategy-driven chart system that supports: - Technical indicator overlays (SMA, EMA, Bollinger Bands) - Subplot management (RSI, MACD) - Strategy-specific configurations - Future bot signal integration Main Components: - ChartBuilder: Main orchestrator for chart creation - Layer System: Modular rendering components - Configuration System: Strategy-driven chart configs """ import plotly.graph_objects as go from typing import List from .builder import ChartBuilder from .utils import ( validate_market_data, prepare_chart_data, get_indicator_colors ) from .config import ( get_available_indicators, calculate_indicators, get_overlay_indicators, get_subplot_indicators, get_indicator_display_config ) from .data_integration import ( MarketDataIntegrator, DataIntegrationConfig, get_market_data_integrator, fetch_indicator_data, check_symbol_data_quality ) from .error_handling import ( ChartErrorHandler, ChartError, ErrorSeverity, InsufficientDataError, DataValidationError, IndicatorCalculationError, DataConnectionError, check_data_sufficiency, get_error_message, create_error_annotation ) # Layer imports with error handling from .layers.base import ( LayerConfig, BaseLayer, CandlestickLayer, VolumeLayer, LayerManager ) from .layers.indicators import ( IndicatorLayerConfig, BaseIndicatorLayer, SMALayer, EMALayer, BollingerBandsLayer ) from .layers.subplots import ( SubplotLayerConfig, BaseSubplotLayer, RSILayer, MACDLayer ) # Version information __version__ = "0.1.0" __package_name__ = "charts" # Public API exports __all__ = [ # Core components "ChartBuilder", "validate_market_data", "prepare_chart_data", "get_indicator_colors", # Chart creation functions "create_candlestick_chart", "create_strategy_chart", "create_empty_chart", "create_error_chart", # Data integration "MarketDataIntegrator", "DataIntegrationConfig", "get_market_data_integrator", "fetch_indicator_data", "check_symbol_data_quality", # Error handling "ChartErrorHandler", "ChartError", "ErrorSeverity", "InsufficientDataError", "DataValidationError", "IndicatorCalculationError", "DataConnectionError", "check_data_sufficiency", "get_error_message", "create_error_annotation", # Utility functions "get_supported_symbols", "get_supported_timeframes", "get_market_statistics", "check_data_availability", "create_data_status_indicator", # Base layers "LayerConfig", "BaseLayer", "CandlestickLayer", "VolumeLayer", "LayerManager", # Indicator layers "IndicatorLayerConfig", "BaseIndicatorLayer", "SMALayer", "EMALayer", "BollingerBandsLayer", # Subplot layers "SubplotLayerConfig", "BaseSubplotLayer", "RSILayer", "MACDLayer", # Convenience functions "create_basic_chart", "create_indicator_chart", "create_chart_with_indicators" ] # Initialize logger from utils.logger import get_logger logger = get_logger("charts") def create_candlestick_chart(symbol: str, timeframe: str, days_back: int = 7, **kwargs) -> go.Figure: """ Create a candlestick chart with enhanced data integration. Args: symbol: Trading pair (e.g., 'BTC-USDT') timeframe: Timeframe (e.g., '1h', '1d') days_back: Number of days to look back **kwargs: Additional chart parameters Returns: Plotly figure with candlestick chart """ builder = ChartBuilder() # Check data quality first data_quality = builder.check_data_quality(symbol, timeframe) if not data_quality['available']: logger.warning(f"Data not available for {symbol} {timeframe}: {data_quality['message']}") return builder._create_error_chart(f"No data available: {data_quality['message']}") if not data_quality['sufficient_for_indicators']: logger.warning(f"Insufficient data for indicators: {symbol} {timeframe}") # Use enhanced data fetching try: candles = builder.fetch_market_data_enhanced(symbol, timeframe, days_back) if not candles: return builder._create_error_chart(f"No market data found for {symbol} {timeframe}") # Prepare data for charting df = prepare_chart_data(candles) if df.empty: return builder._create_error_chart("Failed to prepare chart data") # Create chart with data quality info fig = builder._create_candlestick_with_volume(df, symbol, timeframe) # Add data quality annotation if data is stale if not data_quality['is_recent']: age_hours = data_quality['data_age_minutes'] / 60 fig.add_annotation( text=f"⚠️ Data is {age_hours:.1f}h old", xref="paper", yref="paper", x=0.02, y=0.98, showarrow=False, bgcolor="rgba(255,193,7,0.8)", bordercolor="orange", borderwidth=1 ) logger.debug(f"Created enhanced candlestick chart for {symbol} {timeframe} with {len(candles)} candles") return fig except Exception as e: logger.error(f"Error creating enhanced candlestick chart: {e}") return builder._create_error_chart(f"Chart creation failed: {str(e)}") def create_strategy_chart(symbol: str, timeframe: str, strategy_name: str, **kwargs): """ Convenience function to create a strategy-specific chart. Args: symbol: Trading pair timeframe: Timeframe strategy_name: Name of the strategy configuration **kwargs: Additional parameters Returns: Plotly Figure object with strategy indicators """ builder = ChartBuilder() return builder.create_strategy_chart(symbol, timeframe, strategy_name, **kwargs) def get_supported_symbols(): """Get list of symbols that have data in the database.""" builder = ChartBuilder() candles = builder.fetch_market_data("BTC-USDT", "1m", days_back=1) # Test query if candles: from database.operations import get_database_operations from utils.logger import get_logger logger = get_logger("default_logger") try: db = get_database_operations(logger) with db.market_data.get_session() as session: from sqlalchemy import text result = session.execute(text("SELECT DISTINCT symbol FROM market_data ORDER BY symbol")) return [row[0] for row in result] except Exception: pass return ['BTC-USDT', 'ETH-USDT'] # Fallback def get_supported_timeframes(): """Get list of timeframes that have data in the database.""" builder = ChartBuilder() candles = builder.fetch_market_data("BTC-USDT", "1m", days_back=1) # Test query if candles: from database.operations import get_database_operations from utils.logger import get_logger logger = get_logger("default_logger") try: db = get_database_operations(logger) with db.market_data.get_session() as session: from sqlalchemy import text result = session.execute(text("SELECT DISTINCT timeframe FROM market_data ORDER BY timeframe")) return [row[0] for row in result] except Exception: pass return ['5s', '1m', '15m', '1h'] # Fallback def get_market_statistics(symbol: str, timeframe: str = "1h", days_back: int = 1): """Calculate market statistics from recent data over a specified period.""" builder = ChartBuilder() candles = builder.fetch_market_data(symbol, timeframe, days_back=days_back) if not candles: return {'Price': 'N/A', f'Change ({days_back}d)': 'N/A', f'Volume ({days_back}d)': 'N/A', f'High ({days_back}d)': 'N/A', f'Low ({days_back}d)': 'N/A'} import pandas as pd df = pd.DataFrame(candles) latest = df.iloc[-1] current_price = float(latest['close']) # Calculate change over the period if len(df) > 1: price_period_ago = float(df.iloc[0]['open']) change_percent = ((current_price - price_period_ago) / price_period_ago) * 100 else: change_percent = 0 from .utils import format_price, format_volume # Determine label for period (e.g., "24h", "7d", "1h") if days_back == 1/24: period_label = "1h" elif days_back == 4/24: period_label = "4h" elif days_back == 6/24: period_label = "6h" elif days_back == 12/24: period_label = "12h" elif days_back < 1: # For other fractional days, show as hours period_label = f"{int(days_back * 24)}h" elif days_back == 1: period_label = "24h" # Keep 24h for 1 day for clarity else: period_label = f"{days_back}d" return { 'Price': format_price(current_price, decimals=2), f'Change ({period_label})': f"{'+' if change_percent >= 0 else ''}{change_percent:.2f}%", f'Volume ({period_label})': format_volume(df['volume'].sum()), f'High ({period_label})': format_price(df['high'].max(), decimals=2), f'Low ({period_label})': format_price(df['low'].min(), decimals=2) } def check_data_availability(symbol: str, timeframe: str): """Check data availability for a symbol and timeframe.""" from datetime import datetime, timezone, timedelta from database.operations import get_database_operations from utils.logger import get_logger try: logger = get_logger("charts_data_check") db = get_database_operations(logger) latest_candle = db.market_data.get_latest_candle(symbol, timeframe) if latest_candle: latest_time = latest_candle['timestamp'] time_diff = datetime.now(timezone.utc) - latest_time.replace(tzinfo=timezone.utc) return { 'has_data': True, 'latest_timestamp': latest_time, 'time_since_last': time_diff, 'is_recent': time_diff < timedelta(hours=1), 'message': f"Latest data: {latest_time.strftime('%Y-%m-%d %H:%M:%S UTC')}" } else: return { 'has_data': False, 'latest_timestamp': None, 'time_since_last': None, 'is_recent': False, 'message': f"No data available for {symbol} {timeframe}" } except Exception as e: return { 'has_data': False, 'latest_timestamp': None, 'time_since_last': None, 'is_recent': False, 'message': f"Error checking data: {str(e)}" } def create_data_status_indicator(symbol: str, timeframe: str): """Create a data status indicator for the dashboard.""" status = check_data_availability(symbol, timeframe) if status['has_data']: if status['is_recent']: icon, color, status_text = "🟢", "#27ae60", "Real-time Data" else: icon, color, status_text = "🟡", "#f39c12", "Delayed Data" else: icon, color, status_text = "🔴", "#e74c3c", "No Data" return f'{icon} {status_text}
{status["message"]}' def create_error_chart(error_message: str): """Create an error chart with error message.""" builder = ChartBuilder() return builder._create_error_chart(error_message) def create_basic_chart(symbol: str, data: list, indicators: list = None, error_handling: bool = True) -> 'go.Figure': """ Create a basic chart with error handling. Args: symbol: Trading symbol data: OHLCV data as list of dictionaries indicators: List of indicator configurations error_handling: Whether to use comprehensive error handling Returns: Plotly figure with chart or error display """ try: from plotly import graph_objects as go # Initialize chart builder builder = ChartBuilder() if error_handling: # Use error-aware chart creation error_handler = ChartErrorHandler() is_valid = error_handler.validate_data_sufficiency(data, indicators=indicators or []) if not is_valid: # Create error chart fig = go.Figure() error_msg = error_handler.get_user_friendly_message() fig.add_annotation(create_error_annotation(error_msg, position='center')) fig.update_layout( title=f"Chart Error - {symbol}", xaxis={'visible': False}, yaxis={'visible': False}, template='plotly_white', height=400 ) return fig # Create chart normally return builder.create_candlestick_chart(data, symbol=symbol, indicators=indicators or []) except Exception as e: # Fallback error chart from plotly import graph_objects as go fig = go.Figure() fig.add_annotation(create_error_annotation( f"Chart creation failed: {str(e)}", position='center' )) fig.update_layout( title=f"Chart Error - {symbol}", template='plotly_white', height=400 ) return fig def create_indicator_chart(symbol: str, data: list, indicator_type: str, **params) -> 'go.Figure': """ Create a chart focused on a specific indicator. Args: symbol: Trading symbol data: OHLCV data indicator_type: Type of indicator ('sma', 'ema', 'bollinger_bands', 'rsi', 'macd') **params: Indicator parameters Returns: Plotly figure with indicator chart """ try: # Map indicator types to configurations indicator_map = { 'sma': {'type': 'sma', 'parameters': {'period': params.get('period', 20)}}, 'ema': {'type': 'ema', 'parameters': {'period': params.get('period', 20)}}, 'bollinger_bands': { 'type': 'bollinger_bands', 'parameters': { 'period': params.get('period', 20), 'std_dev': params.get('std_dev', 2) } }, 'rsi': {'type': 'rsi', 'parameters': {'period': params.get('period', 14)}}, 'macd': { 'type': 'macd', 'parameters': { 'fast_period': params.get('fast_period', 12), 'slow_period': params.get('slow_period', 26), 'signal_period': params.get('signal_period', 9) } } } if indicator_type not in indicator_map: raise ValueError(f"Unknown indicator type: {indicator_type}") indicator_config = indicator_map[indicator_type] return create_basic_chart(symbol, data, indicators=[indicator_config]) except Exception as e: 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 ) def initialize_indicator_manager(): # Implementation of initialize_indicator_manager function pass