working charts again!!!
This commit is contained in:
153
test_chart_data.py
Normal file
153
test_chart_data.py
Normal file
@@ -0,0 +1,153 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test script to verify chart data loading functionality
|
||||
"""
|
||||
|
||||
import logging
|
||||
import sys
|
||||
import os
|
||||
|
||||
# Add the project root to the path
|
||||
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
from dataprovider_realtime import RealTimeChart, TickStorage, BinanceHistoricalData
|
||||
|
||||
# Set up logging
|
||||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def test_binance_data_fetch():
|
||||
"""Test fetching data from Binance API"""
|
||||
logger.info("Testing Binance historical data fetch...")
|
||||
|
||||
try:
|
||||
binance_data = BinanceHistoricalData()
|
||||
|
||||
# Test fetching 1m data for ETH/USDT
|
||||
df = binance_data.get_historical_candles("ETH/USDT", 60, 100)
|
||||
|
||||
if df is not None and not df.empty:
|
||||
logger.info(f"✅ Successfully fetched {len(df)} 1m candles")
|
||||
logger.info(f" Latest price: ${df.iloc[-1]['close']:.2f}")
|
||||
logger.info(f" Date range: {df.iloc[0]['timestamp']} to {df.iloc[-1]['timestamp']}")
|
||||
return True
|
||||
else:
|
||||
logger.error("❌ Failed to fetch Binance data")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"❌ Error fetching Binance data: {str(e)}")
|
||||
return False
|
||||
|
||||
def test_tick_storage():
|
||||
"""Test TickStorage data loading"""
|
||||
logger.info("Testing TickStorage data loading...")
|
||||
|
||||
try:
|
||||
# Create tick storage
|
||||
tick_storage = TickStorage("ETH/USDT", ["1s", "1m", "5m", "1h"])
|
||||
|
||||
# Load historical data
|
||||
success = tick_storage.load_historical_data("ETH/USDT", limit=100)
|
||||
|
||||
if success:
|
||||
logger.info("✅ TickStorage data loading successful")
|
||||
|
||||
# Check what we have
|
||||
for tf in ["1s", "1m", "5m", "1h"]:
|
||||
candles = tick_storage.get_candles(tf)
|
||||
logger.info(f" {tf}: {len(candles)} candles")
|
||||
|
||||
if candles:
|
||||
latest = candles[-1]
|
||||
logger.info(f" Latest {tf}: {latest['timestamp']} - ${latest['close']:.2f}")
|
||||
|
||||
return True
|
||||
else:
|
||||
logger.error("❌ TickStorage data loading failed")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"❌ Error in TickStorage: {str(e)}")
|
||||
import traceback
|
||||
logger.error(traceback.format_exc())
|
||||
return False
|
||||
|
||||
def test_chart_initialization():
|
||||
"""Test RealTimeChart initialization and data loading"""
|
||||
logger.info("Testing RealTimeChart initialization...")
|
||||
|
||||
try:
|
||||
# Create chart (without app to avoid GUI issues)
|
||||
chart = RealTimeChart(
|
||||
app=None,
|
||||
symbol="ETH/USDT",
|
||||
standalone=False
|
||||
)
|
||||
|
||||
# Test getting candles
|
||||
candles_1s = chart.get_candles(1) # 1 second
|
||||
candles_1m = chart.get_candles(60) # 1 minute
|
||||
|
||||
logger.info(f"✅ Chart initialized successfully")
|
||||
logger.info(f" 1s candles: {len(candles_1s)}")
|
||||
logger.info(f" 1m candles: {len(candles_1m)}")
|
||||
|
||||
if candles_1m:
|
||||
latest = candles_1m[-1]
|
||||
logger.info(f" Latest 1m candle: {latest['timestamp']} - ${latest['close']:.2f}")
|
||||
|
||||
return len(candles_1s) > 0 or len(candles_1m) > 0
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"❌ Error in chart initialization: {str(e)}")
|
||||
import traceback
|
||||
logger.error(traceback.format_exc())
|
||||
return False
|
||||
|
||||
def main():
|
||||
"""Run all tests"""
|
||||
logger.info("🧪 Starting chart data loading tests...")
|
||||
logger.info("=" * 60)
|
||||
|
||||
tests = [
|
||||
("Binance API fetch", test_binance_data_fetch),
|
||||
("TickStorage loading", test_tick_storage),
|
||||
("Chart initialization", test_chart_initialization)
|
||||
]
|
||||
|
||||
results = []
|
||||
for test_name, test_func in tests:
|
||||
logger.info(f"\n📋 Running test: {test_name}")
|
||||
logger.info("-" * 40)
|
||||
try:
|
||||
result = test_func()
|
||||
results.append((test_name, result))
|
||||
except Exception as e:
|
||||
logger.error(f"❌ Test {test_name} crashed: {str(e)}")
|
||||
results.append((test_name, False))
|
||||
|
||||
# Print summary
|
||||
logger.info("\n" + "=" * 60)
|
||||
logger.info("📊 TEST RESULTS SUMMARY")
|
||||
logger.info("=" * 60)
|
||||
|
||||
passed = 0
|
||||
for test_name, result in results:
|
||||
status = "✅ PASS" if result else "❌ FAIL"
|
||||
logger.info(f"{status}: {test_name}")
|
||||
if result:
|
||||
passed += 1
|
||||
|
||||
logger.info(f"\nPassed: {passed}/{len(results)} tests")
|
||||
|
||||
if passed == len(results):
|
||||
logger.info("🎉 All tests passed! Chart data loading is working correctly.")
|
||||
return True
|
||||
else:
|
||||
logger.warning(f"⚠️ {len(results) - passed} test(s) failed. Please check the issues above.")
|
||||
return False
|
||||
|
||||
if __name__ == "__main__":
|
||||
success = main()
|
||||
sys.exit(0 if success else 1)
|
||||
Reference in New Issue
Block a user