diff options
Diffstat (limited to 'services/strategy-engine/strategies/volume_profile_strategy.py')
| -rw-r--r-- | services/strategy-engine/strategies/volume_profile_strategy.py | 52 |
1 files changed, 47 insertions, 5 deletions
diff --git a/services/strategy-engine/strategies/volume_profile_strategy.py b/services/strategy-engine/strategies/volume_profile_strategy.py index 324f1c2..ef2ae14 100644 --- a/services/strategy-engine/strategies/volume_profile_strategy.py +++ b/services/strategy-engine/strategies/volume_profile_strategy.py @@ -56,7 +56,8 @@ class VolumeProfileStrategy(BaseStrategy): self._was_below_va = False self._was_above_va = False - def _compute_value_area(self) -> tuple[float, float, float] | None: + def _compute_value_area(self) -> tuple[float, float, float, list[float], list[float]] | None: + """Compute POC, VA low, VA high, HVN levels, LVN levels.""" data = list(self._candles) if len(data) < self._lookback_period: return None @@ -67,7 +68,7 @@ class VolumeProfileStrategy(BaseStrategy): min_price = prices.min() max_price = prices.max() if min_price == max_price: - return (float(min_price), float(min_price), float(max_price)) + return (float(min_price), float(min_price), float(max_price), [], []) bin_edges = np.linspace(min_price, max_price, self._num_bins + 1) vol_profile = np.zeros(self._num_bins) @@ -84,7 +85,7 @@ class VolumeProfileStrategy(BaseStrategy): # Value Area: expand from POC outward total_volume = vol_profile.sum() if total_volume == 0: - return (poc, float(bin_edges[0]), float(bin_edges[-1])) + return (poc, float(bin_edges[0]), float(bin_edges[-1]), [], []) target_volume = self._value_area_pct * total_volume accumulated = vol_profile[poc_idx] @@ -111,7 +112,20 @@ class VolumeProfileStrategy(BaseStrategy): va_low = float(bin_edges[low_idx]) va_high = float(bin_edges[high_idx + 1]) - return (poc, va_low, va_high) + # HVN/LVN detection + mean_vol = vol_profile.mean() + std_vol = vol_profile.std() + + hvn_levels: list[float] = [] + lvn_levels: list[float] = [] + for i in range(len(vol_profile)): + mid = float((bin_edges[i] + bin_edges[i + 1]) / 2) + if vol_profile[i] > mean_vol + std_vol: + hvn_levels.append(mid) + elif vol_profile[i] < mean_vol - 0.5 * std_vol and vol_profile[i] > 0: + lvn_levels.append(mid) + + return (poc, va_low, va_high, hvn_levels, lvn_levels) def on_candle(self, candle: Candle) -> Signal | None: self._update_filter_data(candle) @@ -123,13 +137,41 @@ class VolumeProfileStrategy(BaseStrategy): if result is None: return None - poc, va_low, va_high = result + poc, va_low, va_high, hvn_levels, lvn_levels = result if close < va_low: self._was_below_va = True if close > va_high: self._was_above_va = True + # HVN bounce signals (stronger than regular VA bounces) + for hvn in hvn_levels: + if abs(close - hvn) / hvn < 0.005: # Within 0.5% of HVN + if self._was_below_va and close >= va_low: + self._was_below_va = False + signal = Signal( + strategy=self.name, + symbol=candle.symbol, + side=OrderSide.BUY, + price=candle.close, + quantity=self._quantity, + conviction=0.85, + reason=f"Price near HVN {hvn:.2f}, bounced from below VA low {va_low:.2f} to {close:.2f}", + ) + return self._apply_filters(signal) + if self._was_above_va and close <= va_high: + self._was_above_va = False + signal = Signal( + strategy=self.name, + symbol=candle.symbol, + side=OrderSide.SELL, + price=candle.close, + quantity=self._quantity, + conviction=0.85, + reason=f"Price near HVN {hvn:.2f}, rejected from above VA high {va_high:.2f} to {close:.2f}", + ) + return self._apply_filters(signal) + # BUY: was below VA, price bounces back between va_low and poc if self._was_below_va and va_low <= close <= poc: self._was_below_va = False |