490 lines
24 KiB
Python
490 lines
24 KiB
Python
"""
|
|
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_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
|
|
loss_5ma = model_info.get('loss_5ma', 0.0)
|
|
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)
|
|
|
|
# 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),
|
|
html.Span(f" ({size_str} params)", className="text-muted small ms-2")
|
|
], 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 $1 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")] |