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

2 files changed, 109 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
diff --git a/services/strategy-engine/tests/test_grid_strategy.py b/services/strategy-engine/tests/test_grid_strategy.py
index 79eb22a..9823f98 100644
--- a/services/strategy-engine/tests/test_grid_strategy.py
+++ b/services/strategy-engine/tests/test_grid_strategy.py
@@ -60,3 +60,41 @@ def test_grid_strategy_no_signal_in_same_zone():
     strategy.on_candle(make_candle(50000))
     signal = strategy.on_candle(make_candle(50100))
     assert signal is None
+
+
+def test_grid_exit_on_trend_break():
+    """Price drops well below grid range → SELL signal emitted."""
+    strategy = _configured_strategy()
+    # Grid range is 48000-52000, exit_threshold_pct defaults to 5%
+    # Lower bound = 48000 * 0.95 = 45600
+    # Establish a zone first
+    strategy.on_candle(make_candle(50000))
+    # Price drops far below the grid range
+    signal = strategy.on_candle(make_candle(45000))
+    assert signal is not None
+    assert signal.side == OrderSide.SELL
+    assert "broke out of range" in signal.reason
+
+
+def test_grid_no_signal_while_out_of_range():
+    """After exit signal, no more grid signals until price returns to range."""
+    strategy = _configured_strategy()
+    # Establish a zone
+    strategy.on_candle(make_candle(50000))
+    # First out-of-range candle → SELL exit signal
+    signal = strategy.on_candle(make_candle(45000))
+    assert signal is not None
+    assert signal.side == OrderSide.SELL
+
+    # Subsequent out-of-range candles → no signals
+    signal = strategy.on_candle(make_candle(44000))
+    assert signal is None
+
+    signal = strategy.on_candle(make_candle(43000))
+    assert signal is None
+
+    # Price returns to grid range → grid signals resume
+    strategy.on_candle(make_candle(50000))
+    signal = strategy.on_candle(make_candle(48100))
+    assert signal is not None
+    assert signal.side == OrderSide.BUY