gogo2/tests/test_training_integration.py

#!/usr/bin/env python3
"""
Test Training Integration with Dashboard

This script tests the enhanced dashboard's ability to:
1. Stream training data to CNN and DQN models
2. Display real-time training metrics and progress
3. Show model learning curves and performance
4. Integrate with the continuous training system
"""

import sys
import logging
import time
import asyncio
from datetime import datetime, timedelta
from pathlib import Path

# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def test_training_integration():
    """Test the training integration functionality"""
    try:
        print("="*60)
        print("TESTING TRAINING INTEGRATION WITH DASHBOARD")
        print("="*60)
        
        # Import dashboard
        from web.clean_dashboard import CleanTradingDashboard as TradingDashboard
        from core.data_provider import DataProvider
        from core.orchestrator import TradingOrchestrator
        
        # Create components
        data_provider = DataProvider()
        orchestrator = TradingOrchestrator(data_provider)
        dashboard = TradingDashboard(data_provider, orchestrator)
        
        print(f"✓ Dashboard created with training integration")
        print(f"✓ Continuous training active: {getattr(dashboard, 'training_active', False)}")
        
        # Test 1: Simulate tick data for training
        print("\n📊 TEST 1: Simulating Tick Data")
        print("-" * 40)
        
        # Add simulated tick data to cache
        base_price = 3500.0
        for i in range(1000):
            tick_data = {
                'timestamp': datetime.now() - timedelta(seconds=1000-i),
                'price': base_price + (i % 100) * 0.1,
                'volume': 100 + (i % 50),
                'side': 'buy' if i % 2 == 0 else 'sell'
            }
            dashboard.tick_cache.append(tick_data)
        
        print(f"✓ Added {len(dashboard.tick_cache)} ticks to cache")
        
        # Test 2: Prepare training data
        print("\n🔄 TEST 2: Preparing Training Data")
        print("-" * 40)
        
        training_data = dashboard._prepare_training_data()
        if training_data:
            print(f"✓ Training data prepared successfully")
            print(f"  - OHLCV bars: {len(training_data['ohlcv'])}")
            print(f"  - Features: {training_data['features']}")
            print(f"  - Symbol: {training_data['symbol']}")
        else:
            print("❌ Failed to prepare training data")
        
        # Test 3: Format data for CNN
        print("\n🧠 TEST 3: CNN Data Formatting")
        print("-" * 40)
        
        if training_data:
            cnn_data = dashboard._format_data_for_cnn(training_data)
            if cnn_data and 'sequences' in cnn_data:
                print(f"✓ CNN data formatted successfully")
                print(f"  - Sequences shape: {cnn_data['sequences'].shape}")
                print(f"  - Targets shape: {cnn_data['targets'].shape}")
                print(f"  - Sequence length: {cnn_data['sequence_length']}")
            else:
                print("❌ Failed to format CNN data")
        
        # Test 4: Format data for RL
        print("\n🤖 TEST 4: RL Data Formatting")
        print("-" * 40)
        
        if training_data:
            rl_experiences = dashboard._format_data_for_rl(training_data)
            if rl_experiences:
                print(f"✓ RL experiences formatted successfully")
                print(f"  - Number of experiences: {len(rl_experiences)}")
                print(f"  - Experience format: (state, action, reward, next_state, done)")
                print(f"  - Sample experience shapes: {[len(exp) for exp in rl_experiences[:3]]}")
            else:
                print("❌ Failed to format RL experiences")
        
        # Test 5: Send training data to models
        print("\n📤 TEST 5: Sending Training Data to Models")
        print("-" * 40)
        
        success = dashboard.send_training_data_to_models()
        print(f"✓ Training data sent: {success}")
        
        if hasattr(dashboard, 'training_stats'):
            stats = dashboard.training_stats
            print(f"  - Total training sessions: {stats.get('total_training_sessions', 0)}")
            print(f"  - CNN training count: {stats.get('cnn_training_count', 0)}")
            print(f"  - RL training count: {stats.get('rl_training_count', 0)}")
            print(f"  - Training data points: {stats.get('training_data_points', 0)}")
        
        # Test 6: Training metrics display
        print("\n📈 TEST 6: Training Metrics Display")
        print("-" * 40)
        
        training_metrics = dashboard._create_training_metrics()
        print(f"✓ Training metrics created: {len(training_metrics)} components")
        
        # Test 7: Model training status
        print("\n🔍 TEST 7: Model Training Status")
        print("-" * 40)
        
        training_status = dashboard._get_model_training_status()
        print(f"✓ Training status retrieved")
        print(f"  - CNN status: {training_status['cnn']['status']}")
        print(f"  - CNN accuracy: {training_status['cnn']['accuracy']:.1%}")
        print(f"  - RL status: {training_status['rl']['status']}")
        print(f"  - RL win rate: {training_status['rl']['win_rate']:.1%}")
        
        # Test 8: Training events log
        print("\n📝 TEST 8: Training Events Log")
        print("-" * 40)
        
        training_events = dashboard._get_recent_training_events()
        print(f"✓ Training events retrieved: {len(training_events)} events")
        
        # Test 9: Mini training chart
        print("\n📊 TEST 9: Mini Training Chart")
        print("-" * 40)
        
        try:
            training_chart = dashboard._create_mini_training_chart(training_status)
            print(f"✓ Mini training chart created")
            print(f"  - Chart type: {type(training_chart)}")
        except Exception as e:
            print(f"❌ Error creating training chart: {e}")
        
        # Test 10: Continuous training loop
        print("\n🔄 TEST 10: Continuous Training Loop")
        print("-" * 40)
        
        print(f"✓ Continuous training active: {getattr(dashboard, 'training_active', False)}")
        if hasattr(dashboard, 'training_thread'):
            print(f"✓ Training thread alive: {dashboard.training_thread.is_alive()}")
        
        # Test 11: Integration with existing continuous training system
        print("\n🔗 TEST 11: Integration with Continuous Training System")
        print("-" * 40)
        
        try:
            # Check if we can get tick cache for external training
            tick_cache = dashboard.get_tick_cache_for_training()
            print(f"✓ Tick cache accessible: {len(tick_cache)} ticks")
            
            # Check if we can get 1-second bars
            one_second_bars = dashboard.get_one_second_bars()
            print(f"✓ 1-second bars accessible: {len(one_second_bars)} bars")
            
        except Exception as e:
            print(f"❌ Error accessing training data: {e}")
        
        print("\n" + "="*60)
        print("TRAINING INTEGRATION TEST COMPLETED")
        print("="*60)
        
        # Summary
        print("\n📋 SUMMARY:")
        print(f"✓ Dashboard with training integration: WORKING")
        print(f"✓ Training data preparation: WORKING")
        print(f"✓ CNN data formatting: WORKING")
        print(f"✓ RL data formatting: WORKING")
        print(f"✓ Training metrics display: WORKING")
        print(f"✓ Continuous training: ACTIVE")
        print(f"✓ Model status tracking: WORKING")
        print(f"✓ Training events logging: WORKING")
        
        return True
        
    except Exception as e:
        logger.error(f"Training integration test failed: {e}")
        import traceback
        traceback.print_exc()
        return False

if __name__ == "__main__":
    success = test_training_integration()
    if success:
        print("\n🎉 All training integration tests passed!")
    else:
        print("\n❌ Some training integration tests failed!")
        sys.exit(1)