86 lines
3.0 KiB
Python
86 lines
3.0 KiB
Python
"""
|
|
Test script to verify data loader integration with DataProvider
|
|
"""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# Add parent directory to path
|
|
parent_dir = Path(__file__).parent.parent
|
|
sys.path.insert(0, str(parent_dir))
|
|
|
|
from core.data_provider import DataProvider
|
|
from ANNOTATE.core.data_loader import HistoricalDataLoader, TimeRangeManager
|
|
from datetime import datetime, timedelta
|
|
|
|
def test_data_loader():
|
|
"""Test the data loader"""
|
|
print("=" * 60)
|
|
print("Testing ANNOTATE Data Loader Integration")
|
|
print("=" * 60)
|
|
|
|
# Initialize DataProvider
|
|
print("\n1. Initializing DataProvider...")
|
|
data_provider = DataProvider()
|
|
print(f" ✓ DataProvider initialized")
|
|
print(f" - Symbols: {data_provider.symbols}")
|
|
print(f" - Timeframes: {data_provider.timeframes}")
|
|
|
|
# Initialize HistoricalDataLoader
|
|
print("\n2. Initializing HistoricalDataLoader...")
|
|
data_loader = HistoricalDataLoader(data_provider)
|
|
print(f" ✓ HistoricalDataLoader initialized")
|
|
|
|
# Test loading data for ETH/USDT
|
|
print("\n3. Testing data loading for ETH/USDT...")
|
|
symbol = 'ETH/USDT'
|
|
timeframes = ['1s', '1m', '1h', '1d']
|
|
|
|
for timeframe in timeframes:
|
|
df = data_loader.get_data(symbol, timeframe, limit=100)
|
|
if df is not None and not df.empty:
|
|
print(f" ✓ {timeframe}: Loaded {len(df)} candles")
|
|
print(f" Latest: {df.index[-1]} - Close: ${df['close'].iloc[-1]:.2f}")
|
|
else:
|
|
print(f" ✗ {timeframe}: No data available")
|
|
|
|
# Test multi-timeframe loading
|
|
print("\n4. Testing multi-timeframe loading...")
|
|
multi_data = data_loader.get_multi_timeframe_data(symbol, timeframes, limit=50)
|
|
print(f" ✓ Loaded data for {len(multi_data)} timeframes")
|
|
for tf, df in multi_data.items():
|
|
print(f" {tf}: {len(df)} candles")
|
|
|
|
# Test TimeRangeManager
|
|
print("\n5. Testing TimeRangeManager...")
|
|
time_manager = TimeRangeManager(data_loader)
|
|
|
|
center_time = datetime.now()
|
|
range_preset = '1d'
|
|
start_time, end_time = time_manager.calculate_time_range(center_time, range_preset)
|
|
|
|
print(f" ✓ Time range calculated for '{range_preset}':")
|
|
print(f" Start: {start_time}")
|
|
print(f" End: {end_time}")
|
|
|
|
increment = time_manager.get_navigation_increment(range_preset)
|
|
print(f" ✓ Navigation increment: {increment}")
|
|
|
|
# Test data boundaries
|
|
print("\n6. Testing data boundaries...")
|
|
earliest, latest = data_loader.get_data_boundaries(symbol, '1m')
|
|
if earliest and latest:
|
|
print(f" ✓ Data available from {earliest} to {latest}")
|
|
print(f" Total span: {latest - earliest}")
|
|
else:
|
|
print(f" ✗ Could not determine data boundaries")
|
|
|
|
print("\n" + "=" * 60)
|
|
print("✓ All tests completed successfully!")
|
|
print("=" * 60)
|
|
print("\nThe data loader is ready to use with the annotation UI.")
|
|
print("It uses the same DataProvider as training/inference systems.")
|
|
|
|
if __name__ == '__main__':
|
|
test_data_loader()
|