feat(strategy): add Volume Profile HVN/LVN and Combined adaptive weighting

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