diff options
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 |