feat(strategy): add Bollinger Bands strategy

3 files changed, 191 insertions, 0 deletions

diff --git a/services/strategy-engine/strategies/bollinger_strategy.py b/services/strategy-engine/strategies/bollinger_strategy.py
new file mode 100644
index 0000000..bee7ee4
--- /dev/null
+++ b/services/strategy-engine/strategies/bollinger_strategy.py
@@ -0,0 +1,86 @@
+from collections import deque
+from decimal import Decimal
+
+import pandas as pd
+
+from shared.models import Candle, Signal, OrderSide
+from strategies.base import BaseStrategy
+
+
+class BollingerStrategy(BaseStrategy):
+    name: str = "bollinger"
+
+    def __init__(self) -> None:
+        self._closes: deque[float] = deque(maxlen=500)
+        self._period: int = 20
+        self._num_std: float = 2.0
+        self._min_bandwidth: float = 0.02
+        self._quantity: Decimal = Decimal("0.01")
+        self._was_below_lower: bool = False
+        self._was_above_upper: bool = False
+
+    @property
+    def warmup_period(self) -> int:
+        return self._period
+
+    def configure(self, params: dict) -> None:
+        self._period = int(params.get("period", 20))
+        self._num_std = float(params.get("num_std", 2.0))
+        self._min_bandwidth = float(params.get("min_bandwidth", 0.02))
+        self._quantity = Decimal(str(params.get("quantity", "0.01")))
+
+    def reset(self) -> None:
+        self._closes.clear()
+        self._was_below_lower = False
+        self._was_above_upper = False
+
+    def on_candle(self, candle: Candle) -> Signal | None:
+        self._closes.append(float(candle.close))
+
+        if len(self._closes) < self._period:
+            return None
+
+        series = pd.Series(list(self._closes))
+        sma = series.rolling(window=self._period).mean().iloc[-1]
+        std = series.rolling(window=self._period).std().iloc[-1]
+
+        upper = sma + self._num_std * std
+        lower = sma - self._num_std * std
+
+        # Bandwidth filter: skip sideways markets
+        if sma != 0 and (upper - lower) / sma < self._min_bandwidth:
+            return None
+
+        price = float(candle.close)
+
+        # Track band penetration
+        if price < lower:
+            self._was_below_lower = True
+        if price > upper:
+            self._was_above_upper = True
+
+        # BUY: was below lower band and recovered back inside
+        if self._was_below_lower and price >= lower:
+            self._was_below_lower = False
+            return Signal(
+                strategy=self.name,
+                symbol=candle.symbol,
+                side=OrderSide.BUY,
+                price=candle.close,
+                quantity=self._quantity,
+                reason=f"Price recovered above lower Bollinger Band ({lower:.2f})",
+            )
+
+        # SELL: was above upper band and recovered back inside
+        if self._was_above_upper and price <= upper:
+            self._was_above_upper = False
+            return Signal(
+                strategy=self.name,
+                symbol=candle.symbol,
+                side=OrderSide.SELL,
+                price=candle.close,
+                quantity=self._quantity,
+                reason=f"Price recovered below upper Bollinger Band ({upper:.2f})",
+            )
+
+        return None
diff --git a/services/strategy-engine/strategies/config/bollinger_strategy.yaml b/services/strategy-engine/strategies/config/bollinger_strategy.yaml
new file mode 100644
index 0000000..153cf81
--- /dev/null
+++ b/services/strategy-engine/strategies/config/bollinger_strategy.yaml
@@ -0,0 +1,4 @@
+period: 20
+num_std: 2.0
+min_bandwidth: 0.02
+quantity: "0.01"
diff --git a/services/strategy-engine/tests/test_bollinger_strategy.py b/services/strategy-engine/tests/test_bollinger_strategy.py
new file mode 100644
index 0000000..b3d17ac
--- /dev/null
+++ b/services/strategy-engine/tests/test_bollinger_strategy.py
@@ -0,0 +1,101 @@
+"""Tests for the Bollinger Bands strategy."""
+from datetime import datetime, timezone
+from decimal import Decimal
+
+import pytest
+
+from shared.models import Candle, OrderSide
+from strategies.bollinger_strategy import BollingerStrategy
+
+
+def make_candle(close: float) -> Candle:
+    return Candle(
+        symbol="BTC/USDT",
+        timeframe="1m",
+        open_time=datetime(2024, 1, 1, tzinfo=timezone.utc),
+        open=Decimal(str(close)),
+        high=Decimal(str(close)),
+        low=Decimal(str(close)),
+        close=Decimal(str(close)),
+        volume=Decimal("1.0"),
+    )
+
+
+def _make_strategy() -> BollingerStrategy:
+    s = BollingerStrategy()
+    s.configure({"period": 5, "num_std": 1.0, "min_bandwidth": 0.0})
+    return s
+
+
+def test_bollinger_warmup_period():
+    strategy = BollingerStrategy()
+    strategy.configure({"period": 20})
+    assert strategy.warmup_period == 20
+
+    strategy.configure({"period": 5})
+    assert strategy.warmup_period == 5
+
+
+def test_bollinger_no_signal_insufficient_data():
+    strategy = _make_strategy()
+    # Feed fewer candles than the period
+    for price in [100.0, 101.0, 102.0, 101.0]:
+        result = strategy.on_candle(make_candle(price))
+        assert result is None
+
+
+def test_bollinger_buy_on_lower_band_recovery():
+    strategy = _make_strategy()
+
+    # Feed stable prices to build up the window
+    for _ in range(5):
+        strategy.on_candle(make_candle(100.0))
+
+    # Drop well below the lower band
+    signal = strategy.on_candle(make_candle(50.0))
+    # No buy yet -- still below lower band
+    assert signal is None
+
+    # Recover back inside the bands
+    signal = strategy.on_candle(make_candle(100.0))
+    assert signal is not None
+    assert signal.side == OrderSide.BUY
+    assert "lower" in signal.reason.lower() or "bollinger" in signal.reason.lower()
+
+
+def test_bollinger_sell_on_upper_band_recovery():
+    strategy = _make_strategy()
+
+    # Feed stable prices to build up the window
+    for _ in range(5):
+        strategy.on_candle(make_candle(100.0))
+
+    # Spike well above the upper band
+    signal = strategy.on_candle(make_candle(150.0))
+    # No sell yet -- still above upper band
+    assert signal is None
+
+    # Recover back inside the bands
+    signal = strategy.on_candle(make_candle(100.0))
+    assert signal is not None
+    assert signal.side == OrderSide.SELL
+    assert "upper" in signal.reason.lower() or "bollinger" in signal.reason.lower()
+
+
+def test_bollinger_reset_clears_state():
+    strategy = _make_strategy()
+
+    # Build some state
+    for _ in range(5):
+        strategy.on_candle(make_candle(100.0))
+    strategy.on_candle(make_candle(50.0))  # penetrate lower band
+
+    strategy.reset()
+
+    # After reset, insufficient data again
+    result = strategy.on_candle(make_candle(100.0))
+    assert result is None
+    # Internal state should be cleared
+    assert len(strategy._closes) == 1
+    assert strategy._was_below_lower is False
+    assert strategy._was_above_upper is False