hyper/clp_auto_hedger/test_enhanced_velocity.py

#!/usr/bin/env python3
"""
Enhanced test script for multi-timeframe velocity calculation with configurable thresholds
Demonstrates the new EnhancedVelocityCalculator capabilities
"""

import time
import random
import logging
from enhanced_velocity_calculator import EnhancedVelocityCalculator, VelocityThresholdAnalyzer
from velocity_config import VelocityConfig, create_default_config, VelocityTimeframe

# Set up logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)


def create_test_scenarios():
    """Create different market scenarios for testing"""
    base_price = 3000.0
    
    scenarios = {
        "Normal Trading": {
            "duration": 20,
            "noise_level": 0.0002,  # 0.02% noise
            "trend": 0.0,
            "description": "Normal market conditions with small random fluctuations"
        },
        "Noisy Market": {
            "duration": 20,
            "noise_level": 0.0008,  # 0.08% noise
            "trend": 0.0,
            "description": "High volatility with large random movements"
        },
        "Sharp Flash Crash": {
            "duration": 10,
            "noise_level": 0.0001,
            "trend": -0.015,  # 1.5% downward over duration
            "description": "Sudden sharp price drop (emergency scenario)"
        },
        "Sustained Uptrend": {
            "duration": 30,
            "noise_level": 0.0003,
            "trend": 0.002,  # 0.2% upward per interval
            "description": "Gradual sustained upward movement"
        },
        "Whale Manipulation": {
            "duration": 15,
            "noise_level": 0.0005,
            "spike_magnitude": 0.008,  # 0.8% sudden spike
            "spike_timing": 8,
            "description": "Large player creates artificial spike"
        }
    }
    
    return base_price, scenarios


def test_enhanced_velocity_calculation():
    """Test the enhanced velocity calculator with different scenarios"""
    print("=== Enhanced Multi-Timeframe Velocity Calculator Demo ===\n")
    
    # Create enhanced configuration
    config = create_default_config()
    calculator = EnhancedVelocityCalculator(config)
    
    base_price, scenarios = create_test_scenarios()
    
    for scenario_name, params in scenarios.items():
        print(f"Scenario: {scenario_name}")
        print(f"Description: {params['description']}")
        print("-" * 60)
        
        current_price = base_price
        total_triggers = 0
        emergency_overrides = 0
        
        for i in range(params["duration"]):
            # Generate price movement
            noise = random.uniform(-params["noise_level"], params["noise_level"])
            trend_component = params.get("trend", 0)
            
            # Handle special spike scenario
            if "spike_magnitude" in params and i == params["spike_timing"]:
                price_change = params["spike_magnitude"]
                print(f"  *** SPIKE at second {i+1}!")
            else:
                price_change = noise + trend_component
            
            # Apply price change
            current_price = current_price * (1 + price_change)
            
            # Calculate enhanced velocity signal
            signal = calculator.update_price(current_price)
            
            # Check for triggers
            if signal.recommendation in ["trigger_protection", "emergency_override"]:
                total_triggers += 1
                if signal.recommendation == "emergency_override":
                    emergency_overrides += 1
                
                trigger_type = "EMERGENCY" if signal.recommendation == "emergency_override" else "PROTECTION"
                print(f"  Second {i+1:2d}: ${current_price:7.2f} | "
                      f"Vel: {signal.final_velocity*100:+6.3f}% ({signal.dominant_timeframe}) | "
                      f"{trigger_type}")
            elif abs(signal.final_velocity) > 0.0001:  # Show interesting movements
                print(f"  Second {i+1:2d}: ${current_price:7.2f} | "
                      f"Vel: {signal.final_velocity*100:+6.3f}% ({signal.dominant_timeframe}) | "
                      f"Conf: {signal.confidence:.2f} | {signal.market_condition}")
            
            time.sleep(0.05)  # Small delay for readability
        
        print(f"\nResults for {scenario_name}:")
        print(f"  Total velocity triggers: {total_triggers}")
        print(f"  Emergency overrides: {emergency_overrides}")
        print(f"  Final price: ${current_price:.2f} ({((current_price/base_price)-1)*100:+.2f}%)")
        
        # Get velocity summary
        summary = calculator.get_velocity_summary()
        print(f"  Market volatility: {summary['market_volatility']*100:.3f}%")
        
        print("\n" + "="*70 + "\n")


