#!/usr/bin/env python3 """ Enhanced multi-timeframe velocity calculator for CLP Scalper Hedger Provides configurable velocity detection with multiple timeframes and smoothing algorithms """ import logging import math from typing import Dict, List, Optional, Tuple from dataclasses import dataclass from velocity_config import VelocityConfig, VelocityTimeframe @dataclass class VelocityReading: """Single velocity reading with metadata""" timeframe: str velocity: float threshold: float timestamp: float is_extreme: bool weight: float @dataclass class VelocitySignal: """Combined velocity signal from all timeframes""" final_velocity: float confidence: float dominant_timeframe: str all_readings: List[VelocityReading] market_condition: str recommendation: str class EnhancedVelocityCalculator: """Enhanced velocity calculator with multi-timeframe support and configurable parameters""" def __init__(self, config: VelocityConfig): """Initialize with configuration""" self.config = config self.price_history: List[float] = [] self.velocity_history: Dict[str, List[float]] = {} self.ema_values: Dict[str, float] = {} self.logger = logging.getLogger(__name__) # Initialize velocity history for each timeframe if config.timeframes: for tf in config.timeframes: self.velocity_history[tf.name] = [] self.ema_values[tf.name] = 0.0 def update_price(self, price: float, timestamp: Optional[float] = None) -> VelocitySignal: """ Update price history and calculate velocity signal Args: price: Current price timestamp: Optional timestamp (defaults to current time) Returns: VelocitySignal with calculated velocities and recommendations """ import time if timestamp is None: timestamp = time.time() # Update price history self.price_history.append(price) if len(self.price_history) > self.config.history_length: self.price_history = self.price_history[-self.config.history_length:] # Calculate velocities for all timeframes readings = [] market_volatility = self._calculate_market_volatility() if self.config.timeframes and len(self.price_history) >= 2: for timeframe in self.config.timeframes: reading = self._calculate_timeframe_velocity(price, timeframe, timestamp, market_volatility) if reading: readings.append(reading) # Generate final signal signal = self._generate_velocity_signal(readings, market_volatility) self.logger.debug(f"Velocity signal: {signal.final_velocity*100:.3f}% " f"({signal.dominant_timeframe}, {signal.market_condition})") return signal def _calculate_timeframe_velocity(self, current_price: float, timeframe: VelocityTimeframe, timestamp: float, market_volatility: float) -> Optional[VelocityReading]: """Calculate velocity for a specific timeframe""" if len(self.price_history) < timeframe.periods + 1: return None # Get price from N periods ago price_n_ago = self.price_history[-(timeframe.periods + 1)] # Calculate velocity as percentage change per period total_change = (current_price - price_n_ago) / price_n_ago velocity = total_change / timeframe.periods # Apply cap to prevent extreme readings if abs(velocity) > self.config.max_velocity_cap: velocity = self.config.max_velocity_cap if velocity > 0 else -self.config.max_velocity_cap self.logger.warning(f"Velocity capped at {self.config.max_velocity_cap*100:.1f}% for {timeframe.name}") # Apply smoothing if enabled if self.config.use_ema_smoothing: velocity = self._apply_ema_smoothing(velocity, timeframe.name) # Update velocity history self.velocity_history[timeframe.name].append(velocity) if len(self.velocity_history[timeframe.name]) > 20: # Keep last 20 readings self.velocity_history[timeframe.name] = self.velocity_history[timeframe.name][-20:] # Get adjusted threshold based on market conditions adjusted_threshold = self.config.get_active_threshold(market_volatility) # Check if this is an extreme move is_extreme = abs(velocity) > self.config.extreme_move_threshold return VelocityReading( timeframe=timeframe.name, velocity=velocity, threshold=adjusted_threshold, timestamp=timestamp, is_extreme=is_extreme, weight=timeframe.weight ) def _apply_ema_smoothing(self, velocity: float, timeframe_name: str) -> float: """Apply EMA smoothing to velocity""" if self.ema_values[timeframe_name] == 0.0: # First reading self.ema_values[timeframe_name] = velocity return velocity # Apply EMA formula: EMA_new = (α * new_value) + ((1-α) * EMA_old) alpha = self.config.ema_alpha ema_new = (alpha * velocity) + ((1 - alpha) * self.ema_values[timeframe_name]) self.ema_values[timeframe_name] = ema_new return ema_new def _calculate_market_volatility(self) -> float: """Calculate current market volatility from recent price changes""" if len(self.price_history) < 10: return 0.001 # Default low volatility # Calculate volatility as standard deviation of recent price changes recent_prices = self.price_history[-10:] price_changes = [] for i in range(1, len(recent_prices)): change = abs(recent_prices[i] - recent_prices[i-1]) / recent_prices[i-1] price_changes.append(change) if not price_changes: return 0.001 # Simple volatility measure (average of recent changes) volatility = sum(price_changes) / len(price_changes) return volatility def _generate_velocity_signal(self, readings: List[VelocityReading], market_volatility: float) -> VelocitySignal: """Generate final velocity signal from all timeframe readings""" if not readings: return VelocitySignal( final_velocity=0.0, confidence=0.0, dominant_timeframe="none", all_readings=[], market_condition="insufficient_data", recommendation="hold" ) # Determine market condition if market_volatility < 0.001: market_condition = "low_volatility" elif market_volatility < 0.003: market_condition = "normal_volatility" else: market_condition = "high_volatility" # Find extreme readings (highest priority) extreme_readings = [r for r in readings if r.is_extreme] if extreme_readings: # Use the most extreme reading dominant = max(extreme_readings, key=lambda r: abs(r.velocity)) final_velocity = dominant.velocity confidence = 0.9 recommendation = "emergency_override" else: # Weighted average of all readings total_weight = sum(r.weight for r in readings) final_velocity = sum(r.velocity * r.weight for r in readings) / total_weight # Calculate confidence based on agreement between timeframes velocity_directions = [1 if r.velocity > 0 else -1 for r in readings] agreement = abs(sum(velocity_directions)) / len(velocity_directions) confidence = agreement * 0.7 # Max 0.7 for non-extreme moves # Determine recommendation dominant = max(readings, key=lambda r: abs(r.velocity)) if abs(final_velocity) > dominant.threshold: recommendation = "trigger_protection" else: recommendation = "normal_operation" return VelocitySignal( final_velocity=final_velocity, confidence=confidence, dominant_timeframe=dominant.timeframe, all_readings=readings, market_condition=market_condition, recommendation=recommendation ) def get_velocity_summary(self) -> Dict: """Get summary of current velocity calculations""" if not self.price_history: return {"status": "no_data"} summary = { "current_price": self.price_history[-1], "price_history_length": len(self.price_history), "market_volatility": self._calculate_market_volatility(), "timeframe_velocities": {} } for timeframe_name, velocities in self.velocity_history.items(): if velocities: summary["timeframe_velocities"][timeframe_name] = { "current": velocities[-1], "average": sum(velocities) / len(velocities), "count": len(velocities) } return summary class VelocityThresholdAnalyzer: """Analyze and recommend optimal velocity thresholds""" def __init__(self, calculator: EnhancedVelocityCalculator): self.calculator = calculator self.logger = logging.getLogger(__name__) def analyze_threshold_performance(self, test_data: List[float], thresholds: List[float]) -> Dict: """Test different thresholds against historical data""" results = {} for threshold in thresholds: triggers = 0 false_triggers = 0 max_velocity = 0.0 for i, price in enumerate(test_data): signal = self.calculator.update_price(price) if abs(signal.final_velocity) > threshold: triggers += 1 # Count as false trigger if no significant price movement follows if i + 5 < len(test_data): future_change = abs(test_data[i + 5] - price) / price if future_change < 0.001: # Less than 0.1% movement false_triggers += 1 max_velocity = max(max_velocity, abs(signal.final_velocity)) false_trigger_rate = (false_triggers / triggers * 100) if triggers > 0 else 0 results[threshold] = { "total_triggers": triggers, "false_triggers": false_triggers, "false_trigger_rate": false_trigger_rate, "max_velocity_seen": max_velocity, "efficiency": (triggers - false_triggers) / len(test_data) if triggers > 0 else 0 } # Find optimal threshold (highest efficiency with low false trigger rate) optimal = min(results.items(), key=lambda x: (x[1]["false_trigger_rate"], -x[1]["efficiency"])) return { "detailed_results": results, "optimal_threshold": optimal[0], "optimal_performance": optimal[1], "recommendation": self._generate_threshold_recommendation(results) } def _generate_threshold_recommendation(self, results: Dict) -> str: """Generate recommendations based on threshold analysis""" best_threshold = min(results.items(), key=lambda x: (x[1]["false_trigger_rate"], -x[1]["efficiency"])) threshold, performance = best_threshold if performance["false_trigger_rate"] < 20: return (f"Recommended threshold: {threshold*100:.3f}% " f"({performance['false_trigger_rate']:.1f}% false trigger rate)") else: return ("Consider increasing threshold to reduce false triggers. " f"Current best: {threshold*100:.3f}% with {performance['false_trigger_rate']:.1f}% false triggers")