333 lines
13 KiB
Python
333 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Data Stream Checker - Consumes Dashboard API
|
|
Checks stream status, gets OHLCV data, COB data, and generates snapshots via API.
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import requests
|
|
import json
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
def check_dashboard_status():
|
|
"""Check if dashboard is running and get basic info."""
|
|
try:
|
|
response = requests.get("http://127.0.0.1:8050/api/health", timeout=5)
|
|
return response.status_code == 200, response.json()
|
|
except:
|
|
return False, {}
|
|
|
|
def get_stream_status_from_api():
|
|
"""Get stream status from the dashboard API."""
|
|
try:
|
|
response = requests.get("http://127.0.0.1:8050/api/stream-status", timeout=10)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error getting stream status: {e}")
|
|
return None
|
|
|
|
def get_ohlcv_data_from_api(symbol='ETH/USDT', timeframe='1m', limit=300):
|
|
"""Get OHLCV data with indicators from the dashboard API."""
|
|
try:
|
|
url = f"http://127.0.0.1:8050/api/ohlcv-data"
|
|
params = {'symbol': symbol, 'timeframe': timeframe, 'limit': limit}
|
|
response = requests.get(url, params=params, timeout=10)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error getting OHLCV data: {e}")
|
|
return None
|
|
|
|
def get_cob_data_from_api(symbol='ETH/USDT', limit=300):
|
|
"""Get COB data with price buckets from the dashboard API."""
|
|
try:
|
|
url = f"http://127.0.0.1:8050/api/cob-data"
|
|
params = {'symbol': symbol, 'limit': limit}
|
|
response = requests.get(url, params=params, timeout=10)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error getting COB data: {e}")
|
|
return None
|
|
|
|
def create_snapshot_via_api():
|
|
"""Create a snapshot via the dashboard API."""
|
|
try:
|
|
response = requests.post("http://127.0.0.1:8050/api/snapshot", timeout=10)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
except Exception as e:
|
|
print(f"Error creating snapshot: {e}")
|
|
return None
|
|
|
|
def check_stream():
|
|
"""Check current stream status from dashboard API."""
|
|
print("=" * 60)
|
|
print("DATA STREAM STATUS CHECK")
|
|
print("=" * 60)
|
|
|
|
# Check dashboard health
|
|
dashboard_running, health_data = check_dashboard_status()
|
|
if not dashboard_running:
|
|
print("❌ Dashboard not running")
|
|
print("💡 Start dashboard first: python run_clean_dashboard.py")
|
|
return
|
|
|
|
print("✅ Dashboard is running")
|
|
print(f"📊 Health: {health_data.get('status', 'unknown')}")
|
|
|
|
# Get stream status
|
|
stream_data = get_stream_status_from_api()
|
|
if stream_data:
|
|
status = stream_data.get('status', {})
|
|
summary = stream_data.get('summary', {})
|
|
|
|
print(f"\n🔄 Stream Status:")
|
|
print(f" Connected: {status.get('connected', False)}")
|
|
print(f" Streaming: {status.get('streaming', False)}")
|
|
print(f" Total Samples: {summary.get('total_samples', 0)}")
|
|
print(f" Active Streams: {len(summary.get('active_streams', []))}")
|
|
|
|
if summary.get('active_streams'):
|
|
print(f" Active: {', '.join(summary['active_streams'])}")
|
|
|
|
print(f"\n📈 Buffer Sizes:")
|
|
buffers = status.get('buffers', {})
|
|
for stream, count in buffers.items():
|
|
status_icon = "🟢" if count > 0 else "🔴"
|
|
print(f" {status_icon} {stream}: {count}")
|
|
|
|
if summary.get('sample_data'):
|
|
print(f"\n📝 Latest Samples:")
|
|
for stream, sample in summary['sample_data'].items():
|
|
print(f" {stream}: {str(sample)[:100]}...")
|
|
else:
|
|
print("❌ Could not get stream status from API")
|
|
|
|
def show_ohlcv_data():
|
|
"""Show OHLCV data with indicators for all required timeframes and symbols."""
|
|
print("=" * 60)
|
|
print("OHLCV DATA WITH INDICATORS")
|
|
print("=" * 60)
|
|
|
|
# Check dashboard health
|
|
dashboard_running, _ = check_dashboard_status()
|
|
if not dashboard_running:
|
|
print("❌ Dashboard not running")
|
|
print("💡 Start dashboard first: python run_clean_dashboard.py")
|
|
return
|
|
|
|
# Check all required datasets for models
|
|
datasets = [
|
|
("ETH/USDT", "1m"),
|
|
("ETH/USDT", "1h"),
|
|
("ETH/USDT", "1d"),
|
|
("BTC/USDT", "1m")
|
|
]
|
|
|
|
print("📊 Checking all required datasets for model training:")
|
|
|
|
for symbol, timeframe in datasets:
|
|
print(f"\n📈 {symbol} {timeframe} Data:")
|
|
data = get_ohlcv_data_from_api(symbol, timeframe, 300)
|
|
|
|
if data and isinstance(data, dict) and 'data' in data:
|
|
ohlcv_data = data['data']
|
|
if ohlcv_data and len(ohlcv_data) > 0:
|
|
print(f" ✅ Records: {len(ohlcv_data)}")
|
|
|
|
latest = ohlcv_data[-1]
|
|
oldest = ohlcv_data[0]
|
|
print(f" 📅 Range: {oldest['timestamp'][:10]} to {latest['timestamp'][:10]}")
|
|
print(f" 💰 Latest Price: ${latest['close']:.2f}")
|
|
print(f" 📊 Volume: {latest['volume']:.2f}")
|
|
|
|
indicators = latest.get('indicators', {})
|
|
if indicators:
|
|
rsi = indicators.get('rsi')
|
|
macd = indicators.get('macd')
|
|
sma_20 = indicators.get('sma_20')
|
|
print(f" 📉 RSI: {rsi:.2f}" if rsi else " 📉 RSI: N/A")
|
|
print(f" 🔄 MACD: {macd:.4f}" if macd else " 🔄 MACD: N/A")
|
|
print(f" 📈 SMA20: ${sma_20:.2f}" if sma_20 else " 📈 SMA20: N/A")
|
|
|
|
# Check if we have enough data for training
|
|
if len(ohlcv_data) >= 300:
|
|
print(f" 🎯 Model Ready: {len(ohlcv_data)}/300 candles")
|
|
else:
|
|
print(f" ⚠️ Need More: {len(ohlcv_data)}/300 candles ({300-len(ohlcv_data)} missing)")
|
|
else:
|
|
print(f" ❌ Empty data array")
|
|
elif data and isinstance(data, list) and len(data) > 0:
|
|
# Direct array format
|
|
print(f" ✅ Records: {len(data)}")
|
|
latest = data[-1]
|
|
oldest = data[0]
|
|
print(f" 📅 Range: {oldest['timestamp'][:10]} to {latest['timestamp'][:10]}")
|
|
print(f" 💰 Latest Price: ${latest['close']:.2f}")
|
|
elif data:
|
|
print(f" ⚠️ Unexpected format: {type(data)}")
|
|
else:
|
|
print(f" ❌ No data available")
|
|
|
|
print(f"\n🎯 Expected: 300 candles per dataset (1200 total)")
|
|
|
|
def show_detailed_ohlcv(symbol="ETH/USDT", timeframe="1m"):
|
|
"""Show detailed OHLCV data for a specific symbol/timeframe."""
|
|
print("=" * 60)
|
|
print(f"DETAILED {symbol} {timeframe} DATA")
|
|
print("=" * 60)
|
|
|
|
# Check dashboard health
|
|
dashboard_running, _ = check_dashboard_status()
|
|
if not dashboard_running:
|
|
print("❌ Dashboard not running")
|
|
return
|
|
|
|
data = get_ohlcv_data_from_api(symbol, timeframe, 300)
|
|
|
|
if data and isinstance(data, dict) and 'data' in data:
|
|
ohlcv_data = data['data']
|
|
if ohlcv_data and len(ohlcv_data) > 0:
|
|
print(f"📈 Total candles loaded: {len(ohlcv_data)}")
|
|
|
|
if len(ohlcv_data) >= 2:
|
|
oldest = ohlcv_data[0]
|
|
latest = ohlcv_data[-1]
|
|
print(f"📅 Date range: {oldest['timestamp']} to {latest['timestamp']}")
|
|
|
|
# Calculate price statistics
|
|
closes = [item['close'] for item in ohlcv_data]
|
|
volumes = [item['volume'] for item in ohlcv_data]
|
|
|
|
print(f"💰 Price range: ${min(closes):.2f} - ${max(closes):.2f}")
|
|
print(f"📊 Average volume: {sum(volumes)/len(volumes):.2f}")
|
|
|
|
# Show sample data
|
|
print(f"\n🔍 First 3 candles:")
|
|
for i in range(min(3, len(ohlcv_data))):
|
|
candle = ohlcv_data[i]
|
|
ts = candle['timestamp'][:19] if len(candle['timestamp']) > 19 else candle['timestamp']
|
|
print(f" {ts} | ${candle['close']:.2f} | Vol:{candle['volume']:.2f}")
|
|
|
|
print(f"\n🔍 Last 3 candles:")
|
|
for i in range(max(0, len(ohlcv_data)-3), len(ohlcv_data)):
|
|
candle = ohlcv_data[i]
|
|
ts = candle['timestamp'][:19] if len(candle['timestamp']) > 19 else candle['timestamp']
|
|
print(f" {ts} | ${candle['close']:.2f} | Vol:{candle['volume']:.2f}")
|
|
|
|
# Model training readiness check
|
|
if len(ohlcv_data) >= 300:
|
|
print(f"\n✅ Model Training Ready: {len(ohlcv_data)}/300 candles loaded")
|
|
else:
|
|
print(f"\n⚠️ Insufficient Data: {len(ohlcv_data)}/300 candles (need {300-len(ohlcv_data)} more)")
|
|
else:
|
|
print("❌ Empty data array")
|
|
elif data and isinstance(data, list) and len(data) > 0:
|
|
# Direct array format
|
|
print(f"📈 Total candles loaded: {len(data)}")
|
|
# ... (same processing as above for array format)
|
|
else:
|
|
print(f"❌ No data returned: {type(data)}")
|
|
|
|
def show_cob_data():
|
|
"""Show COB data with price buckets."""
|
|
print("=" * 60)
|
|
print("COB DATA WITH PRICE BUCKETS")
|
|
print("=" * 60)
|
|
|
|
# Check dashboard health
|
|
dashboard_running, _ = check_dashboard_status()
|
|
if not dashboard_running:
|
|
print("❌ Dashboard not running")
|
|
print("💡 Start dashboard first: python run_clean_dashboard.py")
|
|
return
|
|
|
|
symbol = 'ETH/USDT'
|
|
print(f"\n📊 {symbol} COB Data:")
|
|
|
|
data = get_cob_data_from_api(symbol, 300)
|
|
if data and data.get('data'):
|
|
cob_data = data['data']
|
|
print(f" Records: {len(cob_data)}")
|
|
|
|
if cob_data:
|
|
latest = cob_data[-1]
|
|
print(f" Latest: {latest['timestamp']}")
|
|
print(f" Mid Price: ${latest['mid_price']:.2f}")
|
|
print(f" Spread: {latest['spread']:.4f}")
|
|
print(f" Imbalance: {latest['imbalance']:.4f}")
|
|
|
|
price_buckets = latest.get('price_buckets', {})
|
|
if price_buckets:
|
|
print(f" Price Buckets: {len(price_buckets)} ($1 increments)")
|
|
|
|
# Show some sample buckets
|
|
bucket_count = 0
|
|
for price, bucket in price_buckets.items():
|
|
if bucket['bid_volume'] > 0 or bucket['ask_volume'] > 0:
|
|
print(f" ${price}: Bid={bucket['bid_volume']:.2f} Ask={bucket['ask_volume']:.2f}")
|
|
bucket_count += 1
|
|
if bucket_count >= 5: # Show first 5 active buckets
|
|
break
|
|
else:
|
|
print(f" No COB data available")
|
|
|
|
def generate_snapshot():
|
|
"""Generate a snapshot via API."""
|
|
print("=" * 60)
|
|
print("GENERATING DATA SNAPSHOT")
|
|
print("=" * 60)
|
|
|
|
# Check dashboard health
|
|
dashboard_running, _ = check_dashboard_status()
|
|
if not dashboard_running:
|
|
print("❌ Dashboard not running")
|
|
print("💡 Start dashboard first: python run_clean_dashboard.py")
|
|
return
|
|
|
|
# Create snapshot via API
|
|
result = create_snapshot_via_api()
|
|
if result:
|
|
print(f"✅ Snapshot saved: {result.get('filepath', 'Unknown')}")
|
|
print(f"📅 Timestamp: {result.get('timestamp', 'Unknown')}")
|
|
else:
|
|
print("❌ Failed to create snapshot via API")
|
|
|
|
def main():
|
|
if len(sys.argv) < 2:
|
|
print("Usage:")
|
|
print(" python check_stream.py status # Check stream status")
|
|
print(" python check_stream.py ohlcv # Show all OHLCV datasets")
|
|
print(" python check_stream.py detail [symbol] [timeframe] # Show detailed data")
|
|
print(" python check_stream.py cob # Show COB data")
|
|
print(" python check_stream.py snapshot # Generate snapshot")
|
|
print("\nExamples:")
|
|
print(" python check_stream.py detail ETH/USDT 1h")
|
|
print(" python check_stream.py detail BTC/USDT 1m")
|
|
return
|
|
|
|
command = sys.argv[1].lower()
|
|
|
|
if command == "status":
|
|
check_stream()
|
|
elif command == "ohlcv":
|
|
show_ohlcv_data()
|
|
elif command == "detail":
|
|
symbol = sys.argv[2] if len(sys.argv) > 2 else "ETH/USDT"
|
|
timeframe = sys.argv[3] if len(sys.argv) > 3 else "1m"
|
|
show_detailed_ohlcv(symbol, timeframe)
|
|
elif command == "cob":
|
|
show_cob_data()
|
|
elif command == "snapshot":
|
|
generate_snapshot()
|
|
else:
|
|
print(f"Unknown command: {command}")
|
|
print("Available commands: status, ohlcv, detail, cob, snapshot")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|