128 lines
4.9 KiB
Python
128 lines
4.9 KiB
Python
"""
|
|
Test script for StandardizedDataProvider
|
|
|
|
This script tests the standardized BaseDataInput functionality
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
import logging
|
|
from datetime import datetime
|
|
from core.standardized_data_provider import StandardizedDataProvider
|
|
from core.data_models import create_model_output
|
|
|
|
# Set up logging
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def test_standardized_data_provider():
|
|
"""Test the StandardizedDataProvider functionality"""
|
|
|
|
print("Testing StandardizedDataProvider...")
|
|
|
|
# Initialize the provider
|
|
symbols = ['ETH/USDT', 'BTC/USDT']
|
|
timeframes = ['1s', '1m', '1h', '1d']
|
|
|
|
provider = StandardizedDataProvider(symbols=symbols, timeframes=timeframes)
|
|
|
|
# Test getting BaseDataInput
|
|
print("\n1. Testing BaseDataInput creation...")
|
|
base_input = provider.get_base_data_input('ETH/USDT')
|
|
|
|
if base_input is None:
|
|
print("❌ BaseDataInput is None - this is expected if no historical data is available")
|
|
print(" The provider needs real market data to create BaseDataInput")
|
|
|
|
# Test with mock data
|
|
print("\n2. Testing data structures...")
|
|
|
|
# Test ModelOutput creation
|
|
model_output = create_model_output(
|
|
model_type='cnn',
|
|
model_name='test_cnn',
|
|
symbol='ETH/USDT',
|
|
action='BUY',
|
|
confidence=0.75,
|
|
metadata={'test': True}
|
|
)
|
|
|
|
print(f"✅ Created ModelOutput: {model_output.model_type} - {model_output.predictions['action']} ({model_output.confidence})")
|
|
|
|
# Test storing model output
|
|
provider.store_model_output(model_output)
|
|
stored_outputs = provider.get_model_outputs('ETH/USDT')
|
|
|
|
if 'test_cnn' in stored_outputs:
|
|
print("✅ Model output storage and retrieval working")
|
|
else:
|
|
print("❌ Model output storage failed")
|
|
|
|
else:
|
|
print("✅ BaseDataInput created successfully!")
|
|
print(f" Symbol: {base_input.symbol}")
|
|
print(f" Timestamp: {base_input.timestamp}")
|
|
print(f" OHLCV 1s frames: {len(base_input.ohlcv_1s)}")
|
|
print(f" OHLCV 1m frames: {len(base_input.ohlcv_1m)}")
|
|
print(f" OHLCV 1h frames: {len(base_input.ohlcv_1h)}")
|
|
print(f" OHLCV 1d frames: {len(base_input.ohlcv_1d)}")
|
|
print(f" BTC 1s frames: {len(base_input.btc_ohlcv_1s)}")
|
|
print(f" COB data available: {base_input.cob_data is not None}")
|
|
print(f" Technical indicators: {len(base_input.technical_indicators)}")
|
|
print(f" Pivot points: {len(base_input.pivot_points)}")
|
|
print(f" Last predictions: {len(base_input.last_predictions)}")
|
|
|
|
# Test feature vector creation
|
|
try:
|
|
feature_vector = base_input.get_feature_vector()
|
|
print(f"✅ Feature vector created: shape {feature_vector.shape}")
|
|
except Exception as e:
|
|
print(f"❌ Feature vector creation failed: {e}")
|
|
|
|
# Test validation
|
|
is_valid = base_input.validate()
|
|
print(f"✅ BaseDataInput validation: {'PASSED' if is_valid else 'FAILED'}")
|
|
|
|
print("\n3. Testing data provider capabilities...")
|
|
|
|
# Test historical data fetching
|
|
try:
|
|
eth_data = provider.get_historical_data('ETH/USDT', '1h', 10)
|
|
if eth_data is not None and not eth_data.empty:
|
|
print(f"✅ Historical data available: {len(eth_data)} bars for ETH/USDT 1h")
|
|
else:
|
|
print("⚠️ No historical data available - this is normal if APIs are not accessible")
|
|
except Exception as e:
|
|
print(f"⚠️ Historical data fetch error: {e}")
|
|
|
|
print("\n4. Testing COB data functionality...")
|
|
|
|
# Test COB data creation
|
|
try:
|
|
# Set a mock current price for testing
|
|
provider.current_prices['ETHUSDT'] = 3000.0
|
|
cob_data = provider._get_cob_data('ETH/USDT', datetime.now())
|
|
|
|
if cob_data:
|
|
print(f"✅ COB data created successfully")
|
|
print(f" Current price: ${cob_data.current_price}")
|
|
print(f" Bucket size: ${cob_data.bucket_size}")
|
|
print(f" Price buckets: {len(cob_data.price_buckets)}")
|
|
print(f" MA 1s imbalance: {len(cob_data.ma_1s_imbalance)} buckets")
|
|
print(f" MA 5s imbalance: {len(cob_data.ma_5s_imbalance)} buckets")
|
|
else:
|
|
print("⚠️ COB data creation returned None")
|
|
except Exception as e:
|
|
print(f"❌ COB data creation error: {e}")
|
|
|
|
print("\n✅ StandardizedDataProvider test completed!")
|
|
print("\nNext steps:")
|
|
print("1. Integrate with real market data APIs")
|
|
print("2. Connect to actual COB provider")
|
|
print("3. Test with live data streams")
|
|
print("4. Integrate with model training pipelines")
|
|
|
|
if __name__ == "__main__":
|
|
test_standardized_data_provider() |