From 71e5942632a5a8c7cd555b2d52e5632a67186a8d Mon Sep 17 00:00:00 2001
From: TheSiahxyz <164138827+TheSiahxyz@users.noreply.github.com>
Date: Thu, 2 Apr 2026 09:17:35 +0900
Subject: feat(strategy): add Grid trend guard and Bollinger squeeze detection

---
 .../strategies/bollinger_strategy.py               | 61 +++++++++++++++++++---
 1 file changed, 54 insertions(+), 7 deletions(-)

(limited to 'services/strategy-engine/strategies/bollinger_strategy.py')

diff --git a/services/strategy-engine/strategies/bollinger_strategy.py b/services/strategy-engine/strategies/bollinger_strategy.py
index e53ecaa..a195cb8 100644
--- a/services/strategy-engine/strategies/bollinger_strategy.py
+++ b/services/strategy-engine/strategies/bollinger_strategy.py
@@ -19,6 +19,9 @@ class BollingerStrategy(BaseStrategy):
         self._quantity: Decimal = Decimal("0.01")
         self._was_below_lower: bool = False
         self._was_above_upper: bool = False
+        self._squeeze_threshold: float = 0.01  # Bandwidth below this = squeeze
+        self._in_squeeze: bool = False
+        self._squeeze_bars: int = 0  # How many bars in squeeze
 
     @property
     def warmup_period(self) -> int:
@@ -28,6 +31,7 @@ class BollingerStrategy(BaseStrategy):
         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._squeeze_threshold = float(params.get("squeeze_threshold", 0.01))
         self._quantity = Decimal(str(params.get("quantity", "0.01")))
 
         if self._period < 2:
@@ -46,9 +50,12 @@ class BollingerStrategy(BaseStrategy):
         )
 
     def reset(self) -> None:
+        super().reset()
         self._closes.clear()
         self._was_below_lower = False
         self._was_above_upper = False
+        self._in_squeeze = False
+        self._squeeze_bars = 0
 
     def _bollinger_conviction(self, price: float, band: float, sma: float) -> float:
         """Map distance from band to conviction (0.1-1.0).
@@ -75,12 +82,52 @@ class BollingerStrategy(BaseStrategy):
         upper = sma + self._num_std * std
         lower = sma - self._num_std * std
 
+        price = float(candle.close)
+
+        # %B calculation
+        bandwidth = (upper - lower) / sma if sma > 0 else 0
+        pct_b = (price - lower) / (upper - lower) if (upper - lower) > 0 else 0.5
+
+        # Squeeze detection
+        if bandwidth < self._squeeze_threshold:
+            self._in_squeeze = True
+            self._squeeze_bars += 1
+            return None  # Don't trade during squeeze, wait for breakout
+        elif self._in_squeeze:
+            # Squeeze just ended — breakout!
+            self._in_squeeze = False
+            squeeze_duration = self._squeeze_bars
+            self._squeeze_bars = 0
+
+            if price > sma:
+                # Breakout upward
+                conv = min(0.5 + squeeze_duration * 0.1, 1.0)
+                return self._apply_filters(Signal(
+                    strategy=self.name,
+                    symbol=candle.symbol,
+                    side=OrderSide.BUY,
+                    price=candle.close,
+                    quantity=self._quantity,
+                    conviction=conv,
+                    reason=f"Bollinger squeeze breakout UP after {squeeze_duration} bars",
+                ))
+            else:
+                # Breakout downward
+                conv = min(0.5 + squeeze_duration * 0.1, 1.0)
+                return self._apply_filters(Signal(
+                    strategy=self.name,
+                    symbol=candle.symbol,
+                    side=OrderSide.SELL,
+                    price=candle.close,
+                    quantity=self._quantity,
+                    conviction=conv,
+                    reason=f"Bollinger squeeze breakout DOWN after {squeeze_duration} bars",
+                ))
+
         # Bandwidth filter: skip sideways markets
-        if sma != 0 and (upper - lower) / sma < self._min_bandwidth:
+        if sma != 0 and bandwidth < self._min_bandwidth:
             return None
 
-        price = float(candle.close)
-
         # Track band penetration
         if price < lower:
             self._was_below_lower = True
@@ -90,14 +137,14 @@ 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
-            conviction = self._bollinger_conviction(price, lower, sma)
+            conv = max(1.0 - pct_b, 0.3)  # Closer to lower band = higher conviction
             signal = Signal(
                 strategy=self.name,
                 symbol=candle.symbol,
                 side=OrderSide.BUY,
                 price=candle.close,
                 quantity=self._quantity,
-                conviction=conviction,
+                conviction=conv,
                 reason=f"Price recovered above lower Bollinger Band ({lower:.2f})",
             )
             return self._apply_filters(signal)
@@ -105,14 +152,14 @@ class BollingerStrategy(BaseStrategy):
         # SELL: was above upper band and recovered back inside
         if self._was_above_upper and price <= upper:
             self._was_above_upper = False
-            conviction = self._bollinger_conviction(price, upper, sma)
+            conv = max(pct_b, 0.3)  # Closer to upper band = higher conviction
             signal = Signal(
                 strategy=self.name,
                 symbol=candle.symbol,
                 side=OrderSide.SELL,
                 price=candle.close,
                 quantity=self._quantity,
-                conviction=conviction,
+                conviction=conv,
                 reason=f"Price recovered below upper Bollinger Band ({upper:.2f})",
             )
             return self._apply_filters(signal)
-- 
cgit v1.2.3