showing some cob info
This commit is contained in:
Dobromir Popov
@ -879,12 +879,27 @@ class CleanTradingDashboard:
|
||||
'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': 'Active' if COB_INTEGRATION_AVAILABLE else 'Inactive',
|
||||
'cob_status': 'Simulated' if self.is_streaming else 'Inactive', # Show simulation status
|
||||
'rl_model_status': 'Inactive',
|
||||
'predictions_count': 0,
|
||||
'cache_size': 0
|
||||
}
|
||||
|
||||
# Check COB cache status
|
||||
if hasattr(self, 'cob_cache') and self.cob_cache:
|
||||
active_symbols = []
|
||||
total_updates = 0
|
||||
|
||||
for symbol, cache_data in self.cob_cache.items():
|
||||
if cache_data.get('data') and cache_data.get('last_update', 0) > 0:
|
||||
active_symbols.append(symbol)
|
||||
total_updates += cache_data.get('updates_count', 0)
|
||||
|
||||
if active_symbols:
|
||||
status['cob_status'] = f'Simulated ({len(active_symbols)} symbols)'
|
||||
status['cache_size'] = total_updates
|
||||
status['active_symbols'] = active_symbols
|
||||
|
||||
# Check COB RL trader status
|
||||
if self.cob_rl_trader:
|
||||
status['cob_status'] = 'Active'
|
||||
@ -902,7 +917,12 @@ class CleanTradingDashboard:
|
||||
elif self.orchestrator and hasattr(self.orchestrator, 'cob_integration'):
|
||||
cob_integration = self.orchestrator.cob_integration
|
||||
if cob_integration and hasattr(cob_integration, 'is_active'):
|
||||
status['cob_status'] = 'Active' if cob_integration.is_active else 'Inactive'
|
||||
orchestrator_status = 'Active' if cob_integration.is_active else 'Inactive'
|
||||
# Combine with simulation status
|
||||
if status['cob_status'].startswith('Simulated'):
|
||||
status['cob_status'] = f"{status['cob_status']} + {orchestrator_status} (Orchestrator)"
|
||||
else:
|
||||
status['cob_status'] = orchestrator_status
|
||||
|
||||
return status
|
||||
|
||||
@ -913,20 +933,39 @@ class CleanTradingDashboard:
|
||||
def _get_cob_snapshot(self, symbol: str) -> Optional[Any]:
|
||||
"""Get COB snapshot for symbol"""
|
||||
try:
|
||||
# First try to get from cache (simulated COB data)
|
||||
if symbol in self.cob_cache and self.cob_cache[symbol]['data']:
|
||||
cache_entry = self.cob_cache[symbol]
|
||||
current_time = time.time()
|
||||
|
||||
# Check if data is fresh (within last 10 seconds)
|
||||
if current_time - cache_entry['last_update'] < 10:
|
||||
logger.debug(f"Retrieved cached COB data for {symbol}")
|
||||
return cache_entry['data']
|
||||
else:
|
||||
logger.debug(f"Cached COB data for {symbol} is stale")
|
||||
|
||||
# Fallback to orchestrator COB integration if available
|
||||
if not COB_INTEGRATION_AVAILABLE:
|
||||
logger.debug("COB integration not available")
|
||||
logger.debug("COB integration not available, generating fallback COB data")
|
||||
# Generate fallback COB data for display
|
||||
current_price = self._get_current_price(symbol)
|
||||
if current_price:
|
||||
self._generate_simulated_cob_data(symbol, current_price)
|
||||
if symbol in self.cob_cache and self.cob_cache[symbol]['data']:
|
||||
return self.cob_cache[symbol]['data']
|
||||
return None
|
||||
|
||||
if self.orchestrator and hasattr(self.orchestrator, 'cob_integration'):
|
||||
cob_integration = self.orchestrator.cob_integration
|
||||
if cob_integration and hasattr(cob_integration, 'get_cob_snapshot'):
|
||||
logger.debug(f"Getting COB snapshot for {symbol}")
|
||||
logger.debug(f"Getting COB snapshot for {symbol} from orchestrator")
|
||||
snapshot = cob_integration.get_cob_snapshot(symbol)
|
||||
if snapshot:
|
||||
logger.debug(f"Got COB snapshot for {symbol}: {type(snapshot)}")
|
||||
return snapshot
|
||||
else:
|
||||
logger.debug(f"No COB snapshot available for {symbol}")
|
||||
logger.debug(f"No COB snapshot available for {symbol} from orchestrator")
|
||||
else:
|
||||
logger.debug("COB integration has no get_cob_snapshot method")
|
||||
else:
|
||||
@ -1302,7 +1341,7 @@ class CleanTradingDashboard:
|
||||
logger.error(f"Error initializing streaming: {e}")
|
||||
|
||||
def _start_websocket_streaming(self):
|
||||
"""Start WebSocket streaming for real-time data"""
|
||||
"""Start WebSocket streaming for real-time data including COB"""
|
||||
try:
|
||||
def ws_worker():
|
||||
try:
|
||||
@ -1338,6 +1377,10 @@ class CleanTradingDashboard:
|
||||
if len(self.tick_cache) > 1000:
|
||||
self.tick_cache = self.tick_cache[-1000:]
|
||||
|
||||
# Update COB cache with simulated COB data every few ticks
|
||||
if len(self.tick_cache) % 5 == 0: # Every 5 seconds
|
||||
self._update_cob_cache_from_price_data('ETH/USDT', current_price)
|
||||
|
||||
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:
|
||||
@ -1376,9 +1419,119 @@ class CleanTradingDashboard:
|
||||
ws_thread = threading.Thread(target=ws_worker, daemon=True)
|
||||
ws_thread.start()
|
||||
|
||||
# Start COB data simulation thread
|
||||
self._start_cob_simulation_thread()
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error starting WebSocket: {e}")
|
||||
|
||||
def _start_cob_simulation_thread(self):
|
||||
"""Start COB data simulation for demonstration"""
|
||||
try:
|
||||
def cob_worker():
|
||||
while True:
|
||||
try:
|
||||
if self.is_streaming:
|
||||
# Generate simulated COB data for both symbols
|
||||
for symbol in ['ETH/USDT', 'BTC/USDT']:
|
||||
current_price = self._get_current_price(symbol)
|
||||
if current_price:
|
||||
self._generate_simulated_cob_data(symbol, current_price)
|
||||
|
||||
time.sleep(2) # Update COB data every 2 seconds
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(f"COB simulation error: {e}")
|
||||
time.sleep(5)
|
||||
|
||||
# Start COB simulation thread
|
||||
cob_thread = threading.Thread(target=cob_worker, daemon=True)
|
||||
cob_thread.start()
|
||||
|
||||
logger.info("COB simulation thread started")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error starting COB simulation: {e}")
|
||||
|
||||
def _update_cob_cache_from_price_data(self, symbol: str, current_price: float):
|
||||
"""Update COB cache using price data as a base"""
|
||||
try:
|
||||
# Update COB cache with price-based data
|
||||
if symbol not in self.cob_cache:
|
||||
self.cob_cache[symbol] = {'last_update': 0, 'data': None, 'updates_count': 0}
|
||||
|
||||
# Generate simulated COB data based on current price
|
||||
self._generate_simulated_cob_data(symbol, current_price)
|
||||
|
||||
except Exception as e:
|
||||
logger.debug(f"Error updating COB cache for {symbol}: {e}")
|
||||
|
||||
def _generate_simulated_cob_data(self, symbol: str, current_price: float):
|
||||
"""Generate simulated COB data for display"""
|
||||
try:
|
||||
import random
|
||||
|
||||
# Create simulated COB snapshot
|
||||
simulated_cob = type('COBSnapshot', (), {})()
|
||||
|
||||
# Basic properties
|
||||
simulated_cob.symbol = symbol
|
||||
simulated_cob.volume_weighted_mid = current_price
|
||||
simulated_cob.spread_bps = random.uniform(2.0, 8.0) # 2-8 basis points spread
|
||||
|
||||
# Generate bid/ask liquidity
|
||||
base_liquidity = random.uniform(50000, 200000) # $50k-$200k base liquidity
|
||||
simulated_cob.total_bid_liquidity = base_liquidity * random.uniform(0.8, 1.2)
|
||||
simulated_cob.total_ask_liquidity = base_liquidity * random.uniform(0.8, 1.2)
|
||||
|
||||
# Calculate imbalance
|
||||
total_liquidity = simulated_cob.total_bid_liquidity + simulated_cob.total_ask_liquidity
|
||||
if total_liquidity > 0:
|
||||
simulated_cob.liquidity_imbalance = (simulated_cob.total_bid_liquidity - simulated_cob.total_ask_liquidity) / total_liquidity
|
||||
else:
|
||||
simulated_cob.liquidity_imbalance = 0.0
|
||||
|
||||
# Generate bid/ask levels
|
||||
simulated_cob.consolidated_bids = []
|
||||
simulated_cob.consolidated_asks = []
|
||||
|
||||
# Generate 10 bid levels
|
||||
for i in range(10):
|
||||
bid_price = current_price * (1 - (i + 1) * 0.0001) # 1 basis point increments down
|
||||
bid_volume = random.uniform(1000, 10000) # Random volume
|
||||
bid_level = type('BidLevel', (), {})()
|
||||
bid_level.price = bid_price
|
||||
bid_level.total_volume_usd = bid_volume
|
||||
bid_level.exchange_breakdown = {'Binance': bid_volume * 0.4, 'OKX': bid_volume * 0.3, 'Bybit': bid_volume * 0.3}
|
||||
simulated_cob.consolidated_bids.append(bid_level)
|
||||
|
||||
# Generate 10 ask levels
|
||||
for i in range(10):
|
||||
ask_price = current_price * (1 + (i + 1) * 0.0001) # 1 basis point increments up
|
||||
ask_volume = random.uniform(1000, 10000) # Random volume
|
||||
ask_level = type('AskLevel', (), {})()
|
||||
ask_level.price = ask_price
|
||||
ask_level.total_volume_usd = ask_volume
|
||||
ask_level.exchange_breakdown = {'Binance': ask_volume * 0.4, 'OKX': ask_volume * 0.3, 'Bybit': ask_volume * 0.3}
|
||||
simulated_cob.consolidated_asks.append(ask_level)
|
||||
|
||||
# Update cache
|
||||
self.cob_cache[symbol] = {
|
||||
'last_update': time.time(),
|
||||
'data': simulated_cob,
|
||||
'updates_count': self.cob_cache.get(symbol, {}).get('updates_count', 0) + 1
|
||||
}
|
||||
|
||||
# Log periodic updates
|
||||
update_count = self.cob_cache[symbol]['updates_count']
|
||||
if update_count % 20 == 0: # Every 20 updates
|
||||
logger.info(f"[COB-SIM] {symbol} - Update #{update_count}, "
|
||||
f"Mid: ${current_price:.2f}, Spread: {simulated_cob.spread_bps:.1f}bps, "
|
||||
f"Imbalance: {simulated_cob.liquidity_imbalance:.3f}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error generating simulated COB data for {symbol}: {e}")
|
||||
|
||||
def _start_data_collection(self):
|
||||
"""Start background data collection"""
|
||||
try:
|
||||
|
||||
Reference in New Issue
Block a user