From 828682de5904c8c1d05664a961f7931ebe60fabd Mon Sep 17 00:00:00 2001
From: TheSiahxyz <164138827+TheSiahxyz@users.noreply.github.com>
Date: Thu, 2 Apr 2026 09:17:54 +0900
Subject: feat(strategy): add Volume Profile HVN/LVN and Combined adaptive
 weighting

---
 .../tests/test_combined_strategy.py                | 57 ++++++++++++++++++++++
 .../tests/test_volume_profile_strategy.py          | 53 ++++++++++++++++++++
 2 files changed, 110 insertions(+)

(limited to 'services/strategy-engine/tests')

diff --git a/services/strategy-engine/tests/test_combined_strategy.py b/services/strategy-engine/tests/test_combined_strategy.py
index 3408a89..20a572e 100644
--- a/services/strategy-engine/tests/test_combined_strategy.py
+++ b/services/strategy-engine/tests/test_combined_strategy.py
@@ -167,3 +167,60 @@ def test_combined_invalid_weight():
     c.configure({})
     with pytest.raises(ValueError):
         c.add_strategy(AlwaysBuyStrategy(), weight=-1.0)
+
+
+def test_combined_record_result():
+    """Verify trade history tracking works correctly."""
+    c = CombinedStrategy()
+    c.configure({"adaptive_weights": True, "history_window": 5})
+
+    c.record_result("test_strat", True)
+    c.record_result("test_strat", False)
+    c.record_result("test_strat", True)
+
+    assert len(c._trade_history["test_strat"]) == 3
+    assert c._trade_history["test_strat"] == [True, False, True]
+
+    # Fill beyond window size to test trimming
+    for _ in range(5):
+        c.record_result("test_strat", False)
+
+    assert len(c._trade_history["test_strat"]) == 5  # Trimmed to history_window
+
+
+def test_combined_adaptive_weight_increases_for_winners():
+    """Strategy with high win rate gets higher effective weight."""
+    c = CombinedStrategy()
+    c.configure({"threshold": 0.3, "adaptive_weights": True, "history_window": 20})
+    c.add_strategy(AlwaysBuyStrategy(), weight=1.0)
+
+    # Record high win rate for always_buy (80% wins)
+    for _ in range(8):
+        c.record_result("always_buy", True)
+    for _ in range(2):
+        c.record_result("always_buy", False)
+
+    # Adaptive weight should be > base weight (1.0)
+    adaptive_w = c._get_adaptive_weight("always_buy", 1.0)
+    assert adaptive_w > 1.0
+    # 80% win rate -> scale = 0.5 + 0.8 = 1.3 -> weight = 1.3
+    assert abs(adaptive_w - 1.3) < 0.01
+
+
+def test_combined_adaptive_weight_decreases_for_losers():
+    """Strategy with low win rate gets lower effective weight."""
+    c = CombinedStrategy()
+    c.configure({"threshold": 0.3, "adaptive_weights": True, "history_window": 20})
+    c.add_strategy(AlwaysBuyStrategy(), weight=1.0)
+
+    # Record low win rate for always_buy (20% wins)
+    for _ in range(2):
+        c.record_result("always_buy", True)
+    for _ in range(8):
+        c.record_result("always_buy", False)
+
+    # Adaptive weight should be < base weight (1.0)
+    adaptive_w = c._get_adaptive_weight("always_buy", 1.0)
+    assert adaptive_w < 1.0
+    # 20% win rate -> scale = 0.5 + 0.2 = 0.7 -> weight = 0.7
+    assert abs(adaptive_w - 0.7) < 0.01
diff --git a/services/strategy-engine/tests/test_volume_profile_strategy.py b/services/strategy-engine/tests/test_volume_profile_strategy.py
index 71f0eca..f40261c 100644
--- a/services/strategy-engine/tests/test_volume_profile_strategy.py
+++ b/services/strategy-engine/tests/test_volume_profile_strategy.py
@@ -125,3 +125,56 @@ def test_volume_profile_reset_clears_state():
     # After reset, should not have enough data
     result = strategy.on_candle(make_candle(100.0, 10.0))
     assert result is None
+
+
+def test_volume_profile_hvn_detection():
+    """Feed clustered volume at specific price levels to produce HVN nodes."""
+    strategy = VolumeProfileStrategy()
+    strategy.configure({"lookback_period": 20, "num_bins": 10, "value_area_pct": 0.7})
+
+    # Create a profile with very high volume at price ~100 and low volume elsewhere
+    # Prices range from 90 to 110, heavy volume concentrated at 100
+    candles_data = []
+    # Low volume at extremes
+    for p in [90, 91, 92, 109, 110]:
+        candles_data.append((p, 1.0))
+    # Very high volume around 100
+    for _ in range(15):
+        candles_data.append((100, 100.0))
+
+    for price, vol in candles_data:
+        strategy.on_candle(make_candle(price, vol))
+
+    # Access the internal method to verify HVN detection
+    result = strategy._compute_value_area()
+    assert result is not None
+    poc, va_low, va_high, hvn_levels, lvn_levels = result
+
+    # The bin containing price ~100 should have very high volume -> HVN
+    assert len(hvn_levels) > 0
+    # At least one HVN should be near 100
+    assert any(abs(h - 100) < 5 for h in hvn_levels)
+
+
+def test_volume_profile_reset_thorough():
+    """Verify all state is cleared on reset."""
+    strategy = VolumeProfileStrategy()
+    strategy.configure({"lookback_period": 10, "num_bins": 5})
+
+    # Build up state
+    for _ in range(10):
+        strategy.on_candle(make_candle(100.0, 10.0))
+    # Set below/above VA flags
+    strategy.on_candle(make_candle(50.0, 1.0))  # below VA
+    strategy.on_candle(make_candle(200.0, 1.0))  # above VA
+
+    strategy.reset()
+
+    # Verify all state cleared
+    assert len(strategy._candles) == 0
+    assert strategy._was_below_va is False
+    assert strategy._was_above_va is False
+
+    # Should not produce signal since no data
+    result = strategy.on_candle(make_candle(100.0, 10.0))
+    assert result is None
-- 
cgit v1.2.3