""" Dashboard Component Manager - Clean Trading Dashboard Manages the formatting and creation of dashboard components """ from dash import html, dcc from datetime import datetime import logging logger = logging.getLogger(__name__) class DashboardComponentManager: """Manages dashboard component formatting and creation""" def __init__(self): pass def format_trading_signals(self, recent_decisions): """Format trading signals for display""" try: if not recent_decisions: return [html.P("No recent signals", className="text-muted small")] signals = [] for decision in reversed(recent_decisions[-10:]): # Last 10 signals, reversed # Handle both TradingDecision objects and dictionary formats if hasattr(decision, 'timestamp'): # This is a TradingDecision object (dataclass) timestamp = getattr(decision, 'timestamp', 'Unknown') action = getattr(decision, 'action', 'UNKNOWN') confidence = getattr(decision, 'confidence', 0) price = getattr(decision, 'price', 0) executed = getattr(decision, 'executed', False) blocked = getattr(decision, 'blocked', False) manual = getattr(decision, 'manual', False) else: # This is a dictionary format timestamp = decision.get('timestamp', 'Unknown') action = decision.get('action', 'UNKNOWN') confidence = decision.get('confidence', 0) price = decision.get('price', 0) executed = decision.get('executed', False) blocked = decision.get('blocked', False) manual = decision.get('manual', False) # Determine signal style if executed: badge_class = "bg-success" status = "✓" elif blocked: badge_class = "bg-danger" status = "✗" else: badge_class = "bg-warning" status = "○" action_color = "text-success" if action == "BUY" else "text-danger" manual_indicator = " [M]" if manual else "" signal_div = html.Div([ html.Span(f"{timestamp}", className="small text-muted me-2"), html.Span(f"{status}", className=f"badge {badge_class} me-2"), html.Span(f"{action}{manual_indicator}", className=f"{action_color} fw-bold me-2"), html.Span(f"({confidence:.1f}%)", className="small text-muted me-2"), html.Span(f"${price:.2f}", className="small") ], className="mb-1") signals.append(signal_div) return signals except Exception as e: logger.error(f"Error formatting trading signals: {e}") return [html.P(f"Error: {str(e)}", className="text-danger small")] def format_closed_trades_table(self, closed_trades): """Format closed trades table for display""" try: if not closed_trades: return html.P("No closed trades", className="text-muted small") # Create table headers headers = html.Thead([ html.Tr([ html.Th("Time", className="small"), html.Th("Side", className="small"), html.Th("Size", className="small"), html.Th("Entry", className="small"), html.Th("Exit", className="small"), html.Th("P&L", className="small"), html.Th("Fees", className="small") ]) ]) # Create table rows rows = [] for trade in closed_trades[-20:]: # Last 20 trades # Handle both trade objects and dictionary formats if hasattr(trade, 'entry_time'): # This is a trade object entry_time = getattr(trade, 'entry_time', 'Unknown') side = getattr(trade, 'side', 'UNKNOWN') size = getattr(trade, 'size', 0) entry_price = getattr(trade, 'entry_price', 0) exit_price = getattr(trade, 'exit_price', 0) pnl = getattr(trade, 'pnl', 0) fees = getattr(trade, 'fees', 0) else: # This is a dictionary format entry_time = trade.get('entry_time', 'Unknown') side = trade.get('side', 'UNKNOWN') size = trade.get('size', 0) entry_price = trade.get('entry_price', 0) exit_price = trade.get('exit_price', 0) pnl = trade.get('pnl', 0) fees = trade.get('fees', 0) # Format time if isinstance(entry_time, datetime): time_str = entry_time.strftime('%H:%M:%S') else: time_str = str(entry_time) # Determine P&L color pnl_class = "text-success" if pnl >= 0 else "text-danger" side_class = "text-success" if side == "BUY" else "text-danger" row = html.Tr([ html.Td(time_str, className="small"), html.Td(side, className=f"small {side_class}"), html.Td(f"{size:.3f}", className="small"), html.Td(f"${entry_price:.2f}", className="small"), html.Td(f"${exit_price:.2f}", className="small"), html.Td(f"${pnl:.2f}", className=f"small {pnl_class}"), html.Td(f"${fees:.3f}", className="small text-muted") ]) rows.append(row) tbody = html.Tbody(rows) return html.Table([headers, tbody], className="table table-sm table-striped") except Exception as e: logger.error(f"Error formatting closed trades: {e}") return html.P(f"Error: {str(e)}", className="text-danger small") def format_system_status(self, status_data): """Format system status for display""" try: if not status_data or 'error' in status_data: return [html.P("Status unavailable", className="text-muted small")] status_items = [] # Trading status trading_enabled = status_data.get('trading_enabled', False) simulation_mode = status_data.get('simulation_mode', True) if trading_enabled: if simulation_mode: status_items.append(html.Div([ html.I(className="fas fa-play-circle text-success me-2"), html.Span("Trading: SIMULATION", className="text-warning") ], className="mb-1")) else: status_items.append(html.Div([ html.I(className="fas fa-play-circle text-success me-2"), html.Span("Trading: LIVE", className="text-success fw-bold") ], className="mb-1")) else: status_items.append(html.Div([ html.I(className="fas fa-pause-circle text-danger me-2"), html.Span("Trading: DISABLED", className="text-danger") ], className="mb-1")) # Data provider status data_status = status_data.get('data_provider_status', 'Unknown') status_items.append(html.Div([ html.I(className="fas fa-database text-info me-2"), html.Span(f"Data: {data_status}", className="small") ], className="mb-1")) # WebSocket status ws_status = status_data.get('websocket_status', 'Unknown') ws_class = "text-success" if ws_status == "Connected" else "text-danger" status_items.append(html.Div([ html.I(className="fas fa-wifi text-info me-2"), html.Span(f"WebSocket: {ws_status}", className=f"small {ws_class}") ], className="mb-1")) # COB status cob_status = status_data.get('cob_status', 'Unknown') cob_class = "text-success" if cob_status == "Active" else "text-warning" status_items.append(html.Div([ html.I(className="fas fa-layer-group text-info me-2"), html.Span(f"COB: {cob_status}", className=f"small {cob_class}") ], className="mb-1")) return status_items except Exception as e: logger.error(f"Error formatting system status: {e}") return [html.P(f"Error: {str(e)}", className="text-danger small")] def _format_cnn_pivot_prediction(self, model_info): """Format CNN pivot prediction for display""" try: pivot_prediction = model_info.get('pivot_prediction') if not pivot_prediction: return html.Div() pivot_type = pivot_prediction.get('pivot_type', 'UNKNOWN') predicted_price = pivot_prediction.get('predicted_price', 0) confidence = pivot_prediction.get('confidence', 0) time_horizon = pivot_prediction.get('time_horizon_minutes', 0) # Color coding for pivot types if 'RESISTANCE' in pivot_type: pivot_color = "text-danger" pivot_icon = "fas fa-arrow-up" elif 'SUPPORT' in pivot_type: pivot_color = "text-success" pivot_icon = "fas fa-arrow-down" else: pivot_color = "text-warning" pivot_icon = "fas fa-arrows-alt-h" return html.Div([ html.Div([ html.I(className=f"{pivot_icon} me-1 {pivot_color}"), html.Span("Next Pivot: ", className="text-muted small"), html.Span(f"${predicted_price:.2f}", className=f"small fw-bold {pivot_color}") ], className="mb-1"), html.Div([ html.Span(f"{pivot_type.replace('_', ' ')}", className=f"small {pivot_color}"), html.Span(f" ({confidence:.0%}) in ~{time_horizon}m", className="text-muted small") ]) ], className="mt-1 p-1", style={"backgroundColor": "rgba(255,255,255,0.02)", "borderRadius": "3px"}) except Exception as e: logger.debug(f"Error formatting CNN pivot prediction: {e}") return html.Div() def format_cob_data(self, cob_snapshot, symbol): """Format COB data for display""" try: if not cob_snapshot: return [html.P("No COB data", className="text-muted small")] # Real COB data display cob_info = [] # Symbol header cob_info.append(html.Div([ html.Strong(f"{symbol}", className="text-info"), html.Span(" - COB Snapshot", className="small text-muted") ], className="mb-2")) # Check if we have a real COB snapshot object if hasattr(cob_snapshot, 'volume_weighted_mid'): # Real COB snapshot data mid_price = getattr(cob_snapshot, 'volume_weighted_mid', 0) spread_bps = getattr(cob_snapshot, 'spread_bps', 0) bid_liquidity = getattr(cob_snapshot, 'total_bid_liquidity', 0) ask_liquidity = getattr(cob_snapshot, 'total_ask_liquidity', 0) imbalance = getattr(cob_snapshot, 'liquidity_imbalance', 0) bid_levels = len(getattr(cob_snapshot, 'consolidated_bids', [])) ask_levels = len(getattr(cob_snapshot, 'consolidated_asks', [])) # Price and spread cob_info.append(html.Div([ html.Div([ html.I(className="fas fa-dollar-sign text-success me-2"), html.Span(f"Mid: ${mid_price:.2f}", className="small fw-bold") ], className="mb-1"), html.Div([ html.I(className="fas fa-arrows-alt-h text-warning me-2"), html.Span(f"Spread: {spread_bps:.1f} bps", className="small") ], className="mb-1") ])) # Liquidity info total_liquidity = bid_liquidity + ask_liquidity bid_pct = (bid_liquidity / total_liquidity * 100) if total_liquidity > 0 else 0 ask_pct = (ask_liquidity / total_liquidity * 100) if total_liquidity > 0 else 0 cob_info.append(html.Div([ html.Div([ html.I(className="fas fa-layer-group text-info me-2"), html.Span(f"Liquidity: ${total_liquidity:,.0f}", className="small") ], className="mb-1"), html.Div([ html.Span(f"Bids: {bid_pct:.0f}% ", className="small text-success"), html.Span(f"Asks: {ask_pct:.0f}%", className="small text-danger") ], className="mb-1") ])) # Order book depth cob_info.append(html.Div([ html.Div([ html.I(className="fas fa-list text-secondary me-2"), html.Span(f"Levels: {bid_levels} bids, {ask_levels} asks", className="small") ], className="mb-1") ])) # Imbalance indicator imbalance_color = "text-success" if imbalance > 0.1 else "text-danger" if imbalance < -0.1 else "text-muted" imbalance_text = "Bid Heavy" if imbalance > 0.1 else "Ask Heavy" if imbalance < -0.1 else "Balanced" cob_info.append(html.Div([ html.I(className="fas fa-balance-scale me-2"), html.Span(f"Imbalance: ", className="small text-muted"), html.Span(f"{imbalance_text} ({imbalance:.3f})", className=f"small {imbalance_color}") ], className="mb-1")) else: # Fallback display for other data formats cob_info.append(html.Div([ html.Div([ html.I(className="fas fa-chart-bar text-success me-2"), html.Span("Order Book: Active", className="small") ], className="mb-1"), html.Div([ html.I(className="fas fa-coins text-warning me-2"), html.Span("Liquidity: Good", className="small") ], className="mb-1"), html.Div([ html.I(className="fas fa-balance-scale text-info me-2"), html.Span("Imbalance: Neutral", className="small") ]) ])) return cob_info except Exception as e: logger.error(f"Error formatting COB data: {e}") return [html.P(f"Error: {str(e)}", className="text-danger small")] def format_cob_data_with_buckets(self, cob_snapshot, symbol, price_buckets, memory_stats, bucket_size=1.0): """Format COB data with price buckets for high-frequency display""" try: components = [] # Symbol header with memory stats buffer_count = memory_stats.get('buffer_updates', 0) memory_count = memory_stats.get('memory_snapshots', 0) total_updates = memory_stats.get('total_updates', 0) components.append(html.Div([ html.Strong(f"{symbol}", className="text-info"), html.Span(f" - High-Freq COB", className="small text-muted"), html.Br(), html.Span(f"Buffer: {buffer_count} | Memory: {memory_count} | Total: {total_updates}", className="small text-success") ], className="mb-2")) # COB snapshot data (if available) if cob_snapshot: if hasattr(cob_snapshot, 'volume_weighted_mid'): # Real COB snapshot mid_price = getattr(cob_snapshot, 'volume_weighted_mid', 0) spread_bps = getattr(cob_snapshot, 'spread_bps', 0) imbalance = getattr(cob_snapshot, 'liquidity_imbalance', 0) components.append(html.Div([ html.Div([ html.I(className="fas fa-dollar-sign text-success me-2"), html.Span(f"Mid: ${mid_price:.2f}", className="small fw-bold") ], className="mb-1"), html.Div([ html.I(className="fas fa-arrows-alt-h text-warning me-2"), html.Span(f"Spread: {spread_bps:.1f} bps", className="small") ], className="mb-1") ])) # Imbalance imbalance_color = "text-success" if imbalance > 0.1 else "text-danger" if imbalance < -0.1 else "text-muted" imbalance_text = "Bid Heavy" if imbalance > 0.1 else "Ask Heavy" if imbalance < -0.1 else "Balanced" components.append(html.Div([ html.I(className="fas fa-balance-scale me-2"), html.Span(f"{imbalance_text} ({imbalance:.3f})", className=f"small {imbalance_color}") ], className="mb-2")) else: # Fallback for other data formats components.append(html.Div([ html.I(className="fas fa-chart-bar text-info me-2"), html.Span("COB: Active", className="small") ], className="mb-2")) # Price Buckets Section components.append(html.H6([ html.I(className="fas fa-layer-group me-2 text-primary"), f"${bucket_size:.0f} Price Buckets (±5 levels)" ], className="mb-2")) if price_buckets: # Sort buckets by price sorted_buckets = sorted(price_buckets, key=lambda x: x['price']) bucket_rows = [] for bucket in sorted_buckets: price = bucket['price'] total_vol = bucket['total_volume'] bid_pct = bucket['bid_pct'] ask_pct = bucket['ask_pct'] # Format volume if total_vol > 1000000: vol_str = f"${total_vol/1000000:.1f}M" elif total_vol > 1000: vol_str = f"${total_vol/1000:.0f}K" else: vol_str = f"${total_vol:.0f}" # Color based on bid/ask dominance if bid_pct > 60: row_class = "border-success" dominance = "BID" dominance_class = "text-success" elif ask_pct > 60: row_class = "border-danger" dominance = "ASK" dominance_class = "text-danger" else: row_class = "border-secondary" dominance = "BAL" dominance_class = "text-muted" bucket_row = html.Div([ html.Div([ html.Span(f"${price:.0f}", className="fw-bold me-2"), html.Span(vol_str, className="text-info me-2"), html.Span(f"{dominance}", className=f"small {dominance_class}") ], className="d-flex justify-content-between"), html.Div([ # Bid bar html.Div( style={ "width": f"{bid_pct}%", "height": "4px", "backgroundColor": "#28a745", "display": "inline-block" } ), # Ask bar html.Div( style={ "width": f"{ask_pct}%", "height": "4px", "backgroundColor": "#dc3545", "display": "inline-block" } ) ], className="mt-1") ], className=f"border {row_class} rounded p-2 mb-1 small") bucket_rows.append(bucket_row) components.extend(bucket_rows) else: components.append(html.P("No price bucket data", className="text-muted small")) # High-frequency update rate info components.append(html.Div([ html.Hr(), html.Div([ html.I(className="fas fa-tachometer-alt text-info me-2"), html.Span("High-Freq: 50-100 Hz | UI: 10 Hz", className="small text-muted") ]) ])) return components except Exception as e: logger.error(f"Error formatting COB data with buckets: {e}") return [html.P(f"Error: {str(e)}", className="text-danger small")] def format_training_metrics(self, metrics_data): """Format training metrics for display - Enhanced with loaded models""" try: if not metrics_data or 'error' in metrics_data: return [html.P("No training data", className="text-muted small")] content = [] # Loaded Models Section if 'loaded_models' in metrics_data: loaded_models = metrics_data['loaded_models'] content.append(html.H6([ html.I(className="fas fa-microchip me-2 text-primary"), "Loaded Models" ], className="mb-2")) if loaded_models: for model_name, model_info in loaded_models.items(): # Model status badge is_active = model_info.get('active', True) status_class = "text-success" if is_active else "text-muted" status_icon = "fas fa-check-circle" if is_active else "fas fa-pause-circle" # Last prediction info last_prediction = model_info.get('last_prediction', {}) pred_time = last_prediction.get('timestamp', 'N/A') pred_action = last_prediction.get('action', 'NONE') pred_confidence = last_prediction.get('confidence', 0) # 5MA Loss - with safe comparison handling loss_5ma = model_info.get('loss_5ma', 0.0) if loss_5ma is None: loss_5ma = 0.0 loss_class = "text-muted" else: loss_class = "text-success" if loss_5ma < 0.1 else "text-warning" if loss_5ma < 0.5 else "text-danger" # Model size/parameters model_size = model_info.get('parameters', 0) if model_size > 1e9: size_str = f"{model_size/1e9:.1f}B" elif model_size > 1e6: size_str = f"{model_size/1e6:.1f}M" elif model_size > 1e3: size_str = f"{model_size/1e3:.1f}K" else: size_str = str(model_size) # Get checkpoint filename for tooltip checkpoint_filename = model_info.get('checkpoint_filename', 'No checkpoint info') checkpoint_status = "LOADED" if model_info.get('checkpoint_loaded', False) else "FRESH" # Model card model_card = html.Div([ # Header with model name and toggle html.Div([ html.Div([ html.I(className=f"{status_icon} me-2 {status_class}"), html.Strong(f"{model_name.upper()}", className=status_class, title=f"Checkpoint: {checkpoint_filename}"), html.Span(f" ({size_str} params)", className="text-muted small ms-2"), html.Span(f" [{checkpoint_status}]", className=f"small {'text-success' if checkpoint_status == 'LOADED' else 'text-warning'} ms-1") ], style={"flex": "1"}), # Activation toggle (if easy to implement) html.Div([ dcc.Checklist( id=f"toggle-{model_name}", options=[{"label": "", "value": "active"}], value=["active"] if is_active else [], className="form-check-input", style={"transform": "scale(0.8)"} ) ], className="form-check form-switch") ], className="d-flex align-items-center mb-1"), # Model metrics html.Div([ # Last prediction html.Div([ html.Span("Last: ", className="text-muted small"), html.Span(f"{pred_action}", className=f"small fw-bold {'text-success' if pred_action == 'BUY' else 'text-danger' if pred_action == 'SELL' else 'text-muted'}"), html.Span(f" ({pred_confidence:.1f}%)", className="text-muted small"), html.Span(f" @ {pred_time}", className="text-muted small") ], className="mb-1"), # Loss metrics with improvement tracking html.Div([ html.Span("Current Loss: ", className="text-muted small"), html.Span(f"{loss_5ma:.4f}", className=f"small fw-bold {loss_class}") ] + ([ html.Span(" | Initial: ", className="text-muted small"), html.Span(f"{model_info.get('initial_loss', 0):.4f}", className="text-muted small") ] if model_info.get('initial_loss') else []) + ([ html.Span(" | ", className="text-muted small"), html.Span(f"↑{model_info.get('improvement', 0):.1f}%", className="small text-success") ] if model_info.get('improvement', 0) > 0 else []), className="mb-1"), # CNN Pivot Prediction (if available) *([self._format_cnn_pivot_prediction(model_info)] if model_info.get('pivot_prediction') else []) ]) ], className="border rounded p-2 mb-2", style={"backgroundColor": "rgba(255,255,255,0.05)" if is_active else "rgba(128,128,128,0.1)"}) content.append(model_card) else: content.append(html.P("No models loaded", className="text-warning small")) # COB $1 Buckets Section content.append(html.Hr()) content.append(html.H6([ html.I(className="fas fa-layer-group me-2 text-info"), "COB Buckets" ], className="mb-2")) if 'cob_buckets' in metrics_data: cob_buckets = metrics_data['cob_buckets'] if cob_buckets: for i, bucket in enumerate(cob_buckets[:3]): # Top 3 buckets price_range = f"${bucket['price']:.0f}-${bucket['price']+1:.0f}" volume = bucket.get('total_volume', 0) bid_pct = bucket.get('bid_pct', 0) ask_pct = bucket.get('ask_pct', 0) content.append(html.P([ html.Span(price_range, className="text-warning small fw-bold"), html.Br(), html.Span(f"Vol: ${volume:,.0f} ", className="text-muted small"), html.Span(f"B:{bid_pct:.0f}% ", className="text-success small"), html.Span(f"A:{ask_pct:.0f}%", className="text-danger small") ], className="mb-1")) else: content.append(html.P("COB buckets loading...", className="text-muted small")) else: content.append(html.P("COB data not available", className="text-warning small")) # Training Status (if available) if 'training_status' in metrics_data: training_status = metrics_data['training_status'] content.append(html.Hr()) content.append(html.H6([ html.I(className="fas fa-brain me-2 text-secondary"), "Training Status" ], className="mb-2")) content.append(html.P([ html.Span("Active Sessions: ", className="text-muted small"), html.Span(f"{training_status.get('active_sessions', 0)}", className="text-info small fw-bold") ], className="mb-1")) content.append(html.P([ html.Span("Last Update: ", className="text-muted small"), html.Span(f"{training_status.get('last_update', 'N/A')}", className="text-muted small") ])) return content except Exception as e: logger.error(f"Error formatting training metrics: {e}") return [html.P(f"Error: {str(e)}", className="text-danger small")]