diff options
Diffstat (limited to 'services/strategy-engine/strategies/bollinger_strategy.py')
| -rw-r--r-- | services/strategy-engine/strategies/bollinger_strategy.py | 30 |
1 files changed, 28 insertions, 2 deletions
diff --git a/services/strategy-engine/strategies/bollinger_strategy.py b/services/strategy-engine/strategies/bollinger_strategy.py index 1354182..e53ecaa 100644 --- a/services/strategy-engine/strategies/bollinger_strategy.py +++ b/services/strategy-engine/strategies/bollinger_strategy.py @@ -37,12 +37,32 @@ class BollingerStrategy(BaseStrategy): if self._quantity <= 0: raise ValueError(f"Quantity must be positive, got {self._quantity}") + self._init_filters( + require_trend=False, + adx_threshold=float(params.get("adx_threshold", 25.0)), + min_volume_ratio=float(params.get("min_volume_ratio", 0.5)), + atr_stop_multiplier=float(params.get("atr_stop_multiplier", 2.0)), + atr_tp_multiplier=float(params.get("atr_tp_multiplier", 3.0)), + ) + def reset(self) -> None: self._closes.clear() self._was_below_lower = False self._was_above_upper = False + def _bollinger_conviction(self, price: float, band: float, sma: float) -> float: + """Map distance from band to conviction (0.1-1.0). + + Further from band (relative to band width) = stronger signal. + """ + if sma == 0: + return 0.5 + distance = abs(price - band) / sma + # Scale: 0% distance -> 0.1, 2%+ distance -> ~1.0 + return min(1.0, max(0.1, distance * 50)) + def on_candle(self, candle: Candle) -> Signal | None: + self._update_filter_data(candle) self._closes.append(float(candle.close)) if len(self._closes) < self._period: @@ -70,25 +90,31 @@ class BollingerStrategy(BaseStrategy): # BUY: was below lower band and recovered back inside if self._was_below_lower and price >= lower: self._was_below_lower = False - return Signal( + conviction = self._bollinger_conviction(price, lower, sma) + signal = Signal( strategy=self.name, symbol=candle.symbol, side=OrderSide.BUY, price=candle.close, quantity=self._quantity, + conviction=conviction, reason=f"Price recovered above lower Bollinger Band ({lower:.2f})", ) + return self._apply_filters(signal) # SELL: was above upper band and recovered back inside if self._was_above_upper and price <= upper: self._was_above_upper = False - return Signal( + conviction = self._bollinger_conviction(price, upper, sma) + signal = Signal( strategy=self.name, symbol=candle.symbol, side=OrderSide.SELL, price=candle.close, quantity=self._quantity, + conviction=conviction, reason=f"Price recovered below upper Bollinger Band ({upper:.2f})", ) + return self._apply_filters(signal) return None |