gogo2/test_pivot_normalization_system.py
2025-05-31 00:33:07 +03:00

305 lines
12 KiB
Python

#!/usr/bin/env python3
"""
Test Pivot-Based Normalization System
This script tests the comprehensive pivot-based normalization system:
1. Monthly 1s data collection with pagination
2. Williams Market Structure pivot analysis
3. Pivot bounds extraction and caching
4. Pivot-based feature normalization
5. Integration with model training pipeline
"""
import sys
import os
import logging
from datetime import datetime, timedelta
# Add project root to path
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from core.data_provider import DataProvider
from core.config import get_config
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
def test_pivot_normalization_system():
"""Test the complete pivot-based normalization system"""
print("="*80)
print("TESTING PIVOT-BASED NORMALIZATION SYSTEM")
print("="*80)
# Initialize data provider
symbols = ['ETH/USDT'] # Test with ETH only
timeframes = ['1s']
logger.info("Initializing DataProvider with pivot-based normalization...")
data_provider = DataProvider(symbols=symbols, timeframes=timeframes)
# Test 1: Monthly Data Collection
print("\n" + "="*60)
print("TEST 1: MONTHLY 1S DATA COLLECTION")
print("="*60)
symbol = 'ETH/USDT'
try:
# This will trigger monthly data collection and pivot analysis
logger.info(f"Testing monthly data collection for {symbol}...")
monthly_data = data_provider._collect_monthly_1m_data(symbol)
if monthly_data is not None:
print(f"✅ Monthly data collection SUCCESS")
print(f" 📊 Collected {len(monthly_data):,} 1m candles")
print(f" 📅 Period: {monthly_data['timestamp'].min()} to {monthly_data['timestamp'].max()}")
print(f" 💰 Price range: ${monthly_data['low'].min():.2f} - ${monthly_data['high'].max():.2f}")
print(f" 📈 Volume range: {monthly_data['volume'].min():.2f} - {monthly_data['volume'].max():.2f}")
else:
print("❌ Monthly data collection FAILED")
return False
except Exception as e:
print(f"❌ Monthly data collection ERROR: {e}")
return False
# Test 2: Pivot Bounds Extraction
print("\n" + "="*60)
print("TEST 2: PIVOT BOUNDS EXTRACTION")
print("="*60)
try:
logger.info("Testing pivot bounds extraction...")
bounds = data_provider._extract_pivot_bounds_from_monthly_data(symbol, monthly_data)
if bounds is not None:
print(f"✅ Pivot bounds extraction SUCCESS")
print(f" 💰 Price bounds: ${bounds.price_min:.2f} - ${bounds.price_max:.2f}")
print(f" 📊 Volume bounds: {bounds.volume_min:.2f} - {bounds.volume_max:.2f}")
print(f" 🔸 Support levels: {len(bounds.pivot_support_levels)}")
print(f" 🔹 Resistance levels: {len(bounds.pivot_resistance_levels)}")
print(f" 📈 Candles analyzed: {bounds.total_candles_analyzed:,}")
print(f" ⏰ Created: {bounds.created_timestamp}")
# Store bounds for next tests
data_provider.pivot_bounds[symbol] = bounds
else:
print("❌ Pivot bounds extraction FAILED")
return False
except Exception as e:
print(f"❌ Pivot bounds extraction ERROR: {e}")
return False
# Test 3: Pivot Context Features
print("\n" + "="*60)
print("TEST 3: PIVOT CONTEXT FEATURES")
print("="*60)
try:
logger.info("Testing pivot context features...")
# Get recent data for testing
recent_data = data_provider.get_historical_data(symbol, '1m', limit=100)
if recent_data is not None and not recent_data.empty:
# Add pivot context features
with_pivot_features = data_provider._add_pivot_context_features(recent_data, symbol)
# Check if pivot features were added
pivot_features = [col for col in with_pivot_features.columns if 'pivot' in col]
if pivot_features:
print(f"✅ Pivot context features SUCCESS")
print(f" 🎯 Added features: {pivot_features}")
# Show sample values
latest_row = with_pivot_features.iloc[-1]
print(f" 📊 Latest values:")
for feature in pivot_features:
print(f" {feature}: {latest_row[feature]:.4f}")
else:
print("❌ No pivot context features added")
return False
else:
print("❌ Could not get recent data for testing")
return False
except Exception as e:
print(f"❌ Pivot context features ERROR: {e}")
return False
# Test 4: Pivot-Based Normalization
print("\n" + "="*60)
print("TEST 4: PIVOT-BASED NORMALIZATION")
print("="*60)
try:
logger.info("Testing pivot-based normalization...")
# Get data with technical indicators
data_with_indicators = data_provider.get_historical_data(symbol, '1m', limit=50)
if data_with_indicators is not None and not data_with_indicators.empty:
# Test traditional vs pivot normalization
traditional_norm = data_provider._normalize_features(data_with_indicators.tail(10))
pivot_norm = data_provider._normalize_features(data_with_indicators.tail(10), symbol=symbol)
print(f"✅ Pivot-based normalization SUCCESS")
print(f" 📊 Traditional normalization shape: {traditional_norm.shape}")
print(f" 🎯 Pivot normalization shape: {pivot_norm.shape}")
# Compare price normalization
if 'close' in pivot_norm.columns:
trad_close_range = traditional_norm['close'].max() - traditional_norm['close'].min()
pivot_close_range = pivot_norm['close'].max() - pivot_norm['close'].min()
print(f" 💰 Traditional close range: {trad_close_range:.6f}")
print(f" 🎯 Pivot close range: {pivot_close_range:.6f}")
# Pivot normalization should be better bounded
if 0 <= pivot_norm['close'].min() and pivot_norm['close'].max() <= 1:
print(f" ✅ Pivot normalization properly bounded [0,1]")
else:
print(f" ⚠️ Pivot normalization outside [0,1] bounds")
else:
print("❌ Could not get data for normalization testing")
return False
except Exception as e:
print(f"❌ Pivot-based normalization ERROR: {e}")
return False
# Test 5: Feature Matrix with Pivot Normalization
print("\n" + "="*60)
print("TEST 5: FEATURE MATRIX WITH PIVOT NORMALIZATION")
print("="*60)
try:
logger.info("Testing feature matrix with pivot normalization...")
# Create feature matrix using pivot normalization
feature_matrix = data_provider.get_feature_matrix(symbol, timeframes=['1m'], window_size=20)
if feature_matrix is not None:
print(f"✅ Feature matrix with pivot normalization SUCCESS")
print(f" 📊 Matrix shape: {feature_matrix.shape}")
print(f" 🎯 Data range: [{feature_matrix.min():.4f}, {feature_matrix.max():.4f}]")
print(f" 📈 Mean: {feature_matrix.mean():.4f}")
print(f" 📊 Std: {feature_matrix.std():.4f}")
# Check for proper normalization
if feature_matrix.min() >= -5 and feature_matrix.max() <= 5: # Reasonable bounds
print(f" ✅ Feature matrix reasonably bounded")
else:
print(f" ⚠️ Feature matrix may have extreme values")
else:
print("❌ Feature matrix creation FAILED")
return False
except Exception as e:
print(f"❌ Feature matrix ERROR: {e}")
return False
# Test 6: Caching System
print("\n" + "="*60)
print("TEST 6: CACHING SYSTEM")
print("="*60)
try:
logger.info("Testing caching system...")
# Test pivot bounds caching
original_bounds = data_provider.pivot_bounds[symbol]
data_provider._save_pivot_bounds_to_cache(symbol, original_bounds)
# Clear from memory and reload
del data_provider.pivot_bounds[symbol]
loaded_bounds = data_provider._load_pivot_bounds_from_cache(symbol)
if loaded_bounds is not None:
print(f"✅ Pivot bounds caching SUCCESS")
print(f" 💾 Original price range: ${original_bounds.price_min:.2f} - ${original_bounds.price_max:.2f}")
print(f" 💾 Loaded price range: ${loaded_bounds.price_min:.2f} - ${loaded_bounds.price_max:.2f}")
# Restore bounds
data_provider.pivot_bounds[symbol] = loaded_bounds
else:
print("❌ Pivot bounds caching FAILED")
return False
except Exception as e:
print(f"❌ Caching system ERROR: {e}")
return False
# Test 7: Public API Methods
print("\n" + "="*60)
print("TEST 7: PUBLIC API METHODS")
print("="*60)
try:
logger.info("Testing public API methods...")
# Test get_pivot_bounds
api_bounds = data_provider.get_pivot_bounds(symbol)
if api_bounds is not None:
print(f"✅ get_pivot_bounds() SUCCESS")
print(f" 📊 Returned bounds for {api_bounds.symbol}")
# Test get_pivot_normalized_features
test_data = data_provider.get_historical_data(symbol, '1m', limit=10)
if test_data is not None:
normalized_data = data_provider.get_pivot_normalized_features(symbol, test_data)
if normalized_data is not None:
print(f"✅ get_pivot_normalized_features() SUCCESS")
print(f" 📊 Normalized data shape: {normalized_data.shape}")
else:
print("❌ get_pivot_normalized_features() FAILED")
return False
except Exception as e:
print(f"❌ Public API methods ERROR: {e}")
return False
# Final Summary
print("\n" + "="*80)
print("🎉 PIVOT-BASED NORMALIZATION SYSTEM TEST COMPLETE")
print("="*80)
print("✅ All tests PASSED successfully!")
print("\n📋 System Features Verified:")
print(" ✅ Monthly 1s data collection with pagination")
print(" ✅ Williams Market Structure pivot analysis")
print(" ✅ Pivot bounds extraction and validation")
print(" ✅ Pivot context features generation")
print(" ✅ Pivot-based feature normalization")
print(" ✅ Feature matrix creation with pivot bounds")
print(" ✅ Comprehensive caching system")
print(" ✅ Public API methods")
print(f"\n🎯 Ready for model training with pivot-normalized features!")
return True
if __name__ == "__main__":
try:
success = test_pivot_normalization_system()
if success:
print("\n🚀 Pivot-based normalization system ready for production!")
sys.exit(0)
else:
print("\n❌ Pivot-based normalization system has issues!")
sys.exit(1)
except KeyboardInterrupt:
print("\n⏹️ Test interrupted by user")
sys.exit(1)
except Exception as e:
print(f"\n💥 Unexpected error: {e}")
import traceback
traceback.print_exc()
sys.exit(1)