Files
gogo2/web/component_manager.py
Dobromir Popov d269a1fe6e minor UI changes
2025-07-02 00:17:18 +03:00

803 lines
41 KiB
Python

"""
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 ""
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, 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)
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)", className="small text-muted")
], className="col-4"),
html.Div([
html.Span("Avg Win: ", className="small text-muted"),
html.Span(f"${avg_win:.2f}", className="fw-bold text-success")
], className="col-4"),
html.Div([
html.Span("Avg Loss: ", className="small text-muted"),
html.Span(f"${avg_loss:.2f}", className="fw-bold text-danger")
], className="col-4")
], 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("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)
table = html.Table([headers, tbody], className="table table-sm table-striped")
# Combine statistics header with table
if stats_header:
return html.Div(stats_header + [table])
else:
return table
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):
"""Format COB data into a split view with summary, imbalance stats, and a compact ladder."""
try:
if not cob_snapshot or not hasattr(cob_snapshot, 'stats'):
return html.Div([
html.H6(f"{symbol} COB", className="mb-2"),
html.P("No COB data available", className="text-muted small")
])
stats = cob_snapshot.stats if hasattr(cob_snapshot, 'stats') else {}
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', [])
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")
])
# --- Left Panel: Overview and Stats ---
overview_panel = self._create_cob_overview_panel(symbol, stats, cumulative_imbalance_stats)
# --- 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):
"""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"
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([
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."""
bucket_size = 10
num_levels = 5
def aggregate_buckets(orders):
buckets = {}
for order in orders:
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
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"),
# 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"))
# 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")]
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()