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

wip cob test

Browse Source
This commit is contained in:
Dobromir Popov
2025-07-23 22:56:28 +03:00
parent 4765b1b1e1
commit 8ba52640bd
1 changed files with 25 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
tests/cob/test_cob_data_stability.py
View File

@ -59,6 +59,18 @@ class COBStabilityTester:
def _cob_data_callback(self, symbol: str, cob_data: dict):
"""Callback function to receive COB data from the DataProvider."""
# Debug: Log first few callbacks to see what symbols we're getting
if self.cob_data_received < 5:
logger.info(f"DEBUG: Received COB data for symbol '{symbol}' (target: '{self.symbol}')")
# Filter to only our requested symbol - handle both formats (ETH/USDT and ETHUSDT)
normalized_symbol = symbol.replace('/', '')
normalized_target = self.symbol.replace('/', '')
if normalized_symbol != normalized_target:
if self.cob_data_received < 5:
logger.info(f"DEBUG: Skipping symbol '{symbol}' (normalized: '{normalized_symbol}' vs target: '{normalized_target}')")
return
self.cob_data_received += 1
self.latest_cob_data[symbol] = cob_data
@ -168,11 +180,22 @@ class COBStabilityTester:
timestamp = snapshot['timestamp']
for side in ['bids', 'asks']:
for order in snapshot[side]:
bucketed_price = round(order['price'] / self.price_granularity) * self.price_granularity
# Handle both dict and list formats
if isinstance(order, dict):
price = order['price']
size = order['size']
elif isinstance(order, (list, tuple)) and len(order) >= 2:
price = float(order[0])
size = float(order[1])
else:
logger.warning(f"Unknown order format: {order}")
continue
bucketed_price = round(price / self.price_granularity) * self.price_granularity
heatmap_data.append({
'time': timestamp,
'price': bucketed_price,
'size': order['size'],
'size': size,
'side': side
})