def test_threshold_optimization():
    """Test threshold optimization with historical data"""
    print("=== Threshold Optimization Analysis ===\n")
    
    # Generate synthetic historical data
    base_price = 3000.0
    historical_data = []
    current_price = base_price
    
    # Mix of different market conditions
    for _ in range(100):
        # Randomly choose market condition
        condition = random.choice(["normal", "volatile", "flash_crash", "trend"])
        
        if condition == "normal":
            change = random.uniform(-0.0002, 0.0002)
        elif condition == "volatile":
            change = random.uniform(-0.0008, 0.0008)
        elif condition == "flash_crash":
            change = random.uniform(-0.01, -0.001)
        else:  # trend
            change = random.uniform(0.0001, 0.0005)
        
        current_price = current_price * (1 + change)
        historical_data.append(current_price)
    
    # Test different threshold configurations
    configs = {
        "Conservative": create_default_config().conservative(),
        "Normal": create_default_config(),
        "Aggressive": create_default_config().aggressive()
    }
    
    thresholds_to_test = [0.0003, 0.0005, 0.0008, 0.001, 0.0015, 0.002]
    
    for config_name, config in configs.items():
        print(f"Testing {config_name} Configuration:")
        print(f"Normal threshold: {config.normal_threshold*100:.3f}%")
        
        calculator = EnhancedVelocityCalculator(config)
        analyzer = VelocityThresholdAnalyzer(calculator)
        
        # Reset calculator for clean test
        calculator.price_history = []
        for tf_name in calculator.velocity_history:
            calculator.velocity_history[tf_name] = []
        
        results = analyzer.analyze_threshold_performance(historical_data, thresholds_to_test)
        
        print(f"Optimal threshold: {results['optimal_threshold']*100:.3f}%")
        print(f"Performance: {results['optimal_performance']}")
        print(f"Recommendation: {results['recommendation']}\n")


def test_different_timeframe_configs():
    """Test different timeframe configurations"""
    print("=== Timeframe Configuration Comparison ===\n")
    
    # Custom timeframe configurations
    quick_response_config = create_default_config()
    quick_response_config.timeframes = [
        VelocityTimeframe("1s", 1, 0.6, 0.002, "Emergency detection"),
        VelocityTimeframe("3s", 3, 0.3, 0.001, "Quick response"),
        VelocityTimeframe("10s", 10, 0.1, 0.0005, "Trend confirmation")
    ]
    
    smooth_averaging_config = create_default_config()
    smooth_averaging_config.timeframes = [
        VelocityTimeframe("5s", 5, 0.3, 0.0008, "Short-term smoothing"),
        VelocityTimeframe("15s", 15, 0.4, 0.0005, "Medium-term smoothing"),
        VelocityTimeframe("30s", 30, 0.3, 0.0003, "Long-term smoothing")
    ]
    
    configs = {
        "Quick Response": quick_response_config,
        "Smooth Averaging": smooth_averaging_config,
        "Default Balanced": create_default_config()
    }
    
    # Test with flash crash scenario
    base_price = 3000.0
    current_price = base_price
    
    for config_name, config in configs.items():
        calculator = EnhancedVelocityCalculator(config)
        print(f"Testing {config_name} Configuration:")
        
        # Simulate flash crash
        for i in range(10):
            if i == 3:  # Flash crash at second 4
                price_change = -0.01  # 1% drop
            elif i >= 4 and i <= 6:  # Continued drop
                price_change = -0.003
            else:
                price_change = random.uniform(-0.0002, 0.0002)
            
            current_price = current_price * (1 + price_change)
            signal = calculator.update_price(current_price)
            
            if signal.recommendation in ["trigger_protection", "emergency_override"]:
                trigger_time = i + 1
                trigger_velocity = signal.final_velocity * 100
                trigger_timeframe = signal.dominant_timeframe
                print(f"  *** Trigger at second {trigger_time}: {trigger_velocity:+.3f}% ({trigger_timeframe})")
                break
        else:
            print("  No trigger detected")
        
        print()


def main():
    """Run all enhanced velocity calculation tests"""
    print("Enhanced Multi-Timeframe Velocity Calculator Testing\n")
    print("="*70)
    
    test_enhanced_velocity_calculation()
    test_threshold_optimization()
    test_different_timeframe_configs()
    
    print("KEY Benefits of Enhanced Velocity Calculator:")
    print("  • Configurable multi-timeframe analysis")
    print("  • Market-adaptive thresholds")
    print("  • EMA smoothing for noise reduction")
    print("  • Confidence-based decision making")
    print("  • Comprehensive performance analysis")
    print("  • Flexible configuration for different risk profiles")
    print("\nThe enhanced system is ready for production deployment!")


if __name__ == "__main__":
    main()