268 lines
8.2 KiB
Python
268 lines
8.2 KiB
Python
"""
|
|
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)
|
|
|
|
|
|
# CRITICAL POLICY: NEVER create mock/sample data functions
|
|
# All dashboard data MUST come from real market data or be empty/None
|
|
# This function was removed to prevent synthetic data usage
|
|
# See: reports/REAL_MARKET_DATA_POLICY.md
|
|
#
|
|
# If you need to test the dashboard, use real market data from exchanges
|
|
# or run with empty data to identify what needs to be implemented |