Files
gogo2/web/templated_dashboard.py
Dobromir Popov 6acc1c9296 Add template-based MVC dashboard architecture
- Add HTML templates for clean separation of concerns
- Add structured data models for type safety
- Add template renderer for Jinja2 integration
- Add templated dashboard implementation
- Demonstrates 95% file size reduction potential
2025-07-02 01:56:50 +03:00

598 lines
22 KiB
Python

"""
Template-based Trading Dashboard
Uses MVC architecture with HTML templates and data models
"""
import logging
from typing import Optional, Any, Dict, List
from datetime import datetime
import pandas as pd
import dash
from dash import dcc, html, Input, Output, State, callback_context
import plotly.graph_objects as go
import plotly.express as px
from .dashboard_model import DashboardModel, DashboardDataBuilder, create_sample_dashboard_data
from .template_renderer import DashboardTemplateRenderer
from core.data_provider import DataProvider
from core.orchestrator import TradingOrchestrator
from core.trading_executor import TradingExecutor
# Configure logging
logger = logging.getLogger(__name__)
class TemplatedTradingDashboard:
"""Template-based trading dashboard with MVC architecture"""
def __init__(self, data_provider: Optional[DataProvider] = None,
orchestrator: Optional[TradingOrchestrator] = None,
trading_executor: Optional[TradingExecutor] = None):
"""Initialize the templated dashboard"""
self.data_provider = data_provider
self.orchestrator = orchestrator
self.trading_executor = trading_executor
# Initialize template renderer
self.renderer = DashboardTemplateRenderer()
# Initialize Dash app
self.app = dash.Dash(__name__, external_stylesheets=[
'https://cdn.jsdelivr.net/npm/bootstrap@5.1.3/dist/css/bootstrap.min.css'
])
# Session data
self.session_start_time = datetime.now()
self.session_trades = []
self.session_pnl = 0.0
self.current_position = 0.0
# Setup layout and callbacks
self._setup_layout()
self._setup_callbacks()
logger.info("TEMPLATED DASHBOARD: Initialized with MVC architecture")
def _setup_layout(self):
"""Setup the dashboard layout using templates"""
# Create initial dashboard data
dashboard_data = self._build_dashboard_data()
# Render layout using template
layout = self.renderer.render_dashboard(dashboard_data)
# Add custom CSS
layout.children.insert(0, self._get_custom_css())
self.app.layout = layout
def _get_custom_css(self) -> html.Style:
"""Get custom CSS styles"""
return html.Style(children="""
.metric-card {
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
border-radius: 10px;
padding: 15px;
margin-bottom: 10px;
box-shadow: 0 4px 6px rgba(0, 0, 0, 0.1);
}
.metric-value {
font-size: 1.5rem;
font-weight: bold;
}
.metric-label {
font-size: 0.9rem;
opacity: 0.9;
}
.cob-ladder {
max-height: 400px;
overflow-y: auto;
font-family: 'Courier New', monospace;
font-size: 0.85rem;
}
.bid-row {
background-color: rgba(40, 167, 69, 0.1);
border-left: 3px solid #28a745;
}
.ask-row {
background-color: rgba(220, 53, 69, 0.1);
border-left: 3px solid #dc3545;
}
.training-panel {
background: #f8f9fa;
border-radius: 8px;
padding: 15px;
height: 300px;
overflow-y: auto;
}
.model-status {
padding: 8px 12px;
border-radius: 20px;
font-size: 0.8rem;
font-weight: bold;
margin: 2px;
display: inline-block;
}
.status-training { background-color: #28a745; color: white; }
.status-idle { background-color: #6c757d; color: white; }
.status-loading { background-color: #ffc107; color: black; }
.closed-trades {
max-height: 200px;
overflow-y: auto;
}
.trade-profit { color: #28a745; font-weight: bold; }
.trade-loss { color: #dc3545; font-weight: bold; }
""")
def _setup_callbacks(self):
"""Setup dashboard callbacks"""
@self.app.callback(
[Output('current-price', 'children'),
Output('session-pnl', 'children'),
Output('current-position', 'children'),
Output('trade-count', 'children'),
Output('portfolio-value', 'children'),
Output('mexc-status', 'children')],
[Input('interval-component', 'n_intervals')]
)
def update_metrics(n):
"""Update main metrics"""
try:
# Get current price
current_price = self._get_current_price("ETH/USDT")
# Calculate portfolio value
portfolio_value = 10000.0 + self.session_pnl # Base + PnL
# Get MEXC status
mexc_status = "Connected" if self.trading_executor else "Disconnected"
return (
f"${current_price:.4f}" if current_price else "N/A",
f"${self.session_pnl:.2f}",
f"{self.current_position:.4f}",
str(len(self.session_trades)),
f"${portfolio_value:.2f}",
mexc_status
)
except Exception as e:
logger.error(f"Error updating metrics: {e}")
return "N/A", "N/A", "N/A", "N/A", "N/A", "Error"
@self.app.callback(
Output('price-chart', 'figure'),
[Input('interval-component', 'n_intervals')]
)
def update_price_chart(n):
"""Update price chart"""
try:
return self._create_price_chart("ETH/USDT")
except Exception as e:
logger.error(f"Error updating chart: {e}")
return go.Figure()
@self.app.callback(
Output('recent-decisions', 'children'),
[Input('interval-component', 'n_intervals')]
)
def update_recent_decisions(n):
"""Update recent AI decisions"""
try:
decisions = self._get_recent_decisions()
return self._render_decisions(decisions)
except Exception as e:
logger.error(f"Error updating decisions: {e}")
return html.Div("No recent decisions")
@self.app.callback(
[Output('eth-cob-content', 'children'),
Output('btc-cob-content', 'children')],
[Input('interval-component', 'n_intervals')]
)
def update_cob_data(n):
"""Update COB data"""
try:
eth_cob = self._render_cob_ladder("ETH/USDT")
btc_cob = self._render_cob_ladder("BTC/USDT")
return eth_cob, btc_cob
except Exception as e:
logger.error(f"Error updating COB: {e}")
return html.Div("COB Error"), html.Div("COB Error")
@self.app.callback(
Output('training-metrics', 'children'),
[Input('interval-component', 'n_intervals')]
)
def update_training_metrics(n):
"""Update training metrics"""
try:
return self._render_training_metrics()
except Exception as e:
logger.error(f"Error updating training metrics: {e}")
return html.Div("Training metrics unavailable")
@self.app.callback(
Output('closed-trades-table', 'children'),
[Input('interval-component', 'n_intervals')]
)
def update_closed_trades(n):
"""Update closed trades table"""
try:
return self._render_closed_trades()
except Exception as e:
logger.error(f"Error updating closed trades: {e}")
return html.Div("No trades")
# Trading control callbacks
@self.app.callback(
Output('manual-buy-btn', 'children'),
[Input('manual-buy-btn', 'n_clicks')],
prevent_initial_call=True
)
def handle_manual_buy(n_clicks):
"""Handle manual buy button"""
if n_clicks:
self._execute_manual_trade("BUY")
return "BUY ✓"
return "BUY"
@self.app.callback(
Output('manual-sell-btn', 'children'),
[Input('manual-sell-btn', 'n_clicks')],
prevent_initial_call=True
)
def handle_manual_sell(n_clicks):
"""Handle manual sell button"""
if n_clicks:
self._execute_manual_trade("SELL")
return "SELL ✓"
return "SELL"
@self.app.callback(
Output('leverage-display', 'children'),
[Input('leverage-slider', 'value')]
)
def update_leverage_display(leverage_value):
"""Update leverage display"""
return f"{leverage_value}x"
@self.app.callback(
Output('clear-session-btn', 'children'),
[Input('clear-session-btn', 'n_clicks')],
prevent_initial_call=True
)
def handle_clear_session(n_clicks):
"""Handle clear session button"""
if n_clicks:
self._clear_session()
return "Cleared ✓"
return "Clear Session"
def _build_dashboard_data(self) -> DashboardModel:
"""Build dashboard data model from current state"""
builder = DashboardDataBuilder()
# Basic info
builder.set_basic_info(
title="Live Scalping Dashboard (Templated)",
subtitle="Template-based MVC Architecture",
refresh_interval=1000
)
# Get current metrics
current_price = self._get_current_price("ETH/USDT")
portfolio_value = 10000.0 + self.session_pnl
mexc_status = "Connected" if self.trading_executor else "Disconnected"
# Add metrics
builder.add_metric("current-price", "Current Price", current_price or 0, "currency")
builder.add_metric("session-pnl", "Session PnL", self.session_pnl, "currency")
builder.add_metric("current-position", "Position", self.current_position, "number")
builder.add_metric("trade-count", "Trades", len(self.session_trades), "number")
builder.add_metric("portfolio-value", "Portfolio", portfolio_value, "currency")
builder.add_metric("mexc-status", "MEXC Status", mexc_status, "text")
# Trading controls
builder.set_trading_controls(leverage=10, leverage_range=(1, 50))
# Recent decisions (sample data for now)
builder.add_recent_decision(datetime.now(), "BUY", "ETH/USDT", 0.85, current_price or 3425.67)
# COB data (sample)
builder.add_cob_data("ETH/USDT", "eth-cob-content", 25000.0, 7.3, [])
builder.add_cob_data("BTC/USDT", "btc-cob-content", 35000.0, 0.88, [])
# Model statuses
builder.add_model_status("DQN", True)
builder.add_model_status("CNN", True)
builder.add_model_status("Transformer", False)
builder.add_model_status("COB-RL", True)
# Training metrics
builder.add_training_metric("DQN Loss", 0.0234)
builder.add_training_metric("CNN Accuracy", 0.876)
builder.add_training_metric("Training Steps", 15420)
# Performance stats
builder.add_performance_stat("Win Rate", 68.5)
builder.add_performance_stat("Avg Trade", 8.34)
builder.add_performance_stat("Sharpe Ratio", 1.82)
return builder.build()
def _get_current_price(self, symbol: str) -> Optional[float]:
"""Get current price for symbol"""
try:
if self.data_provider:
return self.data_provider.get_current_price(symbol)
return 3425.67 # Sample price
except Exception as e:
logger.error(f"Error getting price for {symbol}: {e}")
return None
def _create_price_chart(self, symbol: str) -> go.Figure:
"""Create price chart"""
try:
# Get price data
df = self._get_chart_data(symbol)
if df is None or df.empty:
return go.Figure().add_annotation(
text="No data available",
xref="paper", yref="paper",
x=0.5, y=0.5, showarrow=False
)
# Create candlestick chart
fig = go.Figure(data=[go.Candlestick(
x=df.index,
open=df['open'],
high=df['high'],
low=df['low'],
close=df['close'],
name=symbol
)])
fig.update_layout(
title=f"{symbol} Price Chart",
xaxis_title="Time",
yaxis_title="Price (USDT)",
height=500,
showlegend=False
)
return fig
except Exception as e:
logger.error(f"Error creating chart for {symbol}: {e}")
return go.Figure()
def _get_chart_data(self, symbol: str) -> Optional[pd.DataFrame]:
"""Get chart data for symbol"""
try:
if self.data_provider:
return self.data_provider.get_historical_data(symbol, "1m", 100)
# Sample data
import numpy as np
dates = pd.date_range(start='2024-01-01', periods=100, freq='1min')
base_price = 3425.67
df = pd.DataFrame({
'open': base_price + np.random.randn(100) * 10,
'high': base_price + np.random.randn(100) * 15,
'low': base_price + np.random.randn(100) * 15,
'close': base_price + np.random.randn(100) * 10,
'volume': np.random.randint(100, 1000, 100)
}, index=dates)
return df
except Exception as e:
logger.error(f"Error getting chart data: {e}")
return None
def _get_recent_decisions(self) -> List[Dict]:
"""Get recent AI decisions"""
# Sample decisions for now
return [
{
"timestamp": datetime.now().strftime("%H:%M:%S"),
"action": "BUY",
"symbol": "ETH/USDT",
"confidence": 85.3,
"price": 3425.67
},
{
"timestamp": datetime.now().strftime("%H:%M:%S"),
"action": "HOLD",
"symbol": "BTC/USDT",
"confidence": 62.1,
"price": 45123.45
}
]
def _render_decisions(self, decisions: List[Dict]) -> List[html.Div]:
"""Render recent decisions"""
items = []
for decision in decisions:
border_class = {
'BUY': 'border-success bg-success bg-opacity-10',
'SELL': 'border-danger bg-danger bg-opacity-10'
}.get(decision['action'], 'border-secondary bg-secondary bg-opacity-10')
items.append(
html.Div([
html.Small(decision['timestamp'], className="text-muted"),
html.Br(),
html.Strong(f"{decision['action']} - {decision['symbol']}"),
html.Br(),
html.Small(f"Confidence: {decision['confidence']}% | Price: ${decision['price']}")
], className=f"mb-2 p-2 border-start border-3 {border_class}")
)
return items
def _render_cob_ladder(self, symbol: str) -> html.Div:
"""Render COB ladder for symbol"""
# Sample COB data
return html.Table([
html.Thead([
html.Tr([
html.Th("Size"),
html.Th("Price"),
html.Th("Total")
])
]),
html.Tbody([
html.Tr([
html.Td("1.5"),
html.Td("$3426.12"),
html.Td("$5139.18")
], className="ask-row"),
html.Tr([
html.Td("2.3"),
html.Td("$3425.89"),
html.Td("$7879.55")
], className="ask-row"),
html.Tr([
html.Td("1.8"),
html.Td("$3425.45"),
html.Td("$6165.81")
], className="bid-row"),
html.Tr([
html.Td("3.2"),
html.Td("$3425.12"),
html.Td("$10960.38")
], className="bid-row")
])
], className="table table-sm table-borderless")
def _render_training_metrics(self) -> html.Div:
"""Render training metrics"""
return html.Div([
# Model Status
html.Div([
html.H6("Model Status"),
html.Div([
html.Span("DQN: Training", className="model-status status-training"),
html.Span("CNN: Training", className="model-status status-training"),
html.Span("Transformer: Idle", className="model-status status-idle"),
html.Span("COB-RL: Training", className="model-status status-training")
])
], className="mb-3"),
# Training Metrics
html.Div([
html.H6("Training Metrics"),
html.Div([
html.Div([
html.Div([html.Small("DQN Loss:")], className="col-6"),
html.Div([html.Small("0.0234", className="fw-bold")], className="col-6")
], className="row mb-1"),
html.Div([
html.Div([html.Small("CNN Accuracy:")], className="col-6"),
html.Div([html.Small("87.6%", className="fw-bold")], className="col-6")
], className="row mb-1"),
html.Div([
html.Div([html.Small("Training Steps:")], className="col-6"),
html.Div([html.Small("15,420", className="fw-bold")], className="col-6")
], className="row mb-1")
])
], className="mb-3"),
# Performance Stats
html.Div([
html.H6("Performance"),
html.Div([
html.Div([
html.Div([html.Small("Win Rate:")], className="col-8"),
html.Div([html.Small("68.5%", className="fw-bold")], className="col-4")
], className="row mb-1"),
html.Div([
html.Div([html.Small("Avg Trade:")], className="col-8"),
html.Div([html.Small("$8.34", className="fw-bold")], className="col-4")
], className="row mb-1"),
html.Div([
html.Div([html.Small("Sharpe Ratio:")], className="col-8"),
html.Div([html.Small("1.82", className="fw-bold")], className="col-4")
], className="row mb-1")
])
])
])
def _render_closed_trades(self) -> html.Table:
"""Render closed trades table"""
return html.Table([
html.Thead([
html.Tr([
html.Th("Time"),
html.Th("Symbol"),
html.Th("Side"),
html.Th("Size"),
html.Th("Entry"),
html.Th("Exit"),
html.Th("PnL"),
html.Th("Duration")
])
]),
html.Tbody([
html.Tr([
html.Td("14:23:45"),
html.Td("ETH/USDT"),
html.Td([html.Span("BUY", className="badge bg-success")]),
html.Td("1.5"),
html.Td("$3420.45"),
html.Td("$3428.12"),
html.Td("$11.51", className="trade-profit"),
html.Td("2m 34s")
]),
html.Tr([
html.Td("14:21:12"),
html.Td("BTC/USDT"),
html.Td([html.Span("SELL", className="badge bg-danger")]),
html.Td("0.1"),
html.Td("$45150.23"),
html.Td("$45142.67"),
html.Td("-$0.76", className="trade-loss"),
html.Td("1m 12s")
])
])
], className="table table-sm")
def _execute_manual_trade(self, action: str):
"""Execute manual trade"""
try:
logger.info(f"MANUAL TRADE: {action} executed")
# Add to session trades
trade = {
"time": datetime.now(),
"action": action,
"symbol": "ETH/USDT",
"price": self._get_current_price("ETH/USDT") or 3425.67
}
self.session_trades.append(trade)
except Exception as e:
logger.error(f"Error executing manual trade: {e}")
def _clear_session(self):
"""Clear session data"""
self.session_trades = []
self.session_pnl = 0.0
self.current_position = 0.0
self.session_start_time = datetime.now()
logger.info("SESSION: Cleared")
def run_server(self, host='127.0.0.1', port=8051, debug=False):
"""Run the dashboard server"""
logger.info(f"TEMPLATED DASHBOARD: Starting at http://{host}:{port}")
self.app.run_server(host=host, port=port, debug=debug)
def create_templated_dashboard(data_provider: Optional[DataProvider] = None,
orchestrator: Optional[TradingOrchestrator] = None,
trading_executor: Optional[TradingExecutor] = None) -> TemplatedTradingDashboard:
"""Create templated trading dashboard"""
return TemplatedTradingDashboard(data_provider, orchestrator, trading_executor)