""" Dashboard Data Model Provides structured data for template rendering """ from dataclasses import dataclass, field from typing import List, Dict, Any, Optional from datetime import datetime @dataclass class MetricData: """Individual metric for the dashboard""" id: str label: str value: str format_type: str = "text" # text, currency, percentage @dataclass class TradingControlsData: """Trading controls configuration""" buy_text: str = "BUY" sell_text: str = "SELL" clear_text: str = "Clear Session" leverage: int = 10 leverage_min: int = 1 leverage_max: int = 50 @dataclass class RecentDecisionData: """Recent AI decision data""" timestamp: str action: str symbol: str confidence: float price: float @dataclass class COBLevelData: """Order book level data""" side: str # 'bid' or 'ask' size: str price: str total: str @dataclass class COBData: """Complete order book data for a symbol""" symbol: str content_id: str total_usd: str total_crypto: str levels: List[COBLevelData] = field(default_factory=list) @dataclass class ModelData: """Model status data""" name: str status: str # 'training', 'idle', 'loading' status_text: str @dataclass class TrainingMetricData: """Training metric data""" name: str value: str @dataclass class PerformanceStatData: """Performance statistic data""" name: str value: str @dataclass class ClosedTradeData: """Closed trade data""" time: str symbol: str side: str size: str entry_price: str exit_price: str pnl: float duration: str @dataclass class ChartData: """Chart configuration data""" title: str = "Price Chart & Signals" @dataclass class DashboardModel: """Complete dashboard data model""" title: str = "Live Scalping Dashboard" subtitle: str = "Real-time Trading with AI Models" refresh_interval: int = 1000 # Main sections metrics: List[MetricData] = field(default_factory=list) chart: ChartData = field(default_factory=ChartData) trading_controls: TradingControlsData = field(default_factory=TradingControlsData) recent_decisions: List[RecentDecisionData] = field(default_factory=list) cob_data: List[COBData] = field(default_factory=list) models: List[ModelData] = field(default_factory=list) training_metrics: List[TrainingMetricData] = field(default_factory=list) performance_stats: List[PerformanceStatData] = field(default_factory=list) closed_trades: List[ClosedTradeData] = field(default_factory=list) class DashboardDataBuilder: """Builder class to construct dashboard data from various sources""" def __init__(self): self.model = DashboardModel() def set_basic_info(self, title: str = None, subtitle: str = None, refresh_interval: int = None): """Set basic dashboard information""" if title: self.model.title = title if subtitle: self.model.subtitle = subtitle if refresh_interval: self.model.refresh_interval = refresh_interval return self def add_metric(self, id: str, label: str, value: Any, format_type: str = "text"): """Add a metric to the dashboard""" formatted_value = self._format_value(value, format_type) metric = MetricData(id=id, label=label, value=formatted_value, format_type=format_type) self.model.metrics.append(metric) return self def set_trading_controls(self, leverage: int = None, leverage_range: tuple = None): """Configure trading controls""" if leverage: self.model.trading_controls.leverage = leverage if leverage_range: self.model.trading_controls.leverage_min = leverage_range[0] self.model.trading_controls.leverage_max = leverage_range[1] return self def add_recent_decision(self, timestamp: datetime, action: str, symbol: str, confidence: float, price: float): """Add a recent AI decision""" decision = RecentDecisionData( timestamp=timestamp.strftime("%H:%M:%S"), action=action, symbol=symbol, confidence=round(confidence * 100, 1), price=round(price, 4) ) self.model.recent_decisions.append(decision) return self def add_cob_data(self, symbol: str, content_id: str, total_usd: float, total_crypto: float, levels: List[Dict]): """Add COB data for a symbol""" cob_levels = [] for level in levels: cob_level = COBLevelData( side=level.get('side', 'bid'), size=self._format_value(level.get('size', 0), 'number'), price=self._format_value(level.get('price', 0), 'currency'), total=self._format_value(level.get('total', 0), 'currency') ) cob_levels.append(cob_level) cob = COBData( symbol=symbol, content_id=content_id, total_usd=self._format_value(total_usd, 'currency'), total_crypto=self._format_value(total_crypto, 'number'), levels=cob_levels ) self.model.cob_data.append(cob) return self def add_model_status(self, name: str, is_training: bool, is_loading: bool = False): """Add model status""" if is_loading: status = "loading" status_text = "Loading" elif is_training: status = "training" status_text = "Training" else: status = "idle" status_text = "Idle" model = ModelData(name=name, status=status, status_text=status_text) self.model.models.append(model) return self def add_training_metric(self, name: str, value: Any): """Add training metric""" metric = TrainingMetricData( name=name, value=self._format_value(value, 'number') ) self.model.training_metrics.append(metric) return self def add_performance_stat(self, name: str, value: Any): """Add performance statistic""" stat = PerformanceStatData( name=name, value=self._format_value(value, 'number') ) self.model.performance_stats.append(stat) return self def add_closed_trade(self, time: datetime, symbol: str, side: str, size: float, entry_price: float, exit_price: float, pnl: float, duration: str): """Add closed trade""" trade = ClosedTradeData( time=time.strftime("%H:%M:%S"), symbol=symbol, side=side, size=self._format_value(size, 'number'), entry_price=self._format_value(entry_price, 'currency'), exit_price=self._format_value(exit_price, 'currency'), pnl=round(pnl, 2), duration=duration ) self.model.closed_trades.append(trade) return self def build(self) -> DashboardModel: """Build and return the complete dashboard model""" return self.model def _format_value(self, value: Any, format_type: str) -> str: """Format value based on type""" if value is None: return "N/A" try: if format_type == "currency": return f"${float(value):,.4f}" elif format_type == "percentage": return f"{float(value):.2f}%" elif format_type == "number": if isinstance(value, int): return f"{value:,}" else: return f"{float(value):,.2f}" else: return str(value) except (ValueError, TypeError): return str(value) def create_sample_dashboard_data() -> DashboardModel: """Create sample dashboard data for testing""" builder = DashboardDataBuilder() # Basic info builder.set_basic_info( title="Live Scalping Dashboard", subtitle="Real-time Trading with AI Models", refresh_interval=1000 ) # Metrics builder.add_metric("current-price", "Current Price", 3425.67, "currency") builder.add_metric("session-pnl", "Session PnL", 125.34, "currency") builder.add_metric("current-position", "Position", 0.0, "number") builder.add_metric("trade-count", "Trades", 15, "number") builder.add_metric("portfolio-value", "Portfolio", 10250.45, "currency") builder.add_metric("mexc-status", "MEXC Status", "Connected", "text") # Trading controls builder.set_trading_controls(leverage=10, leverage_range=(1, 50)) # Recent decisions builder.add_recent_decision(datetime.now(), "BUY", "ETH/USDT", 0.85, 3425.67) builder.add_recent_decision(datetime.now(), "HOLD", "BTC/USDT", 0.62, 45123.45) # COB data eth_levels = [ {"side": "ask", "size": 1.5, "price": 3426.12, "total": 5139.18}, {"side": "ask", "size": 2.3, "price": 3425.89, "total": 7879.55}, {"side": "bid", "size": 1.8, "price": 3425.45, "total": 6165.81}, {"side": "bid", "size": 3.2, "price": 3425.12, "total": 10960.38} ] builder.add_cob_data("ETH/USDT", "eth-cob-content", 25000.0, 7.3, eth_levels) btc_levels = [ {"side": "ask", "size": 0.15, "price": 45125.67, "total": 6768.85}, {"side": "ask", "size": 0.23, "price": 45123.45, "total": 10378.39}, {"side": "bid", "size": 0.18, "price": 45121.23, "total": 8121.82}, {"side": "bid", "size": 0.32, "price": 45119.12, "total": 14438.12} ] builder.add_cob_data("BTC/USDT", "btc-cob-content", 35000.0, 0.88, btc_levels) # 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) builder.add_training_metric("Learning Rate", 0.0001) # Performance stats builder.add_performance_stat("Win Rate", 68.5) builder.add_performance_stat("Avg Trade", 8.34) builder.add_performance_stat("Max Drawdown", -45.67) builder.add_performance_stat("Sharpe Ratio", 1.82) # Closed trades builder.add_closed_trade( datetime.now(), "ETH/USDT", "BUY", 1.5, 3420.45, 3428.12, 11.51, "2m 34s" ) builder.add_closed_trade( datetime.now(), "BTC/USDT", "SELL", 0.1, 45150.23, 45142.67, -0.76, "1m 12s" ) return builder.build()