popov/gogo2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

cleanup enhanced orchestrator

Browse Source
This commit is contained in:
Dobromir Popov
2025-06-25 15:57:05 +03:00
parent 8a51fcb70a
commit c6094160d7
1 changed files with 125 additions and 289 deletions
Download Patch File Download Diff File Expand all files Collapse all files

280
web/clean_dashboard.py
View File

@ -58,8 +58,6 @@ from core.trading_executor import TradingExecutor
from web.layout_manager import DashboardLayoutManager from web.layout_manager import DashboardLayoutManager
from web.component_manager import DashboardComponentManager from web.component_manager import DashboardComponentManager
# Enhanced RL components are no longer available - using Basic orchestrator only
ENHANCED_RL_AVAILABLE = False
try: try:
from core.cob_integration import COBIntegration from core.cob_integration import COBIntegration
@ -143,6 +141,7 @@ class CleanTradingDashboard:
'BTC/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 self.latest_cob_data = {} # Cache for COB integration data
self.cob_predictions = {} # Cache for COB predictions (both ETH and BTC for display)
# Initialize timezone # Initialize timezone
timezone_name = self.config.get('system', {}).get('timezone', 'Europe/Sofia') timezone_name = self.config.get('system', {}).get('timezone', 'Europe/Sofia')
@ -1031,48 +1030,9 @@ class CleanTradingDashboard:
# Check COB integration in Enhanced orchestrator # Check COB integration in Enhanced orchestrator
if hasattr(self.orchestrator, 'cob_integration'): if hasattr(self.orchestrator, 'cob_integration'):
cob_integration = getattr(self.orchestrator, 'cob_integration', None) cob_integration = getattr(self.orchestrator, 'cob_integration', None)
if cob_integration is not None: # Basic orchestrator only - no enhanced COB features
# 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['cob_status'] = 'Basic Orchestrator (No COB Support)'
status['orchestrator_type'] = 'Basic (Enhanced Not Used)' status['orchestrator_type'] = 'Basic'
return status return status
@ -1081,38 +1041,14 @@ class CleanTradingDashboard:
return {'error': str(e), 'cob_status': 'Error Getting Status', 'orchestrator_type': 'Unknown'} return {'error': str(e), 'cob_status': 'Error Getting Status', 'orchestrator_type': 'Unknown'}
def _get_cob_snapshot(self, symbol: str) -> Optional[Any]: def _get_cob_snapshot(self, symbol: str) -> Optional[Any]:
"""Get COB snapshot for symbol using enhanced orchestrator approach""" """Get COB snapshot for symbol - Basic orchestrator has no COB features"""
try: try:
# Get from Enhanced Orchestrator's COB integration (proper way) # Basic orchestrator has no COB integration
if (ENHANCED_ORCHESTRATOR_AVAILABLE and logger.debug(f"No COB integration available for {symbol} (Basic orchestrator)")
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 return None
except Exception as e: except Exception as e:
logger.warning(f"Error getting Enhanced COB snapshot for {symbol}: {e}") logger.warning(f"Error getting COB snapshot for {symbol}: {e}")
return None return None
def _get_training_metrics(self) -> Dict: def _get_training_metrics(self) -> Dict:
@ -1176,7 +1112,7 @@ class CleanTradingDashboard:
}, },
'loss_5ma': cnn_last_loss, 'loss_5ma': cnn_last_loss,
'model_type': 'CNN', 'model_type': 'CNN',
'description': 'Williams Market Structure CNN' + (' (Enhanced Only)' if not is_enhanced else '') 'description': 'Williams Market Structure CNN (Basic Orchestrator - Inactive)'
} }
loaded_models['cnn'] = cnn_model_info loaded_models['cnn'] = cnn_model_info
@ -1190,12 +1126,12 @@ class CleanTradingDashboard:
'parameters': 400000000, # 400M optimized (Enhanced COB integration) 'parameters': 400000000, # 400M optimized (Enhanced COB integration)
'last_prediction': { 'last_prediction': {
'timestamp': datetime.now().strftime('%H:%M:%S'), 'timestamp': datetime.now().strftime('%H:%M:%S'),
'action': 'ENHANCED_COB_INFERENCE' if cob_active else ('INACTIVE' if is_enhanced else 'NOT_AVAILABLE'), 'action': 'INACTIVE',
'confidence': 0.0 'confidence': 0.0
}, },
'loss_5ma': cob_last_loss, 'loss_5ma': cob_last_loss,
'model_type': 'ENHANCED_COB_RL', 'model_type': 'ENHANCED_COB_RL',
'description': 'Enhanced COB Integration' + (' (Enhanced Only)' if not is_enhanced else ''), 'description': 'Enhanced COB Integration (Basic Orchestrator - Inactive)',
'predictions_count': cob_predictions_count 'predictions_count': cob_predictions_count
} }
loaded_models['cob_rl'] = cob_model_info loaded_models['cob_rl'] = cob_model_info
@ -1212,7 +1148,7 @@ class CleanTradingDashboard:
'last_update': datetime.now().strftime('%H:%M:%S'), 'last_update': datetime.now().strftime('%H:%M:%S'),
'models_loaded': len(loaded_models), 'models_loaded': len(loaded_models),
'total_parameters': sum(m['parameters'] for m in loaded_models.values() if m['active']), 'total_parameters': sum(m['parameters'] for m in loaded_models.values() if m['active']),
'orchestrator_type': 'Enhanced' if is_enhanced else 'Basic' 'orchestrator_type': 'Basic'
} }
# COB $1 Buckets (sample data for now) # COB $1 Buckets (sample data for now)
@ -1272,8 +1208,8 @@ class CleanTradingDashboard:
while True: while True:
try: try:
# Generate signals for both symbols # Generate signals for ETH only (ignore BTC)
for symbol in ['ETH/USDT', 'BTC/USDT']: for symbol in ['ETH/USDT']: # Only ETH signals
try: try:
# Get current price # Get current price
current_price = self._get_current_price(symbol) current_price = self._get_current_price(symbol)
@ -1371,7 +1307,7 @@ class CleanTradingDashboard:
self.recent_decisions = self.recent_decisions[-20:] self.recent_decisions = self.recent_decisions[-20:]
# Log signal generation # Log signal generation
logger.info(f"Generated {signal['action']} signal for {signal['symbol']} " logger.info(f"Generated ETH {signal['action']} signal for {signal['symbol']} "
f"(conf: {signal['confidence']:.2f}, model: {signal.get('model', 'UNKNOWN')})") f"(conf: {signal['confidence']:.2f}, model: {signal.get('model', 'UNKNOWN')})")
# DQN training not available in Basic orchestrator # DQN training not available in Basic orchestrator
@ -1720,141 +1656,31 @@ class CleanTradingDashboard:
logger.error(f"Error clearing session: {e}") logger.error(f"Error clearing session: {e}")
def _initialize_cob_integration_proper(self): def _initialize_cob_integration_proper(self):
"""Initialize COB integration using Enhanced Orchestrator - PROPER APPROACH""" """Initialize COB integration - Basic orchestrator has no COB features"""
try: try:
logger.info("Connecting to COB integration from Enhanced Orchestrator...") logger.info("Basic orchestrator has no COB integration features")
logger.info("COB integration not available with Basic 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: except Exception as e:
logger.error(f"Error starting COB integration from Enhanced Orchestrator: {e}") logger.error(f"Error in COB integration init: {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): def _on_enhanced_cob_update(self, symbol: str, cob_data: Dict):
"""Handle Enhanced COB data updates - NO SIMULATION""" """Handle Enhanced COB data updates - Basic orchestrator has no COB features"""
try: try:
# Process Enhanced COB data update logger.debug("Enhanced COB updates not available with Basic orchestrator")
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: except Exception as e:
logger.error(f"Error handling Enhanced COB update for {symbol}: {e}") logger.error(f"Error handling COB update for {symbol}: {e}")
def _start_cob_data_subscription(self): def _start_cob_data_subscription(self):
"""Start COB data subscription with proper caching""" """Start COB data subscription - Basic orchestrator has no COB features"""
try: try:
# Start the COB RL trader asynchronously logger.info("COB data subscription not available with Basic orchestrator")
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: except Exception as e:
logger.error(f"Error in COB trader loop: {e}") logger.error(f"Error in COB subscription: {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): def _on_cob_prediction(self, prediction: PredictionResult):
"""Handle COB RL predictions""" """Handle COB RL predictions - Display both ETH and BTC for reference"""
try: try:
with self.cob_lock:
# Convert prediction to dashboard format # Convert prediction to dashboard format
prediction_data = { prediction_data = {
'timestamp': prediction.timestamp, 'timestamp': prediction.timestamp,
@ -1865,28 +1691,18 @@ class CleanTradingDashboard:
'color': ['red', 'gray', 'green'][prediction.predicted_direction] 'color': ['red', 'gray', 'green'][prediction.predicted_direction]
} }
# Add to predictions cache # Add predictions to cache for both ETH and BTC (for reference/display)
self.cob_predictions[prediction.symbol].append(prediction_data) if hasattr(prediction, 'symbol') and prediction.symbol:
symbol = prediction.symbol
# Store predictions for display (both ETH and BTC)
if symbol not in self.cob_predictions:
self.cob_predictions[symbol] = deque(maxlen=100)
# Cache COB data (1s buckets for 1 day max, 5 min retention) self.cob_predictions[symbol].append(prediction_data)
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) # Log all predictions but note that only ETH generates trading signals
self.cob_data_cache_1d[prediction.symbol].append(cob_data) signal_note = " (TRADING ENABLED)" if 'ETH' in symbol.upper() else " (REFERENCE ONLY)"
logger.debug(f"COB prediction cached for {symbol}{signal_note}: "
# 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})") f"{prediction_data['direction_text']} (confidence: {prediction.confidence:.3f})")
except Exception as e: except Exception as e:
@ -1905,8 +1721,20 @@ class CleanTradingDashboard:
logger.error(f"Error connecting to orchestrator: {e}") logger.error(f"Error connecting to orchestrator: {e}")
def _on_trading_decision(self, decision): def _on_trading_decision(self, decision):
"""Handle trading decision from orchestrator""" """Handle trading decision from orchestrator - Filter to show only ETH signals"""
try: try:
# Check if this decision is for ETH/USDT - ignore all BTC signals
symbol = None
if hasattr(decision, 'symbol'):
symbol = decision.symbol
elif isinstance(decision, dict) and 'symbol' in decision:
symbol = decision.get('symbol')
# Only process ETH signals, ignore BTC
if symbol and 'BTC' in symbol.upper():
logger.debug(f"Ignoring BTC signal: {symbol}")
return
# Convert orchestrator decision to dashboard format # Convert orchestrator decision to dashboard format
# Handle both TradingDecision objects and dictionary formats # Handle both TradingDecision objects and dictionary formats
if hasattr(decision, 'action'): if hasattr(decision, 'action'):
@ -1916,6 +1744,7 @@ class CleanTradingDashboard:
'action': decision.action, 'action': decision.action,
'confidence': decision.confidence, 'confidence': decision.confidence,
'price': decision.price, 'price': decision.price,
'symbol': getattr(decision, 'symbol', 'ETH/USDT'), # Add symbol field
'executed': True, # Orchestrator decisions are executed 'executed': True, # Orchestrator decisions are executed
'blocked': False, 'blocked': False,
'manual': False 'manual': False
@ -1927,11 +1756,14 @@ class CleanTradingDashboard:
'action': decision.get('action', 'UNKNOWN'), 'action': decision.get('action', 'UNKNOWN'),
'confidence': decision.get('confidence', 0), 'confidence': decision.get('confidence', 0),
'price': decision.get('price', 0), 'price': decision.get('price', 0),
'symbol': decision.get('symbol', 'ETH/USDT'), # Add symbol field
'executed': True, # Orchestrator decisions are executed 'executed': True, # Orchestrator decisions are executed
'blocked': False, 'blocked': False,
'manual': False 'manual': False
} }
# Only show ETH signals in dashboard
if dashboard_decision['symbol'] and 'ETH' in dashboard_decision['symbol'].upper():
# Add to recent decisions # Add to recent decisions
self.recent_decisions.append(dashboard_decision) self.recent_decisions.append(dashboard_decision)
@ -1939,6 +1771,10 @@ class CleanTradingDashboard:
if len(self.recent_decisions) > 50: if len(self.recent_decisions) > 50:
self.recent_decisions = self.recent_decisions[-50:] self.recent_decisions = self.recent_decisions[-50:]
logger.info(f"ETH signal added to dashboard: {dashboard_decision['action']} (conf: {dashboard_decision['confidence']:.2f})")
else:
logger.debug(f"Non-ETH signal ignored: {dashboard_decision.get('symbol', 'UNKNOWN')}")
except Exception as e: except Exception as e:
logger.error(f"Error handling trading decision: {e}") logger.error(f"Error handling trading decision: {e}")
@ -2130,9 +1966,9 @@ class CleanTradingDashboard:
self.tick_cache = self.tick_cache[-1000:] self.tick_cache = self.tick_cache[-1000:]
if 'ohlcv' in data_packet: if 'ohlcv' in data_packet:
# Update multi-timeframe data # Update multi-timeframe data for both ETH and BTC (BTC for reference)
multi_tf_data = data_packet.get('multi_timeframe', {}) multi_tf_data = data_packet.get('multi_timeframe', {})
for symbol in ['ETH/USDT', 'BTC/USDT']: for symbol in ['ETH/USDT', 'BTC/USDT']: # Process both ETH and BTC data
if symbol in multi_tf_data: if symbol in multi_tf_data:
for timeframe in ['1s', '1m', '1h', '1d']: for timeframe in ['1s', '1m', '1h', '1d']:
if timeframe in multi_tf_data[symbol]: if timeframe in multi_tf_data[symbol]: