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

COB fixes

Browse Source
This commit is contained in:
Dobromir Popov
2025-08-08 19:15:07 +03:00
parent 0e469c8e2f
commit be1753c96a
5 changed files with 875 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
core/data_provider.py
View File

@ -1082,7 +1082,7 @@ class DataProvider:
try:
# For 1s timeframe, generate from WebSocket tick data
if timeframe == '1s':
logger.info(f"Generating 1s candles from WebSocket ticks for {symbol}")
# logger.deta(f"Generating 1s candles from WebSocket ticks for {symbol}")
return self._generate_1s_candles_from_ticks(symbol, limit)
# Convert symbol format
@ -1239,7 +1239,7 @@ class DataProvider:
if len(df) > limit:
df = df.tail(limit)
logger.info(f"Generated {len(df)} 1s candles from {len(recent_ticks)} ticks for {symbol}")
# logger.info(f"Generated {len(df)} 1s candles from {len(recent_ticks)} ticks for {symbol}")
return df
except Exception as e:
@ -1253,10 +1253,10 @@ class DataProvider:
# For 1s timeframe, try to generate from WebSocket ticks first
if timeframe == '1s':
logger.info(f"Attempting to generate 1s candles from WebSocket ticks for {symbol}")
# logger.info(f"Attempting to generate 1s candles from WebSocket ticks for {symbol}")
generated_df = self._generate_1s_candles_from_ticks(symbol, limit)
if generated_df is not None and not generated_df.empty:
logger.info(f"Successfully generated 1s candles from WebSocket ticks for {symbol}")
# logger.info(f"Successfully generated 1s candles from WebSocket ticks for {symbol}")
return generated_df
else:
logger.info(f"Could not generate 1s candles from ticks for {symbol}; trying Binance API")
@ -1338,10 +1338,10 @@ class DataProvider:
# For 1s timeframe, try generating from WebSocket ticks first
if timeframe == '1s':
logger.info(f"FALLBACK: Attempting to generate 1s candles from WebSocket ticks for {symbol}")
# logger.info(f"FALLBACK: Attempting to generate 1s candles from WebSocket ticks for {symbol}")
generated_data = self._generate_1s_candles_from_ticks(symbol, limit)
if generated_data is not None and not generated_data.empty:
logger.info(f"FALLBACK: Generated 1s candles from WebSocket ticks for {symbol}: {len(generated_data)} bars")
# logger.info(f"FALLBACK: Generated 1s candles from WebSocket ticks for {symbol}: {len(generated_data)} bars")
return generated_data
# ONLY try cached data
@ -4763,7 +4763,7 @@ class DataProvider:
seconds: int = 300,
bucket_radius: int = 10,
metric: str = 'imbalance'
) -> Tuple[List[datetime], List[float], List[List[float]]]:
) -> Tuple[List[datetime], List[float], List[List[float]], List[float]]:
"""
Build a 1s COB heatmap matrix for ±bucket_radius buckets around current price.
@ -4774,14 +4774,15 @@ class DataProvider:
times: List[datetime] = []
prices: List[float] = []
values: List[List[float]] = []
mids: List[float] = []
latest = self.get_latest_cob_data(symbol)
if not latest or 'stats' not in latest:
return times, prices, values
return times, prices, values, mids
mid = float(latest['stats'].get('mid_price', 0) or 0)
if mid <= 0:
return times, prices, values
return times, prices, values, mids
bucket_size = 1.0 if 'ETH' in symbol else 10.0
center = round(mid / bucket_size) * bucket_size
@ -4821,6 +4822,17 @@ class DataProvider:
except Exception:
continue
# Compute mid price for this snapshot
try:
best_bid = max((float(b[0]) for b in bids), default=0.0)
best_ask = min((float(a[0]) for a in asks), default=0.0)
if best_bid > 0 and best_ask > 0:
mids.append((best_bid + best_ask) / 2.0)
else:
mids.append(0.0)
except Exception:
mids.append(0.0)
row: List[float] = []
for p in prices:
b = float(bucket_map.get(p, {}).get('bid', 0.0))
@ -4833,10 +4845,10 @@ class DataProvider:
row.append(val)
values.append(row)
return times, prices, values
return times, prices, values, mids
except Exception as e:
logger.error(f"Error building COB heatmap matrix for {symbol}: {e}")
return [], [], []
return [], [], [], []
def get_combined_ohlcv_cob_data(self, symbol: str, timeframe: str = '1s', count: int = 60) -> dict:
"""

19
core/enhanced_cob_websocket.py
View File

@ -134,9 +134,8 @@ class EnhancedCOBWebSocket:
self.first_event_u: Dict[str, int] = {} # Track first event U for synchronization
self.snapshot_in_progress: Dict[str, bool] = {} # Track snapshot initialization
# Rate limiting for message processing (Binance: max 5 messages per second)
# Message tracking (no artificial throttling; rely on Binance stream pacing)
self.last_message_time: Dict[str, datetime] = {}
self.min_message_interval = 0.2 # 200ms = 5 messages per second compliance
self.message_count: Dict[str, int] = {}
self.message_window_start: Dict[str, datetime] = {}
@ -150,7 +149,8 @@ class EnhancedCOBWebSocket:
# Configuration
self.max_depth = 1000 # Maximum depth for order book
self.update_speed = '1000ms' # Binance update speed - reduced for stability
# Prefer high-frequency depth stream. Binance supports @100ms diff depth
self.update_speed = '100ms'
# Timezone configuration
if self.timezone_offset == '+08:00':
@ -439,7 +439,7 @@ class EnhancedCOBWebSocket:
logger.info("Using UTC timezone for kline stream")
streams = [
f"{ws_symbol}@depth@1000ms", # Order book depth
f"{ws_symbol}@depth@{self.update_speed}", # Order book diff depth
kline_stream, # 1-second candlesticks (with timezone)
f"{ws_symbol}@ticker", # 24hr ticker with volume
f"{ws_symbol}@aggTrade" # Aggregated trades
@ -487,23 +487,14 @@ class EnhancedCOBWebSocket:
# Handle ping frames (though websockets library handles this automatically)
continue
# Rate limiting: Binance allows max 5 messages per second
now = datetime.now()
# Initialize rate limiting tracking
# Track receive rate for monitoring only
if symbol not in self.message_window_start:
self.message_window_start[symbol] = now
self.message_count[symbol] = 0
# Reset counter every second
if (now - self.message_window_start[symbol]).total_seconds() >= 1.0:
self.message_window_start[symbol] = now
self.message_count[symbol] = 0
# Check rate limit (5 messages per second)
if self.message_count[symbol] >= 5:
continue # Skip this message to comply with rate limit
self.message_count[symbol] += 1
self.last_message_time[symbol] = now

6
core/standardized_data_provider.py
View File

@ -168,7 +168,7 @@ class StandardizedDataProvider(DataProvider):
# Attach COB heatmap (visual+model optional input), fixed scope defaults
try:
times, prices, matrix = self.get_cob_heatmap_matrix(
times, prices, matrix, mids = self.get_cob_heatmap_matrix(
symbol=symbol,
seconds=300,
bucket_radius=10,
@ -177,6 +177,10 @@ class StandardizedDataProvider(DataProvider):
base_input.cob_heatmap_times = times
base_input.cob_heatmap_prices = prices
base_input.cob_heatmap_values = matrix
# We also store mids in market_microstructure for optional use
if not hasattr(base_input, 'market_microstructure') or base_input.market_microstructure is None:
base_input.market_microstructure = {}
base_input.market_microstructure['heatmap_mid_prices'] = mids
except Exception as _hm_ex:
logger.debug(f"COB heatmap not attached for {symbol}: {_hm_ex}")