""" Dashboard Component Manager - Clean Trading Dashboard Manages the formatting and creation of dashboard components """ from dash import html, dcc import dash_bootstrap_components as dbc from datetime import datetime import logging import numpy as np 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 "" # Highlight COB signals cob_indicator = "" if hasattr(decision, 'type') and getattr(decision, 'type', '') == 'cob_liquidity_imbalance': cob_indicator = " [COB]" badge_class = "bg-info" # Use blue for COB signals elif isinstance(decision, dict) and decision.get('type') == 'cob_liquidity_imbalance': cob_indicator = " [COB]" badge_class = "bg-info" # Use blue for COB signals 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}{cob_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, trading_stats=None): """Format closed trades table for display with trading statistics""" try: # Create statistics header if trading stats are provided stats_header = [] if trading_stats and trading_stats.get('total_trades', 0) > 0: win_rate = trading_stats.get('win_rate', 0) avg_win = trading_stats.get('avg_win_size', 0) avg_loss = trading_stats.get('avg_loss_size', 0) total_trades = trading_stats.get('total_trades', 0) winning_trades = trading_stats.get('winning_trades', 0) losing_trades = trading_stats.get('losing_trades', 0) total_fees = trading_stats.get('total_fees', 0) breakeven_trades = trading_stats.get('breakeven_trades', 0) win_rate_class = "text-success" if win_rate >= 50 else "text-warning" if win_rate >= 30 else "text-danger" stats_header = [ html.Div([ html.H6("Trading Performance", className="mb-2"), html.Div([ html.Div([ html.Span("Win Rate: ", className="small text-muted"), html.Span(f"{win_rate:.1f}%", className=f"fw-bold {win_rate_class}"), html.Span(f" ({winning_trades}W/{losing_trades}L/{breakeven_trades}B)", className="small text-muted") ], className="col-3"), html.Div([ html.Span("Avg Win: ", className="small text-muted"), html.Span(f"${avg_win:.2f}", className="fw-bold text-success") ], className="col-3"), html.Div([ html.Span("Avg Loss: ", className="small text-muted"), html.Span(f"${avg_loss:.2f}", className="fw-bold text-danger") ], className="col-3"), html.Div([ html.Span("Total Fees: ", className="small text-muted"), html.Span(f"${total_fees:.2f}", className="fw-bold text-warning") ], className="col-3") ], className="row"), html.Hr(className="my-2") ], className="mb-3") ] if not closed_trades: if stats_header: return html.Div(stats_header + [html.P("No closed trades", className="text-muted small")]) else: 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("Hold (s)", className="small"), html.Th("P&L", className="small"), html.Th("Fees", className="small") ]) ]) # Create table rows rows = [] for trade in closed_trades: # Removed [-20:] to show all 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) hold_time_seconds = getattr(trade, 'hold_time_seconds', 0.0) else: # This is a dictionary format entry_time = trade.get('entry_time', 'Unknown') side = trade.get('side', 'UNKNOWN') size = trade.get('quantity', trade.get('size', 0)) # Try 'quantity' first, then 'size' entry_price = trade.get('entry_price', 0) exit_price = trade.get('exit_price', 0) pnl = trade.get('pnl', 0) fees = trade.get('fees', 0) hold_time_seconds = trade.get('hold_time_seconds', 0.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"{hold_time_seconds:.0f}", className="small text-info"), 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) table = html.Table([headers, tbody], className="table table-sm table-striped") # Wrap the table in a scrollable div scrollable_table_container = html.Div( table, style={'maxHeight': '300px', 'overflowY': 'scroll', 'overflowX': 'hidden'} ) # Combine statistics header with table if stats_header: return html.Div(stats_header + [scrollable_table_container]) else: return scrollable_table_container 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_cob_data(self, cob_snapshot, symbol, cumulative_imbalance_stats=None, cob_mode="Unknown"): """Format COB data into a split view with summary, imbalance stats, and a compact ladder.""" try: if not cob_snapshot: return html.Div([ html.H6(f"{symbol} COB", className="mb-2"), html.P("No COB data available", className="text-muted small"), html.P(f"Mode: {cob_mode}", className="text-muted small") ]) # Handle both old format (with stats attribute) and new format (direct attributes) if hasattr(cob_snapshot, 'stats'): # Old format with stats attribute stats = cob_snapshot.stats mid_price = stats.get('mid_price', 0) spread_bps = stats.get('spread_bps', 0) imbalance = stats.get('imbalance', 0) bids = getattr(cob_snapshot, 'consolidated_bids', []) asks = getattr(cob_snapshot, 'consolidated_asks', []) else: # New COBSnapshot format with direct attributes mid_price = getattr(cob_snapshot, 'volume_weighted_mid', 0) spread_bps = getattr(cob_snapshot, 'spread_bps', 0) imbalance = getattr(cob_snapshot, 'liquidity_imbalance', 0) bids = getattr(cob_snapshot, 'consolidated_bids', []) asks = getattr(cob_snapshot, 'consolidated_asks', []) if mid_price == 0 or not bids or not asks: return html.Div([ html.H6(f"{symbol} COB", className="mb-2"), html.P("Awaiting valid order book data...", className="text-muted small") ]) # Create stats dict for compatibility with existing code stats = { 'mid_price': mid_price, 'spread_bps': spread_bps, 'imbalance': imbalance, 'total_bid_liquidity': getattr(cob_snapshot, 'total_bid_liquidity', 0), 'total_ask_liquidity': getattr(cob_snapshot, 'total_ask_liquidity', 0), 'bid_levels': len(bids), 'ask_levels': len(asks) } # --- Left Panel: Overview and Stats --- overview_panel = self._create_cob_overview_panel(symbol, stats, cumulative_imbalance_stats, cob_mode) # --- Right Panel: Compact Ladder --- ladder_panel = self._create_cob_ladder_panel(bids, asks, mid_price, symbol) return dbc.Row([ dbc.Col(overview_panel, width=5, className="pe-1"), dbc.Col(ladder_panel, width=7, className="ps-1") ], className="g-0") # g-0 removes gutters except Exception as e: logger.error(f"Error formatting split COB data: {e}") return html.P(f"Error: {str(e)}", className="text-danger small") def _create_cob_overview_panel(self, symbol, stats, cumulative_imbalance_stats, cob_mode="Unknown"): """Creates the left panel with summary and imbalance stats.""" mid_price = stats.get('mid_price', 0) spread_bps = stats.get('spread_bps', 0) total_bid_liquidity = stats.get('total_bid_liquidity', 0) total_ask_liquidity = stats.get('total_ask_liquidity', 0) bid_levels = stats.get('bid_levels', 0) ask_levels = stats.get('ask_levels', 0) imbalance = stats.get('imbalance', 0) imbalance_text = f"Bid Heavy ({imbalance:.3f})" if imbalance > 0 else f"Ask Heavy ({imbalance:.3f})" imbalance_color = "text-success" if imbalance > 0 else "text-danger" # COB mode indicator mode_color = "text-success" if cob_mode == "WS" else "text-warning" if cob_mode == "REST" else "text-muted" mode_icon = "fas fa-wifi" if cob_mode == "WS" else "fas fa-globe" if cob_mode == "REST" else "fas fa-question" imbalance_stats_display = [] if cumulative_imbalance_stats: imbalance_stats_display.append(html.H6("Cumulative Imbalance", className="mt-3 mb-2 small text-muted text-uppercase")) for period, value in cumulative_imbalance_stats.items(): imbalance_stats_display.append(self._create_imbalance_stat_row(period, value)) return html.Div([ html.H6(f"{symbol} - COB Overview", className="mb-2"), html.Div([ html.Span([ html.I(className=f"{mode_icon} me-1 {mode_color}"), html.Span(f"Mode: {cob_mode}", className=f"small {mode_color}") ], className="mb-2") ]), html.Div([ self._create_stat_card("Mid Price", f"${mid_price:,.2f}", "fas fa-dollar-sign"), self._create_stat_card("Spread", f"{spread_bps:.1f} bps", "fas fa-arrows-alt-h") ], className="d-flex justify-content-between mb-2"), html.Div([ html.Strong("Snapshot Imbalance: ", className="small"), html.Span(imbalance_text, className=f"fw-bold small {imbalance_color}") ]), html.Div(imbalance_stats_display), html.Hr(className="my-2"), html.Table([ html.Tbody([ html.Tr([html.Td("Bid Liq.", className="small text-muted"), html.Td(f"${total_bid_liquidity/1e6:.2f}M", className="text-end small")]), html.Tr([html.Td("Ask Liq.", className="small text-muted"), html.Td(f"${total_ask_liquidity/1e6:.2f}M", className="text-end small")]), html.Tr([html.Td("Bid Levels", className="small text-muted"), html.Td(f"{bid_levels}", className="text-end small")]), html.Tr([html.Td("Ask Levels", className="small text-muted"), html.Td(f"{ask_levels}", className="text-end small")]) ]) ], className="table table-sm table-borderless") ], className="p-2 border rounded", style={"backgroundColor": "rgba(255,255,255,0.03)"}) def _create_imbalance_stat_row(self, period, value): """Helper to format a single row of cumulative imbalance.""" color = "text-success" if value > 0 else "text-danger" if value < 0 else "text-muted" icon = "fas fa-chevron-up" if value > 0 else "fas fa-chevron-down" if value < 0 else "fas fa-minus" return html.Div([ html.Span(f"{period}:", className="small text-muted", style={"width": "35px", "display": "inline-block"}), html.Span([ html.I(className=f"{icon} me-1 {color}"), html.Span(f"{value:+.3f}", className=f"fw-bold small {color}") ]) ], className="d-flex align-items-center mb-1") def _create_stat_card(self, title, value, icon): """Helper for creating small stat cards.""" return html.Div([ html.Div(title, className="small text-muted"), html.Div(value, className="fw-bold") ], className="text-center") def _create_cob_ladder_panel(self, bids, asks, mid_price, symbol=""): """Creates the right panel with the compact COB ladder.""" # Use symbol-specific bucket sizes: ETH = $1, BTC = $10 bucket_size = 1.0 if "ETH" in symbol else 10.0 num_levels = 5 def aggregate_buckets(orders): buckets = {} for order in orders: # Handle both dictionary format and ConsolidatedOrderBookLevel objects if hasattr(order, 'price'): # ConsolidatedOrderBookLevel object price = order.price size = order.total_size volume_usd = order.total_volume_usd else: # Dictionary format (legacy) price = order.get('price', 0) # Handle both old format (size) and new format (total_size) size = order.get('total_size', order.get('size', 0)) volume_usd = order.get('total_volume_usd', size * price) if price > 0: bucket_key = round(price / bucket_size) * bucket_size if bucket_key not in buckets: buckets[bucket_key] = {'usd_volume': 0, 'crypto_volume': 0} buckets[bucket_key]['usd_volume'] += volume_usd buckets[bucket_key]['crypto_volume'] += size return buckets bid_buckets = aggregate_buckets(bids) ask_buckets = aggregate_buckets(asks) all_usd_volumes = [b['usd_volume'] for b in bid_buckets.values()] + [a['usd_volume'] for a in ask_buckets.values()] max_volume = max(all_usd_volumes) if all_usd_volumes else 1 center_bucket = round(mid_price / bucket_size) * bucket_size ask_levels = [center_bucket + i * bucket_size for i in range(1, num_levels + 1)] bid_levels = [center_bucket - i * bucket_size for i in range(num_levels)] def create_ladder_row(price, bucket_data, max_vol, row_type): usd_volume = bucket_data.get('usd_volume', 0) crypto_volume = bucket_data.get('crypto_volume', 0) progress = (usd_volume / max_vol) * 100 if max_vol > 0 else 0 color = "danger" if row_type == 'ask' else "success" text_color = "text-danger" if row_type == 'ask' else "text-success" # Format USD volume (no $ symbol) if usd_volume > 1e6: usd_str = f"{usd_volume/1e6:.1f}M" elif usd_volume > 1e3: usd_str = f"{usd_volume/1e3:.0f}K" else: usd_str = f"{usd_volume:,.0f}" # Format crypto volume (no unit symbol) if crypto_volume > 1000: crypto_str = f"{crypto_volume/1000:.1f}K" elif crypto_volume > 1: crypto_str = f"{crypto_volume:.1f}" else: crypto_str = f"{crypto_volume:.3f}" return html.Tr([ html.Td(f"${price:,.0f}", className=f"{text_color} price-level small"), html.Td( dbc.Progress(value=progress, color=color, className="vh-25 compact-progress"), className="progress-cell p-0" ), html.Td(usd_str, className="volume-level text-end fw-bold small p-0 pe-1"), html.Td(crypto_str, className="volume-level text-start small text-muted p-0 ps-1") ], className="compact-ladder-row p-0") def get_bucket_data(buckets, price): return buckets.get(price, {'usd_volume': 0, 'crypto_volume': 0}) ask_rows = [create_ladder_row(p, get_bucket_data(ask_buckets, p), max_volume, 'ask') for p in sorted(ask_levels, reverse=True)] bid_rows = [create_ladder_row(p, get_bucket_data(bid_buckets, p), max_volume, 'bid') for p in sorted(bid_levels, reverse=True)] mid_row = html.Tr([ html.Td(f"${mid_price:,.0f}", colSpan=4, className="text-center fw-bold small mid-price-row p-0") ]) ladder_table = html.Table([ html.Thead(html.Tr([ html.Th("Price", className="small p-0"), html.Th("Volume", className="small p-0"), html.Th("USD", className="small text-end p-0 pe-1"), html.Th("Crypto", className="small text-start p-0 ps-1") ])), html.Tbody(ask_rows + [mid_row] + bid_rows) ], className="table table-sm table-borderless cob-ladder-table-compact m-0 p-0") # Compact classes return ladder_table 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'] # Get crypto volume if available (some bucket formats include crypto_volume) crypto_vol = bucket.get('crypto_volume', bucket.get('size', 0)) # Format USD 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}" # Format crypto volume based on symbol crypto_unit = "BTC" if "BTC" in symbol else "ETH" if "ETH" in symbol else "CRYPTO" if crypto_vol > 1000: crypto_str = f"{crypto_vol/1000:.1f}K {crypto_unit}" elif crypto_vol > 1: crypto_str = f"{crypto_vol:.1f} {crypto_unit}" elif crypto_vol > 0: crypto_str = f"{crypto_vol:.3f} {crypto_unit}" else: crypto_str = "" # 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(crypto_str, className="small text-muted me-2") if crypto_str else "", html.Span(f"{dominance}", className=f"small {dominance_class}") ], className="d-flex justify-content-between align-items-center"), 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 with enhanced details 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-warning' if 'PREDICTION' in pred_action else 'text-info'}"), html.Span(f" ({pred_confidence:.1f}%)", className="text-muted small"), html.Span(f" @ {pred_time}", className="text-muted small") ], className="mb-1"), # Additional prediction details if available *([ html.Div([ html.Span("Price: ", className="text-muted small"), html.Span(f"${last_prediction.get('predicted_price', 0):.2f}", className="text-warning small fw-bold") ], className="mb-1") ] if last_prediction.get('predicted_price', 0) > 0 else []), *([ html.Div([ html.Span("Change: ", className="text-muted small"), html.Span(f"{last_prediction.get('price_change', 0):+.2f}%", className=f"small fw-bold {'text-success' if last_prediction.get('price_change', 0) > 0 else 'text-danger'}") ], className="mb-1") ] if last_prediction.get('price_change', 0) != 0 else []), # Timing information (NEW) html.Div([ html.Span("Timing: ", className="text-muted small"), html.Span(f"Inf: {model_info.get('timing', {}).get('last_inference', 'None')}", className="text-info small"), html.Span(" | ", className="text-muted small"), html.Span(f"Train: {model_info.get('timing', {}).get('last_training', 'None')}", className="text-warning small"), html.Br(), html.Span(f"Rate: {model_info.get('timing', {}).get('inferences_per_second', '0.00')}/s", className="text-success small"), html.Span(" | ", className="text-muted small"), html.Span(f"24h: {model_info.get('timing', {}).get('predictions_24h', 0)}", className="text-primary 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")) 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") ])) # Enhanced Training Statistics (if available) if 'enhanced_training_stats' in metrics_data: enhanced_stats = metrics_data['enhanced_training_stats'] if enhanced_stats and not enhanced_stats.get('error'): content.append(html.Hr()) content.append(html.H6([ html.I(className="fas fa-rocket me-2 text-primary"), "Enhanced Training System" ], className="mb-2")) # Training system status is_training = enhanced_stats.get('is_training', False) training_iteration = enhanced_stats.get('training_iteration', 0) content.append(html.Div([ html.Span("Status: ", className="text-muted small"), html.Span("ACTIVE" if is_training else "INACTIVE", className=f"small fw-bold {'text-success' if is_training else 'text-warning'}"), html.Span(f" | Iteration: {training_iteration:,}", className="text-info small ms-2") ], className="mb-1")) # Buffer statistics exp_buffer_size = enhanced_stats.get('experience_buffer_size', 0) priority_buffer_size = enhanced_stats.get('priority_buffer_size', 0) content.append(html.Div([ html.Span("Experience Buffer: ", className="text-muted small"), html.Span(f"{exp_buffer_size:,}", className="text-success small fw-bold"), html.Span(" | Priority: ", className="text-muted small"), html.Span(f"{priority_buffer_size:,}", className="text-warning small fw-bold") ], className="mb-1")) # Data collection stats if 'data_collection_stats' in enhanced_stats: data_stats = enhanced_stats['data_collection_stats'] content.append(html.Div([ html.Span("Data: ", className="text-muted small"), html.Span(f"OHLCV: {data_stats.get('ohlcv_1m_bars', 0)}", className="text-info small"), html.Span(f" | Ticks: {data_stats.get('tick_data_points', 0)}", className="text-primary small"), html.Span(f" | COB: {data_stats.get('cob_snapshots', 0)}", className="text-success small") ], className="mb-1")) # Orchestrator Integration Stats (NEW) if 'orchestrator_integration' in enhanced_stats: orch_stats = enhanced_stats['orchestrator_integration'] content.append(html.Div([ html.Span("Integration: ", className="text-muted small"), html.Span(f"Models: {orch_stats.get('models_connected', 0)}", className="text-success small"), html.Span(f" | COB: {'ON' if orch_stats.get('cob_integration_active') else 'OFF'}", className=f"small {'text-success' if orch_stats.get('cob_integration_active') else 'text-warning'}"), html.Span(f" | Fusion: {'ON' if orch_stats.get('decision_fusion_enabled') else 'OFF'}", className=f"small {'text-success' if orch_stats.get('decision_fusion_enabled') else 'text-warning'}"), html.Span(f" | Symbols: {orch_stats.get('symbols_tracking', 0)}", className="text-info small") ], className="mb-1")) content.append(html.Div([ html.Span("Decisions: ", className="text-muted small"), html.Span(f"{orch_stats.get('recent_decisions_count', 0):,}", className="text-primary small fw-bold"), html.Span(" | RT Processing: ", className="text-muted small"), html.Span("ON" if orch_stats.get('realtime_processing') else "OFF", className=f"small {'text-success' if orch_stats.get('realtime_processing') else 'text-muted'}") ], className="mb-1")) # Model Training Status (NEW) if 'model_training_status' in enhanced_stats: model_status = enhanced_stats['model_training_status'] content.append(html.Div([ html.Span("Model Status: ", className="text-muted small"), html.Br() ] + [ html.Div([ html.Span(f"{model_name.upper()}: ", className="text-muted small"), html.Span("LOADED" if status.get('model_loaded') else "MISSING", className=f"small {'text-success' if status.get('model_loaded') else 'text-danger'}"), html.Span(f" | Mem: {status.get('memory_usage', 0):,}", className="text-info small"), html.Span(f" | Steps: {status.get('training_steps', 0):,}", className="text-warning small"), *([html.Span(f" | Loss: {status['last_loss']:.4f}", className="text-primary small")] if status.get('last_loss') is not None else []) ], className="ms-2 mb-1") for model_name, status in model_status.items() ], className="mb-1")) # Prediction Tracking Stats (NEW) if 'prediction_tracking' in enhanced_stats: pred_stats = enhanced_stats['prediction_tracking'] content.append(html.Div([ html.Span("Predictions: ", className="text-muted small"), html.Span(f"DQN: {pred_stats.get('dqn_predictions_tracked', 0):,}", className="text-success small"), html.Span(f" | CNN: {pred_stats.get('cnn_predictions_tracked', 0):,}", className="text-warning small"), html.Span(f" | Accuracy: {pred_stats.get('accuracy_history_tracked', 0):,}", className="text-info small") ], className="mb-1")) symbols_with_preds = pred_stats.get('symbols_with_predictions', []) if symbols_with_preds: content.append(html.Div([ html.Span("Active Symbols: ", className="text-muted small"), html.Span(", ".join(symbols_with_preds), className="text-primary small fw-bold") ], className="mb-1")) # COB Integration Stats (NEW) if 'cob_integration_stats' in enhanced_stats: cob_stats = enhanced_stats['cob_integration_stats'] content.append(html.Div([ html.Span("COB Data: ", className="text-muted small"), html.Span(f"Symbols: {len(cob_stats.get('latest_cob_data_symbols', []))}", className="text-success small"), html.Span(f" | Features: {len(cob_stats.get('cob_features_available', []))}", className="text-warning small"), html.Span(f" | States: {len(cob_stats.get('cob_state_available', []))}", className="text-info small") ], className="mb-1")) # Recent losses if enhanced_stats.get('dqn_recent_loss') is not None: content.append(html.Div([ html.Span("DQN Loss: ", className="text-muted small"), html.Span(f"{enhanced_stats['dqn_recent_loss']:.4f}", className="text-info small fw-bold") ], className="mb-1")) if enhanced_stats.get('cnn_recent_loss') is not None: content.append(html.Div([ html.Span("CNN Loss: ", className="text-muted small"), html.Span(f"{enhanced_stats['cnn_recent_loss']:.4f}", className="text-warning small fw-bold") ], className="mb-1")) # Validation score if enhanced_stats.get('recent_validation_score') is not None: content.append(html.Div([ html.Span("Validation Score: ", className="text-muted small"), html.Span(f"{enhanced_stats['recent_validation_score']:.3f}", className="text-primary small fw-bold") ], className="mb-1")) 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")] 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()