#!/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()