276 lines
12 KiB
Python
276 lines
12 KiB
Python
"""
|
|
Dashboard Component Manager - Clean Trading Dashboard
|
|
Manages the formatting and creation of dashboard components
|
|
"""
|
|
|
|
from dash import html
|
|
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 recent_decisions[-10:]: # Last 10 signals
|
|
# 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_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")]
|
|
|
|
# Basic COB info
|
|
cob_info = []
|
|
|
|
# Symbol and update count
|
|
cob_info.append(html.Div([
|
|
html.Strong(f"{symbol}", className="text-info"),
|
|
html.Span(" - COB Snapshot", className="small text-muted")
|
|
], className="mb-2"))
|
|
|
|
# Mock COB data display (since we don't have real COB structure)
|
|
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_training_metrics(self, metrics_data):
|
|
"""Format training metrics for display"""
|
|
try:
|
|
if not metrics_data or 'error' in metrics_data:
|
|
return [html.P("No training data", className="text-muted small")]
|
|
|
|
metrics_info = []
|
|
|
|
# CNN metrics
|
|
if 'cnn_metrics' in metrics_data:
|
|
cnn_data = metrics_data['cnn_metrics']
|
|
metrics_info.append(html.Div([
|
|
html.Strong("CNN Model", className="text-primary"),
|
|
html.Br(),
|
|
html.Span(f"Status: Active", className="small text-success")
|
|
], className="mb-2"))
|
|
|
|
# RL metrics
|
|
if 'rl_metrics' in metrics_data:
|
|
rl_data = metrics_data['rl_metrics']
|
|
metrics_info.append(html.Div([
|
|
html.Strong("RL Model", className="text-warning"),
|
|
html.Br(),
|
|
html.Span(f"Status: Training", className="small text-info")
|
|
], className="mb-2"))
|
|
|
|
# Default message if no metrics
|
|
if not metrics_info:
|
|
metrics_info.append(html.P("Training metrics not available", className="text-muted small"))
|
|
|
|
return metrics_info
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error formatting training metrics: {e}")
|
|
return [html.P(f"Error: {str(e)}", className="text-danger small")] |