""" Clean Trading Dashboard - Modular Implementation This dashboard is fully integrated with the Universal Data Stream architecture and receives the standardized 5 timeseries format: UNIVERSAL DATA FORMAT (The Sacred 5): 1. ETH/USDT Ticks (1s) - Primary trading pair real-time data 2. ETH/USDT 1m - Short-term price action and patterns 3. ETH/USDT 1h - Medium-term trends and momentum 4. ETH/USDT 1d - Long-term market structure 5. BTC/USDT Ticks (1s) - Reference asset for correlation analysis The dashboard subscribes to the UnifiedDataStream as a consumer and receives real-time updates for all 5 timeseries through a standardized callback. This ensures consistent data across all models and components. Uses layout and component managers to reduce file size and improve maintainability """ import dash from dash import Dash, dcc, html, Input, Output, State import plotly.graph_objects as go from plotly.subplots import make_subplots import pandas as pd import numpy as np from datetime import datetime, timedelta, timezone import pytz import logging import json import time import threading from typing import Dict, List, Optional, Any, Union import os import asyncio import dash_bootstrap_components as dbc from dash.exceptions import PreventUpdate from collections import deque from threading import Lock import warnings from dataclasses import asdict # Setup logger logger = logging.getLogger(__name__) # Reduce Werkzeug/Dash logging noise logging.getLogger('werkzeug').setLevel(logging.WARNING) logging.getLogger('dash').setLevel(logging.WARNING) logging.getLogger('dash.dash').setLevel(logging.WARNING) # Import core components from core.config import get_config from core.data_provider import DataProvider from core.orchestrator import TradingOrchestrator from core.trading_executor import TradingExecutor # Import layout and component managers from web.layout_manager import DashboardLayoutManager from web.component_manager import DashboardComponentManager # Enhanced RL components are no longer available - using Basic orchestrator only ENHANCED_RL_AVAILABLE = False try: from core.cob_integration import COBIntegration from core.multi_exchange_cob_provider import COBSnapshot COB_INTEGRATION_AVAILABLE = True except ImportError: COB_INTEGRATION_AVAILABLE = False logger.warning("COB integration not available") # Add Universal Data Stream imports try: from core.unified_data_stream import UnifiedDataStream from core.universal_data_adapter import UniversalDataAdapter, UniversalDataStream as UDS UNIFIED_STREAM_AVAILABLE = True except ImportError: UNIFIED_STREAM_AVAILABLE = False logger.warning("Unified Data Stream not available") # Import RL COB trader for 1B parameter model integration from core.realtime_rl_cob_trader import RealtimeRLCOBTrader, PredictionResult # Using Basic orchestrator only - Enhanced orchestrator removed for stability ENHANCED_ORCHESTRATOR_AVAILABLE = False USE_ENHANCED_ORCHESTRATOR = False class CleanTradingDashboard: """Clean, modular trading dashboard implementation""" def __init__(self, data_provider: Optional[DataProvider] = None, orchestrator: Optional[Any] = None, trading_executor: Optional[TradingExecutor] = None): self.config = get_config() # Initialize components self.data_provider = data_provider or DataProvider() self.trading_executor = trading_executor or TradingExecutor() # Initialize orchestrator - USING BASIC ORCHESTRATOR ONLY if orchestrator is None: self.orchestrator = TradingOrchestrator(self.data_provider) logger.info("Using Basic Trading Orchestrator for stability") else: self.orchestrator = orchestrator # Initialize layout and component managers self.layout_manager = DashboardLayoutManager( starting_balance=self._get_initial_balance(), trading_executor=self.trading_executor ) self.component_manager = DashboardComponentManager() # Initialize Universal Data Stream for the 5 timeseries architecture if UNIFIED_STREAM_AVAILABLE: self.unified_stream = UnifiedDataStream(self.data_provider, self.orchestrator) self.stream_consumer_id = self.unified_stream.register_consumer( consumer_name="CleanTradingDashboard", callback=self._handle_unified_stream_data, data_types=['ticks', 'ohlcv', 'training_data', 'ui_data'] ) logger.info(f"Universal Data Stream initialized with consumer ID: {self.stream_consumer_id}") logger.info("Subscribed to Universal 5 Timeseries: ETH(ticks,1m,1h,1d) + BTC(ticks)") else: self.unified_stream = None self.stream_consumer_id = None logger.warning("Universal Data Stream not available - fallback to direct data access") # Dashboard state self.recent_decisions = [] self.closed_trades = [] self.current_prices = {} self.session_pnl = 0.0 self.total_fees = 0.0 self.current_position = None # WebSocket streaming self.ws_price_cache = {} self.is_streaming = False self.tick_cache = [] # COB data cache - using same approach as cob_realtime_dashboard.py self.cob_cache = { 'ETH/USDT': {'last_update': 0, 'data': None, 'updates_count': 0}, 'BTC/USDT': {'last_update': 0, 'data': None, 'updates_count': 0} } self.latest_cob_data = {} # Cache for COB integration data # Initialize timezone timezone_name = self.config.get('system', {}).get('timezone', 'Europe/Sofia') self.timezone = pytz.timezone(timezone_name) # Create Dash app self.app = Dash(__name__, external_stylesheets=[ 'https://cdn.jsdelivr.net/npm/bootstrap@5.1.3/dist/css/bootstrap.min.css', 'https://cdnjs.cloudflare.com/ajax/libs/font-awesome/6.0.0/css/all.min.css' ]) # Suppress Dash development mode logging self.app.enable_dev_tools(debug=False, dev_tools_silence_routes_logging=True) # Setup layout and callbacks self._setup_layout() self._setup_callbacks() # Start data streams self._initialize_streaming() # Connect to orchestrator for real trading signals self._connect_to_orchestrator() # Initialize REAL COB integration - using proper approach from enhanced orchestrator self._initialize_cob_integration_proper() # Start Universal Data Stream if self.unified_stream: import threading threading.Thread(target=self._start_unified_stream, daemon=True).start() logger.info("Universal Data Stream starting...") # Start signal generation loop to ensure continuous trading signals self._start_signal_generation_loop() logger.info("Clean Trading Dashboard initialized with PROPER COB integration and signal generation") def load_model_dynamically(self, model_name: str, model_type: str, model_path: Optional[str] = None) -> bool: """Dynamically load a model at runtime - Not implemented in orchestrator""" logger.warning("Dynamic model loading not implemented in orchestrator") return False def unload_model_dynamically(self, model_name: str) -> bool: """Dynamically unload a model at runtime - Not implemented in orchestrator""" logger.warning("Dynamic model unloading not implemented in orchestrator") return False def get_loaded_models_status(self) -> Dict[str, Any]: """Get status of all loaded models from training metrics""" try: # Get status from training metrics instead metrics = self._get_training_metrics() return { 'loaded_models': metrics.get('loaded_models', {}), 'total_models': len(metrics.get('loaded_models', {})), 'system_status': 'ACTIVE' if metrics.get('training_status', {}).get('active_sessions', 0) > 0 else 'INACTIVE' } except Exception as e: logger.error(f"Error getting model status: {e}") return {'loaded_models': {}, 'total_models': 0, 'system_status': 'ERROR'} def _get_initial_balance(self) -> float: """Get initial balance from trading executor or default""" try: if self.trading_executor and hasattr(self.trading_executor, 'starting_balance'): balance = getattr(self.trading_executor, 'starting_balance', None) if balance and balance > 0: return balance except Exception as e: logger.warning(f"Error getting balance: {e}") return 100.0 # Default balance def _setup_layout(self): """Setup the dashboard layout using layout manager""" self.app.layout = self.layout_manager.create_main_layout() def _setup_callbacks(self): """Setup dashboard callbacks""" @self.app.callback( [Output('current-price', 'children'), Output('session-pnl', 'children'), Output('current-position', 'children'), Output('portfolio-value', 'children'), Output('total-fees', 'children'), Output('trade-count', 'children'), Output('mexc-status', 'children')], [Input('interval-component', 'n_intervals')] ) def update_metrics(n): """Update key metrics""" try: # Get current price current_price = self._get_current_price('ETH/USDT') price_str = f"${current_price:.2f}" if current_price else "Loading..." # Calculate session P&L session_pnl_str = f"${self.session_pnl:.2f}" session_pnl_class = "text-success" if self.session_pnl >= 0 else "text-danger" # Current position position_str = "No Position" if self.current_position: side = self.current_position.get('side', 'UNKNOWN') size = self.current_position.get('size', 0) entry_price = self.current_position.get('price', 0) position_str = f"{side} {size:.3f} @ ${entry_price:.2f}" # Portfolio value initial_balance = self._get_initial_balance() portfolio_value = initial_balance + self.session_pnl portfolio_str = f"${portfolio_value:.2f}" # Total fees fees_str = f"${self.total_fees:.3f}" # Trade count trade_count = len(self.closed_trades) trade_str = f"{trade_count} Trades" # MEXC status mexc_status = "SIM" if self.trading_executor: if hasattr(self.trading_executor, 'trading_enabled') and self.trading_executor.trading_enabled: if hasattr(self.trading_executor, 'simulation_mode') and not self.trading_executor.simulation_mode: mexc_status = "LIVE" return price_str, session_pnl_str, position_str, portfolio_str, fees_str, trade_str, mexc_status except Exception as e: logger.error(f"Error updating metrics: {e}") return "Error", "$0.00", "Error", "$100.00", "$0.00", "0", "ERROR" @self.app.callback( Output('recent-decisions', 'children'), [Input('interval-component', 'n_intervals')] ) def update_recent_decisions(n): """Update recent trading signals""" try: return self.component_manager.format_trading_signals(self.recent_decisions) except Exception as e: logger.error(f"Error updating decisions: {e}") return [html.P(f"Error: {str(e)}", className="text-danger")] @self.app.callback( Output('price-chart', 'figure'), [Input('interval-component', 'n_intervals')] ) def update_price_chart(n): """Update price chart every second (1000ms interval)""" try: return self._create_price_chart('ETH/USDT') except Exception as e: logger.error(f"Error updating chart: {e}") return go.Figure().add_annotation(text=f"Chart Error: {str(e)}", xref="paper", yref="paper", x=0.5, y=0.5, showarrow=False) @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.component_manager.format_closed_trades_table(self.closed_trades) except Exception as e: logger.error(f"Error updating trades table: {e}") return html.P(f"Error: {str(e)}", className="text-danger") @self.app.callback( [Output('cob-status-content', 'children'), Output('eth-cob-content', 'children'), Output('btc-cob-content', 'children')], [Input('interval-component', 'n_intervals')] ) def update_cob_data(n): """Update COB data displays""" try: # COB Status cob_status = self._get_cob_status() status_components = self.component_manager.format_system_status(cob_status) # ETH/USDT COB eth_cob = self._get_cob_snapshot('ETH/USDT') eth_components = self.component_manager.format_cob_data(eth_cob, 'ETH/USDT') # BTC/USDT COB btc_cob = self._get_cob_snapshot('BTC/USDT') btc_components = self.component_manager.format_cob_data(btc_cob, 'BTC/USDT') return status_components, eth_components, btc_components except Exception as e: logger.error(f"Error updating COB data: {e}") error_msg = html.P(f"Error: {str(e)}", className="text-danger") return error_msg, error_msg, error_msg @self.app.callback( Output('training-metrics', 'children'), [Input('interval-component', 'n_intervals')] ) def update_training_metrics(n): """Update training metrics""" try: metrics_data = self._get_training_metrics() return self.component_manager.format_training_metrics(metrics_data) except Exception as e: logger.error(f"Error updating training metrics: {e}") return [html.P(f"Error: {str(e)}", className="text-danger")] # Manual trading buttons @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 [html.I(className="fas fa-arrow-up me-1"), "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 [html.I(className="fas fa-arrow-down me-1"), "SELL"] # Clear session button @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 [html.I(className="fas fa-trash me-1"), "Clear Session"] def _get_current_price(self, symbol: str) -> Optional[float]: """Get current price for symbol""" try: # Try WebSocket cache first ws_symbol = symbol.replace('/', '') if ws_symbol in self.ws_price_cache: return self.ws_price_cache[ws_symbol] # Fallback to data provider if symbol in self.current_prices: return self.current_prices[symbol] # Get fresh price from data provider df = self.data_provider.get_historical_data(symbol, '1m', limit=1) if df is not None and not df.empty: price = float(df['close'].iloc[-1]) self.current_prices[symbol] = price return price except Exception as e: logger.warning(f"Error getting current price for {symbol}: {e}") return None def _create_price_chart(self, symbol: str) -> go.Figure: """Create 1-minute main chart with 1-second mini chart - Updated every second""" try: # FIXED: Always get fresh data on startup to avoid gaps # 1. Get historical 1-minute data as base (180 candles = 3 hours) - FORCE REFRESH on first load is_startup = not hasattr(self, '_chart_initialized') or not self._chart_initialized df_historical = self.data_provider.get_historical_data(symbol, '1m', limit=180, refresh=is_startup) # Mark chart as initialized to use cache on subsequent loads if is_startup: self._chart_initialized = True logger.info(f"[STARTUP] Fetched fresh {symbol} 1m data to avoid gaps") # 2. Get WebSocket 1s data and convert to 1m bars ws_data_raw = self._get_websocket_chart_data(symbol, 'raw') df_live = None if ws_data_raw is not None and len(ws_data_raw) > 60: # Resample 1s data to 1m bars df_live = ws_data_raw.resample('1min').agg({ 'open': 'first', 'high': 'max', 'low': 'min', 'close': 'last', 'volume': 'sum' }).dropna() # 3. Merge historical + live data intelligently if df_historical is not None and not df_historical.empty: if df_live is not None and not df_live.empty: # Find overlap point - where live data starts live_start = df_live.index[0] # Keep historical data up to live data start df_historical_clean = df_historical[df_historical.index < live_start] # Combine: historical (older) + live (newer) df_main = pd.concat([df_historical_clean, df_live]).tail(180) main_source = f"Historical + Live ({len(df_historical_clean)} + {len(df_live)} bars)" else: # No live data, use historical only df_main = df_historical main_source = "Historical 1m" elif df_live is not None and not df_live.empty: # No historical data, use live only df_main = df_live.tail(180) main_source = "Live 1m (WebSocket)" else: # No data at all df_main = None main_source = "No data" # Get 1-second data (mini chart) ws_data_1s = self._get_websocket_chart_data(symbol, '1s') if df_main is None or df_main.empty: return go.Figure().add_annotation(text="No data available", xref="paper", yref="paper", x=0.5, y=0.5, showarrow=False) # Create chart with 3 subplots: Main 1m chart, Mini 1s chart, Volume if ws_data_1s is not None and len(ws_data_1s) > 5: fig = make_subplots( rows=3, cols=1, shared_xaxes=False, # Make 1s chart independent from 1m chart vertical_spacing=0.08, subplot_titles=( f'{symbol} - {main_source} ({len(df_main)} bars)', f'1s Mini Chart - Independent Axis ({len(ws_data_1s)} bars)', 'Volume' ), row_heights=[0.5, 0.25, 0.25], specs=[[{"secondary_y": False}], [{"secondary_y": False}], [{"secondary_y": False}]] ) has_mini_chart = True else: fig = make_subplots( rows=2, cols=1, shared_xaxes=True, vertical_spacing=0.08, subplot_titles=(f'{symbol} - {main_source} ({len(df_main)} bars)', 'Volume'), row_heights=[0.7, 0.3] ) has_mini_chart = False # Main 1-minute candlestick chart fig.add_trace( go.Candlestick( x=df_main.index, open=df_main['open'], high=df_main['high'], low=df_main['low'], close=df_main['close'], name=f'{symbol} 1m', increasing_line_color='#26a69a', decreasing_line_color='#ef5350', increasing_fillcolor='#26a69a', decreasing_fillcolor='#ef5350' ), row=1, col=1 ) # ADD MODEL PREDICTIONS TO MAIN CHART self._add_model_predictions_to_chart(fig, symbol, df_main, row=1) # ADD TRADES TO MAIN CHART self._add_trades_to_chart(fig, symbol, df_main, row=1) # Mini 1-second chart (if available) if has_mini_chart and ws_data_1s is not None: fig.add_trace( go.Scatter( x=ws_data_1s.index, y=ws_data_1s['close'], mode='lines', name='1s Price', line=dict(color='#ffa726', width=1), showlegend=False ), row=2, col=1 ) # ADD ALL SIGNALS TO 1S MINI CHART self._add_signals_to_mini_chart(fig, symbol, ws_data_1s, row=2) # Volume bars (bottom subplot) volume_row = 3 if has_mini_chart else 2 fig.add_trace( go.Bar( x=df_main.index, y=df_main['volume'], name='Volume', marker_color='rgba(100,150,200,0.6)', showlegend=False ), row=volume_row, col=1 ) # Update layout chart_height = 500 if has_mini_chart else 400 fig.update_layout( title=f'{symbol} Live Chart - {main_source} (Updated Every Second)', template='plotly_dark', showlegend=True, # Show legend for model predictions height=chart_height, margin=dict(l=50, r=50, t=60, b=50), xaxis_rangeslider_visible=False ) # Update axes with specific configurations for independent charts if has_mini_chart: # Main 1m chart (row 1) fig.update_xaxes(title_text="Time (1m intervals)", showgrid=True, gridwidth=1, gridcolor='rgba(128,128,128,0.2)', row=1, col=1) fig.update_yaxes(title_text="Price (USD)", showgrid=True, gridwidth=1, gridcolor='rgba(128,128,128,0.2)', row=1, col=1) # Independent 1s chart (row 2) - can zoom/pan separately fig.update_xaxes(title_text="Time (1s ticks)", showgrid=True, gridwidth=1, gridcolor='rgba(128,128,128,0.2)', row=2, col=1) fig.update_yaxes(title_text="Price (USD)", showgrid=True, gridwidth=1, gridcolor='rgba(128,128,128,0.2)', row=2, col=1) # Volume chart (row 3) fig.update_xaxes(title_text="Time", showgrid=True, gridwidth=1, gridcolor='rgba(128,128,128,0.2)', row=3, col=1) fig.update_yaxes(title_text="Volume", showgrid=True, gridwidth=1, gridcolor='rgba(128,128,128,0.2)', row=3, col=1) else: # Main chart only fig.update_xaxes(showgrid=True, gridwidth=1, gridcolor='rgba(128,128,128,0.2)') fig.update_yaxes(showgrid=True, gridwidth=1, gridcolor='rgba(128,128,128,0.2)') chart_info = f"1m bars: {len(df_main)}" if has_mini_chart: chart_info += f", 1s ticks: {len(ws_data_1s)}" logger.debug(f"[CHART] Created combined chart - {chart_info}") return fig except Exception as e: logger.error(f"Error creating chart for {symbol}: {e}") return go.Figure().add_annotation(text=f"Chart Error: {str(e)}", xref="paper", yref="paper", x=0.5, y=0.5, showarrow=False) def _add_model_predictions_to_chart(self, fig: go.Figure, symbol: str, df_main: pd.DataFrame, row: int = 1): """Add model predictions to the chart - ONLY EXECUTED TRADES on main chart""" try: # Only show EXECUTED TRADES on the main 1m chart executed_signals = [signal for signal in self.recent_decisions if signal.get('executed', False)] if executed_signals: # Separate by prediction type buy_trades = [] sell_trades = [] for signal in executed_signals[-20:]: # Last 20 executed trades signal_time = signal.get('timestamp') signal_price = signal.get('price', 0) signal_action = signal.get('action', 'HOLD') signal_confidence = signal.get('confidence', 0) if signal_time and signal_price and signal_confidence > 0: # Convert timestamp if needed if isinstance(signal_time, str): try: # Handle time-only format if ':' in signal_time and len(signal_time.split(':')) == 3: signal_time = datetime.now().replace( hour=int(signal_time.split(':')[0]), minute=int(signal_time.split(':')[1]), second=int(signal_time.split(':')[2]), microsecond=0 ) else: signal_time = pd.to_datetime(signal_time) except: continue if signal_action == 'BUY': buy_trades.append({'x': signal_time, 'y': signal_price, 'confidence': signal_confidence}) elif signal_action == 'SELL': sell_trades.append({'x': signal_time, 'y': signal_price, 'confidence': signal_confidence}) # Add EXECUTED BUY trades (large green circles) if buy_trades: fig.add_trace( go.Scatter( x=[t['x'] for t in buy_trades], y=[t['y'] for t in buy_trades], mode='markers', marker=dict( symbol='circle', size=15, color='rgba(0, 255, 100, 0.9)', line=dict(width=3, color='green') ), name='EXECUTED BUY', showlegend=True, hovertemplate="EXECUTED BUY TRADE
" + "Price: $%{y:.2f}
" + "Time: %{x}
" + "Confidence: %{customdata:.1%}", customdata=[t['confidence'] for t in buy_trades] ), row=row, col=1 ) # Add EXECUTED SELL trades (large red circles) if sell_trades: fig.add_trace( go.Scatter( x=[t['x'] for t in sell_trades], y=[t['y'] for t in sell_trades], mode='markers', marker=dict( symbol='circle', size=15, color='rgba(255, 100, 100, 0.9)', line=dict(width=3, color='red') ), name='EXECUTED SELL', showlegend=True, hovertemplate="EXECUTED SELL TRADE
" + "Price: $%{y:.2f}
" + "Time: %{x}
" + "Confidence: %{customdata:.1%}", customdata=[t['confidence'] for t in sell_trades] ), row=row, col=1 ) except Exception as e: logger.warning(f"Error adding executed trades to main chart: {e}") def _add_signals_to_mini_chart(self, fig: go.Figure, symbol: str, ws_data_1s: pd.DataFrame, row: int = 2): """Add ALL signals (executed and non-executed) to the 1s mini chart""" try: if not self.recent_decisions: return # Show ALL signals on the mini chart all_signals = self.recent_decisions[-50:] # Last 50 signals buy_signals = [] sell_signals = [] for signal in all_signals: signal_time = signal.get('timestamp') signal_price = signal.get('price', 0) signal_action = signal.get('action', 'HOLD') signal_confidence = signal.get('confidence', 0) is_executed = signal.get('executed', False) if signal_time and signal_price and signal_confidence > 0: # Convert timestamp if needed if isinstance(signal_time, str): try: # Handle time-only format if ':' in signal_time and len(signal_time.split(':')) == 3: signal_time = datetime.now().replace( hour=int(signal_time.split(':')[0]), minute=int(signal_time.split(':')[1]), second=int(signal_time.split(':')[2]), microsecond=0 ) else: signal_time = pd.to_datetime(signal_time) except: continue signal_data = { 'x': signal_time, 'y': signal_price, 'confidence': signal_confidence, 'executed': is_executed } if signal_action == 'BUY': buy_signals.append(signal_data) elif signal_action == 'SELL': sell_signals.append(signal_data) # Add ALL BUY signals to mini chart if buy_signals: # Split into executed and non-executed executed_buys = [s for s in buy_signals if s['executed']] pending_buys = [s for s in buy_signals if not s['executed']] # Executed buy signals (solid green triangles) if executed_buys: fig.add_trace( go.Scatter( x=[s['x'] for s in executed_buys], y=[s['y'] for s in executed_buys], mode='markers', marker=dict( symbol='triangle-up', size=10, color='rgba(0, 255, 100, 1.0)', line=dict(width=2, color='green') ), name='BUY (Executed)', showlegend=False, hovertemplate="BUY EXECUTED
" + "Price: $%{y:.2f}
" + "Time: %{x}
" + "Confidence: %{customdata:.1%}", customdata=[s['confidence'] for s in executed_buys] ), row=row, col=1 ) # Pending/non-executed buy signals (hollow green triangles) if pending_buys: fig.add_trace( go.Scatter( x=[s['x'] for s in pending_buys], y=[s['y'] for s in pending_buys], mode='markers', marker=dict( symbol='triangle-up', size=8, color='rgba(0, 255, 100, 0.5)', line=dict(width=2, color='green') ), name='📊 BUY (Signal)', showlegend=False, hovertemplate="📊 BUY SIGNAL
" + "Price: $%{y:.2f}
" + "Time: %{x}
" + "Confidence: %{customdata:.1%}", customdata=[s['confidence'] for s in pending_buys] ), row=row, col=1 ) # Add ALL SELL signals to mini chart if sell_signals: # Split into executed and non-executed executed_sells = [s for s in sell_signals if s['executed']] pending_sells = [s for s in sell_signals if not s['executed']] # Executed sell signals (solid red triangles) if executed_sells: fig.add_trace( go.Scatter( x=[s['x'] for s in executed_sells], y=[s['y'] for s in executed_sells], mode='markers', marker=dict( symbol='triangle-down', size=10, color='rgba(255, 100, 100, 1.0)', line=dict(width=2, color='red') ), name='SELL (Executed)', showlegend=False, hovertemplate="SELL EXECUTED
" + "Price: $%{y:.2f}
" + "Time: %{x}
" + "Confidence: %{customdata:.1%}", customdata=[s['confidence'] for s in executed_sells] ), row=row, col=1 ) # Pending/non-executed sell signals (hollow red triangles) if pending_sells: fig.add_trace( go.Scatter( x=[s['x'] for s in pending_sells], y=[s['y'] for s in pending_sells], mode='markers', marker=dict( symbol='triangle-down', size=8, color='rgba(255, 100, 100, 0.5)', line=dict(width=2, color='red') ), name='📊 SELL (Signal)', showlegend=False, hovertemplate="📊 SELL SIGNAL
" + "Price: $%{y:.2f}
" + "Time: %{x}
" + "Confidence: %{customdata:.1%}", customdata=[s['confidence'] for s in pending_sells] ), row=row, col=1 ) except Exception as e: logger.warning(f"Error adding signals to mini chart: {e}") def _add_trades_to_chart(self, fig: go.Figure, symbol: str, df_main: pd.DataFrame, row: int = 1): """Add executed trades to the chart""" try: if not self.closed_trades: return buy_trades = [] sell_trades = [] for trade in self.closed_trades[-20:]: # Last 20 trades entry_time = trade.get('entry_time') side = trade.get('side', 'UNKNOWN') entry_price = trade.get('entry_price', 0) pnl = trade.get('pnl', 0) if entry_time and entry_price: trade_data = {'x': entry_time, 'y': entry_price, 'pnl': pnl} if side == 'BUY': buy_trades.append(trade_data) elif side == 'SELL': sell_trades.append(trade_data) # Add BUY trades (green circles) if buy_trades: fig.add_trace( go.Scatter( x=[t['x'] for t in buy_trades], y=[t['y'] for t in buy_trades], mode='markers', marker=dict( symbol='circle', size=8, color='rgba(0, 255, 0, 0.7)', line=dict(width=2, color='green') ), name='BUY Trades', showlegend=True, hovertemplate="BUY Trade Executed
" + "Price: $%{y:.2f}
" + "Time: %{x}
" + "P&L: $%{customdata:.2f}", customdata=[t['pnl'] for t in buy_trades] ), row=row, col=1 ) # Add SELL trades (red circles) if sell_trades: fig.add_trace( go.Scatter( x=[t['x'] for t in sell_trades], y=[t['y'] for t in sell_trades], mode='markers', marker=dict( symbol='circle', size=8, color='rgba(255, 0, 0, 0.7)', line=dict(width=2, color='red') ), name='SELL Trades', showlegend=True, hovertemplate="SELL Trade Executed
" + "Price: $%{y:.2f}
" + "Time: %{x}
" + "P&L: $%{customdata:.2f}", customdata=[t['pnl'] for t in sell_trades] ), row=row, col=1 ) except Exception as e: logger.warning(f"Error adding trades to chart: {e}") def _get_price_at_time(self, df: pd.DataFrame, timestamp) -> Optional[float]: """Get price from dataframe at specific timestamp""" try: if isinstance(timestamp, str): timestamp = pd.to_datetime(timestamp) # Find closest timestamp in dataframe closest_idx = df.index.get_indexer([timestamp], method='nearest')[0] if closest_idx >= 0 and closest_idx < len(df): return float(df.iloc[closest_idx]['close']) return None except Exception: return None def _get_websocket_chart_data(self, symbol: str, timeframe: str = '1m') -> Optional[pd.DataFrame]: """Get WebSocket chart data - supports both 1m and 1s timeframes""" try: if not hasattr(self, 'tick_cache') or not self.tick_cache: return None # Filter ticks for symbol symbol_ticks = [tick for tick in self.tick_cache if tick.get('symbol') == symbol.replace('/', '')] if len(symbol_ticks) < 10: return None # Convert to DataFrame df = pd.DataFrame(symbol_ticks) df['datetime'] = pd.to_datetime(df['datetime']) df.set_index('datetime', inplace=True) # Get the price column (could be 'price', 'close', or 'c') price_col = None for col in ['price', 'close', 'c']: if col in df.columns: price_col = col break if price_col is None: logger.warning(f"No price column found in WebSocket data for {symbol}") return None # Create OHLC bars based on requested timeframe if timeframe == '1s': df_resampled = df[price_col].resample('1s').ohlc() # For 1s data, keep last 300 seconds (5 minutes) max_bars = 300 elif timeframe == 'raw': # Return raw 1s kline data for resampling to 1m in chart creation df_resampled = df[['open', 'high', 'low', 'close', 'volume']].copy() # Keep last 3+ hours of 1s data for 1m resampling max_bars = 200 * 60 # 200 minutes worth of 1s data else: # 1m df_resampled = df[price_col].resample('1min').ohlc() # For 1m data, keep last 180 minutes (3 hours) max_bars = 180 if timeframe == '1s': df_resampled.columns = ['open', 'high', 'low', 'close'] # Handle volume data if timeframe == '1s': # FIXED: Better volume calculation for 1s if 'volume' in df.columns and df['volume'].sum() > 0: df_resampled['volume'] = df['volume'].resample('1s').sum() else: # Use tick count as volume proxy with some randomization for variety import random tick_counts = df[price_col].resample('1s').count() df_resampled['volume'] = tick_counts * (50 + random.randint(0, 100)) # For 1m timeframe, volume is already in the raw data # Remove any NaN rows and limit to max bars df_resampled = df_resampled.dropna().tail(max_bars) if len(df_resampled) < 5: logger.debug(f"Insufficient {timeframe} data for {symbol}: {len(df_resampled)} bars") return None logger.debug(f"[WS-CHART] Created {len(df_resampled)} {timeframe} OHLC bars for {symbol}") return df_resampled except Exception as e: logger.warning(f"Error getting WebSocket chart data: {e}") return None def _get_cob_status(self) -> Dict: """Get REAL COB integration status - FIXED TO USE ENHANCED ORCHESTRATOR PROPERLY""" try: status = { 'trading_enabled': bool(self.trading_executor and getattr(self.trading_executor, 'trading_enabled', False)), 'simulation_mode': bool(self.trading_executor and getattr(self.trading_executor, 'simulation_mode', True)), 'data_provider_status': 'Active', 'websocket_status': 'Connected' if self.is_streaming else 'Disconnected', 'cob_status': 'No COB Integration', # Default 'orchestrator_type': 'Basic', 'rl_model_status': 'Inactive', 'predictions_count': 0, 'cache_size': 0 } # Check if we have Enhanced Orchestrator - PROPER TYPE CHECK is_enhanced = (ENHANCED_ORCHESTRATOR_AVAILABLE and self.orchestrator.__class__.__name__ == 'EnhancedTradingOrchestrator') if is_enhanced: status['orchestrator_type'] = 'Enhanced' # Check COB integration in Enhanced orchestrator if hasattr(self.orchestrator, 'cob_integration'): cob_integration = getattr(self.orchestrator, 'cob_integration', None) if cob_integration is not None: # Get real COB integration statistics try: if hasattr(cob_integration, 'get_statistics'): cob_stats = cob_integration.get_statistics() if cob_stats: active_symbols = cob_stats.get('active_symbols', []) total_updates = cob_stats.get('total_updates', 0) provider_status = cob_stats.get('provider_status', 'Unknown') if active_symbols: status['cob_status'] = f'Enhanced COB Active ({len(active_symbols)} symbols)' status['active_symbols'] = active_symbols status['cache_size'] = total_updates status['provider_status'] = provider_status else: status['cob_status'] = 'Enhanced COB Integration Loaded (No Data)' else: status['cob_status'] = 'Enhanced COB Integration (Stats Unavailable)' else: status['cob_status'] = 'Enhanced COB Integration (No Stats Method)' except Exception as e: logger.debug(f"Error getting COB statistics: {e}") status['cob_status'] = 'Enhanced COB Integration (Error Getting Stats)' else: status['cob_status'] = 'Enhanced Orchestrator (COB Integration Not Initialized)' # Don't log warning here to avoid spam, just info level logger.debug("Enhanced orchestrator has COB integration attribute but it's None") else: status['cob_status'] = 'Enhanced Orchestrator Missing COB Integration' logger.debug("Enhanced orchestrator available but has no COB integration attribute") else: status['cob_status'] = 'Enhanced Orchestrator Missing COB Integration' logger.debug("Enhanced orchestrator available but has no COB integration attribute") else: if not ENHANCED_ORCHESTRATOR_AVAILABLE: status['cob_status'] = 'Enhanced Orchestrator Not Available' status['orchestrator_type'] = 'Basic (Enhanced Unavailable)' else: status['cob_status'] = 'Basic Orchestrator (No COB Support)' status['orchestrator_type'] = 'Basic (Enhanced Not Used)' return status except Exception as e: logger.error(f"Error getting COB status: {e}") return {'error': str(e), 'cob_status': 'Error Getting Status', 'orchestrator_type': 'Unknown'} def _get_cob_snapshot(self, symbol: str) -> Optional[Any]: """Get COB snapshot for symbol using enhanced orchestrator approach""" try: # Get from Enhanced Orchestrator's COB integration (proper way) if (ENHANCED_ORCHESTRATOR_AVAILABLE and hasattr(self.orchestrator, 'cob_integration') and self.orchestrator.__class__.__name__ == 'EnhancedTradingOrchestrator'): cob_integration = getattr(self.orchestrator, 'cob_integration', None) if cob_integration is not None: # Get real COB snapshot using the proper method if hasattr(cob_integration, 'get_cob_snapshot'): snapshot = cob_integration.get_cob_snapshot(symbol) if snapshot: logger.debug(f"Retrieved Enhanced COB snapshot for {symbol}") return snapshot else: logger.debug(f"No Enhanced COB data available for {symbol}") elif hasattr(cob_integration, 'get_consolidated_orderbook'): # Alternative method name snapshot = cob_integration.get_consolidated_orderbook(symbol) if snapshot: logger.debug(f"Retrieved Enhanced COB orderbook for {symbol}") return snapshot else: logger.warning("Enhanced COB integration has no recognized snapshot method") else: logger.debug(f"No Enhanced COB integration available for {symbol}") return None except Exception as e: logger.warning(f"Error getting Enhanced COB snapshot for {symbol}: {e}") return None def _get_training_metrics(self) -> Dict: """Get training metrics data - HANDLES BOTH ENHANCED AND BASIC ORCHESTRATORS""" try: metrics = {} # Loaded Models Section - FIXED loaded_models = {} # 1. DQN Model Status and Loss Tracking - FIXED ATTRIBUTE ACCESS dqn_active = False dqn_last_loss = 0.0 dqn_prediction_count = 0 last_action = 'NONE' last_confidence = 0.0 # Using Basic orchestrator only - Enhanced orchestrator removed is_enhanced = False # Basic orchestrator doesn't have DQN agent - create default status try: # Check if Basic orchestrator has any DQN features if hasattr(self.orchestrator, 'some_basic_dqn_method'): dqn_active = True # Get basic stats if available else: dqn_active = False logger.debug("Basic orchestrator - no DQN features available") except Exception as e: logger.debug(f"Error checking Basic orchestrator DQN: {e}") dqn_active = False dqn_model_info = { 'active': dqn_active, 'parameters': 5000000, # ~5M params for DQN 'last_prediction': { 'timestamp': datetime.now().strftime('%H:%M:%S'), 'action': last_action, 'confidence': last_confidence }, 'loss_5ma': dqn_last_loss, # Real loss from training 'model_type': 'DQN', 'description': 'Deep Q-Network Agent' + (' (Enhanced)' if is_enhanced else ' (Basic)'), 'prediction_count': dqn_prediction_count, 'epsilon': 1.0 # Default epsilon for Basic orchestrator } loaded_models['dqn'] = dqn_model_info # 2. CNN Model Status - NOT AVAILABLE IN BASIC ORCHESTRATOR cnn_active = False cnn_last_loss = 0.0234 # Default loss value cnn_model_info = { 'active': cnn_active, 'parameters': 50000000, # ~50M params 'last_prediction': { 'timestamp': datetime.now().strftime('%H:%M:%S'), 'action': 'MONITORING' if cnn_active else 'INACTIVE', 'confidence': 0.0 }, 'loss_5ma': cnn_last_loss, 'model_type': 'CNN', 'description': 'Williams Market Structure CNN' + (' (Enhanced Only)' if not is_enhanced else '') } loaded_models['cnn'] = cnn_model_info # 3. COB RL Model Status - NOT AVAILABLE IN BASIC ORCHESTRATOR cob_active = False cob_last_loss = 0.012 # Default loss value cob_predictions_count = 0 cob_model_info = { 'active': cob_active, 'parameters': 400000000, # 400M optimized (Enhanced COB integration) 'last_prediction': { 'timestamp': datetime.now().strftime('%H:%M:%S'), 'action': 'ENHANCED_COB_INFERENCE' if cob_active else ('INACTIVE' if is_enhanced else 'NOT_AVAILABLE'), 'confidence': 0.0 }, 'loss_5ma': cob_last_loss, 'model_type': 'ENHANCED_COB_RL', 'description': 'Enhanced COB Integration' + (' (Enhanced Only)' if not is_enhanced else ''), 'predictions_count': cob_predictions_count } loaded_models['cob_rl'] = cob_model_info # Add loaded models to metrics metrics['loaded_models'] = loaded_models # Enhanced training status with signal generation signal_generation_active = self._is_signal_generation_active() metrics['training_status'] = { 'active_sessions': len([m for m in loaded_models.values() if m['active']]), 'signal_generation': 'ACTIVE' if signal_generation_active else 'INACTIVE', 'last_update': datetime.now().strftime('%H:%M:%S'), 'models_loaded': len(loaded_models), 'total_parameters': sum(m['parameters'] for m in loaded_models.values() if m['active']), 'orchestrator_type': 'Enhanced' if is_enhanced else 'Basic' } # COB $1 Buckets (sample data for now) metrics['cob_buckets'] = self._get_cob_dollar_buckets() return metrics except Exception as e: logger.error(f"Error getting training metrics: {e}") return {'error': str(e), 'loaded_models': {}, 'training_status': {'active_sessions': 0}} def _is_signal_generation_active(self) -> bool: """Check if signal generation is currently active""" try: # Check if orchestrator has recent decisions if self.orchestrator and hasattr(self.orchestrator, 'recent_decisions'): for symbol, decisions in self.orchestrator.recent_decisions.items(): if decisions and len(decisions) > 0: # Check if last decision is recent (within 5 minutes) last_decision_time = decisions[-1].timestamp time_diff = (datetime.now() - last_decision_time).total_seconds() if time_diff < 300: # 5 minutes return True # Check if we have recent dashboard decisions if len(self.recent_decisions) > 0: last_decision = self.recent_decisions[-1] if 'timestamp' in last_decision: # Parse timestamp string to datetime try: if isinstance(last_decision['timestamp'], str): decision_time = datetime.strptime(last_decision['timestamp'], '%H:%M:%S') decision_time = decision_time.replace(year=datetime.now().year, month=datetime.now().month, day=datetime.now().day) else: decision_time = last_decision['timestamp'] time_diff = (datetime.now() - decision_time).total_seconds() if time_diff < 300: # 5 minutes return True except Exception: pass return False except Exception as e: logger.debug(f"Error checking signal generation status: {e}") return False def _start_signal_generation_loop(self): """Start continuous signal generation loop""" try: def signal_worker(): logger.info("Starting continuous signal generation loop") # Basic orchestrator doesn't have DQN - using momentum signals only logger.info("Using momentum-based signals (Basic orchestrator)") while True: try: # Generate signals for both symbols for symbol in ['ETH/USDT', 'BTC/USDT']: try: # Get current price current_price = self._get_current_price(symbol) if not current_price: continue # 1. Generate basic signal (Basic orchestrator doesn't have DQN) # Skip DQN signals - Basic orchestrator doesn't support them # 2. Generate simple momentum signal as backup momentum_signal = self._generate_momentum_signal(symbol, current_price) if momentum_signal: self._process_dashboard_signal(momentum_signal) except Exception as e: logger.debug(f"Error generating signal for {symbol}: {e}") # Wait 10 seconds before next cycle time.sleep(10) except Exception as e: logger.error(f"Error in signal generation cycle: {e}") time.sleep(30) # Start signal generation thread signal_thread = threading.Thread(target=signal_worker, daemon=True) signal_thread.start() logger.info("Signal generation loop started") except Exception as e: logger.error(f"Error starting signal generation loop: {e}") def _generate_dqn_signal(self, symbol: str, current_price: float) -> Optional[Dict]: """Generate trading signal using DQN agent - NOT AVAILABLE IN BASIC ORCHESTRATOR""" # Basic orchestrator doesn't have DQN features return None def _generate_momentum_signal(self, symbol: str, current_price: float) -> Optional[Dict]: """Generate simple momentum-based signal as backup""" try: # Get recent price data df = self.data_provider.get_historical_data(symbol, '1m', limit=10) if df is None or len(df) < 5: return None prices = df['close'].values # Calculate momentum short_momentum = (prices[-1] - prices[-3]) / prices[-3] # 3-period momentum medium_momentum = (prices[-1] - prices[-5]) / prices[-5] # 5-period momentum # Simple signal generation import random signal_prob = random.random() if short_momentum > 0.002 and medium_momentum > 0.001 and signal_prob > 0.7: action = 'BUY' confidence = min(0.8, 0.4 + abs(short_momentum) * 100) elif short_momentum < -0.002 and medium_momentum < -0.001 and signal_prob > 0.7: action = 'SELL' confidence = min(0.8, 0.4 + abs(short_momentum) * 100) elif signal_prob > 0.95: # Random signals for activity action = 'BUY' if signal_prob > 0.975 else 'SELL' confidence = 0.3 else: return None return { 'action': action, 'symbol': symbol, 'price': current_price, 'confidence': confidence, 'timestamp': datetime.now().strftime('%H:%M:%S'), 'size': 0.005, 'reason': f'Momentum signal (s={short_momentum:.4f}, m={medium_momentum:.4f})', 'model': 'Momentum' } except Exception as e: logger.debug(f"Error generating momentum signal for {symbol}: {e}") return None def _process_dashboard_signal(self, signal: Dict): """Process signal for dashboard display and training""" try: # Add signal to recent decisions signal['executed'] = False signal['blocked'] = False signal['manual'] = False self.recent_decisions.append(signal) # Keep only last 20 decisions for display if len(self.recent_decisions) > 20: self.recent_decisions = self.recent_decisions[-20:] # Log signal generation logger.info(f"Generated {signal['action']} signal for {signal['symbol']} " f"(conf: {signal['confidence']:.2f}, model: {signal.get('model', 'UNKNOWN')})") # DQN training not available in Basic orchestrator # Skip DQN training - Basic orchestrator doesn't support it except Exception as e: logger.error(f"Error processing dashboard signal: {e}") def _train_dqn_on_signal(self, signal: Dict): """Train DQN agent on generated signal - NOT AVAILABLE IN BASIC ORCHESTRATOR""" # Basic orchestrator doesn't have DQN features return def _get_cob_dollar_buckets(self) -> List[Dict]: """Get COB $1 price buckets with volume data""" try: # This would normally come from the COB integration # For now, return sample data structure sample_buckets = [ {'price': 2000, 'total_volume': 150000, 'bid_pct': 45, 'ask_pct': 55}, {'price': 2001, 'total_volume': 120000, 'bid_pct': 52, 'ask_pct': 48}, {'price': 1999, 'total_volume': 98000, 'bid_pct': 38, 'ask_pct': 62}, {'price': 2002, 'total_volume': 87000, 'bid_pct': 60, 'ask_pct': 40}, {'price': 1998, 'total_volume': 76000, 'bid_pct': 35, 'ask_pct': 65} ] return sample_buckets except Exception as e: logger.debug(f"Error getting COB buckets: {e}") return [] def _execute_manual_trade(self, action: str): """Execute manual trading action - FIXED to properly execute and track trades""" try: if not self.trading_executor: logger.warning("No trading executor available") return symbol = 'ETH/USDT' current_price = self._get_current_price(symbol) if not current_price: logger.warning("No current price available for manual trade") return # CAPTURE ALL MODEL INPUTS FOR COLD START TRAINING using core TradeDataManager try: from core.trade_data_manager import TradeDataManager trade_data_manager = TradeDataManager() model_inputs = trade_data_manager.capture_comprehensive_model_inputs( symbol, action, current_price, self.orchestrator, self.data_provider ) except Exception as e: logger.warning(f"Failed to capture model inputs via TradeDataManager: {e}") model_inputs = {} # Create manual trading decision decision = { 'timestamp': datetime.now().strftime('%H:%M:%S'), 'action': action, 'confidence': 1.0, # Manual trades have 100% confidence 'price': current_price, 'symbol': symbol, 'size': 0.01, 'executed': False, 'blocked': False, 'manual': True, 'reason': f'Manual {action} button', 'model_inputs': model_inputs # Store for training } # Execute through trading executor try: result = self.trading_executor.execute_trade(symbol, action, 0.01) # Small size for testing if result: decision['executed'] = True logger.info(f"Manual {action} executed at ${current_price:.2f}") # Create a trade record for tracking WITH model inputs trade_record = { 'symbol': symbol, 'side': action, 'quantity': 0.01, 'entry_price': current_price, 'exit_price': current_price, 'entry_time': datetime.now(), 'exit_time': datetime.now(), 'pnl': 0.0, # Manual test trades have 0 P&L initially 'fees': 0.0, 'confidence': 1.0, 'trade_type': 'manual', 'model_inputs_at_entry': model_inputs, # CRITICAL: Store model inputs for training 'entry_market_state': model_inputs.get('market_state', {}), 'entry_features': model_inputs.get('features', {}), 'entry_predictions': model_inputs.get('predictions', {}), 'training_ready': True # Mark as ready for cold start training } # Add to closed trades for display self.closed_trades.append(trade_record) # Store for cold start training when trade closes using core TradeDataManager try: case_id = trade_data_manager.store_trade_for_training(trade_record) if case_id: logger.info(f"Trade stored for training with case ID: {case_id}") except Exception as e: logger.warning(f"Failed to store trade for training: {e}") # Update session metrics if action == 'BUY': self.session_pnl += 0.0 # No immediate P&L for entry else: # SELL # For demo purposes, simulate small positive P&L demo_pnl = 0.05 # $0.05 demo profit self.session_pnl += demo_pnl trade_record['pnl'] = demo_pnl # TRIGGER COLD START TRAINING on profitable demo trade using core TrainingIntegration try: from core.training_integration import TrainingIntegration training_integration = TrainingIntegration(self.orchestrator) training_success = training_integration.trigger_cold_start_training(trade_record, case_id) if training_success: logger.info("Cold start training completed successfully") else: logger.warning("Cold start training failed") except Exception as e: logger.warning(f"Failed to trigger cold start training: {e}") else: decision['executed'] = False decision['blocked'] = True decision['block_reason'] = "Trading executor returned False" logger.warning(f"Manual {action} failed - executor returned False") except Exception as e: decision['executed'] = False decision['blocked'] = True decision['block_reason'] = str(e) logger.error(f"Manual {action} failed with error: {e}") # Add to recent decisions for display self.recent_decisions.append(decision) # Keep only last 50 decisions if len(self.recent_decisions) > 50: self.recent_decisions = self.recent_decisions[-50:] except Exception as e: logger.error(f"Error executing manual {action}: {e}") # Model input capture moved to core.trade_data_manager.TradeDataManager def _get_comprehensive_market_state(self, symbol: str, current_price: float) -> Dict[str, float]: """Get comprehensive market state features""" try: market_state = {} # Price-based features market_state['current_price'] = current_price # Get historical data for features df = self.data_provider.get_historical_data(symbol, '1m', limit=100) if df is not None and not df.empty: prices = df['close'].values volumes = df['volume'].values # Price features market_state['price_sma_5'] = float(prices[-5:].mean()) market_state['price_sma_20'] = float(prices[-20:].mean()) market_state['price_std_20'] = float(prices[-20:].std()) market_state['price_rsi'] = self._calculate_rsi(prices, 14) # Volume features market_state['volume_current'] = float(volumes[-1]) market_state['volume_sma_20'] = float(volumes[-20:].mean()) market_state['volume_ratio'] = float(volumes[-1] / volumes[-20:].mean()) # Trend features market_state['price_momentum_5'] = float((prices[-1] - prices[-5]) / prices[-5]) market_state['price_momentum_20'] = float((prices[-1] - prices[-20]) / prices[-20]) # Add timestamp features now = datetime.now() market_state['hour_of_day'] = now.hour market_state['minute_of_hour'] = now.minute market_state['day_of_week'] = now.weekday() return market_state except Exception as e: logger.warning(f"Error getting market state: {e}") return {'current_price': current_price} def _calculate_rsi(self, prices, period=14): """Calculate RSI indicator""" try: deltas = np.diff(prices) gains = np.where(deltas > 0, deltas, 0) losses = np.where(deltas < 0, -deltas, 0) avg_gain = np.mean(gains[-period:]) avg_loss = np.mean(losses[-period:]) if avg_loss == 0: return 100.0 rs = avg_gain / avg_loss rsi = 100 - (100 / (1 + rs)) return float(rsi) except: return 50.0 # Neutral RSI def _get_cnn_features_and_predictions(self, symbol: str) -> Dict[str, Any]: """Get CNN features and predictions from orchestrator""" try: cnn_data = {} # Get CNN features if available if hasattr(self.orchestrator, 'latest_cnn_features'): cnn_features = getattr(self.orchestrator, 'latest_cnn_features', {}).get(symbol) if cnn_features is not None: cnn_data['features'] = cnn_features.tolist() if hasattr(cnn_features, 'tolist') else cnn_features # Get CNN predictions if available if hasattr(self.orchestrator, 'latest_cnn_predictions'): cnn_predictions = getattr(self.orchestrator, 'latest_cnn_predictions', {}).get(symbol) if cnn_predictions is not None: cnn_data['predictions'] = cnn_predictions.tolist() if hasattr(cnn_predictions, 'tolist') else cnn_predictions return cnn_data except Exception as e: logger.debug(f"Error getting CNN data: {e}") return {} def _get_dqn_state_features(self, symbol: str, current_price: float) -> Dict[str, Any]: """Get DQN state features from orchestrator""" try: # Get DQN state from orchestrator if available if hasattr(self.orchestrator, 'build_comprehensive_rl_state'): rl_state = self.orchestrator.build_comprehensive_rl_state(symbol) if rl_state is not None: return { 'state_vector': rl_state.tolist() if hasattr(rl_state, 'tolist') else rl_state, 'state_size': len(rl_state) if hasattr(rl_state, '__len__') else 0 } return {} except Exception as e: logger.debug(f"Error getting DQN state: {e}") return {} def _get_cob_features_for_training(self, symbol: str) -> Dict[str, Any]: """Get COB features for training""" try: cob_data = {} # Get COB features from orchestrator if hasattr(self.orchestrator, 'latest_cob_features'): cob_features = getattr(self.orchestrator, 'latest_cob_features', {}).get(symbol) if cob_features is not None: cob_data['features'] = cob_features.tolist() if hasattr(cob_features, 'tolist') else cob_features # Get COB snapshot cob_snapshot = self._get_cob_snapshot(symbol) if cob_snapshot: cob_data['snapshot_available'] = True cob_data['bid_levels'] = len(getattr(cob_snapshot, 'consolidated_bids', [])) cob_data['ask_levels'] = len(getattr(cob_snapshot, 'consolidated_asks', [])) else: cob_data['snapshot_available'] = False return cob_data except Exception as e: logger.debug(f"Error getting COB features: {e}") return {} def _get_technical_indicators(self, symbol: str) -> Dict[str, float]: """Get technical indicators""" try: indicators = {} # Get recent price data df = self.data_provider.get_historical_data(symbol, '1m', limit=50) if df is not None and not df.empty: closes = df['close'].values highs = df['high'].values lows = df['low'].values volumes = df['volume'].values # Moving averages indicators['sma_10'] = float(closes[-10:].mean()) indicators['sma_20'] = float(closes[-20:].mean()) # Bollinger Bands sma_20 = closes[-20:].mean() std_20 = closes[-20:].std() indicators['bb_upper'] = float(sma_20 + 2 * std_20) indicators['bb_lower'] = float(sma_20 - 2 * std_20) indicators['bb_position'] = float((closes[-1] - indicators['bb_lower']) / (indicators['bb_upper'] - indicators['bb_lower'])) # MACD ema_12 = closes[-12:].mean() # Simplified ema_26 = closes[-26:].mean() # Simplified indicators['macd'] = float(ema_12 - ema_26) # Volatility indicators['volatility'] = float(std_20 / sma_20) return indicators except Exception as e: logger.debug(f"Error calculating technical indicators: {e}") return {} def _get_recent_price_history(self, symbol: str, periods: int = 50) -> List[float]: """Get recent price history""" try: df = self.data_provider.get_historical_data(symbol, '1m', limit=periods) if df is not None and not df.empty: return df['close'].tolist() return [] except Exception as e: logger.debug(f"Error getting price history: {e}") return [] # Trade storage moved to core.trade_data_manager.TradeDataManager # Cold start training moved to core.training_integration.TrainingIntegration def _clear_session(self): """Clear session data""" try: # Reset session metrics self.session_pnl = 0.0 self.total_fees = 0.0 self.closed_trades = [] self.recent_decisions = [] logger.info("Session data cleared") except Exception as e: logger.error(f"Error clearing session: {e}") def _initialize_cob_integration_proper(self): """Initialize COB integration using Enhanced Orchestrator - PROPER APPROACH""" try: logger.info("Connecting to COB integration from Enhanced Orchestrator...") # Check if we have Enhanced Orchestrator if not ENHANCED_ORCHESTRATOR_AVAILABLE: logger.error("Enhanced Orchestrator not available - COB integration requires Enhanced Orchestrator") return # Check if Enhanced Orchestrator has COB integration if not hasattr(self.orchestrator, 'cob_integration'): logger.error("Enhanced Orchestrator has no cob_integration attribute") return if self.orchestrator.cob_integration is None: logger.warning("Enhanced Orchestrator COB integration is None - needs to be started") # Try to start the COB integration asynchronously def start_cob_async(): """Start COB integration in async context""" import asyncio async def _start_cob(): try: # Start the COB integration from enhanced orchestrator await self.orchestrator.start_cob_integration() logger.info("COB integration started successfully from Enhanced Orchestrator") # Register dashboard callback if possible if hasattr(self.orchestrator.cob_integration, 'add_dashboard_callback'): self.orchestrator.cob_integration.add_dashboard_callback(self._on_enhanced_cob_update) logger.info("Registered dashboard callback with Enhanced COB integration") except Exception as e: logger.error(f"Error starting COB integration from Enhanced Orchestrator: {e}") # Run in new event loop if needed try: loop = asyncio.get_event_loop() if loop.is_running(): # If loop is already running, schedule as task asyncio.create_task(_start_cob()) else: # If no loop running, run directly loop.run_until_complete(_start_cob()) except RuntimeError: # No event loop, create new one asyncio.run(_start_cob()) # Start COB integration in background thread to avoid blocking dashboard import threading cob_start_thread = threading.Thread(target=start_cob_async, daemon=True) cob_start_thread.start() logger.info("Enhanced COB integration startup initiated in background") else: # COB integration already exists, just register callback cob_integration = self.orchestrator.cob_integration logger.info(f"Enhanced COB integration found: {type(cob_integration)}") # Register callbacks if available if hasattr(cob_integration, 'add_dashboard_callback'): cob_integration.add_dashboard_callback(self._on_enhanced_cob_update) logger.info("Registered dashboard callback with existing Enhanced COB integration") # Verify COB integration is active and working if hasattr(cob_integration, 'get_statistics'): try: stats = cob_integration.get_statistics() logger.info(f"Enhanced COB statistics: {stats}") except Exception as e: logger.debug(f"Could not get COB statistics: {e}") logger.info("Enhanced COB integration connection completed") logger.info("NO SIMULATION - Using Enhanced Orchestrator real market data only") except Exception as e: logger.error(f"CRITICAL: Failed to connect to Enhanced COB integration: {e}") logger.error("Dashboard will operate without COB data") def _on_enhanced_cob_update(self, symbol: str, cob_data: Dict): """Handle Enhanced COB data updates - NO SIMULATION""" try: # Process Enhanced COB data update current_time = time.time() # Update cache with Enhanced COB data (same format as cob_realtime_dashboard.py) if symbol not in self.cob_cache: self.cob_cache[symbol] = {'last_update': 0, 'data': None, 'updates_count': 0} self.cob_cache[symbol] = { 'last_update': current_time, 'data': cob_data, 'updates_count': self.cob_cache[symbol].get('updates_count', 0) + 1 } # Also update latest_cob_data for compatibility self.latest_cob_data[symbol] = cob_data # Log Enhanced COB data updates update_count = self.cob_cache[symbol]['updates_count'] if update_count % 50 == 0: # Every 50 Enhanced updates logger.info(f"[ENHANCED-COB] {symbol} - Enhanced update #{update_count}") except Exception as e: logger.error(f"Error handling Enhanced COB update for {symbol}: {e}") def _start_cob_data_subscription(self): """Start COB data subscription with proper caching""" try: # Start the COB RL trader asynchronously import asyncio def start_cob_trader(): loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: loop.run_until_complete(self.cob_rl_trader.start()) logger.info("COB RL trader started successfully") except Exception as e: logger.error(f"Error in COB trader loop: {e}") finally: loop.close() # Start in separate thread to avoid blocking import threading cob_thread = threading.Thread(target=start_cob_trader, daemon=True) cob_thread.start() except Exception as e: logger.error(f"Error starting COB data subscription: {e}") def _on_cob_prediction(self, prediction: PredictionResult): """Handle COB RL predictions""" try: with self.cob_lock: # Convert prediction to dashboard format prediction_data = { 'timestamp': prediction.timestamp, 'direction': prediction.predicted_direction, # 0=DOWN, 1=SIDEWAYS, 2=UP 'confidence': prediction.confidence, 'predicted_change': prediction.predicted_change, 'direction_text': ['DOWN', 'SIDEWAYS', 'UP'][prediction.predicted_direction], 'color': ['red', 'gray', 'green'][prediction.predicted_direction] } # Add to predictions cache self.cob_predictions[prediction.symbol].append(prediction_data) # Cache COB data (1s buckets for 1 day max, 5 min retention) current_time = datetime.now() cob_data = { 'timestamp': current_time, 'prediction': prediction_data, 'features': prediction.features.tolist() if prediction.features is not None else [] } # Add to 1d cache (1s buckets) self.cob_data_cache_1d[prediction.symbol].append(cob_data) # Add to raw ticks cache (15 seconds max, 10+ updates/sec) self.cob_raw_ticks[prediction.symbol].append({ 'timestamp': current_time, 'prediction': prediction_data, 'raw_features': prediction.features.tolist() if prediction.features is not None else [] }) logger.debug(f"COB prediction cached for {prediction.symbol}: " f"{prediction_data['direction_text']} (confidence: {prediction.confidence:.3f})") except Exception as e: logger.error(f"Error handling COB prediction: {e}") def _connect_to_orchestrator(self): """Connect to orchestrator for real trading signals""" try: if self.orchestrator and hasattr(self.orchestrator, 'add_decision_callback'): # Register callback to receive trading decisions self.orchestrator.add_decision_callback(self._on_trading_decision) logger.info("Connected to orchestrator for trading signals") else: logger.warning("Orchestrator not available or doesn't support callbacks") except Exception as e: logger.error(f"Error connecting to orchestrator: {e}") def _on_trading_decision(self, decision): """Handle trading decision from orchestrator""" try: # Convert orchestrator decision to dashboard format # Handle both TradingDecision objects and dictionary formats if hasattr(decision, 'action'): # This is a TradingDecision object (dataclass) dashboard_decision = { 'timestamp': datetime.now().strftime('%H:%M:%S'), 'action': decision.action, 'confidence': decision.confidence, 'price': decision.price, 'executed': True, # Orchestrator decisions are executed 'blocked': False, 'manual': False } else: # This is a dictionary format dashboard_decision = { 'timestamp': datetime.now().strftime('%H:%M:%S'), 'action': decision.get('action', 'UNKNOWN'), 'confidence': decision.get('confidence', 0), 'price': decision.get('price', 0), 'executed': True, # Orchestrator decisions are executed 'blocked': False, 'manual': False } # Add to recent decisions self.recent_decisions.append(dashboard_decision) # Keep only last 50 decisions if len(self.recent_decisions) > 50: self.recent_decisions = self.recent_decisions[-50:] except Exception as e: logger.error(f"Error handling trading decision: {e}") def _initialize_streaming(self): """Initialize data streaming""" try: # Start WebSocket streaming self._start_websocket_streaming() # Start data collection thread self._start_data_collection() logger.info("Data streaming initialized") except Exception as e: logger.error(f"Error initializing streaming: {e}") def _start_websocket_streaming(self): """Start WebSocket streaming for real-time data - NO COB SIMULATION""" try: def ws_worker(): try: import websocket import json def on_message(ws, message): try: data = json.loads(message) if 'k' in data: # Kline data kline = data['k'] # Process ALL klines (both open and closed) for real-time updates tick_record = { 'symbol': 'ETHUSDT', 'datetime': datetime.fromtimestamp(int(kline['t']) / 1000), 'open': float(kline['o']), 'high': float(kline['h']), 'low': float(kline['l']), 'close': float(kline['c']), 'price': float(kline['c']), # For compatibility 'volume': float(kline['v']), # Real volume data! 'is_closed': kline['x'] # Track if kline is closed } # Update current price every second current_price = float(kline['c']) self.ws_price_cache['ETHUSDT'] = current_price self.current_prices['ETH/USDT'] = current_price # Add to tick cache (keep last 1000 klines for charts) # For real-time updates, we need more data points self.tick_cache.append(tick_record) if len(self.tick_cache) > 1000: self.tick_cache = self.tick_cache[-1000:] # NO COB SIMULATION - Real COB data comes from enhanced orchestrator status = "CLOSED" if kline['x'] else "LIVE" logger.debug(f"[WS] {status} kline: {current_price:.2f}, Vol: {tick_record['volume']:.0f} (cache: {len(self.tick_cache)})") except Exception as e: logger.warning(f"WebSocket message error: {e}") def on_error(ws, error): logger.error(f"WebSocket error: {error}") self.is_streaming = False def on_close(ws, close_status_code, close_msg): logger.warning("WebSocket connection closed") self.is_streaming = False def on_open(ws): logger.info("WebSocket connected") self.is_streaming = True # Binance WebSocket - Use kline stream for OHLCV data ws_url = "wss://stream.binance.com:9443/ws/ethusdt@kline_1s" ws = websocket.WebSocketApp( ws_url, on_message=on_message, on_error=on_error, on_close=on_close, on_open=on_open ) ws.run_forever() except Exception as e: logger.error(f"WebSocket worker error: {e}") self.is_streaming = False # Start WebSocket thread ws_thread = threading.Thread(target=ws_worker, daemon=True) ws_thread.start() # NO COB SIMULATION - Real COB data managed by enhanced orchestrator except Exception as e: logger.error(f"Error starting WebSocket: {e}") def _start_data_collection(self): """Start background data collection""" try: def data_worker(): while True: try: # Update recent decisions from orchestrator if self.orchestrator and hasattr(self.orchestrator, 'get_recent_decisions'): decisions = self.orchestrator.get_recent_decisions('ETH/USDT') if decisions: self.recent_decisions = decisions[-20:] # Keep last 20 # Update closed trades if self.trading_executor and hasattr(self.trading_executor, 'get_closed_trades'): trades = self.trading_executor.get_closed_trades() if trades: self.closed_trades = trades # Update session metrics self._update_session_metrics() time.sleep(5) # Update every 5 seconds except Exception as e: logger.warning(f"Data collection error: {e}") time.sleep(10) # Wait longer on error # Start data collection thread data_thread = threading.Thread(target=data_worker, daemon=True) data_thread.start() except Exception as e: logger.error(f"Error starting data collection: {e}") def _update_session_metrics(self): """Update session P&L and metrics""" try: # Calculate session P&L from closed trades if self.closed_trades: self.session_pnl = sum(trade.get('pnl', 0) for trade in self.closed_trades) self.total_fees = sum(trade.get('fees', 0) for trade in self.closed_trades) # Update current position if self.trading_executor and hasattr(self.trading_executor, 'get_current_position'): position = self.trading_executor.get_current_position() self.current_position = position except Exception as e: logger.warning(f"Error updating session metrics: {e}") def run_server(self, host='127.0.0.1', port=8051, debug=False): """Run the dashboard server""" # Set logging level for Flask/Werkzeug to reduce noise if not debug: logging.getLogger('werkzeug').setLevel(logging.ERROR) logger.info(f"Starting Clean Trading Dashboard at http://{host}:{port}") self.app.run(host=host, port=port, debug=debug, dev_tools_silence_routes_logging=True) def stop(self): """Stop the dashboard and cleanup resources""" try: self.is_streaming = False logger.info("Clean Trading Dashboard stopped") except Exception as e: logger.error(f"Error stopping dashboard: {e}") def _start_unified_stream(self): """Start the unified data stream in background""" try: if self.unified_stream is None: logger.warning("Unified stream is None - cannot start") return import asyncio loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) loop.run_until_complete(self.unified_stream.start_streaming()) except Exception as e: logger.error(f"Error starting unified stream: {e}") def _handle_unified_stream_data(self, data_packet: Dict[str, Any]): """Handle incoming data from the Universal Data Stream (5 timeseries)""" try: # Extract the universal 5 timeseries data if 'ticks' in data_packet and data_packet['ticks']: # Update tick cache with real-time data self.tick_cache.extend(data_packet['ticks'][-50:]) # Last 50 ticks if len(self.tick_cache) > 1000: self.tick_cache = self.tick_cache[-1000:] if 'ohlcv' in data_packet: # Update multi-timeframe data multi_tf_data = data_packet.get('multi_timeframe', {}) for symbol in ['ETH/USDT', 'BTC/USDT']: if symbol in multi_tf_data: for timeframe in ['1s', '1m', '1h', '1d']: if timeframe in multi_tf_data[symbol]: # Update internal cache with universal data tf_data = multi_tf_data[symbol][timeframe] if tf_data: # Update current prices from universal stream latest_bar = tf_data[-1] if 'close' in latest_bar: self.current_prices[symbol] = latest_bar['close'] self.ws_price_cache[symbol.replace('/', '')] = latest_bar['close'] if 'ui_data' in data_packet and data_packet['ui_data']: # Process UI-specific data updates ui_data = data_packet['ui_data'] # This could include formatted data specifically for dashboard display pass if 'training_data' in data_packet and data_packet['training_data']: # Process training data for real-time model updates training_data = data_packet['training_data'] # This includes market state and model features pass # Log periodic universal data stream stats consumer_name = data_packet.get('consumer_name', 'unknown') if hasattr(self, '_stream_update_count'): self._stream_update_count += 1 else: self._stream_update_count = 1 if self._stream_update_count % 100 == 0: # Every 100 updates logger.info(f"Universal Stream: {self._stream_update_count} updates processed for {consumer_name}") logger.debug(f"Current data: ticks={len(data_packet.get('ticks', []))}, " f"tf_symbols={len(data_packet.get('multi_timeframe', {}))}") except Exception as e: logger.error(f"Error handling universal stream data: {e}") def _update_case_index(self, case_dir: str, case_id: str, case_summary: Dict[str, Any], case_type: str): """Update the case index file with new case information""" try: import json import os index_filepath = os.path.join(case_dir, "case_index.json") # Load existing index or create new one if os.path.exists(index_filepath): with open(index_filepath, 'r') as f: index_data = json.load(f) else: index_data = { "cases": [], "last_updated": datetime.now().isoformat(), "case_type": case_type, "total_cases": 0 } # Add new case to index pnl = case_summary.get('pnl', 0) training_priority = 1 # Default priority # Calculate training priority based on P&L and confidence if case_type == "negative": # Higher priority for bigger losses if abs(pnl) > 10: training_priority = 5 # Very high priority elif abs(pnl) > 5: training_priority = 4 elif abs(pnl) > 1: training_priority = 3 else: training_priority = 2 else: # positive # Higher priority for high-confidence profitable trades confidence = case_summary.get('confidence', 0) if pnl > 5 and confidence > 0.8: training_priority = 5 elif pnl > 1 and confidence > 0.6: training_priority = 4 elif pnl > 0.5: training_priority = 3 else: training_priority = 2 case_entry = { "case_id": case_id, "timestamp": case_summary['timestamp'], "symbol": case_summary['symbol'], "side": case_summary['side'], "entry_price": case_summary['entry_price'], "pnl": pnl, "confidence": case_summary.get('confidence', 0), "trade_type": case_summary.get('trade_type', 'unknown'), "training_priority": training_priority, "retraining_count": 0, "model_inputs_captured": case_summary.get('model_inputs_captured', False), "feature_counts": case_summary.get('feature_counts', {}), "created_at": datetime.now().isoformat() } # Add to cases list index_data["cases"].append(case_entry) index_data["last_updated"] = datetime.now().isoformat() index_data["total_cases"] = len(index_data["cases"]) # Sort by training priority (highest first) and timestamp (newest first) index_data["cases"].sort(key=lambda x: (-x['training_priority'], -time.mktime(datetime.fromisoformat(x['timestamp']).timetuple()))) # Keep only last 1000 cases to prevent index from getting too large if len(index_data["cases"]) > 1000: index_data["cases"] = index_data["cases"][:1000] index_data["total_cases"] = 1000 # Save updated index with open(index_filepath, 'w') as f: json.dump(index_data, f, indent=2, default=str) logger.debug(f"Updated {case_type} case index: {len(index_data['cases'])} total cases") except Exception as e: logger.error(f"Error updating case index: {e}") def get_testcase_summary(self) -> Dict[str, Any]: """Get summary of stored testcases for display""" try: import os import json summary = { 'positive_cases': 0, 'negative_cases': 0, 'total_cases': 0, 'latest_cases': [], 'high_priority_cases': 0 } base_dir = "testcases" for case_type in ['positive', 'negative']: case_dir = os.path.join(base_dir, case_type) index_filepath = os.path.join(case_dir, "case_index.json") if os.path.exists(index_filepath): with open(index_filepath, 'r') as f: index_data = json.load(f) case_count = len(index_data.get('cases', [])) summary[f'{case_type}_cases'] = case_count summary['total_cases'] += case_count # Get high priority cases high_priority = len([c for c in index_data.get('cases', []) if c.get('training_priority', 1) >= 4]) summary['high_priority_cases'] += high_priority # Get latest cases latest = index_data.get('cases', [])[:5] # Top 5 latest for case in latest: case['case_type'] = case_type summary['latest_cases'].extend(latest) # Sort latest cases by timestamp summary['latest_cases'].sort(key=lambda x: x.get('timestamp', ''), reverse=True) # Keep only top 10 latest cases summary['latest_cases'] = summary['latest_cases'][:10] return summary except Exception as e: logger.error(f"Error getting testcase summary: {e}") return { 'positive_cases': 0, 'negative_cases': 0, 'total_cases': 0, 'latest_cases': [], 'high_priority_cases': 0, 'error': str(e) } def create_clean_dashboard(data_provider: Optional[DataProvider] = None, orchestrator: Optional[TradingOrchestrator] = None, trading_executor: Optional[TradingExecutor] = None): """Factory function to create a CleanTradingDashboard instance""" return CleanTradingDashboard( data_provider=data_provider, orchestrator=orchestrator, trading_executor=trading_executor )