summaryrefslogtreecommitdiff
path: root/services/strategy-engine/tests
diff options
context:
space:
mode:
Diffstat (limited to 'services/strategy-engine/tests')
-rw-r--r--services/strategy-engine/tests/test_bollinger_strategy.py72
-rw-r--r--services/strategy-engine/tests/test_grid_strategy.py38
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