43 lines
1.6 KiB
Python
43 lines
1.6 KiB
Python
import logging
|
|
|
|
def get_price_momentum_pct(self, current_price):
|
|
"""Calculate price momentum percentage over last 5 intervals"""
|
|
if not hasattr(self, 'price_momentum_history') or len(self.price_momentum_history) < 2:
|
|
return 0.0
|
|
|
|
recent_prices = self.price_momentum_history[-5:] # Last 5 prices
|
|
if len(recent_prices) < 2:
|
|
return 0.0
|
|
|
|
# Calculate momentum as percentage change
|
|
oldest_price = recent_prices[0]
|
|
momentum_pct = (current_price - oldest_price) / oldest_price
|
|
return momentum_pct
|
|
|
|
def get_dynamic_price_buffer(self):
|
|
"""Calculate dynamic price buffer based on market conditions"""
|
|
# These constants should be defined in the main module
|
|
try:
|
|
PRICE_BUFFER_PCT = 0.0015
|
|
MOMENTUM_ADJUSTMENT_ENABLED = True
|
|
|
|
if not MOMENTUM_ADJUSTMENT_ENABLED:
|
|
return PRICE_BUFFER_PCT
|
|
|
|
current_price = self.last_price if hasattr(self, 'last_price') and self.last_price else 0
|
|
momentum_pct = get_price_momentum_pct(self, current_price)
|
|
|
|
base_buffer = PRICE_BUFFER_PCT
|
|
|
|
# Adjust buffer based on momentum and position direction
|
|
momentum_adjustment = abs(momentum_pct) * 0.3 # 30% of momentum as adjustment
|
|
dynamic_buffer = base_buffer + momentum_adjustment
|
|
|
|
# Cap the maximum buffer to prevent excessive thresholds
|
|
max_buffer = base_buffer * 3.0
|
|
dynamic_buffer = min(dynamic_buffer, max_buffer)
|
|
|
|
return dynamic_buffer
|
|
except Exception as e:
|
|
logging.error(f"Error calculating dynamic buffer: {e}")
|
|
return 0.0015 # Return default buffer on error |