308 lines
12 KiB
Python
308 lines
12 KiB
Python
#!/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") |