#!/usr/bin/env python3 """ Crash-Safe Dashboard Runner This runner is designed to prevent crashes by: 1. Isolating imports with try/except blocks 2. Minimal initialization 3. Graceful error handling 4. No complex training loops 5. Safe component loading """ import os import sys import logging import traceback from pathlib import Path # Fix environment issues before any imports os.environ['KMP_DUPLICATE_LIB_OK'] = 'TRUE' os.environ['OMP_NUM_THREADS'] = '1' # Minimal threads os.environ['MPLBACKEND'] = 'Agg' # Add project root to path project_root = Path(__file__).parent sys.path.insert(0, str(project_root)) # Setup basic logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # Reduce noise from other loggers logging.getLogger('werkzeug').setLevel(logging.ERROR) logging.getLogger('dash').setLevel(logging.ERROR) logging.getLogger('matplotlib').setLevel(logging.ERROR) class CrashSafeDashboard: """Crash-safe dashboard with minimal dependencies""" def __init__(self): """Initialize with safe error handling""" self.components = {} self.dashboard_app = None self.initialization_errors = [] logger.info("Initializing crash-safe dashboard...") def safe_import(self, module_name, class_name=None): """Safely import modules with error handling""" try: if class_name: module = __import__(module_name, fromlist=[class_name]) return getattr(module, class_name) else: return __import__(module_name) except Exception as e: error_msg = f"Failed to import {module_name}.{class_name if class_name else ''}: {e}" logger.error(error_msg) self.initialization_errors.append(error_msg) return None def initialize_core_components(self): """Initialize core components safely""" logger.info("Initializing core components...") # Try to import and initialize config try: from core.config import get_config, setup_logging setup_logging() self.components['config'] = get_config() logger.info("✓ Config loaded") except Exception as e: logger.error(f"✗ Config failed: {e}") self.initialization_errors.append(f"Config: {e}") # Try to initialize data provider try: from core.data_provider import DataProvider self.components['data_provider'] = DataProvider() logger.info("✓ Data provider initialized") except Exception as e: logger.error(f"✗ Data provider failed: {e}") self.initialization_errors.append(f"Data provider: {e}") # Try to initialize trading executor try: from core.trading_executor import TradingExecutor self.components['trading_executor'] = TradingExecutor() logger.info("✓ Trading executor initialized") except Exception as e: logger.error(f"✗ Trading executor failed: {e}") self.initialization_errors.append(f"Trading executor: {e}") # Try to initialize orchestrator (WITHOUT training to avoid crashes) try: from core.orchestrator import TradingOrchestrator self.components['orchestrator'] = TradingOrchestrator( data_provider=self.components.get('data_provider'), enhanced_rl_training=False # DISABLED to prevent crashes ) logger.info("✓ Orchestrator initialized (training disabled)") except Exception as e: logger.error(f"✗ Orchestrator failed: {e}") self.initialization_errors.append(f"Orchestrator: {e}") def create_minimal_dashboard(self): """Create minimal dashboard without complex features""" try: import dash from dash import html, dcc # Create minimal Dash app self.dashboard_app = dash.Dash(__name__) # Create simple layout self.dashboard_app.layout = html.Div([ html.H1("Trading Dashboard - Safe Mode", style={'textAlign': 'center'}), html.Hr(), # Status section html.Div([ html.H3("System Status"), html.Div(id="system-status", children=self._get_system_status()), ], style={'margin': '20px'}), # Error section html.Div([ html.H3("Initialization Status"), html.Div(id="init-status", children=self._get_init_status()), ], style={'margin': '20px'}), # Simple refresh interval dcc.Interval( id='interval-component', interval=5000, # Update every 5 seconds n_intervals=0 ) ]) # Add simple callback @self.dashboard_app.callback( [dash.dependencies.Output('system-status', 'children'), dash.dependencies.Output('init-status', 'children')], [dash.dependencies.Input('interval-component', 'n_intervals')] ) def update_status(n): try: return self._get_system_status(), self._get_init_status() except Exception as e: logger.error(f"Callback error: {e}") return f"Callback error: {e}", "Error in callback" logger.info("✓ Minimal dashboard created") return True except Exception as e: logger.error(f"✗ Dashboard creation failed: {e}") logger.error(traceback.format_exc()) return False def _get_system_status(self): """Get system status for display""" try: status_items = [] # Check components for name, component in self.components.items(): if component is not None: status_items.append(html.P(f"✓ {name.replace('_', ' ').title()}: OK", style={'color': 'green'})) else: status_items.append(html.P(f"✗ {name.replace('_', ' ').title()}: Failed", style={'color': 'red'})) # Add timestamp status_items.append(html.P(f"Last update: {datetime.now().strftime('%H:%M:%S')}", style={'color': 'gray', 'fontSize': '12px'})) return status_items except Exception as e: return [html.P(f"Status error: {e}", style={'color': 'red'})] def _get_init_status(self): """Get initialization status for display""" try: if not self.initialization_errors: return [html.P("✓ All components initialized successfully", style={'color': 'green'})] error_items = [html.P("⚠️ Some components failed to initialize:", style={'color': 'orange'})] for error in self.initialization_errors: error_items.append(html.P(f"• {error}", style={'color': 'red', 'fontSize': '12px'})) return error_items except Exception as e: return [html.P(f"Init status error: {e}", style={'color': 'red'})] def run(self, port=8051): """Run the crash-safe dashboard""" try: logger.info("=" * 60) logger.info("CRASH-SAFE DASHBOARD") logger.info("=" * 60) logger.info("Mode: Safe mode with minimal features") logger.info("Training: Completely disabled") logger.info("Focus: System stability and basic monitoring") logger.info("=" * 60) # Initialize components self.initialize_core_components() # Create dashboard if not self.create_minimal_dashboard(): logger.error("Failed to create dashboard") return False # Report initialization status if self.initialization_errors: logger.warning(f"Dashboard starting with {len(self.initialization_errors)} component failures") for error in self.initialization_errors: logger.warning(f" - {error}") else: logger.info("All components initialized successfully") # Start dashboard logger.info(f"Starting dashboard on http://127.0.0.1:{port}") logger.info("Press Ctrl+C to stop") self.dashboard_app.run_server( host='127.0.0.1', port=port, debug=False, use_reloader=False, threaded=True ) return True except KeyboardInterrupt: logger.info("Dashboard stopped by user") return True except Exception as e: logger.error(f"Dashboard failed: {e}") logger.error(traceback.format_exc()) return False def main(): """Main function with comprehensive error handling""" try: dashboard = CrashSafeDashboard() success = dashboard.run() if success: logger.info("Dashboard completed successfully") else: logger.error("Dashboard failed") except Exception as e: logger.error(f"Fatal error: {e}") logger.error(traceback.format_exc()) sys.exit(1) if __name__ == "__main__": main()