feat(strategy): add Grid trend guard and Bollinger squeeze detection

1 files changed, 71 insertions, 1 deletions

diff --git a/services/strategy-engine/tests/test_bollinger_strategy.py b/services/strategy-engine/tests/test_bollinger_strategy.py
index 348a9e0..473d9b4 100644
--- a/services/strategy-engine/tests/test_bollinger_strategy.py
+++ b/services/strategy-engine/tests/test_bollinger_strategy.py
@@ -23,7 +23,7 @@ def make_candle(close: float) -> Candle:
 
 def _make_strategy() -> BollingerStrategy:
     s = BollingerStrategy()
-    s.configure({"period": 5, "num_std": 1.0, "min_bandwidth": 0.0})
+    s.configure({"period": 5, "num_std": 1.0, "min_bandwidth": 0.0, "squeeze_threshold": 0.0})
     return s
 
 
@@ -99,3 +99,73 @@ def test_bollinger_reset_clears_state():
     assert len(strategy._closes) == 1
     assert strategy._was_below_lower is False
     assert strategy._was_above_upper is False
+    assert strategy._in_squeeze is False
+    assert strategy._squeeze_bars == 0
+
+
+def test_bollinger_squeeze_detection():
+    """Tight bandwidth → no signal during squeeze."""
+    # Use a strategy with a high squeeze threshold so constant prices trigger squeeze
+    s = BollingerStrategy()
+    s.configure({
+        "period": 5,
+        "num_std": 2.0,
+        "min_bandwidth": 0.0,
+        "squeeze_threshold": 0.5,  # Very high threshold to ensure squeeze triggers
+    })
+
+    # Feed identical prices → bandwidth = 0 (below any threshold)
+    for _ in range(6):
+        result = s.on_candle(make_candle(100.0))
+
+    # With identical prices, std=0, bandwidth=0 < 0.5 → squeeze, no signal
+    assert s._in_squeeze is True
+    assert result is None
+
+
+def test_bollinger_squeeze_breakout_buy():
+    """Squeeze ends with price above SMA → BUY signal."""
+    s = BollingerStrategy()
+    s.configure({
+        "period": 5,
+        "num_std": 1.0,
+        "min_bandwidth": 0.0,
+        "squeeze_threshold": 0.01,
+    })
+
+    # Feed identical prices to create a squeeze (bandwidth = 0)
+    for _ in range(6):
+        s.on_candle(make_candle(100.0))
+
+    assert s._in_squeeze is True
+
+    # Now feed a price that creates enough spread to exit squeeze AND is above SMA
+    signal = s.on_candle(make_candle(120.0))
+    assert signal is not None
+    assert signal.side == OrderSide.BUY
+    assert "squeeze breakout UP" in signal.reason
+
+
+def test_bollinger_pct_b_conviction():
+    """Signals near band extremes have higher conviction via %B."""
+    s = BollingerStrategy()
+    s.configure({
+        "period": 5,
+        "num_std": 1.0,
+        "min_bandwidth": 0.0,
+        "squeeze_threshold": 0.0,  # Disable squeeze for this test
+    })
+
+    # Build up with stable prices
+    for _ in range(5):
+        s.on_candle(make_candle(100.0))
+
+    # Drop below lower band
+    s.on_candle(make_candle(50.0))
+
+    # Recover just at the lower band edge — %B close to 0 → high conviction
+    signal = s.on_candle(make_candle(100.0))
+    assert signal is not None
+    assert signal.side == OrderSide.BUY
+    # conviction = max(1.0 - pct_b, 0.3), with pct_b near lower → conviction should be >= 0.3
+    assert signal.conviction >= 0.3